#!/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 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 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 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 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 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())