"""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 four 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, and what
is currently wrong. It writes nothing; the deciding half (confirming a merge, clearing a
check) stays with the commands that already record who decided.
"""

from __future__ import annotations

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,
    Source,
)
from app.models.quality import AuditSnapshot, RejectedObservation, VerificationCheck
from app.services import audit as audit_service
from app.services import catalog_queries as cq
from app.services import verify as verify_service
from app.services.collectors.registry import COLLECTORS

#: 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: 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]
    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 _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 duplicate queue: pairs the merge rules found but would not merge on their own.
    pending = db.execute(
        select(MergeCandidate).where(MergeCandidate.decision.is_(None))
        .order_by(MergeCandidate.score.desc()).limit(limit)
    ).scalars().all()
    pairs = []
    for candidate in pending:
        left, right = db.get(Product, candidate.product_id), db.get(Product, candidate.candidate_id)
        if left is None or right is None:
            continue
        pairs.append({
            "id": candidate.id, "reason": candidate.reason, "score": candidate.score,
            "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))
    ) 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}