From d8521aef40f8831543777289353cd6dcd597e335 Mon Sep 17 00:00:00 2001 From: Alexander Bell Date: Fri, 5 Sep 2025 18:39:29 +0200 Subject: [PATCH] initial commit --- main.py | 247 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 247 insertions(+) create mode 100644 main.py diff --git a/main.py b/main.py new file mode 100644 index 0000000..25553ea --- /dev/null +++ b/main.py @@ -0,0 +1,247 @@ +#!/usr/bin/env python3 +import argparse +import os +import re +import shlex +import subprocess +import sys +from pathlib import Path +from typing import Iterable, List, Optional, Tuple + +VIDEO_EXTS = {".mkv", ".mp4", ".m4v", ".avi", ".ts", ".mov", ".wmv", ".webm"} + + +def log(msg: str) -> None: + print(msg, flush=True) + + +def normalize_title(name: str) -> str: + name = name.replace("_", " ").replace(".", " ") + return re.sub(r"\s+", " ", name).strip() + + +def parse_title_year_from_folder(folder_name: str) -> Tuple[str, Optional[str]]: + m = re.match(r"^(?P.+)\s\((?P<year>\d{4})\)$", folder_name) + if m: + return normalize_title(m.group("title")), m.group("year") + return normalize_title(folder_name), None + + +def has_video_file(path: Path) -> bool: + for p in path.iterdir(): + if p.is_file() and p.suffix.lower() in VIDEO_EXTS: + return True + return False + + +def build_queries(title: str, year: Optional[str], variants: List[str]) -> List[str]: + queries: List[str] = [] + for v in variants: + if year: + queries.append(f"{title} ({year}) {v}") + queries.append(f"{title} {v}") + seen, deduped = set(), [] + for q in queries: + if q not in seen: + seen.add(q) + deduped.append(q) + return deduped + + +def run(cmd: str, timeout: Optional[int] = None) -> Tuple[int, str, str]: + proc = subprocess.Popen( + cmd, + shell=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True + ) + try: + out, err = proc.communicate(timeout=timeout) + except subprocess.TimeoutExpired: + proc.kill() + out, err = proc.communicate() + return 124, out, err + return proc.returncode, out, err + + +def yt_dlp_find_first_id(query: str, min_sec: int, max_sec: int, yt_dlp_path: str, verbose: bool) -> Optional[str]: + match_filter = f'duration >= {min_sec} & duration <= {max_sec}' + cmd = f'{shlex.quote(yt_dlp_path)} --quiet --no-warnings --match-filter {shlex.quote(match_filter)} --get-id {shlex.quote("ytsearch1:" + query)}' + if verbose: + log(f" [search] {cmd}") + rc, out, err = run(cmd, timeout=60) + if verbose and err.strip(): + log(f" [stderr] {err.strip()}") + vid = out.strip() + return vid if (rc == 0 and vid) else None + + +def format_command(template: str, url: str, out_path: Path) -> str: + return template.replace("{URL}", url).replace("{OUT}", str(out_path)) + + +def process_movie_dir( + movie_dir: Path, + *, + theme_filename: str, + query_variants: List[str], + min_sec: int, + max_sec: int, + yt_dlp_path: str, + dl_cmd_template: str, + dry_run: bool, + verbose: bool, + retries: int +) -> None: + out_mp3 = movie_dir / theme_filename + + # ✅ Skip if theme already exists + if out_mp3.exists(): + log(f"✔ Skipping (already has {theme_filename}): {movie_dir}") + return + + if not has_video_file(movie_dir): + if verbose: + log(f" (no video files) Skipping directory: {movie_dir}") + return + + title, year = parse_title_year_from_folder(movie_dir.name) + log(f"→ Searching theme for: '{title}'{f' ({year})' if year else ''}") + + queries = build_queries(title, year, query_variants) + found_id: Optional[str] = None + + for idx, q in enumerate(queries, start=1): + if verbose: + log(f" [{idx}/{len(queries)}] Query: {q}") + vid_id = yt_dlp_find_first_id(q, min_sec, max_sec, yt_dlp_path, verbose) + if vid_id: + found_id = vid_id + log(f" Found candidate: {found_id} for query '{q}'") + break + else: + if verbose: + log(" No match in duration window.") + + if not found_id: + log(f"✖ No suitable theme found for: '{title}'{f' ({year})' if year else ''}") + return + + url = f"https://www.youtube.com/watch?v={found_id}" + final_cmd = format_command(dl_cmd_template, url, out_mp3) + log(f"↓ Downloading: {url}") + log(f" → to: {out_mp3}") + if verbose: + log(f" [exec] {final_cmd}") + + if dry_run: + log(" [dry-run] Skipping execution.") + return + + out_mp3.parent.mkdir(parents=True, exist_ok=True) + + attempt = 0 + while attempt <= retries: + attempt += 1 + rc, out, err = run(final_cmd) + if verbose and out.strip(): + log(f" [stdout] {out.strip()}") + if rc == 0 and out_mp3.exists() and out_mp3.stat().st_size > 0: + log(f"✔ Done: {out_mp3}") + return + else: + log(f" Attempt {attempt} failed (rc={rc}).") + if err.strip(): + log(f" [stderr] {err.strip()}") + if attempt <= retries: + log(" Retrying...") + + log(f"✖ Download failed after {retries+1} attempts: {movie_dir.name}") + + +def iter_movie_dirs(root: Path, max_depth: int) -> Iterable[Path]: + queue: List[Tuple[Path, int]] = [(root, 0)] + while queue: + cur, depth = queue.pop(0) + if depth > max_depth: + continue + try: + for p in cur.iterdir(): + if p.is_dir(): + if depth >= 1: + yield p + queue.append((p, depth + 1)) + except PermissionError: + log(f" [warn] Permission denied: {cur}") + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Fetch theme songs for Plex movie folders by searching YouTube and downloading as theme.mp3." + ) + parser.add_argument("library_root", type=str, help="Path to Plex Movies root directory.") + parser.add_argument( + "--theme-filename", type=str, default="theme.mp3", + help="File name to save in each movie folder (default: theme.mp3)." + ) + parser.add_argument( + "--dl-cmd", type=str, + default='yt-dlp -x --audio-format mp3 --audio-quality 0 -o "{OUT}" "{URL}"', + help="Download command template with {URL} and {OUT} placeholders." + ) + parser.add_argument("--min-sec", type=int, default=45, help="Minimum duration to accept (default: 45).") + parser.add_argument("--max-sec", type=int, default=240, help="Maximum duration to accept (default: 240).") + parser.add_argument("--yt-dlp-path", type=str, default="yt-dlp", help="Path to yt-dlp binary.") + parser.add_argument("--max-depth", type=int, default=2, help="Max folder depth to scan (default: 2).") + parser.add_argument("--dry-run", action="store_true", help="Preview actions without downloading.") + parser.add_argument("--retries", type=int, default=1, help="Retries on failure (default: 1).") + parser.add_argument("--quiet", action="store_true", help="Less verbose output.") + + args = parser.parse_args() + + library_root = Path(args.library_root).expanduser().resolve() + if not library_root.is_dir(): + log(f"Error: library_root is not a directory: {library_root}") + sys.exit(2) + + query_variants = [ + "main theme", + "theme song", + "soundtrack main theme", + "opening theme", + "OST main theme", + ] + + verbose = not args.quiet + + log("===== Plex Theme Fetcher =====") + log(f"Library root : {library_root}") + log(f"Theme filename : {args.theme_filename}") + log(f"Duration range : {args.min_sec}–{args.max_sec} sec") + log(f"yt-dlp path : {args.yt_dlp_path}") + log(f"Dry run : {'yes' if args.dry_run else 'no'}") + log("=============================\n") + + rc, _, _ = run(f'{shlex.quote(args.yt_dlp_path)} --version', timeout=10) + if rc != 0: + log("Error: yt-dlp not found or not executable.") + sys.exit(3) + + for movie_dir in iter_movie_dirs(library_root, args.max_depth): + process_movie_dir( + movie_dir, + theme_filename=args.theme_filename, + query_variants=query_variants, + min_sec=args.min_sec, + max_sec=args.max_sec, + yt_dlp_path=args.yt_dlp_path, + dl_cmd_template=args.dl_cmd, + dry_run=args.dry_run, + verbose=verbose, + retries=args.retries + ) + + +if __name__ == "__main__": + main()