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()