From 0153adb9ac69e49749c040c3d95fcc4dad279e1f Mon Sep 17 00:00:00 2001 From: umair Date: Tue, 8 Sep 2026 13:53:48 +0500 Subject: [PATCH] feat(indexing): add bulk video indexing command and coordinator --- src/vidxp/bulk_indexing.py | 201 +++++++++ src/vidxp/cli_commands/index.py | 178 +++++++- tests/test_bulk_indexing.py | 697 ++++++++++++++++++++++++++++++++ 3 files changed, 1075 insertions(+), 1 deletion(-) create mode 100644 src/vidxp/bulk_indexing.py create mode 100644 tests/test_bulk_indexing.py diff --git a/src/vidxp/bulk_indexing.py b/src/vidxp/bulk_indexing.py new file mode 100644 index 00000000..6cbe5e2b --- /dev/null +++ b/src/vidxp/bulk_indexing.py @@ -0,0 +1,201 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Callable, Sequence + +from vidxp.application_models import ( + ApplicationError, + CreateIndexCommand, + ListMediaCommand, + MediaAsset, + MediaState, +) +from vidxp.core.snapshots import IndexSnapshot + +if TYPE_CHECKING: + from vidxp.application import VidXPApplication + from vidxp.control_plane import ControlPlaneApplication + from vidxp.job_service import JobService + + +@dataclass(frozen=True) +class BulkIndexItemResult: + media_id: str + filename: str + status: str + job_id: str | None = None + error_code: str | None = None + error_message: str | None = None + + +@dataclass(frozen=True) +class BulkIndexSummary: + total: int + indexed: int + skipped: int + failed: int + queued: int = 0 + results: tuple[BulkIndexItemResult, ...] = () + + +def _resolve_all_media( + application: VidXPApplication | ControlPlaneApplication, +) -> list[MediaAsset]: + media_list: list[MediaAsset] = [] + cursor: str | None = None + while True: + page = application.list_media( + ListMediaCommand( + page_size=100, + cursor=cursor, + state=MediaState.ready, + ) + ) + media_list.extend(page.items) + if not page.next_cursor or not page.items: + break + cursor = page.next_cursor + return media_list + + +def _is_already_indexed( + snapshot: IndexSnapshot | None, + media_id: str, + requested_modalities: Sequence[str] | None = None, +) -> bool: + if snapshot is None: + return False + generation = snapshot.generations.get(media_id) + if generation is None: + return False + if requested_modalities is not None: + return set(requested_modalities).issubset(set(generation.modalities)) + return True + + +def run_bulk_index( + application: VidXPApplication | ControlPlaneApplication, + jobs: JobService, + media_ids: Sequence[str] | None = None, + *, + all_eligible: bool = False, + skip_indexed: bool = True, + detach: bool = False, + modalities: Sequence[str] | None = None, + frame_stride: int = 1, + scene_sample_fps: float | None = None, + capability_options: dict[str, dict] | None = None, + on_item_start: Callable[[str, str], None] | None = None, + on_item_progress: Callable[[str, Any], None] | None = None, + on_item_complete: Callable[[BulkIndexItemResult], None] | None = None, +) -> BulkIndexSummary: + items: list[tuple[str, str]] = [] + if all_eligible: + all_media = _resolve_all_media(application) + items = [(asset.media_id, asset.original_filename) for asset in all_media] + elif media_ids: + for mid in media_ids: + asset = application.get_media(mid) + items.append((asset.media_id, asset.original_filename)) + + read_snapshot = getattr(application, "_read_active_snapshot", None) + snapshot: IndexSnapshot | None = ( + read_snapshot() if callable(read_snapshot) else None + ) + + if modalities is not None: + cmd_modalities = tuple(modalities) + elif hasattr(application, "select_index_modalities"): + cmd_modalities = application.select_index_modalities(None) + elif hasattr(application, "list_capabilities"): + cmd_modalities = tuple( + c.name + for c in application.list_capabilities() + if getattr(c, "supports_indexing", True) + ) + else: + cmd_modalities = () + + results: list[BulkIndexItemResult] = [] + + for media_id, filename in items: + if skip_indexed and _is_already_indexed(snapshot, media_id, modalities): + item_result = BulkIndexItemResult( + media_id=media_id, + filename=filename, + status="skipped", + ) + results.append(item_result) + if on_item_complete is not None: + on_item_complete(item_result) + continue + + if on_item_start is not None: + on_item_start(media_id, filename) + + command = CreateIndexCommand( + media_id=media_id, + modalities=cmd_modalities, + frame_stride=frame_stride, + scene_sample_fps=scene_sample_fps, + capability_options=capability_options or {}, + ) + + try: + job = jobs.submit_index(command) + if detach: + item_result = BulkIndexItemResult( + media_id=media_id, + filename=filename, + status="queued", + job_id=job.job_id, + ) + results.append(item_result) + if on_item_complete is not None: + on_item_complete(item_result) + else: + def _progress(current: Any) -> None: + if on_item_progress is not None: + on_item_progress(media_id, current) + + job = jobs.wait(job.job_id, progress=_progress) + item_result = BulkIndexItemResult( + media_id=media_id, + filename=filename, + status="indexed", + job_id=job.job_id, + ) + results.append(item_result) + if on_item_complete is not None: + on_item_complete(item_result) + except ApplicationError as exc: + item_result = BulkIndexItemResult( + media_id=media_id, + filename=filename, + status="failed", + error_code=exc.code, + error_message=str(exc), + ) + results.append(item_result) + if on_item_complete is not None: + on_item_complete(item_result) + except Exception as exc: + item_result = BulkIndexItemResult( + media_id=media_id, + filename=filename, + status="failed", + error_code="unexpected_error", + error_message=str(exc), + ) + results.append(item_result) + if on_item_complete is not None: + on_item_complete(item_result) + + return BulkIndexSummary( + total=len(results), + indexed=sum(1 for r in results if r.status == "indexed"), + skipped=sum(1 for r in results if r.status == "skipped"), + failed=sum(1 for r in results if r.status == "failed"), + queued=sum(1 for r in results if r.status == "queued"), + results=tuple(results), + ) diff --git a/src/vidxp/cli_commands/index.py b/src/vidxp/cli_commands/index.py index 14eb192a..af23423d 100644 --- a/src/vidxp/cli_commands/index.py +++ b/src/vidxp/cli_commands/index.py @@ -1,15 +1,17 @@ from __future__ import annotations -from typing import Annotated, Iterable +from typing import Annotated, Any, Iterable import typer from rich.console import Console +from rich.markup import escape from rich.table import Table from vidxp.application_models import ( CreateIndexCommand, RemoveIndexCommand, ) +from vidxp.bulk_indexing import run_bulk_index from vidxp.cli_support import ( CLIState, IndexProgress, @@ -154,6 +156,180 @@ def index_create( ) +@app.command("bulk") +def index_bulk( + ctx: typer.Context, + media_ids: Annotated[ + list[str] | None, + typer.Argument( + help="Registered media identifiers to index.", + ), + ] = None, + all_eligible: Annotated[ + bool, + typer.Option( + "--all", + help="Index all eligible registered media in the catalog.", + ), + ] = False, + reindex: Annotated[ + bool, + typer.Option( + "--reindex", + help="Reindex media even if already present in the active index.", + ), + ] = False, + modalities: Annotated[ + list[str] | None, + typer.Option( + "--modality", + "-m", + help="Modality to index; repeat to select more than one.", + ), + ] = None, + frame_stride: Annotated[ + int, + typer.Option( + "--frame-stride", + min=1, + help=( + "Materialize every Nth frame for actor and legacy visual " + "indexing." + ), + ), + ] = 1, + scene_sample_fps: Annotated[ + float | None, + typer.Option( + "--scene-sample-fps", + min=0.01, + help=( + "Target scene samples per second; lower-FPS media uses every " + "available frame." + ), + ), + ] = None, + capability_options: Annotated[ + list[str] | None, + typer.Option( + "--option", + help=( + "Capability setting as CAPABILITY.KEY=VALUE; " + "repeat for multiple settings." + ), + ), + ] = None, + detach: Annotated[ + bool, + typer.Option( + "--detach", + help="Return after the durable job is queued.", + ), + ] = False, + json_output: Annotated[ + bool, + typer.Option("--json", help="Emit machine-readable JSON."), + ] = False, +) -> None: + """Index multiple media items or all eligible media in the catalog.""" + + if not media_ids and not all_eligible: + raise typer.BadParameter("Provide either media IDs or pass --all.") + if media_ids and all_eligible: + raise typer.BadParameter("Pass media IDs or --all, not both.") + + state = state_from_context(ctx) + indexable = tuple( + capability.name + for capability in state.service.list_capabilities() + if capability.supports_indexing + ) + selected = selected_modalities(modalities, indexable) + parsed_options = parse_capability_options(capability_options) + output_fmt = effective_output_format(state, json_output) + show_progress = not state.quiet and output_fmt == OutputFormat.rich + + with IndexProgress(show_progress) as progress: + def on_item_start(media_id: str, filename: str) -> None: + if show_progress: + progress.update({ + "stage": "indexing", + "message": f"Indexing {filename} ({media_id[:8]}...)", + }) + + def on_item_progress(media_id: str, current: Any) -> None: + if show_progress: + if hasattr(current, "progress") and current.progress is not None: + progress.update(current.progress.model_dump(mode="python")) + elif isinstance(current, dict): + progress.update(current) + + summary = run_bulk_index( + application=state.service, + jobs=state.jobs, + media_ids=media_ids, + all_eligible=all_eligible, + skip_indexed=not reindex, + detach=detach, + modalities=selected, + frame_stride=frame_stride, + scene_sample_fps=scene_sample_fps, + capability_options=parsed_options, + on_item_start=on_item_start, + on_item_progress=on_item_progress, + ) + + if output_fmt == OutputFormat.json: + payload = { + "total": summary.total, + "indexed": summary.indexed, + "skipped": summary.skipped, + "failed": summary.failed, + "queued": summary.queued, + "results": [ + { + "media_id": r.media_id, + "filename": r.filename, + "status": r.status, + "job_id": r.job_id, + "error_code": r.error_code, + "error_message": r.error_message, + } + for r in summary.results + ], + } + emit_json(payload) + else: + table = Table(title="Bulk indexing summary") + table.add_column("Media ID") + table.add_column("Filename") + table.add_column("Status") + table.add_column("Job ID") + table.add_column("Error") + for r in summary.results: + error_str = ( + f"[{r.error_code}] {r.error_message}" + if r.error_code + else (r.error_message or "—") + ) + table.add_row( + escape(r.media_id), + escape(r.filename), + escape(r.status), + escape(r.job_id or "—"), + escape(error_str), + ) + Console().print(table) + typer.echo( + f"Total: {summary.total}, Indexed: {summary.indexed}, " + f"Skipped: {summary.skipped}, Failed: {summary.failed}, " + f"Queued: {summary.queued}." + ) + + if summary.failed > 0: + raise typer.Exit(code=1) + + @app.command("remove") def index_remove( ctx: typer.Context, diff --git a/tests/test_bulk_indexing.py b/tests/test_bulk_indexing.py new file mode 100644 index 00000000..6a6ba2e6 --- /dev/null +++ b/tests/test_bulk_indexing.py @@ -0,0 +1,697 @@ +from __future__ import annotations + +import json +import unittest +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from unittest.mock import Mock, patch + +from typer.testing import CliRunner + +from vidxp import cli +from vidxp.application_models import ( + ApplicationError, + CreateIndexCommand, + DependencyCheckResult, + ErrorCategory, + IndexJobResult, + IndexResult, + Job, + JobKind, + JobProgress, + JobQueue, + JobState, + ListMediaCommand, + MediaAsset, + MediaPage, + MediaState, + MediaStream, +) +from vidxp.bulk_indexing import ( + BulkIndexItemResult, + BulkIndexSummary, + _is_already_indexed, + _resolve_all_media, + run_bulk_index, +) +from vidxp.capabilities.registry import create_capability_registry +from vidxp.capability_service import CapabilityService +from vidxp.composition import LocalApplicationContext +from vidxp.core.snapshots import GenerationReference, IndexSnapshot +from vidxp.repositories import RepositoryConfig, RepositoryRegistry + +MEDIA_ID_1 = "123456781234423481234567890abcde" +MEDIA_ID_2 = "223456781234423481234567890abcde" +MEDIA_ID_3 = "323456781234423481234567890abcde" +JOB_ID_1 = "423456781234423481234567890abcde" +JOB_ID_2 = "523456781234423481234567890abcde" +SNAPSHOT_ID = "623456781234423481234567890abcde" +GENERATION_ID = "723456781234423481234567890abcde" + + +def make_media(media_id: str, filename: str = "video.mp4") -> MediaAsset: + return MediaAsset( + schema_version=1, + media_id=media_id, + video_id=media_id, + original_filename=filename, + sha256="1" * 64, + byte_size=1024, + detected_mime_type="video/mp4", + container="mp4", + duration_seconds=10.0, + streams=( + MediaStream( + index=0, + kind="video", + codec="h264", + width=640, + height=480, + ), + ), + state=MediaState.ready, + created_at=datetime.now(timezone.utc), + ) + + +def make_job( + job_id: str, + state: JobState = JobState.succeeded, + media_id: str = MEDIA_ID_1, +) -> Job: + result = None + if state == JobState.succeeded: + result = IndexJobResult( + result=IndexResult( + media_id=media_id, + generation_id=GENERATION_ID, + snapshot_id=SNAPSHOT_ID, + active_media_count=1, + record_counts={"scene": 1}, + ) + ) + return Job( + job_id=job_id, + kind=JobKind.index, + state=state, + queue=JobQueue.cpu, + result=result, + progress=JobProgress( + stage="indexing", + current=1, + total=1, + message="Done", + updated_at=datetime.now(timezone.utc), + ), + ) + + +def make_snapshot(generations: dict[str, tuple[str, ...]]) -> IndexSnapshot: + gen_refs: dict[str, GenerationReference] = {} + for mid, modalities in generations.items(): + gen_refs[mid] = GenerationReference( + generation_id=GENERATION_ID, + media_id=mid, + manifest_sha256="1" * 64, + input_sha256="2" * 64, + config_fingerprint="3" * 64, + modalities=modalities, + record_counts={m: 1 for m in modalities}, + store_size_bytes_at_commit=1024, + ) + return IndexSnapshot( + schema_version=1, + snapshot_id=SNAPSHOT_ID, + created_at=datetime.now(timezone.utc), + config_fingerprint="0" * 64, + configuration={}, + generations=gen_refs, + ) + + +class BulkIndexingHelperTests(unittest.TestCase): + def test_resolve_all_media_paginates_cursor(self): + media_1 = make_media(MEDIA_ID_1, "first.mp4") + media_2 = make_media(MEDIA_ID_2, "second.mp4") + media_3 = make_media(MEDIA_ID_3, "third.mp4") + + app = Mock() + app.list_media.side_effect = [ + MediaPage(items=(media_1, media_2), total=3, next_cursor="c1"), + MediaPage(items=(media_3,), total=3, next_cursor=None), + ] + + result = _resolve_all_media(app) + + self.assertEqual(result, [media_1, media_2, media_3]) + self.assertEqual(app.list_media.call_count, 2) + app.list_media.assert_any_call( + ListMediaCommand( + page_size=100, + cursor=None, + state=MediaState.ready, + ) + ) + app.list_media.assert_any_call( + ListMediaCommand( + page_size=100, + cursor="c1", + state=MediaState.ready, + ) + ) + + def test_resolve_all_media_empty_catalog(self): + app = Mock() + app.list_media.return_value = MediaPage( + items=(), total=0, next_cursor=None + ) + + result = _resolve_all_media(app) + + self.assertEqual(result, []) + self.assertEqual(app.list_media.call_count, 1) + + def test_is_already_indexed_snapshot_none(self): + self.assertFalse(_is_already_indexed(None, MEDIA_ID_1)) + + def test_is_already_indexed_media_not_in_generations(self): + snapshot = make_snapshot({MEDIA_ID_2: ("scene",)}) + self.assertFalse(_is_already_indexed(snapshot, MEDIA_ID_1)) + + def test_is_already_indexed_no_modalities_requested(self): + snapshot = make_snapshot({MEDIA_ID_1: ("scene",)}) + self.assertTrue(_is_already_indexed(snapshot, MEDIA_ID_1, None)) + + def test_is_already_indexed_matching_modalities_subset(self): + snapshot = make_snapshot({MEDIA_ID_1: ("scene", "speech", "actor")}) + self.assertTrue( + _is_already_indexed(snapshot, MEDIA_ID_1, ("scene", "speech")) + ) + + def test_is_already_indexed_missing_requested_modality(self): + snapshot = make_snapshot({MEDIA_ID_1: ("scene",)}) + self.assertFalse( + _is_already_indexed(snapshot, MEDIA_ID_1, ("scene", "speech")) + ) + + +class RunBulkIndexTests(unittest.TestCase): + def setUp(self): + self.app = Mock() + self.jobs = Mock() + self.media_1 = make_media(MEDIA_ID_1, "one.mp4") + self.media_2 = make_media(MEDIA_ID_2, "two.mp4") + self.app.get_media.side_effect = lambda mid: ( + self.media_1 if mid == MEDIA_ID_1 else self.media_2 + ) + self.app._read_active_snapshot.return_value = None + + def test_bulk_index_multiple_media_ids_success(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [job_1, job_2] + + started: list[tuple[str, str]] = [] + completed: list[BulkIndexItemResult] = [] + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1, MEDIA_ID_2], + modalities=["scene"], + on_item_start=lambda mid, fn: started.append((mid, fn)), + on_item_complete=lambda r: completed.append(r), + ) + + self.assertEqual( + summary, + BulkIndexSummary( + total=2, + indexed=2, + skipped=0, + failed=0, + queued=0, + results=( + BulkIndexItemResult( + media_id=MEDIA_ID_1, + filename="one.mp4", + status="indexed", + job_id=JOB_ID_1, + ), + BulkIndexItemResult( + media_id=MEDIA_ID_2, + filename="two.mp4", + status="indexed", + job_id=JOB_ID_2, + ), + ), + ), + ) + self.assertEqual( + started, + [(MEDIA_ID_1, "one.mp4"), (MEDIA_ID_2, "two.mp4")], + ) + self.assertEqual(len(completed), 2) + self.assertEqual(self.jobs.submit_index.call_count, 2) + self.assertEqual(self.jobs.wait.call_count, 2) + + def test_bulk_index_all_eligible_paginates(self): + self.app.list_media.side_effect = [ + MediaPage(items=(self.media_1,), total=2, next_cursor="c1"), + MediaPage(items=(self.media_2,), total=2, next_cursor=None), + ] + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [job_1, job_2] + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + all_eligible=True, + modalities=["scene"], + ) + + self.assertEqual(summary.total, 2) + self.assertEqual(summary.indexed, 2) + self.assertEqual(summary.skipped, 0) + self.assertEqual(summary.failed, 0) + + def test_bulk_index_skips_already_indexed(self): + snapshot = make_snapshot({MEDIA_ID_1: ("scene",)}) + self.app._read_active_snapshot.return_value = snapshot + + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.return_value = job_2 + self.jobs.wait.return_value = job_2 + + started: list[tuple[str, str]] = [] + completed: list[BulkIndexItemResult] = [] + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1, MEDIA_ID_2], + skip_indexed=True, + modalities=["scene"], + on_item_start=lambda mid, fn: started.append((mid, fn)), + on_item_complete=lambda r: completed.append(r), + ) + + self.assertEqual(summary.total, 2) + self.assertEqual(summary.indexed, 1) + self.assertEqual(summary.skipped, 1) + self.assertEqual(summary.failed, 0) + self.assertEqual(summary.results[0].status, "skipped") + self.assertEqual(summary.results[0].media_id, MEDIA_ID_1) + self.assertEqual(summary.results[1].status, "indexed") + self.assertEqual(summary.results[1].media_id, MEDIA_ID_2) + # started should NOT include skipped item + self.assertEqual(started, [(MEDIA_ID_2, "two.mp4")]) + self.assertEqual(len(completed), 2) + self.jobs.submit_index.assert_called_once() + + def test_bulk_index_reindex_flag_overrides_skip(self): + snapshot = make_snapshot({MEDIA_ID_1: ("scene",)}) + self.app._read_active_snapshot.return_value = snapshot + + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.return_value = job_1 + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + skip_indexed=False, + modalities=["scene"], + ) + + self.assertEqual(summary.total, 1) + self.assertEqual(summary.indexed, 1) + self.assertEqual(summary.skipped, 0) + self.assertEqual(summary.results[0].status, "indexed") + + def test_bulk_index_detach_queues_jobs(self): + job_1 = make_job(JOB_ID_1, state=JobState.queued, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + detach=True, + modalities=["scene"], + ) + + self.assertEqual(summary.total, 1) + self.assertEqual(summary.queued, 1) + self.assertEqual(summary.indexed, 0) + self.assertEqual(summary.results[0].status, "queued") + self.assertEqual(summary.results[0].job_id, JOB_ID_1) + self.jobs.wait.assert_not_called() + + def test_bulk_index_error_resilience(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [ + ApplicationError( + "transcription_failed", + ErrorCategory.unavailable, + "Model crashed during transcription.", + ), + job_2, + ] + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1, MEDIA_ID_2], + modalities=["scene"], + ) + + self.assertEqual(summary.total, 2) + self.assertEqual(summary.indexed, 1) + self.assertEqual(summary.failed, 1) + self.assertEqual(summary.skipped, 0) + self.assertEqual(summary.results[0].status, "failed") + self.assertEqual( + summary.results[0].error_code, "transcription_failed" + ) + self.assertEqual( + summary.results[0].error_message, + "Model crashed during transcription.", + ) + self.assertEqual(summary.results[1].status, "indexed") + self.assertEqual(summary.results[1].job_id, JOB_ID_2) + + def test_bulk_index_generic_exception_resilience(self): + self.jobs.submit_index.side_effect = RuntimeError("Disk IO error") + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + modalities=["scene"], + ) + + self.assertEqual(summary.total, 1) + self.assertEqual(summary.failed, 1) + self.assertEqual(summary.results[0].status, "failed") + self.assertEqual(summary.results[0].error_code, "unexpected_error") + self.assertIn("Disk IO error", summary.results[0].error_message or "") + + + def test_bulk_index_forwards_options_and_command_fields(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.return_value = job_1 + + progress_events: list[tuple[str, Any]] = [] + + def fake_wait(job_id, progress=None): + if progress: + progress(job_1) + return job_1 + + self.jobs.wait.side_effect = fake_wait + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + modalities=["scene"], + frame_stride=3, + scene_sample_fps=1.5, + capability_options={"scene": {"threshold": 0.7}}, + on_item_progress=lambda mid, curr: progress_events.append((mid, curr)), + ) + + self.assertEqual(summary.total, 1) + self.assertEqual(summary.indexed, 1) + submitted_command = self.jobs.submit_index.call_args.args[0] + self.assertIsInstance(submitted_command, CreateIndexCommand) + self.assertEqual(submitted_command.media_id, MEDIA_ID_1) + self.assertEqual(submitted_command.modalities, ("scene",)) + self.assertEqual(submitted_command.frame_stride, 3) + self.assertEqual(submitted_command.scene_sample_fps, 1.5) + self.assertEqual( + submitted_command.capability_options, + {"scene": {"threshold": 0.7}}, + ) + self.assertEqual(len(progress_events), 1) + self.assertEqual(progress_events[0][0], MEDIA_ID_1) + + def test_bulk_index_default_modalities_resolves_from_application(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.return_value = job_1 + self.app.select_index_modalities.return_value = ("scene", "speech") + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + modalities=None, + ) + + self.assertEqual(summary.total, 1) + self.assertEqual(summary.indexed, 1) + submitted_command = self.jobs.submit_index.call_args.args[0] + self.assertEqual(submitted_command.modalities, ("scene", "speech")) + + def test_bulk_index_default_modalities_resolves_from_list_capabilities(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.return_value = job_1 + del self.app.select_index_modalities + cap1 = Mock() + cap1.name = "scene" + cap1.supports_indexing = True + cap2 = Mock() + cap2.name = "summary" + cap2.supports_indexing = False + self.app.list_capabilities.return_value = [cap1, cap2] + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + modalities=None, + ) + + self.assertEqual(summary.total, 1) + submitted_command = self.jobs.submit_index.call_args.args[0] + self.assertEqual(submitted_command.modalities, ("scene",)) + + def test_bulk_index_empty_items(self): + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[], + modalities=["scene"], + ) + self.assertEqual(summary.total, 0) + self.assertEqual(summary.indexed, 0) + self.assertEqual(summary.skipped, 0) + self.assertEqual(summary.failed, 0) + self.assertEqual(summary.queued, 0) + self.assertEqual(summary.results, ()) + + +class CliBulkIndexTests(unittest.TestCase): + def setUp(self): + self.runner = CliRunner() + self.service = Mock() + self.service.registry = create_capability_registry() + self.service.list_capabilities.return_value = CapabilityService( + self.service.registry + ).list() + self.service.index_directory = Path("repo/indexes") + self.service.layout.root = Path("repo") + self.service.model_cache = Path("model-cache") + self.service.runtime.backends.requested = "cpu" + self.service.model_readiness.return_value = DependencyCheckResult( + ok=True, + modalities=(), + checks=(), + ) + self.service._read_active_snapshot.return_value = None + + self.media_1 = make_media(MEDIA_ID_1, "one.mp4") + self.media_2 = make_media(MEDIA_ID_2, "two.mp4") + self.service.get_media.side_effect = lambda mid: ( + self.media_1 if mid == MEDIA_ID_1 else self.media_2 + ) + self.service.list_media.return_value = MediaPage( + items=(self.media_1, self.media_2), total=2, next_cursor=None + ) + + self.jobs = Mock() + self.registry = Mock(spec=RepositoryRegistry) + self.registry.path = Path("repositories.json") + self.repository = RepositoryConfig( + "default", + Path("repo"), + device="cpu", + configured=False, + ) + + def invoke(self, arguments, *, media_runtime_initialized=True): + with ( + patch.object( + cli, + "create_local_application", + return_value=LocalApplicationContext( + application=self.service, + jobs=self.jobs, + repositories=self.registry, + repository=self.repository, + ), + ) as create_local_application, + patch( + "vidxp.cli_support.media_runtime_is_initialized", + return_value=media_runtime_initialized, + ), + ): + result = self.runner.invoke(cli.app, arguments) + self.create_local_application = create_local_application + return result + + def test_cli_requires_media_ids_or_all(self): + result = self.invoke(["index", "bulk"]) + self.assertEqual(result.exit_code, 2, result.output) + self.assertIn("Provide either media IDs or pass --all.", result.output) + + def test_cli_rejects_both_media_ids_and_all(self): + result = self.invoke(["index", "bulk", MEDIA_ID_1, "--all"]) + self.assertEqual(result.exit_code, 2, result.output) + self.assertIn("Pass media IDs or --all, not both.", result.output) + + def test_cli_bulk_index_success_json(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [job_1, job_2] + + result = self.invoke( + [ + "index", + "bulk", + MEDIA_ID_1, + MEDIA_ID_2, + "--modality", + "scene", + "--json", + ] + ) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["total"], 2) + self.assertEqual(payload["indexed"], 2) + self.assertEqual(payload["failed"], 0) + self.assertEqual(payload["skipped"], 0) + self.assertEqual(len(payload["results"]), 2) + self.assertEqual(payload["results"][0]["media_id"], MEDIA_ID_1) + self.assertEqual(payload["results"][0]["status"], "indexed") + self.assertEqual(payload["results"][0]["job_id"], JOB_ID_1) + + def test_cli_bulk_index_all_success_table(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [job_1, job_2] + + result = self.invoke(["index", "bulk", "--all", "--modality", "scene"]) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertIn("Bulk indexing summary", result.output) + self.assertIn("Total: 2, Indexed: 2, Skipped: 0, Failed: 0, Queued: 0.", result.output) + + def test_cli_bulk_index_forwards_options_and_flags(self): + job_1 = make_job(JOB_ID_1, state=JobState.queued, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + + result = self.invoke( + [ + "index", + "bulk", + MEDIA_ID_1, + "--modality", + "scene", + "--frame-stride", + "4", + "--scene-sample-fps", + "2.5", + "--option", + "scene.threshold=0.8", + "--detach", + "--reindex", + "--json", + ] + ) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["total"], 1) + self.assertEqual(payload["queued"], 1) + submitted_command = self.jobs.submit_index.call_args.args[0] + self.assertEqual(submitted_command.frame_stride, 4) + self.assertEqual(submitted_command.scene_sample_fps, 2.5) + self.assertEqual( + submitted_command.capability_options, + {"scene": {"threshold": 0.8}}, + ) + + def test_cli_bulk_index_with_failure_exits_code_1(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.side_effect = ApplicationError( + "model_error", + ErrorCategory.unavailable, + "Failed to load model weights.", + ) + + result = self.invoke( + ["index", "bulk", MEDIA_ID_1, "--modality", "scene", "--json"] + ) + + self.assertEqual(result.exit_code, 1, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["total"], 1) + self.assertEqual(payload["failed"], 1) + self.assertEqual(payload["results"][0]["status"], "failed") + self.assertEqual(payload["results"][0]["error_code"], "model_error") + + def test_cli_bulk_index_all_json(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [job_1, job_2] + + result = self.invoke(["index", "bulk", "--all", "--modality", "scene", "--json"]) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["total"], 2) + self.assertEqual(payload["indexed"], 2) + + def test_cli_bulk_index_with_failure_table_output(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.side_effect = ApplicationError( + "model_error", + ErrorCategory.unavailable, + "Failed to load model weights.", + ) + + result = self.invoke(["index", "bulk", MEDIA_ID_1, "--modality", "scene"]) + + self.assertEqual(result.exit_code, 1, result.output) + self.assertIn("Bulk indexing summary", result.output) + self.assertIn("model_error", result.output) + self.assertIn("Failed: 1", result.output)