"""The merge desk (Stream L, LT7): the brands table and the lines table a person confirms
suggested matches from in bulk, the batch that applies them as that person's decisions, and
the proposals a session files for a person to confirm, which the rules never withdraw.

Sources of truth: this module, `services/merge_session.py` (what one confirm or reject
does; the desk applies the same functions with `regenerate=False, commit=False`),
`services/suggest.py` (`RULE_REASONS`, which never contains `proposed`), `routers/collectors.py`,
`web/src/components/collectors/MergeDesk.tsx`, `tests/test_merge_desk.py`. Rian, 14 Sep:
"a merge desk beside the giant table where he approves suggested merges in bulk as his own
decisions". The keyboard queue cost one full regeneration per confirm (O(N) for a bulk
session); the batch regenerates once at the end and commits once. A `proposed` suggestion
(`app.cli propose --file`, `POST /merge/propose`) is the session's own reading of the data
written to a file a person approves from; its reason is outside `RULE_REASONS`, so
`suggest.generate` neither withdraws nor rescores it, and nothing is ever merged by filing
it. Every count here is read from the database on request.
"""

from __future__ import annotations

import json
import pathlib
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import case, func, or_, select
from sqlalchemy.orm import Session, aliased

from app.models import Account, Brand, Listing, Suggestion, ProductVariant, ProductLine
from app.services import audit_log, keying, merge_session, suggest

BATCH_MAX = 200
PROPOSED = "proposed"
_DESK_SORTS = ("name", "score", "product_variants", "listings", "lines")


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


def _best_pair(level: str):
    """The best undecided pair per side at a level: (side_id, suggestion_id, other_id, score,
    reason, why), one row per side, highest score first."""
    proposed = case((Suggestion.reason == PROPOSED, True),
                    (Suggestion.detail["proposed_by"].as_string().isnot(None), True), else_=False)
    left = select(
        Suggestion.left_id.label("side_id"), Suggestion.id.label("suggestion_id"),
        Suggestion.right_id.label("other_id"), Suggestion.score.label("score"),
        Suggestion.reason.label("reason"), Suggestion.detail["why"].as_string().label("why"),
        proposed.label("proposed"), Suggestion.detail["proposed_why"].as_string().label("proposed_why"),
    ).where(Suggestion.level == level, Suggestion.is_open())
    right = select(
        Suggestion.right_id, Suggestion.id, Suggestion.left_id, Suggestion.score,
        Suggestion.reason, Suggestion.detail["why"].as_string(), proposed,
        Suggestion.detail["proposed_why"].as_string(),
    ).where(Suggestion.level == level, Suggestion.is_open())
    both = left.union_all(right).subquery()
    ranked = select(
        both.c.side_id, both.c.suggestion_id, both.c.other_id, both.c.score, both.c.reason, both.c.why, both.c.proposed,
        both.c.proposed_why,
        func.row_number().over(partition_by=both.c.side_id, order_by=(both.c.score.desc().nullslast(), both.c.suggestion_id)).label("rn"),
    ).subquery()
    return select(ranked.c.side_id, ranked.c.suggestion_id, ranked.c.other_id, ranked.c.score, ranked.c.reason, ranked.c.why,
                  ranked.c.proposed, ranked.c.proposed_why).where(ranked.c.rn == 1).subquery()


def _decided_pairs(level: str):
    """The newest decided pair per side, so a decided row shows who and when."""
    left = select(Suggestion.left_id.label("side_id"), Suggestion.id.label("suggestion_id"),
                  Suggestion.decision.label("decision"), Suggestion.decided_by.label("decided_by"),
                  Suggestion.decided_at.label("decided_at")).where(Suggestion.level == level, Suggestion.decision.isnot(None))
    right = select(Suggestion.right_id, Suggestion.id, Suggestion.decision, Suggestion.decided_by,
                   Suggestion.decided_at).where(Suggestion.level == level, Suggestion.decision.isnot(None))
    both = left.union_all(right).subquery()
    ranked = select(both.c.side_id, both.c.suggestion_id, both.c.decision, both.c.decided_by, both.c.decided_at,
                    func.row_number().over(partition_by=both.c.side_id, order_by=both.c.decided_at.desc()).label("rn")).subquery()
    return select(ranked.c.side_id, ranked.c.suggestion_id, ranked.c.decision, ranked.c.decided_by, ranked.c.decided_at) \
        .where(ranked.c.rn == 1).subquery()


def _page(stmt, sort_expr: dict, sort: str | None, direction: str | None, page: int, per_page: int, db: Session):
    per_page = max(1, min(per_page, 500))
    page = max(1, page)
    total = int(db.scalar(select(func.count()).select_from(stmt.subquery())) or 0)
    e = sort_expr.get(sort or "", sort_expr["name"])
    order = e.desc().nullslast() if direction == "desc" else e.asc().nullsfirst()
    rows = db.execute(stmt.order_by(order, sort_expr["id"]).offset((page - 1) * per_page).limit(per_page)).all()
    return total, page, per_page, rows


def brands(db: Session, *, text: str | None = None, has_suggestion: bool | None = None, min_score: float | None = None,
           sort: str | None = "score", direction: str | None = "desc", page: int = 1, per_page: int = 100) -> dict[str, Any]:
    """One row per brand (a listed_brand row not aliased away): its lines, product variants and listings,
    the best undecided pair with the other side, score, reason and why, and the newest
    decision on it. Sorted in SQL."""
    other = aliased(Brand, name="other_brand")
    best = _best_pair("brand")
    decided = _decided_pairs("brand")
    lines_n = select(ProductLine.brand_id, func.count(ProductLine.id).label("n")).where(ProductLine.alias_of_id.is_(None)).group_by(ProductLine.brand_id).subquery()
    product_variants_n = select(ProductVariant.brand_id, func.count(ProductVariant.id).label("n")).where(ProductVariant.merged_into_id.is_(None)).group_by(ProductVariant.brand_id).subquery()
    listings_n = (
        select(ProductVariant.brand_id, func.count(Listing.id).label("n"))
        .join(Listing, Listing.variant_id == ProductVariant.id)
        .where(ProductVariant.merged_into_id.is_(None), Listing.ignored_at.is_(None)).group_by(ProductVariant.brand_id).subquery()
    )
    stmt = (
        select(
            Brand.id, Brand.name, Brand.slug, Brand.alias_of_id,
            func.coalesce(lines_n.c.n, 0).label("lines"), func.coalesce(product_variants_n.c.n, 0).label("product_variants"),
            func.coalesce(listings_n.c.n, 0).label("listings"),
            best.c.suggestion_id, best.c.other_id, other.name.label("other_name"), best.c.score, best.c.reason, best.c.why,
            best.c.proposed, best.c.proposed_why,
            decided.c.suggestion_id.label("decided_suggestion_id"), decided.c.decision, decided.c.decided_at,
            Account.display_name.label("decided_by"),
        )
        .select_from(Brand)
        .outerjoin(lines_n, lines_n.c.brand_id == Brand.id)
        .outerjoin(product_variants_n, product_variants_n.c.brand_id == Brand.id)
        .outerjoin(listings_n, listings_n.c.brand_id == Brand.id)
        .outerjoin(best, best.c.side_id == Brand.id)
        .outerjoin(other, other.id == best.c.other_id)
        .outerjoin(decided, decided.c.side_id == Brand.id)
        .outerjoin(Account, Account.id == decided.c.decided_by)
        .where(Brand.alias_of_id.is_(None))
    )
    if text:
        like = f"%{text.strip().lower()}%"
        stmt = stmt.where(or_(func.lower(Brand.name).like(like), func.lower(other.name).like(like)))
    if has_suggestion is not None:
        stmt = stmt.where(best.c.suggestion_id.isnot(None) if has_suggestion else best.c.suggestion_id.is_(None))
    if min_score is not None:
        stmt = stmt.where(best.c.score >= min_score)
    sort_expr = {"id": Brand.id, "name": Brand.name, "score": best.c.score, "product_variants": product_variants_n.c.n,
                 "listings": listings_n.c.n, "lines": lines_n.c.n}
    total, page, per_page, rows = _page(stmt, sort_expr, sort, direction, page, per_page, db)
    return {"total": total, "page": page, "per_page": per_page, "counts": counts(db, "brand"), "rows": [
        {"id": r.id, "name": r.name, "slug": r.slug, "lines": int(r.lines), "product_variants": int(r.product_variants), "listings": int(r.listings),
         "suggestion_id": r.suggestion_id, "other_id": r.other_id, "other_name": r.other_name,
         "score": float(r.score) if r.score is not None else None, "reason": r.reason, "why": r.why,
         "proposed": bool(r.proposed), "proposed_why": r.proposed_why,
         "decided": {"suggestion_id": r.decided_suggestion_id, "decision": r.decision, "by": r.decided_by, "at": _iso(r.decided_at)}
         if r.decided_suggestion_id else None}
        for r in rows
    ]}


def lines(db: Session, *, brand: str | None = None, vertical: str | None = None, kind: str | None = None,
          reason: str | None = None, min_score: float | None = None, text: str | None = None,
          sort: str | None = "score", direction: str | None = "desc", page: int = 1, per_page: int = 100) -> dict[str, Any]:
    """One row per line (not aliased away) within its brand, the same shape as the brands
    table, plus the undecided product-level pairs under it (their count and the best one)."""
    other = aliased(ProductLine, name="other_line")
    best = _best_pair("line")
    decided = _decided_pairs("line")
    product_variants_n = select(ProductVariant.product_line_id, func.count(ProductVariant.id).label("n")).where(ProductVariant.merged_into_id.is_(None)).group_by(ProductVariant.product_line_id).subquery()
    listings_n = (
        select(ProductVariant.product_line_id, func.count(Listing.id).label("n"))
        .join(Listing, Listing.variant_id == ProductVariant.id)
        .where(ProductVariant.merged_into_id.is_(None), Listing.ignored_at.is_(None)).group_by(ProductVariant.product_line_id).subquery()
    )
    vertical_of_line = (
        select(ProductVariant.product_line_id, func.max(ProductVariant.vertical).label("vertical"),
               func.max(ProductVariant.attributes["attribute_kind"].as_string()).label("kind"))
        .where(ProductVariant.merged_into_id.is_(None)).group_by(ProductVariant.product_line_id).subquery()
    )
    pair_left = aliased(ProductVariant, name="pair_left")
    product_pairs = (
        select(pair_left.product_line_id.label("product_line_id"), func.count(Suggestion.id).label("n"), func.max(Suggestion.score).label("best"))
        .join(pair_left, pair_left.id == Suggestion.left_id)
        .where(Suggestion.level == "product", Suggestion.is_open())
        .group_by(pair_left.product_line_id).subquery()
    )
    stmt = (
        select(
            ProductLine.id, ProductLine.name, ProductLine.slug, ProductLine.key, Brand.name.label("brand"), Brand.id.label("brand_of_id"),
            func.coalesce(product_variants_n.c.n, 0).label("product_variants"), func.coalesce(listings_n.c.n, 0).label("listings"),
            vertical_of_line.c.vertical, vertical_of_line.c.kind,
            best.c.suggestion_id, best.c.other_id, other.name.label("other_name"), best.c.score, best.c.reason, best.c.why,
            best.c.proposed, best.c.proposed_why,
            func.coalesce(product_pairs.c.n, 0).label("product_pairs"), product_pairs.c.best.label("product_pairs_best"),
            decided.c.suggestion_id.label("decided_suggestion_id"), decided.c.decision, decided.c.decided_at,
            Account.display_name.label("decided_by"),
        )
        .select_from(ProductLine)
        .join(Brand, Brand.id == ProductLine.brand_id)
        .outerjoin(product_variants_n, product_variants_n.c.product_line_id == ProductLine.id)
        .outerjoin(listings_n, listings_n.c.product_line_id == ProductLine.id)
        .outerjoin(vertical_of_line, vertical_of_line.c.product_line_id == ProductLine.id)
        .outerjoin(product_pairs, product_pairs.c.product_line_id == ProductLine.id)
        .outerjoin(best, best.c.side_id == ProductLine.id)
        .outerjoin(other, other.id == best.c.other_id)
        .outerjoin(decided, decided.c.side_id == ProductLine.id)
        .outerjoin(Account, Account.id == decided.c.decided_by)
        .where(ProductLine.alias_of_id.is_(None))
    )
    if brand:
        like = f"%{brand.strip().lower()}%"
        stmt = stmt.where(or_(func.lower(Brand.name).like(like), func.lower(Brand.slug).like(like)))
    if text:
        like = f"%{text.strip().lower()}%"
        stmt = stmt.where(or_(func.lower(ProductLine.name).like(like), func.lower(other.name).like(like)))
    if vertical:
        stmt = stmt.where(vertical_of_line.c.vertical == vertical)
    if kind:
        stmt = stmt.where(vertical_of_line.c.kind == kind)
    if reason == PROPOSED:
        stmt = stmt.where(best.c.proposed.is_(True))
    elif reason:
        stmt = stmt.where(best.c.reason == reason)
    if min_score is not None:
        stmt = stmt.where(best.c.score >= min_score)
    sort_expr = {"id": ProductLine.id, "name": ProductLine.name, "score": best.c.score, "product_variants": product_variants_n.c.n,
                 "listings": listings_n.c.n, "lines": product_pairs.c.n, "brand": Brand.name}
    total, page, per_page, rows = _page(stmt, sort_expr, sort, direction, page, per_page, db)
    return {"total": total, "page": page, "per_page": per_page, "counts": counts(db, "line"), "rows": [
        {"id": r.id, "name": r.name, "slug": r.slug, "key": r.key, "brand": r.brand, "brand_of_id": r.brand_of_id,
         "product_variants": int(r.product_variants), "listings": int(r.listings), "vertical": r.vertical, "kind": r.kind,
         "suggestion_id": r.suggestion_id, "other_id": r.other_id, "other_name": r.other_name,
         "score": float(r.score) if r.score is not None else None, "reason": r.reason, "why": r.why,
         "proposed": bool(r.proposed), "proposed_why": r.proposed_why,
         "product_pairs": int(r.product_pairs), "product_pairs_best": float(r.product_pairs_best) if r.product_pairs_best is not None else None,
         "decided": {"suggestion_id": r.decided_suggestion_id, "decision": r.decision, "by": r.decided_by, "at": _iso(r.decided_at)}
         if r.decided_suggestion_id else None}
        for r in rows
    ]}


def counts(db: Session, level: str) -> dict[str, int]:
    """Remaining, decided today and decided in all, at one level, from the database."""
    since = datetime.now(UTC) - timedelta(hours=24)
    remaining = int(db.scalar(select(func.count(Suggestion.id)).where(Suggestion.level == level, Suggestion.is_open())) or 0)
    today = int(db.scalar(select(func.count(Suggestion.id)).where(
        Suggestion.level == level, Suggestion.decided_by.isnot(None), Suggestion.decided_at >= since)) or 0)
    total = int(db.scalar(select(func.count(Suggestion.id)).where(Suggestion.level == level, Suggestion.decided_by.isnot(None))) or 0)
    proposed = int(db.scalar(select(func.count(Suggestion.id)).where(
        Suggestion.level == level, Suggestion.is_open(),
        or_(Suggestion.reason == PROPOSED, Suggestion.detail["proposed_by"].as_string().isnot(None)))) or 0)
    return {"remaining": remaining, "decided_today": today, "decided_total": total, "proposed": proposed}


# --------------------------------------------------------------------------- the batch

def batch(db: Session, *, decided_by: int, confirm: list[dict[str, Any]], keep_separate: list[dict[str, Any]]) -> dict[str, Any]:
    """Apply up to `BATCH_MAX` decisions as one person's: each pair in id order through
    `merge_session.confirm` / `reject` with `regenerate=False, commit=False`; a refusal
    (`NOTE_REQUIRED` on a barcode veto, `LINE_HOUSES_DIFFER`, `PAIR_DECIDED`, a side gone) is
    reported per pair and the rest proceed; suggestions regenerate ONCE at the end, one
    commit, one keying invalidation, one audit row per decision."""
    if len(confirm) + len(keep_separate) > BATCH_MAX:
        raise merge_session.Refused("BATCH_TOO_LARGE", f"At most {BATCH_MAX} decisions in one batch.")
    applied: list[dict[str, Any]] = []
    refused: list[dict[str, Any]] = []
    work = sorted(
        [("confirm", item) for item in confirm] + [("keep_separate", item) for item in keep_separate],
        key=lambda pair: int(pair[1].get("suggestion_id") or 0),
    )
    levels_touched: set[str] = set()
    from app.services.decisions import writer

    ledger = writer.batch(db, "desk", "individual", decided_by, commit=False,
                          scope={"suggestion_ids": [int(i.get("suggestion_id") or 0) for _, i in work]})
    b = ledger.__enter__()
    for action, item in work:
        suggestion_id = int(item.get("suggestion_id") or 0)
        try:
            if action == "confirm":
                result = merge_session.confirm(
                    db, suggestion_id, decided_by=decided_by, preferred_name=item.get("preferred_name"),
                    keep=item.get("keep") if item.get("keep") in ("left", "right") else None, note=item.get("note"),
                    regenerate=False, commit=False, batch=b,
                )
            else:
                result = merge_session.keep_separate(db, suggestion_id, decided_by=decided_by, note=item.get("note"), commit=False, batch=b)
            levels_touched.add(result["level"])
            applied.append({"suggestion_id": suggestion_id, "action": action, "level": result["level"],
                            "applied": result.get("applied"), "decision": result.get("decision")})
            audit_log.record(f"merge.{action}", entity_type=result["level"], entity_key=str(suggestion_id),
                             detail={"batch": True, "applied": result.get("applied"), "preferred_name": item.get("preferred_name"),
                                     "note": item.get("note")}, account_id=decided_by)
        except merge_session.Refused as exc:
            db.flush()
            b.refused.append({"suggestion_id": suggestion_id, "code": exc.code})
            refused.append({"suggestion_id": suggestion_id, "action": action, "error_code": exc.code, "summary": exc.summary})
    ledger.__exit__(None, None, None)
    suggested = None
    if levels_touched & {"brand", "line"}:
        suggested = suggest.generate(db)
    db.commit()
    keying.invalidate()
    return {"applied": applied, "refused": refused, "suggested": suggested, "batch": str(b.uid),
            "counts": {level: counts(db, level) for level in suggest.LEVELS}}


# --------------------------------------------------------------------------- proposals

def propose(db: Session, entries: list[dict[str, Any]], *, proposed_by: str, check: bool = False) -> dict[str, Any]:
    """Insert `{level, left_id, right_id, score, why}` entries as suggestions with reason
    `proposed` and `detail.proposed_by`, `left_id < right_id`; a pair the rules already hold
    undecided is ENDORSED instead (its `detail` gains `proposed_by` and `proposed_why`, so
    "tick all proposed" on the desk finds it; its reason and score stay the rule's); a decided
    pair or one with a side that no longer exists is skipped and named; nothing is merged.
    `check` reports and writes nothing."""
    inserted = endorsed = skipped_present = skipped_missing = invalid = 0
    named: list[str] = []
    rows = {(c.level, c.left_id, c.right_id): c for c in db.scalars(select(Suggestion)) if c.left_id and c.right_id}
    existing = set(rows)
    alive = {
        "brand": set(db.scalars(select(Brand.id).where(Brand.alias_of_id.is_(None)))),
        "line": set(db.scalars(select(ProductLine.id).where(ProductLine.alias_of_id.is_(None)))),
        "product": set(db.scalars(select(ProductVariant.id).where(ProductVariant.merged_into_id.is_(None)))),
    }
    for entry in entries:
        level = entry.get("level")
        try:
            left, right = int(entry["left_id"]), int(entry["right_id"])
            score = float(entry.get("score", 0.5))
        except (KeyError, TypeError, ValueError):
            invalid += 1
            continue
        if level not in suggest.LEVELS or left == right or not entry.get("why"):
            invalid += 1
            continue
        left, right = min(left, right), max(left, right)
        if left not in alive[level] or right not in alive[level]:
            skipped_missing += 1
            named.append(f"{level} {left}/{right}: a side is gone")
            continue
        if (level, left, right) in existing:
            row = rows.get((level, left, right))
            if row is not None and row.decision is None and row.closed_reason is None and not (row.detail or {}).get("proposed_by"):
                endorsed += 1
                if not check:
                    row.detail = {**(row.detail or {}), "proposed_by": proposed_by, "proposed_why": str(entry["why"])[:400]}
            else:
                skipped_present += 1
            continue
        existing.add((level, left, right))
        inserted += 1
        if check:
            continue
        db.add(Suggestion(
            level=level, left_id=left, right_id=right,
            reason=PROPOSED, score=max(0.0, min(1.0, score)),
            detail={"why": str(entry["why"])[:400], "proposed_by": proposed_by},
        ))
    if not check:
        db.flush()
    return {"inserted": inserted, "endorsed": endorsed, "skipped_present": skipped_present,
            "skipped_missing": skipped_missing, "invalid": invalid, "check": check, "notes": named[:50]}


def load_proposals(path: str | pathlib.Path) -> tuple[list[dict[str, Any]], str]:
    """A proposals file: `{"proposed_by": "...", "entries": [...]}` or a bare list."""
    data = json.loads(pathlib.Path(path).read_text())
    if isinstance(data, dict):
        return list(data.get("entries") or []), str(data.get("proposed_by") or pathlib.Path(path).name)
    return list(data), pathlib.Path(path).name