"""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 html
import re
import unicodedata
from collections import defaultdict
from datetime import UTC, datetime
from typing import Any

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

from app.models import (
    Brand,
    CollectionRun,
    Listing,
    Location,
    MergeCandidate,
    PriceObservation,
    Product,
    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.changi import channel_price as changi_price
from app.services.collectors.registry import COLLECTORS
from app.services.lines import resolve_alias
from app.services.normalize import brand_key, parse_size, parse_size_ml, size_ml_of

#: 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 _iso(value: datetime | None) -> str | None:
    return value.astimezone(UTC).isoformat(timespec="minutes") if value else None


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

def _story(runs: list[CollectionRun]) -> 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.
    """
    if not runs:
        return {"story": "never run", "runs": 0, "ok": 0, "refused": 0, "failed": 0, "recent": []}
    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"
    return {
        "story": story, "runs": len(kinds), "ok": ok, "refused": refused, "failed": failed,
        "stale_days": stale_days,
        "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)

    by_code = {loc.code: loc for loc in db.scalars(select(Location))}
    counts = {
        loc_id: (listings, products)
        for loc_id, listings, products in db.execute(
            select(Listing.location_id, func.count(Listing.id), func.count(func.distinct(Listing.product_id)))
            .group_by(Listing.location_id)
        )
    }
    newest = {
        loc_id: at
        for loc_id, at in db.execute(
            select(Listing.location_id, func.max(PriceObservation.observed_at))
            .join(PriceObservation, PriceObservation.listing_id == Listing.id)
            .group_by(Listing.location_id)
        )
    }
    blocked_locations = set(cq.blocked_location_ids(db))

    rows = []
    for slug, collector in sorted(COLLECTORS.items()):
        source = sources.get(slug)
        specs = list(collector.locations())
        places, listings_n, products_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
            products_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_locations),
                "listings": l_n, "products": 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,
            "locations": places,
            "airports": [p["iata"] for p in places if p["iata"]],
            "listings": listings_n,
            "products": products_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 []),
            "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 _alive():
    return Product.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(Product.id)).where(alive))
    merged = one(select(func.count(Product.id)).where(Product.merged_into_id.isnot(None)))
    priced = one(
        select(func.count(func.distinct(Listing.product_id)))
        .join(Product, Product.id == Listing.product_id).where(alive)
    )

    # Products by how many airports stock them: the comparison that is the whole point.
    per_product = (
        select(
            Listing.product_id.label("pid"),
            func.count(func.distinct(Listing.location_id)).label("n"),
        )
        .join(Product, Product.id == Listing.product_id)
        .where(alive)
        .group_by(Listing.product_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(Product.id)).where(and_(alive, condition)))

    verticals = [
        {"vertical": v or "(none)", "products": n}
        for v, n in db.execute(
            select(Product.vertical, func.count(Product.id)).where(alive)
            .group_by(Product.vertical).order_by(func.count(Product.id).desc())
        )
    ]
    categories = [
        {"category": c or "(none)", "products": n}
        for c, n in db.execute(
            select(Product.category, func.count(Product.id)).where(alive)
            .group_by(Product.category).order_by(func.count(Product.id).desc())
        )
    ]
    sizes = [
        {"size": f"{v:g} {u}" if v and u else "(unknown)", "products": n}
        for v, u, n in db.execute(
            select(Product.size_value, Product.size_unit, func.count(Product.id)).where(alive)
            .group_by(Product.size_value, Product.size_unit)
            .order_by(func.count(Product.id).desc()).limit(25)
        )
    ]
    return {
        "products": {
            "total": total, "priced": priced, "merged_away": merged,
            "comparable": comparable, "single_airport": spread.get(1, 0),
            "exclusives": one(select(func.count(Product.id)).where(and_(alive, Product.is_exclusive.is_(True)))),
        },
        "brands": {
            "total": one(select(func.count(Brand.id))),
            "houses": one(select(func.count(Brand.id)).where(Brand.canonical_id.is_(None))),
            "aliases": one(select(func.count(Brand.id)).where(Brand.canonical_id.isnot(None))),
        },
        "have": {
            "image": missing(Product.image_url.isnot(None)),
            "gtin": missing(Product.gtin.isnot(None)),
            "size": missing(Product.size_ml.isnot(None)),
            "category": missing(Product.category.isnot(None)),
            "brand": missing(Product.brand_id.isnot(None)),
        },
        "missing": {
            "image": missing(Product.image_url.is_(None)),
            "gtin": missing(Product.gtin.is_(None)),
            "size": missing(Product.size_ml.is_(None)),
            "category": missing(Product.category.is_(None)),
            "brand": missing(Product.brand_id.is_(None)),
        },
        "airport_spread": [{"airports": n, "products": 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))),
    }


# --------------------------------------------------------------------------- products

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


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, size_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(Product.name).like(like), func.lower(Product.brand).like(like)))
    if category:
        where.append(Product.category == category)
    if vertical:
        where.append(Product.vertical == vertical)
    if brand:
        where.append(Product.brand == brand)
    if size_ml is not None:
        where.append(Product.size_ml == size_ml)
    if missing in MISSING_CHOICES:
        column = {"image": Product.image_url, "size": Product.size_ml, "gtin": Product.gtin,
                  "category": Product.category, "brand": Product.brand_id}[missing]
        where.append(column.is_(None))

    location_ids = _scope_location_ids(db, collector, airport)
    if location_ids is not None:
        where.append(Product.id.in_(select(Listing.product_id).where(Listing.location_id.in_(location_ids))))
    return where, min_airports


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


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


def products(
    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, size_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, size_ml=size_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 = Product.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 = Product.name.desc()

    rows = db.execute(
        select(Product, counted.c.airports)
        .join(counted, counted.c.pid == Product.id)
        .order_by(order, Product.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 products, 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.product_id, Location.iata, Location.code, Location.visible)
            .join(Location, Location.id == Listing.location_id)
            .where(Listing.product_id.in_(ids))
        ).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, product_id, iata, code, visible in listing_rows:
            obs = latest.get(listing_id)
            places[product_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": f"{product.size_value:g} {product.size_unit}" if product.size_value and product.size_unit else None,
            "size_ml": product.size_ml, "gtin": product.gtin,
            "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,
    size_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, size_ml=size_ml,
    )
    counted = _counted(where, min_airports).subquery()
    column = GROUPS[group]
    rows = db.execute(
        select(column, func.count(func.distinct(Product.id)), func.sum(counted.c.airports))
        .join(counted, counted.c.pid == Product.id)
        .group_by(column).order_by(func.count(func.distinct(Product.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} ml" if isinstance(v, (int, float)) and group == "size" else v) or "(none)",
                  "products": 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(MergeCandidate).where(MergeCandidate.decision.is_(None), MergeCandidate.level == "product")
        .order_by(MergeCandidate.score.desc()).limit(limit)
    ).scalars().all()
    pairs = []
    for candidate in pending:
        left, right = db.get(Product, candidate.left_id or candidate.product_id), db.get(Product, candidate.right_id or candidate.candidate_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.size_ml, "gtin": left.gtin},
            "right": {"id": right.id, "name": right.name, "brand": right.brand,
                      "size": right.size_ml, "gtin": right.gtin},
        })
    pending_total = db.scalar(
        select(func.count(MergeCandidate.id)).where(MergeCandidate.decision.is_(None), MergeCandidate.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}


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

#: What "differs from collected" can ask for: any field, or one of the three compared.
DIFFERS_CHOICES = ("any", "brand", "name", "size")
_FLAT_RE = re.compile(r"[^a-z0-9]+")
_MONEY_RE = re.compile(r"[^0-9.,]")


def _num(value: Any) -> float | None:
    """A number out of whatever a feed put in a price field: 176, "176.00", "1,234.5", None."""
    if value is None or isinstance(value, bool):
        return None
    if isinstance(value, int | float):
        return float(value)
    text = _MONEY_RE.sub("", str(value))
    if not text:
        return None
    if text.count(",") == 1 and "." not in text and len(text.rsplit(",", 1)[1]) in (1, 2):
        text = text.replace(",", ".")  # a decimal comma
    else:
        text = text.replace(",", "")  # thousands separators
    try:
        return float(text)
    except ValueError:
        return None


def _text(value: Any) -> str | None:
    """A field as text, with the HTML entities a feed ships ("B&G") decoded, as ingest does."""
    text = html.unescape(str(value)).strip() if value is not None else ""
    return text or None


def _brand_name(value: Any) -> str | None:
    """A brand as JSON-LD writes it (an object with a name) or as a feed does (a string)."""
    if isinstance(value, dict):
        value = value.get("name")
    return _text(value)


def _stated_size(name: str | None, size_ml: Any = None) -> tuple[str | None, int | None]:
    """The size as the tile states it, and its millilitres; the collector's figure when the name is silent."""
    stated = parse_size(name)
    if stated:
        value, unit = stated
        return f"{value:g} {unit}", size_ml_of(value, unit)
    millilitres = _num(size_ml)
    if millilitres:
        return f"{millilitres:g} ml", int(millilitres)
    return None, None


def _platform(parser_version: str | None) -> str:
    """The platform a raw record's parser belongs to: the versioned prefix ("avolta/2026-09-05"),
    or, for a record stamped with a bare collector slug, that collector's module."""
    prefix = (parser_version or "").split("/", 1)[0]
    collector = COLLECTORS.get(prefix)
    if collector is not None:
        prefix = type(collector).__module__.rsplit(".", 1)[-1]
    return "heinemann" if prefix in ("heinemann_platform", "heinemann-platform") else prefix


def collected_tile(parser_version: str | None, payload: dict | None, *, currency: str | None = None) -> dict[str, Any]:
    """The tile as the shop showed it, read back out of the fragment the collector kept.

    One reader per platform, taking each field from where that platform's collector takes
    it (`COLLECTORS.md`), so the review page shows what the collector saw and not a second
    parse of it. A shape this does not know yields empty fields, never a guess; a field
    the fragment does not carry stays empty. The currency is the fragment's where it
    declares one, else the location's, passed in by the caller.
    """
    platform = _platform(parser_version)
    p = payload if isinstance(payload, dict) else {}
    name = brand = size = None
    size_ml: int | None = None
    price = was = None
    if platform == "avolta":
        tile, variant = p.get("tile") or {}, p.get("variant") or {}
        name, brand = _text(tile.get("name")), _text(tile.get("brand"))
        price = _num(variant.get("price", tile.get("price")))
        was = _num(variant.get("was_price", tile.get("was_price")))
        size, size_ml = _stated_size(name, variant.get("size_ml"))
    elif platform == "shopify":
        product, variant = p.get("product") or {}, p.get("variant") or {}
        name, brand = _text(product.get("title")), _text(product.get("vendor"))
        option = _text(variant.get("title")) or _text(variant.get("option1"))
        size, size_ml = _stated_size(option if option and option.lower() != "default title" else name)
        price, was = _num(variant.get("price")), _num(variant.get("compare_at_price"))
    elif platform == "ari":
        ld = p.get("jsonld") or {}
        offers = ld.get("offers") or {}
        if isinstance(offers, list):
            offers = offers[0] if offers and isinstance(offers[0], dict) else {}
        name, brand = _text(ld.get("name")), _brand_name(ld.get("brand"))
        price, was = _num(offers.get("price")), _num(p.get("was_price"))
        currency = _text(offers.get("priceCurrency")) or currency
        size, size_ml = _stated_size(name)
    elif platform == "extime":
        ld, variation = p.get("jsonld") or {}, p.get("variation") or {}
        offers = ld.get("offers") or []
        first = offers[0] if isinstance(offers, list) and offers and isinstance(offers[0], dict) else (
            offers if isinstance(offers, dict) else {})
        name = _text(variation.get("product_name")) or _text(ld.get("name"))
        brand = _brand_name(ld.get("brand")) or _text((p.get("main_offer") or {}).get("brand_name"))
        tier = variation.get("duty_free") if isinstance(variation.get("duty_free"), dict) else {}
        price, was = _num(tier.get("price")), _num(tier.get("price_crossed"))
        currency = _text(first.get("priceCurrency")) or currency
        unit = _text(variation.get("capacity_unit"))
        capacity = _num(variation.get("capacity"))
        size = _text(variation.get("name")) or (f"{capacity:g} {unit}" if capacity and unit else None)
        size_ml = size_ml_of(capacity, unit) if capacity and unit else None
        if size_ml is None:
            size, size_ml = _stated_size(size or name)
    elif platform == "shilla":
        specs, tiers, hidden = p.get("specs") or {}, p.get("tiers") or {}, p.get("hidden") or {}
        name, brand = _text(p.get("name")), _text(p.get("brand"))
        price = _num(hidden.get("prdPriceDollar")) or _num(tiers.get("discount")) or _num(tiers.get("list"))
        listed = _num(tiers.get("list"))
        was = listed if listed and price and listed > price else None
        currency = "USD"
        stated = _text(specs.get("Weight,Volume")) or _text(specs.get("Volume"))
        size, size_ml = (stated, parse_size_ml(stated)) if stated else _stated_size(name)
    elif platform == "changi":
        variant, offer = p.get("variant") or {}, p.get("offer") or {}
        name, brand = _text(p.get("name")), _text(p.get("brandName"))
        price, was = changi_price(offer) if isinstance(offer, dict) and offer else (None, None)
        currency = "SGD"
        qualifiers = variant.get("variantOptionQualifiers") or []
        measure = next((_text(q.get("value")) for q in qualifiers
                        if isinstance(q, dict) and q.get("qualifier") == "level1saleMeasure"), None)
        size, size_ml = (measure, parse_size_ml(measure)) if measure else _stated_size(name)
    elif platform == "dubai":
        name, brand = _text(p.get("displayName")), _text(p.get("brand"))
        sale, listed = _num(p.get("salePrice")), _num(p.get("listPrice"))
        price = sale or listed
        was = listed if sale and listed and listed > sale else None
        size, size_ml = _stated_size(name)
    elif platform == "heinemann":
        name = _text(p.get("name"))
        brand = _text(p.get("brand")) or _text(p.get("manufacturerName"))
        price = _num((p.get("price") or {}).get("value"))
        was = _num((p.get("strikethroughPrice") or {}).get("value"))
        content = p.get("contentUnit") or {}
        quantity = _num(content.get("quantity"))
        unit = (_text((content.get("unit") or {}).get("code")) or "").lower()
        if quantity and unit:
            size, size_ml = f"{quantity:g} {unit}", size_ml_of(quantity, unit)
        else:
            size, size_ml = _stated_size(name)
    return {"name": name, "brand": brand, "size": size, "size_ml": size_ml,
            "price": price, "was_price": was, "currency": currency}


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 house 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("size_ml") and int(tile["size_ml"]) != int(ours["size_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.locations()}


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 _houses(db: Session) -> dict[int, Brand]:
    return {b.id: b for b in db.scalars(select(Brand))}


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


def _listing_conditions(db: Session, *, q: str | None, collector: str | None, airport: str | None,
                        brand: str | None) -> list:
    where = []
    if q:
        like = f"%{q.strip().lower()}%"
        where.append(or_(func.lower(Product.name).like(like), func.lower(Product.brand).like(like)))
    if brand:
        where.append(func.lower(Product.brand).like(f"%{brand.strip().lower()}%"))
    location_ids = _scope_location_ids(db, collector, airport)
    if location_ids is not None:
        where.append(Listing.location_id.in_(location_ids))
    return where


_LISTING_ORDER = (Product.brand.asc().nullslast(), Product.name.asc(), Location.code.asc(), Listing.id.asc())


def _lines_of(db: Session, line_ids: set[int]) -> dict[int, ProductLine]:
    """The line rows of a page's products and the canonical rows their aliases point at."""
    rows: dict[int, ProductLine] = {}
    wanted = {i for i in 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.canonical_id for row in rows.values() if row.canonical_id}
    return rows


def _ours_of(product: Product, house: Brand | None, line: ProductLine | None = None) -> dict[str, Any]:
    return {
        "name": product.name, "brand": product.brand,
        "house": house.name if house else product.brand,
        "house_slug": house.slug if house else None,
        "line": line.name if line else None,
        "line_slug": line.slug if line else None,
        "variation": (product.attributes or {}).get("concentration"),
        "size": f"{product.size_value:g} {product.size_unit}" if product.size_value and product.size_unit else None,
        "size_ml": product.size_ml, "gtin": product.gtin,
    }


def listings(
    db: Session, *, q: str | None = None, collector: str | None = None, airport: str | None = None,
    brand: str | None = None, differs: str | None = None, page: int = 1, per_page: int = 50,
) -> dict[str, Any]:
    """One row per listing: the tile as the shop showed it beside what we made of it.

    The surface every identity decision is checked against (Stream M): the collected
    brand, name, size and price from the listing's newest raw record on the left; the
    house, line, variation, size, product and our latest price on the right; and which of
    the three compared fields disagree. A listing whose collector keeps no fragment shows
    an empty left side, honestly. "Differs" reads every candidate's fragment, because the
    tile lives in JSON the database cannot compare, then pages the ones that disagree.
    """
    where = _listing_conditions(db, q=q, collector=collector, airport=airport, brand=brand)
    joined = (
        select(Listing.id, Product, Location)
        .join(Product, Product.id == Listing.product_id)
        .join(Location, Location.id == Listing.location_id)
        .where(and_(*where) if where else True)
    )
    per_page = max(1, min(per_page, 200))
    page = max(1, page)
    brands = _houses(db)
    with_tile = db.scalar(
        select(func.count(func.distinct(RawRecord.listing_id)))
        .select_from(RawRecord).join(Listing, Listing.id == RawRecord.listing_id)
        .join(Product, Product.id == Listing.product_id)
        .join(Location, Location.id == Listing.location_id)
        .where(and_(*where) if where else True)
    ) or 0

    if differs in DIFFERS_CHOICES:
        raws = _newest_raw(db, None, slim=True)
        kept: list[int] = []
        for listing_id, product, location in db.execute(joined.order_by(*_LISTING_ORDER)).all():
            raw = raws.get(listing_id)
            if raw is None:
                continue
            tile = collected_tile(raw[0], raw[1], currency=location.currency)
            fields = tile_differs(tile, _ours_of(product, _house_of(brands, product.brand_id)))
            if fields and (differs == "any" or differs in fields):
                kept.append(listing_id)
        total = len(kept)
        ids = kept[(page - 1) * per_page: page * per_page]
    else:
        total = db.scalar(select(func.count()).select_from(joined.subquery())) or 0
        ids = list(db.scalars(
            select(Listing.id)
            .join(Product, Product.id == Listing.product_id)
            .join(Location, Location.id == Listing.location_id)
            .where(and_(*where) if where else True)
            .order_by(*_LISTING_ORDER).offset((page - 1) * per_page).limit(per_page)
        ))

    rows = _listing_rows(db, ids, brands, raws=None)
    return {"total": total, "with_tile": with_tile, "page": page, "per_page": per_page, "rows": rows}


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, location)
        for listing, product, location in db.execute(
            select(Listing, Product, Location)
            .join(Product, Product.id == Listing.product_id)
            .join(Location, Location.id == Listing.location_id)
            .where(Listing.id.in_(ids))
        ).all()
    }
    raws = _newest_raw(db, ids) if raws is None else raws
    lines = _lines_of(db, {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, location = found[listing_id]
        raw = raws.get(listing_id)
        tile = None
        if raw is not None:
            tile = collected_tile(raw[0], raw[1], currency=location.currency)
            tile["seen_at"] = _iso(raw[2])
            tile["parser"] = raw[0]
        ours = _ours_of(product, _house_of(brands, product.brand_id), resolve_alias(lines, 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, "product_id": product.id,
            "airport": location.iata or location.code, "location": location.code,
            "collector": collector_of.get(location.code), "source_sku": listing.source_sku, "url": listing.url,
            "collected": tile, "ours": ours,
            "differs": tile_differs(tile, ours) if tile else [],
        })
    return out