"""Is this data true? The no-network audit: metrics with thresholds, and the human review's lists.

`audit` reads the database and nothing else. It answers the questions the
verification bar (build plan §3a, Decision 10) says a re-read cannot: verify
re-reads a page through the same parser, so it can never notice a parser that
reads the wrong field. This module looks for the *shapes* those misreads
leave behind: a location whose currency disagrees with its prices, a litre of
vodka for the price of a miniature, a shop where every product is in stock,
a size no bottle comes in.

Two rules shape the output:

* **A threshold lives beside its metric**, in `THRESHOLDS`, and starts at the
  value measured the day the metric was written. Tightening one is then a
  one-line diff with a date on it, not a judgement remembered by nobody. A
  metric is "over" when its value exceeds its threshold; nothing here hides a
  row or blocks a source, because the audit measures, and the tripwire in
  `verify.py` acts.
* **Every metric carries a sample** of the rows behind it, with the retailer
  URL, so the dated human review (build plan §7, Mon 21 Sep) starts from a
  list rather than from a query.

The snapshot is one JSON document: written to `.logs/verification/` and, once
migration #2 has landed, to `audit_snapshots` as well. The pure functions
(`is_multipack`, `duplicate_kind`, `duplicate_rows`, `same_day_flips`,
`cross_shop_ratios`, `stock_smell`, `capacity_disagreement`, `brand_folds`) are what
`tests/test_audit.py` pins. Merged products (`merged_into_id` set) are tombstones and
count in no product metric.
"""

import json
import logging
import random
import re
from collections import Counter, defaultdict
from dataclasses import asdict, dataclass, field
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any

from sqlalchemy import func, inspect, select, text
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session

from app.models import (
    CollectionRun,
    Listing,
    Location,
    MergeCandidate,
    PriceObservation,
    Product,
    RawRecord,
    Source,
)
from app.services import merges
from app.services.collectors.extime import capacity_ml, implied_size_ml, variation_price
from app.services.normalize import brand_key
from app.services.taxonomy import vertical_of

logger = logging.getLogger(__name__)

# The UI's own freshness rule (web/src/lib/format.ts, isStale): older than
# this and a price is shown with its date in a warning tone.
STALE_DAYS = 3
# A run still "running" this long after it started is a process that died.
# Half the ingest STUCK_AFTER: the audit runs nightly and should see a dead
# run before the next collection supersedes it.
STUCK_HOURS = 12
# The cross-shop spread that a third shop must corroborate (featured.py uses
# 3.0 to headline; the audit lists from 2.5 so the review sees the margin).
RATIO_LIMIT = 2.5
# A bottle larger than this is a case or a display, not something a traveller
# carries, unless the name says it is a pack.
MAX_SINGLE_ML = 5_000
SAMPLE = 30

# Measured 2026-09-05 on the dump taken that morning (pre-todos, 13,128
# products, 33,242 observations). A metric is over when value > threshold.
# Tighten by editing the number and dating the change here.
# 2026-09-05 (evening): duplicate_groups 10 -> 0, oversize_singles 6 -> 2 and
# brand_folds (new metric) 0, the numbers measured on a copy of the same dump
# after migration #3's backfills (implausible_sizes, brands, merges); the audit
# now skips merged products, which had kept 41 groups and 122 folds visible.
THRESHOLDS: dict[str, float] = {
    "vertical_vs_category": 0,
    "currency_mismatch": 0,
    "orphan_tiles": 0,
    "duplicate_groups": 0,
    "drinks_without_size": 1560,
    "oversize_singles": 2,
    "brand_folds": 0,
    "extime_capacity_disagreement": 0,
    "same_day_flips": 2,
    "cross_shop_ratio": 18,
    "stock_smell_sources": 3,
    "fallback_fx_latest": 0,
    "uncategorised": 3419,
    "stuck_runs": 0,
    "stale_visible_locations": 4,
}


@dataclass(slots=True)
class Metric:
    name: str
    value: float
    threshold: float | None
    unit: str = "rows"
    note: str = ""
    by: dict[str, Any] = field(default_factory=dict)
    sample: list[dict[str, Any]] = field(default_factory=list)

    @property
    def ok(self) -> bool:
        return self.threshold is None or self.value <= self.threshold

    def as_dict(self) -> dict[str, Any]:
        out = asdict(self)
        out["ok"] = self.ok
        return out


@dataclass(slots=True)
class LatestRow:
    """One listing with its newest observation and everything a check needs."""

    listing_id: int
    product_id: int
    location_id: int
    location_code: str
    location_currency: str
    visible: bool
    source_slug: str | None
    price: float
    currency: str
    price_usd: float | None
    was_price: float | None
    in_stock: bool | None
    observed_at: datetime
    fx_source: str | None
    url: str | None
    source_sku: str
    name: str
    brand: str | None
    gtin: str | None
    size_ml: int | None
    vertical: str
    category: str | None


# --- pure logic (tested) -------------------------------------------------------------------

# "24x0.33L", "6 x 70cl", "Twin Pack", "3L Jeroboam": the size is a pack or a
# large format, and the name says so.
_MULTIPACK_RE = re.compile(
    r"(\d+\s*(bottles?|cans?|btl)?\s*[x\u00d7]\s*\d|\b(pack|case|twin|duo|trio|set|gift|magnum|jeroboam|methuselah|rehoboam)\b)",
    re.I,
)


def is_multipack(name: str | None) -> bool:
    """Whether a name explains a size no single bottle comes in."""
    return bool(name and _MULTIPACK_RE.search(name))


def duplicate_kind(gtins: list[str | None]) -> str:
    """How a group of products sharing one match key relates by barcode.

    Two different barcodes are two bottles the key cannot tell apart (leave
    them); one barcode means the rest should fold into it; none means the
    key alone decides and a merge is safe.
    """
    present = {g for g in gtins if g}
    if len(present) >= 2:
        return "gtin_differs"
    if len(present) == 1:
        return "gtin_one"
    return "gtin_none"


def duplicate_rows(groups: list[merges.Group], product_url: dict[int, str | None]) -> list[dict[str, Any]]:
    """The audit's view of the merge rules' duplicate groups: what the barcodes
    say (`kind`), why the rules keep the group apart (`conflicts`, empty when the
    next `backfill merges` would fold it), and the members with their links.

    The audit once kept its own reading (barcodes only) and listed a gift set
    against its bottle as a merge the rules had deliberately queued for a human;
    the rules are the one definition, so the metric counts exactly what a
    backfill would do.
    """
    return [
        {
            "match_key": g.match_key,
            "kind": duplicate_kind([p.gtin for p in g.members]),
            "conflicts": list(g.conflicts),
            "products": [
                {"id": p.id, "name": p.name, "gtin": p.gtin, "url": product_url.get(p.id)} for p in g.members
            ],
        }
        for g in groups
    ]


def same_day_flips(
    observations: list[tuple[int, float, float | None, datetime]],
) -> list[dict[str, Any]]:
    """Listings whose price changed within one day with no was-price on either side.

    A genuine promotion carries the crossed-out figure; a flip without one is
    usually the parser reading the small-variant or the pre-promotion price
    on one pass and the real one on the next.
    """
    by_day: dict[tuple[int, str], list[tuple[float, float | None]]] = defaultdict(list)
    for listing_id, price, was_price, observed_at in observations:
        by_day[(listing_id, observed_at.date().isoformat())].append((float(price), was_price))
    out = []
    for (listing_id, day), rows in by_day.items():
        prices = {p for p, _ in rows}
        if len(prices) > 1 and not any(w for _, w in rows):
            out.append({"listing_id": listing_id, "day": day, "prices": sorted(prices)})
    return out


def cross_shop_ratios(rows: list[LatestRow], limit: float = RATIO_LIMIT) -> list[dict[str, Any]]:
    """Products whose dearest shop is more than `limit` times the cheapest.

    One shop per product counts once (its cheapest in-stock row). The
    dataset's worst spread so far, $77 against $21 for one sparkling wine,
    was almost certainly a retailer's own entry error; the list is for the
    review, the featuring rule needs a third shop before it headlines one.
    """
    per_product: dict[int, dict[int, LatestRow]] = defaultdict(dict)
    for row in rows:
        if row.price_usd is None:
            continue
        held = per_product[row.product_id].get(row.location_id)
        if held is None or (row.in_stock is False, row.price_usd) < (held.in_stock is False, held.price_usd):
            per_product[row.product_id][row.location_id] = row
    out = []
    for product_id, shops in per_product.items():
        if len(shops) < 2:
            continue
        cheapest = min(shops.values(), key=lambda r: r.price_usd)
        dearest = max(shops.values(), key=lambda r: r.price_usd)
        if cheapest.price_usd <= 0:
            continue
        ratio = dearest.price_usd / cheapest.price_usd
        if ratio > limit:
            out.append(
                {
                    "product_id": product_id,
                    "name": cheapest.name,
                    "brand": cheapest.brand,
                    "size_ml": cheapest.size_ml,
                    "ratio": round(ratio, 2),
                    "shops": len(shops),
                    "cheapest": _shop(cheapest),
                    "dearest": _shop(dearest),
                }
            )
    return sorted(out, key=lambda r: -r["ratio"])


def stock_smell(counts: dict[str, Counter], minimum: int = 20) -> list[str]:
    """Sources where every latest row says in stock.

    A feed that never says "out of stock" is not reporting stock; it is
    reporting a default, and the comparison then prefers a shelf that may be
    empty. Dubai's API was read as in-stock for every row for two weeks.
    """
    smelly = []
    for slug, c in counts.items():
        n = sum(c.values())
        if n >= minimum and c.get("true", 0) == n:
            smelly.append(slug)
    return sorted(smelly)


def capacity_disagreement(variation: dict, price: float | None) -> dict[str, int] | None:
    """Extime's declared capacity against the size its per-litre price implies.

    Both are read from the page's own payload; when they differ by more than
    the 700-vs-750 tolerance, one of them is wrong and the collector filed
    the size as unknown. The audit counts how often that happens so a
    payload change shows up as a rising number rather than as silence.
    """
    declared = capacity_ml(variation.get("capacity"), variation.get("capacity_unit"))
    implied = implied_size_ml(variation, price)
    if declared is None or implied is None:
        return None
    if abs(declared - implied) / max(declared, implied) > 0.12:
        return {"declared_ml": declared, "implied_ml": implied}
    return None


def brand_folds(
    products: list[tuple[str | None, int]], brand_ids: dict[int, int | None] | None = None
) -> list[dict[str, Any]]:
    """Spellings of one brand that the fold key brings together and the brands
    table has not folded.

    Input is (brand, product_id), plus each product's `brand_id` once migration
    #3 has landed. Before the brands table existed the list was everything the
    key WOULD fold, for the review to object to a wrong fold; after `backfill
    brands` a key whose products all share one brand row is folded and drops
    off, so the list, and the metric over it, read zero unless a collector
    wrote a spelling the fold missed or a product has no brand row at all.
    Without the second argument the pre-migration reading is kept.
    """
    spellings: dict[str, Counter] = defaultdict(Counter)
    rows: dict[str, set[int | None]] = defaultdict(set)
    for brand, product_id in products:
        key = brand_key(brand)
        if key and brand:
            spellings[key][brand] += 1
            if brand_ids is not None:
                rows[key].add(brand_ids.get(product_id))
    out = [
        {"key": key, "spellings": dict(c.most_common()), "products": sum(c.values()),
         "brand_ids": sorted(i for i in rows[key] if i is not None) if brand_ids is not None else None,
         "unfolded": (None in rows[key]) if brand_ids is not None else None}
        for key, c in spellings.items()
        if len(c) > 1 and (brand_ids is None or len(rows[key]) != 1 or None in rows[key])
    ]
    return sorted(out, key=lambda r: (-len(r["spellings"]), r["key"]))


def _shop(row: LatestRow) -> dict[str, Any]:
    return {
        "location": row.location_code,
        "price": float(row.price),
        "currency": row.currency,
        "price_usd": row.price_usd,
        "in_stock": row.in_stock,
        "seen": row.observed_at.isoformat(),
        "url": row.url,
    }


def _row_sample(row: LatestRow) -> dict[str, Any]:
    return {
        "listing_id": row.listing_id,
        "product_id": row.product_id,
        "location": row.location_code,
        "source": row.source_slug,
        "brand": row.brand,
        "name": row.name,
        "size_ml": row.size_ml,
        "gtin": row.gtin,
        "price": float(row.price),
        "currency": row.currency,
        "price_usd": row.price_usd,
        "was_price": row.was_price,
        "in_stock": row.in_stock,
        "seen": row.observed_at.isoformat(),
        "url": row.url,
    }


# --- reading the database --------------------------------------------------------------------


def _latest_subquery():
    ranked = select(
        PriceObservation.id.label("obs_id"),
        func.row_number()
        .over(
            partition_by=PriceObservation.listing_id,
            order_by=(PriceObservation.observed_at.desc(), PriceObservation.id.desc()),
        )
        .label("rn"),
    ).subquery()
    return select(ranked.c.obs_id).where(ranked.c.rn == 1).subquery()


def latest_rows(db: Session) -> list[LatestRow]:
    """Every listing's newest observation, joined to what the checks need. One query."""
    latest = _latest_subquery()
    stmt = (
        select(
            Listing.id,
            Listing.product_id,
            Listing.location_id,
            Location.code,
            Location.currency,
            Location.visible,
            Source.slug,
            PriceObservation.price,
            PriceObservation.currency,
            PriceObservation.price_usd,
            PriceObservation.was_price,
            PriceObservation.in_stock,
            PriceObservation.observed_at,
            CollectionRun.fx_source,
            Listing.url,
            Listing.source_sku,
            Product.name,
            Product.brand,
            Product.gtin,
            Product.size_ml,
            Product.vertical,
            Product.category,
        )
        .join(PriceObservation, PriceObservation.listing_id == Listing.id)
        .join(Location, Location.id == Listing.location_id)
        .join(Product, Product.id == Listing.product_id)
        .outerjoin(CollectionRun, CollectionRun.id == PriceObservation.run_id)
        .outerjoin(Source, Source.id == CollectionRun.source_id)
        .where(PriceObservation.id.in_(select(latest.c.obs_id)))
    )
    out = []
    for r in db.execute(stmt):
        out.append(
            LatestRow(
                listing_id=r[0], product_id=r[1], location_id=r[2], location_code=r[3],
                location_currency=r[4], visible=bool(r[5]), source_slug=r[6],
                price=float(r[7]), currency=r[8],
                price_usd=float(r[9]) if r[9] is not None else None,
                was_price=float(r[10]) if r[10] is not None else None,
                in_stock=r[11], observed_at=r[12], fx_source=r[13], url=r[14],
                source_sku=r[15], name=r[16], brand=r[17], gtin=r[18], size_ml=r[19],
                vertical=r[20], category=r[21],
            )
        )
    return out


def _sample(rows: list, rng: random.Random, n: int = SAMPLE) -> list:
    return rng.sample(rows, n) if len(rows) > n else list(rows)


def _live_products_stmt():
    """Every product that is not a tombstone. A merged product keeps its row with
    `merged_into_id` set (Decision 6: recorded, never deleted) and still shares
    its match key and brand spelling with the survivor, so counted it would keep
    every duplicate group and brand fold the merge resolved on the list: 41
    groups and 122 folds on the first copy after `backfill merges`."""
    return select(
        Product.id, Product.name, Product.brand, Product.brand_id, Product.gtin,
        Product.size_ml, Product.vertical, Product.category, Product.image_source,
    ).where(Product.merged_into_id.is_(None))


def _merge_candidates(db: Session, limit: int = 100) -> list[dict[str, Any]]:
    """Undecided merge candidates, for the review's "is this the right product"
    list: the pairs the rules suspect are one bottle but would not merge
    unasked (mostly two barcodes under one key). Empty before migration #3."""
    if not inspect(db.get_bind()).has_table("merge_candidates"):
        return []
    out = []
    for c in db.scalars(
        select(MergeCandidate).where(MergeCandidate.decision.is_(None))
        .order_by(MergeCandidate.reason, MergeCandidate.id).limit(limit)
    ):
        detail = c.detail if isinstance(c.detail, dict) else {}
        sides = [detail.get(k) for k in ("from", "to") if isinstance(detail.get(k), dict)]
        out.append({
            "id": c.id, "reason": c.reason, "match_key": detail.get("match_key"),
            "products": [{"id": p.get("id"), "name": p.get("name"), "gtin": p.get("gtin"),
                          "size_ml": p.get("size_ml")} for p in sides]
                        or [{"id": c.product_id}, {"id": c.candidate_id}],
        })
    return out


def run_audit(db: Session, *, now: datetime | None = None, seed: int = 0) -> dict[str, Any]:
    """The whole snapshot. Read-only; safe against a live database at any time."""
    now = now or datetime.now(UTC)
    rng = random.Random(seed)
    rows = latest_rows(db)
    products = db.execute(_live_products_stmt()).all()
    merged = db.scalar(select(func.count(Product.id)).where(Product.merged_into_id.isnot(None))) or 0
    live_ids = {p.id for p in products}
    metrics: list[Metric] = []
    priced = {r.product_id for r in rows}
    product_url = {}
    for r in rows:
        product_url.setdefault(r.product_id, r.url)

    # 1. vertical vs category: our own category names a family the row is not in.
    bad = [
        {"product_id": p.id, "name": p.name, "category": p.category, "vertical": p.vertical,
         "expected": vertical_of(p.category), "url": product_url.get(p.id)}
        for p in products
        if p.category and vertical_of(p.category) and vertical_of(p.category) != p.vertical
    ]
    metrics.append(Metric("vertical_vs_category", len(bad), THRESHOLDS["vertical_vs_category"],
                          note="products whose category belongs to another vertical",
                          sample=_sample(bad, rng)))

    # 2. location currency vs observation currency (the MEX incident).
    mism = [r for r in rows if r.currency != r.location_currency]
    metrics.append(Metric("currency_mismatch", len(mism), THRESHOLDS["currency_mismatch"],
                          note="latest observations whose currency is not the location's",
                          by=dict(Counter(f"{r.location_code}:{r.currency}" for r in mism)),
                          sample=[_row_sample(r) for r in _sample(mism, rng)]))

    # 3. orphan parent tiles: a "-P" row with no "::size" child carries the family's cheapest price.
    listings = db.execute(select(Listing.id, Listing.location_id, Listing.source_sku, Listing.url)).all()
    children = {(l.location_id, l.source_sku.split("::", 1)[0]) for l in listings if "::" in l.source_sku}
    orphans = [
        {"listing_id": l.id, "location_id": l.location_id, "source_sku": l.source_sku, "url": l.url}
        for l in listings
        if l.source_sku.endswith("-P") and (l.location_id, l.source_sku) not in children
    ]
    metrics.append(Metric("orphan_tiles", len(orphans), THRESHOLDS["orphan_tiles"],
                          note="multi-size parent tiles with no per-size sibling (backfill orphan_tiles purges them)",
                          sample=_sample(orphans, rng)))

    # 4. duplicate match_key groups, read by the merge rules themselves
    # (live products only): a group with no conflict is one the next
    # `backfill merges` would fold, so after the backfill the count is zero
    # until the identity rules let a duplicate through. Two barcodes, a set
    # against its bottle and differing attributes keep a group apart; those
    # are queued in merge_candidates for a human (the review list below).
    dups = duplicate_rows(merges.duplicate_groups(db), product_url)
    mergeable = [d for d in dups if not d["conflicts"]]
    metrics.append(Metric("duplicate_groups", len(mergeable), THRESHOLDS["duplicate_groups"], unit="groups",
                          note="products sharing a match key that the merge rules would fold (backfill merges)",
                          by=dict(Counter(c for d in dups for c in (d["conflicts"] or ["mergeable"]))),
                          sample=_sample(mergeable, rng)))

    # 5. a drink with a price and no size cannot be compared per bottle.
    sizeless = [p for p in products if p.vertical == "liquor" and p.size_ml is None and p.id in priced]
    metrics.append(Metric("drinks_without_size", len(sizeless), THRESHOLDS["drinks_without_size"],
                          note="liquor products with a current price and no size",
                          sample=[{"product_id": p.id, "name": p.name, "brand": p.brand, "url": product_url.get(p.id)}
                                  for p in _sample(sizeless, rng)]))

    # 6. sizes no bottle comes in, unless the name says pack or case.
    oversize = [p for p in products if p.size_ml and p.size_ml > MAX_SINGLE_ML and not is_multipack(p.name)]
    metrics.append(Metric("oversize_singles", len(oversize), THRESHOLDS["oversize_singles"],
                          note=f"size over {MAX_SINGLE_ML} ml with nothing in the name explaining it",
                          sample=[{"product_id": p.id, "name": p.name, "size_ml": p.size_ml, "url": product_url.get(p.id)}
                                  for p in _sample(oversize, rng)]))

    # 7. Extime: declared capacity against the size the per-litre price implies.
    disagreements, read = [], 0
    for record in db.scalars(select(RawRecord).where(RawRecord.parser_version.like("extime%"))):
        variation = record.payload.get("variation") if isinstance(record.payload, dict) else None
        if not isinstance(variation, dict):
            continue
        read += 1
        found = capacity_disagreement(variation, variation_price(variation)[0])
        if found:
            disagreements.append({"listing_id": record.listing_id, **found})
    metrics.append(Metric("extime_capacity_disagreement", len(disagreements),
                          THRESHOLDS["extime_capacity_disagreement"],
                          note=f"of {read} Extime raw records with a variation payload",
                          sample=_sample(disagreements, rng)))

    # 8. same-day price flips with no was-price on either side.
    observations = db.execute(
        select(PriceObservation.listing_id, PriceObservation.price, PriceObservation.was_price,
               PriceObservation.observed_at)
    ).all()
    flips = same_day_flips([(o[0], float(o[1]), o[2], o[3]) for o in observations])
    metrics.append(Metric("same_day_flips", len(flips), THRESHOLDS["same_day_flips"],
                          note="a listing priced two ways in one day, neither a promotion",
                          sample=_sample(flips, rng)))

    # 9. cross-shop spread.
    ratios = cross_shop_ratios(rows)
    metrics.append(Metric("cross_shop_ratio", len(ratios), THRESHOLDS["cross_shop_ratio"], unit="products",
                          note=f"dearest shop over {RATIO_LIMIT}x the cheapest",
                          sample=ratios[:SAMPLE]))

    # 10. in-stock rate per source.
    stock: dict[str, Counter] = defaultdict(Counter)
    for r in rows:
        stock[r.source_slug or "?"][{True: "true", False: "false"}.get(r.in_stock, "unknown")] += 1
    smelly = stock_smell(stock)
    metrics.append(Metric("stock_smell_sources", len(smelly), THRESHOLDS["stock_smell_sources"], unit="sources",
                          note="sources whose every latest row says in stock (a default, not a reading)",
                          by={slug: dict(c) for slug, c in sorted(stock.items())},
                          sample=[{"source": s} for s in smelly]))

    # 11. fallback FX: how much of what is shown was converted on the static table.
    fallback_all = db.scalar(
        select(func.count(PriceObservation.id))
        .join(CollectionRun, CollectionRun.id == PriceObservation.run_id)
        .where(CollectionRun.fx_source == "fallback")
    ) or 0
    fb_latest = [r for r in rows if r.fx_source == "fallback"]
    metrics.append(Metric("fallback_fx_latest", len(fb_latest), THRESHOLDS["fallback_fx_latest"],
                          note=f"latest rows converted on the fallback table ({fallback_all} observations ever; "
                               "runs before migration #1 carry no fx_source)",
                          by={"latest_by_fx_source": dict(Counter(r.fx_source or "unknown" for r in rows))},
                          sample=[_row_sample(r) for r in _sample(fb_latest, rng)]))

    # 12. image provenance.
    metrics.append(Metric("image_provenance", 0, None, unit="split",
                          note="products by where their picture came from",
                          by=dict(Counter(p.image_source or "none" for p in products))))

    # 13. uncategorised, per source (live products only: a listing still on a
    # tombstone is one `backfill merges` re-points, not a second product).
    uncat = [r for r in rows if r.category is None and r.product_id in live_ids]
    uncat_products = {r.product_id for r in uncat}
    metrics.append(Metric("uncategorised", len(uncat_products), THRESHOLDS["uncategorised"], unit="products",
                          note="priced products with no category, by the source that priced them",
                          by=dict(Counter(r.source_slug or "?" for r in uncat)),
                          sample=[_row_sample(r) for r in _sample(uncat, rng)]))

    # 14. stuck runs.
    stuck = [
        {"run_id": run.id, "source_id": run.source_id, "started_at": run.started_at.isoformat(),
         "hours": round((now - run.started_at).total_seconds() / 3600, 1)}
        for run in db.scalars(select(CollectionRun).where(CollectionRun.status == "running"))
        if now - run.started_at > timedelta(hours=STUCK_HOURS)
    ]
    metrics.append(Metric("stuck_runs", len(stuck), THRESHOLDS["stuck_runs"], unit="runs",
                          note=f"still 'running' after {STUCK_HOURS} h", sample=stuck))

    # 15. observation age per visible location, against the UI's threshold.
    by_loc: dict[str, list[LatestRow]] = defaultdict(list)
    for r in rows:
        if r.visible:
            by_loc[r.location_code].append(r)
    ages = {}
    for code, group in sorted(by_loc.items()):
        newest = max(g.observed_at for g in group)
        stale = sum(1 for g in group if now - g.observed_at > timedelta(days=STALE_DAYS))
        ages[code] = {"listings": len(group), "newest": newest.isoformat(),
                      "newest_age_days": round((now - newest).total_seconds() / 86400, 1),
                      "stale_share": round(stale / len(group), 3)}
    stale_locs = [c for c, a in ages.items() if a["newest_age_days"] > STALE_DAYS]
    metrics.append(Metric("stale_visible_locations", len(stale_locs), THRESHOLDS["stale_visible_locations"],
                          unit="locations", note=f"visible locations with nothing newer than {STALE_DAYS} days",
                          by=ages, sample=[{"location": c} for c in stale_locs]))

    # 16. brand spellings one fold key joins that the brands table has not
    # folded (migration #3): zero after `backfill brands` unless a collector
    # wrote a spelling the fold missed or a product has no brand row.
    folds = brand_folds([(p.brand, p.id) for p in products], {p.id: p.brand_id for p in products})
    metrics.append(Metric("brand_folds", len(folds), THRESHOLDS["brand_folds"], unit="groups",
                          note="brand spellings one fold key joins that the brands table has not folded",
                          sample=_sample(folds, rng)))

    # The lists the human review starts from.
    discounts = sorted(
        (r for r in rows if r.was_price and r.was_price > r.price),
        key=lambda r: -(r.was_price - r.price) / r.was_price,
    )
    per_source: dict[str, list[LatestRow]] = defaultdict(list)
    for r in rows:
        per_source[r.source_slug or "?"].append(r)
    lists = {
        "brand_folds": folds,
        "merge_candidates": _merge_candidates(db),
        "ratios": ratios[:100],
        "largest_discounts": [
            {**_row_sample(r), "discount_pct": round(100 * (r.was_price - r.price) / r.was_price, 1)}
            for r in discounts[:SAMPLE]
        ],
        "sample_rows": {slug: [_row_sample(r) for r in _sample(group, rng, 5)]
                        for slug, group in sorted(per_source.items())},
    }
    over = [m.name for m in metrics if not m.ok]
    return {
        "taken_at": now.isoformat(),
        "seed": seed,
        "database": {
            "products": len(products), "merged": merged, "listings": len(listings),
            "latest_rows": len(rows), "observations": len(observations),
        },
        "summary": {"metrics": len(metrics), "over": over},
        "metrics": {m.name: m.as_dict() for m in metrics},
        "lists": lists,
    }


# --- storing ----------------------------------------------------------------------------------


def write_snapshot(snapshot: dict[str, Any], out_dir: Path) -> Path:
    out_dir.mkdir(parents=True, exist_ok=True)
    stamp = datetime.fromisoformat(snapshot["taken_at"]).strftime("%Y-%m-%d-%H%M")
    path = out_dir / f"audit-{stamp}.json"
    path.write_text(json.dumps(snapshot, indent=1, default=str) + "\n")
    return path


def store_snapshot(db: Session, snapshot: dict[str, Any]) -> bool:
    """Into audit_snapshots when migration #2 has landed; a no-op before."""
    if not inspect(db.get_bind()).has_table("audit_snapshots"):
        return False
    try:
        db.execute(
            text("INSERT INTO audit_snapshots (taken_at, metrics) VALUES (:at, CAST(:m AS jsonb))"),
            {"at": snapshot["taken_at"], "m": json.dumps(snapshot, default=str)},
        )
        db.commit()
        return True
    except SQLAlchemyError as exc:
        db.rollback()
        logger.error("audit_snapshot_not_stored error=%s", exc)
        return False


def format_summary(snapshot: dict[str, Any]) -> str:
    lines = [f"audit {snapshot['taken_at'][:16]}  {snapshot['database']}"]
    for name, m in snapshot["metrics"].items():
        state = "  ok " if m["ok"] else " OVER"
        threshold = "-" if m["threshold"] is None else f"{m['threshold']:g}"
        lines.append(f"{state} {name:30s} {m['value']:>7g} / {threshold:<7s} {m['unit']:9s} {m['note']}")
    over = snapshot["summary"]["over"]
    lines.append(f"{len(over)} over threshold" + (f": {', '.join(over)}" if over else ""))
    return "\n".join(lines)