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

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

What may merge unasked: products sharing one `match_key` (identity rules v3: house, line,
variation, size, form) whose barcodes do not disagree (at most one distinct GTIN in the
group) and whose attributes do not disagree. A group the rules cannot settle becomes
`merge_candidates` rows for a person, never a merge.

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 products), re-keys the affected products 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 dataclasses import dataclass, field
from datetime import UTC, datetime

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

from app.models import Award, Brand, Listing, MergeCandidate, Product, ProductLine, ProductMerge, VariationAlias
from app.models.catalog import IDENTITY_RULES_VERSION
from app.services import keying
from app.services import lines as lines_service
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", "size_ml", "size_value", "size_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[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")
    # 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):
                out.append("attribute_differs")
                break
        if "attribute_differs" in out:
            break
    return out


def _quantity_state(product: Product) -> 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.size_ml else "none"


def duplicate_groups(db: Session, keys: set[str] | None = None) -> list[Group]:
    """Every set of live products sharing a match key, with why it may or may not merge;
    `keys` narrows it to the keys an alias just touched."""
    by_key: dict[str, list[Product]] = defaultdict(list)
    stmt = select(Product).where(Product.merged_into_id.is_(None)).order_by(Product.id)
    if keys is not None:
        if not keys:
            return []
        stmt = stmt.where(Product.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)))
    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,
    note: str | 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)}
        if note:
            detail["note"] = note
        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))
        # 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.
        moved = overrides_service.move(db, "product", str(other.id), str(survivor.id))
        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(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 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(MergeCandidate.id)).where(
                MergeCandidate.level == "product", MergeCandidate.left_id == left, MergeCandidate.right_id == right,
            )
        ) or db.scalar(
            select(func.count(MergeCandidate.id)).where(
                MergeCandidate.product_id == anchor.id, MergeCandidate.candidate_id == other.id
            )
        )
        if exists:
            continue
        reason = ",".join(group.conflicts)
        db.add(MergeCandidate(
            level="product", left_id=left, right_id=right,
            product_id=anchor.id, candidate_id=other.id, 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) -> 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, keys):
        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


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

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


def rekey_products(db: Session, products: list[Product], 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 products:
        # 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) -> dict[str, int]:
    counts = merge_duplicates(db, merged_by=merged_by, keys=touched)
    db.flush()
    return counts


def apply_brand_alias(
    db: Session, alias: Brand, house: Brand, *, decided_by: int | None, preferred_name: str | None = None,
) -> dict[str, int]:
    """Point one brand row at another as its house, and let the catalogue follow: every
    line of the alias points at the house's line with the same key (created when the
    house had none) and its products move there; every product of the alias is re-keyed
    under the house; whatever then agrees on all four fields folds. The house may take a
    preferred name. Safe to repeat: a second call moves nothing."""
    brands = {b.id: b for b in db.scalars(select(Brand))}
    house = lines_service.resolve_alias(brands, house.id) or house
    if alias.id == house.id:
        raise ValueError("a brand cannot be an alias of itself")
    now = _now()
    alias.canonical_id = house.id
    alias.decided_by, alias.decided_at = decided_by, now
    if preferred_name and preferred_name.strip():
        house.name = preferred_name.strip()[:160]
        house.decided_by, house.decided_at = decided_by, now
    db.flush()

    house_lines = {row.key: row for row in db.scalars(select(ProductLine).where(ProductLine.brand_id == house.id))}
    lines_pointed = lines_created = 0
    for row in db.scalars(select(ProductLine).where(ProductLine.brand_id == alias.id)):
        target = house_lines.get(row.key)
        if target is None:
            target = ProductLine(brand_id=house.id, key=row.key, name=row.name,
                                 slug=_unique_slug(db, lines_service.line_slug(house.slug, row.key)[:230]))
            db.add(target)
            db.flush()
            house_lines[row.key] = target
            lines_created += 1
        if row.canonical_id != target.id:
            row.canonical_id = target.id
            row.decided_by, row.decided_at = decided_by, now
            lines_pointed += 1
        db.execute(update(Product).where(Product.line_id == row.id).values(line_id=target.id))
    db.flush()

    keying.invalidate()
    maps = keying.load_maps(db)
    products = list(db.scalars(select(Product).where(Product.brand_id == alias.id, Product.merged_into_id.is_(None))))
    touched = rekey_products(db, products, maps)
    folded = _fold(db, touched, decided_by)
    logger.info("brand_alias applied alias=%s house=%s by=%s", alias.slug, house.slug, decided_by)
    return {"lines_pointed": lines_pointed, "lines_created": lines_created, "products_rekeyed": len(products),
            "groups_merged": folded["groups"], "rows_merged": folded["merged_rows"]}


def apply_line_alias(
    db: Session, alias: ProductLine, canonical: ProductLine, *, decided_by: int | None,
    preferred_name: str | None = None,
) -> dict[str, int]:
    """Point one line at another under the same house, move its products, re-key them
    (the canonical's key names the line part) and fold what then agrees. Two houses'
    lines never fold here: the brand alias comes first."""
    rows = {row.id: row for row in db.scalars(select(ProductLine))}
    canonical = lines_service.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 houses; confirm the brand alias first")
    now = _now()
    alias.canonical_id = canonical.id
    alias.decided_by, alias.decided_at = decided_by, now
    if preferred_name and preferred_name.strip():
        canonical.name = preferred_name.strip()[:200]
        canonical.decided_by, canonical.decided_at = decided_by, now
    products = list(db.scalars(select(Product).where(Product.line_id == alias.id, Product.merged_into_id.is_(None))))
    for product in products:
        product.line_id = canonical.id
    db.flush()
    keying.invalidate()
    maps = keying.load_maps(db)
    touched = rekey_products(db, products, maps)
    folded = _fold(db, touched, decided_by)
    logger.info("line_alias applied alias=%s canonical=%s by=%s", alias.slug, canonical.slug, decided_by)
    return {"products_moved": len(products), "groups_merged": folded["groups"], "rows_merged": folded["merged_rows"]}


def apply_variation_alias(
    db: Session, vertical: str, raw: str, canonical: str, *, decided_by: int | None, display: str | None = None,
) -> dict[str, int]:
    """Map one wording of a variation to a canonical one, with the display wording, then
    re-key every product of that vertical whose name carries the wording and fold what
    then agrees."""
    raw, canonical = raw.strip().lower(), canonical.strip().lower()
    if not raw or not canonical:
        raise ValueError("a variation alias needs a wording and what it means")
    row = db.scalar(select(VariationAlias).where(VariationAlias.vertical == vertical, VariationAlias.raw == raw))
    if row is None:
        row = VariationAlias(vertical=vertical, raw=raw, canonical=canonical)
        db.add(row)
    row.canonical = canonical
    row.display = (display or lines_service.display_variation(canonical)).strip()[:80]
    row.decided_by, row.decided_at = decided_by, _now()
    db.flush()
    keying.invalidate()
    maps = keying.load_maps(db)
    products = [
        product for product in db.scalars(
            select(Product).where(Product.vertical == vertical, Product.merged_into_id.is_(None))
        )
        if lines_service.variation_of(product.name, vertical, category=product.category)[0] == raw
    ]
    touched = rekey_products(db, products, maps)
    folded = _fold(db, touched, decided_by)
    logger.info("variation_alias applied vertical=%s raw=%r canonical=%r by=%s", vertical, raw, canonical, decided_by)
    return {"products_rekeyed": len(products), "groups_merged": folded["groups"], "rows_merged": folded["merged_rows"]}