booth/booth/items.py

"""The item record — ONE resolver for what is in a booth.

Before this module, three functions independently walked a booth and derived
overlapping subsets of the same facts: `build_gallery` (kind, caption, blur,
doc), `booth_view_file` (kind, doc, image ring) and `list_booths` (kind counts,
cover thumb). The zoom route's subset was the smallest, and the fact it lacked
was the caption — so an annotated image lost its annotation at exactly the size
where the annotation is most readable.

That was never a rendering bug. It was three readers of one truth. This module
is the one truth; every surface reads its record and derives nothing itself.

See docs/contracts/u1_item_record.contract.md.
"""

from __future__ import annotations

import json
import os
import re
import stat
from dataclasses import dataclass
from pathlib import Path
from typing import Sequence
from urllib.parse import quote

try:  # optional: markdown rendering degrades to raw text without it
    import markdown as _markdown
except ImportError:  # pragma: no cover
    _markdown = None

from booth.asks import is_answer_file, is_ask_file

# Browser-playable media buckets. Anything else renders as a download link.
IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".avif", ".svg", ".bmp"}
VIDEO_EXTS = {".webm", ".mp4", ".ogv", ".m4v", ".mov"}
AUDIO_EXTS = {".mp3", ".wav", ".ogg", ".oga", ".flac", ".m4a", ".opus", ".aac"}

# Loose text docs that render as a readable page rather than a download.
MARKDOWN_EXTS = {".md", ".markdown", ".mdown"}
TEXT_EXTS = {".txt", ".text", ".log"}

CAPTION_MAX = 800  # chars of a sidecar .txt caption we render
DOC_MAX_BYTES = 2 * 1024 * 1024  # above this, a doc is handed back raw, not rendered

BLUR_FILE = ".blurred"


def classify(name: str) -> str:
    """image | video | audio | other, by extension."""
    ext = Path(name).suffix.lower()
    if ext in IMAGE_EXTS:
        return "image"
    if ext in VIDEO_EXTS:
        return "video"
    if ext in AUDIO_EXTS:
        return "audio"
    return "other"


def doc_kind(name: str) -> str | None:
    """'markdown' | 'text' | None — a booth file viewable as a readable page."""
    ext = Path(name).suffix.lower()
    if ext in MARKDOWN_EXTS:
        return "markdown"
    if ext in TEXT_EXTS:
        return "text"
    return None


def render_doc(text: str, kind: str) -> tuple[str, bool]:
    """(rendered, is_html). Markdown → HTML (fenced code, tables, sane lists);
    plain text — or markdown when the lib is unavailable — → raw text for <pre>.

    Text is returned RAW on purpose: the template escapes it inside <pre>, and
    pre-escaping here would double-encode under Jinja autoescape.
    """
    if kind == "markdown" and _markdown is not None:
        html = _markdown.markdown(text, extensions=["fenced_code", "tables", "sane_lists"])
        return html, True
    return text, False


@dataclass(frozen=True)
class Item:
    """One renderable file in a booth, with every fact any surface needs.

    `rel` is the identity — the booth-relative POSIX path. Marks (U2) attach to
    it, blur is keyed by it, and the zoom route resolves by it.
    """

    rel: str
    url: str
    kind: str
    section: str | None
    group: str | None
    caption: str | None
    blurred: bool
    doc: str | None
    size: int
    # R2 C1: the 1-based position in `booth_items` order over ALL items — the
    # number the operator means by "the third one". Set in the resolver loop
    # and nowhere else (INV-1). APPENDED, never inserted: a mid-dataclass field
    # is a positional-construction break.
    ordinal: int


# R2 C2: which items have been looked at full size. UI state, not judgment —
# never exposed to sessions, holds nothing. One viewer: this records WHAT was
# seen, never who saw it.
SEEN_FILE = ".seen"


# A seen marker bigger than this is not one this service wrote: a JSON array of
# every rel in a 270-item booth is a few KB.
SEEN_MAX_BYTES = 1 << 20


def read_seen(booth: Path) -> set[str]:
    """Rels seen at full size (R2 C2). A JSON array of strings, because a rel
    may hold a leading space or a newline and must round-trip exactly.

    NEVER RAISES and NEVER BLOCKS. Any fleet session can write into a booth,
    so the marker may be planted: it is opened without following a link and
    without blocking (a FIFO with no writer), refused unless it is a regular
    file of sane size, and anything unreadable or malformed reads as nothing
    seen — a damaged marker costs the tape its memory, never the page.
    """
    try:
        fd = os.open(booth / SEEN_FILE, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
    except OSError:
        return set()
    try:
        st = os.fstat(fd)
        if not stat.S_ISREG(st.st_mode) or st.st_size > SEEN_MAX_BYTES:
            return set()
        raw = os.read(fd, SEEN_MAX_BYTES + 1)
    except OSError:
        return set()
    finally:
        os.close(fd)
    try:
        data = json.loads(raw.decode("utf-8"))
    except (UnicodeDecodeError, ValueError, RecursionError):
        # RecursionError: a deeply nested array (`[[[[...`) blows the parser's
        # stack, and it is neither a ValueError nor an OSError — the same hole
        # marks.py, manifest.py and benches.py already close.
        return set()
    if not isinstance(data, list):
        return set()
    return {r for r in data if isinstance(r, str)}


def read_blurred(booth: Path) -> set[str]:
    """Blurred item paths for a booth. Missing file -> empty set."""
    try:
        text = (booth / BLUR_FILE).read_text()
    except (OSError, UnicodeDecodeError):
        return set()
    return {ln.strip() for ln in text.splitlines() if ln.strip()}


def _section_of(rel: str) -> str | None:
    """The item's parent directory relative to the booth; None at the root.

    Derived, never stored. This is the whole input to the navigation fix (U7):
    the structure a poster already created on disk, which `rglob` has been
    flattening into one wall at render time.
    """
    parent = Path(rel).parent
    return None if str(parent) == "." else parent.as_posix()


# One separator run between name segments. A filename is the only grouping
# signal the live booths actually carry: 0 of 11 galleries have a subdirectory.
_SEG = re.compile(r"[-_. ]+")


def _group_of(rel: str) -> str | None:
    """The grouping key for an item, or None when it has none.

    THE RULE, in one line: **the first separator-delimited segment of the
    basename's stem — with a trailing digit run stripped only when the stem has
    no separator at all.** `00-sheet-c1-market-noon.png` -> `00`;
    `m-c1-market-noon-9401.png` -> `m`; `flag-rear.png` -> `flag`;
    `ac01.png` -> `ac` (no separator, so the digits are the separator);
    `v30-seed8302.png` -> `v30` (separator present, so `v30` survives and does
    not merge with `v35`, which is the axis that booth is about).

    None for a stem with nothing before the digits -- `01.png` has no prefix to
    group on, and inventing one would file every numbered render under the
    empty string.

    ⚠ THIS IS NOT THE RULE THE CONTRACT FIRST NAMED. `strip ONE trailing run of
    digits` was measured against the live set on 2026-09-22 and yields 24 groups
    for sindra-bakeoff's 40 images and 27 for sindra's 30 -- a rail with one row
    per tile. The contract's own table claimed 5 and 1 for those two booths;
    neither reproduces under the rule it states beside them. The rewritten table
    carries the re-measurement.

    Derived HERE and nowhere else (INV-1). A route body that re-derived it would
    be the caption bug in a new field.
    """
    stem = Path(rel).stem  # basename without its last suffix; `a.tar.gz` -> `a.tar`
    segs = _SEG.split(stem)
    if len(segs) == 1:
        return re.sub(r"\d+$", "", stem) or None
    return segs[0] or None


def _resolve_captions(by_rel: dict[str, Path]) -> tuple[dict[str, str], set[str]]:
    """(caption-by-rel, rels consumed as sidecars).

    Two forms, in this precedence, preserved from the original gallery:
      1. `<file>.txt`  — `a.png.txt` captions `a.png`
      2. `<stem>.txt` beside a same-stem MEDIA sibling — `a.txt` captions
         `a.png`, but NOT `a.bin` (the `classify != "other"` guard, so a stray
         `data.txt` next to `data.bin` stays an item of its own)

    The sibling scan runs in sorted order rather than filesystem order: when two
    media files share a stem in one directory (`a.png` and `a.webm`), the
    original picked whichever `rglob` happened to yield first. Same rule, now
    deterministic.
    """
    caption: dict[str, str] = {}
    sidecars: set[str] = set()

    for rel in sorted(by_rel):
        if not rel.lower().endswith(".txt"):
            continue
        p = by_rel[rel]
        target = None

        base_full = rel[:-4]  # "a.png.txt" -> "a.png"
        if base_full in by_rel:
            target = base_full
        else:
            parent = str(Path(rel).parent)
            stem = Path(rel).stem
            for q_rel in sorted(by_rel):
                if q_rel == rel:
                    continue
                q = by_rel[q_rel]
                if (
                    str(Path(q_rel).parent) == parent
                    and Path(q_rel).stem == stem
                    and classify(q.name) != "other"
                ):
                    target = q_rel
                    break

        if target is not None:
            try:
                # BOUNDED AT THE READ. `read_text()` pulled the whole sidecar
                # into memory before the slice trimmed it, so a pathological
                # file was a MemoryError — which the OSError handler below does
                # not catch — rather than a missing caption.
                #
                # Deliberately NOT bounded by st_size: a FIFO reports 0 and a
                # bound that trusts it inherits what it does not mean, which is
                # the hang in persistent-memory.d/2026-09-22-size-cap-opened-a-hang.md.
                # The factor of 4 is UTF-8's worst case, so CAPTION_MAX
                # characters always survive the byte bound.
                with p.open("r", errors="replace") as fh:
                    caption[target] = fh.read(CAPTION_MAX * 4).strip()[:CAPTION_MAX]
            except OSError:
                pass
            sidecars.add(rel)

    return caption, sidecars


def booth_items(booth: Path) -> list[Item]:
    """Every renderable file in a booth, sorted by relative path.

    Excluded: dotfiles, `*.ask.json` / `*.answer.json` (they render as the asks
    panel, not as tiles), and any file consumed as another item's caption.

    Doc BODIES are deliberately not rendered here. The index calls this once per
    booth to count items and pick a cover; rendering every doc in every booth on
    every page load would be the cost of that convenience. `render_doc_body` is
    the separate step, for the one consumer that needs it.
    """
    by_rel: dict[str, Path] = {}
    for p in booth.rglob("*"):
        if not p.is_file() or p.name.startswith("."):
            continue
        if is_ask_file(p.name) or is_answer_file(p.name):
            continue
        rel = p.relative_to(booth).as_posix()
        try:
            quote(rel, safe="/")
        except UnicodeEncodeError:
            # A non-UTF-8 filename reaches CPython as a surrogate escape, and
            # `quote` raises on it. This used to happen at Item construction,
            # OUTSIDE any per-item handler — so one 0xff byte in one filename
            # took out that booth's page AND the index for every booth, because
            # `list_booths` calls this too. The repo's posture is that a damaged
            # file costs its own tile and never the page.
            #
            # Skipped rather than rescued: a name that cannot be percent-encoded
            # cannot be linked, served or zipped either, so there is no item to
            # render. Found by the heid bug-hunt panel (hulda), 2026-09-22.
            continue
        by_rel[rel] = p

    caption, sidecars = _resolve_captions(by_rel)
    blurred = read_blurred(booth)  # ONE read per call, not one per item

    items: list[Item] = []
    for rel in sorted(by_rel):
        if rel in sidecars:
            continue
        p = by_rel[rel]
        try:
            size = p.stat().st_size
        except OSError:
            size = 0
        items.append(
            Item(
                rel=rel,
                url=quote(rel, safe="/"),
                kind=classify(p.name),
                section=_section_of(rel),
                group=_group_of(rel),
                caption=caption.get(rel),
                blurred=rel in blurred,
                doc=doc_kind(p.name),
                size=size,
                # Counted over items that RENDER: a caption sidecar or a name
                # the quote() guard skipped takes no number, so the numbers
                # stay contiguous over what the operator can see.
                ordinal=len(items) + 1,
            )
        )
    return items


def image_chain(items: Sequence[Item]) -> list[str]:
    """The rels of the image items, in order — the zoom view's prev/next ring.

    Replaces `booth_image_names`, which walked the tree a second time to derive
    what the item list already knows.
    """
    return [it.rel for it in items if it.kind == "image"]


# R2 C2: what the review route steps through. ONE LINE: the item order
# filtered to media. It is a declared change to the zoom-ring rule, which was
# images only: a listening set is reviewed the same way a picture set is.
REVIEW_KINDS = ("image", "video", "audio")


def review_chain(items: Sequence[Item]) -> list[str]:
    """The rels of the media items, in item order — the review's prev/next ring,
    its filmstrip and its tape."""
    return [it.rel for it in items if it.kind in REVIEW_KINDS]


def find_item(items: Sequence[Item], rel: str) -> Item | None:
    """The record for one rel, or None — the zoom/doc route's entry point."""
    for it in items:
        if it.rel == rel:
            return it
    return None


def render_doc_body(booth: Path, item: Item) -> tuple[str, bool] | None:
    """(body, is_html) for a doc item under DOC_MAX_BYTES, else None.

    None means "do not inline this": either it is not a doc, or it is a log big
    enough that inlining it into every page render is the wrong trade.
    """
    if item.doc is None or item.size > DOC_MAX_BYTES:
        return None
    try:
        text = (booth / item.rel).read_text(encoding="utf-8", errors="replace")
    except OSError:
        return None
    return render_doc(text, item.doc)