"""Backfill asset purchase amounts and currencies. Dry-run by default.

Why (2026-09-05). 资产财务完整度 read 27.6%. The 760-row gap is almost
entirely the 2026-04 易盘点 legacy import: ``purchase_date`` is missing on
ZERO rows and ``purchase_entity`` on three, while ``purchase_cost`` and
``currency`` are each missing on 718. Two fixable shapes:

**Pass 1 — 42 rows that have an amount but no currency.**
⚠️ **Do NOT derive the currency from the purchase entity.** That was the
obvious rule and it is wrong: every one of those 42 amounts is a
mainland-China list price (8499 / 10999 / 12999 / 14999 …) including the
rows filed under ``PINGCAP (US), INC.``, and among the rows that ARE
complete the US entity carries 21 CNY against 7 USD. Production evidence
beat the tidy rule, so this pass writes ``CNY`` flat and says so.

**Pass 2 — same-model median for rows with no amount.**
Only where at least ``--min-samples`` same-model devices have both an
amount and a currency, and only where the sample's own spread is within
``--max-spread`` (max/min). A model whose priced siblings range 1.5x+ is
skipped and listed: a median across a genuinely mixed configuration is a
number nobody can defend. Written with
``purchase_cost_source='estimated_model'`` (migration 604) so no read
surface can mistake it for a recorded figure.

**Ceiling — MEASURED on production 2026-09-05, not estimated.** The
number this docstring first carried ("roughly 50%", "~200 of 692 rows
estimable") was a desk estimate and it was wrong by a wide margin. The
actual run:

    live assets                    1050
    complete before                 290  (27.6%)
    pass 1 — currency written        42
    pass 2 — median written          52   ← not ~200
      skipped, <3 same-model priced siblings   502
      skipped, sibling spread > 1.5x             3 models
      skipped, mixed sibling currencies          1 model
    complete after                  384  (36.6%)

So the real ceiling of both passes is **36.6%, not ~50%** — still the
30-point band, so the conclusion is unchanged, but the number was
overstated by ~14 points. Passing 90% needs the finance fixed-asset
register, which is what the reconciliation surface is for. The lesson
this paragraph now carries: **a coverage figure written before the run
is a guess; say which one it is.**

⚠️ **A dry run under-reports pass 2 by construction.** Pass 1 gives 42
rows a currency, and a row needs BOTH an amount and a currency to count
as a priced sibling — so those rows only become eligible references
after pass 1 has actually written. The dry run projected 50 estimable;
the apply wrote 52. Expect the applied count to be >= the dry-run count,
never less.

Usage::

    # always start here — writes nothing, prints every intended change
    .venv/bin/python scripts/backfill_asset_purchase_cost.py

    .venv/bin/python scripts/backfill_asset_purchase_cost.py --apply
    .venv/bin/python scripts/backfill_asset_purchase_cost.py --apply \\
        --only currency        # pass 1 only
    .venv/bin/python scripts/backfill_asset_purchase_cost.py \\
        --min-samples 5 --max-spread 1.2

Every write carries a ``trace_id`` audit row naming the pass, the old
value and the new one, so the whole run is reversible from the audit log.
"""
from __future__ import annotations

import argparse
import asyncio
import statistics
import sys
import uuid
from collections import defaultdict
from decimal import Decimal
from pathlib import Path
from typing import Any

# Running as a script, not a module: put the app package on the path
# before importing anything from it.
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

LEGACY_IMPORT_CURRENCY = "CNY"


def _parse_args() -> argparse.Namespace:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument(
        "--apply",
        action="store_true",
        help="actually write. Omit for a dry run (the default).",
    )
    ap.add_argument(
        "--only",
        choices=("currency", "estimate"),
        help="run only one pass.",
    )
    ap.add_argument(
        "--min-samples",
        type=int,
        default=3,
        help=(
            "minimum same-model priced siblings before estimating "
            "(default 3)."
        ),
    )
    ap.add_argument(
        "--max-spread",
        type=float,
        default=1.5,
        help=(
            "skip a model whose priced siblings' max/min exceeds this "
            "(default 1.5)."
        ),
    )
    ap.add_argument(
        "--limit",
        type=int,
        default=0,
        help="cap the rows touched per pass (0 = no cap).",
    )
    return ap.parse_args()


async def main() -> int:
    args = _parse_args()
    from sqlalchemy import select

    import app.main  # noqa: F401  — registers every ORM model
    from app.audit.service import AuditService
    from app.core.database import async_session_factory
    from app.domains.asset.models import (
        PURCHASE_COST_SOURCE_ESTIMATED_MODEL,
        AssetItem,
    )

    trace_id = f"backfill:asset_purchase_cost:{uuid.uuid4()}"
    mode = "APPLY" if args.apply else "DRY-RUN"
    print(f"=== asset purchase-cost backfill [{mode}] trace={trace_id}")

    async with async_session_factory() as db:
        live = (
            await db.execute(
                select(AssetItem).where(
                    AssetItem.deleted_at.is_(None),
                    AssetItem.status != "disposed",
                )
            )
        ).scalars().all()
        print(f"live assets: {len(live)}")

        def complete(a: AssetItem) -> bool:
            return all(
                getattr(a, f) is not None
                for f in (
                    "purchase_date",
                    "purchase_cost",
                    "currency",
                    "purchase_entity",
                )
            )

        before_complete = sum(1 for a in live if complete(a))
        print(
            f"financial completeness before: {before_complete}/{len(live)}"
            f" = {before_complete / max(1, len(live)) * 100:.1f}%"
        )

        audit = AuditService(db)
        changed = 0

        # ── Pass 1: amount present, currency missing ──────────────────
        if args.only in (None, "currency"):
            targets = [
                a for a in live
                if a.purchase_cost is not None and a.currency is None
            ]
            if args.limit:
                targets = targets[: args.limit]
            print(
                f"\n-- pass 1 (currency): {len(targets)} row(s) "
                f"→ {LEGACY_IMPORT_CURRENCY}"
            )
            by_entity: dict[str, int] = defaultdict(int)
            for a in targets:
                by_entity[a.purchase_entity or "(none)"] += 1
                print(
                    f"   {a.asset_tag or a.id}  cost={a.purchase_cost}  "
                    f"entity={a.purchase_entity}  -> "
                    f"currency={LEGACY_IMPORT_CURRENCY}"
                )
            for entity, n in sorted(
                by_entity.items(), key=lambda kv: -kv[1]
            ):
                print(f"     [{n:>4}] {entity}")
            if args.apply:
                for a in targets:
                    old = a.currency
                    a.currency = LEGACY_IMPORT_CURRENCY
                    await audit.log(
                        trace_id=trace_id,
                        actor_type="system",
                        action="asset.purchase_currency_backfilled",
                        status="success",
                        resource_type="asset_item",
                        resource_id=str(a.id),
                        detail={
                            "pass": "currency",
                            "old_currency": old,
                            "new_currency": LEGACY_IMPORT_CURRENCY,
                            "purchase_cost": str(a.purchase_cost),
                            "purchase_entity": a.purchase_entity,
                            "reason": (
                                "legacy import amounts are mainland-China "
                                "list prices regardless of filing entity"
                            ),
                        },
                    )
                    changed += 1
                await db.flush()

        # ── Pass 2: same-model median estimate ────────────────────────
        if args.only in (None, "estimate"):
            priced: dict[Any, list[tuple[Decimal, str]]] = defaultdict(list)
            for a in live:
                if (
                    a.model_id is not None
                    and a.purchase_cost is not None
                    and a.currency is not None
                ):
                    priced[a.model_id].append((a.purchase_cost, a.currency))

            targets = [
                a for a in live
                if a.purchase_cost is None and a.model_id is not None
            ]
            usable: list[tuple[AssetItem, Decimal, str, int]] = []
            skipped_few: list[AssetItem] = []
            skipped_spread: dict[Any, tuple[Decimal, Decimal, int]] = {}
            skipped_mixed_ccy: dict[Any, set[str]] = {}

            for a in targets:
                samples = priced.get(a.model_id) or []
                if len(samples) < args.min_samples:
                    skipped_few.append(a)
                    continue
                currencies = {c for _amt, c in samples}
                if len(currencies) > 1:
                    # A median across currencies is meaningless; no FX
                    # conversion here on purpose (the rate would have to
                    # be the purchase-date rate, which is unknown).
                    skipped_mixed_ccy[a.model_id] = currencies
                    continue
                amounts = sorted(amt for amt, _c in samples)
                spread = float(amounts[-1] / amounts[0]) if amounts[0] else 0
                if spread > args.max_spread:
                    skipped_spread[a.model_id] = (
                        amounts[0], amounts[-1], len(amounts),
                    )
                    continue
                median = Decimal(
                    str(statistics.median([float(x) for x in amounts]))
                ).quantize(Decimal("0.01"))
                usable.append((a, median, currencies.pop(), len(amounts)))

            if args.limit:
                usable = usable[: args.limit]
            print(
                f"\n-- pass 2 (same-model median): {len(usable)} row(s) "
                f"estimable of {len(targets)} amount-less rows with a model"
            )
            for a, median, ccy, n in usable:
                # ``model`` is the free-text mirror and is often NULL on
                # legacy rows that only carry ``model_id`` — print the id
                # rather than "None", or the operator cannot tell WHICH
                # model's median is about to be written into a money
                # column.
                label = a.model or f"model_id={a.model_id}"
                print(
                    f"   {a.asset_tag or a.id}  {label}  "
                    f"-> {median} {ccy}  (median of {n})"
                )
            print(
                f"   skipped, too few samples (<{args.min_samples}): "
                f"{len(skipped_few)}"
            )
            if skipped_spread:
                print(
                    f"   skipped, spread > {args.max_spread}x — needs a "
                    f"human price ({len(skipped_spread)} model(s)):"
                )
                for mid, (lo, hi, n) in skipped_spread.items():
                    print(f"     model_id={mid}  {lo}–{hi}  (n={n})")
            if skipped_mixed_ccy:
                print(
                    f"   skipped, mixed currencies "
                    f"({len(skipped_mixed_ccy)} model(s)):"
                )
                for mid, ccys in skipped_mixed_ccy.items():
                    print(f"     model_id={mid}  {sorted(ccys)}")

            if args.apply:
                for a, median, ccy, n in usable:
                    a.purchase_cost = median
                    a.purchase_cost_source = (
                        PURCHASE_COST_SOURCE_ESTIMATED_MODEL
                    )
                    if a.currency is None:
                        a.currency = ccy
                    await audit.log(
                        trace_id=trace_id,
                        actor_type="system",
                        action="asset.purchase_cost_estimated",
                        status="success",
                        resource_type="asset_item",
                        resource_id=str(a.id),
                        detail={
                            "pass": "estimate",
                            "model": a.model,
                            "model_id": str(a.model_id),
                            "estimated_cost": str(median),
                            "currency": ccy,
                            "sample_size": n,
                            "source": (
                                PURCHASE_COST_SOURCE_ESTIMATED_MODEL
                            ),
                        },
                    )
                    changed += 1
                await db.flush()

        after_complete = sum(1 for a in live if complete(a))
        print(
            f"\nfinancial completeness after: {after_complete}/{len(live)}"
            f" = {after_complete / max(1, len(live)) * 100:.1f}%"
        )
        print(f"rows changed: {changed}")

        if args.apply:
            await db.commit()
            print("COMMITTED")
        else:
            await db.rollback()
            print("DRY RUN — nothing written. Re-run with --apply.")
    return 0


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))