backend/scene_processing/scene_db.py

"""SQLite storage for scene descriptions + video summaries.

Schema:
    scene_windows    - one row per (natural_key, language, window_index)
    video_summaries  - one row per (natural_key, language): a structured
                       TL;DR + themes + acts, with the TL;DR embedded
    scene_runs       - audit row per indexing run, for resumability

Embeddings are stored as raw float32 BLOBs (bge-m3 is 1024-dim, so 4096
bytes per row). Brute-force cosine similarity over ~30k windows is fast
enough that we don't need a vector index yet; if that changes, switch to
sqlite-vec without changing the row shape.

WAL mode + 30s busy_timeout matches the patterns used elsewhere in the
project (search.py, search_images_db.py) so concurrent shard writers
behave the same way.

This module also exposes :func:`publish_snapshot`, which the batch indexer
calls on a cadence to put a consistent, rsync-safe copy of the DB into a
"pull" directory. The operator can rsync that directory at any time and
get a clean point-in-time snapshot — never a torn mid-write read.
"""

from __future__ import annotations

import errno
import fcntl
import json
import os
import sqlite3
import struct
import time
from contextlib import contextmanager
from typing import Any, Iterable, Iterator, Sequence

from utils import log_message


SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS scene_windows (
    natural_key      TEXT NOT NULL,
    language         TEXT NOT NULL,
    window_index     INTEGER NOT NULL,
    start_seconds    REAL NOT NULL,
    end_seconds      REAL NOT NULL,
    duration_seconds REAL NOT NULL,
    subtitle_text    TEXT,
    description      TEXT NOT NULL,
    embedding        BLOB NOT NULL,
    embedding_dim    INTEGER NOT NULL,
    vlm_model_id     TEXT,
    embed_model_id   TEXT,
    audio_tags       TEXT,
    shot_type        TEXT,
    setting          TEXT,
    era              TEXT,
    generated_at     TEXT NOT NULL,
    PRIMARY KEY (natural_key, language, window_index)
);

CREATE INDEX IF NOT EXISTS idx_scene_windows_nk
    ON scene_windows(natural_key, language);

CREATE TABLE IF NOT EXISTS video_summaries (
    natural_key      TEXT NOT NULL,
    language         TEXT NOT NULL,
    duration_seconds REAL NOT NULL,
    window_count     INTEGER NOT NULL,
    tldr             TEXT NOT NULL,
    themes_json      TEXT,
    acts_json        TEXT,
    parse_status     TEXT,
    embedding        BLOB NOT NULL,
    embedding_dim    INTEGER NOT NULL,
    vlm_model_id     TEXT,
    embed_model_id   TEXT,
    generated_at     TEXT NOT NULL,
    PRIMARY KEY (natural_key, language)
);

CREATE TABLE IF NOT EXISTS scene_runs (
    natural_key      TEXT NOT NULL,
    language         TEXT NOT NULL,
    status           TEXT NOT NULL,
    error_message    TEXT,
    window_count     INTEGER,
    duration_seconds REAL,
    processing_seconds REAL,
    started_at       TEXT NOT NULL,
    completed_at     TEXT,
    PRIMARY KEY (natural_key, language)
);

CREATE TABLE IF NOT EXISTS scene_video_locations (
    natural_key  TEXT NOT NULL,
    language     TEXT NOT NULL,
    place_name   TEXT NOT NULL,
    place_type   TEXT,
    country_iso  TEXT,
    evidence     TEXT,
    PRIMARY KEY (natural_key, language, place_name, country_iso)
);

CREATE INDEX IF NOT EXISTS idx_scene_video_locations_iso
    ON scene_video_locations(country_iso, language);

CREATE TABLE IF NOT EXISTS scene_window_locations (
    natural_key  TEXT NOT NULL,
    language     TEXT NOT NULL,
    window_index INTEGER NOT NULL,
    place_name   TEXT NOT NULL,
    PRIMARY KEY (natural_key, language, window_index, place_name)
);

CREATE INDEX IF NOT EXISTS idx_scene_window_locations_place
    ON scene_window_locations(place_name, language);
"""


# ---------------------------------------------------------------------------
# Connection helpers
# ---------------------------------------------------------------------------

def connect(db_path: str) -> sqlite3.Connection:
    """Open a connection with WAL + 30s busy_timeout. Creates parents as needed."""
    parent = os.path.dirname(os.path.abspath(db_path))
    if parent:
        os.makedirs(parent, exist_ok=True)
    conn = sqlite3.connect(db_path, timeout=30.0, isolation_level=None)
    conn.execute("PRAGMA journal_mode = WAL")
    conn.execute("PRAGMA busy_timeout = 30000")
    conn.execute("PRAGMA synchronous = NORMAL")
    return conn


def initialize_schema(conn: sqlite3.Connection) -> None:
    """Apply the schema. Idempotent."""
    conn.executescript(SCHEMA_SQL)
    _migrate_add_columns(conn)


def _migrate_add_columns(conn: sqlite3.Connection) -> None:
    """Add later columns to an already-existing scene_windows table.

    CREATE TABLE IF NOT EXISTS leaves an existing table's columns untouched, so
    a DB built before the visual-attribute columns needs them added explicitly.
    SQLite ADD COLUMN is a cheap metadata-only change (no table rewrite) and the
    new columns default to NULL. Idempotent: skips columns that already exist.
    """
    existing = {row[1] for row in conn.execute("PRAGMA table_info(scene_windows)")}
    for column in ("shot_type", "setting", "era"):
        if column not in existing:
            conn.execute(f"ALTER TABLE scene_windows ADD COLUMN {column} TEXT")


@contextmanager
def open_db(db_path: str) -> Iterator[sqlite3.Connection]:
    """Context manager that yields a connection with schema applied."""
    conn = connect(db_path)
    try:
        initialize_schema(conn)
        yield conn
    finally:
        conn.close()


# ---------------------------------------------------------------------------
# Embedding (de)serialization
# ---------------------------------------------------------------------------

def encode_embedding(vector: Sequence[float]) -> bytes:
    """Pack a float vector into a binary blob (little-endian float32)."""
    return struct.pack(f"<{len(vector)}f", *vector)


def decode_embedding(blob: bytes) -> list[float]:
    """Unpack a blob produced by :func:`encode_embedding`."""
    count = len(blob) // 4
    return list(struct.unpack(f"<{count}f", blob))


# ---------------------------------------------------------------------------
# Read / query helpers (the search layer)
# ---------------------------------------------------------------------------
#
# The query side loads the stored embeddings into an in-memory float32 matrix
# and brute-forces cosine similarity against the query vector. At ~30k windows
# x 1024 dims that is ~120 MB and a single numpy dot product — fast enough that
# we don't need a vector index yet. If the corpus grows enough that this gets
# slow, this is the seam to swap in sqlite-vec (an ANN MATCH query that returns
# rowids + distances) without changing the row shape or the callers below.


def count_scene_windows(conn: sqlite3.Connection, language: str | None = None) -> int:
    """Number of scene_window rows (optionally for one language).

    Used so the search layer can tell "index is still empty" apart from
    "query genuinely matched nothing" and give the user an actionable message.
    """
    if language is None:
        cur = conn.execute("SELECT COUNT(*) FROM scene_windows")
    else:
        cur = conn.execute(
            "SELECT COUNT(*) FROM scene_windows WHERE language = ?", (language,)
        )
    return int(cur.fetchone()[0])


def count_video_summaries(conn: sqlite3.Connection, language: str | None = None) -> int:
    """Number of video_summaries rows (optionally for one language)."""
    if language is None:
        cur = conn.execute("SELECT COUNT(*) FROM video_summaries")
    else:
        cur = conn.execute(
            "SELECT COUNT(*) FROM video_summaries WHERE language = ?", (language,)
        )
    return int(cur.fetchone()[0])


def _build_embedding_matrix(
    records: list[tuple[dict, bytes, int | None]],
) -> tuple[list[dict], "Any", int]:
    """Decode embedding blobs into a rectangular float32 matrix, robustly.

    ``records`` is a list of ``(metadata, blob, declared_dim)``. Returns
    ``(kept_metadata, matrix, skipped)``. A row is skipped (counted in
    ``skipped``, not fatal) when its blob can't be decoded (e.g. a truncated /
    torn write whose byte length isn't a multiple of 4) or when its decoded
    length disagrees with its own ``declared_dim``. Of the rows that decode
    cleanly, only those matching the *modal* (most common) dimension are kept,
    so a stray row from a different embedding model can't make ``np.vstack``
    raise and take the whole query down with it. This is what backs the
    "one corrupt row can't break a query" promise the loaders document.
    """
    import numpy as np
    from collections import Counter

    decoded: list[tuple[dict, Any]] = []
    skipped = 0
    for metadata, blob, declared_dim in records:
        try:
            vector = np.frombuffer(blob, dtype="<f4")
        except (ValueError, TypeError):
            skipped += 1
            continue
        if declared_dim and vector.shape[0] != declared_dim:
            skipped += 1
            continue
        decoded.append((metadata, vector))

    if not decoded:
        return [], np.empty((0, 0), dtype=np.float32), skipped

    modal_dim = Counter(vec.shape[0] for _, vec in decoded).most_common(1)[0][0]
    kept_metadata: list[dict] = []
    kept_vectors: list[Any] = []
    for metadata, vector in decoded:
        if vector.shape[0] != modal_dim:
            skipped += 1
            continue
        kept_metadata.append(metadata)
        kept_vectors.append(vector)

    matrix = np.vstack(kept_vectors).astype(np.float32, copy=False)
    return kept_metadata, matrix, skipped


def load_window_index(
    conn: sqlite3.Connection, language: str | None = None
) -> tuple[list[dict], "Any"]:
    """Load all scene_window metadata + embeddings for brute-force search.

    Returns ``(metadata, embeddings)`` where ``metadata`` is a list of dicts
    (one per window, in DB order) and ``embeddings`` is a float32 numpy array
    of shape ``(n_windows, dim)`` whose row *i* is the embedding for
    ``metadata[i]``. Rows whose stored blob is undecodable or doesn't match the
    modal embedding dimension are skipped (and logged) so one corrupt row can't
    break a query.

    For an empty index returns ``([], empty (0, 0) array)`` so callers can
    branch on ``len(metadata) == 0`` without importing numpy themselves.
    """
    import numpy as np

    sql = (
        "SELECT natural_key, language, window_index, start_seconds, end_seconds, "
        "duration_seconds, description, subtitle_text, audio_tags, "
        "shot_type, setting, era, embedding, "
        "embedding_dim FROM scene_windows"
    )
    params: tuple = ()
    if language is not None:
        sql += " WHERE language = ?"
        params = (language,)
    sql += " ORDER BY natural_key, window_index"

    rows = conn.execute(sql, params).fetchall()
    if not rows:
        return [], np.empty((0, 0), dtype=np.float32)

    records: list[tuple[dict, bytes, int | None]] = []
    for (
        natural_key, lang, window_index, start_seconds, end_seconds,
        duration_seconds, description, subtitle_text, audio_tags,
        shot_type, setting, era, blob,
        embedding_dim,
    ) in rows:
        metadata = {
            "natural_key": natural_key,
            "language": lang,
            "window_index": int(window_index),
            "start_seconds": float(start_seconds),
            "end_seconds": float(end_seconds),
            "duration_seconds": float(duration_seconds),
            "description": description,
            "subtitle_text": subtitle_text,
            "audio_tags": audio_tags.split(",") if audio_tags else [],
            "shot_type": shot_type,
            "setting": setting,
            "era": era,
        }
        records.append((metadata, blob, embedding_dim))

    metadata_list, matrix, skipped = _build_embedding_matrix(records)
    if skipped:
        log_message(
            f"load_window_index: skipped {skipped} window(s) with undecodable or "
            f"off-modal embeddings (kept dim {matrix.shape[1] if matrix.size else 0})",
            "WARNING",
        )
    return metadata_list, matrix


def load_summary_index(
    conn: sqlite3.Connection, language: str | None = None
) -> tuple[list[dict], "Any"]:
    """Load all video_summaries metadata + TL;DR embeddings for video search.

    Same contract as :func:`load_window_index`. ``themes_json``/``acts_json``
    are decoded into ``themes`` (list[str]) / ``acts`` (list) so the search
    layer can text-match themes and return them without re-parsing JSON.
    """
    import numpy as np

    sql = (
        "SELECT natural_key, language, duration_seconds, window_count, tldr, "
        "themes_json, acts_json, embedding, embedding_dim FROM video_summaries"
    )
    params: tuple = ()
    if language is not None:
        sql += " WHERE language = ?"
        params = (language,)
    sql += " ORDER BY natural_key"

    rows = conn.execute(sql, params).fetchall()
    if not rows:
        return [], np.empty((0, 0), dtype=np.float32)

    records: list[tuple[dict, bytes, int | None]] = []
    for (
        natural_key, lang, duration_seconds, window_count, tldr,
        themes_json, acts_json, blob, embedding_dim,
    ) in rows:
        try:
            themes = json.loads(themes_json) if themes_json else []
        except (ValueError, TypeError):
            themes = []
        try:
            acts = json.loads(acts_json) if acts_json else []
        except (ValueError, TypeError):
            acts = []
        metadata = {
            "natural_key": natural_key,
            "language": lang,
            "duration_seconds": float(duration_seconds),
            "window_count": int(window_count),
            "tldr": tldr,
            "themes": themes if isinstance(themes, list) else [],
            "acts": acts if isinstance(acts, list) else [],
        }
        records.append((metadata, blob, embedding_dim))

    metadata_list, matrix, skipped = _build_embedding_matrix(records)
    if skipped:
        log_message(
            f"load_summary_index: skipped {skipped} summary row(s) with undecodable "
            f"or off-modal embeddings (kept dim {matrix.shape[1] if matrix.size else 0})",
            "WARNING",
        )
    return metadata_list, matrix


def cosine_topk(matrix: "Any", query: Sequence[float], k: int) -> list[tuple[int, float]]:
    """Brute-force cosine similarity: top-``k`` rows of ``matrix`` vs ``query``.

    Returns ``[(row_index, score), ...]`` sorted by score descending. Both the
    matrix rows and the query are L2-normalized defensively here, so cosine is
    valid even if a stored vector wasn't unit length — at this corpus size the
    extra normalization is negligible. ``score`` is in ``[-1, 1]``.

    This is where an ANN index (sqlite-vec) would replace the full dot product
    if ~30k windows ever stops being fast enough.
    """
    import numpy as np

    if matrix.size == 0 or k <= 0:
        return []
    q = np.asarray(query, dtype=np.float32)
    if q.ndim != 1 or q.shape[0] != matrix.shape[1]:
        raise ValueError(
            f"query dim {getattr(q, 'shape', None)} does not match index dim "
            f"{matrix.shape[1]}"
        )
    q_norm = np.linalg.norm(q)
    if q_norm == 0:
        return []
    q = q / q_norm

    row_norms = np.linalg.norm(matrix, axis=1)
    row_norms[row_norms == 0] = 1.0  # avoid divide-by-zero on a degenerate row
    scores = (matrix @ q) / row_norms

    n = scores.shape[0]
    k = min(k, n)
    # argpartition for the top-k, then sort just those k by score descending.
    top_idx = np.argpartition(scores, n - k)[n - k:]
    top_idx = top_idx[np.argsort(scores[top_idx])[::-1]]
    return [(int(i), float(scores[i])) for i in top_idx]


# ---------------------------------------------------------------------------
# Resumability
# ---------------------------------------------------------------------------

def already_indexed_keys(
    conn: sqlite3.Connection,
    language: str,
) -> set[str]:
    """Return the set of natural_keys that have at least one scene_window row.

    A run that crashed partway through a video leaves no scene_windows rows
    (writes happen in one batch at the end of process_video), so this is a
    safe basis for resumability.
    """
    cur = conn.execute(
        "SELECT DISTINCT natural_key FROM scene_windows WHERE language = ?",
        (language,),
    )
    return {row[0] for row in cur.fetchall()}


def successful_run_keys(conn: sqlite3.Connection, language: str) -> set[str]:
    """Return natural_keys whose last scene_runs row is status='ok'."""
    cur = conn.execute(
        "SELECT natural_key FROM scene_runs WHERE language = ? AND status = 'ok'",
        (language,),
    )
    return {row[0] for row in cur.fetchall()}


# ---------------------------------------------------------------------------
# Writers
# ---------------------------------------------------------------------------

def upsert_scene_windows(
    conn: sqlite3.Connection,
    *,
    natural_key: str,
    language: str,
    rows: Iterable[dict],
) -> int:
    """Upsert all scene_window rows for one video atomically. Returns row count.

    Wraps DELETE + INSERTs in a single BEGIN IMMEDIATE / COMMIT so a kill-9
    mid-loop either leaves the previous rows intact or replaces them
    completely — never a partial mix.

    Each row dict must include:
        window_index, start_seconds, end_seconds, duration_seconds,
        subtitle_text, description, embedding (sequence of float),
        embedding_dim, vlm_model_id, embed_model_id, audio_tags (list or
        None), generated_at.
    """
    generated_at = time.strftime("%Y-%m-%dT%H:%M:%S")
    count = 0
    conn.execute("BEGIN IMMEDIATE")
    try:
        conn.execute(
            "DELETE FROM scene_windows WHERE natural_key = ? AND language = ?",
            (natural_key, language),
        )
        for row in rows:
            audio_tags = row.get("audio_tags")
            audio_tags_str = ",".join(audio_tags) if audio_tags else None
            conn.execute(
                """
                INSERT INTO scene_windows (
                    natural_key, language, window_index,
                    start_seconds, end_seconds, duration_seconds,
                    subtitle_text, description, embedding, embedding_dim,
                    vlm_model_id, embed_model_id, audio_tags,
                    shot_type, setting, era, generated_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    natural_key, language, row["window_index"],
                    row["start_seconds"], row["end_seconds"], row["duration_seconds"],
                    row.get("subtitle_text"), row["description"],
                    encode_embedding(row["embedding"]), row["embedding_dim"],
                    row.get("vlm_model_id"), row.get("embed_model_id"),
                    audio_tags_str,
                    row.get("shot_type"), row.get("setting"), row.get("era"),
                    row.get("generated_at", generated_at),
                ),
            )
            count += 1
        conn.execute("COMMIT")
    except Exception:
        try:
            conn.execute("ROLLBACK")
        except sqlite3.Error:
            pass
        raise
    return count


def update_window_visual_attributes(
    conn: sqlite3.Connection,
    *,
    natural_key: str,
    language: str,
    window_index: int,
    shot_type: str,
    setting: str,
    era: str,
    embedding: Sequence[float],
    embedding_dim: int,
    embed_model_id: str | None = None,
) -> None:
    """Set a window's visual attributes and replace its embedding in place.

    Used by the text-backfill pass to enrich existing rows: it re-derives the
    attributes from the stored description and re-embeds (description + attribute
    phrase) so the attributes become searchable, without re-running the VLM.
    ``embed_model_id`` is refreshed when given so the column stays truthful.
    """
    conn.execute(
        "UPDATE scene_windows SET shot_type = ?, setting = ?, era = ?, "
        "embedding = ?, embedding_dim = ?, "
        "embed_model_id = COALESCE(?, embed_model_id) "
        "WHERE natural_key = ? AND language = ? AND window_index = ?",
        (
            shot_type, setting, era,
            encode_embedding(embedding), embedding_dim, embed_model_id,
            natural_key, language, window_index,
        ),
    )


def upsert_video_summary(
    conn: sqlite3.Connection,
    *,
    natural_key: str,
    language: str,
    duration_seconds: float,
    window_count: int,
    tldr: str,
    themes: Sequence[str],
    acts: Sequence[dict],
    embedding: Sequence[float],
    embedding_dim: int,
    parse_status: str | None = None,
    vlm_model_id: str | None = None,
    embed_model_id: str | None = None,
) -> None:
    """Insert or replace the structured video summary.

    ``themes`` and ``acts`` are stored as JSON text; ``embedding`` is the
    bge-m3 embedding of the TL;DR (what "what is this video about" search
    matches against).
    """
    conn.execute(
        """
        INSERT OR REPLACE INTO video_summaries (
            natural_key, language, duration_seconds, window_count,
            tldr, themes_json, acts_json, parse_status,
            embedding, embedding_dim, vlm_model_id, embed_model_id, generated_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            natural_key, language, duration_seconds, window_count,
            tldr, json.dumps(list(themes or [])), json.dumps(list(acts or [])),
            parse_status, encode_embedding(embedding), embedding_dim,
            vlm_model_id, embed_model_id, time.strftime("%Y-%m-%dT%H:%M:%S"),
        ),
    )


def upsert_video_locations(
    conn: sqlite3.Connection,
    *,
    natural_key: str,
    language: str,
    places: Iterable[dict],
    window_locations: Iterable[tuple[int, str]],
) -> None:
    """Replace all location rows for one video atomically.

    ``places`` are the video-level mentions: dicts with keys ``name``,
    ``type``, ``country_iso``, ``evidence`` (the shape produced by
    ``locations.LocationMention.to_dict``). ``window_locations`` is an
    iterable of ``(window_index, place_name)`` pairs.

    DELETE + INSERTs run in one BEGIN IMMEDIATE / COMMIT so a crash leaves
    either the previous rows or the new set, never a partial mix.
    """
    conn.execute("BEGIN IMMEDIATE")
    try:
        conn.execute(
            "DELETE FROM scene_video_locations WHERE natural_key = ? AND language = ?",
            (natural_key, language),
        )
        conn.execute(
            "DELETE FROM scene_window_locations WHERE natural_key = ? AND language = ?",
            (natural_key, language),
        )
        for place in places:
            conn.execute(
                """
                INSERT OR IGNORE INTO scene_video_locations (
                    natural_key, language, place_name, place_type,
                    country_iso, evidence
                ) VALUES (?, ?, ?, ?, ?, ?)
                """,
                (
                    natural_key, language, place.get("name"),
                    place.get("type"), place.get("country_iso"),
                    place.get("evidence"),
                ),
            )
        for window_index, place_name in window_locations:
            conn.execute(
                """
                INSERT OR IGNORE INTO scene_window_locations (
                    natural_key, language, window_index, place_name
                ) VALUES (?, ?, ?, ?)
                """,
                (natural_key, language, window_index, place_name),
            )
        conn.execute("COMMIT")
    except Exception:
        try:
            conn.execute("ROLLBACK")
        except sqlite3.Error:
            pass
        raise


def record_run_start(
    conn: sqlite3.Connection, *, natural_key: str, language: str
) -> None:
    """Mark this (key, language) as started. Used so we don't leave the
    table empty for a video that crashed before any scene_window row landed.
    """
    conn.execute(
        """
        INSERT OR REPLACE INTO scene_runs (
            natural_key, language, status, started_at
        ) VALUES (?, ?, 'running', ?)
        """,
        (natural_key, language, time.strftime("%Y-%m-%dT%H:%M:%S")),
    )


def record_run_ok(
    conn: sqlite3.Connection,
    *,
    natural_key: str,
    language: str,
    window_count: int,
    duration_seconds: float,
    processing_seconds: float,
) -> None:
    conn.execute(
        """
        INSERT OR REPLACE INTO scene_runs (
            natural_key, language, status, error_message,
            window_count, duration_seconds, processing_seconds,
            started_at, completed_at
        )
        VALUES (
            ?, ?, 'ok', NULL, ?, ?, ?,
            COALESCE((SELECT started_at FROM scene_runs
                      WHERE natural_key=? AND language=?), ?),
            ?
        )
        """,
        (
            natural_key, language, window_count, duration_seconds,
            processing_seconds,
            natural_key, language, time.strftime("%Y-%m-%dT%H:%M:%S"),
            time.strftime("%Y-%m-%dT%H:%M:%S"),
        ),
    )


# ---------------------------------------------------------------------------
# Snapshot publishing (continuous delivery during long runs)
# ---------------------------------------------------------------------------

@contextmanager
def _publish_lock(lock_path: str) -> Iterator[bool]:
    """Acquire an exclusive non-blocking file lock at ``lock_path``.

    Yields True if the lock was acquired (caller should proceed) or False
    if another process holds it (caller should skip this publish round).
    Multiple shard workers all call publish on their own cadence; this
    lock makes sure only one of them is writing the destination at a time.
    """
    parent = os.path.dirname(os.path.abspath(lock_path))
    if parent:
        os.makedirs(parent, exist_ok=True)
    fd = open(lock_path, "w")
    try:
        try:
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError as exc:
            if exc.errno in (errno.EAGAIN, errno.EACCES):
                yield False
                return
            raise
        try:
            yield True
        finally:
            fcntl.flock(fd, fcntl.LOCK_UN)
    finally:
        fd.close()


def publish_snapshot(source_db_path: str, dest_db_path: str) -> bool:
    """Atomically publish a consistent snapshot of ``source_db_path``.

    Uses sqlite3.Connection.backup() to copy every page including any data
    still in the WAL, so the destination is internally consistent even if
    the source has an in-flight transaction. The destination is written to
    a PID-suffixed temp path and atomically renamed into place; an rsync
    running concurrently will see either the previous snapshot or the new
    one, never a torn mid-copy file.

    A non-blocking file lock at ``<pull_dir>/.publish.lock`` makes this
    safe to call from multiple shard workers simultaneously. If another
    worker is mid-publish, this call returns ``False`` immediately and the
    caller continues; the next publish boundary will catch up.

    Returns True if a new snapshot was published, False if another worker
    was already publishing.
    """
    parent = os.path.dirname(os.path.abspath(dest_db_path))
    if parent:
        os.makedirs(parent, exist_ok=True)
    lock_path = os.path.join(parent, ".publish.lock")

    with _publish_lock(lock_path) as acquired:
        if not acquired:
            return False

        tmp_path = f"{dest_db_path}.tmp.{os.getpid()}"
        for ext in ("", "-wal", "-shm"):
            candidate = tmp_path + ext
            if os.path.exists(candidate):
                os.remove(candidate)

        source = sqlite3.connect(source_db_path, timeout=30.0)
        try:
            dest = sqlite3.connect(tmp_path)
            try:
                with dest:
                    source.backup(dest)
                # Convert the snapshot out of WAL mode so a read-write
                # consumer doesn't create -wal/-shm files next to it.
                dest.execute("PRAGMA journal_mode = DELETE")
            finally:
                dest.close()
        finally:
            source.close()

        os.replace(tmp_path, dest_db_path)
        for ext in ("-wal", "-shm"):
            stale = dest_db_path + ext
            if os.path.exists(stale):
                os.remove(stale)
        return True


def write_progress_manifest(
    manifest_path: str,
    *,
    payload: dict[str, Any],
) -> None:
    """Atomically write a small JSON progress manifest next to the snapshot.

    Uses a PID-suffixed tmp file so concurrent publishers don't clobber
    each other's intermediate state. The final ``os.replace`` is the only
    visible filesystem change; readers (including rsync) never see a
    half-written manifest.
    """
    parent = os.path.dirname(os.path.abspath(manifest_path))
    if parent:
        os.makedirs(parent, exist_ok=True)
    tmp_path = f"{manifest_path}.tmp.{os.getpid()}"
    with open(tmp_path, "w", encoding="utf-8") as handle:
        json.dump(payload, handle, indent=2, sort_keys=True)
    os.replace(tmp_path, manifest_path)


def aggregate_run_stats(conn: sqlite3.Connection, language: str) -> dict[str, Any]:
    """Summary stats for the manifest (videos done, ok/error counts, last key)."""
    cur = conn.execute(
        """
        SELECT status, COUNT(*), COALESCE(SUM(duration_seconds), 0),
               COALESCE(SUM(processing_seconds), 0)
          FROM scene_runs
         WHERE language = ?
         GROUP BY status
        """,
        (language,),
    )
    by_status: dict[str, dict[str, float]] = {}
    for status, count, duration, processing in cur.fetchall():
        by_status[status] = {
            "count": int(count),
            "source_hours": round((duration or 0) / 3600, 3),
            "processing_hours": round((processing or 0) / 3600, 3),
        }
    cur = conn.execute(
        """
        SELECT natural_key, completed_at FROM scene_runs
         WHERE language = ? AND status = 'ok' AND completed_at IS NOT NULL
         ORDER BY completed_at DESC LIMIT 1
        """,
        (language,),
    )
    row = cur.fetchone()
    last_ok = {"natural_key": row[0], "completed_at": row[1]} if row else None

    cur = conn.execute(
        "SELECT COUNT(DISTINCT natural_key) FROM scene_windows WHERE language = ?",
        (language,),
    )
    videos_with_rows = int(cur.fetchone()[0])

    cur = conn.execute(
        "SELECT COUNT(*) FROM scene_windows WHERE language = ?",
        (language,),
    )
    window_rows = int(cur.fetchone()[0])

    return {
        "language": language,
        "by_status": by_status,
        "videos_with_scene_rows": videos_with_rows,
        "window_rows": window_rows,
        "last_ok_video": last_ok,
    }


def record_run_error(
    conn: sqlite3.Connection,
    *,
    natural_key: str,
    language: str,
    error_message: str,
) -> None:
    conn.execute(
        """
        INSERT OR REPLACE INTO scene_runs (
            natural_key, language, status, error_message,
            started_at, completed_at
        )
        VALUES (
            ?, ?, 'error', ?,
            COALESCE((SELECT started_at FROM scene_runs
                      WHERE natural_key=? AND language=?), ?),
            ?
        )
        """,
        (
            natural_key, language, error_message,
            natural_key, language, time.strftime("%Y-%m-%dT%H:%M:%S"),
            time.strftime("%Y-%m-%dT%H:%M:%S"),
        ),
    )