"""Recorded merges: two product rows that are one bottle become one, with a record; and
the aliases at every level (brand, line, attribute), applied so the catalogue follows.

Sources of truth: this module, `models/catalog.py` (ProductVariant.merged_into_id, Merge,
Suggestion, Brand.alias_of_id, ProductLine.alias_of_id, AttributeAlias),
`cli.backfill_merges`, `tests/test_merges.py`, `tests/test_aliases.py`. Decision 6: merges
are recorded and forward, never delete. A merged row stays with `merged_into_id` set; its
listings and awards move to the survivor; the survivor keeps or gains the barcode;
`merges` says what happened and `resolve_variant_id()` makes the old id answer
under the new one.

What may merge unasked: product variants sharing one `match_key` (the certain key of identity
rules v6: brand, residual name, identity attributes, quantity) whose barcodes do not disagree
(at most one distinct GTIN in the group), whose stated identity attributes do not disagree in
any category (`identity_differs`: ABV included), that no person kept separate and none of which
holds a person's identity decision. A group the rules cannot settle becomes `suggestions` rows
for a person, never a merge. The vetoes bind the machine only.

`prune_lines` is here too, and only here: the rule for which derived line rows a prune may
delete is shared by the whole-catalogue backfill and by the brand split (`services/brands.py`),
which empties rows as it moves variants off them.

An alias is a person's decision (rian, 12 Sep: never applied silently by a rule). Applying
one points the alias row at its canonical with who decided and when, moves what hung off
the alias (a brand's lines, a line's product variants), re-keys the affected product variants through the
same call ingest uses, and folds whatever then agrees on all four fields. The first alias
waiting is Paco Rabanne to Rabanne.
"""

from __future__ import annotations

import logging
from collections import defaultdict
from collections.abc import Iterable
from dataclasses import dataclass, field
from datetime import UTC, datetime

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

from app.models import Award, Brand, Listing, Suggestion, ProductVariant, ProductLine, Merge, AttributeAlias
from app.models.catalog import IDENTITY_RULES_VERSION
from app.services import keying
from app.services import product_lines
from app.services import overrides as overrides_service
from app.services.ingest import attributes_disagree
from app.services.normalize import looks_like_set

logger = logging.getLogger(__name__)

# Fields the survivor takes from a merged row when it has none of its own.
_ENRICH_FIELDS = ("brand", "brand_id", "category", "quantity_ml", "quantity_stated_value", "quantity_stated_unit", "abv",
                  "quantity_value", "quantity_unit", "pack_count", "pack_unit_value", "form", "set_contents",
                  "quantity_state", "name_key", "country_of_origin", "image_url", "thumb_url", "image_source")


@dataclass
class Group:
    match_key: str
    members: list[ProductVariant]
    reason: str = "duplicate_match_key"
    conflicts: list[str] = field(default_factory=list)

    @property
    def mergeable(self) -> bool:
        return not self.conflicts and len(self.members) > 1


def _conflicts(members: list[ProductVariant], maps=None) -> list[str]:
    """Why a group must not merge: two barcodes, two declared attributes that differ, a pair a
    person kept separate (`kept_separate`, spec §3.1) or a member holding an effective decision
    on an identity field (`decided_member`): a machine never undoes a person, and never silently
    loses a decided loser's rows."""
    out = []
    if maps is not None:
        ids = [p.id for p in members]
        if maps.separated and any(frozenset((a, b)) in maps.separated for i, a in enumerate(ids) for b in ids[i + 1:]):
            out.append("kept_separate")
        if any(any(f in ("name", "product_line") or f.startswith("attribute:") for f in maps.decided.get(p.id, {})) for p in members):
            out.append("decided_member")
    if len({p.gtin for p in members if p.gtin}) > 1:
        out.append("gtin_differs")
    if len({looks_like_set(p.name) for p in members}) > 1:
        out.append("set_vs_single")
    # Unknown never agrees with anything (rian, 15 Sep): a group where any member states no
    # quantity is a candidate for a person, never an auto-fold; two things joined because
    # neither said how much was the fold this retires.
    if any(_quantity_state(p) != "stated" for p in members):
        out.append("quantity_unknown")
    for i, a in enumerate(members):
        for b in members[i + 1:]:
            if attributes_disagree(a.attributes, b.attributes) or identity_differs(a, b):
                out.append("attribute_differs")
                break
        if "attribute_differs" in out:
            break
    return out


def identity_differs(a: ProductVariant, b: ProductVariant) -> str | None:
    """The first identity attribute two variants BOTH state and state differently, read through
    the registry's accessor whatever column or JSON key holds it (plan W9): an ABV of 40 against
    43, an age of 12 against 15, a shop's option. Numbers compare as numbers ("40" is 40.0); text
    folds case and spacing; a stated value against none is not a disagreement. This veto binds
    the machine only: a person may still Confirm same across it (`merge_session.confirm`) and
    then set the value as a decision."""
    from app.services import attributes as attributes_service

    def stated(variant) -> dict:
        return {x.kind: x.value for x in attributes_service.of(variant) if x.identity == "always" and x.kind != "quantity"}

    ours, theirs = stated(a), stated(b)
    for kind in sorted(ours.keys() & theirs.keys()):
        left, right = ours[kind], theirs[kind]
        try:
            same = abs(float(left) - float(right)) < 1e-9
        except (TypeError, ValueError):
            same = " ".join(str(left).lower().split()) == " ".join(str(right).lower().split())
        if not same:
            return kind
    return None


def _quantity_state(product: ProductVariant) -> str:
    """A row `backfill quantities` has not reached yet (NULL state) is read as the v3 world
    read it: stated when it holds a millilitre figure, unknown otherwise."""
    if product.quantity_state:
        return product.quantity_state
    return "stated" if product.quantity_ml else "none"


def duplicate_groups(db: Session, keys: set[str] | None = None, maps=None) -> list[Group]:
    """Every set of live product variants sharing a match key, with why it may or may not merge;
    `keys` narrows it to the keys an alias just touched. The maps are loaded fresh when not
    given, so the 300 s cache can never fold a pair kept separate seconds ago."""
    if maps is None:
        maps = keying.load_maps(db)
    by_key: dict[str, list[ProductVariant]] = defaultdict(list)
    stmt = select(ProductVariant).where(ProductVariant.merged_into_id.is_(None)).order_by(ProductVariant.id)
    if keys is not None:
        if not keys:
            return []
        stmt = stmt.where(ProductVariant.match_key.in_(list(keys)))
    for product in db.scalars(stmt):
        by_key[product.match_key].append(product)
    groups = []
    for key, members in by_key.items():
        if len(members) < 2:
            continue
        groups.append(Group(match_key=key, members=members, conflicts=_conflicts(members, maps)))
    return groups


def pick_survivor(members: list[ProductVariant], listing_counts: dict[int, int]) -> ProductVariant:
    """The row the others fold into: the one with the barcode, else the most listed, else the oldest."""
    return sorted(
        members,
        key=lambda p: (p.gtin is None, -listing_counts.get(p.id, 0), p.id),
    )[0]


def _snapshot(product: ProductVariant) -> dict:
    return {
        "id": product.id, "gtin": product.gtin, "brand": product.brand, "name": product.name,
        "size_ml": product.quantity_ml, "category": product.category, "match_key": product.match_key,
        "attributes": dict(product.attributes or {}),
    }


def merge_product_variants(
    db: Session, survivor: ProductVariant, others: list[ProductVariant], *, reason: str, merged_by: int | None = None,
    note: str | None = None, decision=None, batch=None,
) -> int:
    """Fold `others` into `survivor`, recording each move (the record beside its decision and the
    batch whose tail made it, plan W15). Returns the rows merged. The carry of a loser's decisions
    is the `merged_into` row's consequence (`appliers`); a rules fold carries through the shim."""
    merged = 0
    for other in others:
        if other.id == survivor.id or other.merged_into_id is not None:
            continue
        detail = {"from": _snapshot(other), "to": _snapshot(survivor)}
        if note:
            detail["note"] = note
        # What moves is recorded before it moves (the catalogue decisions §6, undo): the listing
        # ids that change product, and below the duplicate awards dropped, so an `unmerge` can
        # reverse one merge as one unit from its own record.
        detail["listings"] = list(db.scalars(select(Listing.id).where(Listing.variant_id == other.id).order_by(Listing.id)))
        db.execute(
            Listing.__table__.update().where(Listing.variant_id == other.id).values(variant_id=survivor.id)
        )
        # An award the survivor already holds for that competition and year is
        # the same medal seen twice; the unique constraint says so, so skip it.
        held = set(
            db.execute(
                select(Award.competition, Award.year).where(Award.variant_id == survivor.id)
            ).all()
        )
        for award in db.scalars(select(Award).where(Award.variant_id == other.id)):
            if (award.competition, award.year) in held:
                detail.setdefault("awards_dropped", []).append(
                    {"competition": award.competition, "year": award.year, "medal": award.medal})
                db.delete(award)
            else:
                detail.setdefault("awards_moved", []).append(award.id)
                award.variant_id = survivor.id
                held.add((award.competition, award.year))
        # The barcode (M7): the survivor keeps its own, or gains the other's and is marked
        # as having done so, because its own shops never published it and markup must be
        # able to tell. Two different barcodes reach here only through a person's confirm;
        # the record says which won and why, and nothing is ever invented.
        if survivor.gtin is None and other.gtin:
            gtin, other.gtin = other.gtin, None
            db.flush()
            survivor.gtin = gtin
            survivor.gtin_source = "merge"
            detail["barcode"] = {"gained": gtin, "from": other.id}
        elif survivor.gtin and other.gtin and survivor.gtin != other.gtin:
            detail["barcode"] = {"kept": survivor.gtin, "dropped": other.gtin, "why": note}
        # The image: the survivor inherits one it lacks, with its provenance (image_source
        # says whether it was found by barcode or by name), and keeps its own otherwise.
        if survivor.image_url is None and other.image_url:
            detail["image"] = {"from": other.id, "source": other.image_source}
        # The decided layer (LT3 f): the loser's overrides move to the survivor where it has
        # none for that field; the rest are recorded; a human value is never overwritten.
        if decision is None and overrides_service.table_present(db):
            moved = overrides_service.move(db, "product", str(other.id), str(survivor.id), batch=batch)
            if moved["moved"] or moved["kept"]:
                detail["overrides"] = moved
        guarded = overrides_service.guarded_columns(overrides_service.fields_of(db, "product", str(survivor.id)))
        for name in _ENRICH_FIELDS:
            if name in guarded:
                continue
            if getattr(survivor, name) is None and getattr(other, name) is not None:
                setattr(survivor, name, getattr(other, name))
        missing = {k: v for k, v in (other.attributes or {}).items() if k not in (survivor.attributes or {})}
        if missing:
            survivor.attributes = {**(survivor.attributes or {}), **missing}
        if other.is_exclusive:
            survivor.is_exclusive = True
        other.merged_into_id = survivor.id
        db.add(Merge(from_id=other.id, to_id=survivor.id, merged_by=merged_by, reason=reason, detail=detail,
                     decision_id=getattr(decision, "id", None), batch_id=batch.row.id if batch is not None else None))
        db.flush()
        logger.info("product_merged from=%s to=%s reason=%s", other.id, survivor.id, reason)
        merged += 1
    return merged


def record_candidates(db: Session, group: Group) -> int:
    """A group the rules cannot settle, queued for a person; one row per pair, once, at the
    product level with the reason a person reads (`suggest.product_reason`)."""
    from app.services.suggest import product_reason

    added = 0
    anchor = group.members[0]
    for other in group.members[1:]:
        left, right = min(anchor.id, other.id), max(anchor.id, other.id)
        exists = db.scalar(
            select(func.count(Suggestion.id)).where(
                Suggestion.level == "product", Suggestion.left_id == left, Suggestion.right_id == right,
            )
        )
        if exists:
            continue
        reason = ",".join(group.conflicts)
        db.add(Suggestion(
            level="product", left_id=left, right_id=right,
            reason=reason, score=0.5,
            detail={"match_key": group.match_key, "from": _snapshot(anchor), "to": _snapshot(other),
                    "why": product_reason(reason)},
        ))
        added += 1
    return added


def merge_duplicates(db: Session, *, merged_by: int | None = None, keys: set[str] | None = None, batch=None, maps=None) -> dict[str, int]:
    """Every mergeable group becomes recorded merges; every other group, candidates."""
    counts = {"groups": 0, "merged_rows": 0, "candidate_groups": 0, "candidates_added": 0}
    listing_counts = dict(
        db.execute(select(Listing.variant_id, func.count(Listing.id)).group_by(Listing.variant_id)).all()
    )
    for group in duplicate_groups(db, keys, maps):
        if not group.mergeable:
            counts["candidate_groups"] += 1
            counts["candidates_added"] += record_candidates(db, group)
            continue
        survivor = pick_survivor(group.members, listing_counts)
        others = [p for p in group.members if p.id != survivor.id]
        counts["groups"] += 1
        counts["merged_rows"] += merge_product_variants(db, survivor, others, reason=group.reason, merged_by=merged_by, batch=batch)
    return counts


# --------------------------------------------------------------------------- aliases

def _now() -> datetime:
    return datetime.now(UTC)


def rekey_product_variants(db: Session, product_variants: list[ProductVariant], maps: keying.Maps) -> set[str]:
    """Every product's key and attributes under the current rules; returns the keys
    touched (old and new), so the fold can be limited to them."""
    touched: set[str] = set()
    for product in product_variants:
        # Tombstones are rekeyed too (LT3 d): a human merge's tombstone must carry the key
        # of the current rules for `ingest._follow_tombstone` to find it.
        key, attributes = keying.product_key(product, maps)
        touched.add(key)
        if product.match_key != key:
            touched.add(product.match_key)
            product.match_key = key
        if attributes != (product.attributes or {}):
            product.attributes = attributes
        if product.identity_rules_version != IDENTITY_RULES_VERSION:
            product.identity_rules_version = IDENTITY_RULES_VERSION
    db.flush()
    return touched


def _unique_slug(db: Session, wanted: str) -> str:
    taken = set(db.scalars(select(ProductLine.slug).where(ProductLine.slug.like(f"{wanted}%"))))
    slug, n = wanted, 2
    while slug in taken:
        slug, n = f"{wanted}-{n}", n + 1
    return slug


def _fold(db: Session, touched: set[str], merged_by: int | None, batch=None) -> dict[str, int]:
    """The fold at a batch's tail: every merge it makes carries the batch id, so batch undo reverses it."""
    counts = merge_duplicates(db, merged_by=merged_by, keys=touched, batch=batch)
    db.flush()
    return counts


def apply_brand_alias(
    db: Session, alias: Brand, brand: Brand, *, decided_by: int | None, preferred_name: str | None = None, batch=None,
) -> dict[str, int]:
    """Point one brand row at another as its alias and let the catalogue follow (the cascade is
    the `alias_of` row's consequences in `decisions/appliers.py`; the rekey and the fold run at the
    batch's tail). Opens a route batch when the caller has none. Safe to repeat."""
    from app.services.decisions import writer

    brands = {b.id: b for b in db.scalars(select(Brand))}
    brand = product_lines.resolve_alias(brands, brand.id) or brand
    if alias.id == brand.id:
        raise ValueError("a brand cannot be an alias of itself")
    if alias.alias_of_id == brand.id:
        return {"lines_pointed": 0, "lines_created": 0, "product_variants_rekeyed": 0, "line_decisions_moved": 0,
                "line_decisions_kept": 0, "groups_merged": 0, "rows_merged": 0}

    def run(b):
        detail = {"preferred_name": preferred_name.strip()[:160]} if preferred_name and preferred_name.strip() else None
        return writer.record(b, "brand", alias, "alias_of", brand, reason=f"alias of {brand.name}", detail=detail)

    if batch is not None:
        row = run(batch)
        d = batch.notes.get(row.id, {})
        return {"lines_pointed": d.get("lines_pointed", 0), "lines_created": d.get("lines_created", 0),
                "product_variants_rekeyed": len(batch.touched), "line_decisions_moved": d.get("line_decisions_moved", 0),
                "line_decisions_kept": d.get("line_decisions_kept", 0), "groups_merged": 0, "rows_merged": 0}
    with writer.batch(db, "route", "individual", decided_by, commit=False) as b:
        row = run(b)
    logger.info("brand_alias applied alias=%s brand=%s by=%s", alias.slug, brand.slug, decided_by)
    d = b.notes.get(row.id, {})
    return {"lines_pointed": d.get("lines_pointed", 0), "lines_created": d.get("lines_created", 0),
            "product_variants_rekeyed": b.row.summary.get("variants_rekeyed", 0),
            "line_decisions_moved": d.get("line_decisions_moved", 0), "line_decisions_kept": d.get("line_decisions_kept", 0),
            "groups_merged": b.folded.get("groups", 0), "rows_merged": b.folded.get("merged_rows", 0)}


def apply_line_alias(
    db: Session, alias: ProductLine, canonical: ProductLine, *, decided_by: int | None,
    preferred_name: str | None = None, batch=None,
) -> dict[str, int]:
    """Point one line at another under the same brand and move its variants (the `alias_of`
    row's applier records every move for undo); the rekey and fold run at the batch's tail."""
    from app.services.decisions import writer

    rows = {row.id: row for row in db.scalars(select(ProductLine))}
    canonical = product_lines.resolve_alias(rows, canonical.id) or canonical
    if alias.id == canonical.id:
        raise ValueError("a line cannot be an alias of itself")
    if alias.brand_id != canonical.brand_id:
        raise ValueError("the two lines belong to different brands; confirm the brand alias first")
    if alias.alias_of_id == canonical.id:
        return {"product_variants_moved": 0, "groups_merged": 0, "rows_merged": 0}

    def run(b):
        detail = {"preferred_name": preferred_name.strip()[:200]} if preferred_name and preferred_name.strip() else None
        return writer.record(b, "product_line", alias, "alias_of", canonical, reason=f"alias of {canonical.name}", detail=detail)

    if batch is not None:
        row = run(batch)
        return {"product_variants_moved": len((row.detail or {}).get("products_moved", [])), "groups_merged": 0, "rows_merged": 0}
    with writer.batch(db, "route", "individual", decided_by, commit=False) as b:
        row = run(b)
    logger.info("line_alias applied alias=%s canonical=%s by=%s", alias.slug, canonical.slug, decided_by)
    return {"product_variants_moved": len((row.detail or {}).get("products_moved", [])),
            "groups_merged": b.folded.get("groups", 0), "rows_merged": b.folded.get("merged_rows", 0)}


def apply_variation_alias(
    db: Session, vertical: str, raw: str, canonical: str, *, decided_by: int | None, display: str | None = None, batch=None,
) -> dict[str, int]:
    """Map one wording of an attribute to what it means, with the display wording: two rows on
    the wording (minted when absent); every variant of that vertical carrying the wording is
    re-keyed at the batch's tail and what then agrees folds."""
    from app.services.decisions import writer

    raw, canonical = raw.strip().lower(), canonical.strip().lower()
    if not raw or not canonical:
        raise ValueError("an attribute alias needs a wording and what it means")
    key = f"wording:{vertical}|{raw}"
    shown = (display or product_lines.display_attribute(canonical)).strip()[:80]

    def run(b):
        writer.record(b, "attribute_wording", key, "meaning", canonical, reason="wording mapped at confirm")
        writer.record(b, "attribute_wording", key, "display", shown, reason="wording mapped at confirm")

    if batch is not None:
        run(batch)
        return {"product_variants_rekeyed": len(batch.touched), "groups_merged": 0, "rows_merged": 0}
    with writer.batch(db, "route", "individual", decided_by, commit=False) as b:
        run(b)
    logger.info("variation_alias applied vertical=%s raw=%r canonical=%r by=%s", vertical, raw, canonical, decided_by)
    return {"product_variants_rekeyed": b.row.summary.get("variants_rekeyed", 0),
            "groups_merged": b.folded.get("groups", 0), "rows_merged": b.folded.get("merged_rows", 0)}


def referenced_lines(db: Session, *, ignore_batches: Iterable[int] | None = None) -> set[int]:
    """Every product line id something points at, so a prune cannot delete a record.

    `ignore_batches`: decision batches whose references do not count. An undo reverses a batch,
    and the fold it reverses minted lines that only that batch's rows (and the undo's own) name;
    without this the minted lines could never be pruned, and every fold-and-undo grew the
    catalogue by its unmatched lines.

    A line row is a pure function of the collected name, so a rule change leaves the rows the
    old rule made with nothing on them; such a row is a cache. Anything that REFERS to one makes
    it a record instead: a variant (live or tombstone), a line aliased into it, a line suggestion
    still open or that a person decided, any decision naming it, and a redirect pointing at it.
    One home for that rule: `cli.backfill_prune_lines` runs it over the whole catalogue and the
    brand split (`services/brands.py`) over the rows its move emptied.
    """
    from sqlalchemy import inspect as sa_inspect

    from app.models import Decision, Redirect

    referenced: set[int] = set()
    referenced.update(i for i in db.scalars(select(ProductVariant.product_line_id)
                                            .where(ProductVariant.product_line_id.isnot(None)).distinct()) if i)
    referenced.update(i for i in db.scalars(select(ProductLine.alias_of_id)
                                            .where(ProductLine.alias_of_id.isnot(None)).distinct()) if i)
    for left, right in db.execute(select(Suggestion.left_id, Suggestion.right_id).where(
            Suggestion.level == "line", Suggestion.is_open() | (Suggestion.decided_by.isnot(None)))):
        referenced.update(i for i in (left, right) if i)
    skip = set(int(i) for i in (ignore_batches or ()))
    if "decisions" in sa_inspect(db.connection()).get_table_names():
        for row in db.scalars(select(Decision).where(Decision.entity_type == "product_variant", Decision.field == "product_line")):
            if row.batch_id in skip:
                continue
            for value in (row.value_ref_id, row.rule_value):
                try:
                    referenced.add(int(value))
                except (TypeError, ValueError):
                    continue
        referenced.update(i for i in db.scalars(select(Decision.entity_id)
                                                .where(Decision.entity_type == "product_line", Decision.batch_id.notin_(skip) if skip else True).distinct()) if i)
        referenced.update(i for i in db.scalars(select(Decision.value_ref_id)
                                                .where(Decision.entity_type == "product_line", Decision.field == "alias_of",
                                                       Decision.batch_id.notin_(skip) if skip else True).distinct()) if i)
        # A redirect written by a decision in an ignored batch is that decision's consequence and
        # does not keep its target alive (the 18 Sep rehearsal: one minted line survived every
        # undo on the redirect its line alias had written at confirm).
        redirects = select(Redirect.to_slug)
        if skip:
            redirects = redirects.outerjoin(Decision, Decision.id == Redirect.decision_id).where(
                Redirect.kind == "product_line", or_(Redirect.decision_id.is_(None), Decision.batch_id.notin_(skip)))
        else:
            redirects = redirects.where(Redirect.kind == "product_line")
        for slug in db.scalars(redirects):
            row_id = db.scalar(select(ProductLine.id).where(ProductLine.slug == slug))
            if row_id:
                referenced.add(row_id)
    return referenced


def prune_lines(db: Session, candidates: Iterable[int] | None = None, *, ignore_batches: Iterable[int] | None = None) -> int:
    """Delete the derived line rows nothing references (the catalogue decisions §2.3), optionally
    narrowed to `candidates`. Never a row a person decided (`decided_by`), never an alias row,
    never a merge tombstone (that is a variant, not a line). Flushes; never commits, so a caller
    inside a batch keeps its one transaction. Safe to repeat: a second run deletes zero."""
    referenced = referenced_lines(db, ignore_batches=ignore_batches)
    stmt = select(ProductLine).where(ProductLine.alias_of_id.is_(None), ProductLine.decided_by.is_(None))
    if candidates is not None:
        ids = [int(i) for i in candidates]
        if not ids:
            return 0
        stmt = stmt.where(ProductLine.id.in_(ids))
    pruned = 0
    for row in db.scalars(stmt):
        if row.id in referenced:
            continue
        db.delete(row)
        pruned += 1
    db.flush()
    return pruned