480 lines
16 KiB
Python
480 lines
16 KiB
Python
"""Pexels video downloader for gnommo shared_assets.
|
|
|
|
Configure API key in ~/.gnommo.conf:
|
|
|
|
[pexels]
|
|
api_key = YOUR_KEY_HERE
|
|
|
|
Get a free key at https://www.pexels.com/api/
|
|
"""
|
|
|
|
import configparser
|
|
import json
|
|
import re
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
def get_pexels_api_key() -> Optional[str]:
|
|
config_path = Path.home() / ".gnommo.conf"
|
|
if not config_path.exists():
|
|
return None
|
|
cfg = configparser.ConfigParser()
|
|
cfg.read(config_path)
|
|
return cfg.get("pexels", "api_key", fallback=None)
|
|
|
|
|
|
def extract_pexels_id(source_file: str) -> Optional[str]:
|
|
"""Extract the numeric Pexels video ID from a source_file path.
|
|
|
|
Handles names like 'pexels/11868263-hd_1920_1080_24fps.mp4'
|
|
and 'pexels/12136677_1080_1920_30fps.mp4'.
|
|
"""
|
|
name = Path(source_file).stem.split("/")[-1]
|
|
m = re.match(r"^(\d+)", name)
|
|
return m.group(1) if m else None
|
|
|
|
|
|
def _fetch_video_info(pexels_id: str, api_key: str) -> Optional[dict]:
|
|
url = f"https://api.pexels.com/videos/videos/{pexels_id}"
|
|
req = urllib.request.Request(
|
|
url,
|
|
headers={"Authorization": api_key, "User-Agent": "Mozilla/5.0 gnommo/1.0"},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
return json.loads(resp.read())
|
|
except urllib.error.HTTPError as e:
|
|
print(f" [{pexels_id}] Pexels API error {e.code} — video may have been deleted", flush=True)
|
|
return None
|
|
except Exception as e:
|
|
print(f" [{pexels_id}] Pexels API error: {e}", flush=True)
|
|
return None
|
|
|
|
|
|
def description_from_url(video_url: str) -> str:
|
|
"""Extract human-readable description from a Pexels video URL slug.
|
|
|
|
'https://www.pexels.com/video/abstract-television-noise-11868263/'
|
|
→ 'Abstract Television Noise'
|
|
"""
|
|
m = re.search(r"/video/([a-z0-9][a-z0-9-]+?)-\d+/?$", video_url)
|
|
if m:
|
|
return m.group(1).replace("-", " ").title()
|
|
return ""
|
|
|
|
|
|
def _pick_best_video_file(video_files: list, source_file: str) -> Optional[dict]:
|
|
"""Select the video_files entry that best matches the hints in source_file."""
|
|
stem = Path(source_file).stem.split("/")[-1]
|
|
|
|
width_hint = height_hint = fps_hint = quality_hint = None
|
|
m = re.search(r"[_-](\d{3,4})[_-](\d{3,4})[_-](\d+)fps", stem)
|
|
if m:
|
|
width_hint = int(m.group(1))
|
|
height_hint = int(m.group(2))
|
|
fps_hint = int(m.group(3))
|
|
for q in ("uhd", "hd", "sd"):
|
|
if q in stem.lower():
|
|
quality_hint = q
|
|
break
|
|
|
|
mp4s = [f for f in video_files if f.get("file_type") == "video/mp4"]
|
|
if not mp4s:
|
|
mp4s = video_files # fall back to any format
|
|
|
|
def score(vf: dict) -> int:
|
|
s = 0
|
|
if quality_hint and vf.get("quality", "").lower() == quality_hint:
|
|
s += 10
|
|
if width_hint and vf.get("width") == width_hint:
|
|
s += 5
|
|
if height_hint and vf.get("height") == height_hint:
|
|
s += 5
|
|
if fps_hint and round(float(vf.get("fps") or 0)) == fps_hint:
|
|
s += 3
|
|
return s
|
|
|
|
return max(mp4s, key=score)
|
|
|
|
|
|
def download_video(
|
|
source_file: str,
|
|
shared_assets_dir: Path,
|
|
api_key: str,
|
|
) -> Optional[dict]:
|
|
"""Download one Pexels video to shared_assets_dir/<source_file>.
|
|
|
|
Returns a metadata dict {description, duration, has_audio=False} on
|
|
success, or None on failure.
|
|
"""
|
|
pexels_id = extract_pexels_id(source_file)
|
|
if not pexels_id:
|
|
print(f" Cannot extract Pexels ID from: {source_file}", file=sys.stderr)
|
|
return None
|
|
|
|
target_path = shared_assets_dir / source_file
|
|
target_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
print(f" [{pexels_id}] Fetching video info...", flush=True)
|
|
info = _fetch_video_info(pexels_id, api_key)
|
|
if not info:
|
|
return None
|
|
|
|
description = description_from_url(info.get("url", ""))
|
|
duration = float(info.get("duration") or 0) or None
|
|
|
|
video_files = info.get("video_files", [])
|
|
if not video_files:
|
|
print(f" [{pexels_id}] No video files in API response", flush=True)
|
|
return None
|
|
|
|
best = _pick_best_video_file(video_files, source_file)
|
|
if not best:
|
|
return None
|
|
|
|
download_url = best["link"]
|
|
w, h, fps = best.get("width", "?"), best.get("height", "?"), best.get("fps", "?")
|
|
q = best.get("quality", "?")
|
|
label = f'"{description}" — ' if description else ""
|
|
print(f" [{pexels_id}] {label}{q} {w}x{h} @ {fps}fps", flush=True)
|
|
print(f" → {target_path}", flush=True)
|
|
|
|
try:
|
|
req = urllib.request.Request(
|
|
download_url, headers={"User-Agent": "Mozilla/5.0 gnommo/1.0"}
|
|
)
|
|
with urllib.request.urlopen(req, timeout=300) as resp:
|
|
total = int(resp.headers.get("Content-Length") or 0)
|
|
downloaded = 0
|
|
chunks: list[bytes] = []
|
|
chunk_size = 1024 * 512 # 512 KB
|
|
while True:
|
|
chunk = resp.read(chunk_size)
|
|
if not chunk:
|
|
break
|
|
chunks.append(chunk)
|
|
downloaded += len(chunk)
|
|
if total:
|
|
pct = downloaded * 100 // total
|
|
mb_done = downloaded / 1024 / 1024
|
|
mb_total = total / 1024 / 1024
|
|
print(f" {pct:3d}% {mb_done:.1f}/{mb_total:.1f} MB\r", end="", flush=True)
|
|
print(f" Done — {downloaded / 1024 / 1024:.1f} MB ", flush=True)
|
|
target_path.write_bytes(b"".join(chunks))
|
|
except Exception as e:
|
|
print(f"\n Download failed: {e}", flush=True)
|
|
return None
|
|
|
|
return {
|
|
"description": description,
|
|
"duration": duration,
|
|
"has_audio": False, # conservative; renderer probes when needed
|
|
}
|
|
|
|
|
|
def update_videos_json(
|
|
json_path: Path,
|
|
video_id: str,
|
|
metadata: dict,
|
|
) -> None:
|
|
"""Write description (and other metadata) into an existing videos.json entry."""
|
|
if not json_path.exists():
|
|
return
|
|
with open(json_path, "r", encoding="utf-8") as f:
|
|
raw = json.load(f)
|
|
if video_id not in raw:
|
|
return
|
|
changed = False
|
|
for key, value in metadata.items():
|
|
if value and raw[video_id].get(key) != value:
|
|
raw[video_id][key] = value
|
|
changed = True
|
|
if changed:
|
|
with open(json_path, "w", encoding="utf-8") as f:
|
|
json.dump(raw, f, indent=2, ensure_ascii=False)
|
|
|
|
|
|
def fetch_metadata(pexels_id: str, api_key: str) -> Optional[dict]:
|
|
"""Fetch only description and duration for a Pexels video (no download)."""
|
|
info = _fetch_video_info(pexels_id, api_key)
|
|
if not info:
|
|
return None
|
|
return {
|
|
"description": description_from_url(info.get("url", "")),
|
|
"duration": float(info.get("duration") or 0) or None,
|
|
}
|
|
|
|
|
|
def enrich_missing_descriptions(
|
|
shared_assets_dir: Path,
|
|
api_key: str,
|
|
) -> int:
|
|
"""Fetch descriptions from Pexels API for entries that have a file on disk but no description.
|
|
|
|
Scans shared_assets/videos.json for pexels/* entries where:
|
|
- description is absent or empty
|
|
- source_file exists on disk (locally or via cache)
|
|
|
|
Returns number of entries updated.
|
|
"""
|
|
from .cache import resolve_with_cache
|
|
|
|
videos_json = shared_assets_dir / "videos.json"
|
|
if not videos_json.exists():
|
|
return 0
|
|
|
|
with open(videos_json, "r", encoding="utf-8") as f:
|
|
raw = json.load(f)
|
|
|
|
candidates = [
|
|
(vid_id, entry)
|
|
for vid_id, entry in raw.items()
|
|
if vid_id.startswith("pexels/") and not entry.get("description")
|
|
]
|
|
|
|
# Filter to those whose file exists on disk
|
|
to_enrich = []
|
|
for vid_id, entry in candidates:
|
|
sf = entry.get("source_file", "")
|
|
if not sf:
|
|
continue
|
|
path = shared_assets_dir / sf
|
|
resolved, _ = resolve_with_cache(path, shared_assets_dir)
|
|
if resolved.exists():
|
|
pexels_id = extract_pexels_id(sf)
|
|
if pexels_id:
|
|
to_enrich.append((vid_id, pexels_id))
|
|
|
|
if not to_enrich:
|
|
return 0
|
|
|
|
print(f" Enriching descriptions for {len(to_enrich)} existing pexels video(s)...", flush=True)
|
|
|
|
updated = 0
|
|
for vid_id, pexels_id in to_enrich:
|
|
meta = fetch_metadata(pexels_id, api_key)
|
|
if meta and meta.get("description"):
|
|
print(f" [{pexels_id}] \"{meta['description']}\"", flush=True)
|
|
update_videos_json(videos_json, vid_id, meta)
|
|
updated += 1
|
|
else:
|
|
print(f" [{pexels_id}] not found or no description — skipped", flush=True)
|
|
|
|
return updated
|
|
|
|
|
|
def _search_videos(
|
|
query: str, api_key: str, per_page: int = 80, page: int = 1
|
|
) -> Optional[dict]:
|
|
"""Call the Pexels video search API and return the raw response."""
|
|
import urllib.parse
|
|
|
|
params = urllib.parse.urlencode({"query": query, "per_page": per_page, "page": page})
|
|
url = f"https://api.pexels.com/videos/search?{params}"
|
|
req = urllib.request.Request(
|
|
url,
|
|
headers={"Authorization": api_key, "User-Agent": "Mozilla/5.0 gnommo/1.0"},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
return json.loads(resp.read())
|
|
except Exception as e:
|
|
print(f" Pexels search error: {e}", flush=True)
|
|
return None
|
|
|
|
|
|
def _pick_best_quality(video_files: list) -> Optional[dict]:
|
|
"""Pick the highest-resolution MP4 from a search result's video_files list."""
|
|
mp4s = [f for f in video_files if f.get("file_type") == "video/mp4"]
|
|
if not mp4s:
|
|
mp4s = video_files
|
|
if not mp4s:
|
|
return None
|
|
return max(mp4s, key=lambda f: f.get("width", 0) * f.get("height", 0))
|
|
|
|
|
|
def _make_source_filename(pexels_id: str, video_file: dict) -> str:
|
|
"""Build a canonical filename like 12345678_1920_1080_30fps.mp4."""
|
|
w = video_file.get("width", 0)
|
|
h = video_file.get("height", 0)
|
|
fps = round(float(video_file.get("fps") or 0))
|
|
return f"{pexels_id}_{w}_{h}_{fps}fps.mp4"
|
|
|
|
|
|
def _download_bytes(url: str, target_path: Path) -> bool:
|
|
"""Stream-download url to target_path with a progress indicator. Returns True on success."""
|
|
try:
|
|
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 gnommo/1.0"})
|
|
with urllib.request.urlopen(req, timeout=300) as resp:
|
|
total = int(resp.headers.get("Content-Length") or 0)
|
|
done = 0
|
|
chunks: list[bytes] = []
|
|
while True:
|
|
chunk = resp.read(524288) # 512 KB
|
|
if not chunk:
|
|
break
|
|
chunks.append(chunk)
|
|
done += len(chunk)
|
|
if total:
|
|
pct = done * 100 // total
|
|
print(
|
|
f" {pct:3d}% {done/1048576:.1f}/{total/1048576:.1f} MB\r",
|
|
end="",
|
|
flush=True,
|
|
)
|
|
print(f" Done — {done/1048576:.1f} MB ", flush=True)
|
|
target_path.write_bytes(b"".join(chunks))
|
|
return True
|
|
except Exception as e:
|
|
print(f"\n Download failed: {e}", flush=True)
|
|
return False
|
|
|
|
|
|
def search_and_download(
|
|
query: str,
|
|
pexels_dir: Path,
|
|
shared_videos_json: Path,
|
|
api_key: str,
|
|
max_results: int = 200,
|
|
) -> tuple[int, int]:
|
|
"""Search Pexels for *query* and download all results to pexels_dir.
|
|
|
|
Each video is saved as ``pexels_dir/{pexels_id}_{w}_{h}_{fps}fps.mp4`` and
|
|
registered in *shared_videos_json* so the renderer can find it.
|
|
|
|
Returns (downloaded_count, skipped_count).
|
|
"""
|
|
print(f"Searching Pexels for '{query}'...", flush=True)
|
|
pexels_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Load existing registry so we can skip already-downloaded videos
|
|
existing: dict = {}
|
|
if shared_videos_json.exists():
|
|
with open(shared_videos_json, "r", encoding="utf-8") as f:
|
|
existing = json.load(f)
|
|
|
|
downloaded = 0
|
|
skipped = 0
|
|
page = 1
|
|
|
|
while downloaded + skipped < max_results:
|
|
per_page = min(80, max_results - downloaded - skipped)
|
|
result = _search_videos(query, api_key, per_page=per_page, page=page)
|
|
if not result:
|
|
break
|
|
|
|
videos = result.get("videos", [])
|
|
if not videos:
|
|
break
|
|
|
|
total_results = result.get("total_results", 0)
|
|
print(
|
|
f" Page {page}: {len(videos)} result(s) (Pexels total: {total_results})",
|
|
flush=True,
|
|
)
|
|
|
|
for video in videos:
|
|
pexels_id = str(video.get("id", ""))
|
|
video_files = video.get("video_files", [])
|
|
if not pexels_id or not video_files:
|
|
continue
|
|
|
|
best = _pick_best_quality(video_files)
|
|
if not best:
|
|
continue
|
|
|
|
filename = _make_source_filename(pexels_id, best)
|
|
video_id = f"pexels/{Path(filename).stem}"
|
|
target_path = pexels_dir / filename
|
|
source_file = f"pexels/{filename}"
|
|
|
|
# Skip if already registered or file already on disk
|
|
if video_id in existing or target_path.exists():
|
|
if video_id not in existing:
|
|
# File exists but not registered — register it
|
|
pass
|
|
else:
|
|
skipped += 1
|
|
continue
|
|
|
|
description = description_from_url(video.get("url", ""))
|
|
duration = float(video.get("duration") or 0) or None
|
|
w = best.get("width", "?")
|
|
h = best.get("height", "?")
|
|
fps = best.get("fps", "?")
|
|
q = best.get("quality", "?")
|
|
label = f'"{description}" — ' if description else ""
|
|
print(f" [{pexels_id}] {label}{q} {w}x{h} @ {fps}fps", flush=True)
|
|
print(f" → {target_path}", flush=True)
|
|
|
|
if not target_path.exists():
|
|
if not _download_bytes(best["link"], target_path):
|
|
continue
|
|
|
|
# Register in shared videos.json
|
|
existing[video_id] = {
|
|
"source_file": source_file,
|
|
"description": description,
|
|
"duration": duration,
|
|
"has_audio": False,
|
|
}
|
|
with open(shared_videos_json, "w", encoding="utf-8") as f:
|
|
json.dump(existing, f, indent=2, ensure_ascii=False)
|
|
|
|
downloaded += 1
|
|
|
|
if not result.get("next_page"):
|
|
break
|
|
page += 1
|
|
|
|
return downloaded, skipped
|
|
|
|
|
|
def find_missing_pexels_videos(
|
|
manuscript_markers: list[str],
|
|
videos: dict,
|
|
shared_assets_dir: Path,
|
|
) -> list[tuple[str, str]]:
|
|
"""Return [(video_id, source_file)] for pexels videos referenced but not on disk."""
|
|
from .cache import resolve_with_cache
|
|
|
|
_VIDEO_PREFIXES = (
|
|
"video:", "narration:",
|
|
"vft:", "vfb:", "vfm:",
|
|
"vf2t:", "vf2b:", "vf2m:",
|
|
"vst:", "vsb:", "vsm:",
|
|
"vftp:", "vfbp:", "vfmp:",
|
|
"vf2tp:", "vf2bp:", "vf2mp:",
|
|
"vstp:", "vsbp:", "vsmp:",
|
|
)
|
|
|
|
seen: set[str] = set()
|
|
missing: list[tuple[str, str]] = []
|
|
|
|
for marker in manuscript_markers:
|
|
prefix = next((p for p in _VIDEO_PREFIXES if marker.startswith(p)), None)
|
|
if prefix is None:
|
|
continue
|
|
video_id = marker[len(prefix):].lower()
|
|
if video_id in seen or not video_id.startswith("pexels/"):
|
|
continue
|
|
seen.add(video_id)
|
|
|
|
source_file = videos.get(video_id, None)
|
|
if source_file is None:
|
|
# Not in videos.json yet — synthesize expected path from the ID
|
|
sf = video_id + ".mp4"
|
|
else:
|
|
sf = source_file.source_file if hasattr(source_file, "source_file") else source_file
|
|
|
|
candidate = shared_assets_dir / sf
|
|
resolved, _ = resolve_with_cache(candidate, shared_assets_dir)
|
|
if not resolved.exists():
|
|
missing.append((video_id, sf))
|
|
|
|
return missing
|