backend/cli.py

#!/usr/bin/env python3
"""Command-line ingestion entrypoint (Phase 3a scaffold).

Exposes the existing ingestion operations as subcommands so they can run
OUTSIDE the web process. This is **purely additive**: it calls the existing
processing functions unchanged. It does NOT move heavy ML (torch/transformers/
deepface/whisper) off the web request path — that cutover is a separate,
runtime-verified step (cold-start RSS < 500 MB gate); see REFACTOR_PLAN
"Phase 3a" and the cutover plan there.

IMPORTANT — lazy imports: ``process_video``, ``search`` and ``search_images``
import torch/transformers at module top, so every heavy import here is done
*inside* the subcommand that needs it. Importing this module and running
``--help`` therefore stays ML-free (a test enforces this), which is the property
the eventual web/ingestion split depends on.

DB paths: resolved via the existing env-driven ``runtime_paths`` machinery, so
exporting the five ``SEARCH_UI_*_DB_PATH`` vars (the single-DB flip) makes this
CLI target the merged DB automatically — no DB flags needed.

Usage:
    python backend/cli.py ingest vod --language E [--redownload]
    python backend/cli.py ingest subtitles --language E
    python backend/cli.py ingest video --natural-key pub-..._VIDEO [--label 720p] [--keep-source]
    python backend/cli.py ingest video --all [--reprocess] [--label 720p]
    python backend/cli.py ingest images --source {publications|web} [--language E]
    python backend/cli.py process subtitles --language E [--keep-html]
    python backend/cli.py reindex embeddings [--language E] [--batch-size 50]
    python backend/cli.py reindex subtitles [--language E]
    python backend/cli.py reindex concepts [--language E] [--batch-size 50]

Run ingestion when the web process is idle (SQLite write locking).
"""

from __future__ import annotations

import argparse
import json
import sys
from types import SimpleNamespace
from typing import Any

# Top-level imports MUST stay ML-free (no process_video/search/search_images here).
from app_config import get_app_config
from runtime_paths import ensure_runtime_dirs


# Optional file path to also write the result dict to (set by --result-json).
# Gives callers (e.g. the web process shelling out to this CLI) a clean JSON
# artifact to read instead of parsing it back out of log-interleaved stdout.
_RESULT_FILE: str | None = None


def _set_result_file(path: str | None) -> None:
    """Set the file path _emit writes the result dict to (None disables it)."""
    global _RESULT_FILE
    _RESULT_FILE = path


def _emit(result: Any) -> int:
    """Print a result as JSON and return an exit code (1 on error status).

    When --result-json was given, also write the dict to that file so a caller
    can read the structured result without scraping stdout (which carries logs).
    """
    if isinstance(result, dict):
        print(json.dumps(result, indent=2, default=str))
        if _RESULT_FILE:
            with open(_RESULT_FILE, "w", encoding="utf-8") as handle:
                json.dump(result, handle, default=str)
        return 1 if result.get("status") == "error" else 0
    if result is not None:
        print(result)
    return 0


# --------------------------------------------------------------------------
# Subcommand handlers. Heavy/ML imports are function-local on purpose.
# --------------------------------------------------------------------------

def _cmd_ingest_vod(args: argparse.Namespace) -> int:
    import jw_api

    vod = jw_api.download_vod(args.language, redownload=args.redownload)
    if not vod:
        return _emit({"status": "error", "message": f"Failed to download VOD for {args.language!r}"})
    media = jw_api.combine_media_info(args.language)
    return _emit(
        {"status": "success", "command": "ingest vod", "language": args.language,
         "media_items": len(media or {})}
    )


def _cmd_ingest_subtitles(args: argparse.Namespace) -> int:
    import jw_api
    import subtitles_download

    media = jw_api.combine_media_info(args.language)
    if not media:
        return _emit({"status": "error", "message": "No catalog found; run `ingest vod` first."})
    result = subtitles_download.process_all_media_for_subtitles(media, args.language)
    return _emit({"status": "success", "command": "ingest subtitles", "language": args.language, **result})


def _cmd_process_subtitles(args: argparse.Namespace) -> int:
    from search import get_search_instance
    from processing_route_helpers import process_subtitles_sync

    cfg = get_app_config()
    runtime = SimpleNamespace(
        subtitles_root=cfg.subtitles_dir,
        json_root=cfg.json_dir,
        search_service=get_search_instance(),
    )
    try:
        result = process_subtitles_sync(runtime, language=args.language, remove_html=not args.keep_html)
    except Exception as exc:  # noqa: BLE001 — translate any error (incl. HTTPException) to a clean CLI message
        return _emit({"status": "error", "message": str(getattr(exc, "detail", exc))})
    return _emit({"status": "success", "command": "process subtitles", "language": args.language, **result})


def _cmd_ingest_video(args: argparse.Namespace) -> int:
    if not args.all and not args.natural_key:
        # validate before the (torch-pulling) import so an arg error stays cheap
        return _emit({"status": "error", "message": "--natural-key is required (or use --all)"})

    import process_video

    delete_after = False if args.keep_source else None
    if args.all:
        result = process_video.process_all_local_videos(
            label=args.label, reprocess=args.reprocess, delete_video_after=delete_after
        )
    else:
        result = process_video.process_video(args.natural_key, args.label, delete_video_after=delete_after)
    return _emit(result)


def _cmd_ingest_images(args: argparse.Namespace) -> int:
    if args.source == "web":
        import web_images_processor

        result = web_images_processor.crawl_web_images(language=args.language)
    else:
        import search_publication_images

        result = search_publication_images.index_all_publication_images()
    return _emit(result)


def _cmd_reindex_embeddings(args: argparse.Namespace) -> int:
    from search import get_search_instance

    result = get_search_instance().rebuild_subtitle_embeddings(
        language=args.language, batch_size=args.batch_size
    )
    return _emit(result)


def _cmd_reindex_subtitles(args: argparse.Namespace) -> int:
    from search import get_search_instance

    result = get_search_instance().rebuild_index(language=args.language)
    return _emit(result)


def _cmd_reindex_concepts(args: argparse.Namespace) -> int:
    from search import get_search_instance

    result = get_search_instance().rebuild_video_concepts(
        language=args.language, batch_size=args.batch_size
    )
    return _emit(result)


# Keep in sync with search_db._init_db's FTS5 CREATE statements (porter).
_FTS_REBUILD_SCHEMA = {
    "subtitles_fts": (
        "CREATE VIRTUAL TABLE subtitles_fts USING fts5("
        "natural_key, language, content, tokenize = 'porter unicode61')"
    ),
    "ad_transcriptions_fts": (
        "CREATE VIRTUAL TABLE ad_transcriptions_fts USING fts5("
        "natural_key, language, content, tokenize = 'porter unicode61')"
    ),
}


def _cmd_backfill_chunk_timestamps(args: argparse.Namespace) -> int:
    """Attach VTT-derived start/end to subtitle_chunks (Q2.2). No re-embed.

    Anchors each chunk's opening words to the cue stream (anchor-or-skip), so a
    chunk is timed only when it can be located — never a wrong guess. Run after
    the VTTs are present (``ingest subtitles``).
    """
    import os

    from chunk_timestamps import compute_chunk_timestamps
    from db_utils import connect_vec_db
    from runtime_paths import get_search_db_path, get_subtitles_dir
    from subtitles_process_vtt import parse_vtt

    lang = args.language
    subroot = get_subtitles_dir()
    conn = connect_vec_db(get_search_db_path())
    videos_timed = chunks_timed = videos_no_vtt = 0
    try:
        cols = {r[1] for r in conn.execute("PRAGMA table_info(subtitle_chunks)")}
        for col in ("start_seconds", "end_seconds"):
            if col not in cols:
                conn.execute(f"ALTER TABLE subtitle_chunks ADD COLUMN {col} REAL")
        conn.commit()

        keys = [
            r[0] for r in conn.execute(
                "SELECT DISTINCT natural_key FROM subtitle_chunks WHERE language = ?",
                (lang,),
            )
        ]
        for nk in keys:
            vtt_path = os.path.join(subroot, lang, f"{nk}.vtt")
            if not os.path.exists(vtt_path):
                videos_no_vtt += 1
                continue
            try:
                with open(vtt_path, "r", encoding="utf-8") as fh:
                    cues = parse_vtt(fh.read())
            except Exception:
                continue
            rows = conn.execute(
                "SELECT chunk_index, chunk_text FROM subtitle_chunks "
                "WHERE natural_key = ? AND language = ? ORDER BY chunk_index",
                (nk, lang),
            ).fetchall()
            timestamps = compute_chunk_timestamps(cues, [r[1] for r in rows])
            timed_here = False
            for (cidx, _text), ts in zip(rows, timestamps):
                if ts is not None:
                    conn.execute(
                        "UPDATE subtitle_chunks SET start_seconds = ?, end_seconds = ? "
                        "WHERE natural_key = ? AND language = ? AND chunk_index = ?",
                        (ts[0], ts[1], nk, lang, cidx),
                    )
                    chunks_timed += 1
                    timed_here = True
            if timed_here:
                videos_timed += 1
        conn.commit()
    finally:
        conn.close()
    return _emit({
        "status": "success",
        "command": "backfill chunk-timestamps",
        "language": lang,
        "videos_timed": videos_timed,
        "chunks_timed": chunks_timed,
        "videos_without_vtt": videos_no_vtt,
    })


def _cmd_reindex_fts(args: argparse.Namespace) -> int:
    """Rebuild the keyword FTS tables in place with the CURRENT schema.

    FTS5 cannot ALTER its tokenizer, so switching to porter stemming needs a
    drop + recreate + repopulate. Content is read back from the existing FTS
    (no re-download, no embeddings touched), so this is fast. ML-free.
    """
    from db_utils import connect_vec_db
    from runtime_paths import get_search_db_path

    conn = connect_vec_db(get_search_db_path())
    rebuilt = {}
    try:
        for table, create_sql in _FTS_REBUILD_SCHEMA.items():
            try:
                rows = conn.execute(
                    f"SELECT natural_key, language, content FROM {table}"
                ).fetchall()
            except Exception:
                continue  # table absent in this DB
            conn.execute(f"DROP TABLE {table}")
            conn.execute(create_sql)
            conn.executemany(
                f"INSERT INTO {table}(natural_key, language, content) VALUES (?, ?, ?)",
                rows,
            )
            conn.commit()
            rebuilt[table] = len(rows)
    finally:
        conn.close()
    return _emit({"status": "success", "command": "reindex fts", "rebuilt": rebuilt})


def _cmd_check_integrity(args: argparse.Namespace) -> int:
    """Report cross-table row-count drift in the search DB (ML-free).

    Surfaces the kind of silent drift the audit found (e.g. embedding rows
    without a matching chunk). Counting vec0 tables needs sqlite-vec loaded but
    no torch, so this stays a lightweight health check.
    """
    from db_utils import connect_vec_db
    from runtime_paths import get_search_db_path

    conn = connect_vec_db(get_search_db_path())
    issues: list[str] = []

    def _count(table: str) -> int | None:
        try:
            return int(conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0])
        except Exception:
            return None

    try:
        chunks = _count("subtitle_chunks")
        chunk_embeddings = _count("subtitle_chunk_embeddings")
        fts_docs = _count("subtitles_fts")
        concepts = _count("video_concepts")
        concept_embeddings = _count("video_concept_embeddings")
        legacy = _count("subtitle_embeddings")

        if chunks is not None and chunk_embeddings is not None and chunks != chunk_embeddings:
            issues.append(
                f"subtitle_chunks ({chunks}) != subtitle_chunk_embeddings "
                f"({chunk_embeddings}) — {abs(chunks - chunk_embeddings)} orphaned rows"
            )
        if concepts is not None and concept_embeddings is not None and concepts != concept_embeddings:
            issues.append(
                f"video_concepts ({concepts}) != video_concept_embeddings "
                f"({concept_embeddings})"
            )
        if legacy:
            issues.append(
                f"legacy mxbai subtitle_embeddings table still present ({legacy} rows) — "
                "retired; drop via migration 0002 when convenient"
            )
    finally:
        conn.close()

    return _emit({
        "ok": not issues,
        "counts": {
            "subtitle_chunks": chunks,
            "subtitle_chunk_embeddings": chunk_embeddings,
            "subtitles_fts": fts_docs,
            "video_concepts": concepts,
            "video_concept_embeddings": concept_embeddings,
            "legacy_subtitle_embeddings": legacy,
        },
        "issues": issues,
    })


def build_parser() -> argparse.ArgumentParser:
    """Build the CLI parser. Importing/parsing pulls in no ML."""
    parser = argparse.ArgumentParser(prog="jws", description=__doc__.split("\n", 1)[0])
    parser.add_argument(
        "--result-json",
        dest="result_json",
        default=None,
        metavar="PATH",
        help="Also write the result dict as JSON to PATH (clean IPC for callers).",
    )
    top = parser.add_subparsers(dest="command", required=True)

    # ingest <noun>
    ingest = top.add_parser("ingest", help="Download/process content into the indexes.")
    ingest_sub = ingest.add_subparsers(dest="noun", required=True)

    p = ingest_sub.add_parser("vod", help="Download VOD catalog metadata for a language.")
    p.add_argument("--language", default="E")
    p.add_argument("--redownload", action="store_true")
    p.set_defaults(func=_cmd_ingest_vod)

    p = ingest_sub.add_parser("subtitles", help="Download VTT subtitles for the catalog.")
    p.add_argument("--language", default="E")
    p.set_defaults(func=_cmd_ingest_subtitles)

    p = ingest_sub.add_parser("video", help="Process local video(s): thumbnails + ML indexing.")
    p.add_argument("--natural-key", dest="natural_key", default=None)
    p.add_argument("--label", default="720p")
    p.add_argument("--all", action="store_true", help="Process all local videos.")
    p.add_argument("--reprocess", action="store_true", help="With --all, reprocess already-done videos.")
    p.add_argument("--keep-source", action="store_true", help="Do not delete the source MP4 after processing.")
    p.set_defaults(func=_cmd_ingest_video)

    p = ingest_sub.add_parser("images", help="Index images (publication or crawled web images).")
    p.add_argument("--source", choices=["publications", "web"], default="publications")
    p.add_argument("--language", default="E")
    p.set_defaults(func=_cmd_ingest_images)

    # process <noun>
    process = top.add_parser("process", help="Transform already-downloaded content.")
    process_sub = process.add_subparsers(dest="noun", required=True)

    p = process_sub.add_parser("subtitles", help="Convert downloaded VTT -> text and index it.")
    p.add_argument("--language", default="E")
    p.add_argument("--keep-html", action="store_true", help="Do not strip HTML tags from subtitle text.")
    p.set_defaults(func=_cmd_process_subtitles)

    # reindex <noun>
    reindex = top.add_parser("reindex", help="Rebuild search indexes from stored content.")
    reindex_sub = reindex.add_subparsers(dest="noun", required=True)

    p = reindex_sub.add_parser("embeddings", help="Rebuild subtitle chunk embeddings.")
    p.add_argument("--language", default=None)
    p.add_argument("--batch-size", dest="batch_size", type=int, default=50)
    p.set_defaults(func=_cmd_reindex_embeddings)

    p = reindex_sub.add_parser("subtitles", help="Rebuild the subtitle FTS/index.")
    p.add_argument("--language", default=None)
    p.set_defaults(func=_cmd_reindex_subtitles)

    p = reindex_sub.add_parser(
        "concepts",
        help="Rebuild the per-video concept layer (blends VLM scene summaries).",
    )
    p.add_argument("--language", default=None)
    p.add_argument("--batch-size", dest="batch_size", type=int, default=50)
    p.set_defaults(func=_cmd_reindex_concepts)

    p = reindex_sub.add_parser(
        "fts",
        help="Rebuild the keyword FTS tables with the current tokenizer (porter).",
    )
    p.set_defaults(func=_cmd_reindex_fts)

    # backfill <noun>
    backfill = top.add_parser("backfill", help="Backfill derived columns on existing data.")
    backfill_sub = backfill.add_subparsers(dest="noun", required=True)
    p = backfill_sub.add_parser(
        "chunk-timestamps",
        help="Attach VTT start/end times to subtitle_chunks (Q2.2, no re-embed).",
    )
    p.add_argument("--language", default="E")
    p.set_defaults(func=_cmd_backfill_chunk_timestamps)

    # check <noun>
    check = top.add_parser("check", help="Health/integrity checks on the indexes.")
    check_sub = check.add_subparsers(dest="noun", required=True)
    p = check_sub.add_parser("integrity", help="Report cross-table row-count drift.")
    p.set_defaults(func=_cmd_check_integrity)

    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    _set_result_file(getattr(args, "result_json", None))
    # Resolve runtime dirs once (env-driven; no ML, no service construction).
    ensure_runtime_dirs(get_app_config())
    return args.func(args)


if __name__ == "__main__":
    raise SystemExit(main())