"""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 brand rows that hold several spellings because the list folded
  them (the fold still runs in `normalize.brand_key`: undoing it re-slugs brand rows, which is a
  decision on the running list), so a person confirms each.

`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 words `normalize.brand_key` folds off a brand spelling's end ("Appleton Estate"),
    in the vertical it is read in. `None` is every list's words, which is what a caller with no
    vertical to hand gets (`normalize.trailers_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 = "4"
GENERATOR = "rule"


def _evidence(name: str, words: list[str]) -> dict:
    return {"field": "name", "text": name, "span": ", ".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, "span": 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, "span": alias.raw}], confidence=None, brand_slug=""))
        write("skin_type_tails", rows)
    if "brand_trailers" in wanted:
        from app.services.normalize import brand_key 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, "span": ", ".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, "span": " ".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, 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(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