"""Operational commands.

    python -m app.cli collect [--source SLUG] [--limit N] [--delay SECONDS]
    python -m app.cli backfill <name>     # idempotent data moves, run after a deploy
    python -m app.cli rederive            # recompute match keys under the current identity rules
    python -m app.cli status

Migrations are schema-only (build plan §2); every data move is a backfill here,
safe to run twice.
"""

import argparse
import logging
import sys

from sqlalchemy import func, select, update

from pathlib import Path

from app.db import SessionLocal
from app.models import (
    Account,
    Award,
    Brand,
    CollectionRun,
    Listing,
    Location,
    PriceObservation,
    Product,
    ProductLine,
    RawRecord,
    Retailer,
    Source,
    VariationAlias,
)
from app.services.collectors.registry import COLLECTORS, get_collector
from app.services.fx import load_rates
from app.services.awards_import import import_awards
from app.services.discussion_import import import_discussion
from app.services.images import enrich_products
from app.services.ingest import (
    DEFAULT_VERTICAL,
    is_stuck,
    mark_stuck,
    run_collector,
    sync_location_currency,
)
from app.models.catalog import IDENTITY_RULES_VERSION
from app.services.ingest import brand_slug, raw_attributes
from app.services.collectors.base import RawListing
from app.services import keying
from app.services import lines as lines_service
from app.services.merges import merge_duplicates
from app.services.normalize import parse_size, size_is_implausible, size_ml_of
from app.services.taxonomy import classify, vertical_of
from app.cli_pages import register as register_pages
from app.cli_quality import register as register_quality
from app.cli_editorial import register as register_editorial
from app.cli_accounts import register as register_accounts
from app.cli_refresh import register as register_refresh
from app.cli_hours import backfill_hours_seed, register as register_hours

DEFAULT_WINNERS_PATH = Path("/srv/import/winners.json")
DEFAULT_DISCUSSION_PATH = Path("/srv/import/discussion.json")

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)


def cmd_collect(args: argparse.Namespace) -> int:
    slugs = [args.source] if args.source else list(COLLECTORS)
    rates = load_rates()
    if rates.is_fallback:
        logging.warning("using fallback exchange rates; USD comparisons may be stale")

    exit_code = 0
    with SessionLocal() as db:
        for slug in slugs:
            try:
                collector = get_collector(slug)
            except KeyError:
                logging.error("unknown source %s (known: %s)", slug, ", ".join(COLLECTORS))
                exit_code = 2
                continue
            run = run_collector(
                db, collector, limit=args.limit, delay=args.delay, rates=rates
            )
            print(f"{slug}: {run.status} prices={run.prices_written} error={run.error or '-'}")
            if run.status not in {"ok", "skipped"}:
                exit_code = 1
    return exit_code


def cmd_awards(args: argparse.Namespace) -> int:
    path = Path(args.file) if args.file else DEFAULT_WINNERS_PATH
    if not path.is_file():
        logging.error("winners file not found: %s", path)
        return 2
    with SessionLocal() as db:
        stats = import_awards(db, path, rebuild=args.rebuild)
    print(
        f"winners={stats['winners']} matched={stats['matched']} "
        f"created={stats['created']} already_present={stats['skipped_existing']} "
        f"ambiguous={stats['ambiguous']}"
    )
    # The reconcile counters (Stream F, e2d50a3): what a re-import changed.
    print(
        f"updated={stats.get('updated', 0)} corrected={stats.get('corrected', 0)} "
        f"removed={stats.get('removed', 0)} unchanged={stats.get('unchanged', 0)} "
        f"ambiguous_same_year={stats.get('ambiguous_same_year', 0)}"
    )
    return 0


def cmd_images(args: argparse.Namespace) -> int:
    with SessionLocal() as db:
        stats = enrich_products(
            db,
            limit=args.limit,
            delay=args.delay,
            recheck=args.recheck,
            by_name=not args.barcode_only,
        )
    print(
        f"checked={stats['checked']} images_found={stats['found']} "
        f"(barcode={stats['by_barcode']} name={stats['by_name']})"
    )
    return 0


def cmd_discussion(args: argparse.Namespace) -> int:
    path = Path(args.file) if args.file else DEFAULT_DISCUSSION_PATH
    if not path.is_file():
        logging.error("discussion file not found: %s", path)
        return 2
    with SessionLocal() as db:
        stats = import_discussion(db, path)
    print(f"created={stats['created']} updated={stats['updated']}")
    return 0


def cmd_discussion_inbox(args: argparse.Namespace) -> int:
    """What rian said to Claude: every comment naming @claude on the running list (the owner's
    subjects), oldest first, with its deep link. A session reads this on its next run
    (agents.md, the checkpoint command); nobody is notified of these, by design."""
    from datetime import UTC, datetime

    from app.services.discussion import claude_inbox

    if args.recipient != "claude":
        print(f"discussion inbox: the only recipient is claude, not {args.recipient!r}")
        return 2
    since = datetime.fromisoformat(args.since).replace(tzinfo=UTC) if args.since else None
    with SessionLocal() as db:
        rows = claude_inbox(db, since)
    if not rows:
        print("inbox for claude: nothing" + (f" since {args.since}" if args.since else ""))
        return 0
    for r in rows:
        where = r["label"] or f"{r['subject_type']}:{r['subject_id']}"
        state = " (resolved)" if r["resolved"] else ""
        print(f"[{(r['at'] or '')[:16]}] {r['author']} on {where}{state}\n  {r['body']}\n  {r['url']}")
    print(f"{len(rows)} comment(s) for claude")
    return 0


def cmd_discussion_rethread(args: argparse.Namespace) -> int:
    """Apply a re-threading map once (T18): the topics it names, one move per comment, the
    listed soft deletes, every action attributed to the owner account. `--check` prints what
    would happen and writes nothing; `--apply` does it; a second `--apply` reports every entry
    as already done. The map is a file rian may edit before applying (`import/rethread-*.json`)."""
    import json as _json

    from sqlalchemy import select as _select

    from app.config import settings as _settings
    from app.models import Account
    from app.services.discussion import rethread
    from app.services.identity import Actor

    if not (args.check or args.apply) or (args.check and args.apply):
        print("discussion rethread: pass exactly one of --check and --apply")
        return 2
    # The workspace's import/ is mounted read-only at /srv/import inside the container (the plan
    # and items routes read it the same way), so a relative default resolves there first.
    candidates = [Path(args.file), Path("/srv/import") / Path(args.file).name]
    path = next((c for c in candidates if c.is_file()), None)
    if path is None:
        print(f"discussion rethread: no such file: {args.file} (looked in /srv/import too)")
        return 2
    plan = _json.loads(path.read_text())
    with SessionLocal() as db:
        owner = db.scalar(_select(Account).where(Account.username == (_settings.account_owner or "")))
        if owner is None:
            print("discussion rethread: no owner account to attribute the moves to (ACCOUNT_OWNER)")
            return 2
        lines = rethread(db, plan, actor=Actor(owner.id, owner.username, owner.display_name), check=args.check)
        for line in lines:
            print(line)
        if args.check:
            db.rollback()  # the walk wrote nothing; this only ends the transaction
            print(f"[check, nothing written] {len(lines)} entr{'y' if len(lines) == 1 else 'ies'} from {path.name}")
            return 0
        db.commit()
    print(f"discussion rethread: applied {len(lines)} entr{'y' if len(lines) == 1 else 'ies'} from {path.name}")
    return 0


OWNER_DISPLAY_NAME = "rian"


def backfill_accounts(db) -> str:
    """Seed the owner's account row (migration #1) and give it its username and status
    (migration #4) where they are NULL. Email is set later, by hand."""
    existing = db.scalar(select(Account).where(Account.display_name == OWNER_DISPLAY_NAME))
    if existing is None:
        db.add(Account(display_name=OWNER_DISPLAY_NAME, username=OWNER_DISPLAY_NAME, status="active"))
        db.commit()
        return "accounts: owner row created"
    changed = []
    if existing.username is None:
        existing.username = OWNER_DISPLAY_NAME
        changed.append("username")
    if not existing.status:
        existing.status = "active"
        changed.append("status")
    db.commit()
    if changed:
        return f"accounts: owner row id={existing.id} gained {', '.join(changed)}"
    return f"accounts: owner row already present (id={existing.id})"


def backfill_levels(db) -> str:
    """Seed the two launch levels (accounts plan §4.6): insert if absent, never update an
    edited level (the kit has no rename, so a name is fixed at its first seed)."""
    from app.models import AccountLevel
    from app.services.accounts import SEED_LEVELS

    created = []
    for name, spec in SEED_LEVELS.items():
        if db.get(AccountLevel, name) is None:
            db.add(AccountLevel(name=name, permissions=list(spec["permissions"]),
                                assignable=list(spec["assignable"])))
            created.append(name)
    db.commit()
    return f"levels: created {', '.join(created) if created else 'none'} (present: {', '.join(sorted(SEED_LEVELS))})"


# Tables whose typed who-column links by folded text. The two the SPA silently defaulted to
# "Adam" for an anonymous actor link only under --include-defaulted (rian's call, 10 Sep).
AUTHOR_COLUMNS = (
    ("discussion_comments", "author", "author_id"),
    ("quote_requests", "author", "author_id"),
    ("client_todos", "completed_by", "completed_by_id"),
    ("client_uploads", "uploaded_by", "uploaded_by_id"),
    ("owner_item_states", "acted_by", "acted_by_id"),
)
DEFAULTED_AUTHOR_COLUMNS = (
    ("feature_priorities", "author", "author_id"),
    ("quote_selections", "author", "author_id"),
)


def backfill_authors(db, args=None) -> str:
    """Link typed names to accounts: `--map Adam=adam` (repeatable) folds the typed text and
    sets the who-column's id where it is NULL; every other spelling stays text with a NULL
    link (empty beats guessed; no guest accounts are minted). `--rename` (rian, 14 Sep) also
    rewrites each mapped spelling to the account's display name on the rows it links or has
    linked ("Mark (by email, 7 Sep)" reads "Mark"): the one deliberate overwrite of a typed
    value, asked for by the person who typed it, never implied by `--map` alone. Idempotent: a
    second run changes zero rows."""
    from sqlalchemy import table as sa_table, column as sa_column

    pairs = list(getattr(args, "map", None) or [])
    if not pairs:
        return "authors: nothing to do (pass --map Typed=username, repeatable)"
    folded: dict[str, Account] = {}
    for pair in pairs:
        typed, _, username = pair.partition("=")
        if not typed or not username:
            raise ValueError(f"authors: --map expects Typed=username, got {pair!r}")
        account = db.scalar(select(Account).where(Account.username == username.strip().lower()))
        if account is None:
            raise ValueError(f"authors: no account with username {username.strip().lower()!r}")
        folded[typed.strip().lower()] = account
    rename = bool(getattr(args, "rename", False))
    columns = AUTHOR_COLUMNS + (DEFAULTED_AUTHOR_COLUMNS if getattr(args, "include_defaulted", False) else ())
    report = []
    for tname, text_col, id_col in columns:
        t = sa_table(tname, sa_column(text_col), sa_column(id_col))
        n = renamed = 0
        for key, account in folded.items():
            result = db.execute(
                update(t).where(t.c[id_col].is_(None), func.lower(func.trim(t.c[text_col])) == key)
                .values({id_col: account.id})
            )
            n += result.rowcount or 0
            if rename and account.display_name and key != account.display_name.strip().lower():
                result = db.execute(
                    update(t).where(t.c[id_col] == account.id, func.lower(func.trim(t.c[text_col])) == key)
                    .values({text_col: account.display_name})
                )
                renamed += result.rowcount or 0
        report.append(f"{tname}={n}" + (f" renamed={renamed}" if rename else ""))
    db.commit()
    return "authors: linked " + " ".join(report)


def backfill_overrides(db) -> str:
    """The home for a later human-edit source; reports what exists and writes nothing."""
    from app.models import Override

    n = db.scalar(select(func.count()).select_from(Override)) or 0
    return f"overrides: 0 changed ({n} row(s) present)"


def backfill_vertical(db) -> str:
    """products.vertical from our own category (taxonomy family); the default otherwise."""
    changed = 0
    for product in db.scalars(select(Product)):
        wanted = vertical_of(product.category) or product.vertical or DEFAULT_VERTICAL
        if product.vertical != wanted:
            product.vertical = wanted
            changed += 1
    db.commit()
    return f"vertical: {changed} product(s) updated"


def backfill_stuck_runs(db) -> str:
    """Runs still 'running' a day after they started belong to dead processes."""
    from datetime import UTC, datetime
    now = datetime.now(UTC)
    changed = 0
    for run in db.scalars(select(CollectionRun).where(CollectionRun.status == "running")):
        if is_stuck(run, now):
            mark_stuck(run, "process ended without finishing (marked by backfill)")
            changed += 1
    db.commit()
    return f"stuck_runs: {changed} run(s) marked error"


def backfill_locations(db) -> str:
    """Existing location rows take their currency from the collector that owns them."""
    changed = 0
    by_key = {}
    for collector in COLLECTORS.values():
        for spec in collector.locations():
            by_key[(collector.retailer_slug, spec.code)] = spec.currency
    for location in db.scalars(select(Location).join(Retailer)):
        declared = by_key.get((location.retailer.slug, location.code))
        if declared and sync_location_currency(location, declared):
            changed += 1
    db.commit()
    return f"locations: {changed} currency value(s) synced from collectors"


def orphan_tile_ids(db) -> list[int]:
    """Listings for a multi-size parent tile that never got its per-size rows.

    Such a row carries the FAMILY'S cheapest price against no particular size,
    which is exactly the misread the per-size rows were introduced to stop.
    """
    parents = db.scalars(select(Listing).where(Listing.source_sku.like("%-P"))).all()
    children = set(
        db.execute(
            select(Listing.location_id, Listing.source_sku).where(Listing.source_sku.like("%::%"))
        ).all()
    )
    child_parents = {(loc, sku.split("::", 1)[0]) for loc, sku in children}
    return [l.id for l in parents if (l.location_id, l.source_sku) not in child_parents]


def backfill_orphan_tiles(db) -> str:
    """Purge parent-tile listings with no per-size sibling, and products left empty.

    Purged, not re-resolved: re-resolving needs the product page, which is a
    crawl. The product page is read on the next collection of that store and
    the per-size rows are created then.
    """
    ids = orphan_tile_ids(db)
    if not ids:
        return "orphan_tiles: nothing to purge"
    product_ids = set(db.scalars(select(Listing.product_id).where(Listing.id.in_(ids))))
    db.execute(RawRecord.__table__.delete().where(RawRecord.listing_id.in_(ids)))
    db.execute(PriceObservation.__table__.delete().where(PriceObservation.listing_id.in_(ids)))
    db.execute(Listing.__table__.delete().where(Listing.id.in_(ids)))
    db.flush()
    still_listed = set(db.scalars(select(Listing.product_id).where(Listing.product_id.in_(product_ids))))
    awarded = set(db.scalars(select(Award.product_id).where(Award.product_id.in_(product_ids))))
    empty = product_ids - still_listed - awarded
    if empty:
        db.execute(Product.__table__.delete().where(Product.id.in_(empty)))
    db.commit()
    return f"orphan_tiles: {len(ids)} listing(s) purged, {len(empty)} empty product(s) removed"


def backfill_image_provenance(db) -> str:
    """Split the old single attribution into barcode-keyed and name-matched.

    Approximate for rows found before the split: a product with a barcode is
    assumed found by it. Rows enriched after the split carry the exact value.
    """
    changed = 0
    for product in db.scalars(select(Product).where(Product.image_source == "Open Food Facts")):
        product.image_source = "Open Food Facts (barcode)" if product.gtin else "Open Food Facts (name)"
        changed += 1
    db.commit()
    return f"image_provenance: {changed} product(s) relabelled"


def backfill_implausible_sizes(db) -> str:
    """Null a size no single item comes in when the product's own name does not state it.

    The first audit found two Extime mists at 7,624 and 5,924 ml (the page's own
    capacity field; the collection-time cross-check now refuses it) and two
    whiskies at 7,000 ml. The size goes, the match key follows it, and the next
    collection re-reads the row through the corrected parser.
    """
    changed = 0
    maps = keying.load_maps(db)
    for product in db.scalars(select(Product).where(Product.size_ml.isnot(None))):
        if size_is_implausible(product.vertical, product.name, product.size_ml):
            logger.info("size_nulled product=%s name=%r was=%s", product.id, product.name, product.size_ml)
            product.size_ml = None
            product.match_key, _ = keying.product_key(product, maps)
            changed += 1
    db.commit()
    return f"implausible_sizes: {changed} product(s) set to unknown size"


def display_spelling(counter) -> str:
    """The spelling to show for a brand: mixed case over SHOUTING, then the most common.

    One storefront capitalises every house ("SISLEY", "DIOR") and lists more
    products than the rest, so the raw majority would shout on every brand
    page. Trademark glyphs are not part of a name.
    """
    def rank(item):
        spelling, count = item
        return (spelling.isupper(), -count, spelling)
    best = sorted(counter.items(), key=rank)[0][0]
    return best.replace("\u00ae", "").replace("\u2122", "").strip() or best


def backfill_brands(db) -> str:
    """One brands row per fold key, products pointed at it (migration #3).

    A row is created with the spelling most products carry; an existing row's
    name is never changed here (a human may have set it). Products keep the
    spelling they were collected with in `brand`; `brand_id` is the fold.
    """
    from collections import Counter, defaultdict
    spellings: dict[str, Counter] = defaultdict(Counter)
    for brand, n in db.execute(
        select(Product.brand, func.count(Product.id)).where(Product.brand.isnot(None)).group_by(Product.brand)
    ):
        slug = brand_slug(brand)
        if slug:
            spellings[slug][brand.strip()] += n
    existing = {b.slug: b for b in db.scalars(select(Brand))}
    created = 0
    for slug, counter in spellings.items():
        if slug not in existing:
            row = Brand(slug=slug, name=display_spelling(counter))
            db.add(row)
            existing[slug] = row
            created += 1
    db.flush()
    pointed = 0
    for product in db.scalars(select(Product).where(Product.brand.isnot(None), Product.brand_id.is_(None))):
        row = existing.get(brand_slug(product.brand))
        if row is not None:
            product.brand_id = row.id
            pointed += 1
    db.commit()
    return f"brands: {created} brand row(s) created, {pointed} product(s) pointed at their brand"


def backfill_sizes(db) -> str:
    """size_value/size_unit as the name states it, for rows that have a size_ml and no stated size."""
    changed = 0
    for product in db.scalars(
        select(Product).where(Product.size_ml.isnot(None), Product.size_value.is_(None))
    ):
        stated = parse_size(product.name)
        if stated and size_ml_of(*stated) == product.size_ml:
            product.size_value, product.size_unit = stated
        else:
            product.size_value, product.size_unit = float(product.size_ml), "ml"
        changed += 1
    db.commit()
    return f"sizes: {changed} product(s) given a stated size"


def backfill_categories(db) -> str:
    """Classify the uncategorised again under the widened rules and the brand map.

    Only rows with no category are touched (a category, once set, is never
    overwritten here); the vertical follows the new category. Rows nothing
    matches stay uncategorised: no category reads better than a wrong one.
    """
    changed = 0
    by_vertical: dict[str, int] = {}
    for product in db.scalars(select(Product).where(Product.category.is_(None))):
        category = classify(product.name, product.brand)
        if category is None:
            continue
        product.category = category
        wanted = vertical_of(category) or product.vertical
        if product.vertical != wanted:
            product.vertical = wanted
        by_vertical[category] = by_vertical.get(category, 0) + 1
        changed += 1
    db.commit()
    detail = ", ".join(f"{k} {v}" for k, v in sorted(by_vertical.items(), key=lambda kv: -kv[1]))
    return f"categories: {changed} product(s) categorised ({detail or 'none'})"


def backfill_merges(db, args=None) -> str:
    """The duplicate groups as recorded merges; the unsettled ones as candidates.

    Run after `rederive`, so the groups are read under the current identity
    rules. Safe to repeat: a merged row is skipped, a candidate pair is
    recorded once. With `--check` (Stream L) the group list is printed first, one line
    per group (key, member ids, names, barcodes; a candidate group with its conflicts), so
    the groups can be read before a real run.
    """
    from app.services.merges import duplicate_groups

    if args is not None and getattr(args, "check", False):
        groups = duplicate_groups(db)
        for group in sorted(groups, key=lambda g: (g.mergeable, g.match_key)):
            kind = "MERGE" if group.mergeable else "CANDIDATE " + ",".join(group.conflicts)
            members = "; ".join(f"{p.id} {p.name!r}" + (f" gtin={p.gtin}" if p.gtin else "") for p in group.members)
            print(f"{kind}  {group.match_key}  {members}")
        print(f"{sum(1 for g in groups if g.mergeable)} mergeable group(s), {sum(1 for g in groups if not g.mergeable)} candidate group(s)")
    counts = merge_duplicates(db)
    db.commit()
    return (
        f"merges: {counts['groups']} group(s) merged ({counts['merged_rows']} row(s) now forward), "
        f"{counts['candidate_groups']} group(s) left for a human ({counts['candidates_added']} new candidate pair(s))"
    )


def cmd_rederive(args: argparse.Namespace) -> int:
    """Recompute every product's match key and attributes under the current identity rules.

    A rules change must reach the rows resolved before it, or two spellings of one house
    (v2) or of one variation (v3) stay two products forever. Rows are re-keyed in place
    through the same call ingest uses (`keying.product_key`: the house with its alias
    followed, the canonical variation from the alias table, then the v3 key) and stamped
    with the rules version; `backfill merges` then folds whatever now shares a key.
    Idempotent: a second run changes nothing.
    """
    rekeyed = stamped = restamped = 0
    with SessionLocal() as db:
        maps = keying.load_maps(db)
        # Every row, tombstones included (Stream L): a human merge's tombstone must carry
        # the current rules' key for ingest to follow it; `merged_into_id` is never touched.
        # The decided layer is read first inside `product_key` (overrides on the name's
        # collected spelling, the line, the variation, the quantity).
        for product in db.scalars(select(Product).order_by(Product.id)):
            key, attributes = keying.product_key(product, maps)
            if product.match_key != key:
                logging.info("rekeyed product=%s %r -> %r", product.id, product.match_key, key)
                product.match_key = key
                rekeyed += 1
            if product.vertical == "beauty" and not (product.attributes or {}).get("concentration"):
                attrs = raw_attributes(RawListing(
                    source_sku="", name=product.name, price=1.0, currency="EUR",
                    location_code="", brand=product.brand, vertical=product.vertical,
                ))
                attributes = {**attributes, **{k: v for k, v in attrs.items() if k not in attributes}}
            if attributes != (product.attributes or {}):
                product.attributes = attributes
                restamped += 1
            if product.identity_rules_version != IDENTITY_RULES_VERSION:
                product.identity_rules_version = IDENTITY_RULES_VERSION
                stamped += 1
        db.commit()
    print(
        f"rederive: {rekeyed} match key(s) changed, {restamped} attribute set(s) changed, "
        f"{stamped} product(s) stamped rules v{IDENTITY_RULES_VERSION}"
    )
    return 0


# name -> function(db) -> summary line. Each must be safe to run twice.
def backfill_perfume(db) -> str:
    """Rename the stored category Fragrance to Perfume (structure review, 9 Sep:
    "Search is always our driver"; the searched word has 19x the volume).

    Idempotent: a row already reading Perfume is untouched, and a second run
    reports zero. The vertical (beauty) does not change; `backfill categories`
    cannot do this because it touches only uncategorised rows.
    """
    changed = (
        db.execute(
            update(Product).where(Product.category == "Fragrance").values(category="Perfume")
        ).rowcount
        or 0
    )
    db.commit()
    return f"perfume: {changed} product(s) renamed from Fragrance to Perfume"


def _backfill_threads(db, args=None) -> str:
    """Migration #6: a thread per legacy comment key, then the legacy mentions rung
    (app/services/discussion.py). The mention step resolves names through the accounts kit,
    which the API process wires at startup and this process must wire itself: the first
    rehearsal on a copy of staging raised NOT_INITIALIZED here."""
    from app.services import accounts
    from app.services.discussion import backfill_threads

    accounts.init()
    return backfill_threads(db, args)


def _backfill_thread_reads_seed(db, args=None) -> str:
    """The one-time read seed: everything before today is read for everyone
    (app/services/discussion.py, the workflow plan §3)."""
    from app.services.discussion import backfill_thread_reads_seed

    return backfill_thread_reads_seed(db, args)


def _backfill_legacy_mentions(db, args=None) -> str:
    """T14: a mention notification for every legacy comment naming a current account that never
    rang, under the write path's own dedupe key (app/services/discussion.py). Wires the
    accounts kit first, as `_backfill_threads` does."""
    from app.services import accounts
    from app.services.discussion import backfill_legacy_mentions

    accounts.init()
    return backfill_legacy_mentions(db, args)
def backfill_lines(db) -> str:
    """One product_lines row per house and line key (`services/lines.py`), products pointed at it.

    The house is the product's brand row with its alias followed; a product with no brand
    gets no line. A row is created with the spelling the most common product name gives
    the key's words; an existing row's name is never changed here (a person may have set
    it), and a product joins the canonical line when its row is an alias. Safe to repeat:
    a second run creates nothing and points nothing.
    """
    from collections import Counter, defaultdict

    brands = {b.id: b for b in db.scalars(select(Brand))}
    rows = list(db.scalars(select(ProductLine)))
    by_key = {(row.brand_id, row.key): row for row in rows}
    by_id = {row.id: row for row in rows}
    slugs = {row.slug for row in rows}
    keyed = []
    samples: dict[tuple[int, str], Counter] = defaultdict(Counter)
    spellings: dict[tuple[int, str], set[str]] = defaultdict(set)
    from app.services import overrides as overrides_service

    decided = overrides_service.read(db, "product", fields=("name", "line_id"))
    for product in db.scalars(
        select(Product).where(Product.merged_into_id.is_(None), Product.brand_id.isnot(None))
    ):
        house = lines_service.resolve_alias(brands, product.brand_id)
        if house is None:
            continue
        rows = decided.get(str(product.id), {})
        if "line_id" in rows:
            continue  # a person chose the line; the rules never point it elsewhere
        # A renamed product keys on the spelling its listings arrive under (LT3 b).
        name = str(rows["name"].collected_value) if "name" in rows and rows["name"].collected_value else product.name
        key = lines_service.line_key(name, brand=product.brand, house=house.name, vertical=product.vertical)
        keyed.append((product, house, key))
        samples[(house.id, key)][name] += 1
        spellings[(house.id, key)].update({product.brand or "", house.name})
    created = pointed = 0
    for product, house, key in keyed:
        row = by_key.get((house.id, key))
        if row is None:
            base = lines_service.line_slug(house.slug, key)[:230]
            slug, n = base, 2
            while slug in slugs:
                slug, n = f"{base}-{n}", n + 1
            row = ProductLine(
                brand_id=house.id, key=key, slug=slug,
                name=lines_service.display_line_name(
                    key, samples[(house.id, key)], house.name,
                    brand_words=lines_service.brand_words_of(*spellings[(house.id, key)]),
                    vertical=product.vertical,
                )[:200],
            )
            db.add(row)
            db.flush()
            by_key[(house.id, key)] = row
            by_id[row.id] = row
            slugs.add(slug)
            created += 1
        target = lines_service.resolve_alias(by_id, row.id) or row
        if product.line_id != target.id:
            product.line_id = target.id
            pointed += 1
    db.commit()
    return f"lines: {created} line(s) created, {pointed} product(s) pointed at their line"


def backfill_quantities(db, args=None) -> str:
    """Every live product's standard quantity (identity rules v4, `services/quantity.py`):
    the name parsed with the stored v3 pair as the hint the text outranks (the pair is our
    past parse; 7293 held 75.00 ml for a 75 g stick), the seven quantity columns and
    `name_key` written, `size_ml` filled where the dimension is ml and it was NULL, and
    `size_ml` (with the derived pair) CLEARED only where the parsed dimension is g or pcs and
    the stored `size_ml` equals the same figure read as ml, the misread's signature; never
    otherwise. A product with a decided `quantity` override is never touched. Run before
    `rederive`, which keys on these columns. Safe to repeat: a second run writes zero.
    """
    from app.services import overrides as overrides_service
    from app.services.ingest import write_quantity
    from app.services.quantity import parse_quantity

    decided = overrides_service.read(db, "product", fields=("quantity", "name"))
    written = cleared = filled = skipped = 0
    misreads: list[str] = []

    def snapshot(p):
        return (None if p.quantity_value is None else float(p.quantity_value), p.quantity_unit, p.pack_count,
                None if p.pack_unit_value is None else float(p.pack_unit_value), p.form, p.set_contents,
                p.quantity_state, p.name_key, p.size_ml, None if p.size_value is None else float(p.size_value), p.size_unit)

    for product in db.scalars(select(Product).where(Product.merged_into_id.is_(None)).order_by(Product.id)):
        rows = decided.get(str(product.id), {})
        if "quantity" in rows:
            skipped += 1
            continue
        name = str(rows["name"].collected_value) if "name" in rows and rows["name"].collected_value else product.name
        if product.size_value is not None and product.size_unit:
            hint = (float(product.size_value), product.size_unit)
        elif product.size_ml:
            hint = (product.size_ml, "ml")
        else:
            hint = None
        before = snapshot(product)
        q = parse_quantity(name, hint=hint, category=product.category)
        write_quantity(product, q, name=name)
        if q.state == "stated" and q.unit == "ml" and product.size_ml is None and q.ml:
            product.size_ml = q.ml
            filled += 1
        elif (q.state == "stated" and q.unit in ("g", "pcs") and q.value is not None
              and product.size_ml is not None and product.size_ml == round(float(q.value))):
            misreads.append(f"{product.id} {product.name!r} {product.size_ml} ml -> {q.value:g} {q.unit}")
            product.size_ml = product.size_value = product.size_unit = None
            cleared += 1
        if snapshot(product) != before:
            written += 1
    db.commit()
    for line in misreads[:40]:
        logging.info("size_ml cleared: %s", line)
    return (f"quantities: {written} product(s) written, {filled} size_ml filled from a millilitre name, "
            f"{cleared} size_ml cleared (a gram or piece figure once read as ml), {skipped} decided and untouched")


def backfill_listed(db, args=None) -> str:
    """Every listing's LISTED columns from its newest fragment (`collected.listed_fields`,
    the same reader ingest uses), `listed_record_id` set to that fragment; every column left
    NULL where no fragment exists (NULL is the honest value: a copy of our own text would be
    labelled as the shop's); `last_seen_at` filled where NULL from the newest of the
    listing's price observations and raw records. The newest fragment is `max(id)` per
    listing through a portable subquery, read slim (`payload - 'tile_html'` on Postgres).
    Safe to repeat: a second run writes zero.
    """
    from app.models import Listing, PriceObservation, RawRecord
    from app.services.collected import write_listed

    newest = (
        select(RawRecord.listing_id.label("listing_id"), func.max(RawRecord.id).label("record_id"))
        .group_by(RawRecord.listing_id).subquery()
    )
    slim = RawRecord.payload.op("-")("tile_html") if db.get_bind().dialect.name == "postgresql" else RawRecord.payload
    written = 0
    for listing, record_id, parser_version, payload in db.execute(
        select(Listing, RawRecord.id, RawRecord.parser_version, slim)
        .join(newest, newest.c.listing_id == Listing.id)
        .join(RawRecord, RawRecord.id == newest.c.record_id)
        .order_by(Listing.id)
    ):
        if write_listed(listing, record_id, parser_version, payload):
            written += 1
    db.flush()
    seen = (
        select(Listing.id.label("listing_id"),
               func.max(PriceObservation.observed_at).label("seen"))
        .join(PriceObservation, PriceObservation.listing_id == Listing.id)
        .where(Listing.last_seen_at.is_(None)).group_by(Listing.id).subquery()
    )
    seen_raw = (
        select(RawRecord.listing_id.label("listing_id"), func.max(RawRecord.created_at).label("seen"))
        .group_by(RawRecord.listing_id).subquery()
    )
    stamped = 0
    latest: dict[int, object] = {}
    for listing_id, at in db.execute(select(seen.c.listing_id, seen.c.seen)):
        latest[listing_id] = at
    for listing_id, at in db.execute(select(seen_raw.c.listing_id, seen_raw.c.seen)):
        if at is not None and (listing_id not in latest or latest[listing_id] is None or at > latest[listing_id]):
            latest[listing_id] = at
    if latest:
        for listing in db.scalars(select(Listing).where(Listing.last_seen_at.is_(None), Listing.id.in_(list(latest)))):
            listing.last_seen_at = latest[listing.id]
            stamped += 1
    db.commit()
    without = int(db.scalar(select(func.count(Listing.id)).where(Listing.listed_record_id.is_(None))) or 0)
    return (f"listed: {written} listing(s) written from their newest fragment, {stamped} last_seen_at filled, "
            f"{without} listing(s) with no fragment (listed columns NULL)")


def backfill_variations(db) -> str:
    """The variation vocabulary seeded from what the names say, and every product of a
    vertical with variations stamped with its canonical one.

    The wording found in a name (`lines.variation_of`, "elixir parfum intense") becomes a
    `variation_aliases` row if absent, canonical by the rule ("elixir"); a row a person
    decided keeps its canonical and display, and that wins for the product.
    `attributes.variation` is set (the dict reassigned, never mutated in place); the parsed
    concentration stays as the raw record. Safe to repeat: a second run seeds nothing and
    stamps nothing.
    """
    aliases = {(a.vertical, a.raw): a for a in db.scalars(select(VariationAlias))}
    seeded = stamped = kinds = 0
    for product in db.scalars(
        select(Product).where(
            Product.merged_into_id.is_(None),
            Product.vertical.in_(sorted(lines_service.VARIATION_VERTICALS)),
        )
    ):
        # The category reaches the rule (Stream L): a Makeup shade is a color, a Skincare
        # " / Grasa" tail is a skin type and gets none; the flavor wordings come from
        # confectionery names.
        raw, canonical = lines_service.variation_of(product.name, product.vertical, category=product.category)
        kind = lines_service.variation_kind_of(product.name, product.vertical, category=product.category) if raw else ""
        if raw:
            alias = aliases.get((product.vertical, raw))
            if alias is None:
                alias = VariationAlias(
                    vertical=product.vertical, raw=raw, canonical=canonical,
                    display=lines_service.display_variation(canonical), kind=kind or None,
                )
                db.add(alias)
                aliases[(product.vertical, raw)] = alias
                seeded += 1
            elif not alias.kind and kind:
                alias.kind = kind
                kinds += 1
            canonical = alias.canonical
            kind = alias.kind or kind
        attrs = dict(product.attributes or {})
        wanted = {k: v for k, v in attrs.items() if k not in ("variation", "variation_kind")}
        if canonical:
            wanted["variation"] = canonical
            if kind:
                wanted["variation_kind"] = kind
        if wanted != attrs:
            product.attributes = wanted
            stamped += 1
    db.commit()
    return (f"variations: {seeded} wording(s) seeded, {stamped} product(s) stamped with their variation"
            + (f", {kinds} kind(s) recorded on existing wordings" if kinds else ""))


BACKFILLS = {
    "accounts": backfill_accounts,
    "levels": backfill_levels,
    "authors": backfill_authors,
    "overrides": backfill_overrides,
    "vertical": backfill_vertical,
    "stuck_runs": backfill_stuck_runs,
    "locations": backfill_locations,
    "orphan_tiles": backfill_orphan_tiles,
    "image_provenance": backfill_image_provenance,
    "implausible_sizes": backfill_implausible_sizes,
    "brands": backfill_brands,
    "sizes": backfill_sizes,
    "categories": backfill_categories,
    "perfume": backfill_perfume,
    "merges": backfill_merges,
    "threads": _backfill_threads,
    "legacy_mentions": _backfill_legacy_mentions,
    "thread_reads_seed": _backfill_thread_reads_seed,
    "hours_seed": backfill_hours_seed,  # app/cli_hours.py (Stream G)
    "lines": backfill_lines,
    "variations": backfill_variations,
    "quantities": backfill_quantities,  # Stream L: before rederive
    "listed": backfill_listed,  # Stream L: after merges
}


def cmd_propose(args: argparse.Namespace) -> int:
    """File a proposals file as merge candidates with reason `proposed` (Stream L): the
    session's own reading of the data, for a person to approve on the desk; the rules never
    withdraw or rescore them; nothing is merged. `--check` reports the counts and writes
    nothing. The file: `{"proposed_by": "...", "entries": [{level, left_id, right_id, score,
    why}, ...]}`; `import/` is mounted read-only at `/srv/import` in the container."""
    from app.services import merge_desk

    entries, proposed_by = merge_desk.load_proposals(args.file)
    with SessionLocal() as db:
        result = merge_desk.propose(db, entries, proposed_by=args.by or proposed_by, check=args.check)
        if args.check:
            db.rollback()
        else:
            db.commit()
    prefix = "[check, nothing written] " if args.check else ""
    print(f"{prefix}propose: {result['inserted']} pair(s) inserted, {result['endorsed']} rule pair(s) endorsed, "
          f"{result['skipped_present']} already present or decided, "
          f"{result['skipped_missing']} with a side gone, {result['invalid']} invalid, of {len(entries)} in {args.file}")
    for note in result["notes"]:
        print("  ", note)
    return 0


def cmd_suggest(args: argparse.Namespace) -> int:
    """Fill the merge queue at brand, line and product level from the rules
    (`services/suggest.py`). Idempotent: a pair is inserted once, an undecided pair is
    refreshed, a decided pair is never touched, a pair one side of which is gone is closed."""
    from app.services import suggest

    with SessionLocal() as db:
        if args.check:
            db.commit = db.flush  # type: ignore[method-assign]
        counts = suggest.generate(db)
        if args.check:
            db.rollback()
        else:
            db.commit()
    print(
        f"{'[check, nothing written] ' if args.check else ''}suggest: {counts['brand']} brand, {counts['line']} line, "
        f"{counts['product']} product pair(s) added; {counts['refreshed']} refreshed, {counts['superseded']} closed as "
        f"superseded, {counts['withdrawn']} withdrawn, {counts['legacy_filled']} earlier pair(s) given their level"
    )
    return 0


def cmd_backfill(args: argparse.Namespace) -> int:
    """Run one backfill. `--check` prints what a run would change and writes nothing: every
    commit inside the backfill becomes a flush, and the transaction is rolled back at the
    end, so a backfill written before the flag existed honours it too."""
    import inspect

    if args.name != "authors" and (args.map or args.include_defaulted or getattr(args, "rename", False)):
        print(f"backfill {args.name}: --map, --include-defaulted and --rename belong to `authors` only")
        return 2
    fn = BACKFILLS[args.name]
    with SessionLocal() as db:
        if args.check:
            db.commit = db.flush  # type: ignore[method-assign]
        try:
            takes_args = len(inspect.signature(fn).parameters) >= 2
            message = fn(db, args) if takes_args else fn(db)
        except ValueError as exc:
            db.rollback()
            print(exc)
            return 2
        if args.check:
            db.rollback()
            message = f"[check, nothing written] {message}"
        print(message)
    return 0


def match_locations(locations: list, tokens: list[str]) -> list:
    """Resolve user-typed codes to locations by code or IATA, case-insensitively.

    A token that matches nothing is an error (raise, don't guess): a demo set
    silently missing an airport is worse than a rejected command.
    """
    wanted = {t.upper() for t in tokens}
    matched, hit = [], set()
    for loc in locations:
        keys = {loc.code.upper()} | ({loc.iata.upper()} if loc.iata else set())
        if keys & wanted:
            matched.append(loc)
            hit |= keys & wanted
    missing = wanted - hit
    if missing:
        raise SystemExit(f"no location matches: {', '.join(sorted(missing))}")
    return matched


def cmd_locations(args: argparse.Namespace) -> int:
    with SessionLocal() as db:
        locations = list(db.scalars(select(Location).order_by(Location.code)))
        if args.only:
            chosen = {loc.id for loc in match_locations(locations, args.only)}
            for loc in locations:
                loc.visible = loc.id in chosen
        elif args.show or args.hide:
            for loc in match_locations(locations, args.show or []):
                loc.visible = True
            for loc in match_locations(locations, args.hide or []):
                loc.visible = False
        db.commit()

        counts = dict(
            db.execute(
                select(Listing.location_id, func.count(Listing.id)).group_by(
                    Listing.location_id
                )
            ).all()
        )
        for loc in locations:
            state = "visible" if loc.visible else "HIDDEN"
            iata = loc.iata or "-"
            print(f"  {loc.code:28s} {iata:4s} {state:8s} {counts.get(loc.id, 0):5d} listings")
    return 0


def cmd_status(_: argparse.Namespace) -> int:
    with SessionLocal() as db:
        multi = (
            select(Listing.product_id)
            .group_by(Listing.product_id)
            .having(func.count(func.distinct(Listing.location_id)) > 1)
            .subquery()
        )
        print(f"products:      {db.scalar(select(func.count(Product.id))) or 0}")
        print(f"comparable:    {db.scalar(select(func.count()).select_from(multi)) or 0}")
        print(f"observations:  {db.scalar(select(func.count(PriceObservation.id))) or 0}")
        for source in db.scalars(select(Source).order_by(Source.slug)):
            last = db.scalar(
                select(CollectionRun)
                .where(CollectionRun.source_id == source.id)
                .order_by(CollectionRun.started_at.desc())
                .limit(1)
            )
            state = "enabled" if source.enabled else "PAUSED"
            detail = f"{last.status} ({last.prices_written} prices)" if last else "never run"
            print(f"  {source.slug:22s} {state:8s} {detail}")
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(prog="app.cli")
    sub = parser.add_subparsers(dest="command", required=True)

    collect = sub.add_parser("collect", help="run collectors into the database")
    collect.add_argument("--source", help="collector slug; default is every enabled collector")
    collect.add_argument("--limit", type=int, default=None, help="max listings per source")
    collect.add_argument("--delay", type=float, default=None, help="politeness delay in seconds")
    collect.set_defaults(func=cmd_collect)

    awards = sub.add_parser("awards", help="import competition medals onto products")
    awards.add_argument("--file", help=f"winners JSON (default {DEFAULT_WINNERS_PATH})")
    awards.add_argument(
        "--rebuild", action="store_true",
        help="drop competition-network medals and re-match from scratch",
    )
    awards.set_defaults(func=cmd_awards)

    images = sub.add_parser("images", help="attach openly licensed product imagery")
    images.add_argument("--limit", type=int, default=None)
    images.add_argument("--delay", type=float, default=0.8)
    images.add_argument("--recheck", action="store_true", help="re-check products already looked up")
    images.add_argument("--barcode-only", action="store_true", help="skip the name-search fallback")
    images.set_defaults(func=cmd_images)

    discussion = sub.add_parser("discussion", help="the decision cards import, and the inbox a session reads")
    dsub = discussion.add_subparsers(dest="discussion_command", required=True)
    dimport = dsub.add_parser("import", help="import the decisions/realities list onto /discuss")
    dimport.add_argument("--file", help=f"discussion JSON (default {DEFAULT_DISCUSSION_PATH})")
    dimport.set_defaults(func=cmd_discussion)
    dinbox = dsub.add_parser("inbox", help="what rian said to @claude on the running list, oldest first")
    dinbox.add_argument("--for", dest="recipient", required=True, metavar="HANDLE", help="claude")
    dinbox.add_argument("--since", metavar="DATE", help="ISO date; only comments on or after it")
    dinbox.set_defaults(func=cmd_discussion_inbox)
    drethread = dsub.add_parser("rethread", help="apply a re-threading map once: topics, moves, soft deletes (T18)")
    drethread.add_argument("--file", default="import/rethread-2026-09-13.json", help="the map (default: the 13 Sep map)")
    drethread.add_argument("--check", action="store_true", help="print what would happen; write nothing")
    drethread.add_argument("--apply", action="store_true", help="apply the map, attributed to the owner account")
    drethread.set_defaults(func=cmd_discussion_rethread)

    backfill = sub.add_parser("backfill", help="run one idempotent data move after a deploy")
    backfill.add_argument("name", choices=sorted(BACKFILLS))
    backfill.add_argument("--check", action="store_true", help="print what would change; write nothing")
    backfill.add_argument("--map", action="append", default=[], metavar="TYPED=USERNAME",
                          help="authors only: link this typed name to this account (repeatable)")
    backfill.add_argument("--rename", action="store_true",
                          help="authors only: also rewrite each mapped spelling to the account's display name")
    backfill.add_argument("--include-defaulted", action="store_true",
                          help="authors only: also link feature_priorities and quote_selections, "
                               "where the page defaulted an anonymous actor to a name")
    backfill.set_defaults(func=cmd_backfill)

    rederive = sub.add_parser("rederive", help="recompute match keys under the current identity rules")
    rederive.set_defaults(func=cmd_rederive)

    suggest = sub.add_parser("suggest", help="fill the merge queue at brand, line and product level from the rules")
    suggest.add_argument("--check", action="store_true", help="print what would change; write nothing")
    suggest.set_defaults(func=cmd_suggest)

    propose = sub.add_parser("propose", help="file a proposals JSON as merge candidates a person approves on the desk")
    propose.add_argument("--file", required=True, help="the proposals file (import/proposals/<date>-<level>.json)")
    propose.add_argument("--by", default=None, help="who proposed (default: the file's proposed_by)")
    propose.add_argument("--check", action="store_true", help="report the counts; write nothing")
    propose.set_defaults(func=cmd_propose)

    status = sub.add_parser("status", help="show catalog and collector status")
    status.set_defaults(func=cmd_status)

    locations = sub.add_parser(
        "locations",
        help="list locations and toggle site visibility (collectors keep running)",
    )
    locations.add_argument("--show", nargs="+", metavar="CODE", help="make these visible")
    locations.add_argument("--hide", nargs="+", metavar="CODE", help="hide these from the site")
    locations.add_argument(
        "--only", nargs="+", metavar="CODE",
        help="make exactly these visible and hide everything else",
    )
    locations.set_defaults(func=cmd_locations)

    register_quality(sub)  # audit, verify: app/cli_quality.py (Stream Q)
    register_pages(sub)  # indexnow: app/cli_pages.py (Stream B)
    register_editorial(sub)  # articles, subscribers: app/cli_editorial.py (Stream D)
    register_accounts(sub)  # accounts, sessions, audit-log: app/cli_accounts.py (Stream R)
    register_refresh(sub)  # staging-refresh export | apply: app/cli_refresh.py (Stream R2)
    register_hours(sub)  # hours collect/set/show: app/cli_hours.py (Stream G)

    args = parser.parse_args(argv)
    return args.func(args)


if __name__ == "__main__":
    sys.exit(main())