"""PlusROI books: the Google-Sheet → DB transform step (R2, ADR #016).

rian runs PlusROI (a 50/50 revenue-share partnership with Rob Cooper) on a
Google Sheet: an **Invoices** ledger (receivables), an **Expenses** ledger
(payables), and a monthly settlement. `with` is replacing that sheet. This
module imports the sheet's full CSV export
(``books-import/source/tab-*.csv`` from the LIVE sheet + ``arch2022-*.csv``
from the 2022 archive) into the R2 schema (alembic 0007).

It runs as one more step in the existing rebuild pipeline
(``transforms.run_all``), AFTER ``step_comments`` and INSIDE the same
transaction, so it inherits that pipeline's discipline:

  * clear-and-rebuild idempotency (its tables are in
    ``transforms.TRUNCATE_TABLES``): a second run reproduces identical
    counts.
  * per-step logging (``report.step``), per-row disposition accounting,
    and reconciliation (``report.reconcile``) that FAILS the run — rolling
    back the WHOLE rebuild — on any unaccounted CSV row.

This only applies while the PlusROI tenant is SHEET-owned
(``settings.books_owner == "sheet"`` — the parallel-run era). Once the
tenant is flipped to app-owned (ADR #021, ``scripts/flip_books_owner.py``)
the step SKIPS it entirely: the tenant-scoped reset in ``transforms``
preserved its rows, and the app is the system of record.

It is NOT the settlement engine (that lives in ``services/money``): it
lands the ledgers + the per-settlement-month totals the settlement gate
verifies against the sheet's own Monthly Split.

Requirements doc: ``/srv/apps/work/docs/rebuild/08-current-tooling.md``
(§1 structure, §2 Invoices, §3 Expenses, §4 settlement, §9). Party
creation goes through ``transforms._insert_party`` (the raw-SQL choke
point that maintains the 1:1 subtype rows — ADR #012/#013).
"""

from __future__ import annotations

import csv
import json
import logging
import re
import time
import uuid
from datetime import date, timedelta
from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
from pathlib import Path

from sqlalchemy import text
from sqlalchemy.engine import Connection

from app.services.migrate.transforms import (
    CURRENCY,
    Ctx,
    MigrationError,
    _insert_engagement,
    _insert_party,
)

log = logging.getLogger("with.migrate.books")

# The PlusROI tenant settings (08 §4): 3% bookkeeping (Danielle), 50/50
# partner split (rian ↔ Rob). books_owner="sheet" (ADR #021, superseding
# the boolean parallel_run) marks this tenant's books as
# rebuilt-from-the-sheet: this step reimports them each run AND the app's
# invoice/expense writes are refused. The FIRST import always runs
# sheet-owned; scripts/flip_books_owner.py flips the tenant to "app"
# (nightly leaves it alone, writes unlock). The legacy-parity harness
# excludes books tenants (extract_new scopes to legacy-mirrored tenants).
PLUSROI_TENANT_SETTINGS = {
    "default_currency": "CAD",
    "bookkeeping_pct": 3.0,
    "books_owner": "sheet",
    "partner_split": 0.5,
}

# 08-doc-identified humans who appear as vendors but are NOT legacy team
# members (so no person party pre-exists) — create them as people, not orgs.
PINNED_PERSON_VENDORS = {
    "Brennan Doyle",
    "Taylor Harris",
    "Alim Rawji",
    "Alden Loutit",
    "Kayla Cooper",
    "Emma Slone",
}

# Both spellings collapse to the existing Bowden Works org party (the Adi
# pass-through — 08 §3).
BOWDEN_WORKS_VENDOR_KEYS = {"bowden works team", "bowden works"}

# Sheet invoice lifecycle → invoices.status (08 §2; the CHECK widened in
# alembic 0007). Blank / unknown defaults to 'paid' (98.5% of non-blank
# statuses are Paid; these are historical closed-book rows) with a warn.
STATUS_MAP = {
    "pending approval": "pending_approval",
    "ready to invoice": "ready",
    "invoiced": "sent",
    "paid": "paid",
    "cancelled": "cancelled",
    "delayed payment": "delayed",
}

# Excel's 1900 date system (with the leap-year bug) — serial 0 is
# 1899-12-30. Only the invoice "Due Date" column carries serials.
EXCEL_EPOCH = date(1899, 12, 30)


# --- CSV reading -----------------------------------------------------------


class _Tab:
    """A sheet tab read positionally with a first-occurrence name→index
    map. Positional access handles the LIVE Projects tab's blank column
    headers (idx0 = "{Client} : {Title}", idx1 = Client); name access
    handles every labelled column and the LIVE/ARCH header differences
    (e.g. Expenses columns are in a different ORDER between the two)."""

    def __init__(self, path: Path):
        with open(path, newline="", encoding="utf-8-sig") as fh:
            reader = csv.reader(fh)
            self.header = next(reader, [])
            self.rows = list(reader)
        self._idx: dict[str, int] = {}
        for i, name in enumerate(self.header):
            key = (name or "").strip().lower()
            if key and key not in self._idx:
                self._idx[key] = i

    def _col(self, *names: str) -> int | None:
        for n in names:
            i = self._idx.get(n.strip().lower())
            if i is not None:
                return i
        return None

    def get(self, row: list[str], *names: str, index: int | None = None) -> str:
        i = index if index is not None else self._col(*names)
        if i is None or i >= len(row):
            return ""
        return (row[i] or "").strip()


# --- value parsers ---------------------------------------------------------


def _money(s: str) -> Decimal | None:
    """Parse a sheet money string. Handles ``$1,234.56`` and parenthesised
    negatives ``(123.45)``; blank / non-numeric (e.g. "no limit") → None."""
    s = (s or "").strip()
    if not s:
        return None
    neg = s.startswith("(") and s.endswith(")")
    body = s.strip("()").replace("$", "").replace(",", "").strip()
    if not body:
        return None
    try:
        d = Decimal(body)
    except InvalidOperation:
        return None
    return -d if neg else d


def _money2(s: str) -> tuple[Decimal | None, bool]:
    """(rounded-to-2dp value, over_2dp?) — HALF_UP, and flag when the
    source carried >2 decimals (08 §9: it does, for FX-converted rows)."""
    d = _money(s)
    if d is None:
        return None, False
    over = (-d.as_tuple().exponent) > 2
    return d.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), over


def _pct(s: str) -> Decimal | None:
    """Commission percentage as stored in the sheet (a fraction: 0.2 =
    20%), quantised to Numeric(7,4)."""
    d = _money(s)
    if d is None:
        return None
    return d.quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP)


def _date(s: str) -> date | None:
    """Parse a sheet date: ISO ``YYYY-MM-DD[ HH:MM:SS]`` OR an Excel serial
    (only the "Due Date" column uses serials, all ≥ 20000)."""
    s = (s or "").strip()
    if not s:
        return None
    m = re.match(r"^(\d{4})-(\d{2})-(\d{2})", s)
    if m:
        try:
            return date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
        except ValueError:
            return None
    try:
        f = float(s)
    except ValueError:
        return None
    if f >= 20000:  # a plausible Excel serial (year ≥ 1954), not a year int
        return EXCEL_EPOCH + timedelta(days=int(f))
    return None


def _first_of_month(s: str) -> date | None:
    d = _date(s)
    return d.replace(day=1) if d else None


def _cur(s: str) -> str | None:
    s = (s or "").strip().upper()
    return s[:3] if s else None


def _norm_company(s: str) -> str:
    """Normalise a "{Client} : {Project}" key for project matching."""
    return re.sub(r"\s+", " ", (s or "").strip()).lower()


def _ext_id(s: str) -> str:
    """Normalise a sheet invoice id. Numeric CAD ids arrive as float
    strings from the xlsx export ("6357.0") — render integral floats as
    ints so the ledger shows "6357"; US-prefixed / non-numeric ids pass
    through untouched. Applied identically to live + archive, so the
    id-based dedupe is unaffected."""
    s = (s or "").strip()
    if not s:
        return s
    try:
        f = float(s)
    except ValueError:
        return s
    return str(int(f)) if f == int(f) else s


def _resolve_source(books_source: str | Path | None) -> Path | None:
    if books_source is None:
        return None
    if books_source == "auto":
        # app-relative: app/services/migrate/sheet_books.py → project root
        return Path(__file__).resolve().parents[3] / "books-import" / "source"
    return Path(books_source)


# --- the step --------------------------------------------------------------


#: Every tab the step consumes. If ANY is absent while a books_source was
#: requested, the run FAILS (rolls back) — see the fail-fast block below.
_REQUIRED_TABS = (
    "tab-Invoices.csv",
    "tab-Expenses.csv",
    "tab-Clients.csv",
    "tab-Projects.csv",
    "tab-Vendors.csv",
    "arch2022-Invoices.csv",
    "arch2022-Expenses.csv",
)

#: A fresh export can only shrink so much vs the last good run (rows are
#: appended, rarely deleted). A bigger shrink = truncated/corrupt file.
_SHRINK_FLOOR = 0.8


def _csv_row_count(path: Path) -> int:
    with path.open(encoding="utf-8") as fh:
        return max(sum(1 for _ in csv.reader(fh)) - 1, 0)


def _guard_against_shrunken_export(conn: Connection, source: Path) -> None:
    """Adversarial-review fix (2026-07-11): a partially-written export
    imports partial books that still reconcile green. Compare the two
    money-bearing tabs' row counts against the last OK run's report and
    fail the run on a >20% shrink."""
    prior = conn.execute(
        text(
            "SELECT report->'reconciliation' FROM migration_runs"
            " WHERE ok ORDER BY started_at DESC LIMIT 1"
        )
    ).scalar()
    if not prior:
        return
    for tab in ("tab-Invoices.csv", "tab-Expenses.csv"):
        key = tab[:-4]
        entry = next(
            (v for k, v in prior.items() if k.endswith(key)), None
        )
        if not entry:
            continue
        previous = int(entry.get("source") or 0)
        current = _csv_row_count(source / tab)
        if previous and current < previous * _SHRINK_FLOOR:
            raise MigrationError(
                f"sheet export sanity: {tab} has {current} data rows but the"
                f" last good run saw {previous} — a >20% shrink means a"
                " truncated/corrupt export; refusing to import it"
            )


def step_sheet_books(conn: Connection, ctx: Ctx) -> None:  # noqa: C901
    step = ctx.report.step("2.10 PlusROI sheet books")
    source = _resolve_source(ctx.books_source)
    if ctx.books_source is None:
        # Legacy-only mode (harness/reporting tests): the step is OFF by
        # design and the run never promises PlusROI books.
        step.details["skipped"] = True
        return

    # App-owned skip (ADR #021): once the PlusROI tenant's books are
    # app-owned (settings.books_owner == "app"), the app is the system of
    # record — this step must neither clear nor reimport them. The
    # tenant-scoped reset already preserved every row; checked BEFORE the
    # missing-source fatal so a stale/absent export can never fail (or
    # touch) an app-owned tenant's nightly.
    owned = conn.execute(
        text(
            "SELECT t.id, t.settings FROM tenants t"
            " JOIN parties p ON p.id = t.party_id WHERE lower(p.name) = 'plusroi'"
        )
    ).fetchone()
    if owned is not None:
        from app.services.tenancy import BOOKS_OWNER_APP, effective_books_owner

        if effective_books_owner(owned[1]) == BOOKS_OWNER_APP:
            step.details["skipped_app_owned"] = str(owned[0])
            log.info(
                "PlusROI tenant %s is app-owned (books_owner=app) — sheet"
                " step skipped, books preserved",
                owned[0],
            )
            return

    # A books_source WAS requested (production passes "auto"), so the
    # PlusROI tables are already truncated — committing a skip here would
    # DELETE the tenant and strand the partner logins (adversarial-review
    # CRITICAL, 2026-07-11). A missing/incomplete export is therefore
    # FATAL: MigrationError rolls the whole transaction back and
    # yesterday's books survive untouched.
    if source is None or not source.is_dir():
        raise MigrationError(
            f"PlusROI books source not found ({ctx.books_source!r} → {source})"
            " — failing the run so yesterday's books survive (a committed"
            " skip would delete the PlusROI tenant)"
        )
    missing = [tab for tab in _REQUIRED_TABS if not (source / tab).exists()]
    if missing:
        raise MigrationError(
            f"PlusROI books source is incomplete — missing {missing};"
            " failing the run so yesterday's books survive"
        )
    _guard_against_shrunken_export(conn, source)

    newest_mtime = max((source / tab).stat().st_mtime for tab in _REQUIRED_TABS)
    age_days = (time.time() - newest_mtime) / 86400
    step.details["source_age_days"] = round(age_days, 1)
    if age_days > 9:
        step.warn(
            f"sheet export is {age_days:.0f} days old — books are drifting"
            " behind the sheet (re-export needed; weekly cadence assumed)"
        )

    # --- load the migrated party/name maps -------------------------------
    ctx.slugs.update(
        r[0] for r in conn.execute(text("SELECT slug FROM parties")).fetchall()
    )
    org_by_name: dict[str, uuid.UUID] = {}
    person_by_name: dict[str, uuid.UUID] = {}
    party_by_email: dict[str, uuid.UUID] = {}
    for pid, kind, name, email in conn.execute(
        text("SELECT id, kind, name, email FROM parties")
    ).fetchall():
        key = (name or "").strip().lower()
        if kind == "org":
            org_by_name.setdefault(key, pid)
        else:
            person_by_name.setdefault(key, pid)
        if email:
            party_by_email.setdefault(email.strip().lower(), pid)

    warns: dict[str, int] = {}

    def warn(cls: str, n: int = 1) -> None:
        warns[cls] = warns.get(cls, 0) + n

    def resolve_existing(name: str) -> uuid.UUID | None:
        key = (name or "").strip().lower()
        return org_by_name.get(key) or person_by_name.get(key) if key else None

    # === 1. tenant ======================================================
    pr_party = org_by_name.get("plusroi")
    if pr_party is None:
        pr_party = _insert_party(conn, ctx, kind="org", name="PlusROI")
        org_by_name["plusroi"] = pr_party
    row = conn.execute(
        text("SELECT id FROM tenants WHERE party_id = :p"), {"p": pr_party}
    ).fetchone()
    if row is not None:
        pr_tenant = row[0]
    else:
        pr_tenant = uuid.uuid4()
        conn.execute(
            text(
                "INSERT INTO tenants (id, party_id, settings)"
                " VALUES (:id, :p, cast(:s AS jsonb))"
            ),
            {"id": pr_tenant, "p": pr_party, "s": json.dumps(PLUSROI_TENANT_SETTINGS)},
        )

    # === 2. owners (people + official_partnership engagements + users) ===
    def person_get_or_create(name: str, email: str | None) -> uuid.UUID:
        if email:
            pid = party_by_email.get(email.lower())
            if pid is not None:
                return pid
        pid = person_by_name.get(name.strip().lower())
        if pid is not None:
            return pid
        pid = _insert_party(conn, ctx, kind="person", name=name, email=email)
        person_by_name[name.strip().lower()] = pid
        if email:
            party_by_email[email.lower()] = pid
        return pid

    rian_p = person_get_or_create("Rian Bowden", "rian@rian.ca")  # reuse by email
    rob_p = person_get_or_create("Rob Cooper", "rob@plusroi.com")  # id-auth email
    dani_p = person_get_or_create("Danielle Cooper", "danielle@plusroi.com")
    for person in (rian_p, rob_p, dani_p):
        _insert_engagement(
            conn,
            ctx,
            tenant_id=pr_tenant,
            type_="official_partnership",
            party_a=pr_party,
            party_b=person,
            role_a="owner",
            role_b="owner",
        )
    # link Rob/Danielle's users rows BY USERNAME (they may not have logged
    # in yet → 0-row updates are silent; re-runs re-link the fresh party).
    # rian keeps his existing BW person party — NOT relinked.
    users_linked = {}
    for username, pid in (("rob", rob_p), ("danielle", dani_p)):
        res = conn.execute(
            text(
                "UPDATE users SET person_party_id = :p"
                " WHERE idauth_username = :u AND person_party_id IS NULL"
            ),
            {"p": pid, "u": username},
        )
        users_linked[username] = res.rowcount

    # === 3. Clients tab → parties + client_profiles + direct_client ======
    client_by_key: dict[str, uuid.UUID] = {}  # lower(short name) → party
    profiled: set[uuid.UUID] = set()
    tab = _Tab(source / "tab-Clients.csv")
    disp: dict[str, int] = {}

    def bump(d: dict[str, int], k: str) -> None:
        d[k] = d.get(k, 0) + 1

    for r in tab.rows:
        short = tab.get(r, "Short Name")
        if not short:
            bump(disp, "skipped_blank")
            continue
        key = short.lower()
        official = tab.get(r, "Official Name")
        pid = org_by_name.get(key)
        if pid is not None:
            bump(disp, "client_matched")
        else:
            notes = official if official and official.lower() != key else None
            pid = _insert_party(conn, ctx, kind="org", name=short, notes=notes)
            org_by_name[key] = pid
            bump(disp, "client_created")
        client_by_key[key] = pid
        if pid in profiled:
            continue
        profiled.add(pid)
        comm_name = tab.get(r, "Commission")
        comm_party = resolve_existing(comm_name)
        if comm_name and comm_party is None:
            warn("client_commission_unresolved")
        contacts = " | ".join(
            v
            for v in (
                tab.get(r, "Contact Name"),
                tab.get(r, "Contact Emails"),
                tab.get(r, "Phone"),
            )
            if v
        ) or None
        conn.execute(
            text(
                "INSERT INTO client_profiles (tenant_id, client_party_id, currency,"
                " terms, invoice_timing, default_commission_pct, commission_party_id,"
                " credit_limit, contacts, archived, notes)"
                " VALUES (:t, :c, :cur, :terms, :timing, :pct, :cparty, :limit,"
                " :contacts, :archived, :notes)"
                " ON CONFLICT (tenant_id, client_party_id) DO NOTHING"
            ),
            {
                "t": pr_tenant,
                "c": pid,
                "cur": _cur(tab.get(r, "Currency")),
                "terms": tab.get(r, "Terms") or None,
                "timing": tab.get(r, "Invoice Timing") or None,
                "pct": _pct(tab.get(r, "Com.%")),
                "cparty": comm_party,
                "limit": (_money2(tab.get(r, "Credit Limit"))[0]),
                "contacts": contacts,
                "archived": tab.get(r, "⛔").lower() == "true",
                "notes": (official if official and official.lower() != key else None)
                or (tab.get(r, "Address") or None),
            },
        )
        if pid != pr_party:
            _insert_engagement(
                conn,
                ctx,
                tenant_id=pr_tenant,
                type_="direct_client",
                party_a=pr_party,
                party_b=pid,
                role_a="operator",
                role_b="client",
            )
    ctx.report.reconcile("sheet.tab-Clients", len(tab.rows), disp)

    # === 4. Vendors tab → parties + in-memory defaults ===================
    vendor_by_name: dict[str, dict] = {}
    tab = _Tab(source / "tab-Vendors.csv")
    disp = {}
    for r in tab.rows:
        name = tab.get(r, "Name")
        if not name:
            bump(disp, "skipped_blank")
            continue
        key = name.lower()
        if key in BOWDEN_WORKS_VENDOR_KEYS:
            pid = org_by_name.get("bowden works")
            if pid is None:
                pid = _insert_party(conn, ctx, kind="org", name="Bowden Works")
                org_by_name["bowden works"] = pid
            bump(disp, "mapped_bowden_works")
        elif key in person_by_name:
            pid = person_by_name[key]
            bump(disp, "matched_person")
        elif key in org_by_name:
            pid = org_by_name[key]
            bump(disp, "matched_org")
        elif name in PINNED_PERSON_VENDORS:
            pid = _insert_party(conn, ctx, kind="person", name=name)
            person_by_name[key] = pid
            bump(disp, "created_person")
        else:
            pid = _insert_party(conn, ctx, kind="org", name=name)
            org_by_name[key] = pid
            bump(disp, "created_org")
        vendor_by_name[key] = {
            "party": pid,
            "category": tab.get(r, "Default Expense Category") or None,
            "paid_by": tab.get(r, "Default Paid By") or None,
        }
    ctx.report.reconcile("sheet.tab-Vendors", len(tab.rows), disp)

    # === 5. Projects tab → projects + project_billing ====================
    project_by_company: dict[str, uuid.UUID] = {}
    seen_project: set[tuple[uuid.UUID, str]] = set()
    tab = _Tab(source / "tab-Projects.csv")
    disp = {}
    for r in tab.rows:
        title = tab.get(r, index=2)  # "Title" — idx2 in both LIVE and ARCH
        if not title:
            bump(disp, "skipped_blank")
            continue
        client_name = tab.get(r, index=1)  # Client (blank header on LIVE)
        key0 = tab.get(r, index=0)  # "{Client} : {Title}"
        ckey = client_name.lower()
        cpid = client_by_key.get(ckey) or org_by_name.get(ckey)
        on_fly = False
        if cpid is None:
            cpid = _insert_party(
                conn, ctx, kind="org", name=client_name or "(unknown client)"
            )
            org_by_name[ckey] = cpid
            client_by_key[ckey] = cpid
            on_fly = True
            warn("project_client_on_fly")
        status, prefix = _project_status(tab.get(r, "Status"))
        if status is None:
            status, prefix = "active", ""
            warn("project_status_unknown")
        name, base, n = title, title, 1
        while (cpid, name.lower()) in seen_project:
            n += 1
            name = f"{base} ({n})"
        if n > 1:
            warn("project_name_suffixed")
        seen_project.add((cpid, name.lower()))
        pid = uuid.uuid4()
        notes = (prefix + (tab.get(r, "Notes") or "")).strip() or None
        conn.execute(
            text(
                "INSERT INTO projects (id, tenant_id, operator_party_id,"
                " client_party_id, name, status, notes)"
                " VALUES (:id, :t, :op, :cp, :name, :status, :notes)"
            ),
            {
                "id": pid,
                "t": pr_tenant,
                "op": pr_party,
                "cp": cpid,
                "name": name,
                "status": status,
                "notes": notes,
            },
        )
        conn.execute(
            text(
                "INSERT INTO project_billing (project_id, billing_type, amount,"
                " currency, invoice_timing, terms, start_date,"
                " default_commission_pct, commission_party_id, expected_next, notes)"
                " VALUES (:id, :bt, :amt, :cur, :timing, :terms, :start, :pct,"
                " :cparty, :next, :notes)"
            ),
            {
                "id": pid,
                "bt": tab.get(r, "Type") or None,
                "amt": _money2(tab.get(r, "Amount"))[0],
                "cur": _cur(tab.get(r, "Cur")),
                "timing": tab.get(r, "Invoice Timing") or None,
                "terms": tab.get(r, "Terms") or None,
                "start": _date(tab.get(r, "Project Start")),
                "pct": _pct(tab.get(r, "Com. %", "Com.%")),
                "cparty": resolve_existing(tab.get(r, "Commission")),
                "next": tab.get(r, "Expected Next Invoice") or None,
                "notes": tab.get(r, "Description of Work") or None,
            },
        )
        project_by_company[_norm_company(f"{client_name} : {title}")] = pid
        if key0:
            project_by_company[_norm_company(key0)] = pid
        bump(disp, "project_client_created_on_fly" if on_fly else "project_created")
    ctx.report.reconcile("sheet.tab-Projects", len(tab.rows), disp)

    # === helpers shared by invoices + expenses ===========================
    def client_from_company(company: str) -> uuid.UUID | None:
        client_name = company.split(":", 1)[0].strip() if company else ""
        if not client_name:
            return None
        ckey = client_name.lower()
        pid = client_by_key.get(ckey) or org_by_name.get(ckey)
        if pid is None:
            pid = _insert_party(conn, ctx, kind="org", name=client_name)
            org_by_name[ckey] = pid
            client_by_key[ckey] = pid
            warn("invoice_client_on_fly")
        return pid

    # === 6. Invoices (live + arch2022, deduped; live wins) ===============
    seen_invoice_names: set[str] = set()

    def parse_invoice(t: _Tab, r: list[str]) -> dict | None:
        company = t.get(r, "Company", "Client")
        amount, over = _money2(t.get(r, "Invoice Amount"))
        ext = _ext_id(t.get(r, "Invoice ID"))
        if not company and amount is None and not ext:
            return None
        if over:
            warn("money_rounded")
        return {
            "company": company,
            "amount": amount,
            "ext": ext,
            "status": t.get(r, "Status"),
            "invoice_date": _date(t.get(r, "Invoice Date")),
            "fee": _money2(t.get(r, "Fee"))[0],
            "relevant_period": t.get(r, "Relevant Period") or None,
            "invoice_timing": t.get(r, "Invoice Timing") or None,
            "type": t.get(r, "Type") or None,
            "currency": _cur(t.get(r, "Currency", "USD")),
            "com_pct": _pct(t.get(r, "Com.%")),
            "commission": _money2(t.get(r, "Commission"))[0],
            "due_date": _date(t.get(r, "Due Date")),
            "project_amount": _money2(t.get(r, "Project Amount"))[0],
            "notes": t.get(r, "Notes") or None,
            "bw_month": _first_of_month(t.get(r, "BW Invoice")),
        }

    live_tab = _Tab(source / "tab-Invoices.csv")
    arch_tab = _Tab(source / "arch2022-Invoices.csv")
    live_recs, live_skipped = _parse_all(live_tab, parse_invoice)
    arch_recs, arch_skipped = _parse_all(arch_tab, parse_invoice)
    live_ext = {r["ext"].lower() for r in live_recs if r["ext"]}
    live_tuples = {
        (r["company"].lower(), r["invoice_date"], r["amount"]) for r in live_recs
    }
    inv_dedup_id = inv_dedup_tuple = 0
    inv_archive_only: list[dict] = []
    for r in arch_recs:
        if r["ext"] and r["ext"].lower() in live_ext:
            inv_dedup_id += 1
        elif (r["company"].lower(), r["invoice_date"], r["amount"]) in live_tuples:
            inv_dedup_tuple += 1
        else:
            inv_archive_only.append(r)

    imported_invoices = 0
    for i, r in enumerate(live_recs + inv_archive_only):
        status = STATUS_MAP.get(r["status"].strip().lower())
        if status is None:
            status = "paid"
            warn("invoice_status_defaulted")
        if not r["currency"]:
            warn("invoice_currency_blank")
        is_usd = (r["currency"] or "") == "USD"
        name = r["ext"] or f"{r['company']} {r['invoice_date'] or ''} #{i}"
        base, n = name, 1
        while name.lower() in seen_invoice_names:
            n += 1
            name = f"{base} ({n})"
        if n > 1:
            warn("invoice_name_suffixed")
        seen_invoice_names.add(name.lower())
        if project_by_company.get(_norm_company(r["company"])) is None:
            warn("invoice_project_unmatched")
        conn.execute(
            text(
                "INSERT INTO invoices (id, tenant_id, name, status, currency,"
                " invoice_date, from_party_id, to_party_id, manual_total, notes,"
                " fee, commission_pct, commission_amount, settlement_month,"
                " relevant_period, original_amount, original_currency,"
                " invoice_timing, billing_type, external_id, due_date)"
                " VALUES (:id, :t, :name, :status, :cur, :idate, :fp, :tp, :total,"
                " :notes, :fee, :cpct, :camt, :sm, :rp, :oamt, :ocur, :timing,"
                " :btype, :ext, :due)"
            ),
            {
                "id": uuid.uuid4(),
                "t": pr_tenant,
                "name": name,
                "status": status,
                "cur": CURRENCY,  # books are CAD
                "idate": r["invoice_date"],
                "fp": pr_party,
                "tp": client_from_company(r["company"]),
                "total": r["amount"],  # CAD book value
                "notes": r["notes"],
                "fee": r["fee"],
                "cpct": r["com_pct"],
                "camt": r["commission"],
                "sm": r["bw_month"],
                "rp": r["relevant_period"],
                "oamt": r["project_amount"] if is_usd else None,
                "ocur": "USD" if is_usd else None,
                "timing": r["invoice_timing"],
                "btype": r["type"],
                "ext": r["ext"] or None,
                "due": r["due_date"],
            },
        )
        imported_invoices += 1
    ctx.report.reconcile(
        "sheet.tab-Invoices",
        len(live_tab.rows),
        {"skipped_blank": live_skipped, "imported": len(live_recs)},
    )
    ctx.report.reconcile(
        "sheet.arch2022-Invoices",
        len(arch_tab.rows),
        {
            "skipped_blank": arch_skipped,
            "deduped_live_wins_id": inv_dedup_id,
            "deduped_live_wins_tuple": inv_dedup_tuple,
            "archive_only": len(inv_archive_only),
        },
    )

    # === 7. Expenses (live + arch2022, deduped; live wins) ===============
    def resolve_vendor(name: str) -> uuid.UUID | None:
        if not name:
            warn("expense_vendor_blank")
            return None
        key = name.lower()
        v = vendor_by_name.get(key)
        if v is not None:
            return v["party"]
        pid = person_by_name.get(key) or org_by_name.get(key)
        if pid is not None:
            return pid
        pid = _insert_party(conn, ctx, kind="org", name=name)
        org_by_name[key] = pid
        warn("expense_vendor_created")
        return pid

    def resolve_paid_by(s: str) -> uuid.UUID | None:
        r = (s or "").strip().lower()
        if r == "plusroi":
            return pr_party
        if r == "bowden works":
            return org_by_name.get("bowden works")
        if s:
            warn("expense_paidby_unmapped")
        return None

    def parse_expense(t: _Tab, r: list[str]) -> dict | None:
        amount, over = _money2(t.get(r, "Amount"))
        if amount is None:  # no amount → padding row, not a payable
            return None
        if over:
            warn("money_rounded")
        return {
            "project": t.get(r, "Project", "Related Project"),
            "vendor": t.get(r, "Vendor"),
            "amount": amount,
            "date": _date(t.get(r, "Expense Date")),
            "ext": _ext_id(t.get(r, "Expense Invoice ID")),
            "paid_by": t.get(r, "Paid By"),
            "category": t.get(r, "Expense Category") or None,
            "type": t.get(r, "Type") or None,
            "relevant_period": t.get(r, "Relevant Period") or None,
            "notes": t.get(r, "Notes") or None,
            "bw_month": _first_of_month(t.get(r, "BW Invoice")),
        }

    live_tab = _Tab(source / "tab-Expenses.csv")
    arch_tab = _Tab(source / "arch2022-Expenses.csv")
    live_recs, live_skipped = _parse_all(live_tab, parse_expense)
    arch_recs, arch_skipped = _parse_all(arch_tab, parse_expense)
    live_ext = {r["ext"].lower() for r in live_recs if r["ext"]}
    live_tuples = {
        (r["vendor"].lower(), r["date"], r["amount"], r["project"].lower())
        for r in live_recs
    }
    exp_dedup_id = exp_dedup_tuple = 0
    exp_archive_only: list[dict] = []
    for r in arch_recs:
        if r["ext"] and r["ext"].lower() in live_ext:
            exp_dedup_id += 1
        elif (
            r["vendor"].lower(),
            r["date"],
            r["amount"],
            r["project"].lower(),
        ) in live_tuples:
            exp_dedup_tuple += 1
        else:
            exp_archive_only.append(r)

    for r in live_recs + exp_archive_only:
        proj = project_by_company.get(_norm_company(r["project"]))
        if proj is None and r["project"]:
            warn("expense_project_unmatched")
        conn.execute(
            text(
                "INSERT INTO expenses (id, tenant_id, project_id, vendor_party_id,"
                " amount, currency, expense_date, external_invoice_id,"
                " paid_by_party_id, category, type, relevant_period,"
                " settlement_month, notes, source)"
                " VALUES (:id, :t, :proj, :vendor, :amt, :cur, :date, :ext, :paid,"
                " :cat, :type, :rp, :sm, :notes, 'sheet')"
            ),
            {
                "id": uuid.uuid4(),
                "t": pr_tenant,
                "proj": proj,
                "vendor": resolve_vendor(r["vendor"]),
                "amt": r["amount"],
                "cur": CURRENCY,
                "date": r["date"],
                "ext": r["ext"] or None,
                "paid": resolve_paid_by(r["paid_by"]),
                "cat": r["category"],
                "type": r["type"],
                "rp": r["relevant_period"],
                "sm": r["bw_month"],
                "notes": r["notes"],
            },
        )
    ctx.report.reconcile(
        "sheet.tab-Expenses",
        len(live_tab.rows),
        {"skipped_blank": live_skipped, "imported": len(live_recs)},
    )
    ctx.report.reconcile(
        "sheet.arch2022-Expenses",
        len(arch_tab.rows),
        {
            "skipped_blank": arch_skipped,
            "deduped_live_wins_id": exp_dedup_id,
            "deduped_live_wins_tuple": exp_dedup_tuple,
            "archive_only": len(exp_archive_only),
        },
    )

    # === 8. per-settlement-month summary (the settlement gate's numbers) ==
    inv_summary = [
        {
            "settlement_month": _iso(sm),
            "invoices": int(cnt),
            "manual_total": str(total),
            "fee": str(fee),
        }
        for sm, cnt, total, fee in conn.execute(
            text(
                "SELECT settlement_month, count(*), coalesce(sum(manual_total),0),"
                " coalesce(sum(fee),0) FROM invoices WHERE tenant_id = :t"
                " GROUP BY 1 ORDER BY 1 NULLS FIRST"
            ),
            {"t": pr_tenant},
        ).fetchall()
    ]
    exp_summary = [
        {
            "settlement_month": _iso(sm),
            "expenses": int(cnt),
            "amount": str(total),
        }
        for sm, cnt, total in conn.execute(
            text(
                "SELECT settlement_month, count(*), coalesce(sum(amount),0)"
                " FROM expenses WHERE tenant_id = :t GROUP BY 1 ORDER BY 1 NULLS FIRST"
            ),
            {"t": pr_tenant},
        ).fetchall()
    ]

    # === bookkeeping on the step ========================================
    for cls, count in sorted(warns.items()):
        step.warn(f"{cls}: {count}")
    written = {
        "invoices": imported_invoices,
        "expenses": _scalar(
            conn, "SELECT count(*) FROM expenses WHERE tenant_id = :t", pr_tenant
        ),
        "projects": _scalar(
            conn, "SELECT count(*) FROM projects WHERE tenant_id = :t", pr_tenant
        ),
        "client_profiles": len(profiled),
    }
    # source_count is the sum of every consumed tab's row count (tracked
    # per-tab in report.reconciliation, keyed "sheet.*").
    step.source_count = sum(
        e["source"]
        for k, e in ctx.report.reconciliation.items()
        if k.startswith("sheet.")
    )
    step.written = sum(written.values())
    step.details = {
        "tenant_id": str(pr_tenant),
        "tenant_party_id": str(pr_party),
        "owners": {"rian": str(rian_p), "rob": str(rob_p), "danielle": str(dani_p)},
        "users_linked": users_linked,
        "written": written,
        "dedupe": {
            "invoices": {
                "archive_dropped_by_id": inv_dedup_id,
                "archive_dropped_by_tuple": inv_dedup_tuple,
                "archive_only_imported": len(inv_archive_only),
            },
            "expenses": {
                "archive_dropped_by_id": exp_dedup_id,
                "archive_dropped_by_tuple": exp_dedup_tuple,
                "archive_only_imported": len(exp_archive_only),
            },
        },
        "warn_counts": warns,
        "settlement_summary": {
            "invoices_by_month": inv_summary,
            "expenses_by_month": exp_summary,
        },
    }


# --- small module helpers --------------------------------------------------


def _project_status(raw: str) -> tuple[str | None, str]:
    r = (raw or "").strip().lower()
    if r == "active" or r == "":
        return "active", ""
    if r == "closed":
        return "archived", ""
    if r == "paused":
        return "archived", "[paused] "
    return None, ""  # unknown → caller defaults + warns


def _parse_all(tab: _Tab, parse) -> tuple[list[dict], int]:
    recs, skipped = [], 0
    for r in tab.rows:
        rec = parse(tab, r)
        if rec is None:
            skipped += 1
        else:
            recs.append(rec)
    return recs, skipped


def _iso(d) -> str | None:
    return d.isoformat() if d is not None else None


def _scalar(conn: Connection, sql: str, tenant) -> int:
    return int(conn.execute(text(sql), {"t": tenant}).scalar_one())