"""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}


# --------------------------------------------------------------------------- the page's status

#: The four words a signed-in admin is shown for the address they are standing on, and the
#: fifth that is not a page at all (`docs/SEO.md` "Page status: what a signed-in admin sees").
#: They are an app word each, not a crawler's: `robots` beside them is what the head sends.
STATUSES = ("indexable", "noindex", "unlisted", "hidden", "missing")
ROBOTS_FOLLOW = "noindex, follow"
ROBOTS_NOFOLLOW = "noindex, nofollow"


@dataclass(frozen=True)
class PageStatus:
    """What one address is, read from the rows and the declared tables alone.

    `robots` is the directive the head would send, or None when it sends none; `forwards_to` is
    where a `hidden` address sends its visitor; `canonical` is the address the page is really at.
    """

    path: str
    status: str
    reason: str
    robots: str | None
    in_sitemap: bool
    forwards_to: str | None = None
    canonical: str | None = None


def _indexed_status(path: str, page, canonical: str) -> PageStatus:
    """A page row that answers at its own address: indexed by a person, or not yet."""
    if is_indexed(page):
        return PageStatus(path, "indexable", "a person approved this page for the index", None, True,
                          canonical=canonical)
    return PageStatus(path, "noindex", "generated, and no person has approved it for the index yet",
                      ROBOTS_FOLLOW, False, canonical=canonical)


def _forwards(path: str, target: str, reason: str, canonical: str | None = None) -> PageStatus:
    """A redirect answers here, so the page is at `canonical` (the bare address; `target` may
    carry the query the redirect keeps, and a canonical never does)."""
    return PageStatus(path, "hidden", reason, None, False, forwards_to=target,
                      canonical=canonical or target.split("?")[0])


def _missing(path: str, reason: str) -> PageStatus:
    """A real 404: not a page, so not one of the four words."""
    return PageStatus(path, "missing", reason, ROBOTS_NOFOLLOW, False)


def _row_status(db: Session, kind: str, ref: str, path: str) -> PageStatus:
    page = find(db, kind, ref)
    if page is None:
        return _missing(path, f"no {kind.replace('_', ' ')} answers this address")
    if is_hidden(page):
        return _forwards(path, hidden_forward(page), "hidden, so the address answers a 302 to the nearest right page")
    canonical = page_path(page)
    if canonical != path:
        return _forwards(path, canonical, "an alias or a retired address; it answers a 301 to the live one")
    return _indexed_status(path, page, canonical)


def _airport_canonical(db: Session, iata: str) -> str:
    """The address an airport page really sits at: the declared word when there is one, else the
    `shops` row's own (`urls.airport_path`, as `catalog_queries._airport_summary` builds it)."""
    from app.services import urls

    if iata in urls.AIRPORT_SLUGS:
        return urls.airport_path(iata, None)
    shops = list(db.scalars(select(Shop).where(Shop.iata == iata, Shop.visible.is_(True)).order_by(Shop.id)))
    city = next((s.city for s in shops if s.city), None)
    return urls.airport_path(iata, city, shops[0].name if shops else None)


def _place_status(db: Session, path: str, slug: str, category_slug: str | None) -> PageStatus:
    from app.services import catalog_queries, urls

    iata = urls.parse_airport_slug(slug)
    place = place_by_identifier(db, "iata", iata) if iata else None
    if place is None:
        return _missing(path, "no airport answers this address")
    if is_hidden(place):
        return _forwards(path, hidden_forward(place), "hidden, so the address answers a 302 to the index of its kind")
    canonical = _airport_canonical(db, iata)
    if category_slug is not None:
        category = urls.category_from_slug(category_slug)
        if category is None:
            return _missing(path, "no category answers this address")
        # The pair only has a page over the coverage bar: under it the route answers a real 404
        # and the sitemap omits it, so the address is nothing to be indexable about. Resolved the
        # way the page resolves it, through the links the airport's own rail draws
        # (`coverage.qualifying_pairs`, the bar `catalog_queries.airport_category_detail` reads),
        # and resolved before the canonical is compared because the route 404s ahead of its 301.
        pages = {link.path for link in catalog_queries.category_page_links(db, iata, canonical)}
        canonical = urls.airport_category_path(canonical, category)
        if canonical not in pages:
            return _missing(path, "no page exists for this category at this airport")
    if canonical != path:
        return _forwards(path, canonical, "a stale address for this airport; it answers a 301 to the live one")
    if category_slug is not None:
        # A category at an airport has no row of its own: it rides on the airport's decision,
        # the way `sitemap_entries` lists the pairing pages of an approved airport and no other.
        return (_indexed_status(path, place, canonical) if is_indexed(place)
                else PageStatus(path, "noindex", "the airport page it belongs to is not approved for the index",
                                ROBOTS_FOLLOW, False, canonical=canonical))
    return _indexed_status(path, place, canonical)


def _line_status(db: Session, path: str, slug: str) -> PageStatus:
    from app.services import catalog_queries, urls

    line = find(db, "product_line", slug)
    if line is not None:
        if is_hidden(line):
            return _forwards(path, hidden_forward(line), "hidden, so the address answers a 302 to its brand")
        canonical = page_path(line)
        if canonical != path:
            return _forwards(path, canonical, "an alias or a retired address; it answers a 301 to the live one")
        return _indexed_status(path, line, canonical)
    old = urls.parse_product_slug(slug)
    landing = catalog_queries.line_of_variant(db, old) if old is not None else None
    if landing is None:
        return _missing(path, "no product line answers this address")
    variant_id, line_slug = landing
    return _forwards(path, urls.line_path(line_slug, variant_id),
                     "an old product variant address; it answers a 301 to its product line",
                     canonical=urls.line_path(line_slug))


def _declared_page_status(db: Session, path: str) -> PageStatus:
    """An address with no row of its own: a family landing, a category page, `/how-we-choose`,
    the privacy policy and the terms of use.
    A person approves it by name (`INDEXED_PAGES`), which is what decides the robots directive.

    Whether the sitemap lists it is a second question with a second answer, so it is read from
    the list the sitemap itself builds rather than derived again here: `/how-we-choose` is
    approved by name yet never emitted, and a family or category address whose page has fallen
    away drops out of `indexed_category_rows` while staying named.
    """
    from app.services import seo

    if not is_indexed_path(path):
        return PageStatus(path, "noindex", "no person has named this address in INDEXED_PAGES yet",
                          ROBOTS_FOLLOW, False, canonical=path)
    in_sitemap = path in seo.static_sitemap_paths() or path in {p for p, _ in seo.indexed_category_rows(db)}
    reason = ("a person approved this address for the index" if in_sitemap else
              "a person approved this address; the sitemap lists it only while a page qualifies")
    return PageStatus(path, "indexable", reason, None, in_sitemap, canonical=path)


def status_of_path(db: Session, path: str) -> PageStatus:
    """What the address `path` is: one of `STATUSES`, with the reason, the robots directive the
    head would send, whether the sitemap lists it, and where it forwards.

    Reads rows and the declared tables only. It calls no head builder (the head builders call the
    same facts, and `tests/test_page_status.py` pins the two in step) and it writes nothing: this
    is a GET, and a GET that writes deadlocked a listing page against a running collection once.
    """
    from app.services import editorial, seo, urls

    clean = (path or "/").split("?")[0].split("#")[0]
    clean = clean.rstrip("/") or "/"
    segments = clean.strip("/").split("/") if clean != "/" else []

    if len(segments) == 2 and segments[0] == "brands":
        return _row_status(db, "brand", segments[1], clean)
    if len(segments) == 2 and segments[0] == "products":
        return _line_status(db, clean, segments[1])
    if segments and segments[0] == "airports" and 2 <= len(segments) <= 3:
        return _place_status(db, clean, segments[1], segments[2] if len(segments) == 3 else None)
    if len(segments) == 2 and segments[0] == "articles":
        row = editorial.article_by_slug(db, segments[1])
        if row is None:
            return _missing(clean, "a draft or an unknown slug, which answers a real 404")
        return PageStatus(clean, "indexable", "a published article; an article indexes on publish",
                          None, True, canonical=editorial.article_path(row.slug))
    # The family landings and their category pages, and any other address approved by name.
    # `/articles` is one of them: the pieces index on publish, the centre only once a person
    # has named it (`seo.head_for_articles` and `seo.static_sitemap_paths` read the same rule).
    family_paths = {f"/{word}" for word in urls.FAMILY_SLUGS.values()}
    if clean in family_paths or clean in (seo.METHOD_PATH, seo.ARTICLES_PATH, *seo.LEGAL_PATHS.values()):
        return _declared_page_status(db, clean)
    if len(segments) == 2 and f"/{segments[0]}" in family_paths:
        if urls.category_from_slug(segments[1]) is None:
            return _missing(clean, "no category answers this address")
        return _declared_page_status(db, clean)

    head = seo.STATIC_HEADS.get(clean)
    if head is not None:
        if head.noindex:
            return PageStatus(clean, "unlisted", "a client, owner or account surface: linked to nobody, never indexed",
                              ROBOTS_FOLLOW if head.follow else ROBOTS_NOFOLLOW, False, canonical=clean)
        in_sitemap = clean in seo.static_sitemap_paths()
        return PageStatus(clean, "indexable", "a hub the site is built around", None, in_sitemap, canonical=clean)
    if clean == "/data":
        return PageStatus(clean, "indexable", "a hub the site is built around", None,
                          "/data" in seo.static_sitemap_paths(), canonical=clean)
    if clean.endswith(".html"):
        return PageStatus(clean, "unlisted", "an explainer file served by rule, for the client and us",
                          ROBOTS_NOFOLLOW, False, canonical=clean)
    if seo.is_known_route(clean):
        return PageStatus(clean, "indexable", "no robots directive, and no sitemap entry either", None, False,
                          canonical=clean)
    return _missing(clean, "no route answers this address")