"""What a review pass is given, and how the review knows it is done (Stream K12.5).

Sources of truth: this module, `docs/REVIEW-PASS.md` (the specification a pass follows),
`docs/REVIEW-PROCESS.md` sections 4 and 5, `cli_pass.py`, `tests/test_pass_packet.py`.

Model independence (the plan's hard constraint 1): a pass runs on staging today with Claude driven
by hand, and live from a button or a schedule through an API. So a pass uses ONE input, the packet
this module builds, and nothing else: no database reads, no conversation, no memory of a previous
pass. Everything a pass may know is here, and if the packet lacks something the pass says so and
stops. The packet is the same document whichever model reads it.

Three other things live here because the packet needs them:

* **The cross-divide candidates.** Two retailer populations cannot join on any key: one publishes
  barcodes and internal shorthand, the other clean names and no barcodes (the 18 Sep simulation's
  structural finding). A candidate pair is a barcoded product variant with no listed text beside
  a name-bearing one with no barcode, same brand family, same quantity (and ABV where both state
  one), some token overlap. It is a hint a pass reads and judges, never a queue row; it is what
  makes the cross-divide layer cheap, and its count per brand is the value order's first measure.
* **The fingerprint.** A hash over the brand family's live listed words (and collected names where
  no fragment exists). A settled pass stamps it with both versions; any of the three changing
  un-settles the brand, which is the terminal condition's honest half.
* **Status.** Per brand: parked, open, deferred, settled, unsettled or never; and one done line
  (`0 open, 0 deferred, 0 wants, 0 parked`) that is measured, not felt.
"""

from __future__ import annotations

import hashlib
import json
from typing import Any

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

from app.models import Brand, Listing, ProductLine, ProductVariant, Proposal, ProposalPass, Retailer, Shop, Suggestion
from app.models.catalog import IDENTITY_RULES_VERSION
from app.services import product_lines
from app.services.decisions import natural_keys
from app.services.proposal_rules import PROCESS_VERSION

#: The layers, outermost first (the plan's B4, C2). A decision in one may raise questions only in
#: the layers after it.
LAYERS = ("brands", "cross_divide", "product_lines", "product_variants", "attributes")

#: Token overlap below which two names are not offered as a cross-divide candidate. Deliberately
#: low: the pass judges, this only keeps "Boli 5,6% 12x0.5L" from being paired with every 1 L bottle.
CROSS_DIVIDE_MIN_OVERLAP = 1


def _family(db: Session, brand_slug: str) -> tuple[Brand | None, list[Brand]]:
    from app.services.proposals import _brand_family

    return _brand_family(db, brand_slug)


# ----------------------------------------------------------------------------- the two populations


def retailer_populations(db: Session) -> dict[str, dict[str, Any]]:
    """Per retailer, from its live listings: does it publish listed words, does it publish
    barcodes. Read, never typed: the populations are what the data says today."""
    rows = db.execute(
        select(Retailer.slug, func.count(Listing.id),
               func.count(Listing.id).filter(Listing.listed_name.isnot(None)),
               func.count(Listing.id).filter(ProductVariant.gtin.isnot(None)))
        .join(Shop, Shop.retailer_id == Retailer.id).join(Listing, Listing.shop_id == Shop.id)
        .join(ProductVariant, ProductVariant.id == Listing.variant_id)
        .where(Listing.ignored_at.is_(None)).group_by(Retailer.slug)).all()
    out = {}
    for slug, n, named, barcoded in rows:
        n = int(n or 0)
        out[slug] = {"listings": n, "named": int(named or 0), "barcoded": int(barcoded or 0),
                     "has_listed_words": n > 0 and named / n >= 0.5, "has_barcodes": n > 0 and barcoded / n >= 0.5}
    return out


def _tokens(text: str | None) -> set[str]:
    return {t for t in product_lines._fold(text or "").split() if len(t) > 1 and not t.isdigit()}


def cross_divide_candidates(db: Session, family_ids: list[int], *, limit: int = 500) -> list[dict[str, Any]]:
    """Candidate pairs across the barcode divide for one brand family (a hint, never a row).

    Side A: a live product variant with a barcode whose listings carry no listed name (the
    shorthand side). Side B: a live variant with no barcode and at least one listed name. Same
    quantity in ml (or both unstated), ABV equal where both state one, and at least
    CROSS_DIVIDE_MIN_OVERLAP tokens shared after the certain folds. A pair a person already kept
    separate is left out."""
    if not family_ids:
        return []
    variants = list(db.scalars(select(ProductVariant).where(ProductVariant.brand_id.in_(family_ids), ProductVariant.merged_into_id.is_(None))))
    if not variants:
        return []
    ids = [v.id for v in variants]
    named = set(db.scalars(select(Listing.variant_id).where(Listing.variant_id.in_(ids), Listing.listed_name.isnot(None),
                                                            Listing.ignored_at.is_(None)).distinct()))
    side_a = [v for v in variants if v.gtin and v.id not in named]
    side_b = [v for v in variants if not v.gtin and v.id in named]
    if not side_a or not side_b:
        return []
    separate = set()
    for left, right in db.execute(select(Suggestion.left_id, Suggestion.right_id).where(Suggestion.level == "product", Suggestion.decision == "separate")):
        separate.add((left, right)); separate.add((right, left))
    out = []
    for a in side_a:
        ta = _tokens(a.name)
        for b in side_b:
            if (a.id, b.id) in separate:
                continue
            if (a.quantity_ml or None) != (b.quantity_ml or None):
                continue
            if a.abv is not None and b.abv is not None and float(a.abv) != float(b.abv):
                continue
            shared = ta & _tokens(b.name)
            if len(shared) < CROSS_DIVIDE_MIN_OVERLAP:
                continue
            key, detail = natural_keys.pair_key("product", a.id, b.id, sides=(natural_keys.side_key("product", a), natural_keys.side_key("product", b)))
            out.append({"pair": key, "barcoded": {"key": natural_keys.side_key("product", a), "name": a.name, "gtin": a.gtin},
                        "named": {"key": natural_keys.side_key("product", b), "name": b.name},
                        "quantity_ml": a.quantity_ml, "shared_tokens": sorted(shared)})
            if len(out) >= limit:
                return out
    out.sort(key=lambda c: (-len(c["shared_tokens"]), c["pair"]))
    return out


# ----------------------------------------------------------------------------- the fingerprint


def fingerprint(db: Session, family_ids: list[int]) -> str:
    """A hash over the brand family's live listings' listed words, and the collected name of each
    listing with no fragment. Stable across hosts (natural listing keys), changed by any sweep that
    changes a word, untouched by a price."""
    rows = db.execute(
        select(Retailer.slug, Shop.code, Listing.source_sku, Listing.listed_name, Listing.listed_variant, Listing.listed_quantity_text,
               Listing.listed_category, Listing.listed_record_id, ProductVariant.name)
        .join(Shop, Listing.shop_id == Shop.id).join(Retailer, Shop.retailer_id == Retailer.id)
        .join(ProductVariant, ProductVariant.id == Listing.variant_id)
        .where(ProductVariant.brand_id.in_(family_ids or [0]), ProductVariant.merged_into_id.is_(None), Listing.ignored_at.is_(None))).all()
    lines = sorted(json.dumps([r[0], r[1], r[2], r[3], r[4], r[5], r[6], None if r[7] is not None else r[8]], ensure_ascii=False) for r in rows)
    return hashlib.sha256("\n".join(lines).encode("utf-8")).hexdigest()


# ----------------------------------------------------------------------------- status and layer


def _open_rows(db: Session, slugs: list[str]) -> list[Proposal]:
    live = select(ProposalPass.id).where(ProposalPass.kind == "session", ProposalPass.withdrawn_at.is_(None))
    return list(db.scalars(select(Proposal).where(Proposal.brand_slug.in_(slugs), Proposal.pass_id.in_(live),
                                                  Proposal.status.in_(("open", "deferred", "parked")))))


def _layer_of(row: Proposal) -> str:
    if row.entity_type == "brand" or row.natural_key.startswith("pair:brand:"):
        return "brands"
    if row.entity_type == "suggestion" and row.natural_key.startswith("pair:product:") or row.field == "merged_into":
        return "cross_divide" if (row.detail or {}).get("cross_divide") or (row.precedent or "").startswith("cross-divide") else "product_variants"
    if row.entity_type == "product_line" or row.field == "product_line":
        return "product_lines"
    if row.field.startswith("attribute:"):
        return "attributes"
    return "product_variants"


def layer_for(db: Session, brand: Brand, family: list[Brand]) -> str | None:
    """The outermost layer with an unanswered proposal, or None when nothing is open: the pass then
    surveys every layer itself and proposes in the outermost where it finds a question, or settles."""
    rows = [r for r in _open_rows(db, [b.slug for b in family]) if r.status == "open"]
    layers = {_layer_of(r) for r in rows}
    for layer in LAYERS:
        if layer in layers:
            return layer
    return None


def brand_status(db: Session, brand: Brand, family: list[Brand]) -> dict[str, Any]:
    slugs = [b.slug for b in family]
    rows = _open_rows(db, slugs)
    open_n = sum(1 for r in rows if r.status == "open")
    deferred = [r for r in rows if r.status == "deferred"]
    wants = sum(1 for r in deferred if (r.detail or {}).get("wants"))
    parked = [r for r in rows if r.status == "parked" and (r.detail or {}).get("parked_to_rian")]
    newest = db.scalar(select(ProposalPass).where(ProposalPass.scope_brand_slug.in_(slugs), ProposalPass.kind == "session",
                                                  ProposalPass.withdrawn_at.is_(None)).order_by(ProposalPass.loaded_at.desc(), ProposalPass.id.desc()))
    state, why = "never", "no pass has run on this brand"
    current = None
    if parked:
        state, why = "parked", f"{len(parked)} question(s) wait on the running list"
    elif open_n:
        state, why = "open", f"{open_n} proposal(s) wait for an answer"
    elif deferred:
        state, why = "deferred", f"{len(deferred)} deferred note(s) wait for the next pass"
    elif newest is not None and newest.verdict == "settled":
        current = fingerprint(db, [b.id for b in family])
        if newest.process_version != PROCESS_VERSION or newest.rules_version != IDENTITY_RULES_VERSION:
            state, why = "unsettled", f"version changed (settled under process {newest.process_version}, rules {newest.rules_version})"
        elif newest.fingerprint != current:
            state, why = "unsettled", "listed words changed since the settled pass"
        else:
            state, why = "settled", f"settled by {newest.name}"
    elif newest is not None:
        state, why = "open", f"the newest pass {newest.name} left no open row but did not declare the brand settled"
    return {"brand_slug": brand.slug, "state": state, "why": why, "open": open_n, "deferred": len(deferred), "wants": wants,
            "parked": len(parked), "layer": layer_for(db, brand, family), "newest_pass": newest.name if newest else None,
            "fingerprint": current}


def status(db: Session, brand_slug: str | None = None) -> dict[str, Any]:
    """Every brand's state and the one done line. With a slug, that brand only."""
    if brand_slug:
        brand, family = _family(db, brand_slug)
        if brand is None:
            raise ValueError(f"no brand {brand_slug!r}")
        return {"brands": [brand_status(db, brand, family)], "done": None}
    out = []
    for brand in db.scalars(select(Brand).where(Brand.alias_of_id.is_(None)).order_by(Brand.slug)):
        family = [brand, *db.scalars(select(Brand).where(Brand.alias_of_id == brand.id))]
        out.append(brand_status(db, brand, family))
    totals = {"open": sum(b["open"] for b in out), "deferred": sum(b["deferred"] for b in out), "wants": sum(b["wants"] for b in out),
              "parked": sum(b["parked"] for b in out), "not_settled": sum(1 for b in out if b["state"] != "settled"), "brands": len(out)}
    done = totals["open"] == 0 and totals["deferred"] == 0 and totals["parked"] == 0 and totals["not_settled"] == 0
    line = (f"done: 0 open, 0 deferred, 0 wants, 0 parked; {totals['brands']} brands settled" if done else
            f"not done: {totals['open']} open, {totals['deferred']} deferred, {totals['wants']} wants, {totals['parked']} parked; "
            f"{totals['not_settled']} of {totals['brands']} brands not settled")
    return {"brands": out, "totals": totals, "done": done, "line": line}


# ----------------------------------------------------------------------------- the packet


def packet(db: Session, brand_slug: str, *, layer: str | None = None) -> dict[str, Any]:
    """Everything a pass may know about one brand, as one JSON document (the plan's C1)."""
    from app.services import guidelines, precedents, proposals

    brand, family = _family(db, brand_slug)
    if brand is None:
        raise ValueError(f"no brand {brand_slug!r}")
    family_ids = [b.id for b in family]
    populations = retailer_populations(db)
    lines = list(db.scalars(select(ProductLine).where(ProductLine.brand_id.in_(family_ids), ProductLine.alias_of_id.is_(None)).order_by(ProductLine.name)))
    variants = list(db.scalars(select(ProductVariant).where(ProductVariant.brand_id.in_(family_ids), ProductVariant.merged_into_id.is_(None))
                               .order_by(ProductVariant.product_line_id, ProductVariant.name)))
    listings = db.execute(
        select(Listing, Shop.code, Shop.name, Retailer.slug).join(Shop, Listing.shop_id == Shop.id).join(Retailer, Shop.retailer_id == Retailer.id)
        .where(Listing.variant_id.in_([v.id for v in variants] or [0]), Listing.ignored_at.is_(None))).all()
    by_variant: dict[int, list[dict]] = {}
    for listing, code, shop_name, retailer in listings:
        variant = next(v for v in variants if v.id == listing.variant_id)
        by_variant.setdefault(listing.variant_id, []).append({
            "key": natural_keys.listing_key(listing), "retailer": retailer, "shop": code, "shop_name": shop_name, "sku": listing.source_sku,
            "url": listing.url, "listed_brand": listing.listed_brand, "listed_name": listing.listed_name, "listed_variant": listing.listed_variant,
            "listed_quantity_text": listing.listed_quantity_text, "listed_category": listing.listed_category, "listed_gtin": listing.listed_gtin,
            "has_fragment": listing.listed_record_id is not None,
            "collected_name": None if listing.listed_record_id is not None else variant.name,
            "population": {"has_listed_words": populations.get(retailer, {}).get("has_listed_words"),
                           "has_barcodes": populations.get(retailer, {}).get("has_barcodes")}})
    variant_rows = []
    for v in variants:
        key, detail = natural_keys.build(v)
        variant_rows.append({"key": key, "detail": detail, "name": v.name, "gtin": v.gtin, "vertical": v.vertical, "category": v.category,
                             "quantity_value": float(v.quantity_value) if v.quantity_value is not None else None, "quantity_unit": v.quantity_unit,
                             "quantity_ml": float(v.quantity_ml) if v.quantity_ml is not None else None, "abv": float(v.abv) if v.abv is not None else None,
                             "pack_count": v.pack_count, "form": v.form, "attributes": v.attributes or {},
                             "product_line": f"line:{next((l.uid for l in lines if l.id == v.product_line_id), None)}" if v.product_line_id else None,
                             "listings": by_variant.get(v.id, [])})
    line_rows = [{"key": natural_keys.build(l)[0], "slug": l.slug, "name": l.name, "product_line_key": l.key, "hidden": l.hidden, "indexed": l.indexed,
                  "variants": sum(1 for v in variants if v.product_line_id == l.id)} for l in lines]
    brand_rows = [{"key": natural_keys.build(b)[0], "slug": b.slug, "name": b.name, "alias_of": next((x.slug for x in family if x.id == b.alias_of_id), None)}
                  for b in family]
    principles = guidelines.path()
    state = brand_status(db, brand, family)
    reg = precedents.describe(db)
    overturned = [p for p in reg if p["status"] == "overturned"]
    for p in overturned:
        p["decisions_under"] = [{"uid": str(d.uid), "natural_key": d.natural_key, "field": d.field, "value": d.value}
                                for d in precedents.decisions_under(db, p["slug"])]
    return {
        "packet_version": 1,
        "process_version": PROCESS_VERSION, "rules_version": IDENTITY_RULES_VERSION,
        "brand": {"slug": brand.slug, "name": brand.name, "family": brand_rows},
        "layer": layer or state["layer"], "layers": list(LAYERS), "state": state,
        "fingerprint": fingerprint(db, family_ids),
        "product_lines": line_rows, "product_variants": variant_rows,
        "populations": populations,
        "cross_divide_candidates": cross_divide_candidates(db, family_ids),
        "hints": proposals.hints(db, brand.slug),
        "notes": proposals.deferred_notes(db, brand.slug),
        "parked": proposals.parked_questions(db, brand.slug),
        "precedents": reg,
        "principles": principles.read_text(encoding="utf-8") if principles else None,
        "attention": {"levels": [l for l in ("critical", "high", "medium", "low", "none")], "leave_high_after": __import__("app.services.attention", fromlist=["LEAVE_HIGH_AFTER"]).LEAVE_HIGH_AFTER},
    }