From 8282213688cf30a1a778479cafea1a202eacd152 Mon Sep 17 00:00:00 2001 From: Hasnain Ibrar Date: Tue, 8 Sep 2026 19:10:30 +0300 Subject: [PATCH] feat(index): add bulk indexing for many media items `vidxp index create` handled one media item per invocation, so indexing a repository meant driving it once per video and tracking the results by hand. Add a transport-neutral planning operation and a thin CLI adapter over it: - `Application.plan_bulk_index` resolves a selection to per-media targets and decides which ones the active snapshot already covers. The plan is read-only, so callers can show it before committing to any work. - `vidxp index bulk` indexes every registered media item, or a selection passed with repeated `--media-id`. `--plan-only` shows the decision without indexing, and `--reindex` plans covered media anyway. Media is skipped when the active snapshot holds a generation for it, that generation covers every requested modality, and its recorded input checksum still matches the registered media. Replacing a video's content or asking for a modality the generation lacks therefore plans it again. Media that is not in the ready state is reported as skipped rather than silently dropped. No new indexing behavior. Each pending target is submitted through the existing `submit_index` durable job, one job per media item, matching how `IngestionCoordinator` already sequences ingestion. That is what gives the batch its guarantees: a failure isolates to its own media, earlier successes stay committed, and rerunning the command retries only what is still missing because completed media is then skipped. The command exits non-zero when any media failed. Co-Authored-By: Claude Opus 5 (1M context) --- INSTALLATION_GUIDE.md | 15 ++ src/vidxp/application.py | 93 ++++++++++- src/vidxp/application_models.py | 79 +++++++++ src/vidxp/cli_commands/index.py | 169 +++++++++++++++++++ tests/test_bulk_index.py | 285 ++++++++++++++++++++++++++++++++ 5 files changed, 640 insertions(+), 1 deletion(-) create mode 100644 tests/test_bulk_index.py diff --git a/INSTALLATION_GUIDE.md b/INSTALLATION_GUIDE.md index a35ba255..490fd0ac 100644 --- a/INSTALLATION_GUIDE.md +++ b/INSTALLATION_GUIDE.md @@ -245,6 +245,21 @@ vidxp search speech "the bread just came out of the oven" Add `--media-id ` to a search command to restrict results to one video. Without it, VidXP searches all indexed videos in the active repository. +### Index more than one video + +Index every registered video that the active index does not already cover: + +```bash +vidxp index bulk --modality scene +``` + +Videos the active index already covers are skipped, so the command is safe to +repeat after importing more. Add `--media-id ` once per video to index +a specific selection, `--plan-only` to see what would be indexed and skipped +without indexing, and `--reindex` to index covered videos again. A video that +fails does not stop the rest; rerun the command to retry only what is still +missing. + ### Start an installed interface | Interface | Command | diff --git a/src/vidxp/application.py b/src/vidxp/application.py index afcf735b..a52651b1 100644 --- a/src/vidxp/application.py +++ b/src/vidxp/application.py @@ -32,6 +32,12 @@ PrepareModelsCommand, PrepareModelsResult, RemoveIndexCommand, + BulkIndexPlan, + BulkIndexSkipReason, + BulkIndexTarget, + BulkIndexTargetState, + ListMediaCommand, + PlanBulkIndexCommand, ResourceNotFoundError, RuntimeReadiness, QueryAnswer, @@ -56,10 +62,11 @@ from vidxp.capabilities.registry import CapabilityRegistry from vidxp.capability_service import CapabilityService from vidxp.capabilities.schemas import SearchResult +from vidxp.core.media import MediaState from vidxp.core.contracts import ( IndexConfig, ) -from vidxp.core.snapshots import IndexSnapshot +from vidxp.core.snapshots import GenerationReference, IndexSnapshot from vidxp.execution import ExecutionContext, execution_context from vidxp.ports import IndexBackend, ModelRuntimePort, QueryModelPort from vidxp.query_service import GroundedQueryService @@ -275,6 +282,90 @@ def create_index( ) return IndexResult.model_validate(result) + @application_boundary + def plan_bulk_index(self, command: PlanBulkIndexCommand) -> BulkIndexPlan: + """Decide which registered media still need indexing. + + The plan is a read-only decision. Callers submit one ordinary indexing + operation per pending target, so a failure isolates to its own media + and can be retried without disturbing the rest of the selection. + """ + + selected = self.registry.validate_names(command.modalities) + non_indexable = [ + name for name in selected if self.registry.get(name).collection_name is None + ] + if non_indexable: + raise CapabilityRequestError( + "One or more selected capabilities do not support indexing." + ) + snapshot = self._read_active_snapshot() + generations = {} if snapshot is None else snapshot.generations + requested = frozenset(selected) + targets = tuple( + self._bulk_index_target( + asset, + generation=generations.get(asset.media_id), + requested=requested, + reindex=command.reindex, + ) + for asset in self._bulk_index_selection(command.media_ids) + ) + return BulkIndexPlan(targets=targets, modalities=selected) + + def _bulk_index_selection( + self, + media_ids: tuple[str, ...], + ) -> tuple[MediaAsset, ...]: + if media_ids: + return tuple(self.get_media(media_id) for media_id in media_ids) + assets: list[MediaAsset] = [] + cursor: str | None = None + while True: + page = self.list_media( + ListMediaCommand(page_size=100, cursor=cursor) + ) + assets.extend(page.items) + cursor = page.next_cursor + if cursor is None: + break + return tuple(assets) + + @staticmethod + def _bulk_index_target( + asset: MediaAsset, + *, + generation: GenerationReference | None, + requested: frozenset[str], + reindex: bool, + ) -> BulkIndexTarget: + if asset.state != MediaState.ready: + return BulkIndexTarget( + media_id=asset.media_id, + original_filename=asset.original_filename, + state=BulkIndexTargetState.skipped, + reason=BulkIndexSkipReason.media_not_ready, + ) + covered = ( + not reindex + and generation is not None + and requested <= frozenset(generation.modalities) + and generation.input_sha256 == asset.sha256 + ) + if covered and generation is not None: + return BulkIndexTarget( + media_id=asset.media_id, + original_filename=asset.original_filename, + state=BulkIndexTargetState.skipped, + reason=BulkIndexSkipReason.already_indexed, + generation_id=generation.generation_id, + ) + return BulkIndexTarget( + media_id=asset.media_id, + original_filename=asset.original_filename, + state=BulkIndexTargetState.pending, + ) + @application_boundary def indexing_in_progress(self) -> bool: return self.index_backend.indexing_in_progress(self._base_config()) diff --git a/src/vidxp/application_models.py b/src/vidxp/application_models.py index 395bde42..249dc755 100644 --- a/src/vidxp/application_models.py +++ b/src/vidxp/application_models.py @@ -642,6 +642,85 @@ class IndexResult(ApplicationModel): record_counts: dict[str, NonNegativeInt] = Field(default_factory=dict) +class BulkIndexTargetState(StrEnum): + pending = "pending" + skipped = "skipped" + + +class BulkIndexSkipReason(StrEnum): + already_indexed = "already_indexed" + media_not_ready = "media_not_ready" + + +class BulkIndexTarget(ApplicationModel): + media_id: MediaId + original_filename: str = Field(min_length=1) + state: BulkIndexTargetState + reason: BulkIndexSkipReason | None = Field( + default=None, + description="Why the media was skipped. Absent for pending targets.", + ) + generation_id: IndexGenerationId | None = Field( + default=None, + description=( + "Generation already covering this media in the active snapshot." + ), + ) + + @model_validator(mode="after") + def _reason_matches_state(self) -> "BulkIndexTarget": + if self.state == BulkIndexTargetState.skipped and self.reason is None: + raise ValueError("A skipped target requires a reason.") + if self.state == BulkIndexTargetState.pending and self.reason is not None: + raise ValueError("A pending target cannot carry a skip reason.") + return self + + +class PlanBulkIndexCommand(ApplicationModel): + media_ids: tuple[MediaId, ...] = Field( + default=(), + description=( + "Registered media to consider. Empty selects every registered " + "media item in the repository." + ), + ) + modalities: tuple[str, ...] + reindex: bool = Field( + default=False, + description=( + "Plan already-indexed media for indexing instead of skipping it." + ), + ) + + @field_validator("media_ids") + @classmethod + def _unique_media_ids(cls, value: tuple[str, ...]) -> tuple[str, ...]: + if len(set(value)) != len(value): + raise ValueError("media_ids must not repeat a media identifier.") + return value + + +class BulkIndexPlan(ApplicationModel): + targets: tuple[BulkIndexTarget, ...] = () + modalities: tuple[str, ...] + + @property + def pending(self) -> tuple[BulkIndexTarget, ...]: + return tuple( + target + for target in self.targets + if target.state == BulkIndexTargetState.pending + ) + + @property + def skipped(self) -> tuple[BulkIndexTarget, ...]: + return tuple( + target + for target in self.targets + if target.state == BulkIndexTargetState.skipped + ) + + class RemoveIndexCommand(ApplicationModel): media_id: MediaId diff --git a/src/vidxp/cli_commands/index.py b/src/vidxp/cli_commands/index.py index 14eb192a..4b466070 100644 --- a/src/vidxp/cli_commands/index.py +++ b/src/vidxp/cli_commands/index.py @@ -7,7 +7,9 @@ from rich.table import Table from vidxp.application_models import ( + BulkIndexTargetState, CreateIndexCommand, + PlanBulkIndexCommand, RemoveIndexCommand, ) from vidxp.cli_support import ( @@ -154,6 +156,173 @@ def index_create( ) +@app.command("bulk") +def index_bulk( + ctx: typer.Context, + media_ids: Annotated[ + list[str] | None, + typer.Option( + "--media-id", + help=( + "Registered media identifier to index; repeat to select more " + "than one. Omit to select every registered media item." + ), + ), + ] = None, + modalities: Annotated[ + list[str] | None, + typer.Option( + "--modality", + "-m", + help="Modality to index; repeat to select more than one.", + ), + ] = None, + reindex: Annotated[ + bool, + typer.Option( + "--reindex", + help="Index already-indexed media instead of skipping it.", + ), + ] = False, + plan_only: Annotated[ + bool, + typer.Option( + "--plan-only", + help="Show what would be indexed and skipped without indexing.", + ), + ] = False, + json_output: Annotated[ + bool, + typer.Option("--json", help="Emit machine-readable JSON."), + ] = False, +) -> None: + """Index many media items, skipping those the active snapshot covers.""" + + 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) + plan = state.service.plan_bulk_index( + PlanBulkIndexCommand( + media_ids=tuple(media_ids or ()), + modalities=selected, + reindex=reindex, + ) + ) + output_format = effective_output_format(state, json_output) + pending = plan.pending + outcomes: list[dict] = [ + { + "media_id": target.media_id, + "original_filename": target.original_filename, + "state": target.state.value, + "reason": None if target.reason is None else target.reason.value, + "generation_id": target.generation_id, + "job_id": None, + } + for target in plan.skipped + ] + + if plan_only: + outcomes.extend( + { + "media_id": target.media_id, + "original_filename": target.original_filename, + "state": target.state.value, + "reason": None, + "generation_id": None, + "job_id": None, + } + for target in pending + ) + else: + for position, target in enumerate(pending, start=1): + if output_format == OutputFormat.rich and not state.quiet: + typer.echo( + f"[{position}/{len(pending)}] Indexing " + f"{target.original_filename} ({target.media_id})." + ) + entry = { + "media_id": target.media_id, + "original_filename": target.original_filename, + "state": "indexed", + "reason": None, + "generation_id": None, + "job_id": None, + } + try: + job = state.jobs.submit_index( + CreateIndexCommand( + media_id=target.media_id, + modalities=plan.modalities, + ) + ) + entry["job_id"] = job.job_id + completed = state.jobs.wait(job.job_id) + entry["job_id"] = completed.job_id + # One media item failing must not abandon the rest of the + # batch, so every error is recorded and the loop continues. + except Exception as exc: + entry["state"] = "failed" + entry["reason"] = str(exc) + outcomes.append(entry) + + failed = [entry for entry in outcomes if entry["state"] == "failed"] + payload = { + "planned": len(pending), + "skipped": len(plan.skipped), + "indexed": len( + [entry for entry in outcomes if entry["state"] == "indexed"] + ), + "failed": len(failed), + "plan_only": plan_only, + "modalities": list(plan.modalities), + "items": outcomes, + } + if output_format == OutputFormat.json: + emit_json(payload) + else: + table = Table( + title="Planned media" if plan_only else "Bulk indexing results" + ) + table.add_column("Filename") + table.add_column("Outcome") + table.add_column("Detail") + for target in plan.skipped: + table.add_row( + target.original_filename, + "skipped", + "" if target.reason is None else target.reason.value, + ) + if plan_only: + for target in pending: + table.add_row(target.original_filename, "would index", "") + else: + for entry in outcomes: + if entry["state"] == BulkIndexTargetState.skipped.value: + continue + table.add_row( + entry["original_filename"], + entry["state"], + entry["reason"] or entry["job_id"] or "", + ) + Console().print(table) + typer.echo( + f"Selected {len(plan.targets)} media item(s): " + f"{payload['skipped']} skipped, " + + ( + f"{payload['planned']} would be indexed." + if plan_only + else f"{payload['indexed']} indexed, {payload['failed']} failed." + ) + ) + if failed: + raise typer.Exit(code=1) + + @app.command("remove") def index_remove( ctx: typer.Context, diff --git a/tests/test_bulk_index.py b/tests/test_bulk_index.py new file mode 100644 index 00000000..f3c6633e --- /dev/null +++ b/tests/test_bulk_index.py @@ -0,0 +1,285 @@ +import unittest +from datetime import datetime, timezone +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import Mock + +from vidxp.application import VidXPApplication +from vidxp.application_models import ( + BulkIndexSkipReason, + BulkIndexTargetState, + InvalidRequestError, + MediaAsset, + MediaPage, + PlanBulkIndexCommand, +) +from vidxp.capabilities.registry import create_capability_registry +from vidxp.core.media import MediaState, MediaStream +from vidxp.core.snapshots import GenerationReference, IndexSnapshot +from vidxp.runtime import ModelRuntime +from vidxp.repository_layout import RepositoryLayout +from vidxp.settings import VidXPSettings + + +FIRST_MEDIA_ID = "123456781234423481234567890abcde" +SECOND_MEDIA_ID = "223456781234423481234567890abcde" +GENERATION_ID = "323456781234423481234567890abcde" +SNAPSHOT_ID = "423456781234423481234567890abcde" +FIRST_SHA256 = "a" * 64 +SECOND_SHA256 = "b" * 64 +CONFIG_FINGERPRINT = "c" * 64 +MANIFEST_SHA256 = "d" * 64 + + +def media_asset( + media_id: str, + *, + sha256: str = FIRST_SHA256, + filename: str = "clip.mp4", + state: MediaState = MediaState.ready, +) -> MediaAsset: + return MediaAsset( + media_id=media_id, + video_id=media_id, + original_filename=filename, + sha256=sha256, + byte_size=1024, + detected_mime_type="video/mp4", + container="mov,mp4,m4a,3gp,3g2,mj2", + duration_seconds=3.0, + streams=(MediaStream(index=0, kind="video", codec="h264"),), + state=state, + created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + ) + + +def generation( + media_id: str, + *, + input_sha256: str = FIRST_SHA256, + modalities: tuple[str, ...] = ("scene",), +) -> GenerationReference: + return GenerationReference( + generation_id=GENERATION_ID, + media_id=media_id, + manifest_sha256=MANIFEST_SHA256, + input_sha256=input_sha256, + config_fingerprint=CONFIG_FINGERPRINT, + modalities=modalities, + record_counts={name: 1 for name in modalities}, + store_size_bytes_at_commit=2048, + ) + + +def snapshot(*references: GenerationReference) -> IndexSnapshot: + return IndexSnapshot( + snapshot_id=SNAPSHOT_ID, + created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + config_fingerprint=CONFIG_FINGERPRINT, + configuration={"enabled_modalities": ["scene"]}, + generations={ + reference.media_id: reference for reference in references + }, + ) + + +class BulkIndexPlanTests(unittest.TestCase): + def application( + self, + root: str | Path, + *, + assets: tuple[MediaAsset, ...], + active_snapshot: IndexSnapshot | None = None, + page_size: int | None = None, + ) -> VidXPApplication: + settings = VidXPSettings( + repository_root=Path(root), + runtime_backend="cpu", + ) + media_service = Mock() + by_id = {asset.media_id: asset for asset in assets} + media_service.get.side_effect = lambda media_id: by_id[media_id] + + def list_media(command): + if page_size is None: + return MediaPage(items=assets, total=len(assets)) + start = 0 if command.cursor is None else int(command.cursor) + window = assets[start : start + page_size] + following = start + page_size + return MediaPage( + items=window, + total=len(assets), + next_cursor=( + str(following) if following < len(assets) else None + ), + ) + + media_service.list.side_effect = list_media + return VidXPApplication( + settings=settings, + layout=RepositoryLayout(root=Path(root)), + registry=create_capability_registry(), + runtime=ModelRuntime(settings), + index_backend=Mock(), + media=media_service, + artifacts=Mock(), + index_status=lambda: None, + active_snapshot=lambda: active_snapshot, + ) + + def test_media_covered_by_the_active_snapshot_is_skipped(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID),), + active_snapshot=snapshot(generation(FIRST_MEDIA_ID)), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",)) + ) + self.assertEqual(len(plan.skipped), 1) + self.assertEqual(plan.pending, ()) + skipped = plan.skipped[0] + self.assertEqual(skipped.reason, BulkIndexSkipReason.already_indexed) + self.assertEqual(skipped.generation_id, GENERATION_ID) + + def test_media_missing_a_requested_modality_is_planned(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID),), + active_snapshot=snapshot( + generation(FIRST_MEDIA_ID, modalities=("scene",)) + ), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene", "speech")) + ) + self.assertEqual(len(plan.pending), 1) + self.assertEqual(plan.skipped, ()) + + def test_replaced_media_content_is_planned_again(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID, sha256=SECOND_SHA256),), + active_snapshot=snapshot( + generation(FIRST_MEDIA_ID, input_sha256=FIRST_SHA256) + ), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",)) + ) + self.assertEqual(len(plan.pending), 1) + self.assertEqual(plan.skipped, ()) + + def test_reindex_plans_media_the_snapshot_already_covers(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID),), + active_snapshot=snapshot(generation(FIRST_MEDIA_ID)), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",), reindex=True) + ) + self.assertEqual(len(plan.pending), 1) + self.assertEqual(plan.skipped, ()) + + def test_media_that_is_not_ready_is_skipped(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=( + media_asset(FIRST_MEDIA_ID, state=MediaState.pending), + ), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",)) + ) + self.assertEqual(len(plan.skipped), 1) + self.assertEqual( + plan.skipped[0].reason, + BulkIndexSkipReason.media_not_ready, + ) + + def test_an_empty_selection_covers_every_registered_media_item(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=( + media_asset(FIRST_MEDIA_ID, filename="one.mp4"), + media_asset(SECOND_MEDIA_ID, filename="two.mp4"), + ), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",)) + ) + self.assertEqual(len(plan.targets), 2) + self.assertEqual( + [target.state for target in plan.targets], + [BulkIndexTargetState.pending] * 2, + ) + + def test_an_explicit_selection_ignores_other_media(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=( + media_asset(FIRST_MEDIA_ID, filename="one.mp4"), + media_asset(SECOND_MEDIA_ID, filename="two.mp4"), + ), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand( + media_ids=(SECOND_MEDIA_ID,), + modalities=("scene",), + ) + ) + self.assertEqual(len(plan.targets), 1) + self.assertEqual(plan.targets[0].media_id, SECOND_MEDIA_ID) + + def test_every_page_of_registered_media_is_selected(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=( + media_asset(FIRST_MEDIA_ID, filename="one.mp4"), + media_asset(SECOND_MEDIA_ID, filename="two.mp4"), + ), + page_size=1, + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",)) + ) + self.assertEqual( + [target.media_id for target in plan.targets], + [FIRST_MEDIA_ID, SECOND_MEDIA_ID], + ) + + def test_an_unknown_capability_is_rejected(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID),), + ) + with self.assertRaises(InvalidRequestError): + application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("nonexistent",)) + ) + + def test_planning_reads_no_media_when_the_selection_is_rejected(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID),), + ) + with self.assertRaises(InvalidRequestError): + application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("nonexistent",)) + ) + application.media.list.assert_not_called() + + +if __name__ == "__main__": + unittest.main()