"""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 products, sizes, variations and airports; at brand level the two houses with their
lines and the spellings the shops use. The preferred name defaults to the spelling most
listings carry. Confirm applies through `merges.merge_products` (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, Location, MergeCandidate, PriceObservation, Product, ProductLine
from app.services import collector_view, keying, lines as lines_service, merges, overrides, 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(MergeCandidate.level, func.count(MergeCandidate.id))
        .where(MergeCandidate.decision.is_(None)).group_by(MergeCandidate.level)
    ):
        remaining[level] = int(n)
    since = datetime.now(UTC) - timedelta(hours=24)
    decided = MergeCandidate.decided_by.isnot(None)
    return {
        "remaining": remaining,
        "decided_today": int(db.scalar(select(func.count(MergeCandidate.id)).where(decided, MergeCandidate.decided_at >= since)) or 0),
        "decided_total": int(db.scalar(select(func.count(MergeCandidate.id)).where(decided)) or 0),
        "merged_total": int(db.scalar(select(func.count(MergeCandidate.id)).where(decided, MergeCandidate.decision == "merged")) or 0),
        "kept_apart_total": int(db.scalar(select(func.count(MergeCandidate.id)).where(decided, MergeCandidate.decision == "kept_apart")) or 0),
    }


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

def _shops_of(db: Session, product_ids: list[int]) -> dict[int, list[dict[str, Any]]]:
    """Each product's listings with the newest price, as the products table shows them."""
    places: dict[int, list[dict[str, Any]]] = defaultdict(list)
    if not product_ids:
        return places
    listing_rows = db.execute(
        select(Listing.id, Listing.product_id, Location.iata, Location.code, Location.visible, Listing.url)
        .join(Location, Location.id == Listing.location_id)
        .where(Listing.product_id.in_(product_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, product_id, iata, code, visible, url in listing_rows:
        obs = latest.get(listing_id)
        places[product_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]]:
    products = {p.id: p for p in db.scalars(select(Product).where(Product.id.in_(ids)))} if ids else {}
    brands = {b.id: b for b in db.scalars(select(Brand))}
    line_ids = {p.line_id for p in products.values() if p.line_id}
    lines = {row.id: row for row in db.scalars(select(ProductLine).where(ProductLine.id.in_(list(line_ids))))} if line_ids else {}
    shops = _shops_of(db, list(products))
    out = {}
    for pid, p in products.items():
        house = lines_service.resolve_alias(brands, p.brand_id)
        line = lines_service.resolve_alias(lines, p.line_id) if p.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, "brand": p.brand, "house": house.name if house else p.brand,
            "line": line.name if line else None,
            "variation": (p.attributes or {}).get("variation") or None,
            "concentration": (p.attributes or {}).get("concentration"),
            "size": collector_view.quantity_label(p),
            "size_ml": p.size_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))}
    products = list(db.scalars(
        select(Product).where(Product.line_id.in_(ids), Product.merged_into_id.is_(None)).order_by(Product.id)
    )) if ids else []
    shops = _shops_of(db, [p.id for p in products])
    by_line: dict[int, list[Product]] = defaultdict(list)
    for p in products:
        by_line[p.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,
            "house": (brands.get(row.brand_id).name if row.brand_id in brands else None),
            "products": len(members), "listings": sum(len(shops.get(p.id, [])) for p in members),
            "airports": len(airports), "airport_codes": sorted(airports),
            "sizes": sorted({int(p.size_ml) for p in members if p.size_ml}),
            "variations": sorted({(p.attributes or {}).get("variation") for p in members if (p.attributes or {}).get("variation")}),
            "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.canonical_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 {}
    products = list(db.scalars(
        select(Product).where(Product.brand_id.in_(ids), Product.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 products])
    by_brand: dict[int, list[Product]] = defaultdict(list)
    for p in products:
        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.line_id].name for p in members if p.line_id in lines)
        out[bid] = {
            "id": bid, "name": row.name, "slug": row.slug,
            "products": 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.canonical_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 = (MergeCandidate.decision.is_(None), MergeCandidate.level == level)
    total = int(db.scalar(select(func.count(MergeCandidate.id)).where(*where)) or 0)
    rows = db.scalars(
        select(MergeCandidate).where(*where)
        .order_by(MergeCandidate.score.desc().nullslast(), MergeCandidate.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, candidate_id: int) -> MergeCandidate:
    candidate = db.get(MergeCandidate, candidate_id)
    if candidate is None:
        raise Refused("PAIR_NOT_FOUND", "No such pair.")
    if candidate.decision is not None:
        raise Refused("PAIR_DECIDED", f"This pair was already decided: {candidate.decision}.")
    if candidate.level not in LEVELS or not candidate.left_id or not candidate.right_id:
        raise Refused("PAIR_UNLEVELLED", "This pair has no level yet; run `app.cli suggest` first.")
    return candidate


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, candidate_id: int, *, decided_by: int, preferred_name: str | None = None,
            keep: str | None = None, note: str | None = None, regenerate: bool = True,
            commit: bool = True) -> dict[str, Any]:
    """Apply the pair and record the decision. Returns what happened, for the page.
    `regenerate` and `commit` default to the keyboard queue's behaviour (a brand or line
    confirm re-runs the suggestions and commits); the desk's batch passes False for both
    and does each once at the end (Stream L)."""
    candidate = _pending(db, candidate_id)
    level = candidate.level
    result: dict[str, Any]
    if level == "brand":
        left, right = db.get(Brand, candidate.left_id), db.get(Brand, candidate.right_id)
        if left is None or right is None:
            raise Refused("PAIR_SIDE_GONE", "One house of this pair no longer exists.")
        house, alias = _pick(level, left, right, preferred_name, keep)
        if house is None:
            n_left = int(db.scalar(select(func.count(Product.id)).where(Product.brand_id == left.id, Product.merged_into_id.is_(None))) or 0)
            n_right = int(db.scalar(select(func.count(Product.id)).where(Product.brand_id == right.id, Product.merged_into_id.is_(None))) or 0)
            house, alias = (left, right) if n_left >= n_right else (right, left)
        result = merges.apply_brand_alias(db, alias, house, decided_by=decided_by, preferred_name=preferred_name)
        result.update({"survivor_id": house.id, "alias_id": alias.id})
    elif level == "line":
        left, right = db.get(ProductLine, candidate.left_id), db.get(ProductLine, candidate.right_id)
        if left is None or right is None:
            raise Refused("PAIR_SIDE_GONE", "One line of this pair no longer exists.")
        canonical, alias = _pick(level, left, right, preferred_name, keep)
        if canonical is None:
            n_left = int(db.scalar(select(func.count(Product.id)).where(Product.line_id == left.id, Product.merged_into_id.is_(None))) or 0)
            n_right = int(db.scalar(select(func.count(Product.id)).where(Product.line_id == right.id, Product.merged_into_id.is_(None))) or 0)
            canonical, alias = (left, right) if n_left >= n_right else (right, left)
        try:
            result = merges.apply_line_alias(db, alias, canonical, decided_by=decided_by, preferred_name=preferred_name)
        except ValueError as exc:
            raise Refused("LINE_HOUSES_DIFFER", str(exc)) from exc
        result.update({"survivor_id": canonical.id, "alias_id": alias.id})
    else:
        left, right = db.get(Product, candidate.left_id), db.get(Product, candidate.right_id)
        if left is None or right is None:
            raise Refused("PAIR_SIDE_GONE", "One product of this pair no longer exists.")
        if left.merged_into_id or right.merged_into_id:
            raise Refused("PAIR_SIDE_MERGED", "One product 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.")
        survivor, other = _pick(level, left, right, preferred_name, keep)
        if survivor is None:
            listing_counts = dict(db.execute(
                select(Listing.product_id, func.count(Listing.id))
                .where(Listing.product_id.in_([left.id, right.id])).group_by(Listing.product_id)
            ).all())
            survivor = merges.pick_survivor([left, right], listing_counts)
            other = right if survivor is left else left
        merged = merges.merge_products(
            db, survivor, [other], reason="confirmed", merged_by=decided_by, note=note,
        )
        if preferred_name and preferred_name.strip():
            typed = preferred_name.strip()[:400]
            if typed not in (left.name, right.name):
                # A typed name is a decision (LT3 c): the override keeps the survivor's
                # collected spelling for the key, so `rederive` cannot re-key the survivor
                # away from its listings.
                existing = overrides.fields_of(db, "product", str(survivor.id)).get("name")
                collected = existing.collected_value if existing is not None and existing.collected_value else survivor.name
                overrides.decide(db, "product", str(survivor.id), "name", typed, set_by=decided_by,
                              reason="chosen at merge", collected_value=collected)
            survivor.name = typed
        result = {"rows_merged": merged, "survivor_id": survivor.id, "alias_id": other.id}
    candidate.decision = "merged"
    candidate.decided_by, candidate.decided_at = decided_by, datetime.now(UTC)
    candidate.detail = {**(candidate.detail or {}), "decision": {
        "survivor_id": result.get("survivor_id"), "preferred_name": preferred_name, "note": note}}
    db.flush()
    refreshed = None
    if level in ("brand", "line") and regenerate:
        # A house 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, "candidate_id": candidate_id, "applied": result, "suggested": refreshed}


def reject(db: Session, candidate_id: int, *, decided_by: int, note: str | None = None,
           commit: bool = True) -> dict[str, Any]:
    """Record that the two are not one; the pair never resurfaces."""
    candidate = _pending(db, candidate_id)
    candidate.decision = "kept_apart"
    candidate.decided_by, candidate.decided_at = decided_by, datetime.now(UTC)
    if note:
        candidate.detail = {**(candidate.detail or {}), "decision": {"note": note}}
    if commit:
        db.commit()
    else:
        db.flush()
    return {"level": candidate.level, "candidate_id": candidate_id, "decision": "kept_apart"}


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