sources: add 11 stock source adapters, expand catalog from 5 to 16 providers
6 API-based adapters: Pixabay Video, Coverr, NARA (U.S. National Archives), Library of Congress, Pond5 Public Domain, Videvo. 5 scraper-based adapters for sites without APIs: ESA, NOAA, Mixkit, Dareful, JAXA. All follow the StockSource protocol and are auto-discovered — no tool code changes needed. Updated scene-director, asset-director, and idea-director skills with source routing guidance so agents know which provider to use for which content type (e.g. nara for historical, noaa for ocean, esa for space). Also includes: grok_video capability updates (native audio, 1-15s duration, new aspect ratios), seedance_video tool (Seedance 2.0 via fal.ai), ProductReveal Remotion composition registration.
This commit is contained in:
calesthio
@@ -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",
|
||||
}}
|
||||
/>
|
||||
<Composition
|
||||
id="ProductReveal"
|
||||
component={ProductReveal}
|
||||
durationInFrames={30 * 8}
|
||||
fps={30}
|
||||
width={1280}
|
||||
height={720}
|
||||
defaultProps={{
|
||||
productImage: "airnothing/product.png",
|
||||
productName: "AirNothing Pro Max Ultra",
|
||||
price: "Starting at $999",
|
||||
tagline: "Nothing included.",
|
||||
closer: "Less is nothing.",
|
||||
accentColor: "#00D4FF",
|
||||
} as ProductRevealProps}
|
||||
/>
|
||||
<Composition
|
||||
id="ProductRevealVertical"
|
||||
component={ProductReveal}
|
||||
durationInFrames={30 * 8}
|
||||
fps={30}
|
||||
width={720}
|
||||
height={1280}
|
||||
defaultProps={{
|
||||
productImage: "airnothing/product.png",
|
||||
productName: "AirNothing Pro Max Ultra",
|
||||
price: "Starting at $999",
|
||||
tagline: "Nothing included.",
|
||||
closer: "Less is nothing.",
|
||||
accentColor: "#00D4FF",
|
||||
} as ProductRevealProps}
|
||||
/>
|
||||
<Composition
|
||||
id="EndTag"
|
||||
component={EndTag}
|
||||
|
||||
@@ -0,0 +1,318 @@
|
||||
import {
|
||||
AbsoluteFill,
|
||||
Img,
|
||||
interpolate,
|
||||
spring,
|
||||
useCurrentFrame,
|
||||
useVideoConfig,
|
||||
staticFile,
|
||||
Easing,
|
||||
} from "remotion";
|
||||
|
||||
export interface ProductRevealProps {
|
||||
productImage: string;
|
||||
productName: string;
|
||||
price: string;
|
||||
tagline: string;
|
||||
closer: string;
|
||||
accentColor?: string;
|
||||
}
|
||||
|
||||
export const ProductReveal: React.FC<ProductRevealProps> = ({
|
||||
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 (
|
||||
<AbsoluteFill
|
||||
style={{
|
||||
background:
|
||||
"radial-gradient(ellipse at 50% 35%, #1a1a2e 0%, #0a0a0f 60%, #000000 100%)",
|
||||
opacity: fadeOut,
|
||||
}}
|
||||
>
|
||||
{/* Ambient glow behind product */}
|
||||
<div
|
||||
style={{
|
||||
position: "absolute",
|
||||
top: "15%",
|
||||
left: "50%",
|
||||
transform: "translateX(-50%)",
|
||||
width: 400,
|
||||
height: 400,
|
||||
borderRadius: "50%",
|
||||
background: `radial-gradient(circle, ${accentColor}${Math.round(glowIntensity * 30).toString(16).padStart(2, "0")} 0%, transparent 70%)`,
|
||||
filter: "blur(50px)",
|
||||
}}
|
||||
/>
|
||||
|
||||
{/* Product image — centered in upper portion */}
|
||||
<div
|
||||
style={{
|
||||
position: "absolute",
|
||||
top: "8%",
|
||||
left: "50%",
|
||||
transform: `translateX(-50%) translateY(${floatY}px) scale(${interpolate(imgScale, [0, 1], [0.7, 1])})`,
|
||||
opacity: imgOpacity,
|
||||
}}
|
||||
>
|
||||
<div
|
||||
style={{
|
||||
width: 260,
|
||||
height: 260,
|
||||
borderRadius: 28,
|
||||
overflow: "hidden",
|
||||
boxShadow: `0 20px 60px rgba(0,0,0,0.6), 0 0 ${40 + glowIntensity * 30}px ${accentColor}22`,
|
||||
border: "1px solid rgba(255,255,255,0.08)",
|
||||
}}
|
||||
>
|
||||
<Img
|
||||
src={staticFile(productImage)}
|
||||
style={{
|
||||
width: "100%",
|
||||
height: "100%",
|
||||
objectFit: "cover",
|
||||
}}
|
||||
/>
|
||||
</div>
|
||||
{/* Reflection */}
|
||||
<div
|
||||
style={{
|
||||
width: 260,
|
||||
height: 60,
|
||||
borderRadius: "0 0 28px 28px",
|
||||
overflow: "hidden",
|
||||
marginTop: 4,
|
||||
opacity: 0.12,
|
||||
transform: "scaleY(-1)",
|
||||
filter: "blur(6px)",
|
||||
maskImage:
|
||||
"linear-gradient(to bottom, rgba(0,0,0,0.5) 0%, transparent 100%)",
|
||||
WebkitMaskImage:
|
||||
"linear-gradient(to bottom, rgba(0,0,0,0.5) 0%, transparent 100%)",
|
||||
}}
|
||||
>
|
||||
<Img
|
||||
src={staticFile(productImage)}
|
||||
style={{
|
||||
width: "100%",
|
||||
height: 260,
|
||||
objectFit: "cover",
|
||||
objectPosition: "bottom",
|
||||
}}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{/* Text content — stacked vertically, centered */}
|
||||
<div
|
||||
style={{
|
||||
position: "absolute",
|
||||
top: "56%",
|
||||
left: 0,
|
||||
right: 0,
|
||||
display: "flex",
|
||||
flexDirection: "column",
|
||||
alignItems: "center",
|
||||
gap: 0,
|
||||
}}
|
||||
>
|
||||
{/* Product name — letter by letter spring */}
|
||||
<div
|
||||
style={{
|
||||
display: "flex",
|
||||
justifyContent: "center",
|
||||
flexWrap: "wrap",
|
||||
gap: 0,
|
||||
marginBottom: 20,
|
||||
}}
|
||||
>
|
||||
{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 (
|
||||
<span
|
||||
key={i}
|
||||
style={{
|
||||
display: "inline-block",
|
||||
fontFamily:
|
||||
"'SF Pro Display', 'Helvetica Neue', 'Inter', system-ui, sans-serif",
|
||||
fontSize: 52,
|
||||
fontWeight: 600,
|
||||
letterSpacing: "0.02em",
|
||||
color: isAccent ? accentColor : "#FFFFFF",
|
||||
opacity: charSpring,
|
||||
transform: `translateY(${interpolate(charSpring, [0, 1], [20, 0])}px)`,
|
||||
whiteSpace: char === " " ? "pre" : undefined,
|
||||
minWidth: char === " " ? "0.3em" : undefined,
|
||||
}}
|
||||
>
|
||||
{char}
|
||||
</span>
|
||||
);
|
||||
})}
|
||||
</div>
|
||||
|
||||
{/* Price — block element, clearly on its own line */}
|
||||
<div
|
||||
style={{
|
||||
position: "relative",
|
||||
overflow: "hidden",
|
||||
marginBottom: 10,
|
||||
}}
|
||||
>
|
||||
<div
|
||||
style={{
|
||||
fontFamily:
|
||||
"'SF Pro Display', 'Helvetica Neue', 'Inter', system-ui, sans-serif",
|
||||
fontSize: 36,
|
||||
fontWeight: 300,
|
||||
color: "#FFFFFF",
|
||||
opacity: priceSpring,
|
||||
transform: `translateY(${interpolate(priceSpring, [0, 1], [15, 0])}px)`,
|
||||
letterSpacing: "0.05em",
|
||||
textAlign: "center",
|
||||
}}
|
||||
>
|
||||
{price}
|
||||
</div>
|
||||
{/* Shimmer across price text */}
|
||||
{frame >= shimmerDelay && frame <= shimmerDelay + fps * 1.0 && (
|
||||
<div
|
||||
style={{
|
||||
position: "absolute",
|
||||
left: shimmerPos,
|
||||
top: 0,
|
||||
width: 80,
|
||||
height: "100%",
|
||||
background:
|
||||
"linear-gradient(90deg, transparent 0%, rgba(255,255,255,0.4) 50%, transparent 100%)",
|
||||
filter: "blur(3px)",
|
||||
pointerEvents: "none",
|
||||
}}
|
||||
/>
|
||||
)}
|
||||
</div>
|
||||
|
||||
{/* Tagline */}
|
||||
<div
|
||||
style={{
|
||||
opacity: taglineOpacity,
|
||||
fontFamily:
|
||||
"'SF Pro Display', 'Helvetica Neue', 'Inter', system-ui, sans-serif",
|
||||
fontSize: 20,
|
||||
fontWeight: 300,
|
||||
color: "#777777",
|
||||
letterSpacing: "0.08em",
|
||||
marginBottom: 28,
|
||||
}}
|
||||
>
|
||||
{tagline}
|
||||
</div>
|
||||
|
||||
{/* Closer */}
|
||||
<div
|
||||
style={{
|
||||
opacity: closerOpacity,
|
||||
fontFamily:
|
||||
"'SF Pro Display', 'Helvetica Neue', 'Inter', system-ui, sans-serif",
|
||||
fontSize: 26,
|
||||
fontWeight: 500,
|
||||
color: "#BBBBBB",
|
||||
letterSpacing: "0.15em",
|
||||
textTransform: "uppercase",
|
||||
}}
|
||||
>
|
||||
{closer}
|
||||
</div>
|
||||
</div>
|
||||
</AbsoluteFill>
|
||||
);
|
||||
};
|
||||
@@ -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,
|
||||
|
||||
@@ -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": {
|
||||
|
||||
@@ -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 |
|
||||
|
||||
|
||||
@@ -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": {
|
||||
|
||||
@@ -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}",
|
||||
)
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user