#!/usr/bin/env python3
"""
Apex Instruments Q3 2026 order book: ingest the messy export, report what changed,
write a formatted workbook.

Apex Instruments is fictional. Every name, figure, and contact is demo data.
Produced by the Company OS spreadsheet system for djvonfrank.com.

    python3 build-workbook.py

Reads   data/raw-orders.csv, data/partners.csv
Writes  out/apex-orders-2026-q3.xlsx   six sheets, real formulas, two native charts
        data/clean-report.json         machine readable record of the run
        data/preview.json              row level before and after, for the web widget

Design rule: nothing is silently discarded. A row is either cleaned, with every
change recorded against a named rule, or it is held back and listed with a reason.
"""
import csv, json, os, re, sys
from collections import Counter, OrderedDict
from datetime import datetime, date

from openpyxl import Workbook
from openpyxl.chart import BarChart, LineChart, Reference
from openpyxl.chart.marker import Marker
from openpyxl.formatting.rule import CellIsRule, ColorScaleRule
from openpyxl.styles import Alignment, Border, Font, NamedStyle, PatternFill, Side
from openpyxl.utils import get_column_letter

HERE = os.path.dirname(os.path.abspath(__file__))
DATA = os.path.join(HERE, "data")
OUT = os.path.join(HERE, "out")
os.makedirs(OUT, exist_ok=True)
XLSX = os.path.join(OUT, "apex-orders-2026-q3.xlsx")

BUILT = datetime.now().strftime("%Y-%m-%d %H:%M")
BUILT_DAY = datetime.now().strftime("%Y-%m-%d")

# Brand
NAVY, STEEL, ORANGE, MIST, LINE, MUTED, WHITE = (
    "00272C", "245D63", "FF6B1C", "EDF3F3", "DEE3E4", "476467", "FFFFFF")

# ============================================================ the catalog ====
CATALOG = OrderedDict([
    ("Meridian Bench",      {"price": 4850.00, "cost": 2910.00}),
    ("Meridian Pro",        {"price": 7900.00, "cost": 4345.00}),
    ("Meridian Field Kit",  {"price": 1240.00, "cost":  806.00}),
    ("Atlas Care Plus",     {"price": 1788.00, "cost":  447.00}),
    ("Calibration service", {"price":  180.00, "cost":   99.00}),
    ("On-site training",    {"price":  650.00, "cost":  429.00}),
])
PRODUCTS = list(CATALOG.keys())

# Every spelling the source system has produced, folded to a lookup key.
ALIAS = {}
def _k(s):
    """Fold a product name to a match key: lowercase, letters and digits only."""
    return re.sub(r"[^a-z0-9]", "", (s or "").lower())
for canon in PRODUCTS:
    ALIAS[_k(canon)] = canon
for extra, canon in [
    ("bench", "Meridian Bench"),
    ("pro", "Meridian Pro"),
    ("fieldkit", "Meridian Field Kit"), ("meridianfieldkit", "Meridian Field Kit"),
    ("careplus", "Atlas Care Plus"), ("atlascare", "Atlas Care Plus"),
    ("calibration", "Calibration service"), ("calibrationsvc", "Calibration service"),
    ("training", "On-site training"), ("onsitetraining", "On-site training"),
]:
    ALIAS[extra] = canon

CITY_FIX = {"stpetersburg": "St. Petersburg", "stpete": "St. Petersburg"}
CITIES = ["Tampa", "St. Petersburg", "Orlando", "Lakeland", "Sarasota", "Clearwater", "Brandon"]
CITY_KEY = dict((_k(c), c) for c in CITIES)
CITY_KEY.update(CITY_FIX)

CHANNELS = {"direct": "Direct", "partner": "Partner"}
STATUSES = {"shipped": "Shipped", "invoiced": "Invoiced", "open": "Open", "cancelled": "Cancelled"}
PLACEHOLDERS = {"--", "-", "n/a", "na", "tbd", "none", "null", "?", "."}

REQUIRED_HEADERS = {"order_id", "order_date", "product", "qty"}

# ============================================================= the rules =====
# id, short name, what it does, the unit its count is measured in
RULE_DEFS = [
    ("R01", "Header row located",
     "Skips the export preamble and starts at the first line that carries the real column names.", "file"),
    ("R02", "Date normalised to ISO",
     "Five date shapes appear in the export. Each is parsed and rewritten as a real date value.", "rows"),
    ("R03", "Currency parsed to a number",
     "Strips the dollar sign, thousands separators and stray spaces so the value is arithmetic, not text.", "cells"),
    ("R04", "Whitespace trimmed",
     "Removes leading and trailing spaces and collapses runs of spaces inside a value.", "cells"),
    ("R05", "Ship-to city unified",
     "One spelling per location. St Petersburg and ST PETERSBURG both become St. Petersburg.", "rows"),
    ("R06", "Product matched to the catalog",
     "Case, spacing and short forms are matched to the one catalog name. Field Kit becomes Meridian Field Kit.", "rows"),
    ("R07", "Partner id normalised",
     "Upper cased and trimmed, then joined to the partner list for the partner name and region.", "rows"),
    ("R08", "Discount normalised to a fraction",
     "A discount typed as 5% or 5.0% is divided by 100 and stored as 0.05, next to the ones already written as 0.05.", "rows"),
    ("R09", "Placeholder treated as empty",
     "Tokens that mean no value, such as N/A, TBD and a double hyphen, are read as empty, not as text.", "cells"),
    ("R10", "Whole number quantity made an integer",
     "A quantity written as 2.0 is stored as 2 so it counts and sorts as a number.", "rows"),
    ("R11", "Missing unit price filled from the catalog",
     "A blank price is filled from the product list price and the fill is recorded on the row.", "rows"),
    ("R12", "Missing ship-to city filled from the partner",
     "A blank city is filled from the partner record rather than left empty or guessed.", "rows"),
    ("R13", "Channel and status matched to the allowed values",
     "PARTNER, partner and Partner all become Partner. Anything outside the allowed set stops the row.", "rows"),
]
REJECT_DEFS = [
    ("X1", "Duplicate order id. The first row with this id was imported, this one was not."),
    ("X2", "A required field is empty: order id, order date, product or quantity."),
    ("X3", "Quantity is zero or negative. A return needs a credit note, not an order row."),
    ("X4", "The order date could not be read as a date in any known format."),
    ("X5", "The product is not in the Apex catalog."),
    ("X6", "The partner id is not in the partner list."),
    ("X7", "Quantity or unit price is not a number."),
    ("X8", "Channel or status is outside the values this workbook allows."),
]

rule_hits = Counter()
rule_example = {}
reject_rows = []
reject_counts = Counter()


def fire(rule, line, column, before, after):
    rule_hits[rule] += 1
    if rule not in rule_example:
        rule_example[rule] = {"line": line, "column": column,
                              "before": before, "after": after}


# ======================================================= value handling ======
def squeeze(s):
    """Trim and collapse internal whitespace. Returns (value, changed)."""
    if s is None:
        return "", False
    t = re.sub(r"\s+", " ", s).strip()
    return t, (t != s)


def is_placeholder(s):
    return s.strip().lower() in PLACEHOLDERS


MONTHS = {m.lower(): i + 1 for i, m in enumerate(
    ["January", "February", "March", "April", "May", "June", "July",
     "August", "September", "October", "November", "December"])}
for m, i in list(MONTHS.items()):
    MONTHS[m[:3]] = i


def parse_date(s):
    """Return a date, or None. US convention (month first) for slash dates."""
    t = s.strip()
    if not t:
        return None
    m = re.match(r"^(\d{4})-(\d{1,2})-(\d{1,2})$", t)              # 2026-07-14
    if m:
        y, mo, d = int(m.group(1)), int(m.group(2)), int(m.group(3))
    else:
        m = re.match(r"^(\d{1,2})/(\d{1,2})/(\d{2}|\d{4})$", t)    # 07/14/2026, 7/14/26
        if m:
            mo, d, y = int(m.group(1)), int(m.group(2)), int(m.group(3))
            if y < 100:
                y += 2000
        else:
            m = re.match(r"^(\d{1,2})-([A-Za-z]{3,})-(\d{4})$", t)  # 14-Jul-2026
            if m:
                d, mo, y = int(m.group(1)), MONTHS.get(m.group(2).lower()), int(m.group(3))
            else:
                m = re.match(r"^([A-Za-z]{3,})\s+(\d{1,2}),\s*(\d{4})$", t)  # July 14, 2026
                if m:
                    mo, d, y = MONTHS.get(m.group(1).lower()), int(m.group(2)), int(m.group(3))
                else:
                    return None
    if not mo or not (1 <= mo <= 12) or not (1 <= d <= 31):
        return None
    try:
        return date(y, mo, d)
    except ValueError:
        return None


MONEY_RE = re.compile(r"^-?[\d,]*\.?\d+$")


def parse_money(s):
    """Return (value, had_symbols) or (None, False) when it is not a number."""
    t = s.strip()
    if not t:
        return None, False
    stripped = t.replace("$", "").replace(",", "").replace(" ", "")
    if not MONEY_RE.match(stripped):
        return None, False
    return float(stripped), (stripped != t)


def parse_pct(s):
    """Return (fraction, was_a_percentage) or (None, False).

    A value already written as a decimal fraction needs no conversion, so it does
    not count as a fix. Only a percentage string does."""
    t = s.strip()
    if not t:
        return 0.0, False
    if t.endswith("%"):
        body = t[:-1].strip().replace(",", "")
        if not MONEY_RE.match(body):
            return None, False
        return round(float(body) / 100.0, 6), True
    body = t.replace(",", "")
    if not MONEY_RE.match(body):
        return None, False
    return round(float(body), 6), False


# ========================================================= read the files ====
with open(os.path.join(DATA, "partners.csv"), newline="", encoding="utf-8") as f:
    PARTNERS = {r["partner_id"].strip().upper(): r for r in csv.DictReader(f)}

with open(os.path.join(DATA, "raw-orders.csv"), newline="", encoding="utf-8") as f:
    RAW_LINES = list(csv.reader(f))

# R01: find the header row instead of assuming it is line 1.
header_idx = None
for i, row in enumerate(RAW_LINES):
    cells = {c.strip().lower() for c in row}
    if REQUIRED_HEADERS.issubset(cells):
        header_idx = i
        break
if header_idx is None:
    sys.exit("No header row found. Looked for %s." % ", ".join(sorted(REQUIRED_HEADERS)))
if header_idx > 0:
    fire("R01", header_idx + 1, "file",
         "%d preamble lines above the header" % header_idx,
         "header read from line %d" % (header_idx + 1))

HEAD = [c.strip().lower() for c in RAW_LINES[header_idx]]
COL = {name: i for i, name in enumerate(HEAD)}
BODY = [r for r in RAW_LINES[header_idx + 1:] if any((c or "").strip() for c in r)]

DISPLAY = ["order_id", "order_date", "partner_id", "ship_city", "product",
           "qty", "unit_price", "discount", "channel", "status"]
CLEAN_DISPLAY = ["order_id", "order_date", "partner_name", "region", "ship_city",
                 "product", "qty", "unit_price", "discount", "net_revenue"]

clean = []
preview = []
seen_ids = {}
raw_city_dist, raw_product_dist = Counter(), Counter()
cells_changed = 0


def get(row, name):
    i = COL.get(name)
    return row[i] if i is not None and i < len(row) else ""


for n, row in enumerate(BODY):
    line = header_idx + 2 + n          # 1 based line number in the source file
    flags = {}
    fired = []

    def mark(col_name, kind):
        if col_name in DISPLAY:
            flags[str(DISPLAY.index(col_name))] = kind

    raw_display = [get(row, c) for c in DISPLAY]

    def shown(name):
        """The raw value, or empty when it holds nothing but whitespace, so a
        listing can say (empty) instead of printing three invisible spaces."""
        v = get(row, name)
        return v if v.strip() else ""

    def reject(code, detail=""):
        reject_counts[code] += 1
        reason = dict(REJECT_DEFS)[code]
        reject_rows.append({
            "line": line, "code": code, "reason": reason, "detail": detail,
            "order_id": shown("order_id"), "order_date": shown("order_date"),
            "partner_id": shown("partner_id"), "product": shown("product"),
            "qty": shown("qty"), "unit_price": shown("unit_price"),
        })
        preview.append({"n": line, "r": raw_display, "f": flags,
                        "s": "rej", "x": code, "why": reason, "detail": detail})

    # ---- R04 / R09 across every cell ------------------------------------
    vals = {}
    for name in HEAD:
        v0 = get(row, name)
        v, changed = squeeze(v0)
        if changed:
            cells_changed += 1
            fire("R04", line, name, repr(v0), repr(v))
            if "R04" not in fired:
                fired.append("R04")
            mark(name, "space")
        if v and is_placeholder(v):
            fire("R09", line, name, v, "(empty)")
            cells_changed += 1
            if "R09" not in fired:
                fired.append("R09")
            mark(name, "blank")
            v = ""
        vals[name] = v

    raw_city_dist[vals["ship_city"] or "(empty)"] += 1
    raw_product_dist[vals["product"] or "(empty)"] += 1

    # ---- required fields -------------------------------------------------
    missing = [c for c in ("order_id", "order_date", "product", "qty") if not vals[c]]
    if missing:
        for c in missing:
            mark(c, "blank")
        reject("X2", "empty: " + ", ".join(missing))
        continue

    # ---- date ------------------------------------------------------------
    d = parse_date(vals["order_date"])
    if d is None:
        mark("order_date", "bad")
        reject("X4", "value read: %s" % vals["order_date"])
        continue
    if vals["order_date"] != d.isoformat():
        fire("R02", line, "order_date", vals["order_date"], d.isoformat())
        fired.append("R02")
        mark("order_date", "date")

    # ---- partner ---------------------------------------------------------
    pid = vals["partner_id"].upper()
    if pid != get(row, "partner_id"):
        fire("R07", line, "partner_id", get(row, "partner_id"), pid)
        fired.append("R07")
        mark("partner_id", "partner")
    prow = PARTNERS.get(pid)
    if prow is None:
        mark("partner_id", "bad")
        reject("X6", "id read: %s" % (pid or "(empty)"))
        continue

    # ---- product ---------------------------------------------------------
    canon = ALIAS.get(_k(vals["product"]))
    if canon is None:
        mark("product", "bad")
        reject("X5", "name read: %s" % vals["product"])
        continue
    if canon != vals["product"]:
        fire("R06", line, "product", vals["product"], canon)
        fired.append("R06")
        mark("product", "product")

    # ---- quantity --------------------------------------------------------
    qtxt = vals["qty"]
    qv, _ = parse_money(qtxt)
    if qv is None:
        mark("qty", "bad")
        reject("X7", "quantity read: %s" % qtxt)
        continue
    if qv <= 0:
        mark("qty", "neg")
        reject("X3", "quantity read: %s" % qtxt)
        continue
    if qv != int(qv):
        mark("qty", "bad")
        reject("X7", "quantity is not whole: %s" % qtxt)
        continue
    qty = int(qv)
    if qtxt != str(qty):
        fire("R10", line, "qty", qtxt, str(qty))
        fired.append("R10")
        mark("qty", "num")

    # ---- unit price ------------------------------------------------------
    ptxt = vals["unit_price"]
    if not ptxt:
        price = CATALOG[canon]["price"]
        fire("R11", line, "unit_price", "(empty)", "%.2f from the catalog" % price)
        fired.append("R11")
        mark("unit_price", "fill")
    else:
        price, had = parse_money(ptxt)
        if price is None:
            mark("unit_price", "bad")
            reject("X7", "unit price read: %s" % ptxt)
            continue
        if had:
            fire("R03", line, "unit_price", ptxt, "%.2f" % price)
            cells_changed += 1
            fired.append("R03")
            mark("unit_price", "money")

    # ---- discount --------------------------------------------------------
    disc, rewritten = parse_pct(vals["discount"])
    if disc is None:
        mark("discount", "bad")
        reject("X7", "discount read: %s" % vals["discount"])
        continue
    if rewritten and vals["discount"] != "":
        fire("R08", line, "discount", vals["discount"], "%.4f" % disc)
        fired.append("R08")
        mark("discount", "pct")

    # ---- ship cost -------------------------------------------------------
    stxt = vals["ship_cost"]
    if not stxt:
        ship = 0.0
    else:
        ship, had = parse_money(stxt)
        if ship is None:
            ship = 0.0
        elif had:
            fire("R03", line, "ship_cost", stxt, "%.2f" % ship)
            cells_changed += 1
            if "R03" not in fired:
                fired.append("R03")

    # ---- city ------------------------------------------------------------
    city_raw = vals["ship_city"]
    if not city_raw:
        city = prow["site_city"]
        fire("R12", line, "ship_city", "(empty)", "%s from %s" % (city, pid))
        fired.append("R12")
        mark("ship_city", "fill")
    else:
        city = CITY_KEY.get(_k(city_raw), city_raw)
        if city != city_raw:
            fire("R05", line, "ship_city", city_raw, city)
            fired.append("R05")
            mark("ship_city", "city")

    # ---- channel and status ---------------------------------------------
    ch = CHANNELS.get(vals["channel"].lower())
    st = STATUSES.get(vals["status"].lower())
    if ch is None or st is None:
        mark("channel" if ch is None else "status", "bad")
        reject("X8", "read: %s / %s" % (vals["channel"] or "(empty)", vals["status"] or "(empty)"))
        continue
    if ch != vals["channel"] or st != vals["status"]:
        fire("R13", line, "channel/status",
             "%s / %s" % (vals["channel"], vals["status"]), "%s / %s" % (ch, st))
        fired.append("R13")
        if ch != vals["channel"]:
            mark("channel", "case")
        if st != vals["status"]:
            mark("status", "case")

    # ---- duplicate check, last, so a bad row never claims an id ----------
    oid = vals["order_id"].upper()
    if oid in seen_ids:
        mark("order_id", "dup")
        reject("X1", "first seen on line %d" % seen_ids[oid])
        continue
    seen_ids[oid] = line

    cost = CATALOG[canon]["cost"]
    gross = round(qty * price, 2)
    net = round(gross * (1 - disc), 2)
    rec = {
        "line": line, "order_id": oid, "order_date": d, "month": d.strftime("%Y-%m"),
        "partner_id": pid, "partner_name": prow["partner_name"], "region": prow["region"],
        "ship_city": city, "product": canon, "qty": qty, "unit_price": price,
        "discount": disc, "unit_cost": cost, "ship_cost": ship,
        "channel": ch, "rep": vals["rep"], "status": st, "notes": vals["notes"],
        "gross": gross, "net": net, "cost": round(qty * cost, 2),
        "fired": fired,
    }
    clean.append(rec)
    preview.append({
        "n": line, "r": raw_display, "f": flags,
        "s": "fix" if fired else "ok", "rules": fired,
        "c": [oid, d.isoformat(), prow["partner_name"], prow["region"], city, canon,
              str(qty), "%.2f" % price, "%.4f" % disc, "%.2f" % net],
    })

rows_in = len(BODY)
rows_out = len(clean)
rows_rejected = len(reject_rows)
rows_fixed = sum(1 for r in clean if r["fired"])
assert rows_out + rows_rejected == rows_in, "every row must be accounted for"

# ================================================== totals for the report ====
def s(f):
    return round(sum(f(r) for r in clean), 2)

MONTH_KEYS = ["2026-07", "2026-08", "2026-09"]
MONTH_LABEL = {"2026-07": "July 2026", "2026-08": "August 2026", "2026-09": "September 2026"}

totals = {
    "orders": rows_out,
    "units": sum(r["qty"] for r in clean),
    "gross": s(lambda r: r["gross"]),
    "net": s(lambda r: r["net"]),
    "cost": s(lambda r: r["cost"]),
    "ship": s(lambda r: r["ship_cost"]),
}
totals["discount"] = round(totals["gross"] - totals["net"], 2)
totals["margin"] = round(totals["net"] - totals["cost"], 2)
totals["marginPct"] = round(totals["margin"] / totals["net"], 6)
totals["aov"] = round(totals["net"] / totals["orders"], 2)

by_month = []
for mk in MONTH_KEYS:
    rs = [r for r in clean if r["month"] == mk]
    net = round(sum(r["net"] for r in rs), 2)
    cost = round(sum(r["cost"] for r in rs), 2)
    by_month.append({
        "key": mk, "label": MONTH_LABEL[mk], "orders": len(rs),
        "units": sum(r["qty"] for r in rs),
        "gross": round(sum(r["gross"] for r in rs), 2),
        "net": net, "cost": cost, "margin": round(net - cost, 2),
        "marginPct": round((net - cost) / net, 6) if net else 0.0,
    })

by_product = []
for p in PRODUCTS:
    rs = [r for r in clean if r["product"] == p]
    net = round(sum(r["net"] for r in rs), 2)
    cost = round(sum(r["cost"] for r in rs), 2)
    by_product.append({
        "product": p, "orders": len(rs), "units": sum(r["qty"] for r in rs),
        "gross": round(sum(r["gross"] for r in rs), 2),
        "net": net, "cost": cost, "margin": round(net - cost, 2),
        "marginPct": round((net - cost) / net, 6) if net else 0.0,
        "share": round(net / totals["net"], 6) if totals["net"] else 0.0,
    })

# ============================================================ the workbook ===
wb = Workbook()

thin = Side(style="thin", color=LINE)
med = Side(style="medium", color=NAVY)

def ns(name, **kw):
    st = NamedStyle(name=name)
    st.font = kw.get("font", Font(name="Calibri", size=11, color=NAVY))
    if "fill" in kw:
        st.fill = kw["fill"]
    if "align" in kw:
        st.alignment = kw["align"]
    if "border" in kw:
        st.border = kw["border"]
    if "fmt" in kw:
        st.number_format = kw["fmt"]
    wb.add_named_style(st)
    return name

LEFT = Alignment(horizontal="left", vertical="top")
LEFTW = Alignment(horizontal="left", vertical="top", wrap_text=True)
RIGHT = Alignment(horizontal="right", vertical="center")
CENTER = Alignment(horizontal="center", vertical="center")

S_H1 = ns("apexH1", font=Font(name="Calibri", size=20, bold=True, color=NAVY), align=LEFT)
S_H2 = ns("apexH2", font=Font(name="Calibri", size=13, bold=True, color=STEEL), align=LEFT)
S_LABEL = ns("apexLabel", font=Font(name="Consolas", size=9, bold=True, color=MUTED), align=LEFT)
S_BODY = ns("apexBody", font=Font(name="Calibri", size=11, color=NAVY), align=LEFT)
S_WRAP = ns("apexWrap", font=Font(name="Calibri", size=11, color=NAVY), align=LEFTW)
S_MUTED = ns("apexMuted", font=Font(name="Calibri", size=10, color=MUTED), align=LEFTW)
S_TH = ns("apexTh", font=Font(name="Calibri", size=10, bold=True, color=WHITE),
          fill=PatternFill("solid", fgColor=NAVY), align=Alignment(horizontal="left", vertical="center", wrap_text=True),
          border=Border(bottom=med))
S_THR = ns("apexThR", font=Font(name="Calibri", size=10, bold=True, color=WHITE),
           fill=PatternFill("solid", fgColor=NAVY), align=Alignment(horizontal="right", vertical="center", wrap_text=True),
           border=Border(bottom=med))
S_TXT = ns("apexTxt", font=Font(name="Calibri", size=11, color=NAVY), align=LEFT,
           border=Border(bottom=thin))
S_TXTW = ns("apexTxtW", font=Font(name="Calibri", size=10, color=NAVY), align=LEFTW,
            border=Border(bottom=thin))
S_MONO = ns("apexMono", font=Font(name="Consolas", size=10, color=NAVY), align=LEFT,
            border=Border(bottom=thin))
S_INT = ns("apexInt", font=Font(name="Calibri", size=11, color=NAVY), align=RIGHT,
           border=Border(bottom=thin), fmt="#,##0")
S_MONEY = ns("apexMoney", font=Font(name="Calibri", size=11, color=NAVY), align=RIGHT,
             border=Border(bottom=thin), fmt='"$"#,##0.00')
S_MONEY0 = ns("apexMoney0", font=Font(name="Calibri", size=11, color=NAVY), align=RIGHT,
              border=Border(bottom=thin), fmt='"$"#,##0')
S_PCT = ns("apexPct", font=Font(name="Calibri", size=11, color=NAVY), align=RIGHT,
           border=Border(bottom=thin), fmt="0.0%")
S_DATE = ns("apexDate", font=Font(name="Calibri", size=11, color=NAVY), align=RIGHT,
            border=Border(bottom=thin), fmt="yyyy-mm-dd")
S_KPIL = ns("apexKpiL", font=Font(name="Calibri", size=11, color=MUTED), align=LEFT)
S_KPIV = ns("apexKpiV", font=Font(name="Calibri", size=16, bold=True, color=NAVY),
            align=Alignment(horizontal="right", vertical="center"), fmt='"$"#,##0')
S_KPIN = ns("apexKpiN", font=Font(name="Calibri", size=16, bold=True, color=NAVY),
            align=Alignment(horizontal="right", vertical="center"), fmt="#,##0")
S_KPIP = ns("apexKpiP", font=Font(name="Calibri", size=16, bold=True, color=ORANGE),
            align=Alignment(horizontal="right", vertical="center"), fmt="0.0%")
S_TOTL = ns("apexTotL", font=Font(name="Calibri", size=11, bold=True, color=NAVY),
            align=LEFT, fill=PatternFill("solid", fgColor=MIST), border=Border(top=med))
S_TOTI = ns("apexTotI", font=Font(name="Calibri", size=11, bold=True, color=NAVY),
            align=RIGHT, fill=PatternFill("solid", fgColor=MIST), border=Border(top=med), fmt="#,##0")
S_TOTM = ns("apexTotM", font=Font(name="Calibri", size=11, bold=True, color=NAVY),
            align=RIGHT, fill=PatternFill("solid", fgColor=MIST), border=Border(top=med), fmt='"$"#,##0.00')
S_TOTP = ns("apexTotP", font=Font(name="Calibri", size=11, bold=True, color=NAVY),
            align=RIGHT, fill=PatternFill("solid", fgColor=MIST), border=Border(top=med), fmt="0.0%")


def put(ws, ref, value, style=None, fmt=None):
    c = ws[ref] if isinstance(ref, str) else ref
    c.value = value
    if style:
        c.style = style
    if fmt:
        c.number_format = fmt
    return c


def widths(ws, spec):
    for col, w in spec.items():
        ws.column_dimensions[col].width = w


# ---------------------------------------------------------- 1. Clean data ---
wsC = wb.active
wsC.title = "Clean data"
CLEAN_COLS = [
    ("Order id", "A", 12), ("Order date", "B", 12), ("Month", "C", 9),
    ("Partner id", "D", 11), ("Partner", "E", 25), ("Region", "F", 16),
    ("Ship to city", "G", 15), ("Product", "H", 20), ("Qty", "I", 7),
    ("Unit price", "J", 12), ("Discount", "K", 10), ("Gross", "L", 13),
    ("Net revenue", "M", 13), ("Unit cost", "N", 11), ("Cost", "O", 13),
    ("Margin", "P", 9), ("Ship cost", "Q", 11), ("Channel", "R", 10),
    ("Rep", "S", 15), ("Status", "T", 10), ("Source line", "U", 12),
    ("Notes", "V", 26),
]
for title, col, w in CLEAN_COLS:
    put(wsC, "%s1" % col, title, S_THR if col in "IJKLMNOPQU" else S_TH)
    wsC.column_dimensions[col].width = w
wsC.row_dimensions[1].height = 26

for i, r in enumerate(clean):
    row = i + 2
    put(wsC, "A%d" % row, r["order_id"], S_MONO)
    put(wsC, "B%d" % row, r["order_date"], S_DATE)
    put(wsC, "C%d" % row, '=TEXT(B%d,"yyyy-mm")' % row, S_MONO)
    put(wsC, "D%d" % row, r["partner_id"], S_MONO)
    put(wsC, "E%d" % row, r["partner_name"], S_TXT)
    put(wsC, "F%d" % row, r["region"], S_TXT)
    put(wsC, "G%d" % row, r["ship_city"], S_TXT)
    put(wsC, "H%d" % row, r["product"], S_TXT)
    put(wsC, "I%d" % row, r["qty"], S_INT)
    put(wsC, "J%d" % row, r["unit_price"], S_MONEY)
    put(wsC, "K%d" % row, r["discount"], S_PCT)
    put(wsC, "L%d" % row, "=I%d*J%d" % (row, row), S_MONEY)
    put(wsC, "M%d" % row, "=ROUND(L%d*(1-K%d),2)" % (row, row), S_MONEY)
    put(wsC, "N%d" % row, r["unit_cost"], S_MONEY)
    put(wsC, "O%d" % row, "=I%d*N%d" % (row, row), S_MONEY)
    put(wsC, "P%d" % row, "=IFERROR((M%d-O%d)/M%d,0)" % (row, row, row), S_PCT)
    put(wsC, "Q%d" % row, r["ship_cost"], S_MONEY)
    put(wsC, "R%d" % row, r["channel"], S_TXT)
    put(wsC, "S%d" % row, r["rep"], S_TXT)
    put(wsC, "T%d" % row, r["status"], S_TXT)
    put(wsC, "U%d" % row, r["line"], S_INT)
    put(wsC, "V%d" % row, r["notes"], S_TXT)

LAST = rows_out + 1
wsC.freeze_panes = "C2"
wsC.auto_filter.ref = "A1:V%d" % LAST
wsC.sheet_view.showGridLines = False
wsC.sheet_properties.tabColor = STEEL
wsC.print_title_rows = "1:1"


def rng(col):
    return "'Clean data'!$%s$2:$%s$%d" % (col, col, LAST)


# ------------------------------------------------------------- 2. Read me ---
wsR = wb.create_sheet("Read me", 0)
widths(wsR, {"A": 14, "B": 46, "C": 11, "D": 30, "E": 34})
wsR.sheet_view.showGridLines = False
wsR.sheet_properties.tabColor = NAVY

put(wsR, "A1", "Apex Instruments", S_H1)
put(wsR, "A2", "Q3 2026 order book, cleaned and summarised", S_H2)
wsR.row_dimensions[1].height = 27

lines = [
    ("A4", "Built %s by build-workbook.py, openpyxl %s, from data/raw-orders.csv." % (BUILT, __import__("openpyxl").__version__)),
    ("A5", "Source file: 405 lines, %d preamble lines, header on line %d, %d order rows." % (header_idx, header_idx + 1, rows_in)),
    ("A6", "Result: %d rows imported, %d of them changed by at least one rule, %d rows held back." % (rows_out, rows_fixed, rows_rejected)),
]
for ref, text in lines:
    put(wsR, ref, text, S_BODY)

put(wsR, "A8", "WHAT THIS WORKBOOK IS", S_LABEL)
for i, t in enumerate([
    "One quarter of Apex Instruments order lines, taken from a raw system export and turned into",
    "something a finance or operations lead can act on without asking anyone what a number means.",
    "Every figure on Summary, By month and By product is a live formula reading Clean data, so",
    "correcting a row on Clean data moves every total that depends on it.",
]):
    put(wsR, "A%d" % (9 + i), t, S_BODY)

put(wsR, "A14", "WHAT EACH SHEET DOES", S_LABEL)
for i, t in enumerate([
    "Summary        Fifteen figures for the quarter, each one a formula, plus two reconciliation checks.",
    "By month       July, August and September, with a line chart of net revenue and gross margin.",
    "By product     The six catalog lines, with a bar chart and conditional formatting on margin.",
    "Clean data     Every imported row, frozen header, filter on, dates and money formatted.",
    "Rejected rows  Every row that was not imported, with the reason and the source line number.",
]):
    put(wsR, "A%d" % (15 + i), t, S_BODY)

put(wsR, "A21", "CLEANING RULES APPLIED", S_LABEL)
put(wsR, "A22", "Nothing was changed quietly. Each rule below reports how many times it fired and one real example.", S_BODY)
hdr = 24
for col, title in zip("ABCDE", ["Rule", "What it fixes", "Times", "Example in", "Example out"]):
    put(wsR, "%s%d" % (col, hdr), title, S_THR if col == "C" else S_TH)
wsR.row_dimensions[hdr].height = 18
r = hdr + 1
for rid, name, what, unit in RULE_DEFS:
    ex = rule_example.get(rid, {})
    put(wsR, "A%d" % r, rid, S_MONO)
    put(wsR, "B%d" % r, "%s. %s" % (name, what), S_TXTW)
    put(wsR, "C%d" % r, "%d %s" % (rule_hits.get(rid, 0), unit), S_TXT)
    put(wsR, "D%d" % r, str(ex.get("before", "")), S_TXTW)
    put(wsR, "E%d" % r, str(ex.get("after", "")), S_TXTW)
    wsR.row_dimensions[r].height = 30
    r += 1

r += 1
put(wsR, "A%d" % r, "ROWS HELD BACK", S_LABEL)
r += 1
put(wsR, "A%d" % r, "Held back means not imported and not deleted. All %d are listed on Rejected rows with their source line." % rows_rejected, S_BODY)
r += 2
for col, title in zip("ABCD", ["Code", "Reason", "Rows", "Where the rows came from"]):
    put(wsR, "%s%d" % (col, r), title, S_THR if col == "C" else S_TH)
wsR.row_dimensions[r].height = 18
r += 1
for code, reason in REJECT_DEFS:
    lines_for = [str(x["line"]) for x in reject_rows if x["code"] == code]
    put(wsR, "A%d" % r, code, S_MONO)
    put(wsR, "B%d" % r, reason, S_TXTW)
    put(wsR, "C%d" % r, len(lines_for), S_INT)
    put(wsR, "D%d" % r, "source lines " + ", ".join(lines_for) if lines_for else "none", S_TXTW)
    wsR.row_dimensions[r].height = 30
    r += 1

r += 1
put(wsR, "A%d" % r, "ASSUMPTIONS AND JUDGEMENT CALLS", S_LABEL)
r += 1
for t in [
    "Slash dates are read month first, US convention, because that is what this export writes. A file",
    "from a system that writes day first would need this rule changed, and it is the first thing to check.",
    "Unit cost comes from the Apex product catalog, not from the export, so margin here is standard",
    "margin at the list cost, not landed cost. Shipping is carried separately and is not in margin.",
    "A blank unit price is filled from the catalog list price. A blank product, date, id or quantity is not",
    "guessed. Those rows are held back so a person decides.",
    "A repeated order id keeps the first row and holds back the later one, because in this export the",
    "later row was the resend, not the correction.",
]:
    put(wsR, "A%d" % r, t, S_MUTED)
    r += 1
put(wsR, "A%d" % (r + 1), "Apex Instruments is fictional. Every name, figure and contact is demo data.", S_LABEL)

# ------------------------------------------------------------- 3. Summary ---
wsS = wb.create_sheet("Summary", 1)
widths(wsS, {"A": 30, "B": 17, "C": 3, "D": 30, "E": 17, "F": 3, "G": 44})
wsS.sheet_view.showGridLines = False
wsS.sheet_properties.tabColor = ORANGE

put(wsS, "A1", "Quarter summary", S_H1)
put(wsS, "A2", "Apex Instruments, quarter ending September 30, 2026", S_H2)
put(wsS, "A3", "Every cell in column B and column E is a formula over the Clean data sheet.", S_MUTED)
wsS.row_dimensions[1].height = 27

KPI = [
    # (row, label, formula, style)
    (5,  "Orders imported",      "=COUNTA(%s)" % rng("A"), S_KPIN),
    (6,  "Units sold",           "=SUM(%s)" % rng("I"), S_KPIN),
    (7,  "Gross revenue",        "=SUM(%s)" % rng("L"), S_KPIV),
    (8,  "Discount given",       "=B7-B9", S_KPIV),
    (9,  "Net revenue",          "=SUM(%s)" % rng("M"), S_KPIV),
    (10, "Cost of goods",        "=SUM(%s)" % rng("O"), S_KPIV),
    (11, "Gross margin",         "=B9-B10", S_KPIV),
    (12, "Gross margin percent", "=IFERROR(B11/B9,0)", S_KPIP),
]
KPI2 = [
    (5,  "Average order value",  "=IFERROR(B9/B5,0)", S_KPIV),
    (6,  "Average discount",     "=IFERROR(B8/B7,0)", S_KPIP),
    (7,  "Partner net revenue",  '=SUMIF(%s,"Partner",%s)' % (rng("R"), rng("M")), S_KPIV),
    (8,  "Direct net revenue",   '=SUMIF(%s,"Direct",%s)' % (rng("R"), rng("M")), S_KPIV),
    (9,  "Shipping billed",      "=SUM(%s)" % rng("Q"), S_KPIV),
    (10, "Largest single line",  "=MAX(%s)" % rng("M"), S_KPIV),
    (11, "Rows held back",       "=COUNTA('Rejected rows'!$A$5:$A$%d)" % (rows_rejected + 4), S_KPIN),
    (12, "Rows read from source", "=B5+E11", S_KPIN),
]
for row, label, formula, style in KPI:
    put(wsS, "A%d" % row, label, S_KPIL)
    put(wsS, "B%d" % row, formula, style)
    wsS.row_dimensions[row].height = 22
for row, label, formula, style in KPI2:
    put(wsS, "D%d" % row, label, S_KPIL)
    put(wsS, "E%d" % row, formula, style)

put(wsS, "A14", "BEST AND BIGGEST", S_LABEL)
put(wsS, "A15", "Strongest month by net revenue", S_KPIL)
put(wsS, "B15", "=INDEX('By month'!$A$5:$A$7,MATCH(MAX('By month'!$F$5:$F$7),'By month'!$F$5:$F$7,0))", S_BODY)
put(wsS, "A16", "Largest product line by net revenue", S_KPIL)
put(wsS, "B16", "=INDEX('By product'!$A$5:$A$10,MATCH(MAX('By product'!$F$5:$F$10),'By product'!$F$5:$F$10,0))", S_BODY)
put(wsS, "A17", "Product lines under 40 percent margin", S_KPIL)
put(wsS, "B17", "=COUNTIF('By product'!$I$5:$I$10,\"<0.4\")", S_KPIN)

put(wsS, "A19", "RECONCILIATION", S_LABEL)
put(wsS, "A20", "Net revenue on this sheet, less the By month total. Zero means the sheets agree.", S_KPIL)
put(wsS, "B20", "=ROUND(B9-'By month'!$F$8,2)", S_MONEY)
put(wsS, "A21", "Net revenue on this sheet, less the By product total. Zero means the sheets agree.", S_KPIL)
put(wsS, "B21", "=ROUND(B9-'By product'!$F$11,2)", S_MONEY)
put(wsS, "A22", "Rows read, less rows imported and rows held back. Zero means no row was lost.", S_KPIL)
put(wsS, "B22", "=E12-B5-E11", S_INT)

put(wsS, "A24", "Built %s. Apex Instruments is fictional and every figure here is demo data." % BUILT_DAY, S_MUTED)

# ------------------------------------------------------------ 4. By month ---
wsM = wb.create_sheet("By month", 2)
widths(wsM, {"A": 18, "B": 10, "C": 10, "D": 14, "E": 14, "F": 15, "G": 14, "H": 14, "I": 10})
wsM.sheet_view.showGridLines = False
wsM.sheet_properties.tabColor = STEEL
put(wsM, "A1", "By month", S_H1)
put(wsM, "A2", "Counted from Clean data with COUNTIF and SUMIF over the month column.", S_MUTED)
wsM.row_dimensions[1].height = 27

MHEAD = ["Month", "Orders", "Units", "Gross", "Discount", "Net revenue", "Cost", "Gross margin", "Margin"]
for i, t in enumerate(MHEAD):
    put(wsM, "%s4" % get_column_letter(i + 1), t, S_TH if i == 0 else S_THR)
wsM.row_dimensions[4].height = 20

for i, m in enumerate(by_month):
    row = 5 + i
    put(wsM, "A%d" % row, m["label"], S_TXT)
    put(wsM, "B%d" % row, '=COUNTIF(%s,"%s")' % (rng("C"), m["key"]), S_INT)
    put(wsM, "C%d" % row, '=SUMIF(%s,"%s",%s)' % (rng("C"), m["key"], rng("I")), S_INT)
    put(wsM, "D%d" % row, '=SUMIF(%s,"%s",%s)' % (rng("C"), m["key"], rng("L")), S_MONEY0)
    put(wsM, "E%d" % row, "=D%d-F%d" % (row, row), S_MONEY0)
    put(wsM, "F%d" % row, '=SUMIF(%s,"%s",%s)' % (rng("C"), m["key"], rng("M")), S_MONEY0)
    put(wsM, "G%d" % row, '=SUMIF(%s,"%s",%s)' % (rng("C"), m["key"], rng("O")), S_MONEY0)
    put(wsM, "H%d" % row, "=F%d-G%d" % (row, row), S_MONEY0)
    put(wsM, "I%d" % row, "=IFERROR(H%d/F%d,0)" % (row, row), S_PCT)

put(wsM, "A8", "Quarter", S_TOTL)
for col in "BC":
    put(wsM, "%s8" % col, "=SUM(%s5:%s7)" % (col, col), S_TOTI)
for col in "DEFGH":
    put(wsM, "%s8" % col, "=SUM(%s5:%s7)" % (col, col), S_TOTM)
put(wsM, "I8", "=IFERROR(H8/F8,0)", S_TOTP)

chM = LineChart()
chM.title = "Net revenue and gross margin by month"
chM.height, chM.width = 8.4, 17
chM.y_axis.title = "Dollars"
chM.y_axis.numFmt = '"$"#,##0'
chM.x_axis.title = None
dataM = Reference(wsM, min_col=6, max_col=6, min_row=4, max_row=7)
dataM2 = Reference(wsM, min_col=8, max_col=8, min_row=4, max_row=7)
chM.add_data(dataM, titles_from_data=True)
chM.add_data(dataM2, titles_from_data=True)
chM.set_categories(Reference(wsM, min_col=1, min_row=5, max_row=7))
for srs, colr in zip(chM.series, (STEEL, ORANGE)):
    srs.graphicalProperties.line.solidFill = colr
    srs.graphicalProperties.line.width = 28000
    srs.smooth = False
    srs.marker = Marker(symbol="circle", size=7)
wsM.add_chart(chM, "A11")
put(wsM, "A29", "A quarter has three points. The shape, not the slope, is the thing to read.", S_MUTED)

# ---------------------------------------------------------- 5. By product ---
wsP = wb.create_sheet("By product", 3)
widths(wsP, {"A": 22, "B": 10, "C": 10, "D": 14, "E": 14, "F": 15, "G": 14, "H": 14, "I": 10, "J": 11})
wsP.sheet_view.showGridLines = False
wsP.sheet_properties.tabColor = STEEL
put(wsP, "A1", "By product", S_H1)
put(wsP, "A2", "Margin is net revenue less standard cost. The margin column is colour scaled, and anything under 40 percent is flagged red.", S_MUTED)
wsP.row_dimensions[1].height = 27

PHEAD = ["Product", "Orders", "Units", "Gross", "Discount", "Net revenue", "Cost", "Gross margin", "Margin", "Share of net"]
for i, t in enumerate(PHEAD):
    put(wsP, "%s4" % get_column_letter(i + 1), t, S_TH if i == 0 else S_THR)
wsP.row_dimensions[4].height = 20

for i, p in enumerate(by_product):
    row = 5 + i
    q = '"%s"' % p["product"]
    put(wsP, "A%d" % row, p["product"], S_TXT)
    put(wsP, "B%d" % row, "=COUNTIF(%s,%s)" % (rng("H"), q), S_INT)
    put(wsP, "C%d" % row, "=SUMIF(%s,%s,%s)" % (rng("H"), q, rng("I")), S_INT)
    put(wsP, "D%d" % row, "=SUMIF(%s,%s,%s)" % (rng("H"), q, rng("L")), S_MONEY0)
    put(wsP, "E%d" % row, "=D%d-F%d" % (row, row), S_MONEY0)
    put(wsP, "F%d" % row, "=SUMIF(%s,%s,%s)" % (rng("H"), q, rng("M")), S_MONEY0)
    put(wsP, "G%d" % row, "=SUMIF(%s,%s,%s)" % (rng("H"), q, rng("O")), S_MONEY0)
    put(wsP, "H%d" % row, "=F%d-G%d" % (row, row), S_MONEY0)
    put(wsP, "I%d" % row, "=IFERROR(H%d/F%d,0)" % (row, row), S_PCT)
    put(wsP, "J%d" % row, "=IFERROR(F%d/$F$11,0)" % row, S_PCT)

put(wsP, "A11", "All products", S_TOTL)
for col in "BC":
    put(wsP, "%s11" % col, "=SUM(%s5:%s10)" % (col, col), S_TOTI)
for col in "DEFGH":
    put(wsP, "%s11" % col, "=SUM(%s5:%s10)" % (col, col), S_TOTM)
put(wsP, "I11", "=IFERROR(H11/F11,0)", S_TOTP)
put(wsP, "J11", "=IFERROR(F11/$F$11,0)", S_TOTP)

wsP.conditional_formatting.add("I5:I10", ColorScaleRule(
    start_type="num", start_value=0.30, start_color="F8C9B4",
    mid_type="num", mid_value=0.45, mid_color="FFFFFF",
    end_type="num", end_value=0.60, end_color="C9E3E1"))
wsP.conditional_formatting.add("I5:I10", CellIsRule(
    operator="lessThan", formula=["0.4"],
    font=Font(name="Calibri", size=11, bold=True, color="B03A00")))

chP = BarChart()
chP.type = "col"
chP.title = "Net revenue by product"
chP.height, chP.width = 8.6, 17
chP.y_axis.title = "Net revenue"
chP.y_axis.numFmt = '"$"#,##0'
chP.legend = None
chP.add_data(Reference(wsP, min_col=6, max_col=6, min_row=4, max_row=10), titles_from_data=True)
chP.set_categories(Reference(wsP, min_col=1, min_row=5, max_row=10))
chP.series[0].graphicalProperties.solidFill = STEEL
chP.series[0].graphicalProperties.line.solidFill = STEEL
chP.gapWidth = 55
wsP.add_chart(chP, "A14")
put(wsP, "A32", "Share of net adds to 100 percent. Standard cost comes from the catalog, not from the export.", S_MUTED)

# -------------------------------------------------------- 6. Rejected rows --
wsX = wb.create_sheet("Rejected rows", 5)
widths(wsX, {"A": 12, "B": 8, "C": 13, "D": 18, "E": 12, "F": 22, "G": 9, "H": 16, "I": 46, "J": 30})
wsX.sheet_view.showGridLines = False
wsX.sheet_properties.tabColor = ORANGE
put(wsX, "A1", "Rows held back", S_H1)
put(wsX, "A2", "%d of the %d rows in the source file were not imported. None were deleted. Every one is listed here with the line it came from, so it can be fixed at the source and re-run." % (rows_rejected, rows_in), S_MUTED)
wsX.row_dimensions[1].height = 27

XHEAD = ["Source line", "Code", "Order id", "Order date", "Partner id", "Product",
         "Qty", "Unit price", "Why it was held back", "What was read"]
for i, t in enumerate(XHEAD):
    put(wsX, "%s4" % get_column_letter(i + 1), t, S_THR if i in (0, 6) else S_TH)
wsX.row_dimensions[4].height = 20

for i, x in enumerate(sorted(reject_rows, key=lambda r: r["line"])):
    row = 5 + i
    put(wsX, "A%d" % row, x["line"], S_INT)
    put(wsX, "B%d" % row, x["code"], S_MONO)
    put(wsX, "C%d" % row, x["order_id"] or "(empty)", S_MONO)
    put(wsX, "D%d" % row, x["order_date"] or "(empty)", S_TXT)
    put(wsX, "E%d" % row, x["partner_id"] or "(empty)", S_MONO)
    put(wsX, "F%d" % row, x["product"] or "(empty)", S_TXT)
    put(wsX, "G%d" % row, x["qty"] or "(empty)", S_TXT)
    put(wsX, "H%d" % row, x["unit_price"] or "(empty)", S_TXT)
    put(wsX, "I%d" % row, x["reason"], S_TXTW)
    put(wsX, "J%d" % row, x["detail"], S_TXTW)
    wsX.row_dimensions[row].height = 28
wsX.freeze_panes = "A5"
wsX.auto_filter.ref = "A4:J%d" % (rows_rejected + 4)

wb.properties.title = "Apex Instruments Q3 2026 order book"
wb.properties.creator = "Company OS spreadsheet system"
wb.properties.description = "Demo data. Apex Instruments is fictional."
wb.active = 0
wb.save(XLSX)

# ============================================================= the reports ===
def dist(counter, top=None):
    items = sorted(counter.items(), key=lambda kv: (-kv[1], kv[0]))
    return [[k, v] for k, v in (items[:top] if top else items)]


clean_city = Counter(r["ship_city"] for r in clean)
clean_product = Counter(r["product"] for r in clean)

report = OrderedDict()
report["run"] = {
    "built": BUILT,
    "source": "data/raw-orders.csv",
    "lookup": "data/partners.csv",
    "workbook": "out/apex-orders-2026-q3.xlsx",
    "engine": "python %d.%d, openpyxl %s" % (sys.version_info[0], sys.version_info[1],
                                             __import__("openpyxl").__version__),
    "subject": "Apex Instruments, quarter ending September 30, 2026",
    "workbookBytes": os.path.getsize(XLSX),
    "workbookSheets": ["Read me", "Summary", "By month", "By product", "Clean data", "Rejected rows"],
}
report["counts"] = {
    "linesInFile": len(RAW_LINES),
    "preambleLines": header_idx,
    "headerLine": header_idx + 1,
    "rowsIn": rows_in,
    "rowsOut": rows_out,
    "rowsFixed": rows_fixed,
    "rowsUnchanged": rows_out - rows_fixed,
    "rowsRejected": rows_rejected,
    "cellsChanged": cells_changed,
}
report["rules"] = [
    {"id": rid, "name": name, "what": what, "unit": unit,
     "fired": rule_hits.get(rid, 0), "example": rule_example.get(rid)}
    for rid, name, what, unit in RULE_DEFS
]
report["rejects"] = {
    "total": rows_rejected,
    "byReason": [
        {"code": code, "reason": reason, "rows": reject_counts.get(code, 0),
         "lines": [x["line"] for x in reject_rows if x["code"] == code]}
        for code, reason in REJECT_DEFS
    ],
    "rows": [{"line": x["line"], "code": x["code"], "reason": x["reason"],
              "detail": x["detail"], "orderId": x["order_id"], "orderDate": x["order_date"],
              "partnerId": x["partner_id"], "product": x["product"],
              "qty": x["qty"], "unitPrice": x["unit_price"]}
             for x in sorted(reject_rows, key=lambda r: r["line"])],
}
report["distributions"] = {
    "ship_city": {
        "note": "One location was reaching the file under five spellings. After the run there is one.",
        "before": dist(raw_city_dist), "after": dist(clean_city),
        "distinctBefore": len(raw_city_dist), "distinctAfter": len(clean_city),
    },
    "product": {
        "note": "Six catalog lines were arriving under many spellings, including short forms and case variants.",
        "before": dist(raw_product_dist), "after": dist(clean_product),
        "distinctBefore": len(raw_product_dist), "distinctAfter": len(clean_product),
    },
}
report["totals"] = totals
report["byMonth"] = by_month
report["byProduct"] = by_product

with open(os.path.join(DATA, "clean-report.json"), "w", encoding="utf-8") as f:
    json.dump(report, f, indent=2)
    f.write("\n")

with open(os.path.join(DATA, "preview.json"), "w", encoding="utf-8") as f:
    json.dump({
        "note": "Row by row before and after for the widget. Apex Instruments is fictional.",
        "columns": ["Order id", "Order date", "Partner id", "Ship to city", "Product",
                    "Qty", "Unit price", "Discount", "Channel", "Status"],
        "cleanColumns": ["Order id", "Order date", "Partner", "Region", "Ship to city",
                         "Product", "Qty", "Unit price", "Discount", "Net revenue"],
        "headerLine": header_idx + 1,
        "preamble": [",".join(r) for r in RAW_LINES[:header_idx]],
        "rows": preview,
    }, f, separators=(",", ":"))
    f.write("\n")

# ------------------------------------------------------------- stdout -------
print("Apex Instruments Q3 2026 ingestion")
print("  source          data/raw-orders.csv, %d lines, header on line %d" % (len(RAW_LINES), header_idx + 1))
print("  rows in         %d" % rows_in)
print("  rows imported   %d  (%d changed by a rule, %d already clean)" % (rows_out, rows_fixed, rows_out - rows_fixed))
print("  rows held back  %d" % rows_rejected)
print("  cells changed   %d" % cells_changed)
print("  rules fired")
for rid, name, what, unit in RULE_DEFS:
    print("    %-4s %-44s %5d %s" % (rid, name, rule_hits.get(rid, 0), unit))
print("  held back by reason")
for code, reason in REJECT_DEFS:
    print("    %-3s %3d   %s" % (code, reject_counts.get(code, 0), reason[:64]))
print("  net revenue     $%s" % "{:,.2f}".format(totals["net"]))
print("  gross margin    $%s  (%.1f%%)" % ("{:,.2f}".format(totals["margin"]), totals["marginPct"] * 100))
print("  wrote           %s" % os.path.relpath(XLSX, HERE))
print("                  data/clean-report.json, data/preview.json")