"""Backfill every FY27 white-hat report (2026-04-01 onward) into Helix.

Source of truth: the Gmail threads on ``security@pingcap.com``, read
2026-08-25. Two ledger files sit next to this script:

  * ``fy27_threads.json`` — one row per thread: researcher, first
    submission date, subject, and the affected host WHEN the subject or
    body names one. Nothing inferred.
  * ``verdicts.json``     — only threads we actually answered, with the
    reply's own words quoted. A thread with no reply gets status ``new``
    and stays visibly unanswered, because that is the truth and it is
    what the 「等我们回复」 report exists to surface.

Deliberately NOT recorded, even though the mail contains it: bank
account / IBAN / SWIFT / tax forms, and researchers' legal names from
invoices. Helix stores no banking data — by design, and this corpus is
exactly why (those details sit in plain text in an inbox).

Idempotent: keyed on (researcher_email, title). A re-run inserts nothing
and prints what it found. Codes are assigned by the service in
chronological order, so a re-run cannot renumber anything either.

Usage — PYTHONPATH must point at the checkout whose ``app`` you mean:

    PYTHONPATH=/app python backfill_bounty_fy27.py --ledger /opt/helix/backfill
    PYTHONPATH=/app python backfill_bounty_fy27.py --ledger ... --apply
"""
from __future__ import annotations

import argparse
import asyncio
import json
import pathlib
import sys
from collections import Counter
from datetime import UTC, datetime
from decimal import Decimal

from sqlalchemy import select

#: The ledger lives OUTSIDE the repo, and is passed with ``--ledger``.
#: It carries 56 external researchers' email addresses; the bounty tables
#: keep those behind ``include_sensitive`` + the security-role gate,
#: whereas a file in git is readable by every clone forever with no gate
#: at all. Same reasoning as the no-banking-data rule one paragraph up.
DEFAULT_LEDGER = pathlib.Path("/opt/helix/backfill")

# Senders that are our own aliases: the thread arrived by forward, so the
# real researcher is not in the header. Skipped rather than guessed — a
# report attributed to security@pingcap.com is worse than one absent.
INTERNAL_SENDERS = {
    "security@pingcap.com", "support@pingcap.com", "shuai.feng@pingcap.com",
    "dong.zhang@pingcap.com",
}

# Host → registry slug. Only hosts the seed registered; anything else
# stays free text and the operator links it by hand.
HOST_TO_SLUG = {
    "tidbcloud.com": "tidb-cloud",
    "www.pingcap.com": "pingcap-www",
    "ticard.pingcap.com": "ticard",
    "clinic.pingcap.com": "clinic",
    "jira-crm.tidbcloud.com": "jira-crm",
    "one-engine.pingcap.com": "one-engine",
    "docs.pingcap.com": "docs",
    "labs.pingcap.com": "labs",
}


def _dt(day: str) -> datetime:
    return datetime.fromisoformat(f"{day}T00:00:00+00:00")


async def main(apply: bool, ledger: pathlib.Path) -> int:
    import app.main  # noqa: F401  — registers every mapper before use
    from app.core.database import async_session_factory
    from app.domains.bounty import service as svc
    from app.domains.bounty.models import BountyReport, BountyResearcher
    from app.identity.person_models import IdentityPerson

    threads = json.loads((ledger / "fy27_threads.json").read_text("utf-8"))
    verdicts = json.loads((ledger / "verdicts.json").read_text("utf-8"))
    verdicts.pop("_doc", None)

    created: list[str] = []
    skipped: list[str] = []
    dropped: list[tuple[str, str]] = []

    async with async_session_factory() as db:
        # Registry slugs → ids, so a report lands on its system.
        slug_to_id: dict[str, object] = {}
        slug_track: dict[str, str] = {}
        try:
            from app.domains.system_registry.models import RegisteredSystem

            rows = (await db.execute(select(RegisteredSystem))).scalars().all()
            slug_to_id = {r.slug: r.id for r in rows}
            slug_track = {r.slug: r.budget_track for r in rows}
        except Exception as exc:  # pragma: no cover - registry not migrated
            print(f"! system registry unavailable ({exc}) — links skipped")

        # Four of this corpus's threads share one subject line ("Bug
        # Bounty Report"). The dedupe key is (researcher, title), so
        # without disambiguation the second and third submissions would
        # be silently swallowed as re-runs of the first — three real
        # reports lost. Suffix the date only where the pair actually
        # repeats, so every other title stays exactly what the
        # researcher wrote.
        repeated = {
            key for key, n in Counter(
                (r["who"].strip().lower(), r["subject"].strip())
                for r in threads if not r.get("skip")
            ).items() if n > 1
        }

        for row in sorted(threads, key=lambda r: r["first"]):
            title = row["subject"].strip()
            email = row["who"].strip().lower()
            if (email, title) in repeated:
                title = f"{title}({row['first']})"

            if row.get("skip"):
                dropped.append((title, row["skip"]))
                continue
            if email in INTERNAL_SENDERS:
                dropped.append((title, "sender is an internal alias"))
                continue

            v = verdicts.get(row["tid"], {})
            reported = _dt(v.get("first") or row["first"])

            existing = (await db.execute(
                select(BountyReport)
                .join(BountyResearcher,
                      BountyResearcher.id == BountyReport.researcher_id)
                .where(BountyResearcher.email == email,
                       BountyReport.title == title,
                       BountyReport.deleted_at.is_(None))
            )).scalars().first()
            if existing is not None:
                skipped.append(f"{existing.code} {title[:48]}")
                continue

            slug = HOST_TO_SLUG.get(row.get("sys") or "")
            # Pot precedence mirrors the HTTP route exactly: what the mail
            # says > what the registry says the system belongs to > the
            # platform default. Calling the service directly would
            # otherwise charge every www.pingcap.com finding to 产研.
            track = v.get("track") or slug_track.get(slug) or "product"
            report = await svc.create_report(
                db,
                title=title[:200],
                researcher_email=email,
                reported_at=reported,
                affected_system=row.get("sys"),
                system_id=slug_to_id.get(slug),
                cve_id=row.get("cve"),
                vuln_type=row.get("vuln"),
                budget_track=track,
                source="email",
                summary=row.get("note"),
                actor=None,
            )

            # ── What we told the researcher, if anything ──
            if v:
                if v.get("quote"):
                    await svc.add_event(
                        db, report=report, event_type="comment", actor=None,
                        actor_kind="system",
                        body=f"邮件答复原文摘录:{v['quote']}",
                        visibility="internal",
                    )
                if v.get("note"):
                    await svc.add_event(
                        db, report=report, event_type="comment", actor=None,
                        actor_kind="system", body=v["note"],
                        visibility="internal",
                    )
                if v.get("decided"):
                    report.last_replied_at = _dt(v["decided"])
                    report.reply_count = max(1, report.reply_count or 0)

                status = v.get("status")
                if status and status != "new":
                    report.status = status
                if status == "confirmed":
                    report.remediation_status = (
                        "fixed" if v.get("remediation") == "fixed"
                        else "pending"
                    )
                    if v.get("remediation") == "fixed":
                        report.fixed_at = _dt(v["decided"])
                if v.get("severity"):
                    report.severity = v["severity"]
                if v.get("reward") is not None:
                    report.reward_amount_usd = Decimal(str(v["reward"]))
                    report.reward_decided_at = _dt(v["decided"])
                    report.fiscal_year = await _fy(db, _dt(v["decided"]))
                    report.payment_status = v.get("payment", "pending")
                    if report.payment_status in ("requested", "paid"):
                        report.payment_requested_at = _dt(v["decided"])
                    if report.payment_status == "paid":
                        # The date the money actually landed, not the date
                        # we decided it — the budget's "paid" figure is a
                        # cash fact.
                        report.paid_at = _dt(v.get("paid") or v["decided"])
                        report.closed_at = report.paid_at
                if v.get("assignee"):
                    person = (await db.execute(
                        select(IdentityPerson).where(
                            IdentityPerson.work_email == v["assignee"],
                            IdentityPerson.deleted_at.is_(None),
                        )
                    )).scalars().first()
                    if person is None:
                        print(f"! assignee {v['assignee']} not found — left "
                              f"unassigned on {report.code}")
                    else:
                        report.assignee_person_id = person.id
                        report.assignee_name = (
                            person.display_name or person.work_email
                        )
                        report.assigned_at = _dt(v["assigned"])

            await svc.add_event(
                db, report=report, event_type="created", actor=None,
                actor_kind="system",
                body="从 security@pingcap.com 邮件补录(FY27)。",
                meta={"backfill": f"gmail:{row['tid']}",
                      "backfilled_at": datetime.now(UTC).isoformat()},
                visibility="internal",
            )
            created.append(f"{report.code} {title[:48]}")

        print(f"\ncreate: {len(created)}")
        for line in created:
            print(f"  + {line}")
        print(f"\nalready present, untouched: {len(skipped)}")
        for line in skipped:
            print(f"  = {line}")
        print(f"\nnot backfilled: {len(dropped)}")
        for title, why in dropped:
            print(f"  - {title[:60]} — {why}")

        if apply:
            await db.commit()
            print("\nCOMMITTED")
        else:
            await db.rollback()
            print("\nDRY RUN — nothing written (pass --apply to write)")
    return 0


async def _fy(db, when: datetime) -> int:
    """The FY a reward DECISION falls in. Deliberately the decision date,
    not the submission date: one of these reports was submitted in
    November 2025 and decided (and paid) in May 2026, and the budget it
    burned is FY27's."""
    from app.core import fiscal_year as fy

    start_month = await fy.get_fiscal_year_start_month(db)
    return fy.fy_of_period(f"{when.year:04d}-{when.month:02d}", start_month)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--apply", action="store_true")
    parser.add_argument(
        "--ledger", type=pathlib.Path, default=DEFAULT_LEDGER,
        help="directory holding fy27_threads.json + verdicts.json",
    )
    args = parser.parse_args()
    sys.exit(asyncio.run(main(args.apply, args.ledger)))