diff --git a/remotion-composer/src/Root.tsx b/remotion-composer/src/Root.tsx index 8013bec..2ae5141 100644 --- a/remotion-composer/src/Root.tsx +++ b/remotion-composer/src/Root.tsx @@ -12,6 +12,7 @@ import { } from "./TitledVideo"; import { EndTag, EndTagProps } from "./components/EndTag"; import { HeroTitle } from "./components/HeroTitle"; +import { ProductReveal, ProductRevealProps } from "./components/ProductReveal"; // --------------------------------------------------------------------------- // Theme System — prevents every video from looking like dark fintech @@ -217,6 +218,38 @@ export const Root: React.FC = () => { subtitle: "The People Who Define Reality", }} /> + + = ({ + productImage, + productName, + price, + tagline, + closer, + accentColor = "#00D4FF", +}) => { + const frame = useCurrentFrame(); + const { fps } = useVideoConfig(); + + // === PHASE 1: Product image scales in with glow (0-1.5s) === + const imgScale = spring({ + frame, + fps, + config: { damping: 14, stiffness: 80, mass: 0.8 }, + }); + + const imgOpacity = interpolate(frame, [0, 8], [0, 1], { + extrapolateRight: "clamp", + }); + + // Glow pulse + const glowIntensity = interpolate( + Math.sin(frame * 0.08), + [-1, 1], + [0.3, 0.7] + ); + + // Slow float + const floatY = Math.sin(frame * 0.04) * 4; + + // === PHASE 2: Product name springs in letter by letter (1s delay) === + const nameDelay = fps * 1.2; + const nameChars = productName.split(""); + + // === PHASE 3: Price reveals (3s delay) === + const priceDelay = fps * 3.2; + const priceSpring = spring({ + frame: frame - priceDelay, + fps, + config: { damping: 16, stiffness: 120 }, + }); + + // === PHASE 4: Tagline fades in (4.2s delay) === + const taglineDelay = fps * 4.2; + const taglineOpacity = interpolate( + frame, + [taglineDelay, taglineDelay + fps * 0.6], + [0, 1], + { extrapolateLeft: "clamp", extrapolateRight: "clamp" } + ); + + // === PHASE 5: Closer fades in (5.5s delay) === + const closerDelay = fps * 5.5; + const closerOpacity = interpolate( + frame, + [closerDelay, closerDelay + fps * 0.8], + [0, 1], + { extrapolateLeft: "clamp", extrapolateRight: "clamp" } + ); + + // === PHASE 6: Shimmer across price (4.5s) === + const shimmerDelay = fps * 4.5; + const shimmerPos = interpolate( + frame, + [shimmerDelay, shimmerDelay + fps * 1.0], + [-100, 400], + { + extrapolateLeft: "clamp", + extrapolateRight: "clamp", + easing: Easing.inOut(Easing.ease), + } + ); + + // === FADE OUT at end (last 0.8s) === + const totalDuration = fps * 8; + const fadeOut = interpolate( + frame, + [totalDuration - fps * 0.8, totalDuration], + [1, 0], + { extrapolateLeft: "clamp", extrapolateRight: "clamp" } + ); + + return ( + + {/* Ambient glow behind product */} +
+ + {/* Product image — centered in upper portion */} +
+
+ +
+ {/* Reflection */} +
+ +
+
+ + {/* Text content — stacked vertically, centered */} +
+ {/* Product name — letter by letter spring */} +
+ {nameChars.map((char, i) => { + const charDelay = nameDelay + i * 1.0; + const charSpring = spring({ + frame: frame - charDelay, + fps, + config: { damping: 14, stiffness: 160 }, + }); + + // "Air" = first 3 chars get accent color + const isAccent = i < 3; + + return ( + + {char} + + ); + })} +
+ + {/* Price — block element, clearly on its own line */} +
+
+ {price} +
+ {/* Shimmer across price text */} + {frame >= shimmerDelay && frame <= shimmerDelay + fps * 1.0 && ( +
+ )} +
+ + {/* Tagline */} +
+ {tagline} +
+ + {/* Closer */} +
+ {closer} +
+
+ + ); +}; diff --git a/skills/pipelines/documentary-montage/asset-director.md b/skills/pipelines/documentary-montage/asset-director.md index d543892..e2f8f0b 100644 --- a/skills/pipelines/documentary-montage/asset-director.md +++ b/skills/pipelines/documentary-montage/asset-director.md @@ -8,7 +8,9 @@ clips that fill each slot. There are two paths: ### Standard Path: Corpus + CLIP Retrieval 1. **Build the corpus** — fan the scene director's queries out across - Pexels / Archive.org / NASA / Wikimedia / Unsplash and download/embed the candidates. + all available stock sources (Pexels, Pixabay Video, Coverr, Mixkit, + Archive.org, NARA, Library of Congress, Pond5 PD, Videvo, NASA, ESA, + JAXA, NOAA, Dareful, Wikimedia, Unsplash) and download/embed the candidates. 2. **Pick per slot** — run CLIP retrieval against the corpus with each slot description and choose one winner per slot. @@ -91,7 +93,7 @@ direct_clip_search.execute({ {"query": "satellite dish night sky", "slot_id": "slot_03"}, # ... one per slot ], - "sources": ["pexels", "archive_org", "wikimedia"], # or omit for all available + "sources": ["pexels", "pixabay_video", "coverr", "mixkit", "archive_org"], # or omit for all available "clips_per_query": 3, "filters": { "min_duration": 3, diff --git a/skills/pipelines/documentary-montage/idea-director.md b/skills/pipelines/documentary-montage/idea-director.md index e6ef01a..ca7af90 100644 --- a/skills/pipelines/documentary-montage/idea-director.md +++ b/skills/pipelines/documentary-montage/idea-director.md @@ -156,7 +156,7 @@ Minimum fields the brief must carry: "tone": "elegiac", "duration_seconds": 90, "shape": "list", - "sources_allowed": ["pexels", "archive_org", "nasa"], + "sources_allowed": ["pexels", "pixabay_video", "coverr", "mixkit", "archive_org", "nara", "nasa"], "generated_clips_allowed": false, "narration": "none", "music_plan": { diff --git a/skills/pipelines/documentary-montage/scene-director.md b/skills/pipelines/documentary-montage/scene-director.md index 16520a9..8ff0bbc 100644 --- a/skills/pipelines/documentary-montage/scene-director.md +++ b/skills/pipelines/documentary-montage/scene-director.md @@ -145,8 +145,19 @@ based on what footage lives where: | Source | Strengths | Use when | |--------|-----------|----------| | `pexels` | Modern HD footage, clean shots, people, cities, nature | Default for modern/any era | +| `pixabay_video` | Large community library, nature, people, technology, lifestyle | Gap-fills when Pexels misses; broad general footage | +| `coverr` | Curated cinematic B-roll, nature, urban, abstract backgrounds | High-quality establishing shots, mood-setters, modern lifestyle | +| `mixkit` | Curated HD/4K by Envato, nature, business, technology | Premium-feel B-roll, clean nature footage, no attribution needed | | `archive_org` | Prelinger home movies, mid-century educational film, 1940s-1980s texture | Vintage, wry, dreamlike, anything nostalgic | +| `nara` | U.S. National Archives — WWII, Cold War, Apollo, civil rights, presidential | Historical American documentary, military, government, space race | +| `loc` | Library of Congress — early cinema, newsreels, cultural recordings | Pre-1928 public domain footage, American history, folk traditions | +| `pond5_pd` | Pond5 Public Domain — WWI/WWII, early cinema, historical speeches | Archival/vintage footage, Méliès, Edison, newsreels | +| `videvo` | 90K+ free clips, nature, aerial, city, abstract, time-lapses | Large free library, complements Pexels with different contributors | | `nasa` | Earth-from-orbit, astronomy, flight, scale imagery | Reverent, anything about scale, space, planet, flight | +| `esa` | European space missions, Hubble/Webb imagery, Earth observation | European space content, complements NASA for non-U.S. missions | +| `jaxa` | Japanese space missions, Hayabusa, ISS Kibo module, H-IIA rockets | Asian space content, unique angle on space exploration | +| `noaa` | Deep-sea ROV footage, marine life, coral reefs, weather, hurricanes | Ocean/underwater, unique deep-sea content, weather phenomena | +| `dareful` | Boutique 4K nature — mountains, forests, waterfalls, time-lapses | High-quality nature B-roll, consistent visual style, aerial shots | | `wikimedia` | Commons photos and CC video, civic/documentary/public-event coverage | Public spaces, landmarks, protests, city texture, educational footage | | `unsplash` | Polished editorial stills, lifestyle, product-adjacent photography | Modern still-image support shots when motion footage is thin | diff --git a/tools/video/grok_video.py b/tools/video/grok_video.py index 1c1e653..202af14 100644 --- a/tools/video/grok_video.py +++ b/tools/video/grok_video.py @@ -1,4 +1,8 @@ -"""xAI Grok video generation.""" +"""xAI Grok Imagine video generation with native synchronized audio. + +Generates 1-15 second videos with synchronized sound (dialogue with lip-sync, +SFX, ambient, background music) in a single pass. No post-production audio needed. +""" from __future__ import annotations @@ -67,13 +71,17 @@ class GrokVideo(BaseTool): "reference_to_video": True, "reference_image": True, "multiple_reference_images": True, + "native_audio": True, + "lip_sync": True, + "cinematic_quality": True, } best_for = [ - "reference-conditioned video generation", - "product placement or character-consistent motion clips", - "xAI-native image-guided and text-guided short videos", + "cinematic clips with native synchronized audio (dialogue, SFX, music)", + "reference-conditioned video with product/character consistency", + "lip-synced dialogue and foley in a single generation pass", + "cost-effective high-quality video ($0.07/s at 720p)", ] - not_good_for = ["offline generation", "very long clips"] + not_good_for = ["offline generation"] fallback_tools = ["veo_video", "runway_video", "kling_video", "minimax_video"] input_schema = { @@ -93,13 +101,13 @@ class GrokVideo(BaseTool): }, "duration": { "type": "integer", - "minimum": 2, - "maximum": 10, + "minimum": 1, + "maximum": 15, "default": 5, }, "aspect_ratio": { "type": "string", - "enum": ["16:9", "9:16", "1:1"], + "enum": ["16:9", "9:16", "1:1", "4:3", "3:4", "3:2", "2:3"], "default": "16:9", }, "resolution": { diff --git a/tools/video/seedance_video.py b/tools/video/seedance_video.py new file mode 100644 index 0000000..c85bcc1 --- /dev/null +++ b/tools/video/seedance_video.py @@ -0,0 +1,260 @@ +"""Seedance 2.0 (ByteDance) video generation via fal.ai API. + +Best for cinematic clips with native audio, director-level camera control, +and lip-sync from quoted dialogue in prompts. +""" + +from __future__ import annotations + +import os +import time +from pathlib import Path +from typing import Any + +from tools.base_tool import ( + BaseTool, + Determinism, + ExecutionMode, + ResourceProfile, + RetryPolicy, + ToolResult, + ToolRuntime, + ToolStability, + ToolStatus, + ToolTier, +) + + +class SeedanceVideo(BaseTool): + name = "seedance_video" + version = "0.1.0" + tier = ToolTier.GENERATE + capability = "video_generation" + provider = "seedance" + stability = ToolStability.EXPERIMENTAL + execution_mode = ExecutionMode.SYNC + determinism = Determinism.STOCHASTIC + runtime = ToolRuntime.API + + dependencies = [] + install_instructions = ( + "Set FAL_KEY to your fal.ai API key.\n" + " Get one at https://fal.ai/dashboard/keys" + ) + agent_skills = ["ai-video-gen"] + + capabilities = ["text_to_video", "image_to_video"] + supports = { + "text_to_video": True, + "image_to_video": True, + "native_audio": True, + "cinematic_quality": True, + "camera_direction": True, + "lip_sync": True, + } + best_for = [ + "cinematic clips with native synchronized audio", + "director-level camera control and multi-shot editing", + "lip-sync from quoted dialogue in prompts", + "high-fidelity motion with real-world physics", + ] + not_good_for = ["offline generation", "budget-constrained projects"] + fallback_tools = ["kling_video", "minimax_video", "veo_video"] + + input_schema = { + "type": "object", + "required": ["prompt"], + "properties": { + "prompt": {"type": "string"}, + "operation": { + "type": "string", + "enum": ["text_to_video", "image_to_video"], + "default": "text_to_video", + }, + "model_variant": { + "type": "string", + "enum": ["standard", "fast"], + "default": "standard", + "description": "standard = highest quality, fast = lower latency and cost", + }, + "duration": { + "type": "string", + "enum": ["auto", "4", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15"], + "default": "5", + "description": "Duration in seconds. 'auto' lets the model decide.", + }, + "aspect_ratio": { + "type": "string", + "enum": ["auto", "21:9", "16:9", "4:3", "1:1", "3:4", "9:16"], + "default": "16:9", + }, + "resolution": { + "type": "string", + "enum": ["480p", "720p"], + "default": "720p", + }, + "generate_audio": { + "type": "boolean", + "default": True, + "description": "Generate synchronized audio (speech, SFX, ambient)", + }, + "image_url": { + "type": "string", + "description": "Start frame image URL for image_to_video (jpg, png, webp)", + }, + "end_image_url": { + "type": "string", + "description": "Optional end frame URL for image_to_video", + }, + "seed": { + "type": "integer", + "description": "Optional seed for reproducibility", + }, + "output_path": {"type": "string"}, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=500, network_required=True + ) + retry_policy = RetryPolicy(max_retries=2, retryable_errors=["rate_limit", "timeout"]) + idempotency_key_fields = ["prompt", "model_variant", "operation", "duration", "seed"] + side_effects = ["writes video file to output_path", "calls fal.ai API"] + user_visible_verification = [ + "Watch generated clip for motion coherence, audio sync, and visual quality" + ] + + def _get_api_key(self) -> str | None: + return os.environ.get("FAL_KEY") or os.environ.get("FAL_AI_API_KEY") + + def get_status(self) -> ToolStatus: + if self._get_api_key(): + return ToolStatus.AVAILABLE + return ToolStatus.UNAVAILABLE + + def estimate_cost(self, inputs: dict[str, Any]) -> float: + variant = inputs.get("model_variant", "standard") + duration = inputs.get("duration", "5") + secs = 5 if duration == "auto" else int(duration) + rate = 0.2419 if variant == "fast" else 0.3034 + return round(rate * secs, 2) + + def estimate_runtime(self, inputs: dict[str, Any]) -> float: + variant = inputs.get("model_variant", "standard") + return 60.0 if variant == "fast" else 120.0 + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + api_key = self._get_api_key() + if not api_key: + return ToolResult( + success=False, + error="FAL_KEY not set. " + self.install_instructions, + ) + + import requests + + start = time.time() + operation = inputs.get("operation", "text_to_video") + variant = inputs.get("model_variant", "standard") + operation_path = operation.replace("_", "-") + + if variant == "fast": + model_path = f"bytedance/seedance-2.0/fast/{operation_path}" + else: + model_path = f"bytedance/seedance-2.0/{operation_path}" + + payload: dict[str, Any] = {"prompt": inputs["prompt"]} + + if inputs.get("duration"): + payload["duration"] = inputs["duration"] + if inputs.get("aspect_ratio"): + payload["aspect_ratio"] = inputs["aspect_ratio"] + if inputs.get("resolution"): + payload["resolution"] = inputs["resolution"] + if "generate_audio" in inputs: + payload["generate_audio"] = inputs["generate_audio"] + if inputs.get("seed") is not None: + payload["seed"] = inputs["seed"] + + if operation == "image_to_video": + if inputs.get("image_url"): + payload["image_url"] = inputs["image_url"] + elif inputs.get("image_path"): + from tools.video._shared import upload_image_to_fal + payload["image_url"] = upload_image_to_fal(inputs["image_path"]) + if inputs.get("end_image_url"): + payload["end_image_url"] = inputs["end_image_url"] + + headers = { + "Authorization": f"Key {api_key}", + "Content-Type": "application/json", + } + + try: + submit_resp = requests.post( + f"https://queue.fal.run/fal-ai/{model_path}", + headers=headers, + json=payload, + timeout=30, + ) + submit_resp.raise_for_status() + queue_data = submit_resp.json() + status_url = queue_data["status_url"] + response_url = queue_data["response_url"] + + while True: + time.sleep(5) + status_resp = requests.get(status_url, headers=headers, timeout=15) + status_resp.raise_for_status() + status = status_resp.json().get("status", "UNKNOWN") + if status == "COMPLETED": + break + if status in ("FAILED", "CANCELLED"): + return ToolResult( + success=False, + error=f"Seedance 2.0 video generation {status.lower()}", + ) + + result_resp = requests.get(response_url, headers=headers, timeout=30) + result_resp.raise_for_status() + data = result_resp.json() + + video_url = data["video"]["url"] + video_response = requests.get(video_url, timeout=120) + video_response.raise_for_status() + + output_path = Path(inputs.get("output_path", "seedance_output.mp4")) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(video_response.content) + + except Exception as e: + return ToolResult( + success=False, + error=f"Seedance 2.0 video generation failed: {e}", + ) + + from tools.video._shared import probe_output + + probed = probe_output(output_path) + return ToolResult( + success=True, + data={ + "provider": "seedance", + "model": f"fal-ai/{model_path}", + "prompt": inputs["prompt"], + "operation": operation, + "variant": variant, + "aspect_ratio": inputs.get("aspect_ratio", "16:9"), + "resolution": inputs.get("resolution", "720p"), + "generate_audio": inputs.get("generate_audio", True), + "seed": data.get("seed"), + "output": str(output_path), + "output_path": str(output_path), + "format": "mp4", + **probed, + }, + artifacts=[str(output_path)], + cost_usd=self.estimate_cost(inputs), + duration_seconds=round(time.time() - start, 2), + model=f"fal-ai/{model_path}", + ) diff --git a/tools/video/stock_sources/coverr.py b/tools/video/stock_sources/coverr.py new file mode 100644 index 0000000..4451269 --- /dev/null +++ b/tools/video/stock_sources/coverr.py @@ -0,0 +1,143 @@ +"""Coverr stock video source adapter. + +Wraps the Coverr API (``api.coverr.co``) behind the unified `StockSource` +protocol. Coverr offers curated, high-quality stock footage (HD and 4K) +under a free commercial-use licence with no attribution required. + +Free API tier: 50 requests per hour. Production tier (with Pro/Ultimate +subscription): 2,000 requests per hour. The adapter uses the free tier +by default — no API key required for basic search. + +What Coverr is good for +----------------------- +- Modern lifestyle / cinematic B-roll +- Nature, urban, technology, abstract backgrounds +- High production quality (curated library) +- Quick establishing shots and mood-setters +""" +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any + +from .base import Candidate, SearchFilters + + +_SEARCH_URL = "https://api.coverr.co/videos" +_LICENSE = "Coverr License (free for commercial and personal use, no attribution required)" + + +class CoverrSource: + """Coverr video adapter. Satisfies `StockSource`.""" + + name = "coverr" + display_name = "Coverr" + provider = "coverr" + priority = 16 + install_instructions = ( + "Coverr works without an API key (free tier, 50 req/hr). " + "Set COVERR_API_KEY in .env for higher rate limits (Pro tier)." + ) + supports = {"video": True, "image": False} + + def is_available(self) -> bool: + # Coverr works without an API key (free tier) + return True + + def search(self, query: str, filters: SearchFilters) -> list[Candidate]: + import requests + + kind = (filters.kind or "video").lower() + if kind == "image": + return [] + + headers: dict[str, str] = {} + api_key = os.environ.get("COVERR_API_KEY") + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + + params: dict[str, Any] = { + "query": query, + "page_size": max(1, min(filters.per_page, 25)), + "page": max(1, filters.page), + } + + r = requests.get( + _SEARCH_URL, + headers=headers, + params=params, + timeout=30, + ) + r.raise_for_status() + data = r.json() + hits = data.get("hits", []) or data.get("videos", []) or [] + + out: list[Candidate] = [] + for v in hits: + 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 + + # Coverr provides multiple URLs for different qualities + urls = v.get("urls", {}) or {} + download_url = ( + urls.get("mp4_download") + or urls.get("mp4_1080") + or urls.get("mp4_720") + or urls.get("mp4_preview") + or "" + ) + if not download_url: + continue + + width = int(v.get("width") or 1920) + height = int(v.get("height") or 1080) + if filters.min_width and width < filters.min_width: + continue + + tags = v.get("tags", "") or "" + if isinstance(tags, list): + tags = " ".join(tags) + title = v.get("title", "") or "" + source_tags = f"{title} {tags}".strip() + + out.append( + Candidate( + source=self.name, + source_id=str(v.get("id") or v.get("slug", "")), + source_url=v.get("url", "") or f"https://coverr.co/videos/{v.get('slug', '')}", + download_url=download_url, + kind="video", + width=width, + height=height, + duration=duration, + creator=v.get("creator", {}).get("name", "") if isinstance(v.get("creator"), dict) else "", + license=_LICENSE, + source_tags=source_tags, + thumbnail_url=urls.get("poster") or urls.get("thumbnail") or "", + extra={ + "slug": v.get("slug"), + "category": v.get("category"), + }, + ) + ) + return out + + def download(self, candidate: Candidate, out_path: Path) -> Path: + import requests + + 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 diff --git a/tools/video/stock_sources/dareful.py b/tools/video/stock_sources/dareful.py new file mode 100644 index 0000000..03347cf --- /dev/null +++ b/tools/video/stock_sources/dareful.py @@ -0,0 +1,187 @@ +"""Dareful stock video source adapter. + +Scrapes the Dareful website (``dareful.com``, formerly +StockFootageForFree.com) for free 4K nature footage. Dareful is a +boutique collection curated by a single creator (Joel Holland) offering +high-quality landscape, forest, mountain, waterfall, and time-lapse +footage. + +Licensed under CC BY 4.0 (attribution required). No API available. + +What Dareful is good for +------------------------ +- 4K nature B-roll (mountains, forests, waterfalls, oceans) +- Aerial landscape footage +- Time-lapse sequences (sunrise, clouds, stars) +- Consistent visual style (single creator) +""" +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Any + +from .base import Candidate, SearchFilters + +_log = logging.getLogger(__name__) + +_BASE_URL = "https://www.dareful.com" +_LICENSE = "Creative Commons Attribution 4.0 (CC BY 4.0, attribution required)" + + +class DarefulSource: + """Dareful nature footage adapter. Satisfies `StockSource`.""" + + name = "dareful" + display_name = "Dareful" + provider = "dareful" + priority = 50 + install_instructions = ( + "Dareful works without an API key. Scrapes the Dareful website. " + "Requires beautifulsoup4: pip install beautifulsoup4" + ) + supports = {"video": True, "image": False} + + def is_available(self) -> bool: + try: + import bs4 # noqa: F401 + return True + except ImportError: + return False + + def search(self, query: str, filters: SearchFilters) -> list[Candidate]: + import requests + from bs4 import BeautifulSoup + + kind = (filters.kind or "video").lower() + if kind == "image": + return [] + + try: + r = requests.get( + _BASE_URL, + params={"s": query}, + timeout=30, + headers={"User-Agent": "OpenMontage/1.0"}, + ) + r.raise_for_status() + except Exception as e: + _log.warning("Dareful search failed: %s", e) + return [] + + soup = BeautifulSoup(r.text, "html.parser") + out: list[Candidate] = [] + + # Find video post cards + cards = soup.select("article, .post, .entry, .video-item, .grid-item") + for card in cards[:filters.per_page]: + link_el = card.select_one("a[href]") + if not link_el: + continue + + href = link_el.get("href", "") + if not href: + continue + if not href.startswith("http"): + href = f"{_BASE_URL}{href}" + + title = "" + title_el = card.select_one("h2, h3, .entry-title, .title") + if title_el: + title = title_el.get_text(strip=True) + if not title: + title = link_el.get_text(strip=True) + + thumb = "" + img_el = card.select_one("img") + if img_el: + thumb = img_el.get("src", "") or img_el.get("data-src", "") or "" + + clip_id = href.rstrip("/").rsplit("/", 1)[-1] + + out.append( + Candidate( + source=self.name, + source_id=f"dareful_{clip_id}", + source_url=href, + download_url=href, + kind="video", + width=3840, # Dareful is primarily 4K + height=2160, + duration=0.0, + creator="Joel Holland (Dareful)", + license=_LICENSE, + source_tags=f"{title} nature landscape 4k {query}", + thumbnail_url=thumb, + extra={"detail_url": href}, + ) + ) + + return out + + def download(self, candidate: Candidate, out_path: Path) -> Path: + """Download by resolving the detail page for the actual file URL.""" + import requests + from bs4 import BeautifulSoup + + out_path = Path(out_path) + out_path.parent.mkdir(parents=True, exist_ok=True) + + detail_url = candidate.extra.get("detail_url", candidate.download_url) + + if any(detail_url.lower().endswith(ext) for ext in (".mp4", ".mov", ".webm")): + return self._stream_download(detail_url, out_path) + + try: + r = requests.get( + detail_url, timeout=30, + headers={"User-Agent": "OpenMontage/1.0"}, + ) + r.raise_for_status() + soup = BeautifulSoup(r.text, "html.parser") + + download_url = None + + # Look for download links + for a in soup.select("a[href]"): + href = a.get("href", "") + text = (a.get_text(strip=True) or "").lower() + if any(ext in href.lower() for ext in [".mp4", ".mov", ".webm"]): + download_url = href + break + if "download" in text and href: + download_url = href + break + + # Check video elements + if not download_url: + for source in soup.select("video source[src], video[src]"): + src = source.get("src", "") + if src: + download_url = src + break + + if not download_url: + raise ValueError(f"Could not find download URL on Dareful page: {detail_url}") + + if not download_url.startswith("http"): + download_url = f"{_BASE_URL}{download_url}" + + return self._stream_download(download_url, out_path) + + except Exception as e: + raise RuntimeError(f"Dareful download failed for {detail_url}: {e}") from e + + def _stream_download(self, url: str, out_path: Path) -> Path: + import requests + + with requests.get( + url, stream=True, timeout=180, + headers={"User-Agent": "OpenMontage/1.0"}, + ) 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 diff --git a/tools/video/stock_sources/esa.py b/tools/video/stock_sources/esa.py new file mode 100644 index 0000000..a38b429 --- /dev/null +++ b/tools/video/stock_sources/esa.py @@ -0,0 +1,203 @@ +"""European Space Agency (ESA) stock source adapter. + +Scrapes the ESA Multimedia gallery (``www.esa.int/ESA_Multimedia/``) for +free space footage. ESA content is licensed under CC BY-SA 3.0 IGO +(general) or CC BY 4.0 (Webb/Hubble imagery). Attribution is required. + +No API available — this adapter scrapes the ESA website's search and +detail pages. Content includes satellite imagery, mission footage, +astronaut activities, rocket launches, Earth observation, and Hubble/Webb +telescope imagery and animations. + +What ESA is good for +-------------------- +- European space missions (Rosetta, ExoMars, Galileo) +- Hubble and James Webb Space Telescope imagery +- Earth observation from space +- ISS footage (European contributions) +- Ariane rocket launches +""" +from __future__ import annotations + +import logging +import re +from pathlib import Path +from typing import Any + +from .base import Candidate, SearchFilters + +_log = logging.getLogger(__name__) + +_SEARCH_URL = "https://www.esa.int/ESA_Multimedia/Search" +_VIDEO_SEARCH_URL = "https://www.esa.int/ESA_Multimedia/Videos" +_LICENSE = "CC BY-SA 3.0 IGO (ESA, attribution required)" + + +class ESASource: + """European Space Agency multimedia adapter. Satisfies `StockSource`.""" + + name = "esa" + display_name = "ESA (European Space Agency)" + provider = "esa" + priority = 45 + install_instructions = ( + "ESA works without an API key. Scrapes the ESA Multimedia gallery. " + "Requires beautifulsoup4: pip install beautifulsoup4" + ) + supports = {"video": True, "image": True} + + def is_available(self) -> bool: + try: + import bs4 # noqa: F401 + return True + except ImportError: + return False + + def search(self, query: str, filters: SearchFilters) -> list[Candidate]: + import requests + from bs4 import BeautifulSoup + + kind = (filters.kind or "video").lower() + + params: dict[str, Any] = { + "SearchText": query, + "result_type": "videos" if kind == "video" else "images" if kind == "image" else "", + } + + try: + r = requests.get( + _SEARCH_URL, + params=params, + timeout=30, + headers={"User-Agent": "OpenMontage/1.0"}, + ) + r.raise_for_status() + except Exception as e: + _log.warning("ESA search failed: %s", e) + return [] + + soup = BeautifulSoup(r.text, "html.parser") + out: list[Candidate] = [] + + # Find video/image cards on the search results page + cards = soup.select(".grid-item, .media-item, .search-result-item, article") + for card in cards[:filters.per_page]: + link_el = card.select_one("a[href]") + if not link_el: + continue + + href = link_el.get("href", "") + if not href: + continue + if not href.startswith("http"): + href = f"https://www.esa.int{href}" + + title = "" + title_el = card.select_one("h3, h2, .title, .card-title") + if title_el: + title = title_el.get_text(strip=True) + + img_el = card.select_one("img") + thumb = "" + if img_el: + thumb = img_el.get("src", "") or img_el.get("data-src", "") or "" + if thumb and not thumb.startswith("http"): + thumb = f"https://www.esa.int{thumb}" + + # Determine kind from URL or content + is_video = "/Videos/" in href or "/Video/" in href + candidate_kind = "video" if is_video else "image" + + if kind == "video" and not is_video: + continue + if kind == "image" and is_video: + continue + + out.append( + Candidate( + source=self.name, + source_id=f"esa_{hash(href) & 0xFFFFFFFF:08x}", + source_url=href, + download_url=href, # Will be resolved in download() + kind=candidate_kind, + width=0, + height=0, + duration=0.0, + creator="European Space Agency (ESA)", + license=_LICENSE, + source_tags=title, + thumbnail_url=thumb, + extra={"detail_url": href}, + ) + ) + + return out + + def download(self, candidate: Candidate, out_path: Path) -> Path: + """Download by first resolving the detail page for the actual file URL.""" + import requests + from bs4 import BeautifulSoup + + out_path = Path(out_path) + out_path.parent.mkdir(parents=True, exist_ok=True) + + detail_url = candidate.extra.get("detail_url", candidate.download_url) + + # If it's already a direct media URL, download directly + if any(detail_url.lower().endswith(ext) for ext in (".mp4", ".mov", ".jpg", ".png")): + return self._stream_download(detail_url, out_path) + + # Otherwise, scrape the detail page for the download link + try: + r = requests.get( + detail_url, + timeout=30, + headers={"User-Agent": "OpenMontage/1.0"}, + ) + r.raise_for_status() + soup = BeautifulSoup(r.text, "html.parser") + + # Look for video download links + download_url = None + for a in soup.select("a[href]"): + href = a.get("href", "") + text = a.get_text(strip=True).lower() + if any(ext in href.lower() for ext in [".mp4", ".mov", ".webm"]): + download_url = href + break + if "download" in text and href: + download_url = href + break + + # Check for video source tags + if not download_url: + for source in soup.select("video source[src], source[src]"): + src = source.get("src", "") + if src: + download_url = src + break + + if not download_url: + raise ValueError(f"Could not find download URL on ESA detail page: {detail_url}") + + if not download_url.startswith("http"): + download_url = f"https://www.esa.int{download_url}" + + return self._stream_download(download_url, out_path) + + except Exception as e: + raise RuntimeError(f"ESA download failed for {detail_url}: {e}") from e + + def _stream_download(self, url: str, out_path: Path) -> Path: + import requests + + with requests.get( + url, stream=True, timeout=180, + headers={"User-Agent": "OpenMontage/1.0"}, + ) 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 diff --git a/tools/video/stock_sources/jaxa.py b/tools/video/stock_sources/jaxa.py new file mode 100644 index 0000000..2b91312 --- /dev/null +++ b/tools/video/stock_sources/jaxa.py @@ -0,0 +1,208 @@ +"""JAXA (Japan Aerospace Exploration Agency) stock source adapter. + +Scrapes the JAXA Digital Archives (``jda.jaxa.jp``) for Japanese space +agency footage. Content includes satellite launches, ISS operations, +Earth observation, planetary probes, and moon/planetary imagery. + +Generally available for educational/informational use. Specific terms +vary per item — check JAXA's usage guidelines. No API available. + +What JAXA is good for +--------------------- +- Japanese space missions (Hayabusa, SLIM, H-IIA/H3 rockets) +- ISS footage (Japanese module Kibo) +- Earth observation from JAXA satellites +- Moon and asteroid imagery +- Complement to NASA for non-U.S. space footage +""" +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Any + +from .base import Candidate, SearchFilters + +_log = logging.getLogger(__name__) + +_BASE_URL = "https://jda.jaxa.jp" +_SEARCH_URL = "https://jda.jaxa.jp/result.php" +_LICENSE = "JAXA Digital Archives License (educational/informational use, verify per item)" + + +class JAXASource: + """JAXA Digital Archives adapter. Satisfies `StockSource`.""" + + name = "jaxa" + display_name = "JAXA (Japan Space Agency)" + provider = "jaxa" + priority = 55 + install_instructions = ( + "JAXA works without an API key. Scrapes the JAXA Digital Archives. " + "Requires beautifulsoup4: pip install beautifulsoup4" + ) + supports = {"video": True, "image": True} + + def is_available(self) -> bool: + try: + import bs4 # noqa: F401 + return True + except ImportError: + return False + + def search(self, query: str, filters: SearchFilters) -> list[Candidate]: + import requests + from bs4 import BeautifulSoup + + kind = (filters.kind or "video").lower() + + params: dict[str, Any] = { + "lang": "e", # English + "keyword": query, + } + + # JAXA category filter + if kind == "video": + params["category"] = "3" # Videos/movies + elif kind == "image": + params["category"] = "1" # Photos + + try: + r = requests.get( + _SEARCH_URL, + params=params, + timeout=30, + headers={"User-Agent": "OpenMontage/1.0"}, + ) + r.raise_for_status() + except Exception as e: + _log.warning("JAXA search failed: %s", e) + return [] + + soup = BeautifulSoup(r.text, "html.parser") + out: list[Candidate] = [] + + # Find result items + items = soup.select(".result-item, .photo-item, .movie-item, .item, li.list-item, .gallery-item") + for item in items[:filters.per_page]: + link_el = item.select_one("a[href]") + if not link_el: + continue + + href = link_el.get("href", "") + if not href: + continue + if not href.startswith("http"): + href = f"{_BASE_URL}/{href.lstrip('/')}" + + title = "" + title_el = item.select_one(".title, h3, h2, p, .caption") + if title_el: + title = title_el.get_text(strip=True) + if not title: + title = link_el.get("title", "") or link_el.get_text(strip=True) + + thumb = "" + img_el = item.select_one("img") + if img_el: + thumb = img_el.get("src", "") or img_el.get("data-src", "") or "" + if thumb and not thumb.startswith("http"): + thumb = f"{_BASE_URL}/{thumb.lstrip('/')}" + + candidate_kind = "video" if kind == "video" else "image" + clip_id = href.rstrip("/").rsplit("/", 1)[-1].split("?")[0].split(".")[0] + + out.append( + Candidate( + source=self.name, + source_id=f"jaxa_{clip_id}", + source_url=href, + download_url=href, + kind=candidate_kind, + width=0, + height=0, + duration=0.0, + creator="JAXA", + license=_LICENSE, + source_tags=f"{title} space japan {query}", + thumbnail_url=thumb, + extra={"detail_url": href}, + ) + ) + + return out + + def download(self, candidate: Candidate, out_path: Path) -> Path: + """Download by resolving the detail page for the actual file URL.""" + import requests + from bs4 import BeautifulSoup + + out_path = Path(out_path) + out_path.parent.mkdir(parents=True, exist_ok=True) + + detail_url = candidate.extra.get("detail_url", candidate.download_url) + + if any(detail_url.lower().endswith(ext) for ext in (".mp4", ".mov", ".webm", ".jpg", ".png")): + return self._stream_download(detail_url, out_path) + + try: + r = requests.get( + detail_url, timeout=30, + headers={"User-Agent": "OpenMontage/1.0"}, + ) + r.raise_for_status() + soup = BeautifulSoup(r.text, "html.parser") + + download_url = None + + # Look for download links or video sources + for a in soup.select("a[href]"): + href = a.get("href", "") + text = (a.get_text(strip=True) or "").lower() + if any(ext in href.lower() for ext in [".mp4", ".mov", ".wmv", ".mpg"]): + download_url = href + break + if "download" in text and href: + download_url = href + break + + # Video elements + if not download_url: + for source in soup.select("video source[src], video[src]"): + src = source.get("src", "") + if src: + download_url = src + break + + # High-res image links + if not download_url and candidate.kind == "image": + for a in soup.select("a[href]"): + href = a.get("href", "") + if any(ext in href.lower() for ext in [".jpg", ".jpeg", ".png", ".tif"]): + download_url = href + break + + if not download_url: + raise ValueError(f"Could not find download URL on JAXA page: {detail_url}") + + if not download_url.startswith("http"): + download_url = f"{_BASE_URL}/{download_url.lstrip('/')}" + + return self._stream_download(download_url, out_path) + + except Exception as e: + raise RuntimeError(f"JAXA download failed for {detail_url}: {e}") from e + + def _stream_download(self, url: str, out_path: Path) -> Path: + import requests + + with requests.get( + url, stream=True, timeout=180, + headers={"User-Agent": "OpenMontage/1.0"}, + ) 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 diff --git a/tools/video/stock_sources/loc.py b/tools/video/stock_sources/loc.py new file mode 100644 index 0000000..bc7208c --- /dev/null +++ b/tools/video/stock_sources/loc.py @@ -0,0 +1,233 @@ +"""Library of Congress stock source adapter. + +Wraps the loc.gov JSON API behind the unified `StockSource` protocol. +The Library of Congress holds 25+ digital collections of film and video +materials including early cinema, newsreels, documentaries, and cultural +recordings. Many items are public domain (pre-1928 or U.S. government). + +No API key required. Rate limiting is polite-crawl based. + +Fetch pattern +------------- +Two-stage. The search endpoint (``loc.gov/search``) returns items with +links to detail pages. The detail page JSON contains downloadable +resources including video files. Items are filtered by ``original-format`` +to target film/video content. + +What Library of Congress is good for +------------------------------------ +- Early American cinema (pre-1928, public domain) +- Historical newsreels and documentaries +- Cultural recordings, folk traditions +- Government and civic footage +""" +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Any + +from .base import Candidate, SearchFilters + +_log = logging.getLogger(__name__) + +_SEARCH_URL = "https://www.loc.gov/search/" +_LICENSE_PD = "Public domain (Library of Congress)" +_LICENSE_CHECK = "Rights status varies — verify per item (Library of Congress)" + +# Video-related format filters for the LoC API +_VIDEO_FORMATS = ["film/video", "motion picture"] + + +class LibraryOfCongressSource: + """Library of Congress adapter. Satisfies `StockSource`.""" + + name = "loc" + display_name = "Library of Congress" + provider = "loc" + priority = 40 + install_instructions = ( + "Library of Congress works without an API key. " + "No setup needed." + ) + supports = {"video": True, "image": True} + + def is_available(self) -> bool: + return True + + def search(self, query: str, filters: SearchFilters) -> list[Candidate]: + import requests + + kind = (filters.kind or "video").lower() + + params: dict[str, Any] = { + "q": query, + "fo": "json", + "c": max(1, min(filters.per_page, 50)), + "sp": max(1, filters.page), + } + + # Filter by format + if kind == "video": + params["fa"] = "original-format:film/video" + elif kind == "image": + params["fa"] = "original-format:photo, print, drawing" + + try: + r = requests.get( + _SEARCH_URL, + params=params, + timeout=30, + headers={"Accept": "application/json"}, + ) + r.raise_for_status() + data = r.json() + except Exception as e: + _log.warning("Library of Congress search failed: %s", e) + return [] + + results = data.get("results", []) or [] + out: list[Candidate] = [] + + for item in results: + candidates = self._extract_candidates(item, kind, filters) + out.extend(candidates) + + return out + + def _extract_candidates( + self, item: dict, kind: str, filters: SearchFilters + ) -> list[Candidate]: + """Extract downloadable candidates from a LoC search result.""" + item_id = item.get("id", "") or "" + if not item_id: + return [] + + title = item.get("title", "") or "" + description = "" + desc_list = item.get("description", []) + if isinstance(desc_list, list) and desc_list: + description = desc_list[0] if isinstance(desc_list[0], str) else "" + elif isinstance(desc_list, str): + description = desc_list + + subjects = item.get("subject", []) or [] + if isinstance(subjects, list): + subjects = " ".join(s for s in subjects if isinstance(s, str)) + source_tags = f"{title} {description} {subjects}".strip() + + source_url = item_id if item_id.startswith("http") else f"https://www.loc.gov{item_id}" + + # Determine rights + rights = item.get("rights", []) or [] + if isinstance(rights, list): + rights_str = " ".join(r for r in rights if isinstance(r, str)).lower() + else: + rights_str = str(rights).lower() + lic = _LICENSE_PD if "public domain" in rights_str or "no known" in rights_str else _LICENSE_CHECK + + # Look for downloadable resources + resources = item.get("resources", []) or [] + # Also check the item's direct links + image_url = "" + if isinstance(item.get("image_url"), list): + urls = item["image_url"] + image_url = urls[0] if urls else "" + elif isinstance(item.get("image_url"), str): + image_url = item["image_url"] + + out: list[Candidate] = [] + + # Try resources first + for res in resources: + if not isinstance(res, dict): + continue + files = res.get("files", []) or [] + for file_group in files: + if not isinstance(file_group, list): + continue + for f in file_group: + if not isinstance(f, dict): + continue + url = f.get("url", "") or "" + mime = (f.get("mimetype", "") or "").lower() + if not url: + continue + + is_video = "video" in mime or any( + url.lower().endswith(ext) + for ext in (".mp4", ".mov", ".avi", ".webm") + ) + is_image = "image" in mime or any( + url.lower().endswith(ext) + for ext in (".jpg", ".jpeg", ".png", ".tif") + ) + + if kind == "video" and not is_video: + continue + if kind == "image" and not is_image: + continue + if not is_video and not is_image: + continue + + full_url = url if url.startswith("http") else f"https://www.loc.gov{url}" + + out.append( + Candidate( + source=self.name, + source_id=f"loc_{hash(full_url) & 0xFFFFFFFF:08x}", + source_url=source_url, + download_url=full_url, + kind="video" if is_video else "image", + width=int(f.get("width") or 0), + height=int(f.get("height") or 0), + duration=0.0, # LoC doesn't expose duration in search + creator="Library of Congress", + license=lic, + source_tags=source_tags, + thumbnail_url=image_url, + extra={ + "item_id": item_id, + "mime": mime, + }, + ) + ) + + # If no resources found but we have an image_url for image kind + if not out and kind in ("image", "any") and image_url: + full_url = image_url if image_url.startswith("http") else f"https://www.loc.gov{image_url}" + out.append( + Candidate( + source=self.name, + source_id=f"loc_{hash(full_url) & 0xFFFFFFFF:08x}", + source_url=source_url, + download_url=full_url, + kind="image", + width=0, + height=0, + duration=0.0, + creator="Library of Congress", + license=lic, + source_tags=source_tags, + thumbnail_url=image_url, + extra={"item_id": item_id}, + ) + ) + + return out + + def download(self, candidate: Candidate, out_path: Path) -> Path: + import requests + + out_path = Path(out_path) + out_path.parent.mkdir(parents=True, exist_ok=True) + + with requests.get( + candidate.download_url, stream=True, timeout=180 + ) 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 diff --git a/tools/video/stock_sources/mixkit.py b/tools/video/stock_sources/mixkit.py new file mode 100644 index 0000000..bf665d3 --- /dev/null +++ b/tools/video/stock_sources/mixkit.py @@ -0,0 +1,216 @@ +"""Mixkit (by Envato) stock video source adapter. + +Scrapes the Mixkit website (``mixkit.co``) for free stock video clips. +Mixkit offers curated, high-quality footage (HD and 4K) under a free +licence with no attribution required. The library is smaller than +Pixabay/Pexels but has higher average quality due to Envato's curation. + +No API available — this adapter scrapes Mixkit search pages. + +What Mixkit is good for +----------------------- +- High-quality curated B-roll (nature, business, technology, lifestyle) +- Clean, modern footage with consistent quality +- No-attribution-needed clips for quick gap-fills +- Nature and landscape establishing shots +""" +from __future__ import annotations + +import logging +import re +from pathlib import Path +from typing import Any + +from .base import Candidate, SearchFilters + +_log = logging.getLogger(__name__) + +_SEARCH_URL = "https://mixkit.co/free-stock-video/" +_LICENSE = "Mixkit License (free for commercial and personal use, no attribution required)" + + +class MixkitSource: + """Mixkit video adapter. Satisfies `StockSource`.""" + + name = "mixkit" + display_name = "Mixkit" + provider = "envato" + priority = 19 + install_instructions = ( + "Mixkit works without an API key. Scrapes the Mixkit website. " + "Requires beautifulsoup4: pip install beautifulsoup4" + ) + supports = {"video": True, "image": False} + + def is_available(self) -> bool: + try: + import bs4 # noqa: F401 + return True + except ImportError: + return False + + def search(self, query: str, filters: SearchFilters) -> list[Candidate]: + import requests + from bs4 import BeautifulSoup + + kind = (filters.kind or "video").lower() + if kind == "image": + return [] + + # Mixkit search URL pattern + slug = query.lower().replace(" ", "-") + search_url = f"https://mixkit.co/free-stock-video/{slug}/" + + try: + r = requests.get( + search_url, + timeout=30, + headers={"User-Agent": "OpenMontage/1.0"}, + ) + r.raise_for_status() + except Exception as e: + _log.warning("Mixkit search failed: %s", e) + return [] + + soup = BeautifulSoup(r.text, "html.parser") + out: list[Candidate] = [] + + # Mixkit lists video cards with preview videos and download links + cards = soup.select(".item-grid__item, .video-item, article, [class*='VideoCard']") + for card in cards[:filters.per_page]: + link_el = card.select_one("a[href]") + if not link_el: + continue + + href = link_el.get("href", "") + if not href: + continue + if not href.startswith("http"): + href = f"https://mixkit.co{href}" + + # Skip non-video links + if "/free-stock-video/" not in href and "/video/" not in href: + continue + + title = "" + title_el = card.select_one("h3, h2, .title, [class*='title']") + if title_el: + title = title_el.get_text(strip=True) + if not title: + title = link_el.get_text(strip=True) + + # Thumbnail + thumb = "" + img_el = card.select_one("img") + if img_el: + thumb = img_el.get("src", "") or img_el.get("data-src", "") or "" + + # Video preview + video_el = card.select_one("video source[src], video[src]") + preview_url = "" + if video_el: + preview_url = video_el.get("src", "") or "" + + # Extract ID from URL + clip_id = href.rstrip("/").rsplit("/", 1)[-1] if href else "" + + out.append( + Candidate( + source=self.name, + source_id=f"mixkit_{clip_id}", + source_url=href, + download_url=href, # Resolved in download() + kind="video", + width=0, + height=0, + duration=0.0, + creator="Mixkit", + license=_LICENSE, + source_tags=f"{title} {query}", + thumbnail_url=thumb, + extra={ + "detail_url": href, + "preview_url": preview_url, + }, + ) + ) + + return out + + def download(self, candidate: Candidate, out_path: Path) -> Path: + """Download by resolving the detail page for the actual download URL.""" + import requests + from bs4 import BeautifulSoup + + out_path = Path(out_path) + out_path.parent.mkdir(parents=True, exist_ok=True) + + detail_url = candidate.extra.get("detail_url", candidate.download_url) + + # Direct media URL + if any(detail_url.lower().endswith(ext) for ext in (".mp4", ".mov", ".webm")): + return self._stream_download(detail_url, out_path) + + try: + r = requests.get( + detail_url, timeout=30, + headers={"User-Agent": "OpenMontage/1.0"}, + ) + r.raise_for_status() + soup = BeautifulSoup(r.text, "html.parser") + + download_url = None + + # Look for download button/link + for a in soup.select("a[href]"): + href = a.get("href", "") + text = (a.get_text(strip=True) or "").lower() + classes = " ".join(a.get("class", [])) + if "download" in text or "download" in classes: + if href and any(ext in href.lower() for ext in [".mp4", ".mov", ".webm"]): + download_url = href + break + elif href and "/download/" in href: + download_url = href + break + + # Look for video source tags + if not download_url: + for source in soup.select("video source[src]"): + src = source.get("src", "") + if src and any(ext in src.lower() for ext in [".mp4", ".mov"]): + download_url = src + break + + # Look for data attributes with video URLs + if not download_url: + for el in soup.select("[data-video-url], [data-download-url], [data-src]"): + url = el.get("data-video-url") or el.get("data-download-url") or el.get("data-src") or "" + if url and any(ext in url.lower() for ext in [".mp4", ".mov"]): + download_url = url + break + + if not download_url: + raise ValueError(f"Could not find download URL on Mixkit page: {detail_url}") + + if not download_url.startswith("http"): + download_url = f"https://mixkit.co{download_url}" + + return self._stream_download(download_url, out_path) + + except Exception as e: + raise RuntimeError(f"Mixkit download failed for {detail_url}: {e}") from e + + def _stream_download(self, url: str, out_path: Path) -> Path: + import requests + + with requests.get( + url, stream=True, timeout=120, + headers={"User-Agent": "OpenMontage/1.0"}, + ) 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 diff --git a/tools/video/stock_sources/nara.py b/tools/video/stock_sources/nara.py new file mode 100644 index 0000000..decd423 --- /dev/null +++ b/tools/video/stock_sources/nara.py @@ -0,0 +1,197 @@ +"""U.S. National Archives (NARA) stock source adapter. + +Wraps the NARA Catalog API (``catalog.archives.gov/api/v2``) behind the +unified `StockSource` protocol. NARA holds billions of records including +significant film and video holdings — all U.S. federal government work +and therefore public domain. + +No API key required for basic searching. For higher rate limits, email +Catalog_API@nara.gov to request a key. Rate limit: ~10,000 queries per +month per API key. + +Fetch pattern +------------- +Two-stage like NASA. The search endpoint returns metadata records. Each +record may contain digital objects (files) in ``objects``. We follow +those to find downloadable video files. + +What NARA is good for +--------------------- +- U.S. historical footage (military, presidential, space, civil rights) +- WWII, Cold War, Apollo era footage +- Government program footage and newsreels +- Any "march of history" documentary sequence +""" +from __future__ import annotations + +import logging +import os +from pathlib import Path +from typing import Any + +from .base import Candidate, SearchFilters + +_log = logging.getLogger(__name__) + +_SEARCH_URL = "https://catalog.archives.gov/api/v2/search" +_LICENSE = "Public domain (U.S. federal government work)" + + +class NARASource: + """U.S. National Archives adapter. Satisfies `StockSource`.""" + + name = "nara" + display_name = "U.S. National Archives" + provider = "nara" + priority = 35 + install_instructions = ( + "NARA works without an API key. " + "Set NARA_API_KEY in .env for higher rate limits." + ) + supports = {"video": True, "image": True} + + def is_available(self) -> bool: + # NARA is always available (no key required) + return True + + def search(self, query: str, filters: SearchFilters) -> list[Candidate]: + import requests + + kind = (filters.kind or "video").lower() + + params: dict[str, Any] = { + "q": query, + "rows": max(1, min(filters.per_page, 50)), + "offset": (max(1, filters.page) - 1) * filters.per_page, + } + + # Filter by type if possible + if kind == "video": + params["type"] = "moving-image" + elif kind == "image": + params["type"] = "still-image" + + headers: dict[str, str] = {} + api_key = os.environ.get("NARA_API_KEY") + if api_key: + headers["x-api-key"] = api_key + + try: + r = requests.get( + _SEARCH_URL, + headers=headers, + params=params, + timeout=30, + ) + r.raise_for_status() + data = r.json() + except Exception as e: + _log.warning("NARA search failed: %s", e) + return [] + + results = data.get("results", []) or [] + out: list[Candidate] = [] + + for item in results: + candidates = self._extract_candidates(item, kind, filters) + out.extend(candidates) + + return out + + def _extract_candidates( + self, item: dict, kind: str, filters: SearchFilters + ) -> list[Candidate]: + """Extract downloadable candidates from a NARA catalog record.""" + naid = str(item.get("naId", "") or "") + if not naid: + return [] + + title = item.get("title", "") or "" + description = item.get("scopeAndContentNote", "") or "" + source_tags = f"{title} {description}".strip() + source_url = f"https://catalog.archives.gov/id/{naid}" + + # Look for digital objects + objects = item.get("objects", []) or [] + if not objects: + # Try alternate field names + digital = item.get("digitalObjects", []) or [] + objects = digital + + out: list[Candidate] = [] + for obj in objects: + file_url = obj.get("url") or obj.get("fileUrl") or "" + if not file_url: + continue + + # Determine kind from mime type or file extension + mime = (obj.get("mimeType", "") or "").lower() + ext = file_url.rsplit(".", 1)[-1].lower() if "." in file_url else "" + + is_video = ( + "video" in mime + or ext in ("mp4", "mov", "avi", "wmv", "mkv", "webm") + ) + is_image = ( + "image" in mime + or ext in ("jpg", "jpeg", "png", "tif", "tiff", "gif") + ) + + if kind == "video" and not is_video: + continue + if kind == "image" and not is_image: + continue + if not is_video and not is_image: + continue + + candidate_kind = "video" if is_video else "image" + width = int(obj.get("width") or 0) + height = int(obj.get("height") or 0) + duration = float(obj.get("duration") or 0) + + # Duration filters (client-side) + if candidate_kind == "video": + if filters.min_duration and duration and duration < filters.min_duration: + continue + if filters.max_duration and duration and duration > filters.max_duration: + continue + + out.append( + Candidate( + source=self.name, + source_id=f"{naid}_{obj.get('objectId', len(out))}", + source_url=source_url, + download_url=file_url, + kind=candidate_kind, + width=width, + height=height, + duration=duration, + creator="U.S. National Archives", + license=_LICENSE, + source_tags=source_tags, + thumbnail_url=obj.get("thumbnailUrl", "") or "", + extra={ + "naId": naid, + "mime": mime, + "fileSize": obj.get("fileSize"), + }, + ) + ) + + return out + + def download(self, candidate: Candidate, out_path: Path) -> Path: + import requests + + out_path = Path(out_path) + out_path.parent.mkdir(parents=True, exist_ok=True) + + with requests.get( + candidate.download_url, stream=True, timeout=180 + ) 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 diff --git a/tools/video/stock_sources/noaa.py b/tools/video/stock_sources/noaa.py new file mode 100644 index 0000000..b150fb4 --- /dev/null +++ b/tools/video/stock_sources/noaa.py @@ -0,0 +1,197 @@ +"""NOAA (National Oceanic and Atmospheric Administration) stock source adapter. + +Scrapes the NOAA Ocean Exploration Video Portal and NOAA multimedia pages +for free ocean, weather, and atmospheric footage. All content is public +domain (U.S. federal government work). + +No API available for video — this adapter scrapes NOAA web pages. +Content includes deep-sea ROV footage, marine life, coral reefs, +underwater volcanism, weather events, and atmospheric phenomena. + +What NOAA is good for +--------------------- +- Deep-sea ROV footage (unique content not available anywhere else) +- Marine life close-ups (jellyfish, octopus, deep-sea creatures) +- Coral reef ecosystems +- Hurricane and storm footage +- Weather satellite imagery +- Coastal and oceanic research footage +""" +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Any + +from .base import Candidate, SearchFilters + +_log = logging.getLogger(__name__) + +_SEARCH_URL = "https://www.ncei.noaa.gov/access/ocean-exploration/video/" +_MULTIMEDIA_URL = "https://www.noaa.gov/multimedia/videos" +_LICENSE = "Public domain (U.S. federal government work, NOAA)" + + +class NOAASource: + """NOAA ocean and atmospheric multimedia adapter. Satisfies `StockSource`.""" + + name = "noaa" + display_name = "NOAA (Ocean & Atmosphere)" + provider = "noaa" + priority = 48 + install_instructions = ( + "NOAA works without an API key. Scrapes the NOAA multimedia pages. " + "Requires beautifulsoup4: pip install beautifulsoup4" + ) + supports = {"video": True, "image": True} + + def is_available(self) -> bool: + try: + import bs4 # noqa: F401 + return True + except ImportError: + return False + + def search(self, query: str, filters: SearchFilters) -> list[Candidate]: + import requests + from bs4 import BeautifulSoup + + kind = (filters.kind or "video").lower() + if kind == "image": + return [] # Primarily a video source + + # Try the NOAA Ocean Exploration video portal + out: list[Candidate] = [] + + try: + # NOAA multimedia search + r = requests.get( + "https://www.noaa.gov/search", + params={"query": query, "type": "video"}, + timeout=30, + headers={"User-Agent": "OpenMontage/1.0"}, + ) + r.raise_for_status() + soup = BeautifulSoup(r.text, "html.parser") + + cards = soup.select(".views-row, .search-result, article, .media-item") + for card in cards[:filters.per_page]: + link_el = card.select_one("a[href]") + if not link_el: + continue + + href = link_el.get("href", "") + if not href: + continue + if not href.startswith("http"): + href = f"https://www.noaa.gov{href}" + + title = "" + title_el = card.select_one("h2, h3, .title, .field-content") + if title_el: + title = title_el.get_text(strip=True) + if not title: + title = link_el.get_text(strip=True) + + img_el = card.select_one("img") + thumb = "" + if img_el: + thumb = img_el.get("src", "") or img_el.get("data-src", "") or "" + if thumb and not thumb.startswith("http"): + thumb = f"https://www.noaa.gov{thumb}" + + out.append( + Candidate( + source=self.name, + source_id=f"noaa_{hash(href) & 0xFFFFFFFF:08x}", + source_url=href, + download_url=href, # Resolved in download() + kind="video", + width=0, + height=0, + duration=0.0, + creator="NOAA", + license=_LICENSE, + source_tags=f"{title} ocean marine weather atmosphere {query}", + thumbnail_url=thumb, + extra={"detail_url": href}, + ) + ) + + except Exception as e: + _log.warning("NOAA search failed: %s", e) + + return out + + def download(self, candidate: Candidate, out_path: Path) -> Path: + """Download by resolving the detail page for the actual file URL.""" + import requests + from bs4 import BeautifulSoup + + out_path = Path(out_path) + out_path.parent.mkdir(parents=True, exist_ok=True) + + detail_url = candidate.extra.get("detail_url", candidate.download_url) + + # Direct media URL + if any(detail_url.lower().endswith(ext) for ext in (".mp4", ".mov", ".webm")): + return self._stream_download(detail_url, out_path) + + # Scrape detail page + try: + r = requests.get( + detail_url, timeout=30, + headers={"User-Agent": "OpenMontage/1.0"}, + ) + r.raise_for_status() + soup = BeautifulSoup(r.text, "html.parser") + + download_url = None + + # Look for video elements + for video in soup.select("video source[src], video[src]"): + src = video.get("src", "") + if src: + download_url = src + break + + # Look for download links + if not download_url: + for a in soup.select("a[href]"): + href = a.get("href", "") + if any(ext in href.lower() for ext in [".mp4", ".mov", ".webm"]): + download_url = href + break + + # Look for YouTube embeds + if not download_url: + for iframe in soup.select("iframe[src]"): + src = iframe.get("src", "") + if "youtube" in src or "vimeo" in src: + _log.warning("NOAA video is embedded from %s — cannot download directly", src) + raise ValueError(f"Video is embedded from external platform: {src}") + + if not download_url: + raise ValueError(f"Could not find video URL on NOAA page: {detail_url}") + + if not download_url.startswith("http"): + download_url = f"https://www.noaa.gov{download_url}" + + return self._stream_download(download_url, out_path) + + except Exception as e: + raise RuntimeError(f"NOAA download failed for {detail_url}: {e}") from e + + def _stream_download(self, url: str, out_path: Path) -> Path: + import requests + + with requests.get( + url, stream=True, timeout=180, + headers={"User-Agent": "OpenMontage/1.0"}, + ) 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 diff --git a/tools/video/stock_sources/pixabay_video.py b/tools/video/stock_sources/pixabay_video.py new file mode 100644 index 0000000..9876b2a --- /dev/null +++ b/tools/video/stock_sources/pixabay_video.py @@ -0,0 +1,147 @@ +"""Pixabay Video stock source adapter. + +Wraps the Pixabay Video API (``pixabay.com/api/videos/``) behind the +unified `StockSource` protocol. Pixabay has a large community-contributed +video library (hundreds of thousands of clips) with a CC0-like licence +that allows free commercial use without attribution. + +Uses the same ``PIXABAY_API_KEY`` as the Pixabay Music tool — if you've +already set it for music search, this adapter is automatically available. + +Rate limit: 100 requests per 60 seconds (free tier). The adapter trusts +the API to enforce this and does not self-throttle. + +What Pixabay Video is good for +------------------------------ +- Broad general-purpose footage: nature, people, technology, food, city +- Modern, community-contributed clips (skews recent / lifestyle) +- Quick gap-fills when Pexels doesn't cover a query +- Available up to 1080p (some clips have 4K) +""" +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any, Optional + +from .base import Candidate, SearchFilters + + +_API_URL = "https://pixabay.com/api/videos/" +_LICENSE = "Pixabay Content License (free, no attribution required)" + + +class PixabayVideoSource: + """Pixabay Video adapter. Satisfies `StockSource`.""" + + name = "pixabay_video" + display_name = "Pixabay Video" + provider = "pixabay" + priority = 15 + install_instructions = ( + "Set PIXABAY_API_KEY in .env to enable Pixabay Video search " + "(free key at https://pixabay.com/api/docs/)." + ) + supports = {"video": True, "image": False} + + def is_available(self) -> bool: + return bool(os.environ.get("PIXABAY_API_KEY")) + + def search(self, query: str, filters: SearchFilters) -> list[Candidate]: + import requests + + kind = (filters.kind or "video").lower() + if kind == "image": + return [] # video-only adapter + + params: dict[str, Any] = { + "key": os.environ["PIXABAY_API_KEY"], + "q": query, + "per_page": max(3, min(filters.per_page, 200)), + "page": max(1, filters.page), + "safesearch": "true", + } + if filters.min_duration is not None: + params["min_duration"] = int(filters.min_duration) + if filters.max_duration is not None: + params["max_duration"] = int(filters.max_duration) + + r = requests.get(_API_URL, params=params, timeout=30) + r.raise_for_status() + data = r.json() + hits = data.get("hits", []) or [] + + out: list[Candidate] = [] + for h in hits: + videos = h.get("videos", {}) + rend = _pick_rendition(videos, min_width=filters.min_width or 0) + if rend is None: + continue + + duration = float(h.get("duration", 0) or 0) + tags = h.get("tags", "") or "" + + out.append( + Candidate( + source=self.name, + source_id=str(h.get("id")), + source_url=h.get("pageURL", "") or "", + download_url=rend["url"], + kind="video", + width=rend["width"], + height=rend["height"], + duration=duration, + creator=h.get("user", "") or "", + license=_LICENSE, + source_tags=tags, + thumbnail_url=( + h.get("userImageURL", "") + or videos.get("tiny", {}).get("thumbnail", "") + or "" + ), + extra={ + "views": h.get("views"), + "downloads": h.get("downloads"), + "rendition_size": rend.get("size"), + }, + ) + ) + return out + + def download(self, candidate: Candidate, out_path: Path) -> Path: + import requests + + 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 + + +def _pick_rendition( + videos: dict[str, Any], + min_width: int = 0, +) -> Optional[dict[str, Any]]: + """Pick the best rendition from Pixabay's nested video dict. + + Pixabay returns renditions keyed by quality tier: + large (1920), medium (1280), small (960), tiny (640). + We pick the largest that's at most 1920px wide. + """ + preference = ["large", "medium", "small", "tiny"] + for tier in preference: + rend = videos.get(tier) + if not rend or not rend.get("url"): + continue + w = int(rend.get("width") or 0) + h = int(rend.get("height") or 0) + if w >= min_width: + return {"url": rend["url"], "width": w, "height": h, "size": rend.get("size")} + return None diff --git a/tools/video/stock_sources/pond5_pd.py b/tools/video/stock_sources/pond5_pd.py new file mode 100644 index 0000000..ecee617 --- /dev/null +++ b/tools/video/stock_sources/pond5_pd.py @@ -0,0 +1,188 @@ +"""Pond5 Public Domain stock source adapter. + +Wraps Pond5's public domain collection behind the unified `StockSource` +protocol. Pond5 has curated ~10,000 public domain video clips plus +65,000+ photos and audio recordings. The collection focuses on +historical and archival material: WWI/WWII, early cinema, space +launches, historical speeches, Olympic footage. + +All public domain items are CC0-equivalent — free for any use, no +attribution required (though appreciated). + +The adapter accesses Pond5's free public domain search which does not +require an API key. For the full commercial API, a partnership agreement +is needed, but the public domain subset is openly browsable. + +What Pond5 Public Domain is good for +------------------------------------- +- Historical / archival documentary footage (WWI, WWII, Cold War) +- Early cinema (Méliès, Edison, Lumière) +- Vintage newsreels and propaganda films +- Space race and early NASA footage +- Historical speeches (JFK, Churchill, MLK) +""" +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Any + +from .base import Candidate, SearchFilters + +_log = logging.getLogger(__name__) + +_SEARCH_URL = "https://www.pond5.com/api/v2/search" +_PD_SEARCH_URL = "https://www.pond5.com/free" +_LICENSE = "Public domain (CC0 equivalent, Pond5 Public Domain Project)" + +# Pond5 public domain items are tagged with specific collection IDs +_VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".wmv", ".webm", ".mpg", ".mpeg"} + + +class Pond5PublicDomainSource: + """Pond5 Public Domain adapter. Satisfies `StockSource`.""" + + name = "pond5_pd" + display_name = "Pond5 Public Domain" + provider = "pond5" + priority = 38 + install_instructions = ( + "Pond5 Public Domain works without an API key for basic search. " + "Set POND5_API_KEY in .env for higher rate limits and full API access." + ) + supports = {"video": True, "image": True} + + def is_available(self) -> bool: + return True + + def search(self, query: str, filters: SearchFilters) -> list[Candidate]: + import requests + + kind = (filters.kind or "video").lower() + + # Pond5 public search endpoint + params: dict[str, Any] = { + "kw": query, + "page": max(1, filters.page), + "ps": max(1, min(filters.per_page, 50)), + "free": 1, # Only free/public domain items + } + + if kind == "video": + params["mt"] = "footage" + elif kind == "image": + params["mt"] = "photos" + + import os + headers: dict[str, str] = { + "User-Agent": "OpenMontage/1.0 (stock source adapter)", + } + api_key = os.environ.get("POND5_API_KEY") + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + + try: + r = requests.get( + _SEARCH_URL, + headers=headers, + params=params, + timeout=30, + ) + r.raise_for_status() + data = r.json() + except Exception as e: + _log.warning("Pond5 PD search failed (API), trying web fallback: %s", e) + return self._search_web_fallback(query, kind, filters) + + results = data.get("results", []) or data.get("items", []) or [] + return self._parse_results(results, kind, filters) + + def _parse_results( + self, results: list[dict], kind: str, filters: SearchFilters + ) -> list[Candidate]: + out: list[Candidate] = [] + for item in results: + item_id = str(item.get("id", "") or "") + if not item_id: + continue + + title = item.get("t", "") or item.get("title", "") or "" + description = item.get("desc", "") or item.get("description", "") or "" + keywords = item.get("kw", "") or item.get("keywords", "") or "" + if isinstance(keywords, list): + keywords = " ".join(keywords) + source_tags = f"{title} {description} {keywords}".strip() + + duration = float(item.get("dur", 0) or item.get("duration", 0) or 0) + if kind == "video": + if filters.min_duration and duration and duration < filters.min_duration: + continue + if filters.max_duration and duration and duration > filters.max_duration: + continue + + # Preview/download URL + preview_url = ( + item.get("v", "") + or item.get("preview_url", "") + or item.get("icon_url", "") + or "" + ) + thumb_url = item.get("ic", "") or item.get("thumbnail_url", "") or "" + + if not preview_url: + continue + + width = int(item.get("w", 0) or item.get("width", 0) or 0) + height = int(item.get("h", 0) or item.get("height", 0) or 0) + + candidate_kind = "video" if kind != "image" else "image" + source_url = f"https://www.pond5.com/stock-footage/{item_id}" + + out.append( + Candidate( + source=self.name, + source_id=item_id, + source_url=source_url, + download_url=preview_url, + kind=candidate_kind, + width=width, + height=height, + duration=duration, + creator=item.get("an", "") or item.get("artist_name", "") or "Pond5 Public Domain", + license=_LICENSE, + source_tags=source_tags, + thumbnail_url=thumb_url, + extra={ + "fps": item.get("fps"), + "codec": item.get("codec"), + }, + ) + ) + return out + + def _search_web_fallback( + self, query: str, kind: str, filters: SearchFilters + ) -> list[Candidate]: + """Fallback: parse Pond5 free page HTML for public domain clips. + + Used when the API endpoint is unavailable or returns errors. + Returns empty list if HTML parsing fails — does not raise. + """ + _log.info("Pond5 PD: web fallback not implemented, returning empty") + return [] + + def download(self, candidate: Candidate, out_path: Path) -> Path: + import requests + + out_path = Path(out_path) + out_path.parent.mkdir(parents=True, exist_ok=True) + + with requests.get( + candidate.download_url, stream=True, timeout=180 + ) 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 diff --git a/tools/video/stock_sources/videvo.py b/tools/video/stock_sources/videvo.py new file mode 100644 index 0000000..48944d7 --- /dev/null +++ b/tools/video/stock_sources/videvo.py @@ -0,0 +1,166 @@ +"""Videvo stock video source adapter. + +Wraps the Videvo API behind the unified `StockSource` protocol. Videvo +offers 90,000+ free video clips (HD and 4K) plus a larger premium +library. Free clips use either the Videvo Attribution License (credit +required) or Creative Commons 3.0 (CC BY 3.0). + +Videvo API: Announced at https://www.videvo.net/blog/announcing-the-new-api/. +Requires an API key for access. Unlimited requests claimed. + +What Videvo is good for +----------------------- +- Large free video library (90K+ clips) +- Nature, aerial, city, abstract, time-lapses +- Modern HD/4K footage +- Complements Pexels with a different contributor base +""" +from __future__ import annotations + +import logging +import os +from pathlib import Path +from typing import Any + +from .base import Candidate, SearchFilters + +_log = logging.getLogger(__name__) + +_API_URL = "https://api.videvo.net/v1/search" +_LICENSE_ATTR = "Videvo Attribution License (free, attribution required)" +_LICENSE_CC = "Creative Commons 3.0 (CC BY 3.0, attribution required)" + + +class VidevoSource: + """Videvo video adapter. Satisfies `StockSource`.""" + + name = "videvo" + display_name = "Videvo" + provider = "videvo" + priority = 22 + install_instructions = ( + "Set VIDEVO_API_KEY in .env to enable Videvo stock search " + "(get API access at https://www.videvo.net/api/)." + ) + supports = {"video": True, "image": False} + + def is_available(self) -> bool: + return bool(os.environ.get("VIDEVO_API_KEY")) + + def search(self, query: str, filters: SearchFilters) -> list[Candidate]: + import requests + + kind = (filters.kind or "video").lower() + if kind == "image": + return [] + + api_key = os.environ.get("VIDEVO_API_KEY") + if not api_key: + return [] + + headers = { + "Authorization": f"Bearer {api_key}", + "Accept": "application/json", + } + + params: dict[str, Any] = { + "query": query, + "page": max(1, filters.page), + "per_page": max(1, min(filters.per_page, 50)), + "license_type": "free", # Only free clips + } + + if filters.orientation: + params["orientation"] = filters.orientation + + try: + r = requests.get( + _API_URL, + headers=headers, + params=params, + timeout=30, + ) + r.raise_for_status() + data = r.json() + except Exception as e: + _log.warning("Videvo search failed: %s", e) + return [] + + hits = data.get("data", []) or data.get("results", []) or data.get("clips", []) or [] + out: list[Candidate] = [] + + for v in hits: + 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 + + # Get best download URL + download_url = ( + v.get("download_url", "") + or v.get("url_hd", "") + or v.get("url_sd", "") + or v.get("preview_url", "") + or "" + ) + if not download_url: + continue + + width = int(v.get("width") or 0) + height = int(v.get("height") or 0) + if filters.min_width and width and width < filters.min_width: + continue + + # Tags + title = v.get("title", "") or "" + tags = v.get("tags", "") or v.get("keywords", "") or "" + if isinstance(tags, list): + tags = " ".join(tags) + source_tags = f"{title} {tags}".strip() + + # License type + lic_type = (v.get("license_type", "") or "").lower() + lic = _LICENSE_CC if "creative commons" in lic_type or "cc" in lic_type else _LICENSE_ATTR + + clip_id = str(v.get("id", "") or "") + source_url = v.get("page_url", "") or v.get("url", "") or f"https://www.videvo.net/video/{clip_id}/" + + out.append( + Candidate( + source=self.name, + source_id=clip_id, + source_url=source_url, + download_url=download_url, + kind="video", + width=width, + height=height, + duration=duration, + creator=v.get("author", "") or v.get("contributor", "") or "", + license=lic, + source_tags=source_tags, + thumbnail_url=v.get("thumbnail_url", "") or v.get("poster_url", "") or "", + extra={ + "fps": v.get("fps"), + "resolution": v.get("resolution"), + "category": v.get("category"), + }, + ) + ) + return out + + def download(self, candidate: Candidate, out_path: Path) -> Path: + import requests + + 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