"""Splitting a brand row the fold joined wrongly, as one recorded decision with an undo.

Sources of truth: this module, `decisions/appliers.py` (the `(brand, split)` applier),
`normalize.brand_key` (the fold it answers), `merges.prune_lines`, `tests/test_brand_split.py`,
`docs/REVIEW-PROCESS.md` section 1.

Why it is here and not in `merges.py`: that module's whole contract is two rows becoming one,
and it says so in its first sentence. A split is the opposite operation, with its own refusals,
and filing it there would make that sentence false. It borrows what it needs (`prune_lines`,
`rekey_product_variants` at the batch's tail) and owns nothing of the merge.

## What a split is

`normalize.brand_key` drops a trailing listed word before a listing's brand text is resolved to
a row, so "Appleton", "Appleton Estate" and "Appleton Rum" compute one key and land on one brand.
Every fold in today's data is right; the first category beyond drinks and beauty is where a
fashion brand named "<name> London" meets another silently. When one IS wrong, nothing undoes it:
the fold is a rule, not a decision, and `undo.unmerge` works on variants only.

So: a person names the spellings that do not belong, and the row they should sit on. That is one
`decisions` row on the source brand, `field = "split"`, whose value carries the new brand's slug,
its name and the spellings it claims, and whose detail carries every moved variant with the brand
and product line it came from. Undo reads that detail back (`appliers._brand_split_release`).

## The claim, and why the split survives the next collection

The new row's slug cannot be the fold key of its own spellings: `brand_key("Appleton Rum")` IS
"appleton", the row we just split away from. Left there, the next collection would resolve the
spelling straight back onto the source and mint duplicates beside the moved variants.

So the split's spellings are a CLAIM, read from the ledger by `keying.load_maps` and consulted
before the fold by `ingest.resolve_brand` and `Maps.brand_of`. The claim lives in the decision,
not in a column: an undo releases the field and the claim goes with it, with no second thing to
keep in step and no migration. The key follows because `keying.brand_words` reads the brand
ROW's slug, so a moved variant re-keys under the new row at the batch's tail.
"""

from __future__ import annotations

import logging

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models import Brand, ProductVariant
from app.services import urls
from app.services.normalize import flat_key

logger = logging.getLogger(__name__)


def spelling_key(text: str | None) -> str:
    """How two spellings of a listed brand are compared for a claim: case, accents and
    punctuation folded, nothing removed. `flat_key`, never `brand_key`: the whole point of a
    split is that the fold key is the thing being disagreed with."""
    return flat_key(text)


def claim_slug(db: Session, name: str) -> str:
    """The address the new brand row answers at, from the name a person chose: the name's words,
    free of every brand slug already taken. A brand's slug is normally the fold key ingest finds
    the row by (`ingest.brand_slug`); a split row is reached by its claim instead, so its slug is
    free to say what the row is."""
    wanted = urls.slugify(name)[:160].strip("-") or "brand"
    taken = set(db.scalars(select(Brand.slug).where(Brand.slug.like(f"{wanted}%"))))
    slug, n = wanted, 2
    while slug in taken:
        slug, n = f"{wanted}-{n}", n + 1
    return slug


def spellings_on(db: Session, brand: Brand) -> dict[str, int]:
    """Every listed brand spelling the row's live variants carry, with how many carry it."""
    counts: dict[str, int] = {}
    for text, in db.execute(select(ProductVariant.brand)
                            .where(ProductVariant.brand_id == brand.id, ProductVariant.merged_into_id.is_(None))):
        cleaned = (text or "").strip()
        if cleaned:
            counts[cleaned] = counts.get(cleaned, 0) + 1
    return counts


def split_brand(db: Session, brand: Brand, spellings: list[str], *, new_name: str, decided_by, batch=None,
                commit: bool = True) -> dict:
    """Move the named spellings of `brand` onto a brand row of their own, as one decision.

    Refuses (all through `writer.Refused`, so a route and the CLI report the same code):
    `VALUE_INVALID` for no spellings or no name; `SPELLING_NOT_ON_ROW` for a spelling no live
    variant of the row carries; `SPLIT_EMPTIES_SOURCE` when the named spellings are all of them
    (that is a rename, not a split); `ENTITY_ALIASED` for an alias row, from the writer itself;
    and `COLLECTION_RUNNING` while a collection runs, like every other batch route.

    With no `batch` it opens its own and commits it (`commit=False` to leave that to a caller
    that owns the transaction).
    """
    from app.services.decisions import writer

    name = (new_name or "").strip()
    wanted = [s.strip() for s in (spellings or []) if s and s.strip()]
    if not wanted:
        raise writer.Refused("VALUE_INVALID", "A split names at least one listed spelling to move.")
    if not name:
        raise writer.Refused("VALUE_INVALID", "A split names the brand the spellings move to.")
    if brand.alias_of_id is not None:
        raise writer.Refused("ENTITY_ALIASED", "This brand is an alias; split the one it points at.")
    present = spellings_on(db, brand)
    by_key = {spelling_key(k): k for k in present}
    missing = [s for s in wanted if spelling_key(s) not in by_key]
    if missing:
        raise writer.Refused("SPELLING_NOT_ON_ROW",
                             f"{brand.name} has no live product variant listed as {', '.join(repr(m) for m in missing)}; "
                             f"it holds {', '.join(repr(s) for s in sorted(present))}.")
    claimed = {spelling_key(s) for s in wanted}
    if not any(spelling_key(s) not in claimed for s in present):
        raise writer.Refused("SPLIT_EMPTIES_SOURCE",
                             f"Those are every spelling on {brand.name}; nothing would be left on it. "
                             "A row that should simply be called something else is renamed, not split.")
    value = {"brand_slug": claim_slug(db, name), "name": name[:160],
             "spellings": sorted({by_key[spelling_key(s)] for s in wanted})}

    def run(b):
        return writer.record(b, "brand", brand, "split", value,
                             reason=f"{', '.join(value['spellings'])} is not {brand.name}")

    if batch is not None:
        row = run(batch)
        return _report(batch, row)
    writer.refuse_if_collecting(db)
    with writer.batch(db, "route", "individual", decided_by, commit=commit) as b:
        row = run(b)
    logger.info("brand_split from=%s to=%s spellings=%s by=%s", brand.slug, value["brand_slug"], value["spellings"], decided_by)
    return {**_report(b, row), "groups_merged": b.folded.get("groups", 0), "rows_merged": b.folded.get("merged_rows", 0)}


def _report(b, row) -> dict:
    """What the caller is told: the applier reports through `b.notes`, never on the row, because
    the ledger is append-only and a row's detail is complete at insert."""
    minted = (row.detail or {}).get("new_brand") or {}
    notes = b.notes.get(row.id, {})
    return {"decision_id": row.id, "decision_uid": str(row.uid), "brand_slug": minted.get("slug"),
            "brand_uid": minted.get("uid"), "minted": bool(minted.get("minted")),
            "product_variants_moved": notes.get("product_variants_moved", 0),
            "lines_created": notes.get("lines_created", 0), "lines_pruned": notes.get("lines_pruned", 0),
            "line_decisions_carried": notes.get("line_decisions_carried", 0), "groups_merged": 0, "rows_merged": 0}


# --------------------------------------------------------------------------- the brand's own card

#: The three states a brand's page can be in, as a person reads them. `hidden` and `indexed` are
#: two flags; this is the one word that names the pair, so a page is never described twice.
STATUS_UNLISTED = "unlisted"
STATUS_NOINDEX = "noindex"
STATUS_INDEXABLE = "indexable"


def status_of(brand: Brand) -> str:
    if brand.hidden:
        return STATUS_UNLISTED
    return STATUS_INDEXABLE if brand.indexed else STATUS_NOINDEX


def card(db: Session, slug: str) -> dict:
    """What a person needs to know about a brand before reading its proposals: the name it shows
    under, every spelling the shops actually sent with how many listings and shops wrote each, its
    address and whether that page is unlisted, noindex or indexable.

    Rian asked for exactly this twice before it was built: *"when I click the lancome brand review,
    I should see some details about lancome including the variety of spellings and the one we're
    going with as a display name for the brand with the ability for me to manually set that."* None
    of it was stored anywhere new; the spellings are a read of the listed layer and the status is
    the two flags the ledger already decides. Writes nothing.
    """
    from sqlalchemy import func

    from app.models import Listing, Shop

    brand = db.scalar(select(Brand).where(Brand.slug == slug))
    if brand is None:
        return {}
    spellings = [
        {"spelling": text or "", "listings": int(n), "shops": int(shops)}
        for text, n, shops in db.execute(
            select(Listing.listed_brand, func.count(Listing.id), func.count(func.distinct(Shop.id)))
            .join(ProductVariant, ProductVariant.id == Listing.variant_id)
            .join(Shop, Shop.id == Listing.shop_id)
            .where(ProductVariant.brand_id == brand.id, ProductVariant.merged_into_id.is_(None),
                   Listing.ignored_at.is_(None))
            .group_by(Listing.listed_brand).order_by(func.count(Listing.id).desc())
        )
    ]
    variants = db.scalar(select(func.count(ProductVariant.id)).where(
        ProductVariant.brand_id == brand.id, ProductVariant.merged_into_id.is_(None))) or 0
    return {
        # `brand_page_path`, not `brand_path`: the latter takes a NAME and falls back to a
        # catalogue search when it is given no slug, which made every brand's card read
        # "/products?q=lancome" when the page is "/brands/lancome". Every brand page exists since
        # K6; whether it answers or redirects is the status below, not the address.
        "slug": brand.slug, "name": brand.name, "path": urls.brand_page_path(brand.slug),
        "status": status_of(brand), "variants": int(variants),
        # A blank spelling is a shop that sent no brand field at all; it is a fact about the data
        # and is shown as one rather than dropped.
        "spellings": spellings,
        "named_spellings": sum(1 for s in spellings if s["spelling"]),
    }


def set_card(db: Session, slug: str, *, name: str | None, status: str | None, by) -> dict:
    """A person's edit of the two things on the card they own: the display name and the page
    status. One batch, so one undo, and each field a decision in the ledger like any other. A
    machine never writes either of these here (REVIEW-PROCESS.md section 2.3: a pass proposes a
    name, it does not set one over a person's)."""
    from app.services.decisions import writer

    brand = db.scalar(select(Brand).where(Brand.slug == slug))
    if brand is None:
        raise writer.Refused("BRAND_MISSING", f"No brand {slug!r}.")
    if status is not None and status not in (STATUS_UNLISTED, STATUS_NOINDEX, STATUS_INDEXABLE):
        raise writer.Refused("VALUE_INVALID", f"A page status is unlisted, noindex or indexable, not {status!r}.")
    if name is not None and not name.strip():
        raise writer.Refused("VALUE_INVALID", "A display name cannot be blank.")
    changed: list[str] = []
    with writer.batch(db, "route", "individual", by, scope={"brand_slug": slug, "what": "brand-card"}) as b:
        if name is not None and name.strip() != brand.name:
            writer.record(b, "brand", brand, "name", name.strip(), reason="set on the brand's card")
            changed.append("name")
        if status is not None and status != status_of(brand):
            writer.record(b, "brand", brand, "hidden", status == STATUS_UNLISTED, reason="set on the brand's card")
            writer.record(b, "brand", brand, "indexed", status == STATUS_INDEXABLE, reason="set on the brand's card")
            changed.append("status")
        batch_uid = str(b.row.uid)
    db.refresh(brand)
    return {**card(db, slug), "changed": changed, "batch_uid": batch_uid if changed else None}