"""The word lists as proposal generators: what each list WOULD have grouped or read, written as
proposals with the rule's name as the reason, never applied (Stream K3; plan W14 option D).

Sources of truth: this module, `services/product_lines.py` (the certain key these readings start
from), `services/proposals_store.py` (the one store; the AI pass writes through it too),
`docs/REVIEW-PROCESS.md` section 1 ("Everything else is a proposal"), `tests/test_proposal_rules.py`.

Until identity rules v6 every list here deleted words from a name on its way to the key, and the
automatic merge then fired on the result. Each incident on record is one of them acting alone:
"triple" on the format list keyed Triple Cask as Cask; the audience fold put Burberry Her on a
line called "women"; dropping a brand's last word anywhere left Yves Saint Laurent's "Y" with an
empty line; a category noun ("spirit") joins Montblanc Legend with Legend Spirit (rehearsed 17
Sep). The knowledge in the lists is kept: a list still computes its reading, and the reading is
shown to a person on the review sheet beside the AI pass's own. One pass per list
(`rule:<list>:<rules version>`), so a list that proves bad is withdrawn alone
(`app.cli proposals withdraw --pass rule:<list>:6`).

Three shapes of proposal, all through `proposals_store.write`:
* a GROUPING list proposes that two product lines of one brand are one (`pair:line:...`, value
  `same`): under that list alone their residual names meet. `lists_together` proposes the pairs
  that meet only when several lists act at once ("12 Year Old Blended Scotch Whisky" and "12").
  `pack_words` also proposes `separate` where it reads a pack word as the expression.
* an EXTRACTING list proposes an attribute on a variant with the span it read (`shade_shapes`,
  `cask_words`), or the kind of a marked wording (`skin_type_tails`).
* `brand_trailers` reports the spellings that differ only by a trailing category or corporate
  word, so a person confirms each. Since K11.3 the fold is GONE from `normalize.brand_key`, so
  these are separate brand rows until somebody joins them -- which is the point: the stage errs
  toward separate and the join is a decision.

`normalize._STOPWORDS` is not here: it never touched identity. It is the tokenizer `verify` and
the image matcher compare two spellings with, and it groups nothing.
"""

from __future__ import annotations

import re
from collections import Counter, defaultdict
from dataclasses import dataclass, field

from app.services import product_lines
from app.services.normalize import _BRAND_TRAILERS, _NOISE_RE

# --------------------------------------------------------------------------- the lists
# Words that describe the format, the packaging or the edition (v5: never the line).
FORMAT_WORDS = frozenset({
    "spray", "vapo", "vaporisateur", "vaporizador", "vaporizer", "natural", "rechargeable",
    "recargable", "refillable", "refill", "recharge", "jumbo", "repack", "new", "travel",
    "exclusive", "edition", "limited", "collector", "collectors", "set", "coffret", "gift",
    "giftset", "duo", "trio", "kit", "pack", "twinpack", "tripack",
    "bundle", "estuche", "miniature", "miniatures", "mini", "deluxe", "size", "bottle",
    "bottles", "btl", "ml", "cl", "oz", "fl", "gr", "g", "ltr", "lt",
    "gp", "gb", "tube", "tin", "box", "carton", "canister", "case", "can", "pet", "trx", "ck",
    "vol", "abv", "proof",
})
#: "triple", "tri", "twin": packaging when the name is a pack or a set or the next word is a drink
#: word ("triple malt"), the expression otherwise ("Triple Cask", "Twin Barrel", "Triple Serum").
PACK_EXPRESSION_WORDS = frozenset({"triple", "tri", "twin"})
ARTICLES = frozenset({"the", "le", "la", "les", "l", "el", "los", "las", "il", "lo", "gli", "der", "die", "das"})
CONNECTORS = frozenset({"by", "for", "pour", "and", "et", "y", "with", "de", "du", "des", "di", "da", "of", "a", "an"}) | ARTICLES
AUDIENCE = {"homme": "men", "hommes": "men", "man": "men", "him": "men", "men": "men", "male": "men",
            "herren": "men", "uomo": "men", "hombre": "men",
            "femme": "women", "femmes": "women", "woman": "women", "her": "women", "female": "women",
            "women": "women", "damen": "women", "donna": "women", "mujer": "women", "ladies": "women"}
#: Category and describing words a drinks shop appends to the expression.
DRINK_WORDS = frozenset({
    "whisky", "whiskey", "whiskies", "scotch", "bourbon", "single", "malt", "malts", "blended",
    "blend", "grain", "gin", "vodka", "rum", "rhum", "ron", "tequila", "mezcal", "cognac",
    "brandy", "armagnac", "calvados", "liqueur", "liqueurs", "liquor", "liquer", "wine", "wines",
    "champagne", "prosecco", "cava", "sparkling", "beer", "lager", "cider", "spirit", "spirits",
    "dry", "london", "premium", "brut",
})
REGION_WORDS = frozenset({
    "highland", "highlands", "speyside", "islay", "lowland", "campbeltown", "scotland",
    "scottish", "ireland", "irish", "kentucky", "tennessee", "sweden", "swedish", "france",
    "french", "mexico", "mexican", "japan", "japanese", "swiss",
})
AGE_WORDS = frozenset({"old", "aged", "years", "year", "yrs", "yr", "yo", "ans", "jahre", "anos"})
CASK_WORDS = product_lines.CASK_WORDS
#: A marked tail that is a skin type (the Skincare shelves' second option, Spanish and English).
SKIN_TYPE_TAILS = frozenset({
    "grasa", "seca", "muy seca", "mixta", "normal", "sensible", "radiante", "todo tipo de piel", "todo tipo",
    "todas las pieles", "dry", "very dry", "oily", "combination", "sensitive", "normal to dry",
    "normal to oily", "all skin types", "all skin",
})
BRAND_TRAILERS = _BRAND_TRAILERS

#: The grouping lists, in the order the sheet shows them, each with the sentence a person reads.
GROUPING_RULES: dict[str, str] = {
    "drink_words": "removed {words} as drink category or describing words",
    "region_words": "removed {words} as region words",
    "age_words": "read {words} as the age's own words, so a bare number and a stated age meet",
    "format_words": "removed {words} as format, packaging or edition words",
    "noise": "removed {words} as packaging or shop wording",
    "connectors": "removed {words} as articles or connectors",
    "audience": "read {words} as one audience word",
    "pack_words": "removed {words} as pack words (the name reads as a pack or a set)",
    "brand_partial": "removed {words} as part of the brand's name",
}
EXTRACTING_RULES = ("shade_shapes", "cask_words", "skin_type_tails", "brand_trailers")
TOGETHER = "lists_together"
ALL_RULES = (*GROUPING_RULES, TOGETHER, *EXTRACTING_RULES)


def pass_name(rule: str, rules_version: str) -> str:
    return f"rule:{rule}:{rules_version}"


# --------------------------------------------------------------------------- the pure readings

@dataclass
class Reading:
    """What one or more grouping lists make of a name: the residual name they leave, and per
    list the words it took (the span a person is shown)."""

    key: str
    removed: dict[str, list[str]] = field(default_factory=dict)

    def reason(self) -> str:
        return "; ".join(GROUPING_RULES[rule].format(words=", ".join(f"'{w}'" for w in words))
                         for rule, words in self.removed.items())


def pack_word_reading(name: str | None) -> tuple[str, str]:
    """`("expression" | "packaging", why)` for a name carrying "triple", "tri" or "twin": the
    quantity parser's form decides. "Macallan Triple Cask 12" is `("expression", "form: single")`;
    "Chivas Regal 12 Triple Pack 3x1L" is `("packaging", "form: pack")`."""
    form = product_lines.form_of(name) or "single"
    return ("packaging" if form in ("pack", "set") else "expression"), f"form: {form}"


def read_line(name: str | None, *, listed_brand: str | None = None, brand: str | None = None,
              vertical: str | None = None, category: str | None = None,
              rules: frozenset[str] | set[str] = frozenset()) -> Reading:
    """The residual name with the given lists applied ON TOP of the certain key, and what each
    took. With every grouping list this is the identity rules v5 line key, which is how the old
    list-deletion tests became proposal tests: the words v5 deleted are the words proposed."""
    rules = frozenset(rules)
    beauty = vertical in product_lines.BEAUTY_VERTICALS
    removed: dict[str, list[str]] = defaultdict(list)
    plain = product_lines.residual_tokens(name, listed_brand=listed_brand, brand=brand, vertical=vertical, category=category)
    tokens = product_lines.residual_tokens(name, listed_brand=listed_brand, brand=brand, vertical=vertical, category=category,
                                           noise="noise" in rules, partial_brand="brand_partial" in rules)
    if "noise" in rules:
        removed["noise"].extend(" ".join(m.group(0).lower().split()) for m in _NOISE_RE.finditer(product_lines._fold(name)))
    if "brand_partial" in rules:
        took = list(plain)
        for t in tokens:
            if t in took:
                took.remove(t)
        removed["brand_partial"].extend(t for t in took if "noise" not in rules or t not in " ".join(removed["noise"]).split())
    packaging = pack_word_reading(name)[0] == "packaging"
    flags = product_lines.boundary_flags()
    kept: list[str] = []
    for index, token in enumerate(tokens):
        following = tokens[index + 1] if index + 1 < len(tokens) else ""
        if product_lines._drops(tokens, index, flags):
            continue
        if "age_words" in rules and re.fullmatch(r"\d{1,2}yo", token):
            removed["age_words"].append(token)
            kept.append(token[:-2])
            continue
        if beauty and "audience" in rules and token in AUDIENCE:
            if AUDIENCE[token] != token:
                removed["audience"].append(token)
            kept.append(AUDIENCE[token])
            continue
        if not beauty and following in CASK_WORDS and token in DRINK_WORDS | REGION_WORDS:
            kept.append(token)  # "Rum Cask", "Port Cask Finish": the word names the cask
            continue
        if "pack_words" in rules and token in PACK_EXPRESSION_WORDS and (packaging or following in DRINK_WORDS):
            removed["pack_words"].append(token)
            continue
        took = None
        if not beauty and "drink_words" in rules and token in DRINK_WORDS:
            took = "drink_words"
        elif not beauty and "region_words" in rules and token in REGION_WORDS:
            took = "region_words"
        elif not beauty and "age_words" in rules and token in AGE_WORDS:
            took = "age_words"
        elif "format_words" in rules and token in FORMAT_WORDS:
            took = "format_words"
        elif "connectors" in rules and token in CONNECTORS:
            took = "connectors"
        if took:
            removed[took].append(token)
            continue
        kept.append(token)
    return Reading(" ".join(kept)[: product_lines.KEY_MAX].rstrip(), {k: v for k, v in removed.items() if v})


_EXTIME_SHADE_RE = re.compile(r"\s-\s+(\d{1,3}[A-Za-z]?\s+[^\d/]{2,40})$")
_BARE_SHADE_RE = re.compile(r"(?<![\d.,])\b(\d{2,3})$")


def read_shade(name: str | None, vertical: str | None, category: str | None) -> tuple[str, str] | None:
    """`(value, span)` for a shade a Makeup name carries UNMARKED, in the two shapes the shops
    use: a trailing " - 447 Mellow Shade" (Extime) and a trailing bare code "01" (Avolta). Read
    from the name, so never certain: "Brush N°13" and "Chanel N°5" end in a number too."""
    if vertical not in product_lines.BEAUTY_VERTICALS or category != "Makeup" or not name:
        return None
    text = product_lines._strip_measures(name).rstrip(" -,.")
    match = _EXTIME_SHADE_RE.search(text)
    if match:
        return " ".join(match.group(1).lower().split()), match.group(0).strip()
    match = _BARE_SHADE_RE.search(text)
    if match and not re.search(r"n[o°º]?\.?\s*\d+$", text, re.I):
        return match.group(1), match.group(1)
    return None


_CASK_PHRASE_RE = re.compile(r"\b((?:[a-z]+\s+){1,2})(cask|casks|finish|barrel|wood)\b")


def read_cask(name: str | None, vertical: str | None) -> tuple[str, str] | None:
    """`(value, span)` for a cask or finish a drink's name states ("Caribbean Rum Cask",
    "Oloroso Sherry Finish"): proposed as the attribute `cask`; the words stay in the name
    (whether a finish is a member of its parent line is a grouping default, REVIEW-PROCESS 2)."""
    if vertical != "liquor" or not name:
        return None
    match = _CASK_PHRASE_RE.search(product_lines._fold(name))
    if not match or match.group(1).split()[-1] in PACK_EXPRESSION_WORDS | {"single", "the", "a"}:
        return None
    phrase = " ".join(match.group(0).split())
    return phrase, phrase


def read_skin_type(tail: str | None) -> str | None:
    text = " ".join((tail or "").lower().split())
    return text if text in SKIN_TYPE_TAILS else None


def trailer_words(spelling: str | None, vertical: str | None = None) -> list[str]:
    """The trailing words that are not the brand ("Appleton Estate" -> ["estate"]), in the
    vertical the spelling is read in (`None` is every list's words).

    `normalize.brand_key` no longer folds these off (K11.3); this list is what proposes the join
    to a person instead, so the knowledge in it is kept while the silent grouping is not. The
    per-vertical scoping (K9) is what keeps the proposal sane in a vertical the lists were not
    written for.
    """
    from app.services.normalize import trailers_for

    trailers = trailers_for(vertical)
    words = re.sub(r"[^a-z0-9]+", " ", product_lines._fold(spelling)).split()
    took: list[str] = []
    while len(words) > 1 and words[-1] in trailers:
        took.insert(0, words.pop())
    return took


# --------------------------------------------------------------------------- the generators (database)
#: The version line of `docs/REVIEW-PROCESS.md` these passes follow (a test holds the two equal).
PROCESS_VERSION = "5"
GENERATOR = "rule"


def _evidence(name: str, words: list[str]) -> dict:
    """A rule's evidence: the text it read and the words it matched in it.

    `span` stays empty here on purpose. The spec's span is a pair of character offsets into ONE
    listing's listed text, which a session's proposal cites; a rule reads a variant's own name and
    matches words in its folded form, so there are no offsets to give. Putting the words there
    instead made the sheet answer 500 (`span: '01'` against `list[int]`) and would have read as a
    position wherever it did parse."""
    return {"field": "name", "text": name, "matched": ", ".join(words)}


def generate(db, rules: tuple[str, ...] | list[str] | None = None, *, check: bool = False) -> dict[str, int]:
    """Every list's reading of the live catalogue, written as proposals: one pass per list, rows
    upserted on the store's idempotency key, so a second run inserts nothing. NOTHING is applied:
    no key, line, attribute or brand changes here. `check` counts without writing. A line pair a
    person already decided (same or separate) is never proposed again."""
    from sqlalchemy import select

    from app.models import AttributeAlias, Brand, ProductLine, ProductVariant, Suggestion
    from app.models.catalog import IDENTITY_RULES_VERSION
    from app.services import keying, proposals_store
    from app.services.decisions import natural_keys

    wanted = tuple(rules) if rules else ALL_RULES
    maps = keying.load_maps(db)
    variants = list(db.scalars(select(ProductVariant).where(ProductVariant.merged_into_id.is_(None)).order_by(ProductVariant.id)))
    lines = {row.id: row for row in db.scalars(select(ProductLine))}
    decided_pairs = {(s.left_id, s.right_id) for s in db.scalars(select(Suggestion).where(Suggestion.level == "line"))
                     if getattr(s, "decision", None)}
    counts: dict[str, int] = {}

    def write(rule: str, rows: list) -> None:
        counts[rule] = len(rows)
        if rows and not check:
            proposals_store.write(db, pass_name(rule, IDENTITY_RULES_VERSION), rows, kind="rule", generator=GENERATOR,
                                  rules_version=IDENTITY_RULES_VERSION, process_version=PROCESS_VERSION,
                                  note=f"what the {rule} list would have grouped or read; nothing applied")

    # ---- the grouping lists: per brand, the lines whose residual names meet under the list
    placed = []
    for v in variants:
        brand = maps.brand_of(v.brand_id, v.brand, v.vertical)
        line = product_lines.resolve_alias(lines, v.product_line_id) if v.product_line_id else None
        if brand is not None and line is not None:
            placed.append((v, brand, line))
    sizes: dict[int, int] = defaultdict(int)
    for _, _, line in placed:
        sizes[line.id] += 1
    already: set[tuple[int, int]] = set()

    def grouping(rule: str, rule_set: frozenset[str], skip: set[tuple[int, int]]) -> list:
        meet: dict[tuple[int, str], dict[int, tuple]] = defaultdict(dict)
        for v, brand, line in placed:
            name = keying.keyed_name(v, maps)
            reading = read_line(name, listed_brand=v.brand, brand=keying.brand_words(brand), vertical=v.vertical,
                                category=v.category, rules=rule_set)
            slot = meet[(brand.id, reading.key)]
            if line.id not in slot or (reading.removed and not slot[line.id][1].removed):
                slot[line.id] = (name, reading, brand)
        rows = []
        for (_brand_id, _key), members in meet.items():
            if len(members) < 2 or not any(r.removed for _, r, _ in members.values()):
                continue
            anchor_id = max(members, key=lambda i: (sizes[i], -i))
            for member_id, (name, reading, brand) in sorted(members.items()):
                if member_id == anchor_id:
                    continue
                pair = (min(anchor_id, member_id), max(anchor_id, member_id))
                if pair in skip or pair in decided_pairs:
                    continue
                skip.add(pair)
                anchor_name, anchor_reading, _ = members[anchor_id]
                said = "; ".join(x for x in (reading.reason(), anchor_reading.reason()) if x)
                key, detail = natural_keys.pair_key("line", *pair, sides=(natural_keys.side_key("line", lines[pair[0]]),
                                                                           natural_keys.side_key("line", lines[pair[1]])))
                if not check:
                    _ensure_pair(db, pair, rule, said)
                rows.append(proposals_store.ProposalRow(
                    entity_type="suggestion", natural_key=key, natural_key_detail=detail, field="decision",
                    value={"decision": "same"}, reason=f"rule:{rule}: {said}; the two product lines then read as one",
                    evidence=[_evidence(name, sum(reading.removed.values(), [])),
                              _evidence(anchor_name, sum(anchor_reading.removed.values(), []))],
                    confidence=None, brand_slug=brand.slug, sheet_line_ref=f"line:{lines[anchor_id].uid}"))
        return rows

    for rule in GROUPING_RULES:
        if rule in wanted:
            write(rule, grouping(rule, frozenset({rule}), already))
    if TOGETHER in wanted:
        # Only the pairs no single list makes: every single-list pair is already in `already`.
        singles = set(already)
        if not any(r in wanted for r in GROUPING_RULES):
            for rule in GROUPING_RULES:
                grouping(rule, frozenset({rule}), singles)
        write(TOGETHER, grouping(TOGETHER, frozenset(GROUPING_RULES), singles))

    # ---- the extracting lists
    def variant_row(v, brand, field_name: str, value: str, span: str, rule: str, why: str):
        key, detail = natural_keys.build(v)
        return proposals_store.ProposalRow(
            entity_type="product_variant", natural_key=key, natural_key_detail=detail, field=field_name, value=value,
            reason=f"rule:{rule}: {why}", evidence=[{"field": "name", "text": v.name, "matched": span}], confidence=None,
            brand_slug=brand.slug, sheet_line_ref=f"line:{lines[v.product_line_id].uid}" if v.product_line_id in lines else None)

    if "shade_shapes" in wanted:
        rows = []
        for v, brand, _line in placed:
            if (v.attributes or {}).get("attribute") or any(k.startswith("option:") for k in (v.attributes or {})):
                continue
            found = read_shade(keying.keyed_name(v, maps), v.vertical, v.category)
            if found:
                rows.append(variant_row(v, brand, "attribute:color", found[0], found[1], "shade_shapes",
                                        "the name ends in a shade the shop did not publish as a field"))
        write("shade_shapes", rows)
    if "cask_words" in wanted:
        rows = []
        for v, brand, _line in placed:
            found = read_cask(keying.keyed_name(v, maps), v.vertical)
            if found and not (v.attributes or {}).get("cask"):
                rows.append(variant_row(v, brand, "attribute:cask", found[0], found[1], "cask_words",
                                        "the name states a cask or a finish; the words stay in the name"))
        write("cask_words", rows)
    if "skin_type_tails" in wanted:
        rows = []
        for alias in db.scalars(select(AttributeAlias).where(AttributeAlias.vertical == "beauty")):
            if read_skin_type(alias.raw) and (alias.kind or "") != "skin_type":
                rows.append(proposals_store.ProposalRow(
                    entity_type="attribute_wording", natural_key=f"wording:{alias.vertical}|{alias.raw}", natural_key_detail=None,
                    field="kind", value="skin_type", reason="rule:skin_type_tails: the marked wording is a skin type, not a shade",
                    evidence=[{"field": "option", "text": alias.raw, "matched": alias.raw}], confidence=None, brand_slug=""))
        write("skin_type_tails", rows)
    if "brand_trailers" in wanted:
        # The FOLDED key: this generator's whole job is to group the brand rows the list would
        # have joined, and since K11.3 the plain key joins nothing. It is now the only thing that
        # finds two rows of one house (Rabanne and Paco Rabanne), which is what rian asked for.
        from app.services.normalize import brand_key_folded as _brand_key

        # A brand row's vertical is the one most of its live variants sit in; that is the list
        # that acted on its spellings, so it is the list a person is shown.
        brand_verticals: dict[int, Counter] = defaultdict(Counter)
        for v in variants:
            if v.brand_id:
                brand_verticals[v.brand_id][v.vertical or ""] += 1
        spellings: dict[int, dict[str, list[str]]] = defaultdict(dict)
        for v in variants:
            if not v.brand_id:
                continue
            took = trailer_words(v.brand, brand_verticals[v.brand_id].most_common(1)[0][0] or None)
            if took:
                spellings[v.brand_id][v.brand.strip()] = took
        rows = []
        for brand_id, found in sorted(spellings.items()):
            brand = maps.brands.get(brand_id)
            if brand is None or brand.alias_of_id:
                continue
            listed = "; ".join(f"'{spelling}' (removed {', '.join(took)})" for spelling, took in sorted(found.items()))
            rows.append(proposals_store.ProposalRow(
                entity_type="brand", natural_key=f"brand:{brand.slug}", natural_key_detail={"name": brand.name}, field="name",
                value=brand.name, reason=f"rule:brand_trailers: these spellings sit on this brand because the list folded them: {listed}; "
                                         "approve to confirm they are this brand",
                evidence=[{"field": "brand", "text": spelling, "matched": ", ".join(took)} for spelling, took in sorted(found.items())],
                confidence=None, brand_slug=brand.slug))
        # And the other half of the same list, since it was scoped per vertical (K9.5): two brand
        # ROWS that the unscoped list WOULD have folded into one, in a vertical nobody has
        # written a list for. The fold no longer acts there, so it is offered as a pair for a
        # person to Confirm same (which writes the alias) or Keep separate. This is what stops
        # two fashion companies whose names differ only by "London" from silently becoming one.
        by_full: dict[str, list] = defaultdict(list)
        for row in maps.brands.values():
            if row.alias_of_id or not brand_verticals.get(row.id):
                continue
            by_full[_brand_key(row.slug.replace("-", " "))].append(row)
        decided_brands = {(s.left_id, s.right_id) for s in db.scalars(select(Suggestion).where(Suggestion.level == "brand"))
                          if getattr(s, "decision", None)}
        for full, members in sorted(by_full.items()):
            if len(members) < 2:
                continue
            anchor = max(members, key=lambda r: (sum(brand_verticals[r.id].values()), -r.id))
            for other in sorted(members, key=lambda r: r.id):
                if other.id == anchor.id:
                    continue
                pair = (min(anchor.id, other.id), max(anchor.id, other.id))
                if pair in decided_brands:
                    continue
                vertical = brand_verticals[other.id].most_common(1)[0][0] or "no vertical"
                took = trailer_words(other.slug.replace("-", " ")) or trailer_words(anchor.slug.replace("-", " "))
                if not check:
                    _ensure_pair(db, pair, "brand_trailers", f"both read as {full!r}", level="brand")
                left, right = maps.brands[pair[0]], maps.brands[pair[1]]
                key, detail = natural_keys.pair_key("brand", *pair, sides=(f"brand:{left.slug}", f"brand:{right.slug}"))
                rows.append(proposals_store.ProposalRow(
                    entity_type="suggestion", natural_key=key, natural_key_detail=detail, field="decision",
                    value={"decision": "same"},
                    reason=f"rule:brand_trailers: removing {', '.join(repr(w) for w in took) or 'a trailing listed word'} would read "
                           f"'{other.name}' and '{anchor.name}' as one brand, but the {vertical} vertical has no trailer list, "
                           "so they were kept apart; confirm only if they are one company",
                    evidence=[{"field": "brand", "text": other.name, "span": ", ".join(took)},
                              {"field": "brand", "text": anchor.name, "span": ""}],
                    confidence=None, brand_slug=anchor.slug))
        write("brand_trailers", rows)
    if not check:
        db.flush()
    return counts


def _ensure_pair(db, pair: tuple[int, int], rule: str, why: str, level: str = "line") -> None:
    """The `suggestions` row a pair proposal resolves to (the store parks a pair with no row)."""
    from sqlalchemy import select
    from sqlalchemy.exc import IntegrityError

    from app.models import Suggestion

    thing = "product line" if level == "line" else level
    row = db.scalar(select(Suggestion).where(Suggestion.level == level, Suggestion.left_id == pair[0], Suggestion.right_id == pair[1]))
    if row is not None:
        return
    savepoint = db.begin_nested()
    try:
        db.add(Suggestion(level=level, left_id=pair[0], right_id=pair[1], reason=f"rule:{rule}"[:40], score=0.5,
                          detail={"why": f"the {rule} list reads the two names as one {thing}: {why}"[:500]}))
        savepoint.commit()
    except IntegrityError:
        savepoint.rollback()


def nearest_approved_line(db, variant):
    """The approved product line of the variant's brand whose residual name shares the most words
    with the variant's (a line is approved when a person's decision names it: its own name or
    alias decision, or a variant decided onto it); None when none shares a word that is not a
    bare number, or the variant already sits on it."""
    from sqlalchemy import select

    from app.models import Decision, ProductLine

    if variant.brand_id is None or variant.product_line_id is None:
        return None, 0.0
    own = db.get(ProductLine, variant.product_line_id)
    if own is None:
        return None, 0.0
    approved = set(db.scalars(select(Decision.value_ref_id).where(Decision.entity_type == "product_variant", Decision.field == "product_line")))
    approved |= set(db.scalars(select(Decision.entity_id).where(Decision.entity_type == "product_line")))
    approved |= set(db.scalars(select(Decision.value_ref_id).where(Decision.entity_type == "product_line", Decision.field == "alias_of")))
    words = set(own.key.split())
    best, best_score = None, 0.0
    for line in db.scalars(select(ProductLine).where(ProductLine.brand_id == own.brand_id, ProductLine.alias_of_id.is_(None),
                                                     ProductLine.id != own.id)):
        if line.id not in approved:
            continue
        theirs = set(line.key.split())
        shared = words & theirs
        if not any(not w.isdigit() for w in shared):
            continue
        score = len(shared) / len(words | theirs)
        if score > best_score:
            best, best_score = line, score
    return best, best_score


def propose_arrival(db, variant) -> bool:
    """A new variant that matched nothing, written as the proposal "addition to <nearest approved
    line>" (pass `arrival:<rules version>`, generator `arrival`); it stays on its own new product
    line until a person approves. Returns whether a proposal was written."""
    from app.models import Brand
    from app.models.catalog import IDENTITY_RULES_VERSION
    from app.services import proposals_store
    from app.services.decisions import natural_keys

    line, score = nearest_approved_line(db, variant)
    if line is None:
        return False
    brand = db.get(Brand, line.brand_id)
    key, detail = natural_keys.build(variant)
    row = proposals_store.ProposalRow(
        entity_type="product_variant", natural_key=key, natural_key_detail=detail, field="product_line",
        value=natural_keys.build(line)[0],
        reason=f"arrival: a new listing matched no variant; addition to {line.name} (the nearest approved product line, "
               f"{score:.0%} of the words shared)",
        evidence=[{"field": "name", "text": variant.name, "matched": " ".join(sorted(set(line.key.split()) & set(product_lines._fold(variant.name).split())))}],
        confidence=None, brand_slug=brand.slug if brand else "", sheet_line_ref=f"line:{line.uid}")
    proposals_store.write(db, f"arrival:{IDENTITY_RULES_VERSION}", [row], kind="arrival", generator="arrival",
                          rules_version=IDENTITY_RULES_VERSION, process_version=PROCESS_VERSION,
                          note="new listings that matched nothing, offered to the nearest approved product line")
    return True


# --------------------------------------------------------------------------- what the lists are, for a person

#: Every word list in the catalogue, in the order the review area shows them: what it is called,
#: the sentence a person reads, the words themselves, the verticals it applies to, and whether it
#: ACTS or only proposes. Rian asked for exactly this, 17 Sep: *"do we have a list of those stop
#: words somewhere maybe in the review dashboard so we know when we're doing that"*. Read-only;
#: nothing here writes, and the panel that draws it writes nothing either.
_EVERY = "every vertical"


def _describe() -> list[dict]:
    from app.services import normalize

    def row(key, what, words, verticals, acts=False, pattern=None):
        return {"key": key, "what": what, "words": sorted(words), "verticals": verticals,
                "acts": acts, "pattern": pattern}

    out = [
        row("brand_trailers",
            "a trailing word dropped off a brand's spelling before the listing is resolved to a brand row, "
            "so two spellings land on one row",
            normalize._TRAILERS_LIQUOR, ["liquor"], acts=True),
        row("brand_trailers_beauty",
            "the same list in beauty: the house words and the two cities that are part of a beauty house's name",
            normalize._TRAILERS_BEAUTY, ["beauty"], acts=True),
        row("brand_trailers_every",
            "a company's legal form and the article, true in any vertical and never a judgement",
            normalize._TRAILERS_EVERY_VERTICAL, [_EVERY], acts=True),
        row("drink_words", GROUPING_RULES["drink_words"].format(words="these"), DRINK_WORDS, ["liquor"]),
        row("region_words", GROUPING_RULES["region_words"].format(words="these"), REGION_WORDS, ["liquor"]),
        row("age_words", GROUPING_RULES["age_words"].format(words="these"), AGE_WORDS, ["liquor"]),
        row("cask_words", "a cask or a finish a drink's name states, read as the attribute `cask`; the words stay in the name",
            CASK_WORDS, ["liquor"]),
        row("format_words", GROUPING_RULES["format_words"].format(words="these"), FORMAT_WORDS, [_EVERY]),
        row("pack_words", GROUPING_RULES["pack_words"].format(words="these"), PACK_EXPRESSION_WORDS, [_EVERY]),
        row("connectors", GROUPING_RULES["connectors"].format(words="these"), CONNECTORS, [_EVERY]),
        row("audience", GROUPING_RULES["audience"].format(words="these"), sorted(AUDIENCE), ["beauty"]),
        row("skin_type_tails", "a marked wording read as a skin type rather than a shade", SKIN_TYPE_TAILS, ["beauty"]),
        row("noise", GROUPING_RULES["noise"].format(words="these"), [], [_EVERY],
            pattern=_NOISE_RE.pattern),
        row("shade_shapes", "a shade at the end of a name the shop did not publish as a field", [], ["beauty"],
            pattern="a number, a code or a colour word at the end of the name"),
        row("brand_partial", GROUPING_RULES["brand_partial"].format(words="these"), [], [_EVERY],
            pattern="part of the brand's own name, repeated at the head of the product's"),
        row("lists_together", "the pairs no single list makes, which only meet when several act at once", [], [_EVERY]),
        row("stopwords", "NOT an identity list: the words two spellings are compared on when a name is matched to "
                         "another (verification, image matching). It groups nothing and removes nothing.",
            normalize._STOPWORDS, [_EVERY]),
    ]
    return out


def word_lists(db) -> list[dict]:
    """Every word list with its words, where it applies, what it did, and where to go and see.

    The counts are read from the proposals store per pass (`rule:<list>:<rules version>`), and
    for the one list that still ACTS -- the brand trailers -- also from the catalogue: how many
    brand rows hold more than one listed spelling because a word of that list was dropped. That
    is the number rian cannot otherwise see, because a fold leaves no record of itself.
    """
    from sqlalchemy import func, select

    from app.models import Brand, ProductVariant, Proposal, ProposalPass
    from app.models.catalog import IDENTITY_RULES_VERSION
    from app.services.normalize import brand_key_folded, trailers_for

    passes = {p.name: p for p in db.scalars(select(ProposalPass))}
    tallies: dict[int, dict[str, int]] = defaultdict(lambda: defaultdict(int))
    brands_with_open: dict[int, list[str]] = defaultdict(list)
    for pass_id, status, brand_slug, n in db.execute(
            select(Proposal.pass_id, Proposal.status, Proposal.brand_slug, func.count(Proposal.id))
            .group_by(Proposal.pass_id, Proposal.status, Proposal.brand_slug)):
        tallies[pass_id][status] += n
        tallies[pass_id]["total"] += n
        if status == "open" and brand_slug:
            brands_with_open[pass_id].append(brand_slug)

    # What the acting list actually folded, counted from the rows themselves.
    spellings: dict[int, set[str]] = defaultdict(set)
    verticals: dict[int, Counter] = defaultdict(Counter)
    for brand_id, text, vertical in db.execute(
            select(ProductVariant.brand_id, ProductVariant.brand, ProductVariant.vertical)
            .where(ProductVariant.merged_into_id.is_(None), ProductVariant.brand_id.isnot(None))):
        if text and text.strip():
            spellings[brand_id].add(text.strip())
            verticals[brand_id][vertical or ""] += 1
    from app.services.normalize import _TRAILERS_BEAUTY, _TRAILERS_EVERY_VERTICAL, _TRAILERS_LIQUOR, brand_spelling

    folded: Counter = Counter()
    for brand_id, held in spellings.items():
        # A fold worth counting is one that JOINED two spellings that are otherwise different:
        # a row holding only "X Beauty" and "X Beauty(R)" had a word dropped but nothing joined.
        if len(held) < 2 or len({brand_spelling(s) for s in held}) < 2:
            continue
        vertical = verticals[brand_id].most_common(1)[0][0] or None
        trailers = trailers_for(vertical)
        took = {w for s in held for w in trailer_words(s, vertical)}
        if not took or len({brand_key_folded(s, vertical) for s in held}) != 1:
            continue
        for key, words in (("brand_trailers", _TRAILERS_LIQUOR), ("brand_trailers_beauty", _TRAILERS_BEAUTY),
                           ("brand_trailers_every", _TRAILERS_EVERY_VERTICAL)):
            if took & words & trailers:
                folded[key] += 1

    out = []
    for item in _describe():
        pass_row = passes.get(pass_name(item["key"].replace("_beauty", "").replace("_every", ""), IDENTITY_RULES_VERSION))
        counts = dict(tallies.get(pass_row.id, {})) if pass_row is not None else {}
        brands = sorted(set(brands_with_open.get(pass_row.id, []))) if pass_row is not None else []
        out.append({**item, "word_count": len(item["words"]),
                    "pass_name": pass_row.name if pass_row is not None else None,
                    "withdrawn": bool(pass_row is not None and pass_row.withdrawn_at is not None),
                    "proposals": {k: int(v) for k, v in counts.items()},
                    "brands": brands[:8], "brand_count": len(brands),
                    "brand_rows_folded": int(folded.get(item["key"], 0)) if item["acts"] else None})
    return out