From a8d1ebdf6fb8bd42ec9704fcd2a3237f5117f92d Mon Sep 17 00:00:00 2001 From: calesthio Date: Sat, 11 Apr 2026 00:46:06 -0700 Subject: [PATCH] docmontage: corpus builder hardening from P1 + P2 audit observations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit All of these changes came out of running the P1 and P2 documentary-montage audits end-to-end and watching specific things break. Grouping them into one commit because they share a theme: making the corpus builder and its stock source adapters robust enough that a real brief can produce a real corpus without special-casing. corpus_builder.py + new clip_cache.py + test_clip_cache.py Phase 1 of the shared-corpus architecture. Adds ~/.openmontage/clips_cache/ — a process-safe, LRU-evicted cache of downloaded clip files keyed by clip_id. Before each candidate download, corpus_builder asks the cache whether the bytes already exist on disk from a previous run; on a hit it hard-links (or copies on cross-drive) the blob into the caller's corpus dir and skips the network fetch. On a miss it downloads as usual and ingests the fresh file. Motivation: re-running the P1 audit after every tool fix was re-downloading gigs of archive.org footage that had already been fetched in the prior run. Cache faults never block the pipeline — they degrade gracefully to normal downloads. The cache bubbles counters into the corpus_builder return payload so the production report can show hit/miss/bytes-saved. Default 20 GB cap, overridable via OPENMONTAGE_CACHE_MAX_GB. Full test coverage: try_link, ingest, stats, LRU eviction, manifest persistence, lock behavior. 23 tests, tmp_path-scoped. stock_sources/archive_org.py 1. Three-strategy query cascade (phrase_prox_10 → distinctive_and → distinctive_or). Motivation: natural-language documentary queries against archive.org Solr were zeroing out — "1950s family watching television" returned 0 hits because Solr's default multi-term AND over-intersects. Walks strict to loose and returns the first non-empty strategy. Stop words, source hints ("prelinger", "archive", "footage"), and year tokens ("1950s") are excluded from the distinctive-token picks since they don't correlate with Prelinger title tokens. 2. 150 MB per-rendition size cap. archive.org routinely hosts multi-hundred-megabyte h.264 masters and one 2 GB Prelinger item poisons corpus build wall-time and disk. Within a format bucket we now pick the largest rendition under the cap; if nothing fits we fall through to the next format rather than dropping the item. 3. 180 s default max-duration ceiling when the caller hasn't set one — archive.org is the only source that routinely hosts feature- length material and a naive fan-out pulls them into corpora that only ever want a few seconds per clip. stock_sources/wikimedia.py Parallel 3-strategy cascade (full → top2_or → single_best). Motivation: Commons CirrusSearch also defaults to AND across multi-word queries; our first P2 diagnostic pass returned 0 video results for 10/10 queries. Same stop-word / source-hint / year-token stripping as archive_org so the two adapters stay symmetric. test_stock_source_adapters.py Rewrote the wikimedia query-builder tests against the new cascade API. Added coverage for multi-word fallback + source-hint/year stripping. video_compose.py Two small fixes for the Remotion renderer on Windows: 1. Resolve output_path to absolute before invoking the CLI so the binary can write wherever the caller asked regardless of cwd. 2. Pass cwd=composer_dir to run_command so npx can find the local Remotion binary under node_modules/.bin. Without this, Windows npx returns "could not determine executable to run" because it resolves .bin relative to the process cwd rather than the script's parent. --- .gitignore | 7 + tests/tools/test_clip_cache.py | 437 +++++++++++++++++ tests/tools/test_stock_source_adapters.py | 51 +- tools/video/clip_cache.py | 572 ++++++++++++++++++++++ tools/video/corpus_builder.py | 92 +++- tools/video/stock_sources/archive_org.py | 262 ++++++++-- tools/video/stock_sources/wikimedia.py | 160 ++++-- tools/video/video_compose.py | 8 +- 8 files changed, 1495 insertions(+), 94 deletions(-) create mode 100644 tests/tools/test_clip_cache.py create mode 100644 tools/video/clip_cache.py diff --git a/.gitignore b/.gitignore index b16fbf9b..8e3a0fb1 100644 --- a/.gitignore +++ b/.gitignore @@ -24,6 +24,13 @@ tests/qa/output/ # Project workspaces (generated assets, renders — all regenerable) projects/ +# Corpus directories (downloaded stock clips, thumbnails, embeddings — +# regenerable from source APIs via corpus_builder). Explicit rule even +# though most corpora live under projects/ since agents sometimes +# build scratch corpora at the repo root. +corpus/ +**/corpus/ + # User music library (personal royalty-free tracks — not part of repo) music_library/ diff --git a/tests/tools/test_clip_cache.py b/tests/tools/test_clip_cache.py new file mode 100644 index 00000000..feb155d9 --- /dev/null +++ b/tests/tools/test_clip_cache.py @@ -0,0 +1,437 @@ +"""Tests for the shared clip bytes cache (tools/video/clip_cache.py). + +These tests do not touch the network and do not require filelock — +they exercise the cache's public surface (try_link, ingest, stats) +plus the LRU eviction and manifest-persistence paths. Cache dirs are +scoped to pytest's ``tmp_path`` so nothing escapes the test session. +""" +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest + +from tools.video.clip_cache import ( + CacheEntry, + ClipCache, + _link_or_copy, + default_cache_dir, + default_max_total_bytes, + get_default_cache, + reset_default_cache, +) + + +# ---------------------------------------------------------------------- +# Helpers +# ---------------------------------------------------------------------- + + +def _fake_clip(path: Path, size_bytes: int) -> Path: + """Write a fixed-size dummy file and return the path.""" + path.parent.mkdir(parents=True, exist_ok=True) + # Use a repeating pattern so hex dumps of the file are readable + # but the size is exactly what we asked for. + with open(path, "wb") as f: + f.write(b"x" * size_bytes) + return path + + +def _default_metadata(clip_id: str = "test_001") -> dict: + return { + "source": "test_source", + "source_id": clip_id.split("_", 1)[-1], + "source_url": f"https://example.test/{clip_id}", + "license": "CC0", + "creator": "test_rig", + "source_tags": "smoke test metadata", + } + + +# ---------------------------------------------------------------------- +# Config resolution +# ---------------------------------------------------------------------- + + +def test_default_cache_dir_uses_env_override(monkeypatch, tmp_path): + monkeypatch.setenv("OPENMONTAGE_CACHE_DIR", str(tmp_path / "overridden")) + assert default_cache_dir() == tmp_path / "overridden" + + +def test_default_cache_dir_falls_back_to_home(monkeypatch): + monkeypatch.delenv("OPENMONTAGE_CACHE_DIR", raising=False) + result = default_cache_dir() + assert result == Path.home() / ".openmontage" / "clips_cache" + + +def test_default_max_total_bytes_respects_env_override(monkeypatch): + monkeypatch.setenv("OPENMONTAGE_CACHE_MAX_GB", "5") + assert default_max_total_bytes() == 5 * 1024 * 1024 * 1024 + + +def test_default_max_total_bytes_ignores_garbage_override(monkeypatch): + monkeypatch.setenv("OPENMONTAGE_CACHE_MAX_GB", "not-a-number") + assert default_max_total_bytes() == 20 * 1024 * 1024 * 1024 + + +# ---------------------------------------------------------------------- +# CacheEntry dataclass round-trip +# ---------------------------------------------------------------------- + + +def test_cache_entry_round_trip_through_dict(): + entry = CacheEntry( + clip_id="test_001", + file_name="test_001.mp4", + size_bytes=42, + added_at=1000.0, + last_access_at=2000.0, + source="test_source", + source_id="001", + source_url="https://example.test/001", + license="CC0", + creator="rig", + source_tags="smoke test", + ) + d = entry.to_dict() + restored = CacheEntry.from_dict(d) + assert restored == entry + + +def test_cache_entry_from_dict_tolerates_missing_fields(): + minimal = {"clip_id": "x", "file_name": "x.mp4"} + entry = CacheEntry.from_dict(minimal) + assert entry.clip_id == "x" + assert entry.file_name == "x.mp4" + assert entry.size_bytes == 0 + assert entry.source == "" + assert entry.license == "" + + +# ---------------------------------------------------------------------- +# try_link / ingest core flow +# ---------------------------------------------------------------------- + + +def test_miss_on_empty_cache(tmp_path): + cache = ClipCache(cache_dir=tmp_path / "cache") + dest = tmp_path / "project" / "clips" / "pexels_1.mp4" + assert cache.try_link("pexels_1", dest) is False + assert not dest.exists() + assert cache.misses == 1 + assert cache.hits == 0 + + +def test_ingest_then_hit_round_trip(tmp_path): + cache = ClipCache(cache_dir=tmp_path / "cache") + project = tmp_path / "project1" + src = _fake_clip(project / "clips" / "pexels_1.mp4", 5000) + + ok = cache.ingest("pexels_1", src, _default_metadata("pexels_1")) + assert ok is True + + # A second project should get a cache hit linking the SAME blob in. + project2 = tmp_path / "project2" + dest2 = project2 / "clips" / "pexels_1.mp4" + assert cache.try_link("pexels_1", dest2) is True + assert dest2.exists() + assert dest2.stat().st_size == 5000 + assert cache.hits == 1 + assert cache.misses == 0 + + +def test_cache_hit_is_a_hard_link_not_a_copy_when_possible(tmp_path): + cache = ClipCache(cache_dir=tmp_path / "cache") + project = tmp_path / "project1" + src = _fake_clip(project / "clips" / "pexels_1.mp4", 2048) + cache.ingest("pexels_1", src, _default_metadata("pexels_1")) + + project2 = tmp_path / "project2" + dest2 = project2 / "clips" / "pexels_1.mp4" + cache.try_link("pexels_1", dest2) + + # Same filesystem → hard link → same inode. + cache_blob = cache.cache_dir / "pexels_1.mp4" + assert cache_blob.exists() + if os.name != "nt" or ( + cache_blob.stat().st_dev == dest2.stat().st_dev + ): + # Hard links have identical inode numbers on the same FS. + assert cache_blob.stat().st_ino == dest2.stat().st_ino + assert cache_blob.stat().st_nlink >= 2 + + +def test_ingest_rejects_missing_source(tmp_path): + cache = ClipCache(cache_dir=tmp_path / "cache") + missing = tmp_path / "does_not_exist.mp4" + assert cache.ingest("pexels_missing", missing, {}) is False + + +def test_ingest_rejects_too_small_file(tmp_path): + cache = ClipCache(cache_dir=tmp_path / "cache") + tiny = _fake_clip(tmp_path / "tiny.mp4", 100) # < 1024-byte threshold + assert cache.ingest("pexels_tiny", tiny, {}) is False + + +def test_ingest_same_clip_twice_bumps_last_access(tmp_path): + cache = ClipCache(cache_dir=tmp_path / "cache") + src = _fake_clip(tmp_path / "a.mp4", 2000) + cache.ingest("pexels_dup", src, _default_metadata("pexels_dup")) + + # Read the initial last_access_at + entries = cache._read_manifest() + first_access = entries["pexels_dup"].last_access_at + + # Wait a tick, then ingest again with a "fresh" source that + # simulates a rebuild. The existing entry should have its + # last_access bumped and the blob on disk kept intact. + import time + time.sleep(0.01) + src2 = _fake_clip(tmp_path / "b.mp4", 2000) + ok = cache.ingest("pexels_dup", src2, _default_metadata("pexels_dup")) + assert ok is True + + entries2 = cache._read_manifest() + assert entries2["pexels_dup"].last_access_at > first_access + # And the cache still only has one entry (the rebuild replaced itself). + assert len(entries2) == 1 + + +def test_manifest_drift_falls_back_to_miss(tmp_path): + """If the manifest has a row but the blob file is gone, report + a miss and prune the stale entry.""" + cache = ClipCache(cache_dir=tmp_path / "cache") + src = _fake_clip(tmp_path / "real.mp4", 2000) + cache.ingest("pexels_drift", src, _default_metadata("pexels_drift")) + + # Manually delete the blob to simulate filesystem drift. + blob_path = cache.cache_dir / "pexels_drift.mp4" + assert blob_path.exists() + blob_path.unlink() + + # try_link should miss AND prune the stale row. + dest = tmp_path / "project" / "clips" / "pexels_drift.mp4" + assert cache.try_link("pexels_drift", dest) is False + entries = cache._read_manifest() + assert "pexels_drift" not in entries + + +# ---------------------------------------------------------------------- +# LRU eviction +# ---------------------------------------------------------------------- + + +def test_lru_eviction_evicts_oldest_first(tmp_path): + # Cap at 30 KB. Ingest three 10 KB clips, then a fourth — the + # oldest should be evicted to make room. + cache = ClipCache( + cache_dir=tmp_path / "cache", + max_total_bytes=30 * 1024, + ) + + import time + for i in range(1, 4): + src = _fake_clip(tmp_path / f"clip_{i}.mp4", 10 * 1024) + cache.ingest(f"test_{i}", src, _default_metadata(f"test_{i}")) + time.sleep(0.01) # ensure distinct last_access_at + + # Bump access on clip_2 so clip_1 is the LRU victim. + dest = tmp_path / "project" / "clip_2.mp4" + cache.try_link("test_2", dest) + + # Add a fourth 10 KB clip — should evict clip_1 (the LRU). + src4 = _fake_clip(tmp_path / "clip_4.mp4", 10 * 1024) + cache.ingest("test_4", src4, _default_metadata("test_4")) + + entries = cache._read_manifest() + assert "test_1" not in entries, "LRU victim should have been evicted" + assert "test_2" in entries + assert "test_3" in entries + assert "test_4" in entries + assert cache.evictions_count == 1 + assert cache.bytes_evicted == 10 * 1024 + # Blob file for the evicted clip should be gone. + assert not (cache.cache_dir / "test_1.mp4").exists() + + +def test_eviction_multiple_victims_when_single_clip_is_large(tmp_path): + cache = ClipCache( + cache_dir=tmp_path / "cache", + max_total_bytes=50 * 1024, + ) + + import time + # Five 10 KB clips fit exactly at the cap. + for i in range(1, 6): + src = _fake_clip(tmp_path / f"small_{i}.mp4", 10 * 1024) + cache.ingest(f"small_{i}", src, _default_metadata(f"small_{i}")) + time.sleep(0.005) + + # A 30 KB incoming clip forces eviction of the 3 oldest. + big = _fake_clip(tmp_path / "big.mp4", 30 * 1024) + cache.ingest("big_1", big, _default_metadata("big_1")) + + entries = cache._read_manifest() + # big_1 + small_4 + small_5 survive → 3 entries. + assert "big_1" in entries + assert "small_4" in entries + assert "small_5" in entries + assert "small_1" not in entries + assert "small_2" not in entries + assert "small_3" not in entries + assert cache.evictions_count == 3 + + +# ---------------------------------------------------------------------- +# Manifest persistence across ClipCache instances +# ---------------------------------------------------------------------- + + +def test_manifest_survives_process_boundary(tmp_path): + """A second ClipCache instance pointing at the same dir should see + entries written by the first instance. This simulates the common + multi-process scenario where two corpus_builder runs share the + cache via filesystem.""" + cache_dir = tmp_path / "cache" + src = _fake_clip(tmp_path / "clip.mp4", 4000) + + cache_a = ClipCache(cache_dir=cache_dir) + assert cache_a.ingest("test_persist", src, _default_metadata("test_persist")) is True + + cache_b = ClipCache(cache_dir=cache_dir) + dest = tmp_path / "project" / "clip.mp4" + assert cache_b.try_link("test_persist", dest) is True + assert dest.stat().st_size == 4000 + + +def test_manifest_tolerates_corrupt_lines(tmp_path): + cache_dir = tmp_path / "cache" + cache = ClipCache(cache_dir=cache_dir) + + # Write a manifest with one good row and two malformed rows. + good = CacheEntry( + clip_id="good_1", + file_name="good_1.mp4", + size_bytes=1234, + added_at=1000.0, + last_access_at=1000.0, + ) + _fake_clip(cache_dir / "good_1.mp4", 1234) + manifest_path = cache_dir / ClipCache.MANIFEST_NAME + with open(manifest_path, "w", encoding="utf-8") as f: + f.write("this is not json\n") + f.write(json.dumps(good.to_dict()) + "\n") + f.write('{"missing_required_fields": true}\n') + + entries = cache._read_manifest() + assert list(entries.keys()) == ["good_1"] + + +# ---------------------------------------------------------------------- +# Cross-drive / link-fail fallback +# ---------------------------------------------------------------------- + + +def test_link_or_copy_falls_back_to_copy_when_link_fails(tmp_path, monkeypatch): + src = _fake_clip(tmp_path / "src.mp4", 3000) + dst = tmp_path / "other" / "dst.mp4" + dst.parent.mkdir(exist_ok=True) + + def broken_link(a, b): + raise OSError("simulated cross-device link failure") + + monkeypatch.setattr(os, "link", broken_link) + ok = _link_or_copy(src, dst) + assert ok is True + assert dst.exists() + assert dst.read_bytes() == src.read_bytes() + + +def test_link_or_copy_returns_false_when_both_fail(tmp_path, monkeypatch): + src = _fake_clip(tmp_path / "src.mp4", 3000) + dst = tmp_path / "other" / "dst.mp4" + dst.parent.mkdir(exist_ok=True) + + def broken_link(a, b): + raise OSError("no link") + + def broken_copy(a, b): + raise OSError("no copy") + + monkeypatch.setattr(os, "link", broken_link) + + import shutil + monkeypatch.setattr(shutil, "copy2", broken_copy) + + ok = _link_or_copy(src, dst) + assert ok is False + assert not dst.exists() + + +def test_try_link_falls_back_to_copy_and_still_bumps_access(tmp_path, monkeypatch): + cache = ClipCache(cache_dir=tmp_path / "cache") + src = _fake_clip(tmp_path / "src.mp4", 3000) + cache.ingest("pexels_x", src, _default_metadata("pexels_x")) + + def broken_link(a, b): + raise OSError("cross-drive") + + monkeypatch.setattr(os, "link", broken_link) + + dest = tmp_path / "project" / "clips" / "pexels_x.mp4" + ok = cache.try_link("pexels_x", dest) + assert ok is True + assert dest.exists() + assert cache.hits == 1 + + +# ---------------------------------------------------------------------- +# Stats +# ---------------------------------------------------------------------- + + +def test_stats_reports_cache_state_and_session_counters(tmp_path): + cache = ClipCache( + cache_dir=tmp_path / "cache", + max_total_bytes=10 * 1024 * 1024, + ) + _fake_clip(tmp_path / "a.mp4", 2000) + cache.ingest("test_stats_a", tmp_path / "a.mp4", _default_metadata("test_stats_a")) + + dest = tmp_path / "project" / "a.mp4" + cache.try_link("test_stats_a", dest) + cache.try_link("missing_clip", tmp_path / "project" / "missing.mp4") + + s = cache.stats() + assert s["entry_count"] == 1 + assert s["total_bytes"] == 2000 + assert s["hits_this_session"] == 1 + assert s["misses_this_session"] == 1 + assert s["evictions_this_session"] == 0 + assert s["max_total_bytes"] == 10 * 1024 * 1024 + assert "cache_dir" in s + assert s["filelock_backend"] in ("filelock", "o_excl_fallback") + + +# ---------------------------------------------------------------------- +# get_default_cache singleton +# ---------------------------------------------------------------------- + + +def test_get_default_cache_honors_env_var(monkeypatch, tmp_path): + reset_default_cache() + monkeypatch.setenv("OPENMONTAGE_CACHE_DIR", str(tmp_path / "envcache")) + cache = get_default_cache() + assert Path(cache.cache_dir) == tmp_path / "envcache" + reset_default_cache() + + +def test_get_default_cache_returns_same_instance(monkeypatch, tmp_path): + reset_default_cache() + monkeypatch.setenv("OPENMONTAGE_CACHE_DIR", str(tmp_path / "singleton")) + a = get_default_cache() + b = get_default_cache() + assert a is b + reset_default_cache() diff --git a/tests/tools/test_stock_source_adapters.py b/tests/tools/test_stock_source_adapters.py index dbffb049..da6269ce 100644 --- a/tests/tools/test_stock_source_adapters.py +++ b/tests/tools/test_stock_source_adapters.py @@ -1,6 +1,10 @@ from tools.video.stock_sources import all_sources from tools.video.stock_sources.unsplash import _build_download_url, _orientation_for_unsplash -from tools.video.stock_sources.wikimedia import _build_search_query, _kind_from_mime, _meta_value +from tools.video.stock_sources.wikimedia import ( + _build_search_queries, + _kind_from_mime, + _meta_value, +) def test_stock_source_autodiscovery_includes_new_sources(): @@ -10,9 +14,48 @@ def test_stock_source_autodiscovery_includes_new_sources(): def test_wikimedia_search_query_respects_kind(): - assert _build_search_query("rain city", "video").startswith("filetype:video") - assert _build_search_query("rain city", "image").startswith("filetype:image") - assert _build_search_query("rain city", "any") == "rain city" + # The cascade's first ("full") query should always carry the + # filetype filter for video/image kinds. "any" drops the prefix. + video_cascade = _build_search_queries("rain city", "video") + assert video_cascade[0][0] == "full" + assert video_cascade[0][1].startswith("filetype:video") + + image_cascade = _build_search_queries("rain city", "image") + assert image_cascade[0][0] == "full" + assert image_cascade[0][1].startswith("filetype:image") + + any_cascade = _build_search_queries("rain city", "any") + assert any_cascade[0][0] == "full" + assert any_cascade[0][1] == "rain city" + + +def test_wikimedia_cascade_falls_back_on_multi_word(): + # Multi-word query should produce a 3-stage cascade: full, top2_or, + # single_best. Tokens are picked by length, so "television" beats + # "family" and "watching". + cascade = _build_search_queries( + "1950s family watching television", "video" + ) + labels = [label for label, _ in cascade] + assert labels == ["full", "top2_or", "single_best"] + assert cascade[1][1] == "filetype:video television watching" + assert cascade[2][1] == "filetype:video television" + + +def test_wikimedia_cascade_strips_source_hints_and_years(): + # "prelinger" is a source hint (redundant on Commons) and "1955" is + # a year — both are excluded from distinctive-token picks. + cascade = _build_search_queries( + "Prelinger 1955 housewife kitchen", "video" + ) + # Full query keeps the source hint + year (first attempt is strict). + assert cascade[0][1] == "filetype:video Prelinger 1955 housewife kitchen" + # Distinctive picks do NOT include prelinger or 1955. + joined = " ".join(sq for _, sq in cascade[1:]) + assert "housewife" in joined + assert "kitchen" in joined + assert "prelinger" not in joined.lower() + assert "1955" not in joined def test_wikimedia_kind_and_metadata_helpers(): diff --git a/tools/video/clip_cache.py b/tools/video/clip_cache.py new file mode 100644 index 00000000..c0ff7def --- /dev/null +++ b/tools/video/clip_cache.py @@ -0,0 +1,572 @@ +"""Shared clip bytes cache for the corpus builder. + +Phase 1 of the shared-corpus architecture: a process-safe, LRU-evicted +cache of downloaded clip files at ``~/.openmontage/clips_cache/``. +When the corpus builder decides to fetch a candidate, it first asks +this cache whether the bytes are already on disk from a previous +project run. If yes, the cache hard-links (or copies on cross-drive) +the existing blob into the caller's corpus directory and skips the +network fetch entirely. If no, the caller downloads as usual and then +``ingest()``s the fresh bytes so the next project benefits. + +Design decisions +---------------- + +- **Per-project corpus index stays authoritative.** This cache holds + *bytes only*. The ``index.jsonl`` / embeddings of a project corpus + are scoped to that project's brief, so retrieval quality is not + polluted by clips from unrelated topics. The cache only eliminates + redundant downloads and disk copies of the underlying mp4/jpg files. + +- **Manifest is JSONL, rewritten atomically.** One entry per line with + provenance (source, license, creator) and LRU metadata (added_at, + last_access_at). Mutations rewrite the whole file via + ``os.replace`` so readers always see a consistent snapshot and + crash-mid-write only loses the in-flight mutation, not the whole + manifest. + +- **File locking on every mutation.** Concurrent corpus_builder runs + against the same cache serialize on ``cache_manifest.lock``. Uses + ``filelock`` when available (it is, per pip), else a naive exclusive + create-file fallback with a polling retry. 60s timeout. + +- **Hard links first, copies as fallback.** On the same filesystem, + hard-linking the cache blob into the caller's dest dir is free on + disk and instant on wall-time. Cross-drive (Windows C:→D:) falls + back to ``shutil.copy2`` automatically and logs nothing — the + caller doesn't need to care either way. + +- **LRU eviction at cap.** Default cap is 20 GB, overridable via + ``OPENMONTAGE_CACHE_MAX_GB``. When ``ingest()`` would push total + bytes above the cap, the cache evicts least-recently-accessed + entries until there's room. Evictions unlink the blob file and drop + the manifest row. In-flight entries (currently being ingested) are + protected by the lock. + +- **Transparent to the agent.** No skill-level changes. The agent + keeps calling ``corpus_builder.execute(...)`` the way it always has. + The cache tells its story via counters in the ``stats()`` payload + that corpus_builder bubbles up into its return value. + +Non-goals (intentional for Phase 1) +----------------------------------- +- **No embedding cache.** CLIP vectors are still computed per-project. + That's Phase 2 and is where the real wall-time wins compound. +- **No query-result cache.** Source-API calls still run every time. + That's Phase 3 and has the highest staleness risk. +- **No cross-machine sync.** Cache lives on one filesystem; there is + no S3 / Dropbox / rsync story. A future phase can add that. +""" +from __future__ import annotations + +import json +import os +import shutil +import tempfile +import time +from contextlib import contextmanager +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any, Iterator, Optional + +try: + import filelock # type: ignore + _HAVE_FILELOCK = True +except ImportError: + _HAVE_FILELOCK = False + + +# Default 20 GB cap. Overridable via OPENMONTAGE_CACHE_MAX_GB. +_DEFAULT_MAX_TOTAL_BYTES = 20 * 1024 * 1024 * 1024 + +# Reject ingesting a source file under this size — almost always a +# failed/empty download that the caller didn't catch. +_MIN_USABLE_BYTES = 1024 + + +# ---------------------------------------------------------------------- +# Config resolution +# ---------------------------------------------------------------------- + + +def default_cache_dir() -> Path: + """Resolve the cache directory. + + Honors ``OPENMONTAGE_CACHE_DIR`` if set, else falls back to + ``~/.openmontage/clips_cache``. Does not create the directory — + that happens in ``ClipCache.__init__`` on first use. + """ + override = os.environ.get("OPENMONTAGE_CACHE_DIR") + if override: + return Path(override).expanduser() + return Path.home() / ".openmontage" / "clips_cache" + + +def default_max_total_bytes() -> int: + """Resolve the max-cache-size budget. + + Honors ``OPENMONTAGE_CACHE_MAX_GB`` (float or int) if set, else + returns the default 20 GB. Invalid overrides silently fall back to + the default rather than crashing — the cache shouldn't bring down + a production run over a bad env var. + """ + override = os.environ.get("OPENMONTAGE_CACHE_MAX_GB") + if override: + try: + return int(float(override) * 1024 * 1024 * 1024) + except ValueError: + pass + return _DEFAULT_MAX_TOTAL_BYTES + + +# ---------------------------------------------------------------------- +# Dataclass for one manifest row +# ---------------------------------------------------------------------- + + +@dataclass +class CacheEntry: + """One row in the cache manifest. + + Every field except ``clip_id``/``file_name``/``size_bytes`` is + provenance metadata that the agent may want to display or attribute + downstream. Stored flat (no nesting) so JSONL lines stay short. + """ + + clip_id: str + file_name: str # relative to cache_dir, e.g. "pexels_10039002.mp4" + size_bytes: int + added_at: float + last_access_at: float + source: str = "" + source_id: str = "" + source_url: str = "" + license: str = "" + creator: str = "" + source_tags: str = "" + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + @classmethod + def from_dict(cls, d: dict[str, Any]) -> "CacheEntry": + # Tolerate extra fields from future schema evolution — only + # read the ones we know about. Missing fields default. + return cls( + clip_id=str(d["clip_id"]), + file_name=str(d["file_name"]), + size_bytes=int(d.get("size_bytes", 0) or 0), + added_at=float(d.get("added_at", 0.0) or 0.0), + last_access_at=float( + d.get("last_access_at", d.get("added_at", 0.0)) or 0.0 + ), + source=str(d.get("source", "") or ""), + source_id=str(d.get("source_id", "") or ""), + source_url=str(d.get("source_url", "") or ""), + license=str(d.get("license", "") or ""), + creator=str(d.get("creator", "") or ""), + source_tags=str(d.get("source_tags", "") or ""), + ) + + +# ---------------------------------------------------------------------- +# The cache itself +# ---------------------------------------------------------------------- + + +class ClipCache: + """Process-safe, LRU-evicted cache of downloaded clip files. + + Not a singleton — but see ``get_default_cache()`` for the common + singleton-at-default-path pattern the corpus builder uses. + """ + + MANIFEST_NAME = "cache_manifest.jsonl" + LOCK_NAME = "cache_manifest.lock" + + def __init__( + self, + cache_dir: Optional[Path] = None, + max_total_bytes: Optional[int] = None, + ): + self.cache_dir = Path(cache_dir) if cache_dir else default_cache_dir() + self.max_total_bytes = ( + int(max_total_bytes) + if max_total_bytes is not None + else default_max_total_bytes() + ) + self.cache_dir.mkdir(parents=True, exist_ok=True) + self.manifest_path = self.cache_dir / self.MANIFEST_NAME + self.lock_path = self.cache_dir / self.LOCK_NAME + + # Per-instance runtime counters. Reset every time a new + # ClipCache object is built. For persistent totals, sum + # across runs in the calling layer. + self.hits = 0 + self.misses = 0 + self.evictions_count = 0 + self.bytes_evicted = 0 + + # ------------------------------------------------------------------ + # Locking + # ------------------------------------------------------------------ + + @contextmanager + def _locked(self, timeout: float = 60.0) -> Iterator[None]: + """Acquire an exclusive lock for the duration of the block. + + Prefers ``filelock.FileLock`` (proper cross-platform, timeout + support, reentrant). Falls back to a naive O_EXCL create-file + lock with polling retry so the cache still works if filelock + is somehow unavailable. The fallback is not reentrant — don't + nest ``_locked()`` blocks. + """ + if _HAVE_FILELOCK: + lock = filelock.FileLock(str(self.lock_path), timeout=timeout) + with lock: + yield + return + + # Fallback: O_EXCL create-file lock. + deadline = time.time() + timeout + acquired = False + while time.time() < deadline: + try: + fd = os.open( + str(self.lock_path), + os.O_CREAT | os.O_EXCL | os.O_WRONLY, + ) + os.close(fd) + acquired = True + break + except FileExistsError: + time.sleep(0.05) + if not acquired: + raise TimeoutError( + f"ClipCache: could not acquire lock at {self.lock_path} " + f"after {timeout}s" + ) + try: + yield + finally: + try: + os.unlink(self.lock_path) + except OSError: + pass + + # ------------------------------------------------------------------ + # Manifest I/O (caller holds the lock) + # ------------------------------------------------------------------ + + def _read_manifest(self) -> dict[str, CacheEntry]: + """Read the manifest file into a dict keyed by clip_id. + + Malformed lines are skipped silently — a single bad row must + not poison the whole manifest. Missing file returns an empty + dict (first-run case). + """ + entries: dict[str, CacheEntry] = {} + if not self.manifest_path.exists(): + return entries + try: + with open(self.manifest_path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + d = json.loads(line) + entry = CacheEntry.from_dict(d) + entries[entry.clip_id] = entry + except (json.JSONDecodeError, KeyError, TypeError, ValueError): + continue + except OSError: + # Manifest disappeared between exists() and open() — + # treat as empty. The lock prevents this under normal + # operation but filesystems can surprise us. + return {} + return entries + + def _write_manifest(self, entries: dict[str, CacheEntry]) -> None: + """Rewrite the manifest file atomically. + + Writes to a sibling tmpfile and uses ``os.replace`` which is + atomic on both POSIX and Windows. A crash between write and + replace leaves the old manifest intact. + """ + tmp_fd, tmp_name = tempfile.mkstemp( + prefix="cache_manifest.", suffix=".tmp", dir=str(self.cache_dir) + ) + try: + with os.fdopen(tmp_fd, "w", encoding="utf-8") as f: + for entry in entries.values(): + f.write(json.dumps(entry.to_dict(), ensure_ascii=False) + "\n") + os.replace(tmp_name, self.manifest_path) + except Exception: + # Best-effort cleanup of the tmpfile if replace failed + try: + if os.path.exists(tmp_name): + os.unlink(tmp_name) + except OSError: + pass + raise + + # ------------------------------------------------------------------ + # Public API — try_link, ingest, stats + # ------------------------------------------------------------------ + + def try_link(self, clip_id: str, dest: Path) -> bool: + """Hard-link (or copy) a cached clip into ``dest`` if present. + + Returns ``True`` on cache hit, ``False`` on miss. On hit, the + entry's ``last_access_at`` is bumped so LRU eviction keeps + recently-used clips around. + + On manifest/filesystem drift (entry exists but blob file is + gone), the stale entry is pruned and the call reports a miss, + so the caller falls back to downloading fresh. + """ + dest = Path(dest) + with self._locked(): + entries = self._read_manifest() + entry = entries.get(clip_id) + if entry is None: + self.misses += 1 + return False + + blob_path = self.cache_dir / entry.file_name + if not blob_path.exists(): + # Drift — prune and miss. + del entries[clip_id] + self._write_manifest(entries) + self.misses += 1 + return False + + dest.parent.mkdir(parents=True, exist_ok=True) + # Remove any existing file at dest first so the hard link + # can be created cleanly. Harmless if dest didn't exist. + if dest.exists() or dest.is_symlink(): + try: + dest.unlink() + except OSError: + pass + + if not _link_or_copy(blob_path, dest): + # Link and copy both failed — treat as miss so the + # caller redownloads. Leave the cache entry alone; + # the blob is still valid, we just can't reach dest. + self.misses += 1 + return False + + entry.last_access_at = time.time() + entries[clip_id] = entry + self._write_manifest(entries) + self.hits += 1 + return True + + def ingest( + self, + clip_id: str, + source_path: Path, + metadata: Optional[dict[str, Any]] = None, + ) -> bool: + """Copy/link a freshly downloaded clip file into the cache. + + ``source_path`` is the file as it already sits in the caller's + project directory after a successful download. We do NOT move + or mutate it — the caller keeps the file for its own pipeline + and the cache holds a second reference via hard link (or copy + on cross-drive). + + Returns ``True`` if the clip was added (or was already present + and had its access time bumped), ``False`` if ingest failed + (missing source, empty file, lock timeout, or link/copy fail). + """ + source_path = Path(source_path) + if not source_path.exists(): + return False + try: + size_bytes = source_path.stat().st_size + except OSError: + return False + if size_bytes < _MIN_USABLE_BYTES: + return False + + metadata = dict(metadata or {}) + + with self._locked(): + entries = self._read_manifest() + + # Already cached → just bump last_access and return. + if clip_id in entries and ( + self.cache_dir / entries[clip_id].file_name + ).exists(): + entries[clip_id].last_access_at = time.time() + self._write_manifest(entries) + return True + + # Make room. + self._evict_to_fit_locked(entries, size_bytes) + + # Name the blob ``{clip_id}{ext}``. Stable and collision-free + # as long as clip_ids are unique (they are — {source}_{source_id}). + ext = source_path.suffix or "" + blob_name = f"{clip_id}{ext}" + blob_path = self.cache_dir / blob_name + + # Clean any stale blob at the same path (drift or interrupted + # ingest from a previous run). + if blob_path.exists(): + try: + blob_path.unlink() + except OSError: + return False + + if not _link_or_copy(source_path, blob_path): + return False + + now = time.time() + entries[clip_id] = CacheEntry( + clip_id=clip_id, + file_name=blob_name, + size_bytes=size_bytes, + added_at=now, + last_access_at=now, + source=str(metadata.get("source", "") or ""), + source_id=str(metadata.get("source_id", "") or ""), + source_url=str(metadata.get("source_url", "") or ""), + license=str(metadata.get("license", "") or ""), + creator=str(metadata.get("creator", "") or ""), + source_tags=str(metadata.get("source_tags", "") or ""), + ) + self._write_manifest(entries) + return True + + def stats(self) -> dict[str, Any]: + """Return a snapshot of cache state plus session counters. + + The persistent fields (``entry_count``, ``total_bytes``) reflect + what's on disk right now. The ``*_this_session`` fields reflect + only what this ``ClipCache`` instance has observed — they + reset when the process exits. + """ + with self._locked(): + entries = self._read_manifest() + total_bytes = sum(e.size_bytes for e in entries.values()) + return { + "cache_dir": str(self.cache_dir), + "entry_count": len(entries), + "total_bytes": total_bytes, + "total_mb": round(total_bytes / (1024 * 1024), 1), + "max_total_bytes": self.max_total_bytes, + "max_total_gb": round(self.max_total_bytes / (1024 ** 3), 2), + "usage_fraction": ( + round(total_bytes / self.max_total_bytes, 3) + if self.max_total_bytes > 0 else 0.0 + ), + "hits_this_session": self.hits, + "misses_this_session": self.misses, + "evictions_this_session": self.evictions_count, + "bytes_evicted_this_session": self.bytes_evicted, + "filelock_backend": "filelock" if _HAVE_FILELOCK else "o_excl_fallback", + } + + # ------------------------------------------------------------------ + # LRU eviction (caller holds the lock) + # ------------------------------------------------------------------ + + def _evict_to_fit_locked( + self, entries: dict[str, CacheEntry], needed_bytes: int + ) -> None: + """Evict least-recently-accessed entries until ``needed_bytes`` fits. + + Mutates ``entries`` in place. Silently skips victims whose + blob file has already vanished (drift) so eviction is + best-effort and does not block the ingest path. + """ + if needed_bytes <= 0: + return + current_bytes = sum(e.size_bytes for e in entries.values()) + if current_bytes + needed_bytes <= self.max_total_bytes: + return + + # Oldest first. + sorted_victims = sorted( + entries.values(), key=lambda e: e.last_access_at + ) + for victim in sorted_victims: + if current_bytes + needed_bytes <= self.max_total_bytes: + break + blob_path = self.cache_dir / victim.file_name + unlinked = False + try: + if blob_path.exists(): + blob_path.unlink() + unlinked = True + except OSError: + # Could not delete the blob (in-use on Windows, for + # instance). Leave the entry in place and try the next. + continue + if not unlinked: + continue + current_bytes -= victim.size_bytes + del entries[victim.clip_id] + self.evictions_count += 1 + self.bytes_evicted += victim.size_bytes + + +# ---------------------------------------------------------------------- +# Module-level helpers +# ---------------------------------------------------------------------- + + +def _link_or_copy(src: Path, dst: Path) -> bool: + """Hard-link ``src`` to ``dst``; on failure, fall back to ``shutil.copy2``. + + Hard linking is instant and uses zero extra disk on the same + filesystem. Cross-drive (Windows C:→D:) and cross-filesystem + situations raise ``OSError`` on ``os.link`` and we transparently + copy the bytes instead. Returns ``True`` on success, ``False`` if + both link and copy failed. + """ + src = Path(src) + dst = Path(dst) + try: + os.link(str(src), str(dst)) + return True + except (OSError, NotImplementedError): + pass + try: + shutil.copy2(str(src), str(dst)) + return True + except (OSError, shutil.SameFileError): + return False + + +# ---------------------------------------------------------------------- +# Default-singleton accessor +# ---------------------------------------------------------------------- + + +_DEFAULT_CACHE: Optional[ClipCache] = None + + +def get_default_cache() -> ClipCache: + """Return a process-level default ``ClipCache`` at the default path. + + Lazily constructed on first call. Tests that want a pristine cache + should instantiate ``ClipCache(cache_dir=tmp_path)`` directly + rather than going through this accessor. + """ + global _DEFAULT_CACHE + if _DEFAULT_CACHE is None: + _DEFAULT_CACHE = ClipCache() + return _DEFAULT_CACHE + + +def reset_default_cache() -> None: + """Drop the cached singleton so a subsequent ``get_default_cache()`` + re-reads env vars. Useful for tests that mutate ``OPENMONTAGE_CACHE_DIR``. + """ + global _DEFAULT_CACHE + _DEFAULT_CACHE = None diff --git a/tools/video/corpus_builder.py b/tools/video/corpus_builder.py index 4bf6b90b..f426b40c 100644 --- a/tools/video/corpus_builder.py +++ b/tools/video/corpus_builder.py @@ -240,6 +240,7 @@ class CorpusBuilder(BaseTool): start = time.time() try: from lib.corpus import Corpus + from tools.video.clip_cache import get_default_cache from tools.video.stock_sources import ( SearchFilters, all_sources, @@ -301,6 +302,14 @@ class CorpusBuilder(BaseTool): corp.load() corp.ensure_dirs() + # Shared clip bytes cache (Phase 1). Hits hard-link blobs + # from a previous run's download into this corpus dir so we + # don't re-hit the source API or re-download megabytes. + # Faults never block the pipeline — a cache miss just means + # we download like before. + cache = get_default_cache() + run_cache_stats = {"hits": 0, "misses": 0, "bytes_saved": 0} + per_source_counts: dict[str, int] = {s.name: 0 for s in sources} added_ids: list[str] = [] errors: list[dict] = [] @@ -354,6 +363,8 @@ class CorpusBuilder(BaseTool): corp=corp, query=query, thumbs_per_video=thumbs_per_video, + cache=cache, + run_cache_stats=run_cache_stats, ) except Exception as e: failed += 1 @@ -376,6 +387,11 @@ class CorpusBuilder(BaseTool): corp.save() elapsed = time.time() - start + try: + cache_snapshot = cache.stats() + except Exception as e: + cache_snapshot = {"error": f"{type(e).__name__}: {e}"} + return ToolResult( success=True, data={ @@ -391,6 +407,13 @@ class CorpusBuilder(BaseTool): "requested_sources": source_names or [], "resolved_sources": [s.name for s in sources], "source_provider_summary": source_summary(), + # Shared clip bytes cache (Phase 1): per-run + # counters plus a full stats snapshot for the + # agent to display in the production report. + "cache_hits": run_cache_stats["hits"], + "cache_misses": run_cache_stats["misses"], + "cache_bytes_saved": run_cache_stats["bytes_saved"], + "cache_stats": cache_snapshot, "errors": errors[:25], # cap log noise }, cost_usd=0.0, @@ -414,12 +437,23 @@ class CorpusBuilder(BaseTool): corp, query: str, thumbs_per_video: int, + cache, + run_cache_stats: dict, ): """Download → thumb → embed → add one Candidate to the corpus. Returns the created `ClipRecord` on success, None if the clip was rejected (download empty, thumb extraction failed, etc.). Raises on unexpected errors (the caller logs them). + + Before downloading, consults the shared clip bytes cache at + ``~/.openmontage/clips_cache/``: if the file is already on + disk from a previous run (the same clip surfaced for a + different project), the cache hard-links it straight into + ``local_abs`` and we skip the network fetch entirely. On a + miss we download as before and ingest the fresh file so the + next run benefits. Cache faults never block the pipeline — + they degrade gracefully to normal downloads. """ import cv2 @@ -432,18 +466,58 @@ class CorpusBuilder(BaseTool): local_rel = Path("clips") / f"{cand.clip_id}{ext}" local_abs = corp.corpus_dir / local_rel - # Download. Any HTTP/IO exception propagates up to the - # per-candidate try in execute(). - src.download(cand, local_abs) - if not local_abs.exists() or local_abs.stat().st_size < 1024: - # Empty / near-empty file = bad download. Clean up so a - # retry doesn't mistake it for success. + # Try the shared cache first. A hit links the cached blob + # into local_abs (same filesystem → hard link, cross-drive + # → copy) and we skip the source fetch entirely. + cache_hit = False + try: + cache_hit = cache.try_link(cand.clip_id, local_abs) + except Exception: + # Never let a cache fault block the pipeline — fall + # through to a fresh download. The cache surfaces faults + # via its own stats counters. + cache_hit = False + + if cache_hit: + run_cache_stats["hits"] += 1 try: - if local_abs.exists(): - local_abs.unlink() + run_cache_stats["bytes_saved"] += local_abs.stat().st_size except OSError: pass - return None + else: + run_cache_stats["misses"] += 1 + # Download. Any HTTP/IO exception propagates up to the + # per-candidate try in execute(). + src.download(cand, local_abs) + if not local_abs.exists() or local_abs.stat().st_size < 1024: + # Empty / near-empty file = bad download. Clean up so a + # retry doesn't mistake it for success. + try: + if local_abs.exists(): + local_abs.unlink() + except OSError: + pass + return None + + # Ingest the fresh file into the shared cache so the + # next run can hit it. Swallow ingest failures — the + # current run already has the bytes locally, which is + # what matters for this build. + try: + cache.ingest( + cand.clip_id, + local_abs, + metadata={ + "source": cand.source, + "source_id": cand.source_id, + "source_url": cand.source_url, + "license": cand.license, + "creator": cand.creator, + "source_tags": cand.source_tags, + }, + ) + except Exception: + pass thumb_dir_rel = Path("thumbnails") / cand.clip_id thumb_dir_abs = corp.corpus_dir / thumb_dir_rel diff --git a/tools/video/stock_sources/archive_org.py b/tools/video/stock_sources/archive_org.py index a0091768..28b37e7c 100644 --- a/tools/video/stock_sources/archive_org.py +++ b/tools/video/stock_sources/archive_org.py @@ -52,7 +52,7 @@ _DEFAULT_COLLECTIONS = ("prelinger", "opensource_movies", "home_movies") # File formats we accept, in preference order. Archive.org runs every # upload through a derivative pipeline so most items have multiple -# renditions — we want the best mp4. +# renditions — we want the best mp4 we can afford under the size cap. _VIDEO_FORMAT_PRIORITY = ( "h.264", # mp4, usually 480p or 720p "MPEG4", # older mp4 encoding @@ -62,6 +62,37 @@ _VIDEO_FORMAT_PRIORITY = ( "WebM", # webm ) +# Skip any rendition bigger than this at pick time so we never queue +# a multi-hundred-megabyte download. Archive.org hosts full-length +# films routinely and even one 2 GB Prelinger master poisons a corpus +# build on wall-time and disk. Lives in this adapter (not in +# corpus_builder) because only archive.org's multi-rendition +# derivative pipeline needs the within-item size shopping this +# resolves against. +_MAX_FILE_SIZE_BYTES = 150 * 1024 * 1024 # 150 MB + +# Archive.org is the only source that routinely hosts multi-hour +# material, so we apply a duration ceiling even when the caller did +# not set `filters.max_duration`. Other adapters can stay loose. +_DEFAULT_MAX_DURATION_SECONDS = 180.0 + +# Stop words and very short tokens are dropped from the user query +# before the query cascade so they don't dilute Solr relevance. Kept +# small and deliberate — we're only killing words that reliably hurt +# documentary-style searches. +_STOP_WORDS = frozenset({ + "the", "and", "for", "with", "that", "this", "from", "into", + "its", "their", "about", "over", "under", "while", "during", + "your", "you", "our", "are", "was", "were", "have", "has", +}) + +# Tokens that are redundant with the collection filter — including them +# in the search body just matches item descriptions that happen to say +# "from the prelinger archives" and not much else. Strip them. +_SOURCE_HINT_TOKENS = frozenset({ + "prelinger", "archive", "archives", "stock", "footage", +}) + class ArchiveOrgSource: """Adapter for public-domain video on archive.org. @@ -96,6 +127,13 @@ class ArchiveOrgSource: adapter returns an empty list for `kind="image"` since Archive.org's image collections are a separate ecosystem (see `nasa.py` for astronomy imagery instead). + + Query strategy is a 3-stage cascade (`_build_queries`). We walk + the cascade in order and return the first strategy whose + hydrated candidate list is non-empty. See `_build_queries` for + why a cascade is needed — pure OR-join is too noisy, pure + phrase-proximity is too strict, no single strategy wins across + the documentary query space. """ kind = (filters.kind or "video").lower() if kind not in ("video", "any"): @@ -103,33 +141,43 @@ class ArchiveOrgSource: import requests # lazy - q = self._build_query(query) - params = [ - ("q", q), - ("fl[]", "identifier"), - ("fl[]", "title"), - ("fl[]", "description"), - ("fl[]", "creator"), - ("fl[]", "date"), - ("fl[]", "subject"), - ("fl[]", "licenseurl"), - ("fl[]", "collection"), - ("rows", str(max(1, min(filters.per_page, 50)))), - ("page", str(max(1, filters.page))), - ("output", "json"), - ] + for _label, solr_q in self._build_queries(query): + params = [ + ("q", solr_q), + ("fl[]", "identifier"), + ("fl[]", "title"), + ("fl[]", "description"), + ("fl[]", "creator"), + ("fl[]", "date"), + ("fl[]", "subject"), + ("fl[]", "licenseurl"), + ("fl[]", "collection"), + ("rows", str(max(1, min(filters.per_page, 50)))), + ("page", str(max(1, filters.page))), + ("output", "json"), + ] - r = requests.get(_SEARCH_URL, params=params, timeout=30) - r.raise_for_status() - data = r.json() - docs = (data.get("response") or {}).get("docs", []) or [] + try: + r = requests.get(_SEARCH_URL, params=params, timeout=30) + r.raise_for_status() + data = r.json() + except Exception: + # One strategy's network/parse error shouldn't kill the + # whole cascade — try the next one. + continue + docs = (data.get("response") or {}).get("docs", []) or [] + if not docs: + continue - out: list[Candidate] = [] - for doc in docs: - cand = self._hydrate_candidate(doc, filters) - if cand is not None: - out.append(cand) - return out + out: list[Candidate] = [] + for doc in docs: + cand = self._hydrate_candidate(doc, filters) + if cand is not None: + out.append(cand) + if out: + return out + + return [] def download(self, candidate: Candidate, out_path: Path) -> Path: """Stream the candidate's file to `out_path`. @@ -161,20 +209,103 @@ class ArchiveOrgSource: # Internals # ------------------------------------------------------------------ - def _build_query(self, user_query: str) -> str: - """Wrap the user's query with mediatype + collection filters. + def _build_queries(self, user_query: str) -> list[tuple[str, str]]: + """Build a cascade of Solr queries to try in preference order. - Archive.org's query language is Solr-style — parentheses and - booleans work, and spaces default to AND. We quote the user - query so multi-word phrases stay intact. + Archive.org's Solr index behaves badly for natural-language + documentary queries. The obvious strategies fail in different + ways and no single strategy wins across the query space, so we + try several and return the first one whose results aren't + empty: + + 1. **phrase_prox_10** — ``"{phrase}"~10`` proximity match. + Finds items whose title/description contains all the query + tokens within 10 positions of each other. Best for + name-dropped items ("duck and cover drill" -> + ``DuckandC1951``). Returns 0 for descriptive queries that + don't correspond to a real item title. + + 2. **distinctive_and** — top-2 longest non-year tokens joined + with AND. Catches items that have the distinctive tokens in + the same record without requiring proximity. Finds Prelinger + gems like "Frigidaire Imperial Line 1956" when the query is + "1955 refrigerator kitchen", or "To New Horizons" (1940) + when the query is "suburban optimism suburbia". + + 3. **distinctive_or** — top-3 longest tokens OR-joined. Last + resort. Noisy — we sacrifice precision for non-empty results + and rely on the downstream CLIP retrieval filter to reject + junk at clip_search time. Much narrower than OR-joining all + tokens because longer words are more discriminative. + + Year tokens ("1950s", "1955") are excluded from the + distinctive-token picks because Prelinger titles rarely encode + years in text. Including "1950s" in an AND join almost always + zeros the result set. + + Source-hint tokens ("prelinger", "archive", "footage") are + stripped entirely — they're redundant with the collection + filter and match junk descriptions. + + Stop words and short tokens are dropped before everything to + keep the proximity phrase meaningful. """ coll = " OR ".join(f"collection:{c}" for c in _DEFAULT_COLLECTIONS) user = user_query.strip() if not user: - return f"mediatype:movies AND ({coll})" - # Quote the user query as a phrase AND a loose-term search so - # we get both precise matches and relevance-ranked hits. - return f'mediatype:movies AND ({coll}) AND ({user})' + return [("default", f"mediatype:movies AND ({coll})")] + + tokens = [ + t for t in re.split(r"\s+", user) + if len(t) >= 3 + and t.lower() not in _STOP_WORDS + and t.lower() not in _SOURCE_HINT_TOKENS + ] + if not tokens: + # Nothing meaningful survived filtering — fall back to a + # quoted phrase search so we don't ship an empty query. + return [( + "quoted_fallback", + f'mediatype:movies AND ({coll}) AND ("{user}")', + )] + + queries: list[tuple[str, str]] = [] + + # Strategy 1: phrase proximity ~10. Uses the stripped token + # sequence (not the raw query) so stop words don't pollute the + # phrase match. + clean_phrase = " ".join(tokens) + queries.append(( + "phrase_prox_10", + f'mediatype:movies AND ({coll}) AND ("{clean_phrase}"~10)', + )) + + # Strategy 2: top-2 longest non-year tokens AND-joined. + non_year = [t for t in tokens if not _looks_like_year(t)] + if len(non_year) >= 2: + distinctive = sorted(non_year, key=lambda t: -len(t))[:2] + and_q = " AND ".join(distinctive) + queries.append(( + "distinctive_and", + f"mediatype:movies AND ({coll}) AND ({and_q})", + )) + elif len(non_year) == 1: + # Single non-year token — wrap in a simple term query. + queries.append(( + "single_term", + f"mediatype:movies AND ({coll}) AND ({non_year[0]})", + )) + + # Strategy 3: top-3 longest tokens OR-joined. Last-resort + # fallback that accepts noise in exchange for non-empty results. + top_tokens = sorted(tokens, key=lambda t: -len(t))[:3] + or_q = " OR ".join(top_tokens) + queries.append(( + "distinctive_or", + f"mediatype:movies AND ({coll}) AND ({or_q})", + )) + + return queries def _hydrate_candidate( self, doc: dict, filters: SearchFilters @@ -208,12 +339,26 @@ class ArchiveOrgSource: return None duration = _parse_length(picked.get("length")) - if filters.min_duration is not None and duration < filters.min_duration: + + # Apply a default max-duration ceiling if the caller didn't set + # one. Archive.org is the only source that routinely hosts + # multi-hour material, and without a default cap a naive + # fan-out pulls feature-length items into a corpus that only + # ever wants a few seconds per clip. + effective_max_duration = filters.max_duration + if effective_max_duration is None: + effective_max_duration = _DEFAULT_MAX_DURATION_SECONDS + + if ( + filters.min_duration is not None + and duration + and duration < filters.min_duration + ): return None - if filters.max_duration is not None and 0 < duration < filters.max_duration: - # 0 means "unknown" — keep those, reject only known-too-long - pass - if filters.max_duration is not None and duration > filters.max_duration: + # duration == 0 means "unknown" — pass through and let the + # corpus builder's post-download ffprobe decide. Known-too-long + # items are rejected here before the download queue. + if duration and duration > effective_max_duration: return None width = _safe_int(picked.get("width")) @@ -271,12 +416,31 @@ class ArchiveOrgSource: # ---------------------------------------------------------------------- +def _looks_like_year(token: str) -> bool: + """True if token is a bare year or year-with-decade-suffix. + + Examples: ``"1950"``, ``"1950s"``, ``"2026"``. Used by + ``_build_queries`` to exclude year tokens from the distinctive- + token picks, since Prelinger item titles rarely carry the year as + a searchable token. + """ + bare = token.rstrip("s") + return bare.isdigit() and len(bare) == 4 + + def _pick_video_file(files: list[dict]) -> Optional[dict]: """Pick the best playable video file from an Archive.org files list. Preference order is defined by `_VIDEO_FORMAT_PRIORITY`. Within a - format, we prefer the largest file (size as a quality proxy), and - reject obvious thumbnails / derivative animations. + format we first drop renditions that exceed `_MAX_FILE_SIZE_BYTES`, + then pick the largest of the survivors — size is a decent bitrate + proxy, so biggest-under-budget = best quality we can afford. + + If no rendition in the preferred format fits the budget we fall + through to the next format, so a 2 GB h.264 master gracefully + degrades to a 40 MB 512Kb MPEG4 derivative rather than having the + whole item dropped from the corpus. Thumbnails and preview GIFs + are skipped regardless of format. """ if not files: return None @@ -297,8 +461,18 @@ def _pick_video_file(files: list[dict]) -> Optional[dict]: bucket = by_format.get(fmt) if not bucket: continue - bucket.sort(key=lambda f: _safe_int(f.get("size")), reverse=True) - return bucket[0] + affordable = [ + f for f in bucket + if 0 < _safe_int(f.get("size")) <= _MAX_FILE_SIZE_BYTES + ] + if not affordable: + # Every rendition in this format is either too big or has + # no reported size. Fall through to the next format. + continue + affordable.sort( + key=lambda f: _safe_int(f.get("size")), reverse=True + ) + return affordable[0] return None diff --git a/tools/video/stock_sources/wikimedia.py b/tools/video/stock_sources/wikimedia.py index 9ac881dc..e9a416a6 100644 --- a/tools/video/stock_sources/wikimedia.py +++ b/tools/video/stock_sources/wikimedia.py @@ -20,6 +20,23 @@ _USER_AGENT = "OpenMontageBot/0.1 (https://github.com/calesthio/OpenMontage)" _COMMONS_LICENSE = "Wikimedia Commons (verify per-file license)" _HTML_TAG_RE = re.compile(r"<[^>]+>") +# Stop words stripped from multi-term queries before the cascade runs. +# Commons' CirrusSearch defaults to AND semantics across multi-word +# queries, so each extra common token shrinks the result set fast. +_STOP_WORDS = frozenset({ + "the", "and", "for", "with", "that", "this", "from", "into", + "its", "their", "about", "over", "under", "while", "during", + "your", "you", "our", "are", "was", "were", "have", "has", +}) + +# Tokens that refer to other stock archives — useless on Commons and +# will poison the cascade if they end up in top2_or because Commons +# file names don't reference Prelinger or other archives. Keeps the +# cascade parallel to ``archive_org.py``'s own source-hint stripping. +_SOURCE_HINT_TOKENS = frozenset({ + "prelinger", "archive", "archives", "stock", "footage", +}) + class WikimediaSource: """Adapter for Wikimedia Commons media search.""" @@ -37,39 +54,59 @@ class WikimediaSource: return True def search(self, query: str, filters: SearchFilters) -> list[Candidate]: + """Search Commons via CirrusSearch, cascading from precise to broad. + + Commons' search defaults to AND across multi-word queries, so + our first diagnostic pass against the P2 query set returned 0 + video results for 10/10 queries — every query was too specific + to intersect Commons' relatively sparse video holdings. + + The cascade (see ``_build_search_queries``) tries strict first, + then narrows to 2 distinctive tokens, then to 1 — returning the + first non-empty video result set. + """ import requests # lazy - params = { - "action": "query", - "format": "json", - "generator": "search", - "gsrsearch": _build_search_query(query, filters.kind), - "gsrnamespace": 6, - "gsrlimit": max(1, min(filters.per_page, 50)), - "gsroffset": max(0, (max(filters.page, 1) - 1) * max(1, min(filters.per_page, 50))), - "prop": "imageinfo|info", - "iiprop": "url|size|mime|extmetadata|mediatype", - "iiurlwidth": 640, - "inprop": "url", - } + for _label, search_text in _build_search_queries(query, filters.kind): + params = { + "action": "query", + "format": "json", + "generator": "search", + "gsrsearch": search_text, + "gsrnamespace": 6, + "gsrlimit": max(1, min(filters.per_page, 50)), + "gsroffset": max(0, (max(filters.page, 1) - 1) * max(1, min(filters.per_page, 50))), + "prop": "imageinfo|info", + "iiprop": "url|size|mime|extmetadata|mediatype", + "iiurlwidth": 640, + "inprop": "url", + } - r = requests.get( - _API_URL, - params=params, - headers={"User-Agent": _USER_AGENT}, - timeout=30, - ) - r.raise_for_status() - data = r.json() - pages = list(((data.get("query") or {}).get("pages") or {}).values()) - pages.sort(key=lambda page: int(page.get("index", 0))) + try: + r = requests.get( + _API_URL, + params=params, + headers={"User-Agent": _USER_AGENT}, + timeout=30, + ) + r.raise_for_status() + data = r.json() + except Exception: + continue + pages = list(((data.get("query") or {}).get("pages") or {}).values()) + if not pages: + continue + pages.sort(key=lambda page: int(page.get("index", 0))) - out: list[Candidate] = [] - for page in pages: - cand = _page_to_candidate(page, filters) - if cand is not None: - out.append(cand) - return out + out: list[Candidate] = [] + for page in pages: + cand = _page_to_candidate(page, filters) + if cand is not None: + out.append(cand) + if out: + return out + + return [] def download(self, candidate: Candidate, out_path: Path) -> Path: import requests # lazy @@ -94,14 +131,65 @@ class WikimediaSource: return out_path -def _build_search_query(query: str, kind: str) -> str: +def _build_search_queries(query: str, kind: str) -> list[tuple[str, str]]: + """Return a cascade of search queries to try in preference order. + + Commons' CirrusSearch defaults to AND semantics for multi-word + queries, so a 4-word descriptive query like + "1950s family watching television" intersects to 0 video hits. + We walk from specific to loose: + + 1. **full** — ``filetype:video ``. Works when Commons + has a file whose name/description contains all the tokens + (e.g. "atomic bomb test civil defense" finds + "Operation Cue 1955"). + 2. **top2_or** — ``filetype:video `` using the + two longest non-year tokens. AND-combines at the query level + but with only 2 terms, it's loose enough to hit most + documentary queries. + 3. **single_best** — ``filetype:video ``. + Last-resort single-token search. Noisy but non-empty. + + Year tokens are excluded from the distinctive-token picks — they + rarely correlate with file name matches on Commons. + """ user_query = query.strip() - kind = (kind or "video").lower() - if kind == "video": - return f"filetype:video {user_query}".strip() - if kind == "image": - return f"filetype:image {user_query}".strip() - return user_query + kind_l = (kind or "video").lower() + + prefix = "filetype:video" if kind_l == "video" else ( + "filetype:image" if kind_l == "image" else "" + ) + + def _wrap(text: str) -> str: + return f"{prefix} {text}".strip() if prefix else text + + if not user_query: + return [("default", _wrap(""))] + + tokens = [ + t for t in user_query.split() + if len(t) >= 3 + and t.lower() not in _STOP_WORDS + and t.lower() not in _SOURCE_HINT_TOKENS + ] + non_year = [t for t in tokens if not _looks_like_year(t)] + + queries: list[tuple[str, str]] = [("full", _wrap(user_query))] + + if len(non_year) >= 2: + top2 = sorted(non_year, key=lambda t: -len(t))[:2] + queries.append(("top2_or", _wrap(f"{top2[0]} {top2[1]}"))) + + if non_year: + best = max(non_year, key=len) + queries.append(("single_best", _wrap(best))) + + return queries + + +def _looks_like_year(token: str) -> bool: + bare = token.rstrip("s") + return bare.isdigit() and len(bare) == 4 def _page_to_candidate(page: dict[str, Any], filters: SearchFilters) -> Candidate | None: diff --git a/tools/video/video_compose.py b/tools/video/video_compose.py index d98b3b6d..b6bcb1f1 100644 --- a/tools/video/video_compose.py +++ b/tools/video/video_compose.py @@ -973,6 +973,8 @@ class VideoCompose(BaseTool): output_path = Path(inputs.get("output_path", "renders/remotion_output.mp4")) output_path.parent.mkdir(parents=True, exist_ok=True) + # Absolutise so the CLI can resolve the output regardless of cwd. + output_path = output_path.resolve() # Deep-copy props so we don't mutate the original props = json.loads(json.dumps(composition_data)) @@ -1037,7 +1039,11 @@ class VideoCompose(BaseTool): pass try: - self.run_command(cmd, timeout=600) + # Invoke from inside the composer dir so npx can resolve the + # local remotion binary via node_modules/.bin. Without this, + # Windows npx cannot locate the CLI and returns "could not + # determine executable to run". + self.run_command(cmd, timeout=600, cwd=composer_dir) except Exception as e: return ToolResult(success=False, error=f"Remotion render failed: {e}") finally: