jellyfin-scene-search/scene_search/caption_cleanup.py

"""Dry-run first cleanup of meta-prefaced captions already saved in SQLite.

Only known introductory boilerplate is removed. No Ollama inference,
embedding regeneration or full Meilisearch rebuild is needed.
"""
from __future__ import annotations

from pathlib import Path
from typing import Any, Callable

from .caption_text import remove_image_preface
from .database import connect, init_db


def clean_stored_caption_prefaces(
    db_path: Path,
    *,
    apply: bool = False,
    limit: int = 0,
    sync_fn: Callable[[list[str]], int] | None = None,
) -> dict[str, Any]:
    if limit < 0:
        raise ValueError("limit must be >= 0")
    init_db(db_path)
    with connect(db_path) as con:
        # When the Caption Editor is installed, do not rewrite captions that
        # the user has already manually corrected.
        history_exists = con.execute(
            """SELECT 1 FROM sqlite_master
               WHERE type = 'table' AND name = 'caption_edit_history'"""
        ).fetchone() is not None

    window_guard = (
        """AND NOT EXISTS (
            SELECT 1 FROM caption_edit_history h
            WHERE h.caption_kind = 'window' AND h.caption_id = cw.id
        )""" if history_exists else ""
    )
    frame_guard = (
        """AND NOT EXISTS (
            SELECT 1 FROM caption_edit_history h
            WHERE h.caption_kind = 'frame' AND h.caption_id = f.id
        )""" if history_exists else ""
    )
    sql = f"""
        SELECT 'window' AS kind, cw.id, cw.media_id, cw.caption,
               cw.representative_frame_id AS frame_id
        FROM caption_windows cw
        WHERE cw.caption IS NOT NULL AND TRIM(cw.caption) != ''
          {window_guard}

        UNION ALL

        SELECT 'frame' AS kind, f.id, f.media_id, f.caption,
               f.id AS frame_id
        FROM frames f
        WHERE f.caption IS NOT NULL AND TRIM(f.caption) != ''
          AND NOT EXISTS (
              SELECT 1 FROM caption_window_frames cwf
              WHERE cwf.frame_id = f.id
          )
          {frame_guard}
    """
    stats: dict[str, Any] = {
        "scanned": 0, "matched": 0, "updated": 0,
        "meili_synced": 0, "meili_errors": [],
        "published_overviews_unchanged": 0, "examples": [],
    }
    sync = sync_fn
    if apply and sync is None:
        from .sync import sync_frame_metadata
        sync = sync_frame_metadata

    def process_batch(edits: list[tuple[dict[str, Any], str]]) -> None:
        if not edits or not apply:
            return
        frame_ids: list[str] = []
        with connect(db_path) as write_con:
            for row, cleaned in edits:
                table = "caption_windows" if row["kind"] == "window" else "frames"
                extra = ", updated_at = CURRENT_TIMESTAMP" if row["kind"] == "window" else ""
                changed = write_con.execute(
                    f"""UPDATE {table} SET caption = ?, meili_synced_at = NULL{extra}
                        WHERE id = ? AND caption = ?""",
                    (cleaned, row["id"], row["caption"]),
                ).rowcount
                if not changed:
                    continue
                frame_ids.append(str(row["frame_id"]))
                stats["updated"] += 1
                write_con.execute(
                    "UPDATE frames SET meili_synced_at = NULL WHERE id = ?",
                    (row["frame_id"],),
                )
                if history_exists:
                    write_con.execute(
                        """INSERT INTO caption_edit_history
                           (media_id, caption_kind, caption_id,
                            previous_caption, corrected_caption)
                           VALUES (?, ?, ?, ?, ?)""",
                        (row["media_id"], row["kind"], row["id"],
                         row["caption"], cleaned),
                    )
                overview = write_con.execute(
                    """SELECT applied_overview FROM jellyfin_overview_drafts
                       WHERE media_id = ?""",
                    (row["media_id"],),
                ).fetchone()
                if overview and overview["applied_overview"] is not None:
                    stats["published_overviews_unchanged"] += 1
                else:
                    write_con.execute(
                        """DELETE FROM jellyfin_overview_drafts
                           WHERE media_id = ? AND applied_overview IS NULL""",
                        (row["media_id"],),
                    )
        if not frame_ids:
            return
        try:
            assert sync is not None
            stats["meili_synced"] += sync(list(dict.fromkeys(frame_ids)))
        except Exception as exc:
            # Partial updates were saved canonically with null sync markers:
            # the usual caption catch-up can retry without regenerating.
            stats["meili_errors"].append(str(exc)[:300])

    with connect(db_path) as con:
        cursor = con.execute(sql)
        done = False
        while not done:
            rows = cursor.fetchmany(500)
            if not rows:
                break
            batch: list[tuple[dict[str, Any], str]] = []
            for raw in rows:
                row = dict(raw)
                stats["scanned"] += 1
                new_caption = remove_image_preface(row["caption"])
                if new_caption == row["caption"]:
                    continue
                stats["matched"] += 1
                if len(stats["examples"]) < 8:
                    stats["examples"].append({
                        "before": row["caption"][:200],
                        "after": new_caption[:200],
                    })
                batch.append((row, new_caption))
                if limit and stats["matched"] >= limit:
                    done = True
                    break
            process_batch(batch)
    return stats