"""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

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


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). 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, int] = {}
    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.id
    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 = 0
        for key, account_id 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
        report.append(f"{tname}={n}")
    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) -> 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.
    """
    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)
        for product in db.scalars(select(Product).where(Product.merged_into_id.is_(None))):
            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_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)
    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
        key = lines_service.line_key(product.name, brand=product.brand, house=house.name, vertical=product.vertical)
        keyed.append((product, house, key))
        samples[(house.id, key)][product.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_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 = 0
    for product in db.scalars(
        select(Product).where(
            Product.merged_into_id.is_(None),
            Product.vertical.in_(sorted(lines_service.VARIATION_VERTICALS)),
        )
    ):
        raw, canonical = lines_service.variation_of(product.name, product.vertical)
        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),
                )
                db.add(alias)
                aliases[(product.vertical, raw)] = alias
                seeded += 1
            canonical = alias.canonical
        attrs = dict(product.attributes or {})
        wanted = {k: v for k, v in attrs.items() if k != "variation"}
        if canonical:
            wanted["variation"] = canonical
        if wanted != attrs:
            product.attributes = wanted
            stamped += 1
    db.commit()
    return f"variations: {seeded} wording(s) seeded, {stamped} product(s) stamped with their variation"


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,
    "lines": backfill_lines,
    "variations": backfill_variations,
}


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):
        print(f"backfill {args.name}: --map and --include-defaulted 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="import the decisions/realities list")
    discussion.add_argument("--file", help=f"discussion JSON (default {DEFAULT_DISCUSSION_PATH})")
    discussion.set_defaults(func=cmd_discussion)

    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("--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)

    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)

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


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