"""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 urllib.parse import urlsplit

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

from app.models import (
    Account,
    Brand,
    CollectionRun,
    Listing,
    Shop,
    Suggestion,
    PriceObservation,
    ProductVariant,
    ProductLine,
    RawRecord,
    Source,
)
from app.models.quality import AuditSnapshot, RejectedObservation, VerificationCheck
from app.models.schemas import (
    ControlOut,
    LiveCollector,
    LiveEvent,
    LiveHoursAirport,
    LiveHoursCollector,
    LiveLastRun,
    LiveMemory,
    LiveOut,
    LivePace,
    LiveRecentRun,
    LiveRefused,
    LiveRun,
    LiveStartBlocked,
    LiveTotals,
)
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 _aware(value: datetime | None) -> datetime | None:
    """Postgres hands these back aware; SQLite does not, so comparisons in `live` must not care."""
    return value.replace(tzinfo=UTC) if value is not None and value.tzinfo is None else value


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, "neutral": 0, "recent": [],
                "refused_since": _iso(refused_since)}
    ordered = sorted(runs, key=lambda r: r.started_at or datetime.min.replace(tzinfo=UTC))
    # Four kinds: ok, refused (they said no), failed (our side), and neutral (a stop from the page,
    # a skip, or a run still walking: neither a success nor a fault; the strip draws them grey,
    # Stream AW4). A `running` row counted as failed once told a person five hosts had closed
    # their doors mid-sweep ("worked before, refused now") while the collectors were on them.
    kinds = ["ok" if r.status == "ok" else "refused" if r.status == "blocked"
             else "neutral" if r.status in ("stopped", "skipped", "running") else "failed" for r in ordered]
    ok, refused, failed = kinds.count("ok"), kinds.count("refused"), kinds.count("failed")
    bad = refused + 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 and not bad:
        story = "clean every run"
    elif not ok and bad:
        story = "never once succeeded"
    elif not ok:
        story = "nothing finished yet"
    elif kinds[-1] == "ok":
        story = "working now, after trouble"
    elif kinds[-1] == "neutral":
        story = "on and off"
    else:
        story = "worked before, refused now"
    # 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,
        "neutral": kinds.count("neutral"), "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}


# --------------------------------------------------------------------------- live (Stream AW4)
#
# The control room's read: one typed `LiveOut` per poll, every figure computed here from the
# run rows the collectors commit as they go, the container's cgroup and `/proc`. A GET marks
# nothing. The nine state words come from one pure `state_of`, and `start_blocked` is exactly
# what `POST /start` would answer, because the SPA's `ApiError` carries only a code.

#: A page word, never a row state: a running row whose heartbeat is older than
#: max(STALL_FLOOR, STALL_DELAYS x its pace) is called stalled. Ten minutes is five beats of the
#: slowest pace met (60 s) and more; a render caps at 90 s, so a slow host is never "stalled".
STALL_FLOOR = timedelta(minutes=10)
STALL_DELAYS = 5
#: No percent and no ETA before this many units are done: one unit says nothing about the rate.
ETA_MIN_DONE = 3
#: A running run is never shown at 100: the estimate came from an earlier walk.
PERCENT_CAP_RUNNING = 99
#: Audit rows shown per collector (the six page actions).
EVENTS_PER_COLLECTOR = 5
#: The live read's budget in SQL statements, tested: the page polls every few seconds.
LIVE_MAX_STATEMENTS = 12

LIVE_STATES = ("dead", "stalled", "stopping", "paused", "pausing", "running")
STATES = LIVE_STATES + ("off", "refused", "idle")


def percent_done(done: int, expected: int | None, *, finished: bool) -> int | None:
    """None without an expected total or under `ETA_MIN_DONE`; capped at 99 while running (an
    estimate is a previous walk's count, and 100 is a claim only a finished run may make)."""
    if expected is None or expected <= 0 or done < ETA_MIN_DONE:
        return None
    raw = int(done * 100 // expected)
    return min(100, raw) if finished else min(PERCENT_CAP_RUNNING, raw)


def eta_seconds(done: int, expected: int | None, elapsed: float, *, finished: bool) -> float | None:
    """`elapsed * (expected - done) / done`, None on the same conditions as the percent and once
    the run is over. A duration, never a clock time: the page says "about 38 min to go"."""
    if finished or expected is None or expected <= 0 or done < ETA_MIN_DONE or elapsed <= 0:
        return None
    return max(0.0, elapsed * (expected - done) / done)


def _span(value: timedelta | float) -> str:
    seconds = max(0, int(value.total_seconds() if isinstance(value, timedelta) else value))
    if seconds < 60:
        return f"{seconds} s"
    minutes = seconds // 60
    if minutes < 60:
        return f"{minutes} min"
    hours, minutes = divmod(minutes, 60)
    return f"{hours} h {minutes:02d}"


def _when(value: datetime | None) -> str:
    return _aware(value).astimezone(UTC).strftime("%d %b %H:%M") if value else "an unknown time"


def state_of(source: Any, run: CollectionRun | None, last: CollectionRun | None, refused_since: datetime | None,
             now: datetime, alive: bool | None, *, boot: Any = ...) -> tuple[str, str]:
    """One of the nine words and the sentence beside it, from the rows alone (D6). `alive` is what
    `/proc` says about the run's pid (None where unreadable). Pure: a GET marks nothing. The
    order is the table's: a dead row before a stalled one, a stall before any control word (a
    stalled run says stalled, whatever was pressed), and the control words before "running"."""
    from app.services.collectors import control
    from app.services.ingest import STUCK_AFTER, STUCK_HEARTBEAT, is_stuck

    if run is not None:
        started = _aware(run.started_at)
        beat = _aware(run.heartbeat_at)
        if is_stuck(run, now) if boot is ... else is_stuck(run, now, boot):
            if beat is not None and now - beat > STUCK_HEARTBEAT:
                return "dead", f"dead: no heartbeat for {_span(now - beat)}"
            if started is not None and now - started > STUCK_AFTER:
                return "dead", f"dead: running for {_span(now - started)}, past the {_span(STUCK_AFTER)} timer"
            return "dead", "dead: began before this container"
        delay = float(source.delay_seconds or 0.0)
        stall_after = max(STALL_FLOOR, timedelta(seconds=STALL_DELAYS * delay))
        reference = beat or started
        age = (now - reference) if reference is not None else timedelta(0)
        if alive is False:
            return "stalled", "stalled: process gone"
        if age > stall_after:
            silence = f"no heartbeat for {_span(age)}" if beat is not None else f"no heartbeat since it began {_span(age)} ago"
            return "stalled", f"stalled: {silence} (pace {delay:g} s, {_span(stall_after)} expected)"
        effective = control.effective_control(source.control, source.control_set_at, run.started_at)
        by = source.control_set_by or "the page"
        set_at = _aware(source.control_set_at)
        if not source.enabled:
            return "stopping", "stopping: switched off (the kill switch); waits for the request in flight"
        if effective == "stop":
            asked = f" by {by} {_span(now - set_at)} ago" if set_at else ""
            return "stopping", f"stopping{asked}; waits for the request in flight"
        if effective == "pause":
            if beat is not None and set_at is not None and beat >= set_at:
                return "paused", f"paused {_span(now - set_at)} by {by} · holds approvals while paused"
            boundary = max(control.CONTROL_POLL, delay - age.total_seconds())
            return "pausing", f"pausing: next request boundary in about {int(round(boundary))} s"
        return "running", f"collecting since {_when(started)}: {run.prices_written} prices, {run.requests_made} requests"
    if not source.enabled:
        return "off", "switched off (the kill switch)"
    if last is not None and last.status == "blocked":
        return "refused", f"refused since {_when(last.started_at)}: {(last.error or 'refused')[:160]}"
    if refused_since is not None and (last is None or last.started_at is None or _aware(last.started_at) < _aware(refused_since)):
        return "refused", f"refused since {_when(refused_since)}: our declared identity was refused (a verification check)"
    if last is None:
        return "idle", "never run"
    took = ""
    if last.started_at and last.finished_at:
        took = f" in {_span(_aware(last.finished_at) - _aware(last.started_at))}"
    return "idle", f"last: {last.status} {_when(last.started_at)}, {last.prices_written} prices{took}"


def _unsaved_source(slug: str, collector: Any) -> Source:
    """A collector the sources table has never seen, as the row a first run would write; never
    added to the session (a GET writes nothing)."""
    return Source(slug=slug, name=collector.retailer_name, enabled=True, delay_seconds=1.0, control="run",
                  mode="discover", identity_mode="declared")


def _live_run_out(run: CollectionRun, source: Any, now: datetime, alive: bool | None, rss: int | None,
                  effective: str) -> LiveRun:
    started = _aware(run.started_at)
    beat = _aware(run.heartbeat_at)
    elapsed = (now - started).total_seconds() if started else 0.0
    minutes = elapsed / 60
    finished = run.status != "running"
    if run.mode == "recheck":
        done = run.existing_checked + run.existing_missing
        kind = "exact" if run.expected_total is not None else None
    else:
        done = run.products_seen
        kind = "estimate" if run.expected_total is not None else None
    return LiveRun(
        id=run.id, status=run.status, mode=run.mode, limit_n=run.limit_n, started_at=started, finished_at=_aware(run.finished_at),
        heartbeat_at=beat, heartbeat_age_seconds=(now - beat).total_seconds() if beat else None,
        elapsed_seconds=elapsed, requests_made=run.requests_made,
        requests_per_minute=round(run.requests_made / minutes, 1) if elapsed >= 30 else None,
        pid=run.pid, alive=alive, rss_bytes=rss, started_by=run.started_by, stopped_by=run.stopped_by,
        stop_requested_at=_aware(source.control_set_at) if effective == "stop" else None,
        products_seen=run.products_seen, prices_written=run.prices_written,
        prices_per_minute=round(run.prices_written / minutes, 1) if minutes > 0.5 else None,
        skipped_no_price=run.skipped_no_price, skip_counts=run.skip_counts or {}, error=run.error,
        existing_checked=run.existing_checked, existing_changed=run.existing_changed, existing_missing=run.existing_missing,
        existing_missing_measured=run.mode == "recheck" or (run.limit_n is None and run.status == "ok"),
        new_found=run.new_found, new_brands=run.new_brands, new_lines=run.new_lines, new_variants=run.new_variants,
        expected_total=run.expected_total, expected_kind=kind, done=done,
        percent=percent_done(done, run.expected_total, finished=finished),
        over_expected=run.expected_total is not None and done > run.expected_total,
        eta_seconds=eta_seconds(done, run.expected_total, elapsed, finished=finished),
    )


def _last_run_out(run: CollectionRun) -> LiveLastRun:
    minutes = None
    if run.started_at and run.finished_at:
        minutes = round((_aware(run.finished_at) - _aware(run.started_at)).total_seconds() / 60, 1)
    return LiveLastRun(
        id=run.id, status=run.status, mode=run.mode, limit_n=run.limit_n, started_at=_aware(run.started_at),
        finished_at=_aware(run.finished_at), minutes=minutes, 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,
        stopped_by=run.stopped_by, started_by=run.started_by,
        existing_checked=run.existing_checked, existing_changed=run.existing_changed, existing_missing=run.existing_missing,
        existing_missing_measured=run.mode == "recheck" or (run.limit_n is None and run.status == "ok"),
        new_found=run.new_found, new_brands=run.new_brands, new_lines=run.new_lines, new_variants=run.new_variants,
    )


def _hours_group(db: Session, now: datetime) -> tuple[list[LiveHoursCollector], list[LiveHoursAirport], list[str]]:
    """The opening hours collectors as their own group: which airports each covers and what the
    airport currently shows (a hand row outranks a collected one, as the store rules), every
    public airport whose hours were typed in, and every one with no hours at all."""
    from app.models.hours import AirportHours
    from app.services.hours.registry import HOURS_COLLECTORS
    from app.services.ingest import _has_table

    shops = db.execute(select(Shop.id, Shop.iata, Shop.visible, Shop.is_catalogue_only)).all()
    by_iata: dict[str, list[int]] = defaultdict(list)
    for shop in shops:
        if shop.iata:
            by_iata[shop.iata].append(shop.id)
    public = sorted({s.iata for s in shops if s.iata and s.visible and not s.is_catalogue_only})
    rows: dict[int, dict[str, tuple[AirportHours, str | None]]] = defaultdict(dict)
    if _has_table(db, "airport_hours"):
        rn = func.row_number().over(
            partition_by=(AirportHours.shop_id, AirportHours.source_kind),
            order_by=(AirportHours.observed_at.desc(), AirportHours.id.desc()),
        ).label("rn")
        ranked = select(AirportHours.id.label("id"), rn).subquery()
        for row, username in db.execute(
            select(AirportHours, Account.username)
            .join(ranked, ranked.c.id == AirportHours.id)
            .outerjoin(Account, Account.id == AirportHours.entered_by_id)
            .where(ranked.c.rn == 1)
        ):
            rows[row.shop_id][row.source_kind] = (row, username)

    def current(iata: str) -> LiveHoursAirport:
        for kind in ("hand", "collected"):
            found = [rows[i][kind] for i in by_iata.get(iata, []) if kind in rows.get(i, {})]
            if found:
                row, username = max(found, key=lambda pair: (_aware(pair[0].observed_at), pair[0].id))
                host = (urlsplit(row.source_url).hostname or None) if row.source_url else None
                return LiveHoursAirport(iata=iata, kind=kind, observed_at=_aware(row.observed_at),
                                        source_host=host, entered_by=username)
        return LiveHoursAirport(iata=iata, kind="none")

    collectors_out = [
        LiveHoursCollector(slug=c.slug, operator=c.operator, host=urlsplit(c.homepage).hostname or None,
                           airports=[current(iata) for iata in c.airports])
        for c in HOURS_COLLECTORS
    ]
    current_public = [current(iata) for iata in public]
    return (collectors_out, [a for a in current_public if a.kind == "hand"],
            [a.iata for a in current_public if a.kind == "none"])


def live(db: Session, *, now: datetime | None = None, cgroup: str | None = None, proc: str | None = None) -> LiveOut:
    """What is collecting right now and how it is going: the live half of the collectors page,
    typed (D8). Rian, 19 Sep: during a sweep he wants progress and stats as they happen, and the
    page actions in place of `docker exec`. Every statement is portable (window functions, no
    dialect branch) and the whole read stays under `LIVE_MAX_STATEMENTS`; then `/sys` and
    `/proc`. A run `ingest.is_stuck` calls dead is shown dead, as the approval gate sees it (the
    18 Sep orphans were drawn as live collections for the 21 hours the timer had left)."""
    from app.models import AuditLog, Retailer
    from app.services import procinfo
    from app.services.collectors import control
    from app.services.ingest import _has_table, is_stuck

    now = now or datetime.now(UTC)
    cgroup = cgroup or procinfo.CGROUP
    proc = proc or procinfo.PROC
    since = now - timedelta(hours=24)

    sources = {s.slug: s for s in db.scalars(select(Source))}
    running_by_source: dict[int, CollectionRun] = {}
    for run in db.scalars(select(CollectionRun).where(CollectionRun.status == "running")
                          .order_by(CollectionRun.started_at.desc(), CollectionRun.id.desc())):
        running_by_source.setdefault(run.source_id, run)
    rn = func.row_number().over(
        partition_by=CollectionRun.source_id, order_by=(CollectionRun.started_at.desc(), CollectionRun.id.desc())
    ).label("rn")
    ranked = select(CollectionRun.id.label("id"), rn).where(CollectionRun.status != "running").subquery()
    last_by_source = {
        run.source_id: run
        for run in db.scalars(select(CollectionRun).join(ranked, ranked.c.id == CollectionRun.id).where(ranked.c.rn == 1))
    }
    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_table(db, "verification_checks") else {}
    recent_rows = list(db.scalars(
        select(CollectionRun).where(CollectionRun.status != "running", CollectionRun.started_at >= since)
        .order_by(CollectionRun.started_at.desc())
    ))
    held = {
        (retailer, code): int(n)
        for retailer, code, n in db.execute(
            select(Retailer.slug, Shop.code, func.count(Listing.id))
            .join(Shop, Shop.retailer_id == Retailer.id).join(Listing, Listing.shop_id == Shop.id)
            .where(Listing.url.isnot(None)).group_by(Retailer.slug, Shop.code)
        )
    }
    events: dict[str, list[LiveEvent]] = defaultdict(list)
    if _has_table(db, "audit_log"):
        rn_e = func.row_number().over(partition_by=AuditLog.entity_key, order_by=(AuditLog.at.desc(), AuditLog.id.desc())).label("rn")
        ranked_e = select(AuditLog.id.label("id"), rn_e).where(AuditLog.entity_type == "source").subquery()
        for row, username in db.execute(
            select(AuditLog, Account.username).join(ranked_e, ranked_e.c.id == AuditLog.id)
            .outerjoin(Account, Account.id == AuditLog.account_id)
            .where(ranked_e.c.rn <= EVENTS_PER_COLLECTOR)
            .order_by(AuditLog.entity_key, AuditLog.at.desc(), AuditLog.id.desc())
        ):
            events[row.entity_key or ""].append(LiveEvent(at=_aware(row.at), action=row.action, by=username, detail=row.detail))
    hours, hours_by_hand, hours_missing = _hours_group(db, now)

    raw_memory = procinfo.container_memory(cgroup)
    freeze = procinfo.freeze_state(control.freeze_marker(), now)
    memory = LiveMemory(
        used_bytes=raw_memory.used_bytes, limit_bytes=raw_memory.limit_bytes, shed_at_bytes=procinfo.SHED_AT_BYTES,
        start_refused_above_bytes=procinfo.START_HEADROOM_BYTES, readable=raw_memory.readable,
        frozen=freeze.frozen, frozen_stamp=freeze.stamp, frozen_stale=freeze.stale,
    )

    out: list[LiveCollector] = []
    for slug, collector in sorted(COLLECTORS.items()):
        source = sources.get(slug) or _unsaved_source(slug, collector)
        run = running_by_source.get(source.id) if source.id is not None else None
        last = last_by_source.get(source.id) if source.id is not None else None
        refused_at = refused_since.get(source.id) if source.id is not None else None
        # A row with no pid (a run from before the hook) is unknown to /proc, never "gone".
        alive = procinfo.alive(run.pid, slug, proc) if run is not None and run.pid else None
        rss = procinfo.rss_of(run.pid, proc) if run is not None and alive else None
        state, detail = state_of(source, run, last, refused_at, now, alive)
        effective = control.effective_control(source.control, source.control_set_at, run.started_at) if run else "run"
        held_n = sum(held.get((collector.retailer_slug, spec.code), 0) for spec in collector.shops())
        refusal = control.start_refusal(
            db, source, collector, source.mode, None, raw_memory, now,
            last=last, running=run, refused_since=refused_at, prefetched=True, held_listings=held_n,
        )
        pace = control.pace_of(source, collector)
        refused = None
        if state == "refused":
            since_at = last.started_at if last is not None and last.status == "blocked" else refused_at
            refused = LiveRefused(since=_aware(since_at), why=detail)
        out.append(LiveCollector(
            slug=slug, name=source.name, platform=platform_of(slug), enabled=bool(source.enabled),
            rendered=control.is_rendered(collector), state=state, state_detail=detail,
            control=source.control or "run", control_set_by=source.control_set_by, control_set_at=_aware(source.control_set_at),
            effective_control=effective, mode_default=source.mode or "discover",
            recheck_supported=control.recheck_refusal(collector) is None, held_listings=held_n,
            run=_live_run_out(run, source, now, alive, rss, effective) if run is not None else None,
            pace=LivePace(**pace.as_dict()), last=_last_run_out(last) if last is not None else None,
            refused=refused, start_blocked=LiveStartBlocked(code=refusal.code, summary=refusal.summary) if refusal else None,
            holds_approvals=run is not None and not is_stuck(run, now, heartbeat=False),
            events=events.get(slug, []),
        ))
    out.sort(key=lambda c: (0 if c.state in LIVE_STATES else 1, c.slug))

    names = {s.id: s.slug for s in sources.values()}
    recent = [
        LiveRecentRun(
            id=run.id, collector=names.get(run.source_id, str(run.source_id)), mode=run.mode,
            started_at=_aware(run.started_at), finished_at=_aware(run.finished_at),
            minutes=round((_aware(run.finished_at) - _aware(run.started_at)).total_seconds() / 60, 1)
            if run.started_at and run.finished_at else None,
            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, stopped_by=run.stopped_by,
        )
        for run in recent_rows
    ]
    counts = {state: sum(1 for c in out if c.state == state) for state in STATES}
    totals = LiveTotals(
        **counts,
        prices_24h=sum(r.prices for r in recent) + sum(c.run.prices_written for c in out if c.run is not None),
        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 in ("error", "blocked")),
    )
    return LiveOut(now=now, memory=memory, collectors=out, hours=hours, hours_by_hand=hours_by_hand,
                   hours_missing=hours_missing, recent=recent, totals=totals)


def control_out(db: Session, slug: str, now: datetime, *, changed: bool = True, started: bool = False,
                pid: int | None = None) -> ControlOut:
    """What a control route returns: the collector as the live read sees it after the write."""
    found = next((c for c in live(db, now=now).collectors if c.slug == slug), None)
    if found is None:
        raise KeyError(slug)
    return ControlOut(**found.model_dump(), changed=changed, started=started, pid=pid)


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