"""Collection registry identity — one active row per path/name, never resurrect duplicates.

Short brochure codes (ANRWO) and canonical codes (ANRO) used to create two rows
for the same collection. Deactivating the duplicate is not enough: code-keyed
upserts would flip ``is_active`` back on. This module prefers the live twin and
treats inactive rows as tombstones that occupy their codes.
"""

from __future__ import annotations

from typing import Any, Dict, Iterable, List, Optional

from sqlalchemy import case, or_, select, update
from sqlalchemy.orm import Session

from channel.url_canonical import normalize_plp_path
from db.collection_landing_pages import canonical_collection, collection_path_slug, DEFAULT_CATEGORY_L1
from db.models import MasterCollectionRegistry, MasterProduct


def find_active_collection_registry(
    session: Session,
    *,
    registry_id: Optional[int] = None,
    code: Optional[str] = None,
    name: Optional[str] = None,
    path_slug: Optional[str] = None,
    category_l1: Optional[str] = None,
) -> Optional[MasterCollectionRegistry]:
    """Return the active canonical row, never an inactive duplicate."""
    hinted_slug = normalize_plp_path(path_slug) if path_slug else None
    hinted_name = canonical_collection(name)
    hinted_code = str(code or "").strip()

    if registry_id is not None:
        row = session.get(MasterCollectionRegistry, int(registry_id))
        if row is not None and row.is_active:
            return row
        if row is not None:
            hinted_slug = hinted_slug or normalize_plp_path(row.path_slug or "")
            hinted_name = hinted_name or canonical_collection(row.name)
            hinted_code = hinted_code or str(row.code or "").strip()

    if not hinted_slug and hinted_name:
        hinted_slug = collection_path_slug(category_l1 or DEFAULT_CATEGORY_L1, hinted_name)

    actives = list(
        session.scalars(
            select(MasterCollectionRegistry).where(MasterCollectionRegistry.is_active.is_(True))
        ).all()
    )
    if hinted_slug:
        for row in actives:
            if normalize_plp_path(row.path_slug or "") == hinted_slug:
                return row
    if hinted_name:
        wanted = _norm(hinted_name)
        for row in actives:
            names = {_norm(row.name), _norm(row.code)}
            names.update(_norm(alias) for alias in _identity_alias_names(row.aliases))
            if wanted in names:
                return row
    if hinted_code:
        wanted = hinted_code.upper()
        for row in actives:
            codes = {str(row.code or "").strip().upper()}
            codes.update(str(item).strip().upper() for item in _alias_codes(row.aliases))
            if wanted in codes:
                return row
    return None


def merge_collection_identity_aliases(
    row: MasterCollectionRegistry,
    *,
    extra_codes: Optional[Iterable[str]] = None,
    extra_names: Optional[Iterable[str]] = None,
) -> None:
    aliases = dict(row.aliases) if isinstance(row.aliases, dict) else {}
    codes = [str(item).strip() for item in (aliases.get("code_aliases") or []) if str(item).strip()]
    names = [str(item).strip() for item in (aliases.get("items") or []) if str(item).strip()]
    for code in extra_codes or []:
        text = str(code or "").strip()
        if text and text.upper() != str(row.code or "").strip().upper() and text not in codes:
            codes.append(text)
    for name in extra_names or []:
        text = canonical_collection(name) or str(name or "").strip()
        if text and _norm(text) != _norm(row.name) and text not in names:
            names.append(text)
    if codes:
        aliases["code_aliases"] = codes
    if names:
        aliases["items"] = names
    row.aliases = aliases or None


def seal_inactive_collection_duplicates(session: Session) -> Dict[str, Any]:
    """Keep inactive rows as tombstones, alias them onto the live twin, remap SKUs."""
    inactive = list(
        session.scalars(
            select(MasterCollectionRegistry).where(MasterCollectionRegistry.is_active.is_(False))
        ).all()
    )
    stats = {
        "inactive_rows": len(inactive),
        "sealed": 0,
        "unmapped_inactive": 0,
        "aliases_merged": 0,
        "skus_relinked": 0,
    }
    pairs: List[tuple[MasterCollectionRegistry, MasterCollectionRegistry]] = []
    for row in inactive:
        twin = find_active_collection_registry(
            session,
            path_slug=row.path_slug,
            name=row.name,
            code=row.code,
        )
        if twin is None or twin.id == row.id:
            stats["unmapped_inactive"] += 1
            continue
        before = dict(twin.aliases) if isinstance(twin.aliases, dict) else {}
        merge_collection_identity_aliases(
            twin,
            extra_codes=[row.code, str((row.aliases or {}).get("source_collection_code") or "")],
            extra_names=[row.name],
        )
        if twin.aliases != before:
            stats["aliases_merged"] += 1
        note = f"tombstone duplicate of master_collection_registry.id={twin.id} code={twin.code}"
        if note not in str(row.notes or ""):
            row.notes = f"{row.notes}; {note}".strip("; ") if row.notes else note
        pairs.append((row, twin))
        stats["sealed"] += 1
    stats["skus_relinked"] += _relink_sealed_products(session, pairs)
    session.flush()
    session.expire_all()
    return stats


def _relink_sealed_products(
    session: Session,
    pairs: List[tuple[MasterCollectionRegistry, MasterCollectionRegistry]],
) -> int:
    if not pairs:
        return 0
    tombstone_to_twin = {row.id: twin.id for row, twin in pairs}
    name_to_twin: Dict[str, int] = {}
    for row, twin in pairs:
        for name in (twin.name, row.name):
            text = str(name or "").strip()
            if text:
                name_to_twin[text] = twin.id
    moved = 0
    tombstone_update = session.execute(
        update(MasterProduct)
        .where(MasterProduct.collection_registry_id.in_(list(tombstone_to_twin)))
        .values(collection_registry_id=case(tombstone_to_twin, value=MasterProduct.collection_registry_id))
    )
    moved += int(tombstone_update.rowcount or 0)
    if name_to_twin:
        name_case = case(name_to_twin, value=MasterProduct.collection)
        name_update = session.execute(
            update(MasterProduct)
            .where(MasterProduct.collection.in_(list(name_to_twin)))
            .where(
                or_(
                    MasterProduct.collection_registry_id.is_(None),
                    MasterProduct.collection_registry_id != name_case,
                )
            )
            .values(collection_registry_id=name_case)
        )
        moved += int(name_update.rowcount or 0)
    return moved


def _identity_alias_names(value: Any) -> List[str]:
    """Name aliases that identify the same collection. Matching styles are related, not identity."""
    values: List[str] = []
    if isinstance(value, dict):
        items = value.get("items")
        if isinstance(items, list):
            values.extend(str(item).strip() for item in items if str(item).strip())
        line_name = str(value.get("line_name") or "").strip()
        color_label = str(value.get("color_label") or "").strip()
        if line_name and color_label:
            values.append(f"{line_name} {color_label}")
        elif line_name:
            values.append(line_name)
    return values


def _alias_codes(value: Any) -> List[str]:
    from db.master_taxonomy_registry import alias_codes_from_json

    return alias_codes_from_json(value)


def _norm(value: Optional[str]) -> str:
    return " ".join(str(value or "").strip().lower().replace("grey", "gray").split())