"""Re-read a sample of published listings and say what moved, what broke, what refused.

What verify measures, plainly: **freshness and parser drift**. Each check
reads one published listing back through the collector's own `read_one()`
and compares the page as it is now with the observation we show. It can
tell that a price moved, a size or name no longer agrees, a listing is gone,
a page no longer parses, or a host refuses us. It cannot tell that the
parser reads the wrong field, because the re-read uses the same parser; that
is the audit's job (`audit.py`) and the dated human review's.

The rules, from build plan §3a and Decision 10:

* **A verification request is a request to a retailer.** Honest identity,
  robots re-read per host, crawl-delay honoured, and the source is abandoned
  at the first refusal: one BLOCKED check, no second request. Twenty
  declared-bot requests into an edge 403 is what this project says it does
  not do. A source whose latest collection was blocked is not verified. A
  page that is NOT THERE is not a refusal: that check is GONE, exactly as a
  read_one answering None is, and the sample carries on.
* **The tripwire, not a rate.** A correctness failure (MISMATCH_SIZE,
  MISMATCH_IDENTITY, MISMATCH_CURRENCY, PARSE_FAIL) blocks publication of its
  source until a human clears the check. PRICE_MOVED and GONE are reported.
  The rolling pass rate is a number for the owner area, never a launch gate
  at n=20 (20/20 bounds the true rate only at 86 percent).
* **Sampling v1 is hardcoded** (`plan_sample`): a census of every listing
  behind the unscoped featured eight and the cheapest and dearest listing of
  the top-20 savings, plus every row that moved more than 20 percent since
  its previous observation, then uniform random to N per source. The seed is
  stored so a run can be re-drawn.
* **Store, do not log.** Every check is a row; a BLOCKED, a FETCH_ERROR and a
  REVIEW are rows too. PRICE_MOVED writes a new observation with
  `source_kind='verify'` so the site shows the newer figure (an assumption
  pending §10 #11; the old rows are untouched either way).

Verdicts: PASS, PRICE_MOVED, MISMATCH_SIZE, MISMATCH_IDENTITY,
MISMATCH_CURRENCY, GONE, PARSE_FAIL, BLOCKED, REVIEW, and FETCH_ERROR (a
timeout or 5xx: try later; reported, never a failure). REVIEW needs no
network judgement: the re-read price sits outside the plausible band (under
half or over 2.5x the product's cross-shop median), moved by more than half,
or the page's two price channels disagree.
"""

import logging
import random
import time
from collections.abc import Callable
from dataclasses import dataclass
from datetime import UTC, datetime
from statistics import median
from typing import Any

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.models import (
    Account,
    CollectionRun,
    Listing,
    Shop,
    PriceObservation,
    Source,
    VerificationCheck,
    VerificationRun,
)
from app.models.quality import CORRECTNESS_FAILURES
from app.services.audit import LatestRow, latest_rows
from app.services.collectors.base import RawListing
from app.services.collectors.fetch import FetchError, PageGone, SourceBlocked
from app.services.collectors.registry import COLLECTORS
from app.services.collectors.robots import RobotsUnavailable
from app.services.fx import FxRates, load_rates
from app.services.ingest import _sizes_disagree
from app.services.normalize import name_tokens

logger = logging.getLogger(__name__)

DEFAULT_N = 20
WEEKLY_N = 200
# A host that renders pages in a browser costs us and them more per read.
RENDERED_CAP = 20
# Identity: the share of the shorter name's significant words the other
# name must contain. Below it, the page is a different product.
NAME_OVERLAP_MIN = 0.75
# A move this large is not a price change to record silently.
MOVED_REVIEW_PCT = 50.0
# Rows that moved this much since their previous observation are always in
# the sample: they are where a misread shows first.
TARGET_MOVE_PCT = 20.0
# The plausible band around the product's cross-shop median (ingest
# quarantines under 0.45; verify reviews a little wider on both sides).
BAND_LOW, BAND_HIGH = 0.5, 2.5
TOP_SAVINGS = 20
FEATURED = 8

Reader = Callable[[Any, Any], RawListing | None]
Sleeper = Callable[[float], None]


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


@dataclass(slots=True)
class Verdict:
    verdict: str
    detail: str = ""

    @property
    def is_failure(self) -> bool:
        return self.verdict in CORRECTNESS_FAILURES


def name_overlap(a: str, b: str) -> float:
    """How much of the shorter name's significant words the other contains.

    Containment rather than Jaccard: a retailer appends "Eau de Parfum" or a
    size label to the same bottle, and that must not read as a different
    product. The ATH Lancôme row ("Lancome La Vie Est Belle 50ml" against the
    page's "La Vie Est Belle Eau de Parfum 50ml") sits exactly at the line.
    """
    ta, tb = name_tokens(a), name_tokens(b)
    if not ta or not tb:
        return 0.0
    return len(ta & tb) / min(len(ta), len(tb))


def channel_prices(raw: dict | None) -> tuple[float | None, float | None]:
    """The two price channels an Extime page carries, when the record has both.

    The JSON-LD offer states one price for the page; the RSC variation
    states the priced size's. On a single-size page they must agree; when
    they do not, one channel is stale or the parser picked the wrong one.
    """
    if not isinstance(raw, dict):
        return None, None
    jsonld = raw.get("jsonld") if isinstance(raw.get("jsonld"), dict) else {}
    offers = jsonld.get("offers")
    offer = offers[0] if isinstance(offers, list) and offers and isinstance(offers[0], dict) else (
        offers if isinstance(offers, dict) else {}
    )
    variation = raw.get("variation") if isinstance(raw.get("variation"), dict) else {}
    duty_free = variation.get("duty_free") if isinstance(variation.get("duty_free"), dict) else {}
    try:
        ld = float(offer["price"]) if "price" in offer else None
    except (TypeError, ValueError):
        ld = None
    try:
        rsc = float(duty_free["price"]) if "price" in duty_free else None
    except (TypeError, ValueError):
        rsc = None
    return ld, rsc


def compare(published: LatestRow, live: RawListing | None, *, median_usd: float | None,
            rate: float | None) -> Verdict:
    """The verdict for one listing, given what the page says now.

    Order matters: identity and size before price, because a moved price on
    the wrong bottle is a mismatch, not a move. `rate` is units of the live
    currency per USD, for the plausibility band.
    """
    if live is None:
        return Verdict("GONE", "the page no longer offers this listing")
    if live.price is None or live.price <= 0:
        return Verdict("PARSE_FAIL", f"no readable price (got {live.price!r})")
    if (live.currency or "").upper() != (published.currency or "").upper():
        return Verdict("MISMATCH_CURRENCY", f"published {published.currency}, page says {live.currency}")
    if published.gtin and live.gtin and published.gtin != live.gtin:
        return Verdict("MISMATCH_IDENTITY", f"gtin published {published.gtin}, page says {live.gtin}")
    if _sizes_disagree(published.quantity_ml, live.quantity_ml):
        return Verdict("MISMATCH_SIZE", f"published {published.quantity_ml} ml, page says {live.quantity_ml} ml")
    same_gtin = bool(published.gtin and live.gtin and published.gtin == live.gtin)
    if not same_gtin:
        overlap = name_overlap(published.name, live.name)
        if overlap < NAME_OVERLAP_MIN:
            return Verdict(
                "MISMATCH_IDENTITY",
                f"name overlap {overlap:.2f}: published {published.name!r}, page says {live.name!r}",
            )
    notes = []
    ld, rsc = channel_prices(live.raw)
    if ld is not None and rsc is not None and abs(ld - rsc) / max(ld, rsc) > 0.01:
        notes.append(f"channels disagree: json-ld {ld:g}, payload {rsc:g}")
    live_usd = round(live.price / rate, 2) if rate else None
    if median_usd and live_usd is not None and not (BAND_LOW * median_usd <= live_usd <= BAND_HIGH * median_usd):
        notes.append(f"live {live_usd} USD outside {BAND_LOW}-{BAND_HIGH}x the cross-shop median {median_usd}")
    moved_pct = 100.0 * (live.price - published.price) / published.price if published.price else 0.0
    if abs(live.price - published.price) >= 0.005:
        move = f"{published.price:g} -> {live.price:g} {live.currency} ({moved_pct:+.1f}%)"
        if abs(moved_pct) > MOVED_REVIEW_PCT:
            notes.insert(0, f"moved {move}")
            return Verdict("REVIEW", "; ".join(notes))
        if notes:
            return Verdict("REVIEW", "; ".join([f"moved {move}", *notes]))
        return Verdict("PRICE_MOVED", move)
    if notes:
        return Verdict("REVIEW", "; ".join(notes))
    return Verdict("PASS")


@dataclass(slots=True)
class SamplePlan:
    """Which listings a source's run reads, and why each is there."""

    targeted: dict[int, str]  # listing_id -> reason
    filled: list[int]

    @property
    def listing_ids(self) -> list[int]:
        return [*self.targeted, *self.filled]

    def reason(self, listing_id: int) -> str:
        return self.targeted.get(listing_id, "random")


def plan_sample(candidates: list[int], targeted: dict[int, str], n: int, seed: int) -> SamplePlan:
    """The census first, then uniform random from the rest up to N.

    Deterministic for a seed, so a run's sample can be re-drawn exactly. The
    targeted set can exceed N: a census is a census.
    """
    chosen = {lid: why for lid, why in targeted.items() if lid in set(candidates)}
    rest = sorted(set(candidates) - set(chosen))
    rng = random.Random(seed)
    fill = rng.sample(rest, min(max(n - len(chosen), 0), len(rest)))
    return SamplePlan(targeted=chosen, filled=sorted(fill))


def targets_from_rows(rows: list[LatestRow], previous: dict[int, float]) -> dict[int, str]:
    """The listings that must be read: featured eight, top-20 savings, big movers.

    `rows` are the latest rows at VISIBLE shops (what a shopper sees);
    `previous` maps listing_id to the observation before the latest, for the
    movers. The featured pick is the same function the home page uses.
    """
    from app.services.featured import SavingRecord, pick_featured

    per_product: dict[int, dict[int, LatestRow]] = {}
    for r in rows:
        if r.price_usd is None:
            continue
        held = per_product.setdefault(r.variant_id, {}).get(r.shop_id)
        if held is None or (r.in_stock is False, r.price_usd) < (held.in_stock is False, held.price_usd):
            per_product[r.variant_id][r.shop_id] = r
    savings = []
    records = []
    for variant_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)
        saving = dearest.price_usd - cheapest.price_usd
        if saving <= 0:
            continue
        savings.append((saving, variant_id, cheapest.listing_id, dearest.listing_id))
        records.append(SavingRecord(
            variant_id=variant_id, saving_usd=saving, saving_pct=saving / dearest.price_usd,
            awarded=False, has_image=False, shop_count=len(shops),
        ))
    targeted: dict[int, str] = {}
    featured = set(pick_featured(records, total=FEATURED))
    for r in rows:
        if r.variant_id in featured:
            targeted.setdefault(r.listing_id, "featured")
    for _, _, cheap, dear in sorted(savings, reverse=True)[:TOP_SAVINGS]:
        targeted.setdefault(cheap, "top_saving_cheapest")
        targeted.setdefault(dear, "top_saving_dearest")
    for r in rows:
        before = previous.get(r.listing_id)
        if before and abs(r.price - before) / before * 100 > TARGET_MOVE_PCT:
            targeted.setdefault(r.listing_id, "moved")
    return targeted


def cross_shop_medians(rows: list[LatestRow]) -> dict[int, float]:
    """Per product, the median of its shops' latest USD prices (one row per shop)."""
    per_product: dict[int, dict[int, float]] = {}
    for r in rows:
        if r.price_usd is None:
            continue
        held = per_product.setdefault(r.variant_id, {}).get(r.shop_id)
        if held is None or r.price_usd < held:
            per_product[r.variant_id][r.shop_id] = r.price_usd
    return {pid: median(shops.values()) for pid, shops in per_product.items() if len(shops) >= 2}


def summarise(checks: list[Verdict]) -> dict[str, Any]:
    out: dict[str, Any] = {"checked": len(checks)}
    for c in checks:
        out[c.verdict] = out.get(c.verdict, 0) + 1
    out["failures"] = sum(1 for c in checks if c.is_failure)
    out["blocked"] = any(c.verdict == "BLOCKED" for c in checks)
    return out


# --- reading the database and the retailers ---------------------------------------------------


def previous_prices(db: Session) -> dict[int, float]:
    """The observation before the latest, per listing, for the movers target."""
    ranked = select(
        PriceObservation.listing_id,
        PriceObservation.price,
        func.row_number()
        .over(
            partition_by=PriceObservation.listing_id,
            order_by=(PriceObservation.observed_at.desc(), PriceObservation.id.desc()),
        )
        .label("rn"),
    ).subquery()
    return {
        r[0]: float(r[1])
        for r in db.execute(select(ranked.c.listing_id, ranked.c.price).where(ranked.c.rn == 2))
    }


def latest_run_status(db: Session, source: Source) -> str | None:
    run = db.scalar(
        select(CollectionRun).where(CollectionRun.source_id == source.id)
        .order_by(CollectionRun.started_at.desc()).limit(1)
    )
    return run.status if run else None


def verify_source(
    db: Session,
    run: VerificationRun,
    source: Source,
    rows: list[LatestRow],
    *,
    targeted: dict[int, str],
    medians: dict[int, float],
    rates: FxRates,
    n: int,
    reader: Reader | None = None,
    sleeper: Sleeper = time.sleep,
    now: Callable[[], datetime] = lambda: datetime.now(UTC),
) -> dict[str, Any]:
    """One source, one host, in order; abandoned at the first refusal."""
    collector = COLLECTORS.get(source.slug)
    if collector is None:
        return {"checked": 0, "skipped": "no collector registered"}
    if getattr(collector, "rendered", False):
        n = min(n, RENDERED_CAP)
    by_listing = {r.listing_id: r for r in rows}
    plan = plan_sample(list(by_listing), targeted, n, run.seed)
    listings = {
        l.id: l for l in db.scalars(select(Listing).where(Listing.id.in_(plan.listing_ids)))
    }
    shop_code = {
        loc.id: loc.code for loc in db.scalars(select(Shop).where(Shop.id.in_({r.shop_id for r in rows})))
    }
    read = reader or (lambda c, ref: c.read_one(ref))
    delay = float(source.delay_seconds or 0)
    verdicts: list[Verdict] = []
    for i, listing_id in enumerate(plan.listing_ids):
        published = by_listing[listing_id]
        listing = listings[listing_id]
        ref = _ref(listing, shop_code.get(listing.shop_id, published.shop_code))
        if i and delay:
            sleeper(delay)
        live: RawListing | None = None
        try:
            live = read(collector, ref)
            verdict = compare(
                published, live,
                median_usd=medians.get(published.variant_id),
                rate=rates.rates.get((live.currency if live else published.currency or "").upper()),
            )
        except PageGone as exc:
            # The listing is gone, which is this check's finding and nothing about the host:
            # the same verdict a read_one that answers None produces. Before the fetch port
            # told the two apart, one delisted bottle abandoned the whole source at the first
            # check and the run reported the retailer as refusing us.
            verdict = Verdict("GONE", f"the page no longer offers this listing ({exc})"[:500])
        except (SourceBlocked, RobotsUnavailable) as exc:
            verdict = Verdict("BLOCKED", str(exc)[:500])
        except FetchError as exc:
            verdict = Verdict("FETCH_ERROR", str(exc)[:500])
        except Exception as exc:  # the parser threw: that is the finding
            verdict = Verdict("PARSE_FAIL", f"{type(exc).__name__}: {exc}"[:500])
            logger.exception("verify_parse_fail source=%s listing=%s", source.slug, listing_id)
        check = VerificationCheck(
            run_id=run.id,
            listing_id=listing_id,
            observation_id=_latest_observation_id(db, listing_id),
            source_id=source.id,
            live_price=live.price if live else None,
            live_currency=live.currency if live else None,
            live_quantity_ml=live.quantity_ml if live else None,
            live_name=(live.name or "")[:400] if live else None,
            live_gtin=live.gtin if live else None,
            live_in_stock=live.in_stock if live else None,
            live_was_price=live.was_price if live else None,
            verdict=verdict.verdict,
            detail=f"[{plan.reason(listing_id)}] {verdict.detail}".strip(),
            checked_at=now(),
            url=published.url,
        )
        db.add(check)
        if verdict.verdict == "PRICE_MOVED" and live is not None:
            db.add(_moved_observation(listing_id, live, rates, now()))
        verdicts.append(verdict)
        db.commit()
        if verdict.verdict == "BLOCKED":
            logger.warning("verify_blocked source=%s after=%d detail=%s", source.slug, i, verdict.detail)
            break
    summary = summarise(verdicts)
    summary["planned"] = len(plan.listing_ids)
    summary["targeted"] = len(plan.targeted)
    return summary


def _ref(listing: Listing, code: str):
    from app.services.collectors.base import ListingRef

    return ListingRef(source_sku=listing.source_sku, url=listing.url, shop_code=code)


def _latest_observation_id(db: Session, listing_id: int) -> int | None:
    return db.scalar(
        select(PriceObservation.id).where(PriceObservation.listing_id == listing_id)
        .order_by(PriceObservation.observed_at.desc(), PriceObservation.id.desc()).limit(1)
    )


def _moved_observation(listing_id: int, live: RawListing, rates: FxRates, at: datetime) -> PriceObservation:
    was = live.was_price if live.was_price and live.was_price > live.price else None
    return PriceObservation(
        listing_id=listing_id,
        price=live.price,
        currency=live.currency,
        price_usd=rates.to_usd(live.price, live.currency),
        was_price=was,
        price_type=live.price_type,
        in_stock=live.in_stock,
        observed_at=at,
        run_id=None,
        source_kind="verify",
        fx_rate=rates.rates.get((live.currency or "").upper()),
    )


def run_verification(
    db: Session,
    *,
    n: int = DEFAULT_N,
    sources: list[str] | None = None,
    seed: int | None = None,
    mode: str = "on_demand",
    rates: FxRates | None = None,
    reader: Reader | None = None,
    sleeper: Sleeper = time.sleep,
) -> VerificationRun:
    """Every eligible source in turn (one host at a time), recorded as one run.

    Eligible: enabled, has a collector, and its latest collection was not
    blocked. Per-source summaries land in `per_source`; the checks are rows.
    """
    seed = random.randrange(1 << 30) if seed is None else seed
    run = VerificationRun(started_at=datetime.now(UTC), seed=seed, n=n, per_source={}, mode=mode)
    db.add(run)
    db.commit()
    rates = rates or load_rates()
    rows = latest_rows(db)
    visible = [r for r in rows if r.visible]
    targeted = targets_from_rows(visible, previous_prices(db))
    medians = cross_shop_medians(rows)
    by_source: dict[str, list[LatestRow]] = {}
    for r in rows:
        by_source.setdefault(r.source_slug or "", []).append(r)
    per_source: dict[str, Any] = {}
    for source in db.scalars(select(Source).order_by(Source.slug)):
        if sources and source.slug not in sources:
            continue
        if not source.enabled:
            per_source[source.slug] = {"checked": 0, "skipped": "source disabled"}
            continue
        status = latest_run_status(db, source)
        if status == "blocked":
            per_source[source.slug] = {"checked": 0, "skipped": "latest collection was blocked"}
            continue
        source_rows = by_source.get(source.slug, [])
        if not source_rows:
            per_source[source.slug] = {"checked": 0, "skipped": "no published rows"}
            continue
        per_source[source.slug] = verify_source(
            db, run, source, source_rows, targeted=targeted, medians=medians, rates=rates,
            n=n, reader=reader, sleeper=sleeper,
        )
        run.per_source = dict(per_source)
        db.commit()
    run.finished_at = datetime.now(UTC)
    run.per_source = per_source
    db.commit()
    return run


# --- what the rows mean for the site --------------------------------------------------------


def blocked_sources(db: Session) -> dict[int, list[VerificationCheck]]:
    """Sources with an uncleared correctness failure: publication blocked."""
    out: dict[int, list[VerificationCheck]] = {}
    for check in db.scalars(
        select(VerificationCheck).where(
            VerificationCheck.verdict.in_(CORRECTNESS_FAILURES),
            VerificationCheck.cleared_at.is_(None),
            VerificationCheck.source_id.isnot(None),
        ).order_by(VerificationCheck.checked_at.desc())
    ):
        out.setdefault(check.source_id, []).append(check)
    return out


def clear_checks(db: Session, check_ids: list[int], *, by: str, note: str) -> int:
    """A human clears failing checks; who and when are the record."""
    account = db.scalar(select(Account).where(Account.display_name == by))
    if account is None:
        raise ValueError(f"no account named {by!r}; a clearance needs one")
    now = datetime.now(UTC)
    cleared = 0
    for check in db.scalars(select(VerificationCheck).where(VerificationCheck.id.in_(check_ids))):
        if check.cleared_at is None:
            check.cleared_at, check.cleared_by, check.cleared_note = now, account.id, note
            cleared += 1
    db.commit()
    return cleared


def spot_checked(db: Session, source_id: int) -> datetime | None:
    """When the source last had a run with checks and zero correctness failures.

    This is the only date the public site may show per source ("Spot-checked
    <date>"), and only once such a run exists. Never a rate, never a list.
    """
    runs = db.execute(
        select(VerificationRun.id, VerificationRun.finished_at)
        .join(VerificationCheck, VerificationCheck.run_id == VerificationRun.id)
        .where(VerificationCheck.source_id == source_id, VerificationRun.finished_at.isnot(None))
        .group_by(VerificationRun.id)
        .order_by(VerificationRun.finished_at.desc())
    ).all()
    for run_id, finished_at in runs:
        failures = db.scalar(
            select(func.count(VerificationCheck.id)).where(
                VerificationCheck.run_id == run_id,
                VerificationCheck.source_id == source_id,
                VerificationCheck.verdict.in_((*CORRECTNESS_FAILURES, "BLOCKED")),
            )
        )
        if not failures:
            return finished_at
    return None


def pass_rates(db: Session) -> dict[int, dict[str, Any]]:
    """Rolling per-source counts over every check ever made. Owner area only."""
    out: dict[int, dict[str, Any]] = {}
    for source_id, verdict, count in db.execute(
        select(VerificationCheck.source_id, VerificationCheck.verdict, func.count(VerificationCheck.id))
        .where(VerificationCheck.source_id.isnot(None))
        .group_by(VerificationCheck.source_id, VerificationCheck.verdict)
    ):
        entry = out.setdefault(source_id, {"checks": 0, "judged": 0, "pass": 0, "failures": 0})
        entry["checks"] += count
        entry[verdict] = count
        # A refusal or a timeout is not a judgement on the data, so it is
        # outside the rate; GONE and REVIEW are judgements that did not pass.
        if verdict not in ("BLOCKED", "FETCH_ERROR"):
            entry["judged"] += count
        if verdict in ("PASS", "PRICE_MOVED"):
            entry["pass"] += count
        if verdict in CORRECTNESS_FAILURES:
            entry["failures"] += count
    for entry in out.values():
        entry["pass_rate"] = round(entry["pass"] / entry["judged"], 3) if entry["judged"] else None
    return out