#!/usr/bin/env python3 """Search/download Jamendo Creative Commons tracks and mix them under a video. Strict licensing posture for AutoShorts review assets: - reads credentials from ~/.hermes/secrets/jamendo_api - uses Jamendo official API only - defaults to Creative Commons BY tracks without NC/ND/SA restrictions where possible - stores per-track attribution/license metadata next to the downloaded file - mixes quietly with ffmpeg sidechain ducking under the existing voice """ from __future__ import annotations import argparse import html import json import re import subprocess from pathlib import Path from typing import Any import requests SECRET_PATH = Path.home() / ".hermes" / "secrets" / "jamendo_api" API = "https://api.jamendo.com/v3.0/tracks/" DEFAULT_OUT_DIR = Path("data/audio_beds/jamendo") SAFE_LICENSE_PREFIXES = ( "http://creativecommons.org/licenses/by/", "https://creativecommons.org/licenses/by/", ) BLOCKED_LICENSE_PARTS = ("/by-nc", "/by-nd", "/by-sa") def load_credentials() -> dict[str, str]: vals: dict[str, str] = {} for line in SECRET_PATH.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) vals[k.strip()] = v.strip().strip('"').strip("'") if not vals.get("JAMENDO_CLIENT_ID"): raise SystemExit(f"Missing JAMENDO_CLIENT_ID in {SECRET_PATH}") return vals def license_is_safe_for_default(ccurl: str | None) -> bool: if not ccurl: return False low = ccurl.lower() if any(part in low for part in BLOCKED_LICENSE_PARTS): return False return low.startswith(SAFE_LICENSE_PREFIXES) def api_search( query: str, *, limit: int, tags: str | None, allow_nc: bool, allow_nd: bool, allow_sa: bool, vocalinstrumental: str, order: str, ) -> list[dict[str, Any]]: creds = load_credentials() params: dict[str, Any] = { "client_id": creds["JAMENDO_CLIENT_ID"], "format": "json", "limit": limit, "include": "licenses musicinfo stats", "audioformat": "mp32", "order": order, "audiodownload_allowed": "true", } if tags: params["tags"] = tags else: params["search"] = query if vocalinstrumental: params["vocalinstrumental"] = vocalinstrumental if not allow_nc: params["ccnc"] = "false" if not allow_nd: params["ccnd"] = "false" if not allow_sa: params["ccsa"] = "false" r = requests.get(API, params=params, timeout=30) if not r.ok: raise SystemExit(f"Jamendo API failed HTTP {r.status_code}: {r.text[:300]}") data = r.json() headers = data.get("headers", {}) if headers.get("status") != "success": raise SystemExit(f"Jamendo API status failed: {headers}") hits = data.get("results", []) # Enforce locally as well; API params are helpful but license safety should not be trusted blindly. filtered = [] for h in hits: ccurl = h.get("license_ccurl") if not h.get("audiodownload_allowed", True): continue if not allow_nc and ccurl and "/by-nc" in ccurl.lower(): continue if not allow_nd and ccurl and "/by-nd" in ccurl.lower(): continue if not allow_sa and ccurl and "/by-sa" in ccurl.lower(): continue if not allow_nc and not allow_nd and not allow_sa and not license_is_safe_for_default(ccurl): continue filtered.append(h) return filtered def summarize(hit: dict[str, Any]) -> dict[str, Any]: return { "id": hit.get("id"), "name": html.unescape(str(hit.get("name") or "")), "artist_name": html.unescape(str(hit.get("artist_name") or "")), "duration": hit.get("duration"), "license_ccurl": hit.get("license_ccurl"), "shareurl": hit.get("shareurl"), "audiodownload_allowed": hit.get("audiodownload_allowed"), } def safe_name(text: str) -> str: text = html.unescape(text) text = re.sub(r"[^a-zA-Z0-9_.-]+", "_", text).strip("_") return text[:90] or "jamendo_track" def download(hit: dict[str, Any], out_dir: Path) -> tuple[Path, Path]: url = hit.get("audiodownload") or hit.get("audio") if not isinstance(url, str) or not url.startswith("http"): raise SystemExit("Selected Jamendo hit has no audiodownload/audio URL.") out_dir.mkdir(parents=True, exist_ok=True) base = f"{hit.get('id')}_{safe_name(hit.get('artist_name','artist'))}_{safe_name(hit.get('name','track'))}" audio_path = out_dir / f"{base}.mp3" meta_path = out_dir / f"{base}.license.json" with requests.get(url, timeout=90, stream=True, headers={"User-Agent": "AutoShortsBot/1.0"}) as r: if not r.ok: raise SystemExit(f"Jamendo download failed HTTP {r.status_code}: {r.text[:300]}") with audio_path.open("wb") as f: for chunk in r.iter_content(1024 * 256): if chunk: f.write(chunk) meta = { "source": "Jamendo", "track": summarize(hit), "attribution": f"{html.unescape(str(hit.get('name') or 'Untitled'))} by {html.unescape(str(hit.get('artist_name') or 'Unknown artist'))} — {hit.get('license_ccurl') or 'license unknown'} — {hit.get('shareurl') or ''}", "downloaded_file": str(audio_path), "license_checked_default_safe": license_is_safe_for_default(hit.get("license_ccurl")), } meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") return audio_path, meta_path def ffprobe_duration(path: Path) -> float: return float( subprocess.check_output( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nk=1", str(path)], text=True, ).strip() ) def mix(video: Path, music: Path, output: Path, volume: float) -> None: output.parent.mkdir(parents=True, exist_ok=True) cmd = [ "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-i", str(video), "-stream_loop", "-1", "-i", str(music), "-filter_complex", ( f"[1:a]volume={volume},afade=t=in:st=0:d=1.2[m0];" "[m0][0:a]sidechaincompress=threshold=0.012:ratio=10:attack=20:release=450:makeup=1[mduck];" "[0:a][mduck]amix=inputs=2:duration=first:dropout_transition=0,alimiter=limit=0.95[a]" ), "-map", "0:v:0", "-map", "[a]", "-c:v", "copy", "-c:a", "aac", "-b:a", "192k", "-movflags", "+faststart", str(output), ] subprocess.check_call(cmd) def main() -> int: p = argparse.ArgumentParser() p.add_argument("--query", default="dark suspense background") p.add_argument("--tags", help="Jamendo tags query, e.g. cinematic,ambient. Overrides --query.") p.add_argument("--limit", type=int, default=10) p.add_argument("--order", default="popularity_total") p.add_argument("--vocalinstrumental", default="instrumental") p.add_argument("--allow-nc", action="store_true", help="Allow Creative Commons NonCommercial tracks. Default blocks NC.") p.add_argument("--allow-nd", action="store_true", help="Allow Creative Commons NoDerivatives tracks. Default blocks ND.") p.add_argument("--allow-sa", action="store_true", help="Allow Creative Commons ShareAlike tracks. Default blocks SA for simpler platform attribution.") p.add_argument("--pick", type=int, default=0, help="0-based hit index after filtering") p.add_argument("--video", type=Path) p.add_argument("--output", type=Path) p.add_argument("--volume", type=float, default=0.13) p.add_argument("--out-dir", type=Path, default=DEFAULT_OUT_DIR) args = p.parse_args() hits = api_search( args.query, limit=args.limit, tags=args.tags, allow_nc=args.allow_nc, allow_nd=args.allow_nd, allow_sa=args.allow_sa, vocalinstrumental=args.vocalinstrumental, order=args.order, ) summary = [summarize(h) for h in hits] print(json.dumps({"query": args.query, "tags": args.tags, "hits": summary}, ensure_ascii=False, indent=2)) if not args.video: return 0 if not hits: raise SystemExit("No Jamendo hits after license/download filtering.") if args.pick < 0 or args.pick >= len(hits): raise SystemExit(f"--pick {args.pick} out of range for {len(hits)} hits") hit = hits[args.pick] track, meta = download(hit, args.out_dir) output = args.output or args.video.with_name(args.video.stem + "_jamendo_music.mp4") mix(args.video, track, output, args.volume) print( json.dumps( { "selected": summarize(hit), "track_path": str(track), "license_metadata": str(meta), "mixed_output": str(output), "mixed_duration": ffprobe_duration(output), }, ensure_ascii=False, indent=2, ) ) return 0 if __name__ == "__main__": raise SystemExit(main())