248 lines
7.1 KiB
Python
248 lines
7.1 KiB
Python
#!/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<title>.+)\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()
|