"""Recorded merges: two product rows that are one bottle become one, with a record.

Sources of truth: this module, `models/catalog.py` (Product.merged_into_id, ProductMerge,
MergeCandidate) and `cli.backfill_merges`. 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; `product_merges` says what
happened and `resolve_product_id()` makes the old id answer under the new one.

What may merge: products sharing one `match_key` (brand fold + name + size) whose
barcodes do not disagree (at most one distinct GTIN in the group) and whose per-vertical
attributes do not disagree (an EDP and an EDT stay apart even under one key). A group the
rules cannot settle becomes `merge_candidates` rows for a human, never a merge.
"""

from __future__ import annotations

import logging
from collections import defaultdict
from dataclasses import dataclass, field

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

from app.models import Award, Listing, MergeCandidate, Product, ProductMerge
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", "size_ml", "size_value", "size_unit", "abv",
                  "country_of_origin", "image_url", "thumb_url", "image_source")


@dataclass
class Group:
    match_key: str
    members: list[Product]
    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[Product]) -> list[str]:
    """Why a group must not merge: two barcodes, or two declared attributes that differ."""
    out = []
    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")
    for i, a in enumerate(members):
        for b in members[i + 1:]:
            if attributes_disagree(a.attributes, b.attributes):
                out.append("attribute_differs")
                break
        if "attribute_differs" in out:
            break
    return out


def duplicate_groups(db: Session) -> list[Group]:
    """Every set of live products sharing a match key, with why it may or may not merge."""
    by_key: dict[str, list[Product]] = defaultdict(list)
    for product in db.scalars(
        select(Product).where(Product.merged_into_id.is_(None)).order_by(Product.id)
    ):
        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)))
    return groups


def pick_survivor(members: list[Product], listing_counts: dict[int, int]) -> Product:
    """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: Product) -> dict:
    return {
        "id": product.id, "gtin": product.gtin, "brand": product.brand, "name": product.name,
        "size_ml": product.size_ml, "category": product.category, "match_key": product.match_key,
        "attributes": dict(product.attributes or {}),
    }


def merge_products(
    db: Session, survivor: Product, others: list[Product], *, reason: str, merged_by: int | None = None,
) -> int:
    """Fold `others` into `survivor`, recording each move. Returns the rows merged."""
    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)}
        db.execute(
            Listing.__table__.update().where(Listing.product_id == other.id).values(product_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.product_id == survivor.id)
            ).all()
        )
        for award in db.scalars(select(Award).where(Award.product_id == other.id)):
            if (award.competition, award.year) in held:
                db.delete(award)
            else:
                award.product_id = survivor.id
                held.add((award.competition, award.year))
        if survivor.gtin is None and other.gtin:
            gtin, other.gtin = other.gtin, None
            db.flush()
            survivor.gtin = gtin
        for name in _ENRICH_FIELDS:
            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(ProductMerge(from_id=other.id, to_id=survivor.id, merged_by=merged_by, reason=reason, detail=detail))
        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 human; one row per pair, once."""
    added = 0
    anchor = group.members[0]
    for other in group.members[1:]:
        exists = db.scalar(
            select(func.count(MergeCandidate.id)).where(
                MergeCandidate.product_id == anchor.id, MergeCandidate.candidate_id == other.id
            )
        )
        if exists:
            continue
        db.add(MergeCandidate(
            product_id=anchor.id, candidate_id=other.id, reason=",".join(group.conflicts),
            detail={"match_key": group.match_key, "from": _snapshot(anchor), "to": _snapshot(other)},
        ))
        added += 1
    return added


def merge_duplicates(db: Session, *, merged_by: int | None = 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.product_id, func.count(Listing.id)).group_by(Listing.product_id)).all()
    )
    for group in duplicate_groups(db):
        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_products(db, survivor, others, reason=group.reason, merged_by=merged_by)
    return counts