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

Documentary Montage hardening plus governance fixes

Browse Source
This commit is contained in:
calesthio
2026-04-10 16:42:39 -07:00
parent 44baede67f
commit de94d4dba3
27 changed files with 1107 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/video/corpus_builder.py
+56 -9
View File
@@ -94,7 +94,8 @@ class CorpusBuilder(BaseTool):
"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"
" UNSPLASH_ACCESS_KEY for Unsplash (see https://unsplash.com/documentation)\n"
" archive.org, nasa, and wikimedia work without API keys"
)
agent_skills = []
@@ -201,10 +202,32 @@ class CorpusBuilder(BaseTool):
def get_status(self) -> ToolStatus:
try:
from tools.video.stock_sources import available_sources
from tools.video.stock_sources import all_sources, available_sources
except Exception:
return ToolStatus.UNAVAILABLE
return ToolStatus.AVAILABLE if available_sources() else ToolStatus.UNAVAILABLE
total = len(all_sources())
available = len(available_sources())
if available == 0:
return ToolStatus.UNAVAILABLE
if available < total:
return ToolStatus.DEGRADED
return ToolStatus.AVAILABLE
def get_info(self) -> dict[str, Any]:
info = super().get_info()
try:
from tools.video.stock_sources import source_catalog, source_summary
info["source_provider_menu"] = source_catalog()
info["source_provider_summary"] = source_summary()
except Exception:
info["source_provider_menu"] = []
info["source_provider_summary"] = {
"configured": 0,
"total": 0,
"available_source_names": [],
"unavailable_source_names": [],
}
return info
def estimate_cost(self, inputs: dict[str, Any]) -> float:
return 0.0 # all sources are free-tier
@@ -219,8 +242,10 @@ class CorpusBuilder(BaseTool):
from lib.corpus import Corpus
from tools.video.stock_sources import (
SearchFilters,
all_sources,
available_sources,
get_source,
source_summary,
)
corpus_dir = Path(inputs["corpus_dir"])
@@ -232,17 +257,36 @@ class CorpusBuilder(BaseTool):
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.
# must not silently degrade: pinned-but-unavailable sources
# are a provider substitution the agent needs to surface.
if source_names:
requested: list = []
unavailable_requested: list[str] = []
known_sources = {src.name: src for src in all_sources()}
for name in source_names:
try:
s = get_source(name)
except KeyError as e:
return ToolResult(success=False, error=str(e))
s = known_sources.get(name)
if s is None:
try:
s = get_source(name)
except KeyError as e:
return ToolResult(success=False, error=str(e))
if s.is_available():
requested.append(s)
else:
unavailable_requested.append(name)
if unavailable_requested:
summary = source_summary()
return ToolResult(
success=False,
error=(
"Requested stock sources are unavailable: "
f"{', '.join(unavailable_requested)}. "
"Available now: "
f"{', '.join(summary['available_source_names']) or 'none'}. "
"Check corpus_builder.source_provider_menu during preflight "
"before rerunning."
),
)
sources = requested
else:
sources = available_sources()
@@ -344,6 +388,9 @@ class CorpusBuilder(BaseTool):
"per_source_counts": per_source_counts,
"added_ids": added_ids,
"total_corpus_size": len(corp),
"requested_sources": source_names or [],
"resolved_sources": [s.name for s in sources],
"source_provider_summary": source_summary(),
"errors": errors[:25], # cap log noise
},
cost_usd=0.0,
tools/video/stock_sources/__init__.py
+83 -24
View File
@@ -5,44 +5,69 @@ 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.
Discovery
---------
Adapters are auto-discovered from this package. `all_sources()` returns
one instance of every concrete adapter class found under
`tools.video.stock_sources`, ordered by each class's optional
`priority` and then by name. `available_sources()` filters to the ones
whose `is_available()` returns True right now.
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.
This keeps source discoverability aligned with the rest of the repo:
adding a new adapter file is enough to make it visible to
`corpus_builder` and to preflight metadata.
"""
from __future__ import annotations
from .archive_org import ArchiveOrgSource
import importlib
import inspect
import pkgutil
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",
"source_catalog",
"source_summary",
]
# 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 _is_source_adapter_class(cls: type) -> bool:
"""Return True for concrete source adapters in this package."""
return (
inspect.isclass(cls)
and cls.__module__.startswith(f"{__name__}.")
and cls.__module__ != f"{__name__}.base"
and isinstance(getattr(cls, "name", None), str)
and bool(getattr(cls, "name", None))
and callable(getattr(cls, "is_available", None))
and callable(getattr(cls, "search", None))
and callable(getattr(cls, "download", None))
)
def _source_classes() -> list[type]:
"""Auto-discover stock source classes under this package."""
discovered: dict[str, type] = {}
for module_info in pkgutil.iter_modules(__path__, f"{__name__}."):
if module_info.ispkg or module_info.name.endswith(".base"):
continue
module = importlib.import_module(module_info.name)
for _, cls in inspect.getmembers(module, inspect.isclass):
if not _is_source_adapter_class(cls):
continue
discovered[getattr(cls, "name")] = cls
return sorted(
discovered.values(),
key=lambda cls: (
int(getattr(cls, "priority", 100)),
getattr(cls, "display_name", getattr(cls, "name")).lower(),
),
)
def all_sources() -> list[StockSource]:
@@ -53,7 +78,7 @@ def all_sources() -> list[StockSource]:
want to show the user what sources exist regardless of whether
their credentials are configured.
"""
return [cls() for cls in _REGISTRY]
return [cls() for cls in _source_classes()]
def available_sources() -> list[StockSource]:
@@ -67,6 +92,40 @@ def available_sources() -> list[StockSource]:
return [s for s in all_sources() if s.is_available()]
def source_catalog() -> list[dict[str, object]]:
"""Return discoverability metadata for every stock source."""
catalog: list[dict[str, object]] = []
for source in all_sources():
cls = source.__class__
available = bool(source.is_available())
catalog.append({
"name": source.name,
"display_name": getattr(cls, "display_name", source.name),
"provider": getattr(cls, "provider", source.name),
"status": "available" if available else "unavailable",
"install_instructions": getattr(
cls,
"install_instructions",
"See the source adapter docs for setup details.",
),
"supports": getattr(cls, "supports", {}),
})
return catalog
def source_summary() -> dict[str, object]:
"""Summarize source availability for preflight and tool contracts."""
catalog = source_catalog()
available = [entry["name"] for entry in catalog if entry["status"] == "available"]
unavailable = [entry["name"] for entry in catalog if entry["status"] != "available"]
return {
"configured": len(available),
"total": len(catalog),
"available_source_names": available,
"unavailable_source_names": unavailable,
}
def get_source(name: str) -> StockSource:
"""Look up a single adapter by its `name` attribute.
tools/video/stock_sources/archive_org.py
+7
View File
@@ -70,6 +70,13 @@ class ArchiveOrgSource:
"""
name = "archive_org"
display_name = "Archive.org"
provider = "archive_org"
priority = 20
install_instructions = (
"No setup required. Archive.org is available without API keys."
)
supports = {"video": True, "image": False}
def is_available(self) -> bool:
# No API key, no config. As long as the network is up, we're
tools/video/stock_sources/base.py
+3 -2
View File
@@ -29,8 +29,9 @@ Adding a new source
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.
3. Give the class a stable `name` attribute and optional discoverability
metadata such as `display_name`, `install_instructions`, `supports`,
and `priority`. The package auto-discovers concrete adapters.
"""
from __future__ import annotations
tools/video/stock_sources/nasa.py
+8
View File
@@ -55,6 +55,14 @@ class NasaSource:
"""
name = "nasa"
display_name = "NASA"
provider = "nasa"
priority = 30
install_instructions = (
"No setup required. NASA media search works without an API key; "
"NASA_API_KEY is optional for higher rate limits."
)
supports = {"video": True, "image": True}
def is_available(self) -> bool:
# The public API is unauthenticated; as long as the network is
tools/video/stock_sources/pexels.py
+8
View File
@@ -36,6 +36,14 @@ class PexelsSource:
"""
name = "pexels"
display_name = "Pexels"
provider = "pexels"
priority = 10
install_instructions = (
"Set PEXELS_API_KEY in .env to enable Pexels stock search "
"(free key at https://www.pexels.com/api/)."
)
supports = {"video": True, "image": True}
def is_available(self) -> bool:
return bool(os.environ.get("PEXELS_API_KEY"))
tools/video/stock_sources/unsplash.py
+184
View File
@@ -0,0 +1,184 @@
"""Unsplash stock photo adapter.
Unsplash is image-only in this pipeline. It widens the corpus for
modern, polished, lifestyle, and product-adjacent scenes where a
high-quality still can still be valuable to the edit.
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Any
from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
from .base import Candidate, SearchFilters
_SEARCH_URL = "https://api.unsplash.com/search/photos"
_UNSPLASH_LICENSE = "Unsplash License (use returned hotlinked image URLs)"
_USER_AGENT = "OpenMontageBot/0.1 (https://github.com/calesthio/OpenMontage)"
class UnsplashSource:
"""Adapter for Unsplash photo search."""
name = "unsplash"
display_name = "Unsplash"
provider = "unsplash"
priority = 18
install_instructions = (
"Set UNSPLASH_ACCESS_KEY in .env to enable Unsplash image search "
"(see https://unsplash.com/documentation)."
)
supports = {"video": False, "image": True}
def is_available(self) -> bool:
return bool(os.environ.get("UNSPLASH_ACCESS_KEY"))
def search(self, query: str, filters: SearchFilters) -> list[Candidate]:
import requests # lazy
kind = (filters.kind or "video").lower()
if kind == "video":
return []
params: dict[str, Any] = {
"query": query,
"page": max(1, filters.page),
"per_page": max(1, min(filters.per_page, 30)),
"content_filter": "high",
}
orientation = _orientation_for_unsplash(filters.orientation)
if orientation:
params["orientation"] = orientation
r = requests.get(
_SEARCH_URL,
params=params,
headers=self._headers(),
timeout=30,
)
r.raise_for_status()
data = r.json()
results = data.get("results") or []
out: list[Candidate] = []
for photo in results:
cand = _photo_to_candidate(photo, filters)
if cand is not None:
out.append(cand)
return out
def download(self, candidate: Candidate, out_path: Path) -> Path:
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=180,
headers={"User-Agent": _USER_AGENT},
) 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
def _headers(self) -> dict[str, str]:
key = os.environ.get("UNSPLASH_ACCESS_KEY")
if not key:
raise RuntimeError(
"UNSPLASH_ACCESS_KEY not set. Create an app at "
"https://unsplash.com/documentation and add the access key to .env."
)
return {
"Authorization": f"Client-ID {key}",
"Accept-Version": "v1",
"User-Agent": _USER_AGENT,
}
def _photo_to_candidate(photo: dict[str, Any], filters: SearchFilters) -> Candidate | None:
width = int(photo.get("width") or 0)
height = int(photo.get("height") or 0)
if filters.min_width is not None and width and width < filters.min_width:
return None
if filters.orientation and not _matches_orientation(filters.orientation, width, height):
return None
user = photo.get("user") or {}
links = photo.get("links") or {}
urls = photo.get("urls") or {}
raw_url = urls.get("raw") or urls.get("regular") or ""
if not raw_url:
return None
description_parts = [
photo.get("description") or "",
photo.get("alt_description") or "",
photo.get("slug") or "",
]
source_tags = " ".join(part.strip() for part in description_parts if part).strip()
if len(source_tags) > 500:
source_tags = source_tags[:500]
return Candidate(
source=UnsplashSource.name,
source_id=str(photo.get("id") or ""),
source_url=links.get("html", "") or "",
download_url=_build_download_url(raw_url, target_width=max(filters.min_width or 0, 1920)),
kind="image",
width=width,
height=height,
duration=0.0,
creator=user.get("name", "") or "",
license=_UNSPLASH_LICENSE,
source_tags=source_tags,
thumbnail_url=urls.get("small", "") or urls.get("thumb", "") or raw_url,
extra={
"color": photo.get("color"),
"blur_hash": photo.get("blur_hash"),
"download_location": links.get("download_location"),
"photographer_url": user.get("links", {}).get("html"),
},
)
def _orientation_for_unsplash(orientation: str | None) -> str | None:
if orientation == "landscape":
return "landscape"
if orientation == "portrait":
return "portrait"
if orientation == "square":
return "squarish"
return None
def _matches_orientation(orientation: str, width: int, height: int) -> bool:
if not width or not height:
return True
if orientation == "landscape":
return width >= height
if orientation == "portrait":
return height > width
if orientation == "square":
return width == height
return True
def _build_download_url(raw_url: str, target_width: int) -> str:
parts = urlparse(raw_url)
params = dict(parse_qsl(parts.query, keep_blank_values=True))
params.setdefault("fm", "jpg")
params.setdefault("q", "80")
if target_width > 0:
params["w"] = str(target_width)
params.setdefault("fit", "max")
return urlunparse(parts._replace(query=urlencode(params)))
tools/video/stock_sources/wikimedia.py
+196
View File
@@ -0,0 +1,196 @@
"""Wikimedia Commons stock media adapter.
Provides image and video search over Wikimedia Commons using the
MediaWiki API. Commons is a uniquely useful documentary source because
it mixes public-domain historical imagery, recent CC-licensed videos,
and educational media under one searchable catalogue.
"""
from __future__ import annotations
import html
import re
from pathlib import Path
from typing import Any
from .base import Candidate, SearchFilters
_API_URL = "https://commons.wikimedia.org/w/api.php"
_USER_AGENT = "OpenMontageBot/0.1 (https://github.com/calesthio/OpenMontage)"
_COMMONS_LICENSE = "Wikimedia Commons (verify per-file license)"
_HTML_TAG_RE = re.compile(r"<[^>]+>")
class WikimediaSource:
"""Adapter for Wikimedia Commons media search."""
name = "wikimedia"
display_name = "Wikimedia Commons"
provider = "wikimedia"
priority = 25
install_instructions = (
"No setup required. Wikimedia Commons media search works without API keys."
)
supports = {"video": True, "image": True}
def is_available(self) -> bool:
return True
def search(self, query: str, filters: SearchFilters) -> list[Candidate]:
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",
}
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)))
out: list[Candidate] = []
for page in pages:
cand = _page_to_candidate(page, filters)
if cand is not None:
out.append(cand)
return out
def download(self, candidate: Candidate, out_path: Path) -> Path:
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,
headers={"User-Agent": _USER_AGENT},
) 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
def _build_search_query(query: str, kind: str) -> str:
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
def _page_to_candidate(page: dict[str, Any], filters: SearchFilters) -> Candidate | None:
infos = page.get("imageinfo") or []
if not infos:
return None
info = infos[0]
mime = (info.get("mime") or "").lower()
kind = _kind_from_mime(mime, page.get("title", ""))
requested_kind = (filters.kind or "video").lower()
if requested_kind == "video" and kind != "video":
return None
if requested_kind == "image" and kind != "image":
return None
width = int(info.get("width") or 0)
height = int(info.get("height") or 0)
duration = float(info.get("duration") or 0.0)
if filters.min_width is not None and width and width < filters.min_width:
return None
if filters.min_duration is not None and duration and duration < filters.min_duration:
return None
if filters.max_duration is not None and duration and duration > filters.max_duration:
return None
if filters.orientation and not _matches_orientation(filters.orientation, width, height):
return None
meta = info.get("extmetadata") or {}
object_name = _meta_value(meta, "ObjectName")
description = _meta_value(meta, "ImageDescription")
categories = _meta_value(meta, "Categories")
creator = _meta_value(meta, "Artist")
license_name = _meta_value(meta, "LicenseShortName")
usage_terms = _meta_value(meta, "UsageTerms")
source_tags = " ".join(part for part in (object_name, description, categories) if part).strip()
if len(source_tags) > 500:
source_tags = source_tags[:500]
title = page.get("title", "")
page_id = str(page.get("pageid") or title.replace("File:", "", 1))
source_url = info.get("descriptionurl") or page.get("canonicalurl") or ""
return Candidate(
source=WikimediaSource.name,
source_id=page_id,
source_url=source_url,
download_url=info.get("url", "") or "",
kind=kind,
width=width,
height=height,
duration=duration,
creator=creator,
license=license_name or usage_terms or _COMMONS_LICENSE,
source_tags=source_tags,
thumbnail_url=info.get("thumburl", "") or info.get("url", "") or "",
extra={
"mime": mime,
"title": title,
"mediatype": info.get("mediatype"),
"descriptionshorturl": info.get("descriptionshorturl"),
},
)
def _kind_from_mime(mime: str, title: str) -> str:
if mime.startswith("video/") or title.lower().endswith((".webm", ".ogv", ".ogg")):
return "video"
return "image"
def _matches_orientation(orientation: str, width: int, height: int) -> bool:
if not width or not height:
return True
if orientation == "landscape":
return width >= height
if orientation == "portrait":
return height > width
if orientation == "square":
return width == height
return True
def _meta_value(meta: dict[str, Any], key: str) -> str:
raw = ((meta.get(key) or {}).get("value")) or ""
if not raw:
return ""
text = html.unescape(str(raw))
text = _HTML_TAG_RE.sub(" ", text)
text = re.sub(r"\s+", " ", text).strip()
return text
tools/video/video_compose.py
+6 -6
View File
@@ -554,6 +554,7 @@ class VideoCompose(BaseTool):
"explainer-data": "Explainer",
"explainer-teacher": "Explainer",
"cinematic-trailer": "CinematicRenderer",
"documentary-montage": "CinematicRenderer",
"product-reveal": "Explainer",
"screen-demo": "Explainer",
"presenter": "TalkingHead",
@@ -691,12 +692,11 @@ class VideoCompose(BaseTool):
component types, transitions, and mixed content — all in a single
React-based render pass.
Returns False (i.e. use FFmpeg) ONLY when ALL of these are true:
1. Every cut source is a video file (no images, no component types)
2. No cut requests animations or transitions
3. No cut uses a Remotion scene type
4. The edit_decisions explicitly set renderer_family to "ffmpeg-only"
OR Remotion is not available
Returns False (i.e. use FFmpeg) only when Remotion is not
available. For `operation="render"` the governance default is
Remotion-first: the renderer family was chosen earlier, and the
tool should preserve that decision instead of silently
downgrading to FFmpeg.
This "Remotion-first" policy means mixed content (video clips +
animated stills + text cards) is always composed in Remotion, which