"""Publish (Stream K6; plan W18): the one reader of a page's `indexed` and `hidden`, the
forwarding address of a retired slug, the slug a person's chosen name fixes at confirm, and the
candidate rules that SUGGEST a page for indexing. A page here is a brand row, a product line row
or a place row. Three facts and a fourth:

- reachable: every page answers from the moment its row exists; no floor decides existence.
- quality: an approved review sheet (K4); read here only for the desk row.
- indexed: a person's decision, never a rule's. A rule only suggests (`candidates`); falling
  below the rule suggests removal and never removes. Until then the head says `noindex, follow`.
- hidden: a page that is wrong rather than thin. It answers 302 to the nearest right page,
  never 404, and a hidden page is never indexed whatever its column says.

`indexed` and `hidden` are the columns the ledger's appliers materialise from the newest
decision (`decisions/appliers._flag`; `app.cli decisions verify` reports drift), so reading the
column IS reading the newest decision, without a query per page.

A brand's slug is not renamed at confirm: it is the fold key ingest finds the row by and the
ledger's natural key (`brand:<slug>`), so a renamed brand row would be minted again at the next
collection. Its alias row is the forwarding address (plan W4) and `redirect_for` follows it.
A product line is found by its key, never its slug, so its slug is free to follow the chosen name.

Sources of truth: this module, `models/decisions.py` (Redirect), `models/catalog.py` and
`models/places.py` (hidden, indexed), `catalog_queries._brand_counts` (the 9 Sep floor),
`places.unit_count` (the comparison unit), the K6 brief. What it cost before: the brand floor
decided existence, so a quarter of the visible product pages would have linked a 404.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from datetime import UTC, datetime

from sqlalchemy import distinct, func, select
from sqlalchemy.orm import Session, object_session

from app.models import Brand, Listing, PriceObservation, ProductLine, ProductVariant, Redirect, Shop, Suggestion
from app.models.places import Place, ShopPlace

logger = logging.getLogger(__name__)

#: The family landings and category pages a person has approved for indexing, by address.
#: Empty means every one of them is `noindex, follow`: reachable, walked through, out of the
#: index until someone has read the page. They are the one page kind with no row of its own
#: (a category is a taxonomy constant, not a table), so the approval cannot live in the
#: decisions ledger the way a brand's or a place's does; it is declared here and changed by
#: the same review any other line of the site's copy gets. Move it into the ledger on the day
#: categories become rows.
INDEXED_PAGES: frozenset[str] = frozenset()


def is_indexed_path(path: str) -> bool:
    """Has a person approved this address for the index? False for everything by default."""
    return path in INDEXED_PAGES


#: The kinds of page, as the `redirects` table and the ledger's entity types spell them.
KINDS = ("brand", "product_line", "place")
_MODELS = {"brand": Brand, "product_line": ProductLine, "place": Place}
#: `suggestions.level` of an index suggestion (String(16)); `left_id` is the page's row id.
LEVEL_OF = {"brand": "index:brand", "product_line": "index:line", "place": "index:place"}
KIND_OF_LEVEL = {level: kind for kind, level in LEVEL_OF.items()}
REASON_ADD = "index:candidate"
REASON_REMOVE = "index:below_rule"

#: A product line is a candidate when one of its variants is priced at this many comparison units.
LINE_MIN_UNITS = 2
#: An airport or place is a candidate at this many published variants: the category-at-airport
#: bar (`coverage`), proposed by K6 for rian to confirm on the running list.
PLACE_MIN_VARIANTS = 15
_CHAIN_LIMIT = 12


# --------------------------------------------------------------------------- the facts

def kind_of(page) -> str:
    for kind, model in _MODELS.items():
        if isinstance(page, model):
            return kind
    raise TypeError(f"{type(page).__name__} is not a page")


def is_hidden(page) -> bool:
    return bool(getattr(page, "hidden", False))


def is_indexed(page) -> bool:
    """A person approved indexing it (the newest `indexed` decision, materialised), and it is
    neither hidden nor a forwarding row: those answer a redirect, which no engine should hold."""
    if page is None or is_hidden(page) or getattr(page, "alias_of_id", None) is not None:
        return False
    return bool(getattr(page, "indexed", False))


def page_path(page) -> str:
    from app.services import places, urls

    if isinstance(page, Brand):
        return urls.brand_page_path(page.slug)
    if isinstance(page, ProductLine):
        return urls.line_path(page.slug)
    kind = places.KINDS.get(page.kind)
    return f"{kind.address_prefix if kind else '/places/'}{page.slug}"


def hidden_forward(page) -> str:
    """Where a hidden page sends its visitor (302, never 404): a product line to its brand's
    page, a brand to its products in the catalogue, a place to the index of its kind."""
    from app.services import places, urls

    if isinstance(page, ProductLine):
        db = object_session(page)
        brand = _canonical(db, db.get(Brand, page.brand_id)) if db is not None else None
        if brand is not None and not is_hidden(brand):
            return urls.brand_page_path(brand.slug)
        return "/products"
    if isinstance(page, Brand):
        return f"/products?brand={urls.urlquote(page.name)}"
    kind = places.KINDS.get(page.kind)
    return (kind.address_prefix if kind else "/places/").rstrip("/") or "/"


def hidden_target(db: Session, kind: str, ref: str) -> str | None:
    """The page routes' one question: is what this address names hidden, and if so where does
    its visitor go? `ref` is a slug (an old one too), or an airport's IATA code for a place."""
    page = place_by_identifier(db, "iata", ref) if kind == "place" else find(db, kind, ref)
    return hidden_forward(page) if page is not None and is_hidden(page) else None


def _canonical(db: Session | None, row):
    hops = 0
    while row is not None and db is not None and getattr(row, "alias_of_id", None) and row.alias_of_id != row.id and hops < _CHAIN_LIMIT:
        row, hops = db.get(type(row), row.alias_of_id), hops + 1
    return row


def find(db: Session, kind: str, ref: str | int):
    """The live page a slug, an old slug or an id names: forwarding rows and redirects followed."""
    model = _MODELS[kind]
    row = db.get(model, int(ref)) if isinstance(ref, int) or str(ref).isdigit() else db.scalar(select(model).where(model.slug == str(ref)))
    if row is None:
        forward = redirect_for(str(ref), kind, db)
        row = db.scalar(select(model).where(model.slug == forward)) if forward else None
    return _canonical(db, row)


def place_by_identifier(db: Session, scheme: str, value: str) -> Place | None:
    """The place an identifier names (`identifiers` is an open list, so it is read in Python:
    a registry of tens of rows, cached on the session like `blocked_shop_ids`)."""
    info = getattr(db, "info", None)
    cache = info.setdefault("places_by_identifier", {}) if isinstance(info, dict) else {}
    if not cache:
        for place in db.scalars(select(Place)):
            for ident in place.identifiers or []:
                cache[(str(ident.get("scheme", "")).lower(), str(ident.get("value", "")).upper())] = place.id
    place_id = cache.get((scheme.lower(), (value or "").upper()))
    return db.get(Place, place_id) if place_id is not None else None


# --------------------------------------------------------------------------- redirects

def redirect_for(slug: str, kind: str, db: Session | None = None) -> str | None:
    """The live slug a retired one forwards to, or None: the `redirects` row (flattened at
    write, so one read), else the forwarding row that still holds the slug (a brand alias, a
    product line alias), followed to its end."""
    if db is None:
        return None
    row = db.scalar(select(Redirect).where(Redirect.kind == kind, Redirect.from_slug == slug))
    if row is not None:
        return row.to_slug
    model = _MODELS.get(kind)
    if model is None or not hasattr(model, "alias_of_id"):
        return None
    holder = db.scalar(select(model).where(model.slug == slug))
    target = _canonical(db, holder)
    return target.slug if holder is not None and target is not None and target.id != holder.id else None


def redirect_write(db: Session, kind: str, from_slug: str, to_slug: str, *, decision_id: int | None = None) -> Redirect | None:
    """A forwarding address, chains flattened at write (A to B then B to C stores A to C and
    B to C). A slug that becomes live again stops forwarding (A to B then back to A leaves only
    B to A), so no row ever points at itself. The table's key is the slug alone: a slug another
    kind already forwards is left to it and logged, never overwritten."""
    if kind not in KINDS:
        raise ValueError(f"no redirects of kind {kind!r}")
    if from_slug == to_slug:
        return None
    for older in db.scalars(select(Redirect).where(Redirect.kind == kind, Redirect.to_slug == from_slug)):
        older.to_slug = to_slug
    db.flush()
    for live_again in db.scalars(select(Redirect).where(Redirect.kind == kind, Redirect.from_slug == to_slug)):
        db.delete(live_again)
    db.flush()
    row = db.get(Redirect, from_slug)
    if row is not None and row.kind != kind:
        logger.warning("redirect_kept from=%s kind=%s wanted_kind=%s", from_slug, row.kind, kind)
        return None
    if row is None:
        row = Redirect(from_slug=from_slug, kind=kind, to_slug=to_slug, decision_id=decision_id)
        db.add(row)
    else:
        row.to_slug, row.decision_id, row.since = to_slug, decision_id, datetime.now(UTC)
    db.flush()
    return row


def final_slug(db: Session, line: ProductLine, chosen_name: str) -> str:
    """The address a product line's chosen name gives it: the brand's slug, then the name's
    words (the brand not said twice), free of every other line's slug and forwarding address."""
    from app.services import urls

    brand = _canonical(db, db.get(Brand, line.brand_id))
    brand_slug = brand.slug if brand is not None else ""
    words = urls.slugify(chosen_name) if chosen_name and chosen_name.strip() else ""
    if brand_slug and (words == brand_slug or words.startswith(brand_slug + "-")):
        wanted = words
    else:
        wanted = "-".join(p for p in (brand_slug, words) if p)
    wanted = wanted[:230].strip("-")
    if not wanted:
        return line.slug
    taken = set(db.scalars(select(ProductLine.slug).where(ProductLine.slug.like(f"{wanted}%"), ProductLine.id != line.id)))
    for row in db.scalars(select(Redirect).where(Redirect.kind == "product_line", Redirect.from_slug.like(f"{wanted}%"))):
        if row.to_slug != line.slug:  # its own old address is its to take back
            taken.add(row.from_slug)
    slug, n = wanted, 2
    while slug in taken:
        slug, n = f"{wanted}-{n}", n + 1
    return slug


def slug_at_confirm(row, chosen_name: str, *, decision=None, rename_indexed: bool = False) -> str | None:
    """The slug hook (K3 calls it where an alias is recorded): fix the page's final slug from the
    chosen name and forward the old one. Returns the new slug, or None when nothing moved.

    A brand keeps its slug (the module docstring says why); its alias row forwards. An indexed
    page is not renamed unless the decision says so (`rename_indexed`, or `rename_indexed` in the
    decision's detail): an engine holds that address, and a rename costs it a crawl cycle."""
    db = object_session(row)
    if db is None or not isinstance(row, ProductLine) or row.alias_of_id is not None:
        return None
    allowed = rename_indexed or bool((getattr(decision, "detail", None) or {}).get("rename_indexed"))
    wanted = final_slug(db, row, chosen_name)
    if wanted == row.slug:
        return None
    if row.indexed and not allowed:
        logger.info("slug_kept line=%s wanted=%s reason=indexed", row.slug, wanted)
        return None
    old, row.slug = row.slug, wanted
    db.flush()
    redirect_write(db, "product_line", old, wanted, decision_id=getattr(decision, "id", None))
    logger.info("slug_fixed line=%s was=%s", wanted, old)
    return wanted


# --------------------------------------------------------------------------- the candidate rules

@dataclass
class Candidate:
    kind: str
    id: int
    slug: str
    name: str
    path: str
    why: str
    figures: dict = field(default_factory=dict)


def _priced(stmt, db: Session):
    """The joins every rule counts over: a variant's priced listings at the shops the site shows."""
    from app.services import catalog_queries as cq, places

    return (
        stmt.select_from(ProductVariant).join(Listing, Listing.variant_id == ProductVariant.id)
        .join(PriceObservation, PriceObservation.listing_id == Listing.id)
        .join(Shop, Shop.id == Listing.shop_id).outerjoin(ShopPlace, places.primary_on())
        .where(cq.publishable(db), cq.shown_category(), Shop.is_catalogue_only.is_(False))
    )


def brand_candidates(db: Session) -> list[Candidate]:
    """The 9 Sep floor (`catalog_queries._brand_counts`): three published variants priced across
    two comparison units. Once the rule of existence; now only of candidacy."""
    from app.services import catalog_queries as cq

    counted = {brand_id: n for brand_id, n, _ in cq._brand_counts(db)}
    rows = db.scalars(select(Brand).where(Brand.id.in_(list(counted) or [0]), Brand.alias_of_id.is_(None), Brand.hidden.is_(False)))
    return [Candidate("brand", b.id, b.slug, b.name, page_path(b),
                      f"{counted[b.id]} variants priced across {cq.BRAND_PAGE_MIN_AIRPORTS} or more places",
                      {"variants": counted[b.id]}) for b in rows]


def line_candidates(db: Session) -> list[Candidate]:
    """A product line one of whose variants is priced at two or more comparison units: the page
    has a comparison to show. A single-shop line is helpful and reachable, not a candidate."""
    from app.services import places

    per_variant = _priced(
        select(ProductVariant.product_line_id.label("product_line_id"), ProductVariant.id.label("variant_id"),
               places.unit_count().label("units")), db,
    ).where(ProductVariant.product_line_id.isnot(None)).group_by(ProductVariant.product_line_id, ProductVariant.id) \
        .having(places.unit_count() >= LINE_MIN_UNITS).subquery()
    live = func.coalesce(ProductLine.alias_of_id, ProductLine.id)
    found = {product_line_id: (n, best) for product_line_id, n, best in db.execute(
        select(live, func.count(distinct(per_variant.c.variant_id)), func.max(per_variant.c.units))
        .join(per_variant, per_variant.c.product_line_id == ProductLine.id).group_by(live))}
    rows = db.scalars(select(ProductLine).where(ProductLine.id.in_(list(found) or [0]), ProductLine.alias_of_id.is_(None),
                                                ProductLine.hidden.is_(False)))
    return [Candidate("product_line", r.id, r.slug, r.name, page_path(r),
                      f"{found[r.id][0]} variant(s) compared, the widest across {found[r.id][1]} places",
                      {"compared_variants": found[r.id][0], "widest": found[r.id][1]}) for r in rows]


def place_candidates(db: Session) -> list[Candidate]:
    from app.services import places

    counted = dict(db.execute(_priced(select(ShopPlace.place_id, func.count(distinct(ProductVariant.id))), db)
                              .where(ShopPlace.place_id.isnot(None)).group_by(ShopPlace.place_id)
                              .having(func.count(distinct(ProductVariant.id)) >= PLACE_MIN_VARIANTS)).all())
    rows = db.scalars(select(Place).where(Place.id.in_(list(counted) or [0]), Place.hidden.is_(False)))
    return [Candidate("place", p.id, p.slug, p.name, page_path(p), f"{counted[p.id]} variants priced here",
                      {"variants": counted[p.id]})
            for p in rows if (places.KINDS.get(p.kind) is None or places.KINDS[p.kind].pages)]


_RULES = {"brand": brand_candidates, "product_line": line_candidates, "place": place_candidates}


def candidates(db: Session, kinds: tuple[str, ...] = KINDS) -> list[Candidate]:
    """Every page that meets its rule, indexed or not, in a stable order."""
    out: list[Candidate] = []
    for kind in kinds:
        out.extend(sorted(_RULES[kind](db), key=lambda c: c.slug))
    return out


# --------------------------------------------------------------------------- suggestions to index

def suggest(db: Session, *, check: bool = False) -> dict[str, int]:
    """Write the candidates as suggestions (`suggestions.level` `index:<kind>`): a page over its
    rule and not indexed is suggested for indexing; an indexed page under its rule is suggested
    for removal, never removed. A suggestion a person answered never resurfaces; an open one the
    rule no longer makes is withdrawn. Safe to repeat: the second run inserts none."""
    counts = {"suggested": 0, "suggested_removal": 0, "refreshed": 0, "withdrawn": 0, "kept": 0}
    for kind in KINDS:
        model, level = _MODELS[kind], LEVEL_OF[kind]
        meeting = {c.id: c for c in _RULES[kind](db)}
        indexed = {r.id: r for r in db.scalars(select(model).where(model.indexed.is_(True), model.hidden.is_(False)))}
        wanted: dict[int, tuple[str, dict]] = {}
        for c in meeting.values():
            if c.id not in indexed:
                wanted[c.id] = (REASON_ADD, {"kind": kind, "slug": c.slug, "name": c.name, "path": c.path, "why": c.why, **c.figures})
        for r in indexed.values():
            if r.id not in meeting and getattr(r, "alias_of_id", None) is None:
                wanted[r.id] = (REASON_REMOVE, {"kind": kind, "slug": r.slug, "name": r.name, "path": page_path(r),
                                                "why": "indexed, and now under its rule; remove only if the page is thin"})
        existing: dict[int, list[Suggestion]] = {}
        for s in db.scalars(select(Suggestion).where(Suggestion.level == level)):
            existing.setdefault(s.left_id, []).append(s)
        for page_id, (reason, detail) in wanted.items():
            rows = [s for s in existing.get(page_id, []) if s.reason == reason]
            answers = [s for s in rows if s.decision is not None]
            row = next((s for s in rows if s.decision is None), None)
            stands = any(s.decision == "dismissed" for s in answers) or any(not _answer_lapsed(db, kind, page_id, s) for s in answers)
            if answers and row is None and stands:
                counts["kept"] += 1
                continue
            if row is None:
                counts["suggested" if reason == REASON_ADD else "suggested_removal"] += 1
                if not check:
                    db.add(Suggestion(level=level, left_id=page_id, right_id=None, reason=reason, score=None, detail=detail))
            elif row.closed_reason is not None or (row.detail or {}) != detail:
                counts["refreshed"] += 1
                if not check:
                    row.closed_reason, row.detail = None, detail
        for page_id, rows in existing.items():
            for s in rows:
                if s.decision is None and s.closed_reason is None and (page_id not in wanted or wanted[page_id][0] != s.reason):
                    counts["withdrawn"] += 1
                    if not check:
                        s.closed_reason = "withdrawn"
    if not check:
        db.flush()
    return counts


def _answer_lapsed(db: Session, kind: str, page_id: int, answered: Suggestion) -> bool:
    """An approved suggestion stands while the page is as it left it. A dismissal always stands,
    and a person removing a page from the index IS a dismissal of indexing it (`_close`): on the
    staging copy a brand removed by hand was offered again by the very next `index suggest`."""
    if answered.decision != "approved":
        return False
    page = db.get(_MODELS[kind], page_id)
    want_indexed = answered.reason == REASON_ADD
    return page is not None and bool(page.indexed) != want_indexed


def open_suggestions(db: Session) -> list[Suggestion]:
    return list(db.scalars(select(Suggestion).where(Suggestion.level.in_(list(KIND_OF_LEVEL)), Suggestion.is_open())
                           .order_by(Suggestion.level, Suggestion.id)))


# --------------------------------------------------------------------------- the decision to index

class PageMissing(LookupError):
    """No live page answers the reference a person gave."""


def facts(db: Session, page) -> dict:
    """The three facts the desk row shows beside a page (plan W18), and the fourth.
    `quality` is K4's fact, read here and decided nowhere: a product line a person named on an
    approved sheet, a brand with an approved row on any sheet; a place has no sheet (None)."""
    from app.models import Proposal

    kind = kind_of(page)
    quality: bool | None = None
    if kind == "product_line":
        quality = str(page.key or "").startswith("decided:")
    elif kind == "brand":
        slugs = [page.slug, *db.scalars(select(Brand.slug).where(Brand.alias_of_id == page.id))]
        quality = db.scalar(select(func.count()).select_from(Proposal).where(Proposal.brand_slug.in_(slugs), Proposal.status == "approved")) > 0
    return {"reachable": not is_hidden(page), "quality": quality, "indexed": is_indexed(page), "hidden": is_hidden(page)}


def set_indexed(db: Session, pages: list[tuple[str, str | int]], indexed: bool, by, *, batch_kind: str = "cli",
                reason: str | None = None) -> dict:
    """A person's decision to index pages, or to remove them from the index: one batch (one
    undo), a `decisions` row per page on its `indexed` field, the column following through the
    ledger's applier. More than one page is a bulk batch, so it never overrules a decision a
    person made on a single page (`DECISION_HELD`, reported, never forced). A hidden page is
    refused: it answers a redirect. Open suggestions the decision answers are closed with it.
    Returns the counts and the paths newly indexed (the caller pings them once, after commit)."""
    from app.services.decisions import writer

    mode = "bulk" if len(pages) > 1 else "individual"
    done: list[dict] = []
    refused: list[dict] = []
    paths: list[str] = []
    work: list[tuple[str, str, object]] = []
    seen: set[tuple[str, int]] = set()
    for kind, ref in pages:
        page = find(db, kind, ref) if kind in _MODELS else None
        label = f"{kind}:{ref}"
        if page is not None and (kind, page.id) in seen:  # named twice (an old slug and the live one): one decision
            done.append({"page": label, "path": page_path(page), "changed": False})
            continue
        if page is not None:
            seen.add((kind, page.id))
        if page is None:
            refused.append({"page": label, "error_code": "PAGE_NOT_FOUND", "summary": f"No {kind.replace('_', ' ')} answers {ref!r}."})
        elif indexed and is_hidden(page):
            refused.append({"page": label, "error_code": "PAGE_HIDDEN", "summary": f"{page.name} is hidden; a hidden page is never indexed."})
        elif bool(page.indexed) == indexed:
            done.append({"page": label, "path": page_path(page), "changed": False})
            _close(db, kind, page.id, indexed, None)
        else:
            work.append((kind, label, page))
    uid = None
    if work:  # a batch is opened only for something to record: an empty one is noise on the ledger
        # `tail=False`: an index decision touches no identity, so there is nothing to rekey or fold.
        with writer.batch(db, batch_kind, mode, by, scope={"index": "approve" if indexed else "remove", "pages": len(work)},
                          commit=False, tail=False) as b:
            for kind, label, page in work:
                try:
                    row = writer.record(b, kind, page, "indexed", indexed,
                                        reason=reason or ("approved for indexing" if indexed else "removed from the index"))
                except writer.Refused as exc:
                    refused.append({"page": label, "error_code": exc.code, "summary": exc.summary})
                    continue
                _close(db, kind, page.id, indexed, row)
                done.append({"page": label, "path": page_path(page), "changed": True})
                if indexed:
                    paths.append(page_path(page))
        uid = str(b.uid)
    db.flush()
    changed = sum(1 for d in done if d["changed"])
    verb = "approved for indexing" if indexed else "removed from the index"
    return {"batch_uid": uid if changed else None, "mode": mode, "pages": done, "refusals": refused, "ping": paths,
            "counts": {"changed": changed, "unchanged": len(done) - changed, "refused": len(refused)},
            "message": f"{changed} {verb}, {len(done) - changed} already so, {len(refused)} refused"}


def _close(db: Session, kind: str, page_id: int, indexed: bool, row) -> None:
    """Close what the decision answers. Removing a page also answers "no" to indexing it, so the
    rule does not offer it again the next morning."""
    agrees, opposite = (REASON_ADD, REASON_REMOVE) if indexed else (REASON_REMOVE, REASON_ADD)
    now = datetime.now(UTC)
    rows = list(db.scalars(select(Suggestion).where(Suggestion.level == LEVEL_OF[kind], Suggestion.left_id == page_id)))
    for s in rows:
        if s.decision is None and s.closed_reason is None:
            s.decision, s.decision_id, s.decided_at = ("approved" if s.reason == agrees else "dismissed"), getattr(row, "id", None), now
    # Only a removal leaves a standing "no": an approval must not silence the later suggestion to
    # remove the page when it falls under its rule (that suggestion is the whole safeguard).
    if row is not None and not indexed and not any(s.reason == opposite and s.decision == "dismissed" for s in rows):
        db.add(Suggestion(level=LEVEL_OF[kind], left_id=page_id, right_id=None, reason=opposite, score=None, decision="dismissed",
                          decision_id=row.id, decided_at=now, detail={"kind": kind, "why": "answered by a person's own decision on the page"}))


def dismiss(db: Session, suggestion_ids: list[int], by) -> int:
    """A person read the suggestion and said no: it never resurfaces, and nothing about the
    page changes (a page stays as it is until a person decides otherwise)."""
    from app.services.decisions.writer import _username

    by_id, _ = _username(db, by)
    n = 0
    for s in db.scalars(select(Suggestion).where(Suggestion.id.in_(suggestion_ids or [0]), Suggestion.level.in_(list(KIND_OF_LEVEL)), Suggestion.is_open())):
        s.decision, s.decided_by, s.decided_at = "dismissed", by_id, datetime.now(UTC)
        n += 1
    db.flush()
    return n


def desk(db: Session, kind: str | None = None, limit: int = 200, offset: int = 0, brand: str | None = None,
         q: str | None = None) -> dict:
    """The index candidates a person reads on `/review`: the open suggestions, each with its
    page's three facts and why the rule offers it; with `brand`, that brand's page and its
    product lines only (the panel under the brand's sheet). Writes nothing.

    `q` matches a page's name or slug, and `shown` is how many the filter leaves, so the tab can
    page through 1,701 product lines instead of showing the first hundred and asking a person to
    decide them before seeing the rest (K10.4)."""
    from sqlalchemy import and_, or_

    levels = [LEVEL_OF[kind]] if kind in LEVEL_OF else list(KIND_OF_LEVEL)
    scope = [Suggestion.level.in_(list(KIND_OF_LEVEL)), Suggestion.is_open()]
    if brand:
        row = find(db, "brand", brand)
        family = [row.id, *db.scalars(select(Brand.id).where(Brand.alias_of_id == row.id))] if row is not None else [0]
        scope.append(or_(and_(Suggestion.level == LEVEL_OF["brand"], Suggestion.left_id.in_(family)),
                         and_(Suggestion.level == LEVEL_OF["product_line"],
                              Suggestion.left_id.in_(select(ProductLine.id).where(ProductLine.brand_id.in_(family))))))
    where = [*scope, Suggestion.level.in_(levels)]
    totals = {KIND_OF_LEVEL[level]: int(n) for level, n in db.execute(
        select(Suggestion.level, func.count(Suggestion.id)).where(*scope).group_by(Suggestion.level))}
    text = " ".join((q or "").split()).lower()
    if text:
        # The name a person reads and the address they would land on, one clause per kind because
        # the three page tables are different tables reached through the suggestion's left id.
        wanted = []
        for level, page_kind in KIND_OF_LEVEL.items():
            model = _MODELS[page_kind]
            ids = select(model.id).where(func.lower(model.name).contains(text) | func.lower(model.slug).contains(text))
            wanted.append(and_(Suggestion.level == level, Suggestion.left_id.in_(ids)))
        where.append(or_(*wanted))
    shown = db.scalar(select(func.count(Suggestion.id)).where(*where)) or 0
    rows = []
    for s in db.scalars(select(Suggestion).where(*where).order_by(Suggestion.level, Suggestion.reason.desc(), Suggestion.id).limit(limit).offset(offset)):
        page_kind = KIND_OF_LEVEL[s.level]
        page = db.get(_MODELS[page_kind], s.left_id)
        if page is None:
            continue
        rows.append({"suggestion_id": s.id, "kind": page_kind, "action": "remove" if s.reason == REASON_REMOVE else "approve",
                     "slug": page.slug, "name": page.name, "path": page_path(page), "why": (s.detail or {}).get("why") or "",
                     **facts(db, page)})
    return {"totals": {k: totals.get(k, 0) for k in KINDS}, "shown": int(shown), "offset": offset, "limit": limit, "rows": rows}