"""The merge session: the queue a person works through, and what confirming or rejecting a
pair does (Stream M, task M6).

Sources of truth: this module, `services/suggest.py` (what fills the queue),
`services/merges.py` (what a confirm applies), `routers/collectors.py` (the routes),
`tests/test_merge_session.py`. Rian, 12 Sep: "plow through and confirm or reject". Every
count here is computed on request; nothing is stored but the decision itself.

A pair is shown side by side with what a person needs to decide it: at product level the
two rows with their shops, prices, barcodes and images; at line level the two lines with
their product variants, sizes, attribute_values and airports; at brand level the two brands with their
lines and the spellings the shops use. The preferred name defaults to the spelling most
listings carry. Confirm applies through `merges.merge_product_variants` (product) or the alias
functions (brand, line) and records who decided; reject records the same and the pair
never resurfaces; skip is the person's, recorded nowhere.
"""

from __future__ import annotations

from collections import Counter, defaultdict
from datetime import UTC, datetime, timedelta
from typing import Any

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

from app.models import Brand, Listing, Shop, Suggestion, PriceObservation, ProductVariant, ProductLine
from app.services import collector_view, keying, merges, overrides, product_lines, suggest

LEVELS = suggest.LEVELS


def _iso(value: datetime | None) -> str | None:
    return value.astimezone(UTC).isoformat(timespec="minutes") if value else None


# --------------------------------------------------------------------------- progress

def progress(db: Session) -> dict[str, Any]:
    """What is left at each level, and what this person and everyone decided."""
    remaining = {level: 0 for level in LEVELS}
    for level, n in db.execute(
        select(Suggestion.level, func.count(Suggestion.id))
        .where(Suggestion.is_open()).group_by(Suggestion.level)
    ):
        remaining[level] = int(n)
    since = datetime.now(UTC) - timedelta(hours=24)
    decided = Suggestion.decided_by.isnot(None)
    return {
        "remaining": remaining,
        "decided_today": int(db.scalar(select(func.count(Suggestion.id)).where(decided, Suggestion.decided_at >= since)) or 0),
        "decided_total": int(db.scalar(select(func.count(Suggestion.id)).where(decided)) or 0),
        "same_total": int(db.scalar(select(func.count(Suggestion.id)).where(decided, Suggestion.decision == "same")) or 0),
        "separate_total": int(db.scalar(select(func.count(Suggestion.id)).where(decided, Suggestion.decision == "separate")) or 0),
    }


# --------------------------------------------------------------------------- sides

def _shops_of(db: Session, variant_ids: list[int]) -> dict[int, list[dict[str, Any]]]:
    """Each product's listings with the newest price, as the product variants table shows them."""
    places: dict[int, list[dict[str, Any]]] = defaultdict(list)
    if not variant_ids:
        return places
    listing_rows = db.execute(
        select(Listing.id, Listing.variant_id, Shop.iata, Shop.code, Shop.visible, Listing.url)
        .join(Shop, Shop.id == Listing.shop_id)
        .where(Listing.variant_id.in_(variant_ids), Listing.ignored_at.is_(None))
    ).all()
    latest: dict[int, PriceObservation] = {}
    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, variant_id, iata, code, visible, url in listing_rows:
        obs = latest.get(listing_id)
        places[variant_id].append({
            "iata": iata or code, "visible": bool(visible), "url": url,
            "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,
        })
    return places


def _product_sides(db: Session, ids: list[int]) -> dict[int, dict[str, Any]]:
    product_variants = {p.id: p for p in db.scalars(select(ProductVariant).where(ProductVariant.id.in_(ids)))} if ids else {}
    brands = {b.id: b for b in db.scalars(select(Brand))}
    product_line_ids = {p.product_line_id for p in product_variants.values() if p.product_line_id}
    lines = {row.id: row for row in db.scalars(select(ProductLine).where(ProductLine.id.in_(list(product_line_ids))))} if product_line_ids else {}
    shops = _shops_of(db, list(product_variants))
    out = {}
    for pid, p in product_variants.items():
        brand = product_lines.resolve_alias(brands, p.brand_id)
        line = product_lines.resolve_alias(lines, p.product_line_id) if p.product_line_id else None
        places = sorted(shops.get(pid, []), key=lambda s: (s["price_usd"] is None, s["price_usd"] or 0))
        out[pid] = {
            "id": pid, "name": p.name, "listed_brand": p.brand, "brand": brand.name if brand else p.brand,
            "line": line.name if line else None,
            "attribute": (p.attributes or {}).get("attribute") or None,
            "concentration": (p.attributes or {}).get("concentration"),
            "size": collector_view.quantity_label(p),
            "quantity_ml": p.quantity_ml, "gtin": p.gtin, "gtin_source": p.gtin_source,
            "image_url": p.image_url, "thumb_url": p.thumb_url, "image_source": p.image_source,
            "category": p.category, "vertical": p.vertical,
            "listings": len(places), "airports": len({s["iata"] for s in places}), "shops": places,
            "merged": p.merged_into_id is not None,
        }
    return out


def _line_sides(db: Session, ids: list[int]) -> dict[int, dict[str, Any]]:
    lines = {row.id: row for row in db.scalars(select(ProductLine).where(ProductLine.id.in_(ids)))} if ids else {}
    brands = {b.id: b for b in db.scalars(select(Brand))}
    product_variants = list(db.scalars(
        select(ProductVariant).where(ProductVariant.product_line_id.in_(ids), ProductVariant.merged_into_id.is_(None)).order_by(ProductVariant.id)
    )) if ids else []
    shops = _shops_of(db, [p.id for p in product_variants])
    by_line: dict[int, list[ProductVariant]] = defaultdict(list)
    for p in product_variants:
        by_line[p.product_line_id].append(p)
    out = {}
    for lid, row in lines.items():
        members = by_line.get(lid, [])
        airports = {s["iata"] for p in members for s in shops.get(p.id, [])}
        names = Counter(p.name for p in members)
        out[lid] = {
            "id": lid, "name": row.name, "slug": row.slug, "key": row.key,
            "brand": (brands.get(row.brand_id).name if row.brand_id in brands else None),
            "product_variants": len(members), "listings": sum(len(shops.get(p.id, [])) for p in members),
            "airports": len(airports), "airport_codes": sorted(airports),
            "sizes": sorted({int(p.quantity_ml) for p in members if p.quantity_ml}),
            "attribute_values": sorted({(p.attributes or {}).get("attribute") for p in members if (p.attributes or {}).get("attribute")}),
            "samples": [name for name, _ in names.most_common(6)],
            "image_url": next((p.image_url for p in members if p.image_url), None),
            "thumb_url": next((p.thumb_url for p in members if p.thumb_url), None),
            "merged": row.alias_of_id is not None,
        }
    return out


def _brand_sides(db: Session, ids: list[int]) -> dict[int, dict[str, Any]]:
    brands = {b.id: b for b in db.scalars(select(Brand).where(Brand.id.in_(ids)))} if ids else {}
    product_variants = list(db.scalars(
        select(ProductVariant).where(ProductVariant.brand_id.in_(ids), ProductVariant.merged_into_id.is_(None))
    )) if ids else []
    lines = {row.id: row for row in db.scalars(select(ProductLine).where(ProductLine.brand_id.in_(ids)))} if ids else {}
    shops = _shops_of(db, [p.id for p in product_variants])
    by_brand: dict[int, list[ProductVariant]] = defaultdict(list)
    for p in product_variants:
        by_brand[p.brand_id].append(p)
    out = {}
    for bid, row in brands.items():
        members = by_brand.get(bid, [])
        airports = {s["iata"] for p in members for s in shops.get(p.id, [])}
        line_names = Counter(lines[p.product_line_id].name for p in members if p.product_line_id in lines)
        out[bid] = {
            "id": bid, "name": row.name, "slug": row.slug,
            "product_variants": len(members), "listings": sum(len(shops.get(p.id, [])) for p in members),
            "airports": len(airports), "airport_codes": sorted(airports),
            "lines": [name for name, _ in line_names.most_common(10)], "line_count": len(line_names),
            "spellings": [{"name": n, "count": c} for n, c in Counter(p.brand for p in members if p.brand).most_common(6)],
            "samples": [p.name for p in members[:6]],
            "image_url": next((p.image_url for p in members if p.image_url), None),
            "thumb_url": next((p.thumb_url for p in members if p.thumb_url), None),
            "merged": row.alias_of_id is not None,
        }
    return out


_SIDES = {"product": _product_sides, "line": _line_sides, "brand": _brand_sides}


def _names(level: str, left: dict, right: dict) -> tuple[list[dict[str, Any]], str]:
    """The spellings a person may pick as the preferred name, and the default: the one the
    most listings carry (brand: the shops' own spellings; line and product: the two names)."""
    if level == "brand":
        counter: Counter = Counter()
        for side in (left, right):
            for spelling in side.get("spellings", []):
                counter[spelling["name"]] += spelling["count"]
        for side in (left, right):
            counter.setdefault(side["name"], 0)
        options = [{"name": n, "count": c} for n, c in counter.most_common()]
    else:
        weight = "listings"
        options = sorted(
            [{"name": left["name"], "count": left.get(weight, 0)}, {"name": right["name"], "count": right.get(weight, 0)}],
            key=lambda o: -o["count"],
        )
    return options, options[0]["name"] if options else ""


# --------------------------------------------------------------------------- the queue

def queue(db: Session, *, level: str, page: int = 1, per_page: int = 20) -> dict[str, Any]:
    """One page of the undecided pairs at a level, best score first, each side by side."""
    if level not in LEVELS:
        return {"level": level, "total": 0, "page": page, "per_page": per_page, "pairs": [], "progress": progress(db)}
    per_page = max(1, min(per_page, 100))
    page = max(1, page)
    where = (Suggestion.is_open(), Suggestion.level == level)
    total = int(db.scalar(select(func.count(Suggestion.id)).where(*where)) or 0)
    rows = db.scalars(
        select(Suggestion).where(*where)
        .order_by(Suggestion.score.desc().nullslast(), Suggestion.id)
        .offset((page - 1) * per_page).limit(per_page)
    ).all()
    ids = sorted({i for c in rows for i in (c.left_id, c.right_id) if i})
    sides = _SIDES[level](db, ids)
    pairs = []
    for c in rows:
        left, right = sides.get(c.left_id), sides.get(c.right_id)
        if left is None or right is None:
            continue
        options, default = _names(level, left, right)
        pairs.append({
            "id": c.id, "level": level, "score": float(c.score) if c.score is not None else None,
            "reason": c.reason, "why": (c.detail or {}).get("why"), "detected_at": _iso(c.detected_at),
            "detail": {k: v for k, v in (c.detail or {}).items() if k not in ("from", "to", "why")},
            "left": left, "right": right, "names": options, "default_name": default,
        })
    return {"level": level, "total": total, "page": page, "per_page": per_page, "pairs": pairs, "progress": progress(db)}


# --------------------------------------------------------------------------- decisions

class Refused(Exception):
    def __init__(self, code: str, summary: str) -> None:
        super().__init__(summary)
        self.code, self.summary = code, summary


def _pending(db: Session, suggestion_id: int) -> Suggestion:
    suggestion = db.get(Suggestion, suggestion_id)
    if suggestion is None:
        raise Refused("PAIR_NOT_FOUND", "No such pair.")
    if suggestion.decision is not None:
        raise Refused("PAIR_DECIDED", f"This pair was already decided: {suggestion.decision}.")
    if suggestion.level not in LEVELS or not suggestion.left_id or not suggestion.right_id:
        raise Refused("PAIR_UNLEVELLED", "This pair has no level yet; run `app.cli suggest` first.")
    return suggestion


def _pick(level: str, left, right, preferred_name: str | None, keep: str | None):
    """Which side survives: the one whose name the person chose, else the side they named,
    else the better-evidenced one (more listings, a barcode)."""
    if preferred_name:
        wanted = preferred_name.strip().lower()
        if left.name.strip().lower() == wanted and right.name.strip().lower() != wanted:
            return left, right
        if right.name.strip().lower() == wanted and left.name.strip().lower() != wanted:
            return right, left
    if keep == "left":
        return left, right
    if keep == "right":
        return right, left
    return None, None


def confirm(db: Session, suggestion_id: int, *, decided_by: int, preferred_name: str | None = None,
            keep: str | None = None, note: str | None = None, regenerate: bool = True,
            commit: bool = True, batch=None) -> dict[str, Any]:
    """Apply the pair and record the decision: one `suggestion` row (Confirm same, the survivor
    named) whose consequences are the merge or the alias and a typed name (spec §5). Returns what
    happened, for the page. The desk's batch passes its own `batch` and `regenerate=False,
    commit=False`, doing each once at the end."""
    from app.services.decisions import natural_keys, writer

    suggestion = _pending(db, suggestion_id)
    level = suggestion.level
    model = {"brand": Brand, "line": ProductLine, "product": ProductVariant}[level]
    left, right = db.get(model, suggestion.left_id), db.get(model, suggestion.right_id)
    if left is None or right is None:
        raise Refused("PAIR_SIDE_GONE", f"One {level} of this pair no longer exists.")
    if level == "product":
        if left.merged_into_id or right.merged_into_id:
            raise Refused("PAIR_SIDE_MERGED", "One product variant of this pair was already merged away.")
        if left.gtin and right.gtin and left.gtin != right.gtin and not (note and note.strip()):
            raise Refused("NOTE_REQUIRED", "The two rows carry different barcodes. Say which one wins and why; the record keeps both.")
    elif level == "line" and left.brand_id != right.brand_id:
        raise Refused("LINE_HOUSES_DIFFER", "The two product lines belong to different brands; confirm the brand alias first.")
    survivor, other = _pick(level, left, right, preferred_name, keep)
    if survivor is None:
        if level == "product":
            listing_counts = dict(db.execute(
                select(Listing.variant_id, func.count(Listing.id))
                .where(Listing.variant_id.in_([left.id, right.id])).group_by(Listing.variant_id)
            ).all())
            survivor = merges.pick_survivor([left, right], listing_counts)
        else:
            column = ProductVariant.brand_id if level == "brand" else ProductVariant.product_line_id
            n_left = int(db.scalar(select(func.count(ProductVariant.id)).where(column == left.id, ProductVariant.merged_into_id.is_(None))) or 0)
            n_right = int(db.scalar(select(func.count(ProductVariant.id)).where(column == right.id, ProductVariant.merged_into_id.is_(None))) or 0)
            survivor = left if n_left >= n_right else right
        other = right if survivor is left else left
    value = {"decision": "same", "survivor": natural_keys.build(survivor)[0],
             "name": preferred_name.strip()[:400] if preferred_name and preferred_name.strip() else None, "note": note}

    def run(b):
        try:
            return writer.record(b, "suggestion", suggestion, "decision", value, reason=note)
        except writer.Refused as exc:
            raise Refused(exc.code, exc.summary) from exc

    if batch is not None:
        row = run(batch)
        folded = {}
    else:
        with writer.batch(db, "desk", "individual", decided_by, commit=False) as b:
            row = run(b)
        folded = b.folded
    consequence = next((d for d in (batch or b).decisions if d.caused_by_id == row.id and d.field in ("alias_of", "merged_into")), None)
    cdetail = (consequence.detail or {}) if consequence is not None else {}
    if level == "product":
        result = {"rows_merged": 1, "survivor_id": survivor.id, "alias_id": other.id}
    elif level == "brand":
        result = {"lines_pointed": cdetail.get("lines_pointed", 0), "lines_created": cdetail.get("lines_created", 0),
                  "product_variants_rekeyed": (batch or b).row.summary.get("variants_rekeyed", len((batch or b).touched)),
                  "line_decisions_moved": 0, "line_decisions_kept": 0,
                  "groups_merged": folded.get("groups", 0), "rows_merged": folded.get("merged_rows", 0),
                  "survivor_id": survivor.id, "alias_id": other.id}
    else:
        result = {"product_variants_moved": len(cdetail.get("products_moved", [])), "groups_merged": folded.get("groups", 0),
                  "rows_merged": folded.get("merged_rows", 0), "survivor_id": survivor.id, "alias_id": other.id}
    refreshed = None
    if level in ("brand", "line") and regenerate:
        # A brand or a line that just folded may put new pairs within reach (the bare
        # "1 Million 10cl" beside the Eau de Toilette once Paco Rabanne is Rabanne).
        refreshed = suggest.generate(db)
    if commit:
        db.commit()
        keying.invalidate()
    return {"level": level, "suggestion_id": suggestion_id, "applied": result, "suggested": refreshed, "decision": str(row.uid)}


def keep_separate(db: Session, suggestion_id: int, *, decided_by: int, note: str | None = None,
                  commit: bool = True, batch=None) -> dict[str, Any]:
    """Record that the two are not one: a `suggestion` row (Keep separate); the pair never
    resurfaces and the automatic fold honours it (spec §5)."""
    from app.services.decisions import writer

    suggestion = _pending(db, suggestion_id)
    value = {"decision": "separate", "survivor": None, "name": None, "note": note}

    def run(b):
        try:
            return writer.record(b, "suggestion", suggestion, "decision", value, reason=note)
        except writer.Refused as exc:
            raise Refused(exc.code, exc.summary) from exc

    if batch is not None:
        run(batch)
    else:
        with writer.batch(db, "desk", "individual", decided_by, commit=False, tail=False) as b:
            run(b)
    if commit:
        db.commit()
    else:
        db.flush()
    return {"level": suggestion.level, "suggestion_id": suggestion_id, "decision": "separate"}


def refresh(db: Session) -> dict[str, int]:
    counts = suggest.generate(db)
    db.commit()
    return counts