"""The collectors and catalogue, as the review page reads them.

Sources of truth: the collector registry (which airports a collector powers, declared in
code), and the database (what it actually wrote). Nothing here is stored: every number is
computed on request, so a figure on the page can never be a stale copy of one.

This is the reading half of the QA review area. It answers five questions a person asks
when something looks wrong: which collectors are running and what do they power, what is
in the catalogue, what does one product actually look like across the airports, what did
each shop's tile actually say beside what we made of it (the listings table, the surface
every identity decision is checked against), and what is currently wrong. It writes
nothing; the deciding half (confirming a merge, clearing a check) stays with the routes
and commands that record who decided.
"""

from __future__ import annotations

import re
import unicodedata
from collections import defaultdict
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import Select, and_, func, or_, select
from sqlalchemy.orm import Session

from app.models import (
    Brand,
    CollectionRun,
    Listing,
    Shop,
    Suggestion,
    PriceObservation,
    ProductVariant,
    ProductLine,
    RawRecord,
    Source,
)
from app.models.quality import AuditSnapshot, RejectedObservation, VerificationCheck
from app.services import audit as audit_service
from app.services import catalog_queries as cq
from app.services import verify as verify_service
from app.services.collectors.registry import COLLECTORS
from app.services.collected import _brand_name, _num, _platform, _stated_size, _text, collected_tile  # noqa: F401  (moved to collected.py; re-exported for the tests and the page)
from app.services.product_lines import resolve_alias
from app.services.normalize import brand_key

#: How a collector's module maps to the commerce platform behind it. One collector per
#: platform is the design (build plan §3b): ten Avolta airports are one parser, not ten.
PLATFORM_NAMES = {
    "avolta": "Avolta (Magento)",
    "shopify": "Shopify",
    "heinemann": "Heinemann (SAP)",
    "heinemann_platform": "Heinemann (SAP)",
    "dubai": "Dubai (Oracle Commerce)",
    "ari": "ARI (Salesforce)",
    "extime": "Extime (Next.js)",
    "changi": "iShopChangi (rendered)",
    "shilla": "Shilla (rendered)",
}

MISSING_CHOICES = ("image", "size", "gtin", "category", "brand")


def platform_of(slug: str) -> str:
    collector = COLLECTORS.get(slug)
    if collector is None:
        return "unknown"
    module = type(collector).__module__.rsplit(".", 1)[-1]
    return PLATFORM_NAMES.get(module, module)


def _has_checks(db: Session) -> bool:
    from sqlalchemy import inspect

    return "verification_checks" in inspect(db.connection()).get_table_names()


def _iso(value: datetime | None) -> str | None:
    return value.astimezone(UTC).isoformat(timespec="minutes") if value else None


# --------------------------------------------------------------------------- collectors

def _story(runs: list[CollectionRun], refused_since: datetime | None = None) -> dict[str, Any]:
    """What this collector's run history says, in the terms a person actually asks in.

    Not every refusal is the same thing: one that has never worked is a door we have never
    been through, one that worked and stopped is a door that closed on us, and one that
    fails now and then is usually pace or a bad hour rather than a policy. `refused_since` is
    the newest BLOCKED verification check on the source (Stream L): a host that refuses our
    declared identity has no run to show it, and "nothing attempted for N days" was the wrong
    story for Dubai (check 396, 11 Sep; Decision 3). It wins when no run has started since.
    """
    if not runs:
        story = f"refused since {refused_since.date().isoformat()}" if refused_since else "never run"
        return {"story": story, "runs": 0, "ok": 0, "refused": 0, "failed": 0, "recent": [],
                "refused_since": _iso(refused_since)}
    ordered = sorted(runs, key=lambda r: r.started_at or datetime.min.replace(tzinfo=UTC))
    kinds = ["ok" if r.status == "ok" else "refused" if r.status == "blocked" else "failed" for r in ordered]
    ok, refused, failed = kinds.count("ok"), kinds.count("refused"), kinds.count("failed")
    last_ok = next((r for r, k in zip(reversed(ordered), reversed(kinds)) if k == "ok"), None)
    first_bad = next((r for r, k in zip(ordered, kinds) if k != "ok"), None)

    newest = ordered[-1].started_at
    stale_days = (datetime.now(UTC) - newest).days if newest else None

    if ok == len(kinds):
        story = "clean every run"
    elif ok == 0:
        story = "never once succeeded"
    elif kinds[-1] == "ok" and refused + failed:
        story = "working now, after trouble"
    elif kinds[-1] != "ok" and ok:
        story = "worked before, refused now"
    else:
        story = "on and off"
    # A collector we stopped attempting reads as clean forever, which is the opposite of the
    # truth: the last thing it did worked, and that was weeks ago. Say so.
    if stale_days is not None and stale_days >= 7 and kinds[-1] == "ok":
        story = f"nothing attempted for {stale_days} days"
    if refused_since is not None and (newest is None or newest < refused_since):
        story = f"refused since {refused_since.date().isoformat()}"
    return {
        "story": story, "runs": len(kinds), "ok": ok, "refused": refused, "failed": failed,
        "stale_days": stale_days, "refused_since": _iso(refused_since),
        "first_at": _iso(ordered[0].started_at), "last_ok_at": _iso(last_ok.started_at) if last_ok else None,
        "first_bad_at": _iso(first_bad.started_at) if first_bad else None,
        "recent": kinds[-12:],
    }


def collectors(db: Session) -> dict[str, Any]:
    """One row per collector: what it powers, when it last ran, and whether it is trusted."""
    sources = {s.slug: s for s in db.scalars(select(Source))}
    blocked = verify_service.blocked_sources(db)
    rates = verify_service.pass_rates(db)

    # The last run of each source, and the counts its shops hold, in one query each rather
    # than one per collector.
    last_runs: dict[int, CollectionRun] = {}
    all_runs: dict[int, list[CollectionRun]] = defaultdict(list)
    for run in db.scalars(select(CollectionRun).order_by(CollectionRun.started_at.desc())):
        last_runs.setdefault(run.source_id, run)
        all_runs[run.source_id].append(run)
    refused_since: dict[int, datetime] = dict(db.execute(
        select(VerificationCheck.source_id, func.max(VerificationCheck.checked_at))
        .where(VerificationCheck.verdict == "BLOCKED", VerificationCheck.source_id.isnot(None))
        .group_by(VerificationCheck.source_id)
    ).all()) if _has_checks(db) else {}

    by_code = {loc.code: loc for loc in db.scalars(select(Shop))}
    counts = {
        loc_id: (listings, product_variants)
        for loc_id, listings, product_variants in db.execute(
            select(Listing.shop_id, func.count(Listing.id), func.count(func.distinct(Listing.variant_id)))
            .group_by(Listing.shop_id)
        )
    }
    newest = {
        loc_id: at
        for loc_id, at in db.execute(
            select(Listing.shop_id, func.max(PriceObservation.observed_at))
            .join(PriceObservation, PriceObservation.listing_id == Listing.id)
            .group_by(Listing.shop_id)
        )
    }
    blocked_shops = set(cq.blocked_shop_ids(db))

    rows = []
    for slug, collector in sorted(COLLECTORS.items()):
        source = sources.get(slug)
        specs = list(collector.shops())
        places, listings_n, product_variants_n, newest_at = [], 0, 0, None
        for spec in specs:
            loc = by_code.get(spec.code)
            l_n, p_n = counts.get(loc.id, (0, 0)) if loc else (0, 0)
            at = newest.get(loc.id) if loc else None
            listings_n += l_n
            product_variants_n += p_n
            newest_at = max(newest_at, at) if (newest_at and at) else (at or newest_at)
            places.append({
                "code": spec.code, "iata": spec.iata, "name": spec.name,
                "city": spec.city, "country": spec.country,
                "catalogue_only": spec.is_catalogue_only,
                "visible": bool(loc.visible) if loc else False,
                "publication_blocked": bool(loc and loc.id in blocked_shops),
                "listings": l_n, "product_variants": p_n, "newest_price_at": _iso(at),
            })

        run = last_runs.get(source.id) if source else None
        rate = rates.get(source.id, {}) if source else {}
        open_failures = [
            {"id": c.id, "verdict": c.verdict, "detail": c.detail, "url": c.url,
             "checked_at": _iso(c.checked_at)}
            for c in (blocked.get(source.id, []) if source else [])
        ]
        rows.append({
            "slug": slug,
            "name": source.name if source else slug,
            "platform": platform_of(slug),
            "enabled": bool(source.enabled) if source else False,
            "identity_mode": source.identity_mode if source else None,
            "permission_record": source.permission_record if source else None,
            "shops": places,
            "airports": [p["iata"] for p in places if p["iata"]],
            "listings": listings_n,
            "product_variants": product_variants_n,
            "newest_price_at": _iso(newest_at),
            "last_run": {
                "at": _iso(run.started_at), "finished_at": _iso(run.finished_at),
                "status": run.status, "prices": run.prices_written,
                "products_seen": run.products_seen, "error": run.error,
                "skipped_no_price": run.skipped_no_price, "skip_counts": run.skip_counts or {},
            } if run else None,
            "history": _story(all_runs.get(source.id, []) if source else [],
                              refused_since.get(source.id) if source else None),
            "verify": {
                "blocked": bool(open_failures),
                "open_failures": open_failures,
                "checks": rate.get("checks", 0),
                "pass_rate": rate.get("rate"),
                "spot_checked": _iso(verify_service.spot_checked(db, source.id)) if source else None,
            },
        })
    return {"collectors": rows, "platforms": sorted({r["platform"] for r in rows})}


# --------------------------------------------------------------------------- catalogue

def quantity_label(product: ProductVariant) -> str | None:
    """What a person reads for a product's quantity: the v4 columns ("75 g", "24 x 330 ml",
    "5ml+15ml+15ml+50ml set"), else the stated v3 pair, else nothing."""
    state = getattr(product, "quantity_state", None)
    if state == "stated":
        if product.form == "set" and product.set_contents:
            return f"{product.set_contents} set"
        if product.quantity_value is None or not product.quantity_unit:
            return None
        value = float(product.quantity_value)
        if product.form == "pack" and product.pack_count and product.pack_unit_value is not None:
            return f"{product.pack_count} x {float(product.pack_unit_value):g} {product.quantity_unit}"
        label = f"{value:g} {product.quantity_unit}"
        return f"{label} {product.form}" if product.form in ("refill",) else label
    if product.quantity_stated_value and product.quantity_stated_unit:
        return f"{product.quantity_stated_value:g} {product.quantity_stated_unit}"
    return None


def _alive():
    return ProductVariant.merged_into_id.is_(None)


def catalogue(db: Session) -> dict[str, Any]:
    """The counts a person asks for first: how much do we hold, and how good is it."""
    alive = _alive()
    one = lambda stmt: db.scalar(stmt) or 0  # noqa: E731

    total = one(select(func.count(ProductVariant.id)).where(alive))
    merged = one(select(func.count(ProductVariant.id)).where(ProductVariant.merged_into_id.isnot(None)))
    priced = one(
        select(func.count(func.distinct(Listing.variant_id)))
        .join(ProductVariant, ProductVariant.id == Listing.variant_id).where(alive)
    )

    # Products by how many airports stock them: the comparison that is the whole point.
    per_product = (
        select(
            Listing.variant_id.label("pid"),
            func.count(func.distinct(Listing.shop_id)).label("n"),
        )
        .join(ProductVariant, ProductVariant.id == Listing.variant_id)
        .where(alive)
        .group_by(Listing.variant_id)
        .subquery()
    )
    spread = {
        int(n): int(count)
        for n, count in db.execute(
            select(per_product.c.n, func.count())
            .select_from(per_product)
            .group_by(per_product.c.n)
            .order_by(per_product.c.n)
        ).all()
    }
    comparable = sum(count for n, count in spread.items() if n >= 2)

    def missing(condition) -> int:
        return one(select(func.count(ProductVariant.id)).where(and_(alive, condition)))

    verticals = [
        {"vertical": v or "(none)", "product_variants": n}
        for v, n in db.execute(
            select(ProductVariant.vertical, func.count(ProductVariant.id)).where(alive)
            .group_by(ProductVariant.vertical).order_by(func.count(ProductVariant.id).desc())
        )
    ]
    categories = [
        {"category": c or "(none)", "product_variants": n}
        for c, n in db.execute(
            select(ProductVariant.category, func.count(ProductVariant.id)).where(alive)
            .group_by(ProductVariant.category).order_by(func.count(ProductVariant.id).desc())
        )
    ]
    sizes = [
        {"size": f"{float(v):g} {u}" if v is not None and u else "(unknown)", "product_variants": n}
        for v, u, n in db.execute(
            select(ProductVariant.quantity_value, ProductVariant.quantity_unit, func.count(ProductVariant.id)).where(alive)
            .group_by(ProductVariant.quantity_value, ProductVariant.quantity_unit)
            .order_by(func.count(ProductVariant.id).desc()).limit(25)
        )
    ]
    return {
        "product_variants": {
            "total": total, "priced": priced, "merged_away": merged,
            "comparable": comparable, "single_airport": spread.get(1, 0),
            "exclusives": one(select(func.count(ProductVariant.id)).where(and_(alive, ProductVariant.is_exclusive.is_(True)))),
        },
        "brands": {
            "total": one(select(func.count(Brand.id))),
            "resolved": one(select(func.count(Brand.id)).where(Brand.alias_of_id.is_(None))),
            "aliases": one(select(func.count(Brand.id)).where(Brand.alias_of_id.isnot(None))),
        },
        "have": {
            "image": missing(ProductVariant.image_url.isnot(None)),
            "gtin": missing(ProductVariant.gtin.isnot(None)),
            "size": missing(ProductVariant.quantity_state == "stated"),
            "category": missing(ProductVariant.category.isnot(None)),
            "brand": missing(ProductVariant.brand_id.isnot(None)),
        },
        "missing": {
            "image": missing(ProductVariant.image_url.is_(None)),
            "gtin": missing(ProductVariant.gtin.is_(None)),
            "size": missing(or_(ProductVariant.quantity_state.is_(None), ProductVariant.quantity_state != "stated")),
            "category": missing(ProductVariant.category.is_(None)),
            "brand": missing(ProductVariant.brand_id.is_(None)),
        },
        "airport_spread": [{"airports": n, "product_variants": c} for n, c in sorted(spread.items())],
        "verticals": verticals,
        "categories": categories,
        "sizes": sizes,
        "listings": one(select(func.count(Listing.id))),
        "observations": one(select(func.count(PriceObservation.id))),
    }


# --------------------------------------------------------------------------- product variants

#: What a person may sort the product table by, and the column behind it.
SORTS = {
    "name": ProductVariant.name,
    "brand": ProductVariant.brand,
    "category": ProductVariant.category,
    "size": ProductVariant.quantity_value,
    "airports": None,   # the computed count
    "newest": None,     # the newest observation
}
GROUPS = {
    "brand": ProductVariant.brand,
    "category": ProductVariant.category,
    "vertical": ProductVariant.vertical,
    "size": ProductVariant.quantity_value,
}


def _product_filter(
    db: Session, *, q: str | None, collector: str | None, airport: str | None,
    category: str | None, vertical: str | None, missing: str | None, min_airports: int | None,
    brand: str | None = None, quantity_ml: float | None = None,
):
    """The WHERE clauses shared by the table and its grouped view."""
    where = [_alive()]
    if q:
        like = f"%{q.strip().lower()}%"
        where.append(or_(func.lower(ProductVariant.name).like(like), func.lower(ProductVariant.brand).like(like)))
    if category:
        where.append(ProductVariant.category == category)
    if vertical:
        where.append(ProductVariant.vertical == vertical)
    if brand:
        where.append(ProductVariant.brand == brand)
    if quantity_ml is not None:
        # The `quantity_ml` filter is the quantity figure in its own unit (the grouped view's key).
        where.append(or_(ProductVariant.quantity_value == quantity_ml,
                         and_(ProductVariant.quantity_value.is_(None), ProductVariant.quantity_ml == quantity_ml)))
    if missing == "size":
        where.append(or_(ProductVariant.quantity_state.is_(None), ProductVariant.quantity_state != "stated"))
    elif missing in MISSING_CHOICES:
        column = {"image": ProductVariant.image_url, "gtin": ProductVariant.gtin,
                  "category": ProductVariant.category, "brand": ProductVariant.brand_id}[missing]
        where.append(column.is_(None))

    shop_ids = _scope_shop_ids(db, collector, airport)
    if shop_ids is not None:
        where.append(ProductVariant.id.in_(select(Listing.variant_id).where(Listing.shop_id.in_(shop_ids), cq.live_listings())))
    return where, min_airports


def _scope_shop_ids(db: Session, collector: str | None, airport: str | None) -> list[int] | None:
    """The shops a collector or airport filter names; None for no filter, [-1] for none."""
    shop_ids: list[int] | None = None
    if collector:
        codes = [spec.code for spec in COLLECTORS[collector].shops()] if collector in COLLECTORS else []
        shop_ids = list(db.scalars(select(Shop.id).where(Shop.code.in_(codes)))) or [-1]
    if airport:
        ids = list(db.scalars(select(Shop.id).where(Shop.iata == airport.upper()))) or [-1]
        shop_ids = [i for i in ids if shop_ids is None or i in shop_ids] or [-1]
    return shop_ids


def _counted(where: list, min_airports: int | None) -> Select:
    """Product ids with the number of airports that stock each, filtered."""
    stmt = (
        select(ProductVariant.id.label("pid"), func.count(func.distinct(Listing.shop_id)).label("airports"))
        .select_from(ProductVariant).outerjoin(Listing, and_(Listing.variant_id == ProductVariant.id, cq.live_listings()))
        .where(and_(*where)).group_by(ProductVariant.id)
    )
    if min_airports:
        stmt = stmt.having(func.count(func.distinct(Listing.shop_id)) >= min_airports)
    return stmt


def product_variants(
    db: Session, *, q: str | None = None, collector: str | None = None, airport: str | None = None,
    category: str | None = None, vertical: str | None = None, missing: str | None = None,
    min_airports: int | None = None, brand: str | None = None, quantity_ml: float | None = None,
    sort: str = "name", descending: bool = False,
    page: int = 1, per_page: int = 50,
) -> dict[str, Any]:
    """One page of the product table, with each product's airports and price range."""
    where, min_airports = _product_filter(
        db, q=q, collector=collector, airport=airport, category=category,
        vertical=vertical, missing=missing, min_airports=min_airports, brand=brand, quantity_ml=quantity_ml,
    )
    counted = _counted(where, min_airports).subquery()
    total = db.scalar(select(func.count()).select_from(counted)) or 0

    per_page = max(1, min(per_page, 200))
    page = max(1, page)
    order = ProductVariant.name.asc()
    if sort == "airports":
        order = counted.c.airports.desc() if not descending else counted.c.airports.asc()
    elif SORTS.get(sort) is not None:
        column = SORTS[sort]
        order = column.desc().nullslast() if descending else column.asc().nullsfirst()
    elif descending:
        order = ProductVariant.name.desc()

    rows = db.execute(
        select(ProductVariant, counted.c.airports)
        .join(counted, counted.c.pid == ProductVariant.id)
        .order_by(order, ProductVariant.id).offset((page - 1) * per_page).limit(per_page)
    ).all()

    ids = [p.id for p, _ in rows]
    places: dict[int, list[dict]] = defaultdict(list)
    if ids:
        # The listings of this page's product variants, then the newest observation of each in one
        # more query: a correlated per-row lookup is what makes a table like this crawl.
        listing_rows = db.execute(
            select(Listing.id, Listing.variant_id, Shop.iata, Shop.code, Shop.visible)
            .join(Shop, Shop.id == Listing.shop_id)
            .where(Listing.variant_id.in_(ids), cq.live_listings())
        ).all()
        latest: dict[int, Any] = {}
        if listing_rows:
            for obs in db.execute(
                select(PriceObservation)
                .where(PriceObservation.listing_id.in_([r[0] for r in listing_rows]))
                .distinct(PriceObservation.listing_id)
                .order_by(PriceObservation.listing_id, PriceObservation.observed_at.desc())
            ).scalars():
                latest[obs.listing_id] = obs
        for listing_id, variant_id, iata, code, visible in listing_rows:
            obs = latest.get(listing_id)
            places[variant_id].append({
                "iata": iata or code,
                "visible": bool(visible),
                "price": float(obs.price) if obs and obs.price is not None else None,
                "currency": obs.currency if obs else None,
                "price_usd": float(obs.price_usd) if obs and obs.price_usd is not None else None,
                "observed_at": _iso(obs.observed_at) if obs else None,
            })

    out = []
    for product, airports in rows:
        shops = sorted(places.get(product.id, []), key=lambda s: (s["price_usd"] is None, s["price_usd"] or 0))
        usd = [s["price_usd"] for s in shops if s["price_usd"] is not None]
        seen = [s["observed_at"] for s in shops if s["observed_at"]]
        out.append({
            "id": product.id, "name": product.name, "brand": product.brand,
            "category": product.category, "vertical": product.vertical,
            "size": quantity_label(product),
            "quantity_ml": product.quantity_ml, "gtin": product.gtin,
            "quantity_unit": product.quantity_unit, "quantity_state": product.quantity_state, "form": product.form,
            "has_image": bool(product.image_url), "is_exclusive": bool(product.is_exclusive),
            "airports": airports, "shops": shops,
            "cheapest_usd": min(usd) if usd else None, "dearest_usd": max(usd) if usd else None,
            "newest_price_at": max(seen) if seen else None,
        })
    return {"total": total, "page": page, "per_page": per_page, "rows": out}


def grouped(
    db: Session, *, group: str, q: str | None = None, collector: str | None = None,
    airport: str | None = None, category: str | None = None, vertical: str | None = None,
    missing: str | None = None, min_airports: int | None = None, brand: str | None = None,
    quantity_ml: float | None = None,
) -> dict[str, Any]:
    """The same filtered set, counted by one column: the quickest way to see a long tail."""
    if group not in GROUPS:
        return {"group": group, "rows": []}
    where, min_airports = _product_filter(
        db, q=q, collector=collector, airport=airport, category=category,
        vertical=vertical, missing=missing, min_airports=min_airports, brand=brand, quantity_ml=quantity_ml,
    )
    counted = _counted(where, min_airports).subquery()
    column = GROUPS[group]
    rows = db.execute(
        select(column, func.count(func.distinct(ProductVariant.id)), func.sum(counted.c.airports))
        .join(counted, counted.c.pid == ProductVariant.id)
        .group_by(column).order_by(func.count(func.distinct(ProductVariant.id)).desc()).limit(200)
    ).all()
    return {
        "group": group,
        # `key` is the raw value the filters take; `value` is what a person reads.
        "rows": [{"key": v,
                  "value": (f"{v:g}" if v is not None and group == "size" else v) or "(none)",
                  "product_variants": n, "listings": int(total or 0)} for v, n, total in rows],
    }


# --------------------------------------------------------------------------- problems

def problems(db: Session, *, limit: int = 40) -> dict[str, Any]:
    """What is wrong right now, in the order a person should look at it.

    Publication blocks first (they are hiding shops from the site this minute), then the
    duplicate queue the merge rules refused to decide, then the nightly audit's metrics
    that are over their threshold, then what ingest threw away.
    """
    blocked = verify_service.blocked_sources(db)
    names = {s.id: s.slug for s in db.scalars(select(Source))}
    blocks = [
        {"collector": names.get(source_id, str(source_id)),
         "checks": [{"id": c.id, "verdict": c.verdict, "detail": c.detail, "url": c.url,
                     "checked_at": _iso(c.checked_at)} for c in checks]}
        for source_id, checks in blocked.items()
    ]

    # The product pairs the merge rules found but would not merge on their own; the brand
    # and line levels of the queue belong to the Merge view.
    pending = db.execute(
        select(Suggestion).where(Suggestion.is_open(), Suggestion.level == "product")
        .order_by(Suggestion.score.desc()).limit(limit)
    ).scalars().all()
    pairs = []
    for candidate in pending:
        left, right = db.get(ProductVariant, candidate.left_id), db.get(ProductVariant, candidate.right_id)
        if left is None or right is None:
            continue
        pairs.append({
            "id": candidate.id, "reason": candidate.reason, "score": candidate.score,
            "why": (candidate.detail or {}).get("why"),
            "detected_at": _iso(candidate.detected_at),
            "left": {"id": left.id, "name": left.name, "brand": left.brand,
                     "size": left.quantity_ml, "gtin": left.gtin},
            "right": {"id": right.id, "name": right.name, "brand": right.brand,
                      "size": right.quantity_ml, "gtin": right.gtin},
        })
    pending_total = db.scalar(
        select(func.count(Suggestion.id)).where(Suggestion.is_open(), Suggestion.level == "product")
    ) or 0

    snapshot = db.scalar(select(AuditSnapshot).order_by(AuditSnapshot.taken_at.desc()).limit(1))
    over = []
    if snapshot and isinstance(snapshot.metrics, dict):
        for metric, threshold in audit_service.THRESHOLDS.items():
            entry = snapshot.metrics.get(metric)
            value = entry.get("value") if isinstance(entry, dict) else entry
            if isinstance(value, (int, float)) and value > threshold:
                over.append({"metric": metric, "value": value, "threshold": threshold})

    stuck = [
        {"collector": names.get(r.source_id, str(r.source_id)), "started_at": _iso(r.started_at),
         "status": r.status, "error": r.error}
        for r in db.scalars(
            select(CollectionRun).where(CollectionRun.status.notin_(("ok", "blocked")))
            .order_by(CollectionRun.started_at.desc()).limit(limit)
        )
    ]
    rejected = [
        {"stage": stage, "reason": reason, "count": n}
        for stage, reason, n in db.execute(
            select(RejectedObservation.stage, RejectedObservation.reason, func.count(RejectedObservation.id))
            .group_by(RejectedObservation.stage, RejectedObservation.reason)
            .order_by(func.count(RejectedObservation.id).desc()).limit(limit)
        )
    ]
    return {
        "publication_blocks": blocks,
        "duplicates": {"total": pending_total, "pairs": pairs},
        "audit": {"taken_at": _iso(snapshot.taken_at) if snapshot else None, "over_threshold": over},
        "runs_not_ok": stuck,
        "rejected": rejected,
    }


def runs(db: Session, *, collector: str | None = None, limit: int = 60) -> dict[str, Any]:
    """The collection history, newest first: what ran, how long it took, what it wrote."""
    names = {s.id: s.slug for s in db.scalars(select(Source))}
    stmt = select(CollectionRun).order_by(CollectionRun.started_at.desc()).limit(max(1, min(limit, 400)))
    if collector:
        source_id = db.scalar(select(Source.id).where(Source.slug == collector))
        stmt = stmt.where(CollectionRun.source_id == (source_id or -1))
    rows = []
    for run in db.scalars(stmt):
        minutes = None
        if run.started_at and run.finished_at:
            minutes = round((run.finished_at - run.started_at).total_seconds() / 60, 1)
        rows.append({
            "id": run.id, "collector": names.get(run.source_id, str(run.source_id)),
            "started_at": _iso(run.started_at), "finished_at": _iso(run.finished_at),
            "minutes": minutes, "status": run.status, "prices": run.prices_written,
            "products_seen": run.products_seen, "skipped_no_price": run.skipped_no_price,
            "skip_counts": run.skip_counts or {}, "error": run.error,
        })
    return {"runs": rows}


def live(db: Session) -> dict[str, Any]:
    """What is collecting right now, and how it is going: the live half of the collectors page.

    Rian, 19 Sep: the collectors page shows past collections; during a sweep he wants progress and
    stats as they happen. A run commits its counters every `ingest.COMMIT_EVERY` prices, so this
    is a read over `collection_runs` with the arithmetic done here: elapsed, prices per minute,
    the last 24 hours of finished runs, and every enabled source with its state. Nothing is
    stored; the page polls. A run older than `ingest.STUCK_AFTER` is shown as dead, which is
    what `backfill stuck_runs` tidies."""
    from app.services.ingest import STUCK_AFTER

    now = datetime.now(UTC)
    names = {s.id: s for s in db.scalars(select(Source))}
    running = []
    for run in db.scalars(select(CollectionRun).where(CollectionRun.status == "running").order_by(CollectionRun.started_at)):
        started = run.started_at
        elapsed = (now - started).total_seconds() if started else 0.0
        minutes = elapsed / 60
        source = names.get(run.source_id)
        running.append({
            "id": run.id, "collector": source.slug if source else str(run.source_id),
            "started_at": _iso(started), "elapsed_minutes": round(minutes, 1),
            "products_seen": run.products_seen, "prices": run.prices_written,
            "skipped_no_price": run.skipped_no_price, "skip_counts": run.skip_counts or {},
            "prices_per_minute": round(run.prices_written / minutes, 1) if minutes > 0.5 else None,
            "dead": bool(started) and (now - started) > STUCK_AFTER,
        })
    since = now - timedelta(hours=24)
    recent = []
    for run in db.scalars(select(CollectionRun).where(CollectionRun.status != "running", CollectionRun.started_at >= since)
                          .order_by(CollectionRun.started_at.desc())):
        minutes = round((run.finished_at - run.started_at).total_seconds() / 60, 1) if run.started_at and run.finished_at else None
        source = names.get(run.source_id)
        recent.append({"id": run.id, "collector": source.slug if source else str(run.source_id), "started_at": _iso(run.started_at),
                       "finished_at": _iso(run.finished_at), "minutes": minutes, "status": run.status, "prices": run.prices_written,
                       "products_seen": run.products_seen, "skipped_no_price": run.skipped_no_price,
                       "skip_counts": run.skip_counts or {}, "error": run.error})
    last_by_source: dict[int, CollectionRun] = {}
    for run in db.scalars(select(CollectionRun).order_by(CollectionRun.started_at.desc())):
        if run.source_id not in last_by_source:
            last_by_source[run.source_id] = run
    running_ids = {r["collector"] for r in running}
    sources = []
    for source in sorted(names.values(), key=lambda s: s.slug):
        last = last_by_source.get(source.id)
        sources.append({"slug": source.slug, "enabled": source.enabled, "running": source.slug in running_ids,
                        "last_status": last.status if last else None, "last_finished_at": _iso(last.finished_at) if last else None,
                        "last_prices": last.prices_written if last else None, "last_error": last.error if last else None,
                        "ran_in_last_24h": bool(last and last.started_at and last.started_at >= since)})
    prices_24h = sum(r["prices"] for r in recent) + sum(r["prices"] for r in running)
    return {"now": _iso(now), "running": running, "recent": recent, "sources": sources,
            "totals": {"running": len(running), "finished_24h": len(recent), "ok_24h": sum(1 for r in recent if r["status"] == "ok"),
                       "failed_24h": sum(1 for r in recent if r["status"] not in ("ok", "skipped")), "prices_24h": prices_24h,
                       "sources_enabled": sum(1 for s in sources if s["enabled"]),
                       "sources_done_24h": sum(1 for s in sources if s["ran_in_last_24h"] and not s["running"])}}


# --------------------------------------------------------------------------- listings

#: What "differs from collected" can ask for: any field, or one of the three compared.
DIFFERS_CHOICES = ("any", "brand", "name", "quantity", "size")
_FLAT_RE = re.compile(r"[^a-z0-9]+")
def _flat(text: str | None) -> str:
    folded = unicodedata.normalize("NFKD", text or "").encode("ascii", "ignore").decode().lower()
    return " ".join(_FLAT_RE.sub(" ", folded).split())


def tile_differs(tile: dict[str, Any], ours: dict[str, Any]) -> list[str]:
    """Where what we made of a listing disagrees with the tile it came from.

    Brands compare on the fold key (a spelling is not a disagreement; another brand is);
    names on the flattened text; sizes exactly, both known. A field the tile does not
    carry never differs: silence is not a contradiction.
    """
    out = []
    if tile.get("brand") and ours.get("brand") and brand_key(tile["brand"]) != brand_key(ours["brand"]):
        out.append("brand")
    if tile.get("name") and ours.get("name") and _flat(tile["name"]) != _flat(ours["name"]):
        out.append("name")
    if tile.get("size_ml") and ours.get("quantity_ml") and int(tile["size_ml"]) != int(ours["quantity_ml"]):
        out.append("size")
    return out


def _collector_of_code() -> dict[str, str]:
    return {spec.code: slug for slug, collector in COLLECTORS.items() for spec in collector.shops()}


def _newest_raw(db: Session, listing_ids: list[int] | None, *, slim: bool = False) -> dict[int, tuple]:
    """Each listing's newest raw record: (parser_version, payload, created_at). `slim` drops the
    tile's HTML, the one bulky key, when every listing is being read at once; None reads all."""
    payload = RawRecord.payload.op("-")("tile_html") if slim else RawRecord.payload
    stmt = (
        select(RawRecord.listing_id, RawRecord.parser_version, payload, RawRecord.created_at)
        .distinct(RawRecord.listing_id)
        .order_by(RawRecord.listing_id, RawRecord.id.desc())
    )
    if listing_ids is not None:
        if not listing_ids:
            return {}
        stmt = stmt.where(RawRecord.listing_id.in_(listing_ids))
    return {lid: (version, body, at) for lid, version, body, at in db.execute(stmt).all()}


def _brands(db: Session) -> dict[int, Brand]:
    return {b.id: b for b in db.scalars(select(Brand))}


def _brand_of(brands: dict[int, Brand], brand_id: int | None) -> Brand | None:
    """The listed_brand row a product's fold points at, its alias followed to the brand."""
    return resolve_alias(brands, brand_id)


def _listing_conditions(db: Session, *, q: str | None, collector: str | None, airport: str | None,
                        brand: str | None, ignored: str | None = None) -> list:
    """`ignored`: None hides ignored listings (the default everywhere), "only" shows just
    them, "any" shows both."""
    where = []
    if ignored == "only":
        where.append(Listing.ignored_at.isnot(None))
    elif ignored != "any":
        where.append(cq.live_listings())
    if q:
        like = f"%{q.strip().lower()}%"
        where.append(or_(func.lower(ProductVariant.name).like(like), func.lower(ProductVariant.brand).like(like)))
    if brand:
        where.append(func.lower(ProductVariant.brand).like(f"%{brand.strip().lower()}%"))
    shop_ids = _scope_shop_ids(db, collector, airport)
    if shop_ids is not None:
        where.append(Listing.shop_id.in_(shop_ids))
    return where


_LISTING_ORDER = (ProductVariant.brand.asc().nullslast(), ProductVariant.name.asc(), Shop.code.asc(), Listing.id.asc())


def _lines_of(db: Session, product_line_ids: set[int]) -> dict[int, ProductLine]:
    """The line rows of a page's product variants and the canonical rows their aliases point at."""
    rows: dict[int, ProductLine] = {}
    wanted = {i for i in product_line_ids if i}
    for _ in range(4):
        missing = wanted - set(rows)
        if not missing:
            break
        for row in db.scalars(select(ProductLine).where(ProductLine.id.in_(list(missing)))):
            rows[row.id] = row
        wanted |= {row.alias_of_id for row in rows.values() if row.alias_of_id}
    return rows


def _ours_of(product: ProductVariant, brand: Brand | None, line: ProductLine | None = None) -> dict[str, Any]:
    return {
        "name": product.name, "brand": product.brand,
        "brand": brand.name if brand else product.brand,
        "brand_slug": brand.slug if brand else None,
        "line": line.name if line else None,
        "line_slug": line.slug if line else None,
        "attribute": (product.attributes or {}).get("concentration"),
        "size": quantity_label(product),
        "quantity_ml": product.quantity_ml, "gtin": product.gtin,
    }


def listings(db: Session, **kw) -> dict[str, Any]:
    """The listings table: delegated to `services/listings_table.query` (Stream L), one SQL
    statement with every column sortable and filterable; kept here so the route's older
    callers and tests find it under the same name."""
    from app.services import listings_table

    return listings_table.query(db, **kw)


def _listing_rows(db: Session, ids: list[int], brands: dict[int, Brand], raws: dict | None) -> list[dict[str, Any]]:
    if not ids:
        return []
    found = {
        listing.id: (listing, product, shop)
        for listing, product, shop in db.execute(
            select(Listing, ProductVariant, Shop)
            .join(ProductVariant, ProductVariant.id == Listing.variant_id)
            .join(Shop, Shop.id == Listing.shop_id)
            .where(Listing.id.in_(ids))
        ).all()
    }
    raws = _newest_raw(db, ids) if raws is None else raws
    lines = _lines_of(db, {product.product_line_id for _, product, _ in found.values()})
    latest: dict[int, PriceObservation] = {}
    for obs in db.execute(
        select(PriceObservation).where(PriceObservation.listing_id.in_(ids))
        .distinct(PriceObservation.listing_id)
        .order_by(PriceObservation.listing_id, PriceObservation.observed_at.desc())
    ).scalars():
        latest[obs.listing_id] = obs
    collector_of = _collector_of_code()
    out = []
    for listing_id in ids:
        if listing_id not in found:
            continue
        listing, product, shop = found[listing_id]
        raw = raws.get(listing_id)
        tile = None
        if raw is not None:
            tile = collected_tile(raw[0], raw[1], currency=shop.currency)
            tile["seen_at"] = _iso(raw[2])
            tile["parser"] = raw[0]
        ours = _ours_of(product, _brand_of(brands, product.brand_id), resolve_alias(lines, product.product_line_id))
        obs = latest.get(listing_id)
        ours.update({
            "price": float(obs.price) if obs and obs.price is not None else None,
            "currency": obs.currency if obs else None,
            "price_usd": float(obs.price_usd) if obs and obs.price_usd is not None else None,
            "observed_at": _iso(obs.observed_at) if obs else None,
        })
        out.append({
            "listing_id": listing.id, "variant_id": product.id,
            "airport": shop.iata or shop.code, "shop": shop.code,
            "collector": collector_of.get(shop.code), "source_sku": listing.source_sku, "url": listing.url,
            "pinned_variant_id": listing.pinned_variant_id, "ignored_at": _iso(listing.ignored_at),
            "ignore_reason": listing.ignore_reason,
            "collected": tile, "ours": ours,
            "differs": tile_differs(tile, ours) if tile else [],
        })
    return out