"""The keying call: what ingest and rederive ask before a product is keyed (Stream M).

Sources of truth: this module, `normalize.match_key` (identity rules v4), `product_lines.py`,
`quantity.py`, `overrides.py` (the decided layer, read first),
`models/catalog.py` (Brand.alias_of_id, AttributeAlias, IDENTITY_RULES_VERSION),
`tests/test_keying.py`. Identity rules v3 key a product on brand | line | attribute |
size, with a form marker for a set, a refill or a pack, so "automatic matching = same
brand, same line, same canonical attribute, same size, no barcode disagreement" (rian, 12
Sep) is the key itself: an unknown attribute keys apart and is only ever a suggestion.

Two things need the database before the pure key can be computed: the brand (the listed_brand
row with its alias followed, so a confirmed "Paco Rabanne is Rabanne" reaches every key of
both spellings) and the canonical attribute (the rule's reading, unless a person mapped
that wording elsewhere in `attribute_aliases`). Both are read into `Maps` once and reused:
a collection keys thousands of listings, a rederive every product. `maps_for()` caches per
database for a few minutes, long enough for a run and short enough that a confirm in the
merge session reaches the next one; the merge routes call `invalidate()` after writing.
"""

from __future__ import annotations

import time
from dataclasses import dataclass, field
from typing import Any

from sqlalchemy import inspect, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.models import Brand, ProductVariant, ProductLine, AttributeAlias
from app.services import product_lines
from app.services import quantity as quantity_service
from app.services.collectors.base import RawListing
from app.services.normalize import brand_key, flat_key, match_key
from app.services.taxonomy import classify, vertical_of

CACHE_SECONDS = 300


@dataclass
class Maps:
    """What the key needs from the database, read once: brand rows by id and by slug, and
    the attribute aliases a person or the backfill recorded, by (vertical, raw wording)."""

    brands: dict[int, Brand] = field(default_factory=dict)
    by_slug: dict[str, Brand] = field(default_factory=dict)
    aliases: dict[tuple[str, str], str] = field(default_factory=dict)
    #: The kind a person or the backfill recorded for a wording, by (vertical, raw).
    alias_kinds: dict[tuple[str, str], str] = field(default_factory=dict)
    lines: dict[int, ProductLine] = field(default_factory=dict)
    lines_by_key: dict[tuple[int, str], ProductLine] = field(default_factory=dict)
    #: The decided layer (Stream L, the ledger in K2): a variant's effective decisions by id then
    #: field (`name`, `product_line`, `attribute:<kind>`), every rekey reading it before the rules.
    #: The legacy field names (`product_line_id`, `quantity`, `attribute`) are aliases for one
    #: release (`_LEGACY_FIELDS`), so today's readers stand until the shim is retired.
    decided: dict[int, dict[str, Any]] = field(default_factory=dict)
    #: Every variant pair a person kept separate, survivor-chased (spec §3.1); the fold's veto.
    separated: set[frozenset[int]] = field(default_factory=set)
    #: The listed spellings a brand SPLIT claims, by `normalize.flat_key` of the spelling
    #: (K9; `services/brands.py`). Read BEFORE the fold, because the fold key of a split-off
    #: spelling is the row it was split away from: "Appleton Rum" folds to `appleton`, so
    #: without the claim the next sighting lands straight back on the row a person emptied.
    spelling_claims: dict[str, Brand] = field(default_factory=dict)
    #: Where each listed brand spelling ALREADY sits: the brand row its live variants are on,
    #: when they agree on one (K9.5). Read after a split's claim and before the fold, so a change
    #: to the trailer list can never silently re-home a spelling that is already placed -- it
    #: only decides where a spelling nobody has seen before goes. The same principle
    #: REVIEW-PROCESS.md section 1 already states for a listing and its variant.
    placed_spellings: dict[str, Brand] = field(default_factory=dict)
    #: `{id: merged_into_id}` for the chase.
    merged_into: dict[int, int] = field(default_factory=dict)
    loaded_at: float = 0.0

    def brand_of(self, brand_id: int | None = None, listed_brand: str | None = None,
                 vertical: str | None = None) -> Brand | None:
        """The brand row (aliases followed) for a variant's brand id, else for a listed spelling:
        a spelling a split claims first, then the fold key scoped to the vertical."""
        row = product_lines.resolve_alias(self.brands, brand_id) if brand_id else None
        if row is None and listed_brand:
            found = (self.spelling_claims.get(flat_key(listed_brand))
                     or self.placed_spellings.get(flat_key(listed_brand))
                     or self.by_slug.get(brand_key(listed_brand).replace(" ", "-")))
            row = product_lines.resolve_alias(self.brands, found.id) if found else None
        return row

    def line(self, brand: Brand | None, computed_key: str) -> ProductLine | None:
        """The canonical line row for a brand and a computed line key: the row itself, or
        the line it was folded into (a line alias), or None when no row exists yet."""
        if brand is None:
            return None
        row = self.lines_by_key.get((brand.id, computed_key))
        return product_lines.resolve_alias(self.lines, row.id) if row else None

    def remember_line(self, row: ProductLine) -> None:
        self.lines[row.id] = row
        self.lines_by_key[(row.brand_id, row.key)] = row

    def attribute(self, name: str | None, vertical: str | None, *, category: str | None = None) -> str:
        """The canonical attribute of a name: the alias table's word for its wording when a
        row exists (a person's decision or the seeded rule), else the rule's. The category
        reaches the rule (a Makeup shade, Stream L)."""
        raw, rule = product_lines.attribute_of(name, vertical, category=category)
        if not raw:
            return ""
        return self.aliases.get((vertical or "", raw), rule)

    def attribute_kind(self, name: str | None, vertical: str | None, *, category: str | None = None) -> str:
        """The kind of a name's attribute: the alias row's when a person or the backfill set
        one, else the rule's (`product_lines.attribute_kind_of`)."""
        raw, _ = product_lines.attribute_of(name, vertical, category=category)
        if not raw:
            return ""
        return self.alias_kinds.get((vertical or "", raw)) or product_lines.attribute_kind_of(name, vertical, category=category)


_cache: dict[int, Maps] = {}


def load_maps(db: Session) -> Maps:
    maps = Maps(loaded_at=time.monotonic())
    # Inspect through the session's own connection: an inspector on the engine borrows a
    # second connection, and on a single-connection pool (the tests' SQLite) that is the
    # same connection, whose release rolled back the caller's unflushed alias.
    tables = inspect(db.connection()).get_table_names()
    if "brands" in tables:
        for row in db.scalars(select(Brand)):
            maps.brands[row.id] = row
            maps.by_slug[row.slug] = row
    if "attribute_aliases" in tables:
        for alias in db.scalars(select(AttributeAlias)):
            maps.aliases[(alias.vertical, alias.raw)] = alias.canonical
            if getattr(alias, "kind", None):
                maps.alias_kinds[(alias.vertical, alias.raw)] = alias.kind
    if "product_lines" in tables:
        for row in db.scalars(select(ProductLine)):
            maps.remember_line(row)
    if "product_variants" in tables:
        placed: dict[str, Brand | None] = {}
        for text, brand_id in db.execute(select(ProductVariant.brand, ProductVariant.brand_id)
                                         .where(ProductVariant.merged_into_id.is_(None), ProductVariant.brand_id.isnot(None),
                                                ProductVariant.brand.isnot(None)).distinct()):
            key = flat_key(text)
            row = maps.brands.get(brand_id)
            if not key or row is None:
                continue
            placed[key] = row if placed.get(key, row) is row else None  # two rows hold it: ambiguous, so no answer
        maps.placed_spellings = {k: v for k, v in placed.items() if v is not None}
    if "decisions" in tables:
        from app.services.decisions.effective import effective, merged_into_map, separated

        for (variant_id, fld), row in effective(db, "product_variant").items():
            slot = maps.decided.setdefault(variant_id, {})
            slot[fld] = row
            for legacy in _legacy_names(fld):
                slot[legacy] = _LegacyView(row, fld)
        if "product_variants" in tables:
            maps.merged_into = merged_into_map(db)
            maps.separated = separated(db, maps.merged_into)
        for (_brand_id, fld), row in effective(db, "brand", None, ["split"]).items():
            target = maps.by_slug.get((row.value or {}).get("brand_slug") or "")
            if target is None:
                continue
            for spelling in (row.value or {}).get("spellings") or []:
                maps.spelling_claims[flat_key(spelling)] = target
    return maps


def _legacy_names(field_name: str) -> tuple[str, ...]:
    """The pre-ledger field names a ledger field answers to, for one release."""
    if field_name == "product_line":
        return ("product_line_id",)
    if field_name == "attribute:quantity":
        return ("quantity",)
    if field_name.startswith("attribute:") and field_name not in ("attribute:abv", "attribute:country_of_origin", "attribute:is_exclusive"):
        return ("attribute",)
    return ()


class _LegacyView:
    """A ledger row seen through a legacy field name: `product_line_id` reads the resolved line
    id (`value_ref_id`), the rest read `value`; `collected_value` is `rule_value`."""

    def __init__(self, row, field_name: str) -> None:
        self.row, self.field = row, field_name
        self.value = row.value_ref_id if field_name == "product_line" else row.value
        self.collected_value = row.rule_value
        self.set_by, self.set_at, self.reason = row.decided_by, row.decided_at, row.reason


def maps_for(db: Session) -> Maps:
    """The maps for this database, cached for CACHE_SECONDS; `invalidate()` after a write
    that changes a brand or a attribute alias."""
    key = id(db.get_bind())
    maps = _cache.get(key)
    if maps is None or time.monotonic() - maps.loaded_at > CACHE_SECONDS:
        maps = load_maps(db)
        _cache[key] = maps
    return maps


def invalidate() -> None:
    _cache.clear()


def brand_words(brand: Brand | None) -> str | None:
    """The brand as the key names it: the canonical row's slug as words, so a renamed brand
    keeps its key ("ysl" stays "ysl" when the name becomes Yves Saint Laurent)."""
    return brand.slug.replace("-", " ") if brand is not None else None


def product_line_for(db: Session, maps: Maps, brand: Brand | None, name: str | None, listed_brand: str | None,
             vertical: str | None, *, create: bool, category: str | None = None) -> ProductLine | None:
    """The canonical line a name belongs to under its brand: found by the computed key,
    followed through a line alias; created on first sight when `create` (ingest, like the
    brand row), inside a savepoint so a concurrent collector creating the same line does
    not poison the session. None without a brand."""
    if brand is None or not name:
        return None
    computed = product_lines.product_line_key(name, listed_brand=listed_brand, brand=brand_words(brand), vertical=vertical, category=category)
    row = maps.line(brand, computed)
    if row is not None or not create:
        return row
    slug = product_lines.line_slug(brand.slug, computed)[:230]
    savepoint = db.begin_nested()
    try:
        row = ProductLine(brand_id=brand.id, key=computed, slug=slug,
                          name=product_lines.display_line_name(computed, [name], brand.name,
                                                       brand_words=product_lines.brand_words_of(listed_brand, brand.name),
                                                       vertical=vertical, category=category)[:200])
        db.add(row)
        savepoint.commit()
    except IntegrityError:
        savepoint.rollback()
        row = db.scalar(select(ProductLine).where(ProductLine.brand_id == brand.id, ProductLine.key == computed))
        if row is None:
            return None
    maps.remember_line(row)
    return product_lines.resolve_alias(maps.lines, row.id) or row


#: A product nothing placed (no shelf, no category of ours) that states no quantity has none as
#: far as anyone can tell (a ring, a watch): its slot reads `n/a`, which equals itself, so the
#: same ring at two shops of one operator meets on brand, name and options. A placed product
#: that states no quantity stays `unknown`, which never equals anything (bottle sizes differ).
UNPLACED_VERTICALS = frozenset({"", "unknown"})


def quantity_slot(quantity: quantity_service.Quantity | None, vertical: str | None) -> str:
    slot = quantity_service.quantity_key(quantity)
    if slot == "unknown" and (vertical or "") in UNPLACED_VERTICALS:
        return "n/a"
    return slot


def _key(listed_brand: str | None, name: str | None, quantity_ml: int | None, *, maps: Maps, brand: Brand | None,
         residual: str | None, vertical: str | None, category: str | None, attribute: str,
         quantity: quantity_service.Quantity | None, options: dict) -> str:
    """The certain key through `normalize.match_key`, with the `n/a` slot for an unplaced product."""
    key = match_key(listed_brand, name or "", quantity_ml, vertical=vertical, brand=brand_words(brand), attribute=attribute,
                    line=residual, quantity=quantity, category=category, options=options)
    if key.endswith("|unknown") and (vertical or "") in UNPLACED_VERTICALS:
        key = key[: -len("unknown")] + "n/a"
    return key


def option_attributes_of(attributes: dict | None) -> dict:
    return {k: v for k, v in (attributes or {}).items() if k.startswith("option:")}


def listing_key(db: Session, raw: RawListing, attributes: dict,
                quantity: quantity_service.Quantity | None = None) -> tuple[str, dict, int | None]:
    """The key, the attributes and the line a listing resolves under: the collector's
    attributes (the concentration as parsed, the shop's published options) plus the canonical
    `attribute`; the certain key (identity rules v6) computed with the brand row, the line (an
    alias followed to its canonical) and the identity slot; and the line's id for a product
    variant created from it."""
    maps = maps_for(db)
    category = classify(raw.name, raw.brand, raw.feed_categories)
    # The same precedence a stored variant's `vertical` column was written with
    # (`ingest.resolve_vertical`): our category first, then the shelf the collector walked. The
    # two orders differed until v6, so a listing whose shelf and category disagreed could key
    # apart from its own variant.
    vertical = vertical_of(category) or raw.vertical
    attribute = maps.attribute(raw.name, vertical, category=category)
    attributes = dict(attributes)
    if attribute:
        attributes["attribute"] = attribute
        kind = maps.attribute_kind(raw.name, vertical, category=category)
        if kind:
            attributes["attribute_kind"] = kind
    brand = maps.brand_of(listed_brand=raw.brand, vertical=vertical)
    line = product_line_for(db, maps, brand, raw.name, raw.brand, vertical, create=True, category=category)
    key = _key(raw.brand, raw.name, raw.quantity_ml, maps=maps, brand=brand, residual=line.key if line else None,
               vertical=vertical, category=category, attribute=attribute, quantity=quantity,
               options=option_attributes_of(attributes))
    return key, attributes, line.id if line else None


def keyed_name(product: ProductVariant, maps: Maps) -> str:
    """The spelling every name-derived key part is computed from: the collected spelling
    kept in a `name` override's `collected_value` when a person renamed the product, else the
    name itself. A typed name is display only for the key: keying from it once lost the
    concentration ("1 Million" keys `rabanne|1-million||100ml`) and the next collection
    created a new product beside the renamed one."""
    row = maps.decided.get(product.id, {}).get("name")
    if row is not None and row.collected_value:
        return str(row.collected_value)
    return product.name


def product_quantity(product: ProductVariant, maps: Maps, *, decided: bool = True) -> quantity_service.Quantity:
    """A stored product's quantity, override first: a decided `quantity` override; else the
    v4 columns when a backfill or ingest wrote them; else the name parsed with the stored
    v3 pair as the hint, which is exactly what `backfill quantities` writes. `decided=False` is
    the key's reading: what the rules read, never a person's correction (`product_key`)."""
    row = maps.decided.get(product.id, {}).get("quantity")
    if row is not None and not decided:
        # The columns hold the decided quantity (the writer materialises it), so the key's reading
        # is what the rules had when the person decided, else the name read afresh.
        if not isinstance(row.collected_value, dict):
            return quantity_service.parse_quantity(keyed_name(product, maps), category=product.category)
    value = None if row is None else (row.value if decided else row.collected_value)
    if isinstance(value, dict):
        v = value
        return quantity_service.from_stored(
            v.get("value"), v.get("unit"), pack_count=v.get("pack_count"), pack_unit_value=v.get("pack_unit_value"),
            form=v.get("form") or "single", set_contents=v.get("set_contents"),
            state="stated" if v.get("value") is not None or v.get("set_contents") else "none",
        )
    if getattr(product, "quantity_state", None):
        return quantity_service.from_stored(
            product.quantity_value, product.quantity_unit, pack_count=product.pack_count,
            pack_unit_value=product.pack_unit_value, form=product.form, set_contents=product.set_contents,
            state=product.quantity_state,
        )
    hint = None
    if product.quantity_stated_value is not None and product.quantity_stated_unit:
        hint = (product.quantity_stated_value, product.quantity_stated_unit)
    elif product.quantity_ml:
        hint = (product.quantity_ml, "ml")
    return quantity_service.parse_quantity(keyed_name(product, maps), hint=hint, category=product.category)


def rule_residual(product: ProductVariant, maps: Maps, brand: Brand | None) -> str | None:
    """The key of the line the RULES give a stored variant, a confirmed line alias followed:
    exactly what ingest computes for the variant's own listing. Never the decided line."""
    if brand is None:
        return None
    computed = product_lines.product_line_key(keyed_name(product, maps), listed_brand=product.brand, brand=brand_words(brand),
                                              vertical=product.vertical, category=product.category)
    row = maps.line(brand, computed)
    return row.key if row is not None else computed


def product_key(product: ProductVariant, maps: Maps) -> tuple[str, dict]:
    """The certain key and the attributes of a stored variant under the current rules (rederive
    and the alias routes). The KEY is a pure function of the listed facts and the alias maps: the
    collected spelling (`keyed_name`), the brand row and the rules' line with their aliases
    followed, the options the shop published, the quantity the rules read. A person's decision on
    the variant (its line, an attribute, its quantity) changes the columns a page reads and never
    the key: under v5 a decided line moved the key, and the variant's own listing then keyed
    elsewhere at its next sighting and would have minted a duplicate beside every reviewed
    variant (no decision existed yet, so it never fired). An arrival lands on a decided variant
    because both compute this key; the automatic merge still refuses a group holding a decided
    member (`merges._conflicts`). The ATTRIBUTES returned do read the decided attribute first."""
    decided = maps.decided.get(product.id, {})
    name = keyed_name(product, maps)
    rule_attribute = maps.attribute(name, product.vertical, category=product.category)
    decided_attribute = decided.get("attribute")
    decided_kind = ""
    if decided_attribute is not None and decided_attribute.value is not None:
        attribute = str(decided_attribute.value)
        # The KIND of a decided attribute is the decided row's own field suffix
        # (`attribute:color`), never the rule's reading. The rule names no kind for a shade the
        # shop wrote without one ("ROUGE ALLURE 3.5gr / 196 A DEMI-MOT"), so taking the kind from
        # the rule wrote the batch tail's rekey with no `attribute_kind` and `decisions verify`
        # reported drift on every approved shade (K4 found it; one approval, 1 drift).
        field = str(getattr(decided_attribute, "field", "") or "")
        decided_kind = field.partition(":")[2] if field.startswith("attribute:") else ""
    else:
        attribute = rule_attribute
    attributes = {k: v for k, v in (product.attributes or {}).items() if k not in ("attribute", "attribute_kind")}
    if attribute:
        attributes["attribute"] = attribute
        kind = decided_kind or maps.attribute_kind(name, product.vertical, category=product.category)
        if kind:
            attributes["attribute_kind"] = kind
    brand = maps.brand_of(product.brand_id, product.brand, product.vertical)
    key = _key(product.brand, name, product.quantity_ml, maps=maps, brand=brand, residual=rule_residual(product, maps, brand),
               vertical=product.vertical, category=product.category, attribute=rule_attribute,
               quantity=product_quantity(product, maps, decided=False), options=option_attributes_of(product.attributes))
    return key, attributes