freedak/OpenMontage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add documentary-montage pipeline for retrieval-first motion-clip montage

Browse Source 
New end-to-end pipeline for building thematic documentary montages from
a locally-indexed corpus of free stock footage (Pexels, Archive.org,
NASA). The agent builds a project-local corpus, CLIP-ranks candidates
per scene slot, edits with motion-aware arc logic, and composes via
ffmpeg. No paid APIs required for the full path.

Pipeline definition and director skills:
- pipeline_defs/documentary-montage.yaml: 5-stage manifest
  (idea -> scene_plan -> assets -> edit -> compose)
- skills/pipelines/documentary-montage/: 6 director skills
  (executive-producer + idea/scene/asset/edit/compose directors)

Corpus and retrieval infrastructure:
- tools/video/corpus_builder.py: multi-source stock fan-out with
  resumable append-only corpus index
- tools/video/clip_search.py: CLIP ViT-B/32 retrieval —
  rank_for_slot, find_similar_set, diversify, stats
- tools/video/stock_sources/: base + pexels + archive_org + nasa
  adapters with a pluggable BaseStockSource contract
- lib/clip_embedder.py: CLIP wrapper
- lib/corpus.py: corpus schema, jsonl append/read, motion-score
  caching

video_compose fix rolled in because any concat-based pipeline depends
on it:
- Replace ambiguous -to with -t duration (was double-trimming cuts)
- Force re-encode + normalize to 1920x1080 @ 30fps (was keyframe-
  snapping with -c copy and breaking concat on mixed-source corpora)
- Add silent-audio anullsrc fallback for clips without an audio
  stream

README: add Documentary Montage row to the pipeline table and bump
the pipeline count from 11 to 12.
This commit is contained in:
calesthio
2026-04-10 16:04:28 -07:00
parent e815126f5a
commit 44baede67f
18 changed files with 4446 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/video/clip_search.py
+356
View File
@@ -0,0 +1,356 @@
"""Clip search: unified retrieval interface over a local clip corpus.
This is the tool the documentary-montage director calls at edit time.
It loads a corpus built by `corpus_builder` and exposes every
retrieval operation the agent needs through a single dispatch
interface.
Operations
----------
- **rank_for_slot**: embed a text description of a scene slot and
return the top-k clips by fused visual+tag similarity. The agent's
main building block — "for this slot in the montage, what clips
match?"
- **find_similar_set**: given one seed clip, return N clips that share
the seed's visual register but are diverse from each other (MMR).
Used for "collection" shots — all the doorways, all the footsteps,
all the keys-in-locks.
- **diversify**: given a pre-selected list of clip_ids, greedily keep
the most mutually-dissimilar subset. Used at arrangement time to
prevent visually-redundant adjacent cuts.
- **get**: look up one clip_id and return its full provenance dict.
- **stats**: summary counts (rows, per-source breakdown, mean motion).
All operations return JSON-serialisable dicts so the tool contract
stays clean across process boundaries. ClipRecords are converted via
`dataclasses.asdict`.
The corpus is loaded fresh on every call. This keeps the tool
stateless — the agent can call it from multiple stages without
worrying about caches drifting out of sync. For a 1000-row corpus
the load cost is <50 ms.
"""
from __future__ import annotations
import time
from dataclasses import asdict
from pathlib import Path
from typing import Any, Optional
from tools.base_tool import (
BaseTool,
Determinism,
ExecutionMode,
ResourceProfile,
ToolResult,
ToolRuntime,
ToolStability,
ToolStatus,
ToolTier,
)
class ClipSearch(BaseTool):
name = "clip_search"
version = "0.1.0"
tier = ToolTier.ANALYZE
capability = "clip_retrieval"
provider = "openmontage"
stability = ToolStability.EXPERIMENTAL
execution_mode = ExecutionMode.SYNC
determinism = Determinism.DETERMINISTIC
runtime = ToolRuntime.LOCAL
dependencies = [
"python:numpy",
"python:transformers",
"python:torch",
]
install_instructions = (
"pip install numpy transformers torch\n"
"Requires a corpus built by corpus_builder at <corpus_dir>."
)
agent_skills = []
capabilities = [
"text_to_clip_ranking",
"visual_knn",
"mmr_diversification",
"provenance_lookup",
]
supports = {
"fused_visual_tag_scoring": True,
"motion_filter": True,
"kind_filter": True,
"exclude_list": True,
}
best_for = [
"picking clips for a specific slot in a montage",
"finding collection-style sets from one seed clip",
"de-duplicating a candidate list before edit arrangement",
]
not_good_for = [
"searching the internet (use corpus_builder to populate first)",
"editing or composing video (use video_compose)",
]
input_schema = {
"type": "object",
"required": ["operation", "corpus_dir"],
"properties": {
"operation": {
"type": "string",
"enum": [
"rank_for_slot",
"find_similar_set",
"diversify",
"get",
"stats",
],
},
"corpus_dir": {
"type": "string",
"description": "Path to the corpus built by corpus_builder.",
},
# rank_for_slot
"query_text": {
"type": "string",
"description": "Text description of the scene slot. "
"Embedded by CLIP for similarity ranking.",
},
"k": {"type": "integer", "default": 10, "minimum": 1},
"tag_weight": {
"type": "number",
"default": 0.3,
"minimum": 0.0,
"maximum": 1.0,
"description": "Blend between visual (1-w) and tag (w) channels.",
},
"motion_min": {
"type": "number",
"description": "Reject clips with motion_score below this. "
"Use ~1.5 to filter dead-still clips.",
},
"kind": {
"type": "string",
"enum": ["video", "image"],
"description": "Filter to only one media type.",
},
"exclude_ids": {
"type": "array",
"items": {"type": "string"},
"description": "Clip ids to skip (already used in this edit).",
},
# find_similar_set
"seed_clip_id": {"type": "string"},
"n": {"type": "integer", "default": 5, "minimum": 1},
"diversity": {
"type": "number",
"default": 0.3,
"minimum": 0.0,
"maximum": 1.0,
},
"candidate_pool": {"type": "integer", "default": 30},
# diversify
"candidate_ids": {"type": "array", "items": {"type": "string"}},
# get
"clip_id": {"type": "string"},
},
}
resource_profile = ResourceProfile(
cpu_cores=1, ram_mb=1024, vram_mb=0, disk_mb=50, network_required=False
)
side_effects = []
user_visible_verification = [
"Inspect returned clip_ids and visit thumb_dir/frame_02.jpg "
"to verify the retrieval matches the slot description.",
]
def get_status(self) -> ToolStatus:
try:
import numpy # noqa: F401
import torch # noqa: F401
import transformers # noqa: F401
except ImportError:
return ToolStatus.UNAVAILABLE
return ToolStatus.AVAILABLE
def estimate_cost(self, inputs: dict[str, Any]) -> float:
return 0.0
# ------------------------------------------------------------------
# Execute
# ------------------------------------------------------------------
def execute(self, inputs: dict[str, Any]) -> ToolResult:
start = time.time()
try:
from lib.corpus import Corpus
operation = inputs["operation"]
corpus_dir = Path(inputs["corpus_dir"])
corp = Corpus(corpus_dir)
corp.load()
if operation == "stats":
payload = _op_stats(corp)
elif operation == "rank_for_slot":
payload = _op_rank_for_slot(corp, inputs)
elif operation == "find_similar_set":
payload = _op_find_similar_set(corp, inputs)
elif operation == "diversify":
payload = _op_diversify(corp, inputs)
elif operation == "get":
payload = _op_get(corp, inputs)
else:
return ToolResult(
success=False,
error=f"Unknown operation: {operation!r}",
)
return ToolResult(
success=True,
data={
"operation": operation,
"corpus_dir": str(corpus_dir),
"corpus_size": len(corp),
**payload,
},
duration_seconds=round(time.time() - start, 3),
cost_usd=0.0,
)
except Exception as e:
import traceback
return ToolResult(
success=False,
error=f"{type(e).__name__}: {e}\n{traceback.format_exc()[-800:]}",
)
# ----------------------------------------------------------------------
# Operations
# ----------------------------------------------------------------------
def _op_stats(corp) -> dict[str, Any]:
"""Summary counts and per-source breakdown.
Useful as a sanity check before running expensive retrieval loops —
"does the corpus I loaded actually have enough clips to satisfy the
edit plan?"
"""
import numpy as np
if len(corp) == 0:
return {
"rows": 0,
"per_source": {},
"per_kind": {},
"mean_motion_score": 0.0,
"mean_duration": 0.0,
}
per_source: dict[str, int] = {}
per_kind: dict[str, int] = {}
motion_scores: list[float] = []
durations: list[float] = []
for rec in corp.records:
per_source[rec.source] = per_source.get(rec.source, 0) + 1
per_kind[rec.kind] = per_kind.get(rec.kind, 0) + 1
motion_scores.append(rec.motion_score)
durations.append(rec.duration)
return {
"rows": len(corp),
"per_source": per_source,
"per_kind": per_kind,
"mean_motion_score": float(np.mean(motion_scores)) if motion_scores else 0.0,
"mean_duration": float(np.mean(durations)) if durations else 0.0,
}
def _op_rank_for_slot(corp, inputs: dict[str, Any]) -> dict[str, Any]:
"""Embed `query_text` and return top-k clips by fused similarity.
This is the agent's main retrieval move. The returned list is
ordered best-first and every entry carries a score so the agent
can decide whether the match is strong enough (>= 0.25 is a rough
"acceptable" threshold for CLIP ViT-B/32).
"""
from lib.clip_embedder import embed_texts
query_text = inputs.get("query_text", "").strip()
if not query_text:
raise ValueError("rank_for_slot requires 'query_text'")
q_vec = embed_texts([query_text])[0]
results = corp.rank_by_text(
query_embedding=q_vec,
k=int(inputs.get("k", 10)),
tag_weight=float(inputs.get("tag_weight", 0.3)),
motion_min=inputs.get("motion_min"),
kind=inputs.get("kind"),
exclude_ids=inputs.get("exclude_ids") or [],
)
return {
"query_text": query_text,
"results": [
{"score": score, "record": asdict(rec)}
for rec, score in results
],
}
def _op_find_similar_set(corp, inputs: dict[str, Any]) -> dict[str, Any]:
"""MMR-based similar-set retrieval from one seed clip."""
seed = inputs.get("seed_clip_id")
if not seed:
raise ValueError("find_similar_set requires 'seed_clip_id'")
results = corp.find_similar_set(
seed_clip_id=seed,
n=int(inputs.get("n", 5)),
diversity=float(inputs.get("diversity", 0.3)),
candidate_pool=int(inputs.get("candidate_pool", 30)),
exclude_ids=inputs.get("exclude_ids") or [],
)
return {
"seed_clip_id": seed,
"results": [
{"score": score, "record": asdict(rec)}
for rec, score in results
],
}
def _op_diversify(corp, inputs: dict[str, Any]) -> dict[str, Any]:
"""Pick the most mutually-dissimilar subset of a candidate list."""
candidate_ids = inputs.get("candidate_ids") or []
if not candidate_ids:
raise ValueError("diversify requires 'candidate_ids'")
kept = corp.diversify(
candidate_ids=list(candidate_ids),
n=int(inputs.get("n", 5)),
diversity=float(inputs.get("diversity", 0.5)),
)
return {
"input_count": len(candidate_ids),
"kept_count": len(kept),
"kept_ids": kept,
}
def _op_get(corp, inputs: dict[str, Any]) -> dict[str, Any]:
"""Look up one clip_id and return its full record."""
clip_id = inputs.get("clip_id")
if not clip_id:
raise ValueError("get requires 'clip_id'")
rec = corp.get(clip_id)
if rec is None:
return {"clip_id": clip_id, "found": False, "record": None}
return {"clip_id": clip_id, "found": True, "record": asdict(rec)}
tools/video/corpus_builder.py
+553
View File
@@ -0,0 +1,553 @@
"""Corpus builder: fan out across stock sources, download, thumb, embed, index.
This is the tool the agent calls to populate a local clip corpus for
the documentary-montage pipeline. It is deliberately the ONLY place
where adapters, embedding, and the `Corpus` class meet — everything
downstream (retrieval, selection, edit planning) reads from the corpus
on disk and never touches sources directly.
What it does, per query, per source, per candidate
---------------------------------------------------
1. Call `source.search(query, filters)` to get a flat list of
`Candidate`s normalised across sources.
2. Skip candidates whose `clip_id` is already in the corpus (unless
`skip_existing=false`).
3. Download the file to ``<corpus_dir>/clips/<clip_id>.<ext>``.
4. For videos: extract N evenly-spaced frames to
``<corpus_dir>/thumbnails/<clip_id>/frame_NN.jpg``, probe real
dimensions and duration, and compute a cheap motion score
(mean-abs-diff between first and middle frame).
5. For images: copy the image as ``frame_00.jpg`` in the same thumb
directory so the embedder has a consistent input.
6. Run CLIP on the thumbnails, pool frames to one 512-d vector for the
visual channel. Run CLIP on `source_tags` (falling back to the
query itself) for the tag channel.
7. Materialise a `ClipRecord` with every provenance field and append
it to the corpus via `Corpus.add()`.
8. After ALL candidates are processed, call `Corpus.save()` once. Per-
add saves would burn disk I/O for large runs.
Caps
----
- `max_new_clips` halts the whole run once that many new rows have
been added. The remaining candidates in the current loop iteration
are discarded.
- Per-source search errors and per-candidate processing errors are
caught and collected into `errors` in the return payload. One flaky
URL or one broken codec must not poison the whole run.
Idempotence
-----------
Re-running with the same inputs is safe. `skip_existing=True` (default)
causes the tool to short-circuit on any `clip_id` already in the
corpus JSONL. Crash-recovery is handled by `Corpus.load()`, which
truncates the in-memory state to the shorter of the JSONL and the
`.npy` lengths.
Agent surface
-------------
Input schema keeps the agent's decisions at the top: WHAT to search
for (`queries`), WHERE to search (`sources`), WHAT to filter
(`filters`), and HOW MUCH (`max_new_clips`). Everything else has
sensible defaults.
"""
from __future__ import annotations
import time
import urllib.parse
from pathlib import Path
from typing import Any, Optional
from tools.base_tool import (
BaseTool,
Determinism,
ExecutionMode,
ResourceProfile,
ToolResult,
ToolRuntime,
ToolStability,
ToolStatus,
ToolTier,
)
class CorpusBuilder(BaseTool):
name = "corpus_builder"
version = "0.1.0"
tier = ToolTier.SOURCE
capability = "corpus_population"
provider = "openmontage"
stability = ToolStability.EXPERIMENTAL
execution_mode = ExecutionMode.SYNC
determinism = Determinism.DETERMINISTIC
runtime = ToolRuntime.HYBRID # local compute + network APIs
dependencies = [
"python:cv2",
"python:numpy",
"python:requests",
"python:PIL",
"python:transformers",
"python:torch",
]
install_instructions = (
"pip install opencv-python numpy requests pillow transformers torch\n"
"At least one stock source must be configured:\n"
" PEXELS_API_KEY for Pexels (free at https://www.pexels.com/api/)\n"
" archive.org and nasa work without API keys"
)
agent_skills = []
capabilities = [
"stock_fanout_search",
"corpus_population",
"clip_indexing",
"clip_embedding",
]
supports = {
"multi_source": True,
"video_and_image": True,
"append_only": True,
"resumable": True,
}
best_for = [
"documentary-montage retrieval corpora",
"topic-based offline clip indexing",
"collecting candidate B-roll without repeated API calls per edit",
]
not_good_for = [
"single-clip downloads (use pexels_video instead)",
"semantic retrieval itself (use clip_search)",
]
fallback_tools = []
input_schema = {
"type": "object",
"required": ["corpus_dir", "queries"],
"properties": {
"corpus_dir": {
"type": "string",
"description": "Project-local corpus directory, e.g. projects/foo/corpus",
},
"queries": {
"type": "array",
"minItems": 1,
"items": {
"type": "object",
"required": ["query"],
"properties": {
"query": {"type": "string"},
"kind": {
"type": "string",
"enum": ["video", "image", "any"],
"default": "video",
},
"per_source": {
"type": "integer",
"default": 10,
"minimum": 1,
"maximum": 80,
},
},
},
},
"sources": {
"type": "array",
"items": {"type": "string"},
"description": "Source names to use (e.g. ['pexels','archive_org']). "
"Defaults to all available.",
},
"filters": {
"type": "object",
"properties": {
"min_duration": {"type": "number"},
"max_duration": {"type": "number"},
"orientation": {
"type": "string",
"enum": ["landscape", "portrait", "square"],
},
"min_width": {"type": "integer"},
},
},
"max_new_clips": {
"type": "integer",
"default": 100,
"minimum": 1,
"description": "Halt after this many NEW rows have been added.",
},
"skip_existing": {"type": "boolean", "default": True},
"thumbs_per_video": {
"type": "integer",
"default": 5,
"minimum": 1,
"maximum": 20,
},
},
}
resource_profile = ResourceProfile(
cpu_cores=2, ram_mb=2048, vram_mb=0, disk_mb=4000, network_required=True
)
side_effects = [
"downloads clips to <corpus_dir>/clips",
"writes thumbnails under <corpus_dir>/thumbnails",
"appends rows to <corpus_dir>/index.jsonl + embedding .npy files",
"calls external stock APIs",
]
user_visible_verification = [
"Open <corpus_dir>/index.jsonl and inspect a few added rows",
"Open <corpus_dir>/thumbnails/<some_clip_id>/frame_02.jpg visually",
]
def get_status(self) -> ToolStatus:
try:
from tools.video.stock_sources import available_sources
except Exception:
return ToolStatus.UNAVAILABLE
return ToolStatus.AVAILABLE if available_sources() else ToolStatus.UNAVAILABLE
def estimate_cost(self, inputs: dict[str, Any]) -> float:
return 0.0 # all sources are free-tier
# ------------------------------------------------------------------
# Execute
# ------------------------------------------------------------------
def execute(self, inputs: dict[str, Any]) -> ToolResult:
start = time.time()
try:
from lib.corpus import Corpus
from tools.video.stock_sources import (
SearchFilters,
available_sources,
get_source,
)
corpus_dir = Path(inputs["corpus_dir"])
queries: list[dict] = list(inputs["queries"])
source_names: Optional[list[str]] = inputs.get("sources")
filters_in: dict = inputs.get("filters") or {}
max_new = int(inputs.get("max_new_clips", 100))
skip_existing = bool(inputs.get("skip_existing", True))
thumbs_per_video = int(inputs.get("thumbs_per_video", 5))
# Resolve sources. If the caller passed an explicit list we
# respect it even if some are unavailable — their search
# calls will simply fail and be logged.
if source_names:
requested: list = []
for name in source_names:
try:
s = get_source(name)
except KeyError as e:
return ToolResult(success=False, error=str(e))
if s.is_available():
requested.append(s)
sources = requested
else:
sources = available_sources()
if not sources:
return ToolResult(
success=False,
error="No stock sources available. " + self.install_instructions,
)
corp = Corpus(corpus_dir)
corp.load()
corp.ensure_dirs()
per_source_counts: dict[str, int] = {s.name: 0 for s in sources}
added_ids: list[str] = []
errors: list[dict] = []
skipped = 0
failed = 0
candidates_seen = 0
def filters_for(q_spec: dict) -> SearchFilters:
return SearchFilters(
kind=q_spec.get("kind", "video"),
per_page=int(q_spec.get("per_source", 10)),
min_duration=filters_in.get("min_duration"),
max_duration=filters_in.get("max_duration"),
orientation=filters_in.get("orientation"),
min_width=filters_in.get("min_width"),
)
for q_spec in queries:
if len(added_ids) >= max_new:
break
query = q_spec["query"]
f = filters_for(q_spec)
for src in sources:
if len(added_ids) >= max_new:
break
try:
cands = src.search(query, f)
except Exception as e:
errors.append({
"phase": "search",
"source": src.name,
"query": query,
"error": f"{type(e).__name__}: {e}",
})
continue
candidates_seen += len(cands)
for cand in cands:
if len(added_ids) >= max_new:
break
if skip_existing and corp.has(cand.clip_id):
skipped += 1
continue
try:
rec = self._process_candidate(
cand=cand,
src=src,
corp=corp,
query=query,
thumbs_per_video=thumbs_per_video,
)
except Exception as e:
failed += 1
errors.append({
"phase": "process",
"clip_id": cand.clip_id,
"error": f"{type(e).__name__}: {e}",
})
continue
if rec is None:
failed += 1
continue
added_ids.append(rec.clip_id)
per_source_counts[src.name] = per_source_counts.get(src.name, 0) + 1
# Single save at the end. Corpus.save() writes JSONL first
# (source of truth) then both .npy files, so a crash mid-save
# still leaves a loadable corpus.
corp.save()
elapsed = time.time() - start
return ToolResult(
success=True,
data={
"corpus_dir": str(corpus_dir),
"queries_run": len(queries),
"candidates_seen": candidates_seen,
"clips_added": len(added_ids),
"clips_skipped_existing": skipped,
"clips_failed": failed,
"per_source_counts": per_source_counts,
"added_ids": added_ids,
"total_corpus_size": len(corp),
"errors": errors[:25], # cap log noise
},
cost_usd=0.0,
duration_seconds=round(elapsed, 2),
)
except Exception as e:
import traceback
return ToolResult(
success=False,
error=f"{type(e).__name__}: {e}\n{traceback.format_exc()[-800:]}",
)
# ------------------------------------------------------------------
# Per-candidate pipeline
# ------------------------------------------------------------------
def _process_candidate(
self,
cand,
src,
corp,
query: str,
thumbs_per_video: int,
):
"""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).
"""
import cv2
from lib.clip_embedder import embed_images, embed_texts, pool_frames
from lib.corpus import ClipRecord
# Pick file extension from the URL path (sources give us
# stable .mp4/.jpg/.png URLs) with a kind-aware fallback.
ext = _guess_ext(cand)
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:
if local_abs.exists():
local_abs.unlink()
except OSError:
pass
return None
thumb_dir_rel = Path("thumbnails") / cand.clip_id
thumb_dir_abs = corp.corpus_dir / thumb_dir_rel
thumb_dir_abs.mkdir(parents=True, exist_ok=True)
width = cand.width
height = cand.height
duration = cand.duration
motion_score = 0.0
if cand.kind == "video":
thumb_paths, probe = _extract_video_thumbs(
local_abs, thumb_dir_abs, thumbs_per_video
)
if not thumb_paths:
return None
if probe:
width = probe.get("width") or width
height = probe.get("height") or height
duration = probe.get("duration") or duration
motion_score = float(probe.get("motion_score", 0.0))
else:
dst = thumb_dir_abs / "frame_00.jpg"
if not _save_as_jpeg(local_abs, dst):
return None
thumb_paths = [dst]
img = cv2.imread(str(local_abs))
if img is not None:
height, width = img.shape[:2]
# CLIP embeddings. embed_images loads the model lazily on
# first call; subsequent candidates reuse the cached model.
clip_frames = embed_images(thumb_paths)
clip_vec = pool_frames(clip_frames)
# Tag channel: prefer source-supplied tags/description,
# fall back to the query so the row still carries SOME text
# signal. pool_frames returns zeros if empty so this never
# breaks the fused ranking math.
tag_text = cand.source_tags or query
tag_vec = embed_texts([tag_text])[0]
rec = ClipRecord(
clip_id=cand.clip_id,
source=cand.source,
source_id=cand.source_id,
source_url=cand.source_url,
local_path=str(local_rel).replace("\\", "/"),
kind=cand.kind,
thumb_dir=str(thumb_dir_rel).replace("\\", "/"),
query=query,
creator=cand.creator,
license=cand.license,
duration=float(duration or 0.0),
width=int(width or 0),
height=int(height or 0),
motion_score=motion_score,
dominant_colors=[],
source_tags=cand.source_tags,
)
corp.add(rec, clip_vec, tag_vec)
return rec
# ----------------------------------------------------------------------
# Module-level helpers (kept outside the class so tests can hit them)
# ----------------------------------------------------------------------
def _guess_ext(cand) -> str:
"""Extract a sensible file extension from a candidate's URL."""
known = {".mp4", ".mov", ".mkv", ".webm", ".ogv", ".m4v",
".jpg", ".jpeg", ".png", ".tif", ".tiff"}
path = urllib.parse.urlparse(cand.download_url).path
ext = Path(path).suffix.lower()
if ext in known:
# Normalise .jpeg→.jpg for consistent clip paths
return ".jpg" if ext == ".jpeg" else ext
return ".mp4" if cand.kind == "video" else ".jpg"
def _extract_video_thumbs(
video_path: Path, out_dir: Path, n_frames: int
) -> tuple[list[Path], dict]:
"""Extract `n` evenly-spaced JPEG thumbnails from a video.
Returns ``(thumb_paths, probe_dict)``. The probe dict carries the
real dimensions, duration, and a cheap motion score (mean abs
pixel diff between frame 0 and the middle frame). Used to backfill
`ClipRecord` fields that the source API didn't give us.
"""
import cv2
cap = cv2.VideoCapture(str(video_path))
if not cap.isOpened():
return [], {}
total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
fps = float(cap.get(cv2.CAP_PROP_FPS) or 0)
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0)
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0)
duration = total / fps if fps > 0 else 0.0
if total < 2:
cap.release()
return [], {}
n = max(1, min(n_frames, total))
# Space positions at the midpoints of equal segments so we skip the
# first/last frame (often black or a splash).
positions = [int(round((i + 0.5) * total / n)) for i in range(n)]
thumb_paths: list[Path] = []
captured: list = []
for idx, pos in enumerate(positions):
cap.set(cv2.CAP_PROP_POS_FRAMES, max(0, min(pos, total - 1)))
ok, frame = cap.read()
if not ok or frame is None:
continue
dst = out_dir / f"frame_{idx:02d}.jpg"
cv2.imwrite(str(dst), frame, [int(cv2.IMWRITE_JPEG_QUALITY), 88])
thumb_paths.append(dst)
captured.append(frame)
cap.release()
motion = 0.0
if len(captured) >= 2:
import numpy as np
a = cv2.cvtColor(captured[0], cv2.COLOR_BGR2GRAY).astype(np.float32)
b = cv2.cvtColor(captured[len(captured) // 2], cv2.COLOR_BGR2GRAY).astype(np.float32)
motion = float(np.abs(a - b).mean())
return thumb_paths, {
"width": width,
"height": height,
"duration": duration,
"motion_score": motion,
}
def _save_as_jpeg(src_path: Path, dst_path: Path) -> bool:
"""Load an arbitrary image file and re-save as JPEG.
Handles PNG/JPEG/TIFF/WebP inputs — anything cv2.imread understands.
Returns True on success, False on unreadable input (so the caller
can reject the candidate cleanly).
"""
import cv2
img = cv2.imread(str(src_path))
if img is None:
return False
cv2.imwrite(str(dst_path), img, [int(cv2.IMWRITE_JPEG_QUALITY), 88])
return True
tools/video/stock_sources/__init__.py
+80
View File
@@ -0,0 +1,80 @@
"""Stock media source adapters.
Unified-protocol wrappers around free stock APIs (Pexels, Archive.org,
NASA, ...) used by `corpus_builder` to populate the local clip corpus.
See `base.py` for the protocol contract and the "adding a new source"
checklist.
Registry
--------
`all_sources()` returns one instance of every adapter known to the
package, in a stable order. `available_sources()` filters to the ones
whose `is_available()` returns True right now. The corpus builder
queries the latter during fan-out; it never imports adapters directly.
To register a new adapter: add its class to the `_REGISTRY` tuple
below. Order matters only as a tiebreak when two sources return
matching clips.
"""
from __future__ import annotations
from .archive_org import ArchiveOrgSource
from .base import Candidate, SearchFilters, StockSource
from .nasa import NasaSource
from .pexels import PexelsSource
__all__ = [
"Candidate",
"SearchFilters",
"StockSource",
"PexelsSource",
"ArchiveOrgSource",
"NasaSource",
"all_sources",
"available_sources",
"get_source",
]
# Explicit, ordered list of every adapter class the package exposes.
# The corpus builder iterates this (filtered by availability) during
# fan-out. Order matters only as a tiebreak for identical clip ids.
_REGISTRY: tuple[type, ...] = (
PexelsSource,
ArchiveOrgSource,
NasaSource,
)
def all_sources() -> list[StockSource]:
"""Instantiate every registered adapter, whether available or not.
Returned instances are cheap — adapters keep no state beyond env
var reads, so constructing them has no cost. Use this when you
want to show the user what sources exist regardless of whether
their credentials are configured.
"""
return [cls() for cls in _REGISTRY]
def available_sources() -> list[StockSource]:
"""Return only the adapters whose `is_available()` is True.
This is what the corpus builder uses during a normal run. An empty
list means no sources are configured — the caller should surface
that to the user with install instructions, not silently produce
an empty corpus.
"""
return [s for s in all_sources() if s.is_available()]
def get_source(name: str) -> StockSource:
"""Look up a single adapter by its `name` attribute.
Raises `KeyError` if no registered adapter claims that name. Useful
for tests and for agents that want to pin to a specific source
(e.g. "only Archive.org for this topic").
"""
for s in all_sources():
if s.name == name:
return s
raise KeyError(f"No stock source registered with name={name!r}")
tools/video/stock_sources/archive_org.py
+368
View File
@@ -0,0 +1,368 @@
"""Archive.org stock video adapter.
Targets public-domain film and home-movie collections on archive.org:
- **prelinger**: Rick Prelinger's archive of ephemeral films
(industrial, educational, advertising, 1920s-1980s). The single
best source of documentary-grade B-roll for 20th century themes.
- **opensource_movies**: broader public-domain and CC-licensed film
uploads — a grab bag but useful when Prelinger is thin on a topic.
- **home_movies**: anonymous personal footage. The soul of any
nostalgic or observational documentary montage lives here.
Archive.org requires no API key. Everything is open.
Fetch pattern
-------------
The search API (``advancedsearch.php``) returns metadata records with
identifiers but not file lists. To get downloadable URLs we need a
second call to ``/metadata/<identifier>`` per hit. This adapter pays
that round-trip cost during `search()` so the `Candidate` it returns
carries a ready-to-use `download_url`. For a per_page of 20, expect
~21 HTTP calls per search — slow but fully serial and cache-friendly.
Dimensions are sometimes missing from file metadata. When they are,
the `Candidate` carries `width=0, height=0` and the corpus builder is
expected to probe the clip with ffprobe post-download.
Licence
-------
Prelinger collection items are public domain. Broader
`opensource_movies` items are usually CC0 or CC-BY. We record the
collection and `licenseurl` (when present) so the agent can attribute
correctly if it wants to, but no attribution is legally required.
"""
from __future__ import annotations
import re
from pathlib import Path
from typing import Any, Optional
from .base import Candidate, SearchFilters
_SEARCH_URL = "https://archive.org/advancedsearch.php"
_METADATA_URL = "https://archive.org/metadata"
_DOWNLOAD_URL = "https://archive.org/download"
# Default collections to bias toward. Overridable via SearchFilters.extra
# in a future refinement; for now these are baked in because they give
# the best documentary-montage hit rate.
_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.
_VIDEO_FORMAT_PRIORITY = (
"h.264", # mp4, usually 480p or 720p
"MPEG4", # older mp4 encoding
"h.264 HD", # less common, but best quality
"512Kb MPEG4", # last resort — lowest quality derivative
"Matroska", # mkv
"WebM", # webm
)
class ArchiveOrgSource:
"""Adapter for public-domain video on archive.org.
Satisfies `StockSource`. Stateless, no credentials.
"""
name = "archive_org"
def is_available(self) -> bool:
# No API key, no config. As long as the network is up, we're
# available. The corpus builder will catch network errors in
# the per-source try block.
return True
# ------------------------------------------------------------------
# Public protocol
# ------------------------------------------------------------------
def search(self, query: str, filters: SearchFilters) -> list[Candidate]:
"""Search Archive.org for matching video items.
Two-stage: (1) advancedsearch for identifiers, (2) per-item
metadata fetch for file lists. Images are not supported — this
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).
"""
kind = (filters.kind or "video").lower()
if kind not in ("video", "any"):
return []
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"),
]
r = requests.get(_SEARCH_URL, params=params, timeout=30)
r.raise_for_status()
data = r.json()
docs = (data.get("response") or {}).get("docs", []) or []
out: list[Candidate] = []
for doc in docs:
cand = self._hydrate_candidate(doc, filters)
if cand is not None:
out.append(cand)
return out
def download(self, candidate: Candidate, out_path: Path) -> Path:
"""Stream the candidate's file to `out_path`.
Same pattern as the Pexels adapter — no caching, corpus builder
decides.
"""
import requests # lazy
if not candidate.download_url:
raise ValueError(
f"Candidate {candidate.clip_id} has no download_url"
)
out_path = Path(out_path)
out_path.parent.mkdir(parents=True, exist_ok=True)
with requests.get(
candidate.download_url, stream=True, timeout=300
) as r:
r.raise_for_status()
with open(out_path, "wb") as f:
for chunk in r.iter_content(chunk_size=1 << 16):
if chunk:
f.write(chunk)
return out_path
# ------------------------------------------------------------------
# Internals
# ------------------------------------------------------------------
def _build_query(self, user_query: str) -> str:
"""Wrap the user's query with mediatype + collection filters.
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.
"""
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})'
def _hydrate_candidate(
self, doc: dict, filters: SearchFilters
) -> Optional[Candidate]:
"""Turn a search-doc identifier into a full Candidate.
Fetches the item's file list and picks the best playable
rendition. Returns None if the item has no usable files or if
the item's duration falls outside the filter range.
"""
import requests # lazy
identifier = doc.get("identifier")
if not identifier:
return None
try:
r = requests.get(f"{_METADATA_URL}/{identifier}", timeout=30)
r.raise_for_status()
meta = r.json()
except Exception:
# Swallow per-item fetch failures — one bad item shouldn't
# poison the whole search. Alternative would be to raise and
# have corpus_builder catch per-source, but at this layer we
# can keep going.
return None
files = meta.get("files") or []
picked = _pick_video_file(files)
if picked is None:
return None
duration = _parse_length(picked.get("length"))
if filters.min_duration is not None 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:
return None
width = _safe_int(picked.get("width"))
height = _safe_int(picked.get("height"))
if filters.min_width is not None and width and width < filters.min_width:
return None
# Build the direct download URL. archive.org/download/<id>/<file>
# is the stable public pattern and works without auth.
file_name = picked.get("name", "")
download_url = f"{_DOWNLOAD_URL}/{identifier}/{file_name}"
# Tags: flatten title + description + subject. Archive.org is
# verbose, so we truncate to keep the CLIP text encoder focused
# on the most important tokens (77-token limit anyway).
title = _to_text(doc.get("title"))
description = _to_text(doc.get("description"))
subject = _to_text(doc.get("subject"))
source_tags = " ".join(
s for s in (title, description, subject) if s
).strip()
if len(source_tags) > 500:
source_tags = source_tags[:500]
creator = _to_text(doc.get("creator"))
collection = _to_text(doc.get("collection"))
license_url = _to_text(doc.get("licenseurl"))
license_text = license_url or _license_from_collection(collection)
return Candidate(
source=self.name,
source_id=identifier,
source_url=f"https://archive.org/details/{identifier}",
download_url=download_url,
kind="video",
width=width,
height=height,
duration=duration,
creator=creator,
license=license_text,
source_tags=source_tags,
thumbnail_url=f"https://archive.org/services/img/{identifier}",
extra={
"collection": collection,
"date": _to_text(doc.get("date")),
"format": picked.get("format"),
"file_name": file_name,
"file_size_bytes": _safe_int(picked.get("size")),
},
)
# ----------------------------------------------------------------------
# Module-level helpers
# ----------------------------------------------------------------------
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.
"""
if not files:
return None
by_format: dict[str, list[dict]] = {}
for f in files:
fmt = (f.get("format") or "").strip()
name = (f.get("name") or "").lower()
# Skip non-video
if fmt not in _VIDEO_FORMAT_PRIORITY:
continue
# Skip degenerate renditions and thumbnails regardless of format
if any(tag in name for tag in ("thumb", "preview", ".gif")):
continue
by_format.setdefault(fmt, []).append(f)
for fmt in _VIDEO_FORMAT_PRIORITY:
bucket = by_format.get(fmt)
if not bucket:
continue
bucket.sort(key=lambda f: _safe_int(f.get("size")), reverse=True)
return bucket[0]
return None
_HMS_RE = re.compile(r"^(\d+):(\d+):(\d+(?:\.\d+)?)$")
_MS_RE = re.compile(r"^(\d+):(\d+(?:\.\d+)?)$")
def _parse_length(value: Any) -> float:
"""Parse Archive.org's `length` field into seconds.
The field is usually "HH:MM:SS.ss" or "MM:SS.ss" or a bare float
as a string. Missing / unparseable values return 0.0 which the
caller interprets as "unknown, pass any duration filter".
"""
if value is None:
return 0.0
if isinstance(value, (int, float)):
return float(value)
s = str(value).strip()
if not s:
return 0.0
m = _HMS_RE.match(s)
if m:
h, mn, sec = m.groups()
return int(h) * 3600 + int(mn) * 60 + float(sec)
m = _MS_RE.match(s)
if m:
mn, sec = m.groups()
return int(mn) * 60 + float(sec)
try:
return float(s)
except ValueError:
return 0.0
def _safe_int(value: Any) -> int:
"""Parse a value to int, tolerating strings, None, and garbage."""
if value is None:
return 0
try:
return int(value)
except (TypeError, ValueError):
try:
return int(float(value))
except (TypeError, ValueError):
return 0
def _to_text(value: Any) -> str:
"""Flatten Archive.org's sometimes-list-sometimes-string fields.
The search API returns `creator`, `subject`, and `description` as
either a string or a list of strings depending on the item. We
always join with spaces so the caller sees a single str.
"""
if value is None:
return ""
if isinstance(value, list):
return " ".join(str(x) for x in value if x is not None).strip()
return str(value).strip()
def _license_from_collection(collection: str) -> str:
"""Infer licence text from a collection name when no licenseurl is set.
Prelinger items are universally public domain; broader opensource
collections usually are too but we're less sure, so we hedge.
"""
col = collection.lower()
if "prelinger" in col:
return "Public Domain (Prelinger Archives)"
if "home_movies" in col:
return "Public Domain (archive.org home movies)"
return "Public Domain / CC (archive.org — verify per item)"
tools/video/stock_sources/base.py
+140
View File
@@ -0,0 +1,140 @@
"""Unified protocol for stock media source adapters.
Every source (Pexels, Archive.org, NASA, Wikimedia, Unsplash, ...)
implements the same small interface. The corpus builder fans out across
all enabled sources, normalises their results into `Candidate` objects,
downloads what it decides to keep, and writes everything to the local
corpus. During selection the agent never sees which source a clip came
from — it only sees `ClipRecord` rows retrieved by similarity.
Design intent
-------------
- **Separation of concerns.** Adapters handle API shape, licensing
metadata, and downloading. `lib/corpus.py` handles indexing and
retrieval math. The agent handles WHAT to search for and WHICH
results to accept.
- **Interchangeable.** Any adapter satisfies the same protocol, so the
corpus builder can iterate `for src in sources: src.search(q, f)`
without branching on source type.
- **Dumb by design.** No ranking, no de-dup, no filtering beyond what
the API itself accepts. Judgment work happens after embedding, in
the agent. Adapters just convert "API JSON" → "normalised Candidate".
Adding a new source
-------------------
1. Create `tools/video/stock_sources/<name>.py` with a class that
satisfies `StockSource` — a `name` attribute plus `is_available`,
`search`, and `download` methods.
2. Normalise the API's response into `Candidate` objects. Keep every
field the corpus might later want to display or attribute (creator,
licence, source URL, description/tags for the text channel of the
CLIP fused ranking).
3. Register the class so `corpus_builder` can discover it. The
registration pattern lives alongside the first concrete adapter.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Optional, Protocol, runtime_checkable
@dataclass
class Candidate:
"""One pre-download search result, normalised across sources.
A `Candidate` is what a source adapter returns from `search()`. It
carries everything the corpus builder needs to decide whether to
download, plus the provenance fields that later get copied into the
`ClipRecord`.
Fields that don't apply to a given kind default to zero/empty — for
an image, `duration` is 0.0; for a source that doesn't expose fps,
`extra` may hold it or it may be absent entirely. Adapters should
populate as much as they cheaply can and leave the rest alone.
"""
source: str # adapter name, e.g. "pexels"
source_id: str # unique within that source
source_url: str # landing page (human readable)
download_url: str # direct file URL
kind: str # "video" or "image"
width: int = 0
height: int = 0
duration: float = 0.0 # seconds (0 for images)
creator: str = "" # attribution name
license: str = "" # licence string or URL
source_tags: str = "" # title + description + tags joined
thumbnail_url: str = "" # for previews and image-fallback embeds
extra: dict[str, Any] = field(default_factory=dict) # source-specific junk
@property
def clip_id(self) -> str:
"""Stable ID used as the corpus row key.
Format is ``"<source>_<source_id>"``. Matches the convention in
`lib/corpus.ClipRecord`, so the corpus builder can copy this
directly when it materialises the row.
"""
return f"{self.source}_{self.source_id}"
@dataclass
class SearchFilters:
"""Filters a source adapter applies when searching.
Not every source supports every filter — adapters MAY ignore fields
they don't understand. The corpus builder sets these liberally and
trusts each source to do the best it can with what it has. The
missing filters are caught later by the agent at retrieval time.
"""
kind: str = "video" # "video", "image", or "any"
min_duration: Optional[float] = None # seconds; None = no floor
max_duration: Optional[float] = None # seconds; None = no ceiling
orientation: Optional[str] = None # "landscape" | "portrait" | "square"
min_width: Optional[int] = None # resolution floor in pixels
per_page: int = 20
page: int = 1
@runtime_checkable
class StockSource(Protocol):
"""Protocol every stock source adapter must satisfy.
Attributes
----------
name:
Stable key used in the corpus (becomes the prefix of each
`ClipRecord.clip_id`) and in any agent-facing source list. Do
NOT change it after a corpus has been built against it —
existing rows would orphan.
Methods
-------
is_available:
Cheap, non-network check. Answers "does the environment have
the keys / dependencies this source needs?" The corpus builder
silently skips unavailable sources during fan-out.
search:
Returns a flat list of `Candidate` objects in whatever order
the source's own relevance ranking gave them. An empty list is
legal (and common for niche queries). Network errors should be
raised — the corpus builder catches and logs per-source so one
flaky API doesn't poison the whole run.
download:
Saves the candidate's file to `out_path` (creating parent
directories if needed) and returns the final path. This is a
blocking call; any caching is the corpus builder's
responsibility, not the adapter's.
"""
name: str
def is_available(self) -> bool: ...
def search(self, query: str, filters: SearchFilters) -> list[Candidate]: ...
def download(self, candidate: Candidate, out_path: Path) -> Path: ...
tools/video/stock_sources/nasa.py
+330
View File
@@ -0,0 +1,330 @@
"""NASA Image and Video Library adapter.
Wraps ``images-api.nasa.gov`` behind the `StockSource` protocol. NASA's
catalogue is the best free source of space, mission, Earth-observation,
and scientific-visualisation footage and stills. Everything on
images.nasa.gov is free to use (public domain / NASA-created), with the
caveat that third-party material occasionally sneaks in via partner
missions — we record that caveat in the licence field and trust the
user to vet before publishing.
No API key required. An optional `NASA_API_KEY` increases rate limits
but is not part of this adapter's availability check — the default
`DEMO_KEY` path handles typical corpus-building loads.
Fetch pattern
-------------
Like `archive_org`, this is a two-stage fetch. The search endpoint
returns items with an `href` pointing at a per-item asset manifest (a
plain JSON array of file URLs). We follow that manifest to pick the
best rendition and build a `Candidate` with a ready-to-use
`download_url`. One extra HTTP call per hit.
What NASA is good for
---------------------
- space missions, rockets, astronauts,
- Earth observation (weather, city lights, glaciers, hurricanes),
- planetary-science animation,
- historical aeronautics (NACA, shuttle era),
- any "tiny human on a big planet" interstitial.
What it is *not* good for: most documentary subjects that live on
Earth at human scale. For a "walking home" montage this adapter will
contribute atmosphere shots at best (sunrise over Earth, city lights
from orbit) — the core material comes from Pexels and Archive.org.
"""
from __future__ import annotations
import os
import re
from pathlib import Path
from typing import Any, Optional
from urllib.parse import quote, urlparse, urlunparse
from .base import Candidate, SearchFilters
_SEARCH_URL = "https://images-api.nasa.gov/search"
_UNSAFE_ID_CHARS = re.compile(r"[^A-Za-z0-9._\-]+")
class NasaSource:
"""Adapter for images.nasa.gov.
Satisfies `StockSource`. Stateless, no required credentials.
"""
name = "nasa"
def is_available(self) -> bool:
# The public API is unauthenticated; as long as the network is
# reachable we're available. `NASA_API_KEY` is optional and
# only affects rate limiting.
return True
# ------------------------------------------------------------------
# Public protocol
# ------------------------------------------------------------------
def search(self, query: str, filters: SearchFilters) -> list[Candidate]:
"""Search NASA's media library for images and/or videos.
Routes by `filters.kind`:
"video" → `media_type=video`
"image" → `media_type=image`
"any" → both (NASA accepts repeated ``media_type`` params)
"""
import requests # lazy
kind = (filters.kind or "video").lower()
media_types: list[str] = []
if kind in ("video", "any"):
media_types.append("video")
if kind in ("image", "any"):
media_types.append("image")
if not media_types:
return []
params: list[tuple[str, str]] = [("q", query)]
for mt in media_types:
params.append(("media_type", mt))
# NASA uses `page_size` (max 100) and `page` (1-indexed).
params.append(("page_size", str(max(1, min(filters.per_page, 100)))))
params.append(("page", str(max(1, filters.page))))
r = requests.get(_SEARCH_URL, params=params, timeout=30)
r.raise_for_status()
data = r.json()
items = ((data.get("collection") or {}).get("items") or [])
out: list[Candidate] = []
for item in items:
cand = self._hydrate_candidate(item, filters)
if cand is not None:
out.append(cand)
return out
def download(self, candidate: Candidate, out_path: Path) -> Path:
"""Stream the candidate's file to `out_path`.
Same pattern as the other adapters.
"""
import requests # lazy
if not candidate.download_url:
raise ValueError(
f"Candidate {candidate.clip_id} has no download_url"
)
out_path = Path(out_path)
out_path.parent.mkdir(parents=True, exist_ok=True)
with requests.get(
candidate.download_url, stream=True, timeout=300
) as r:
r.raise_for_status()
with open(out_path, "wb") as f:
for chunk in r.iter_content(chunk_size=1 << 16):
if chunk:
f.write(chunk)
return out_path
# ------------------------------------------------------------------
# Internals
# ------------------------------------------------------------------
def _hydrate_candidate(
self, item: dict, filters: SearchFilters
) -> Optional[Candidate]:
"""Turn a search-result item into a Candidate with a download URL.
Fetches the per-item asset manifest (``item['href']`` is a
plain JSON array of file URLs) and picks the best rendition.
Returns None if the manifest is empty or no playable file
exists.
"""
import requests # lazy
data_list = item.get("data") or []
if not data_list:
return None
meta = data_list[0]
nasa_id = meta.get("nasa_id")
media_type = (meta.get("media_type") or "").lower()
if not nasa_id or media_type not in ("video", "image"):
return None
asset_href = item.get("href")
if not asset_href:
return None
try:
r = requests.get(asset_href, timeout=30)
r.raise_for_status()
file_urls = r.json()
except Exception:
return None
if not isinstance(file_urls, list) or not file_urls:
return None
if media_type == "video":
download_url = _pick_video_url(file_urls)
else:
download_url = _pick_image_url(file_urls)
if not download_url:
return None
# NASA asset URLs sometimes contain unencoded spaces and other
# characters (especially older items whose nasa_id is a title).
# URL-encode the path component so `requests.get` handles them.
download_url = _encode_url_path(download_url)
# NASA doesn't give us dimensions or duration on the search
# result. Width/height default to 0 and the corpus builder is
# expected to probe post-download. Duration is 0 for both
# images and (unfortunately) videos.
width = 0
height = 0
duration = 0.0
# Filters that need dims/duration can't apply here. Skipping
# them is the honest thing — "unknown" should not be rejected.
title = (meta.get("title") or "").strip()
description = (meta.get("description") or "").strip()
keywords = meta.get("keywords") or []
if isinstance(keywords, list):
kw_text = " ".join(str(k) for k in keywords if k)
else:
kw_text = str(keywords)
source_tags = " ".join(
s for s in (title, description, kw_text) if s
).strip()
if len(source_tags) > 500:
source_tags = source_tags[:500]
# Thumbnail: NASA exposes a "preview" link in item["links"]
thumbnail_url = ""
for link in item.get("links") or []:
if (link or {}).get("rel") == "preview":
thumbnail_url = link.get("href", "") or ""
break
creator = (meta.get("photographer") or meta.get("center") or "").strip()
safe_id = _sanitize_source_id(nasa_id)
return Candidate(
source=self.name,
source_id=safe_id,
source_url=f"https://images.nasa.gov/details/{quote(nasa_id, safe='')}",
download_url=download_url,
kind=media_type,
width=width,
height=height,
duration=duration,
creator=creator,
license="NASA Media Usage Guidelines (public domain with caveats)",
source_tags=source_tags,
thumbnail_url=thumbnail_url,
extra={
"center": meta.get("center"),
"date_created": meta.get("date_created"),
"secondary_creator": meta.get("secondary_creator"),
},
)
# ----------------------------------------------------------------------
# Module-level helpers
# ----------------------------------------------------------------------
def _pick_video_url(file_urls: list[str]) -> str:
"""Pick the best video rendition from a NASA asset manifest.
NASA file URLs follow a ``<nasa_id>~<tag>.<ext>`` naming convention
where tag is one of: ``orig``, ``large``, ``medium``, ``small``,
``preview``, ``thumb``. Preference order (best quality first):
orig → large → medium → small. We skip previews and thumbnails.
"""
priority = ("orig", "large", "medium", "small")
buckets: dict[str, list[str]] = {p: [] for p in priority}
for url in file_urls:
lower = url.lower()
if not lower.endswith((".mp4", ".mov", ".m4v")):
continue
for p in priority:
if f"~{p}." in lower:
buckets[p].append(url)
break
for p in priority:
if buckets[p]:
return buckets[p][0]
# Fallback: any mp4 at all
for url in file_urls:
if url.lower().endswith(".mp4"):
return url
return ""
def _pick_image_url(file_urls: list[str]) -> str:
"""Pick the best image rendition from a NASA asset manifest.
Same pattern as videos but for .jpg/.png/.tif. Prefer ``orig``,
fall back to ``large``. We skip ``thumb`` and ``small`` because
they're too low-res for CLIP embedding quality.
"""
priority = ("orig", "large", "medium")
buckets: dict[str, list[str]] = {p: [] for p in priority}
for url in file_urls:
lower = url.lower()
if not lower.endswith((".jpg", ".jpeg", ".png", ".tif", ".tiff")):
continue
for p in priority:
if f"~{p}." in lower:
buckets[p].append(url)
break
for p in priority:
if buckets[p]:
return buckets[p][0]
# Fallback: any jpg/png
for url in file_urls:
if url.lower().endswith((".jpg", ".jpeg", ".png")):
return url
return ""
def _sanitize_source_id(raw: str) -> str:
"""Collapse arbitrary NASA nasa_ids into a filesystem-safe token.
Some older items use their title as the nasa_id, which means the
raw string contains spaces, apostrophes, en-dashes, and other
characters that break file paths and command-line tooling. We
replace anything outside ``[A-Za-z0-9._-]`` with an underscore,
collapse runs, and strip edges. Length is capped at 120 chars so
the resulting clip_id stays well under filesystem limits.
"""
if not raw:
return "unknown"
cleaned = _UNSAFE_ID_CHARS.sub("_", raw.strip())
cleaned = re.sub(r"_+", "_", cleaned).strip("_.")
if not cleaned:
return "unknown"
return cleaned[:120]
def _encode_url_path(url: str) -> str:
"""URL-encode the path component of a URL, leaving the host alone.
NASA occasionally returns asset URLs whose path literally contains
spaces and unicode — those will 400 if passed unquoted to
``requests.get``. We use ``urllib.parse`` to surgically re-encode
only the path segment without double-encoding the scheme/host.
"""
try:
parts = urlparse(url)
except Exception:
return url
safe_path = quote(parts.path, safe="/")
return urlunparse(parts._replace(path=safe_path))
tools/video/stock_sources/pexels.py
+289
View File
@@ -0,0 +1,289 @@
"""Pexels stock media source adapter.
Wraps the Pexels video and image search APIs behind the unified
`StockSource` protocol. Pexels is the workhorse for the
documentary-montage pipeline: large catalogue, fast API, free, no
attribution required, and stable URLs for cacheable downloads.
Pexels exposes videos and images on two separate endpoints
(``/videos/search`` and ``/v1/search``), so this adapter fans out
internally and normalises both into the same `Candidate` shape. The
corpus builder never branches on kind.
Uses `PEXELS_API_KEY` from the environment. `.env` is loaded at process
start by `tools.base_tool._load_dotenv`.
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Any, Optional
from .base import Candidate, SearchFilters
_VIDEO_SEARCH_URL = "https://api.pexels.com/videos/search"
_IMAGE_SEARCH_URL = "https://api.pexels.com/v1/search"
_PEXELS_LICENSE = "Pexels License (free, no attribution required)"
class PexelsSource:
"""Unified Pexels adapter for videos and images.
Satisfies `StockSource`. Stateless — all config comes from the
environment, so the corpus builder can instantiate one per run
without caching the instance.
"""
name = "pexels"
def is_available(self) -> bool:
return bool(os.environ.get("PEXELS_API_KEY"))
# ------------------------------------------------------------------
# Public protocol
# ------------------------------------------------------------------
def search(self, query: str, filters: SearchFilters) -> list[Candidate]:
"""Search Pexels, routing to video/image endpoints by `filters.kind`.
`kind="video"` hits only the videos endpoint, `kind="image"`
only the images endpoint, `kind="any"` queries both and
concatenates the results (videos first). The caller decides
ordering semantics downstream — this adapter does not re-rank.
"""
kind = (filters.kind or "video").lower()
out: list[Candidate] = []
if kind in ("video", "any"):
out.extend(self._search_videos(query, filters))
if kind in ("image", "any"):
out.extend(self._search_images(query, filters))
return out
def download(self, candidate: Candidate, out_path: Path) -> Path:
"""Stream the candidate's file to `out_path`.
Creates parent directories as needed. Uses streamed chunks so
large 1080p clips don't blow RAM. No caching — the corpus
builder is responsible for deciding whether to call this at
all.
"""
import requests # lazy — avoid pulling requests at import time
if not candidate.download_url:
raise ValueError(
f"Candidate {candidate.clip_id} has no download_url"
)
out_path = Path(out_path)
out_path.parent.mkdir(parents=True, exist_ok=True)
with requests.get(
candidate.download_url, stream=True, timeout=120
) as r:
r.raise_for_status()
with open(out_path, "wb") as f:
for chunk in r.iter_content(chunk_size=1 << 16):
if chunk:
f.write(chunk)
return out_path
# ------------------------------------------------------------------
# Internals
# ------------------------------------------------------------------
def _headers(self) -> dict[str, str]:
key = os.environ.get("PEXELS_API_KEY")
if not key:
raise RuntimeError(
"PEXELS_API_KEY not set. Get a free key at "
"https://www.pexels.com/api/ and add it to .env."
)
return {"Authorization": key}
def _search_videos(
self, query: str, filters: SearchFilters
) -> list[Candidate]:
import requests # lazy
params: dict[str, Any] = {
"query": query,
"per_page": max(1, min(filters.per_page, 80)),
"page": max(1, filters.page),
}
if filters.orientation:
params["orientation"] = filters.orientation
r = requests.get(
_VIDEO_SEARCH_URL,
headers=self._headers(),
params=params,
timeout=30,
)
r.raise_for_status()
data = r.json()
videos = data.get("videos", []) or []
out: list[Candidate] = []
for v in videos:
# Duration filter. Pexels doesn't expose this server-side,
# so we filter client-side and accept that per_page may be
# partially consumed by clips we throw away.
duration = float(v.get("duration", 0) or 0)
if filters.min_duration is not None and duration < filters.min_duration:
continue
if filters.max_duration is not None and duration > filters.max_duration:
continue
rend = _pick_video_rendition(
v.get("video_files", []) or [],
min_width=filters.min_width or 0,
max_width=1920,
)
if rend is None:
continue
user = v.get("user") or {}
tag_text = _slug_tags_from_url(v.get("url", "") or "")
out.append(
Candidate(
source=self.name,
source_id=str(v.get("id")),
source_url=v.get("url", "") or "",
download_url=rend.get("link", "") or "",
kind="video",
width=int(rend.get("width") or v.get("width") or 0),
height=int(rend.get("height") or v.get("height") or 0),
duration=duration,
creator=user.get("name", "") or "",
license=_PEXELS_LICENSE,
source_tags=tag_text,
thumbnail_url=v.get("image", "") or "",
extra={
"fps": rend.get("fps"),
"rendition_quality": rend.get("quality"),
"user_url": user.get("url"),
},
)
)
return out
def _search_images(
self, query: str, filters: SearchFilters
) -> list[Candidate]:
import requests # lazy
params: dict[str, Any] = {
"query": query,
"per_page": max(1, min(filters.per_page, 80)),
"page": max(1, filters.page),
}
if filters.orientation:
params["orientation"] = filters.orientation
r = requests.get(
_IMAGE_SEARCH_URL,
headers=self._headers(),
params=params,
timeout=30,
)
r.raise_for_status()
data = r.json()
photos = data.get("photos", []) or []
out: list[Candidate] = []
for p in photos:
width = int(p.get("width", 0) or 0)
height = int(p.get("height", 0) or 0)
if filters.min_width is not None and width < filters.min_width:
continue
src = p.get("src") or {}
# "large2x" is the sweet spot between detail and filesize
# for montage use — usually 1.5-2 MP. Fall back to
# "original" if the CDN gave us something weird.
download_url = src.get("large2x") or src.get("original") or ""
if not download_url:
continue
alt = (p.get("alt") or "").strip()
out.append(
Candidate(
source=self.name,
source_id=str(p.get("id")),
source_url=p.get("url", "") or "",
download_url=download_url,
kind="image",
width=width,
height=height,
duration=0.0,
creator=p.get("photographer", "") or "",
license=_PEXELS_LICENSE,
source_tags=alt,
thumbnail_url=src.get("medium", "") or "",
extra={
"photographer_url": p.get("photographer_url"),
"avg_color": p.get("avg_color"),
},
)
)
return out
# ----------------------------------------------------------------------
# Module-level helpers (also used by tests)
# ----------------------------------------------------------------------
def _pick_video_rendition(
video_files: list[dict],
min_width: int = 0,
max_width: int = 1920,
) -> Optional[dict]:
"""Pick the largest mp4/mov rendition within [min_width, max_width].
Pexels returns HLS, SD, HD, and sometimes UHD renditions per video.
We want the biggest file that fits our cap — "biggest" as a quality
proxy, "cap" so we don't waste bandwidth downloading 4K when we'll
scale to 720p anyway.
"""
candidates = [
f for f in video_files
if (f.get("file_type", "") or "").startswith("video/")
and min_width <= int(f.get("width") or 0) <= max_width
and f.get("link")
]
if not candidates:
return None
candidates.sort(key=lambda f: int(f.get("width") or 0), reverse=True)
return candidates[0]
def _slug_tags_from_url(url: str) -> str:
"""Extract a readable tag string from a Pexels video landing URL.
Pexels video JSON does not expose tags or descriptions reliably,
but the landing URL is a slug like::
https://www.pexels.com/video/aerial-view-of-city-at-night-3571264/
The slug is the closest thing to keywords the API gives us, and it
matches what the uploader described the clip as. We strip the
trailing numeric id and rejoin on spaces so the CLIP text encoder
can work with it.
Returns an empty string if the URL cannot be parsed — the caller
should treat that as "no tag signal for this clip".
"""
if not url:
return ""
tail = url.rstrip("/").rsplit("/", 1)
if len(tail) != 2:
return ""
slug = tail[1]
parts = slug.rsplit("-", 1)
if len(parts) == 2 and parts[1].isdigit():
slug = parts[0]
return slug.replace("-", " ").strip()
tools/video/video_compose.py
+110 -9
View File
@@ -247,6 +247,32 @@ class VideoCompose(BaseTool):
"""Check if a file is a still image (routes to Remotion, not FFmpeg)."""
return path.suffix.lower() in VideoCompose._IMAGE_EXTENSIONS
@staticmethod
def _has_audio_stream(path: Path) -> bool:
"""Return True iff ffprobe reports at least one audio stream.
Many stock video clips (especially from Pexels) ship with no audio
stream at all. If we blindly tell ffmpeg to transcode the 0:a stream
on such a file it errors out. This helper lets the segment builder
branch on stream presence so it can synthesize a silent track when
needed, keeping the concat segment layout consistent.
"""
try:
out = subprocess.check_output(
[
"ffprobe", "-v", "error",
"-select_streams", "a",
"-show_entries", "stream=codec_type",
"-of", "default=nw=1:nk=1",
str(path),
],
stderr=subprocess.STDOUT,
text=True,
)
return "audio" in out
except Exception:
return False
def _compose(self, inputs: dict[str, Any]) -> ToolResult:
"""FFmpeg composition: concat video cuts, add audio, burn subtitles.
@@ -326,23 +352,98 @@ class VideoCompose(BaseTool):
)
else:
# Video source: trim to segment.
# -ss before -i for input-level seeking — more reliable
# with -c copy when stream order is non-standard (e.g.
# audio-first containers from Kling/fal.ai).
#
# Semantics:
# -ss BEFORE -i → fast input-level seek to in_s
# -t AFTER -i → "play for `duration` seconds"
# (unambiguous regardless of seek mode)
#
# We MUST re-encode here — `-c copy` cannot do frame-accurate
# cuts because it snaps to keyframes. With sparse GOPs (common
# in Pexels / AI-generated clips), stream-copy can produce
# segments significantly longer than `duration`, breaking the
# target timeline. Re-encoding with libx264/AAC is slower but
# gives exact cut boundaries. Same resolution in → same
# resolution out, so same-res inputs concat cleanly.
cmd = [
"ffmpeg", "-y",
"-ss", str(in_s),
"-t", str(duration),
"-i", str(source),
"-to", str(out_s - in_s),
]
# Normalize every segment to a consistent container so the
# concat-copy step is always safe. The concat demuxer with
# `-c copy` requires identical codec / resolution / fps /
# pix_fmt / sar across ALL segments — otherwise it throws
# "Non-monotonous DTS" or silently produces corrupt output.
#
# Default target is 1920x1080 @ 30fps, yuv420p, sar=1. If the
# source is smaller it letterboxes; if larger it downscales.
# Callers can override via edit_decisions.metadata.compose_target
# (future extension) but the defaults match the most common
# delivery profile (YouTube landscape).
vf_parts: list[str] = [
"scale=1920:1080:force_original_aspect_ratio=decrease",
"pad=1920:1080:(ow-iw)/2:(oh-ih)/2:color=black",
"setsar=1",
"fps=30",
]
af_parts: list[str] = []
if speed != 1.0:
vf = f"setpts={1.0/speed}*PTS"
af = self._build_atempo(speed)
cmd.extend(["-filter:v", vf, "-filter:a", af])
cmd.extend(["-c:v", codec, "-crf", str(crf), "-c:a", "aac"])
vf_parts.append(f"setpts={1.0/speed}*PTS")
af_parts.append(self._build_atempo(speed))
cmd.extend(["-filter:v", ",".join(vf_parts)])
if af_parts:
cmd.extend(["-filter:a", ",".join(af_parts)])
cmd.extend([
"-c:v", codec,
"-crf", str(crf),
"-preset", preset,
"-pix_fmt", "yuv420p",
"-r", "30",
])
# Audio handling: some source clips have no audio stream
# (Pexels stock often ships silent). If we unconditionally
# ask ffmpeg to copy/encode the 0:a stream it errors out.
# Probe for an audio stream first — if present, transcode
# to AAC; if absent, synthesize a silent stereo track so
# concat segments have a consistent stream layout.
has_audio = self._has_audio_stream(source)
if has_audio:
cmd.extend(["-c:a", "aac", "-b:a", "192k", "-ar", "48000", "-ac", "2"])
else:
cmd.extend(["-c", "copy"])
# Inject silent audio via lavfi before the output.
# We have to rebuild cmd to add the lavfi input
# before the output path and map streams explicitly.
cmd = [
"ffmpeg", "-y",
"-ss", str(in_s),
"-t", str(duration),
"-i", str(source),
"-f", "lavfi",
"-t", str(duration),
"-i", "anullsrc=channel_layout=stereo:sample_rate=48000",
"-filter:v", ",".join(vf_parts),
]
if af_parts:
cmd.extend(["-filter:a", ",".join(af_parts)])
cmd.extend([
"-map", "0:v:0",
"-map", "1:a:0",
"-c:v", codec,
"-crf", str(crf),
"-preset", preset,
"-pix_fmt", "yuv420p",
"-r", "30",
"-c:a", "aac",
"-b:a", "192k",
"-ar", "48000",
"-ac", "2",
])
cmd.append(str(seg_path))
self.run_command(cmd)