commit 658d1986593ba23839fc1c7dae8663afa904b8b8 Author: Leo Vasanko Date: Tue Feb 3 19:43:14 2026 +0000 Initial commit of hivescan. diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..fae0205 --- /dev/null +++ b/.gitignore @@ -0,0 +1,9 @@ +__pycache__/ +*.py[oc] +build/ +dist/ +wheels/ +*.egg-info +*.lock +.* +!.gitignore diff --git a/.python-version b/.python-version new file mode 100644 index 0000000..6324d40 --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.14 diff --git a/README.md b/README.md new file mode 100644 index 0000000..8f3f617 --- /dev/null +++ b/README.md @@ -0,0 +1,75 @@ +# Torrent Manager + +Tools for managing a media torrent library: scanning, indexing, metadata fetching, and preview generation. + +## Project Structure + +``` +hivescan/ Indexing & previews (installable package) + indexer.py Media index generation + scanning.py File system scanning + parsing.py Torrent name parsing (PTN) + models.py Data models + images.py TMDb cover/backdrop downloading + showreel.py Video preview clip generation (ffmpeg) + tmdb_client.py TMDb API client with caching + utils.py Path, size, and timestamp helpers +scripts/ + rtorrent-manager.py Torrent scanning & rtorrent management +rtorrent_client.py RTorrent XMLRPC/SCGI client +``` + +## Hivescan + +Scans downloaded content, categorizes it (Movies, Series, Other), fetches metadata from TMDb, generates preview clips, and produces a JSON index. + +```bash +# Scan downloads, auto-detect common root, create .mediahive folder +hivescan /media/torrents/* + +# Scan multiple locations +hivescan /mnt/disk1/* /mnt/disk2/* + +# Override output directory +hivescan /media/torrents/* -o /srv/media/.mediahive + +# Skip cover/showreel generation +hivescan /media/torrents/* --no-covers --no-showreels +``` + +## RTorrent Manager + +Scans `.torrent` files, filters by tracker, verifies downloads exist on disk, loads verified torrents into rtorrent, and cleans up unregistered torrents. + +```bash +# Scan .torrents directories and manage rtorrent +python scripts/rtorrent-manager.py /media/torrents*/.torrents/ + +# Multiple paths +python scripts/rtorrent-manager.py /mnt/disk1/torrents/.torrents/ /mnt/disk2/torrents/.torrents/ + +# Filter by tracker, dry run +python scripts/rtorrent-manager.py /media/torrents*/.torrents/ --tracker example.org --dry +``` + +## Path Mapping + +Hivescan auto-detects the common root of scanned paths and creates a `.mediahive` folder there. All paths in the index are stored relative to that root. + +| Location | Example | +|----------|---------| +| Torrents | `/media/torrents*/` | +| Index | `/media/.mediahive/index.json` | +| Covers | `/media/.mediahive/movies/` | +| Showreels | `/media/.mediahive/movies//reel1.webm` | + +Use `-o` to override the output directory if the auto-detected root isn't suitable. + +## Requirements + +- Python ≥ 3.14 +- ffmpeg (for showreel generation) + +```bash +pip install -e . +``` diff --git a/hivescan/__init__.py b/hivescan/__init__.py new file mode 100644 index 0000000..522228a --- /dev/null +++ b/hivescan/__init__.py @@ -0,0 +1,52 @@ +""" +Hivescan - Scans downloaded torrent directories and generates a media index. + +Usage: + hivescan [path] [options] + +Or as a library: + from hivescan import scan_downloads, generate_media_index +""" + +from hivescan.models import ContentType, ContentHash, ParsedContent +from hivescan.scanning import scan_downloads, categorize_downloads, find_playable_file, find_episode_files +from hivescan.indexer import generate_media_index +from hivescan.utils import DEFAULT_OUTPUT_FOLDER, find_common_root +from hivescan.showreel import generate_showreel_images, generate_episode_reel +from hivescan.tmdb_client import ( + TMDbInfo, + TMDbSeasonInfo, + TMDbEpisodeInfo, + fetch_movie_info, + fetch_series_info, + fetch_season_details, + set_cache_dir, +) + +__all__ = [ + # Models + "ContentType", + "ContentHash", + "ParsedContent", + # Scanning + "scan_downloads", + "categorize_downloads", + "find_playable_file", + "find_episode_files", + # Index generation + "generate_media_index", + # Showreel generation + "generate_showreel_images", + "generate_episode_reel", + # TMDb client + "TMDbInfo", + "TMDbSeasonInfo", + "TMDbEpisodeInfo", + "fetch_movie_info", + "fetch_series_info", + "fetch_season_details", + "set_cache_dir", + # Utilities + "DEFAULT_OUTPUT_FOLDER", + "find_common_root", +] diff --git a/hivescan/__main__.py b/hivescan/__main__.py new file mode 100644 index 0000000..0837fec --- /dev/null +++ b/hivescan/__main__.py @@ -0,0 +1,121 @@ +"""CLI entry point for hivescan.""" + +import argparse +import glob +import sys +from pathlib import Path + +from hivescan.scanning import scan_downloads, categorize_downloads +from hivescan.indexer import generate_media_index +from hivescan.utils import DEFAULT_OUTPUT_FOLDER, find_common_root +from hivescan.tmdb_client import set_cache_dir + + +def main(): + parser = argparse.ArgumentParser( + description="Scan downloaded torrents and generate a media index.", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + %(prog)s /path/to/torrents/* # Scan paths, auto-detect common root + %(prog)s /mnt/disk1/* /mnt/disk2/* # Scan multiple locations + %(prog)s /torrents/* -o /srv/media # Override output directory + %(prog)s /torrents/* --no-showreels # Skip showreel generation + %(prog)s /torrents/* --no-covers # Skip cover/backdrop downloads + +Output: + By default, creates a .mediahive folder at the common root of scanned paths. + All paths in the index are stored relative to the .mediahive parent folder. + Use -o/--output-dir to override the output location. + """, + ) + + parser.add_argument( + "paths", + nargs="+", + help="Folders or glob patterns to scan for downloads", + ) + parser.add_argument( + "-o", "--output-dir", + metavar="DIR", + help=f"Output directory for index and covers (default: {DEFAULT_OUTPUT_FOLDER} at common root)", + ) + parser.add_argument( + "--no-showreels", + action="store_true", + help="Skip generating showreel images", + ) + parser.add_argument( + "--no-covers", + action="store_true", + help="Skip downloading cover and backdrop images from TMDb", + ) + + args = parser.parse_args() + + # Expand glob patterns and collect all paths + all_paths = [] + for pattern in args.paths: + expanded = glob.glob(pattern) + if expanded: + all_paths.extend(Path(p) for p in expanded) + else: + # Treat as literal path if no glob match + all_paths.append(Path(pattern)) + + if not all_paths: + print("Error: No paths found to scan", file=sys.stderr) + sys.exit(1) + + # Determine output directory + if args.output_dir: + output_dir = Path(args.output_dir) + media_root = output_dir.parent + else: + # Find common root of all scan paths + media_root = find_common_root(all_paths) + if media_root is None: + print("Error: Cannot determine common root for paths (different drives?)", file=sys.stderr) + print(" Use -o/--output-dir to specify output location", file=sys.stderr) + sys.exit(1) + output_dir = media_root / DEFAULT_OUTPUT_FOLDER + + output_dir.mkdir(parents=True, exist_ok=True) + index_path = output_dir / "index.json" + + # Set TMDb cache directory within output dir + set_cache_dir(output_dir / ".tmdb-cache") + + print(f"Media root: {media_root}") + print(f"Output dir: {output_dir}") + print(f"Scanning {len(all_paths)} paths...") + + # Scan all paths + downloads = [] + for path in all_paths: + if path.is_dir(): + # Scan directory contents + for item in path.iterdir(): + if not item.name.startswith("."): + from hivescan.parsing import parse_download + downloads.append(parse_download(item)) + elif path.exists(): + from hivescan.parsing import parse_download + downloads.append(parse_download(path)) + + print(f"Found {len(downloads)} items") + + categories = categorize_downloads(downloads) + + generate_media_index( + categories, + index_path, + output_dir, + media_root=media_root, + fetch_covers=not args.no_covers, + generate_showreels=not args.no_showreels, + ) + + +if __name__ == "__main__": + main() diff --git a/hivescan/images.py b/hivescan/images.py new file mode 100644 index 0000000..f528ad5 --- /dev/null +++ b/hivescan/images.py @@ -0,0 +1,95 @@ +"""TMDb image downloading functions.""" + +import urllib.request +from pathlib import Path +from typing import Optional + +from hivescan.utils import get_media_folder_path + + +# TMDb image configuration +TMDB_IMAGE_BASE = "https://image.tmdb.org/t/p" +DEFAULT_POSTER_SIZE = "w500" +DEFAULT_BACKDROP_SIZE = "w1280" + + +def _download_image(url: str, output_path: Path, description: str) -> Optional[str]: + """Download an image from URL to output path.""" + if output_path.exists(): + return str(output_path) + + try: + req = urllib.request.Request(url, headers={"User-Agent": "TorrentManager/1.0"}) + with urllib.request.urlopen(req, timeout=30) as response: + output_path.parent.mkdir(parents=True, exist_ok=True) + with open(output_path, "wb") as f: + f.write(response.read()) + return str(output_path) + except Exception as e: + print(f" Failed to download {description}: {e}") + return None + + +def download_cover_image( + poster_path: str, + title: str, + year: Optional[int], + media_type: str, + cover_dir: Path, + size: str = DEFAULT_POSTER_SIZE, +) -> Optional[str]: + """Download a cover image from TMDb.""" + if not poster_path: + return None + + media_folder = get_media_folder_path(title, year, media_type, cover_dir) + cover_path = media_folder / "cover.jpg" + + if cover_path.exists(): + return str(cover_path) + + url = f"{TMDB_IMAGE_BASE}/{size}{poster_path}" + print(f" Downloading cover: {title}") + return _download_image(url, cover_path, f"cover for {title}") + + +def download_backdrop_image( + backdrop_path: str, + title: str, + year: Optional[int], + media_type: str, + cover_dir: Path, + size: str = DEFAULT_BACKDROP_SIZE, +) -> Optional[str]: + """Download a backdrop image from TMDb.""" + if not backdrop_path: + return None + + media_folder = get_media_folder_path(title, year, media_type, cover_dir) + local_path = media_folder / "backdrop.jpg" + + if local_path.exists(): + return str(local_path) + + url = f"{TMDB_IMAGE_BASE}/{size}{backdrop_path}" + print(f" Downloading backdrop: {title}") + return _download_image(url, local_path, f"backdrop for {title}") + + +def download_season_poster( + poster_path: str, + media_folder: Path, + season_num: int, +) -> Optional[str]: + """Download a season poster image from TMDb.""" + if not poster_path: + return None + + output_path = media_folder / f"season{season_num:02d}.jpg" + + if output_path.exists(): + return str(output_path) + + media_folder.mkdir(parents=True, exist_ok=True) + url = f"{TMDB_IMAGE_BASE}/{DEFAULT_POSTER_SIZE}{poster_path}" + return _download_image(url, output_path, f"season {season_num} poster") diff --git a/hivescan/indexer.py b/hivescan/indexer.py new file mode 100644 index 0000000..fbafe0c --- /dev/null +++ b/hivescan/indexer.py @@ -0,0 +1,698 @@ +"""Media index generation.""" + +import hashlib +import json +from datetime import datetime +from pathlib import Path +from typing import Dict, List, Optional, Tuple + +from tqdm import tqdm + +from hivescan.showreel import ( + episode_reel_exists, + generate_episode_reel, + generate_showreel_images, + get_expected_episode_reel_path, + get_expected_showreel_paths, + movie_showreels_exist, +) +from hivescan.tmdb_client import ( + TMDbInfo, + TMDbSeasonInfo, + TMDbEpisodeInfo, + fetch_movie_info, + fetch_series_info, + fetch_season_details, +) + +from hivescan.models import ContentType, ParsedContent +from hivescan.scanning import find_cover_image, find_episode_files, find_playable_file +from hivescan.images import download_cover_image, download_backdrop_image, download_season_poster +from hivescan.utils import ( + get_added_timestamp, + get_media_folder_path, + make_relative_path, + sort_by_quality, +) + + +def _build_version_info(item: ParsedContent, media_root: Optional[str] = None) -> dict: + """Build version/release info dict for a single torrent.""" + playable_file = find_playable_file(item.path) + size = item.content_hash.size if item.content_hash else None + newest = get_added_timestamp(item.path) + + return { + "path": make_relative_path(str(item.path), media_root), + "playable_file": make_relative_path(playable_file, media_root), + "resolution": item.resolution, + "quality": item.quality, + "codec": item.codec, + "audio": item.audio, + "encoder": item.encoder, + "size": size, + "newest": newest, + } + + +def _collect_episode_files(items: List[ParsedContent]) -> Dict[Tuple[int, int], List[Dict]]: + """ + Collect all episode files from a list of torrent items. + + Returns dict mapping (season, episode) to list of file info dicts. + """ + all_episode_files: Dict[Tuple[int, int], List[Dict]] = {} + + for item in items: + episode_files = find_episode_files(item.path) + + for (season_num, episode_num), files in episode_files.items(): + key = (season_num, episode_num) + if key not in all_episode_files: + all_episode_files[key] = [] + for file_path, file_size in files: + all_episode_files[key].append({ + "path": file_path, + "size": file_size, + "resolution": item.resolution, + "quality": item.quality, + "codec": item.codec, + "audio": item.audio, + "encoder": item.encoder, + "torrent_path": str(item.path), + }) + + # Handle individual episodes from PTN parsing + if item.episode is not None and item.season is not None: + season_nums = item.season if isinstance(item.season, list) else [item.season] + episode_nums = item.episode if isinstance(item.episode, list) else [item.episode] + + playable = find_playable_file(item.path) + if playable: + for sn in season_nums: + for ep in episode_nums: + key = (sn, ep) + if key not in all_episode_files: + all_episode_files[key] = [] + already_added = any(f["path"] == playable for f in all_episode_files.get(key, [])) + if not already_added: + size = item.content_hash.size if item.content_hash else 0 + all_episode_files[key].append({ + "path": playable, + "size": size, + "resolution": item.resolution, + "quality": item.quality, + "codec": item.codec, + "audio": item.audio, + "encoder": item.encoder, + "torrent_path": str(item.path), + }) + + return all_episode_files + + +def _build_episodes_data( + episodes_in_season: Dict[int, List[Dict]], + tmdb_episodes: Dict[int, TMDbEpisodeInfo], + series_folder: Path, + season_num: int, + generate_showreels: bool, + episode_reel_tasks: List, + series_title: str, + media_root: Optional[str] = None, +) -> List[dict]: + """Build episode data list for a season.""" + episodes_data = [] + + for episode_num in sorted(episodes_in_season.keys()): + episode_files = episodes_in_season[episode_num] + sort_by_quality(episode_files) + + tmdb_ep = tmdb_episodes.get(episode_num) + + reel_path = None + if generate_showreels and episode_files: + best_file = episode_files[0]["path"] + if best_file and not best_file.endswith(".bdmv"): + reel_path = get_expected_episode_reel_path(series_folder, season_num, episode_num, media_root=Path(media_root) if media_root else None) + episode_reel_tasks.append((best_file, series_folder, season_num, episode_num, series_title)) + + releases = [] + for f in episode_files: + releases.append({ + "path": make_relative_path(f["torrent_path"], media_root), + "playable_file": make_relative_path(f["path"], media_root), + "resolution": f.get("resolution"), + "quality": f.get("quality"), + "codec": f.get("codec"), + "audio": f.get("audio"), + "encoder": f.get("encoder"), + "size": f.get("size"), + }) + + episode_data = { + "episode_number": episode_num, + "name": tmdb_ep.name if tmdb_ep else None, + "overview": tmdb_ep.overview if tmdb_ep else None, + "air_date": tmdb_ep.air_date if tmdb_ep else None, + "runtime": tmdb_ep.runtime if tmdb_ep else None, + "still_path": tmdb_ep.still_path if tmdb_ep else None, + "rating": tmdb_ep.vote_average if tmdb_ep else None, + "director": tmdb_ep.director if tmdb_ep else None, + "reel_image": reel_path, + "releases": releases, + } + episodes_data.append(episode_data) + + return episodes_data + + +def _build_seasons_data( + all_episode_files: Dict[Tuple[int, int], List[Dict]], + tmdb_id: Optional[int], + series_folder: Path, + display_title: str, + fetch_covers: bool, + generate_showreels: bool, + season_cache: Dict, + episode_reel_tasks: List, + media_root: Optional[str] = None, +) -> List[dict]: + """Build seasons data structure for a series.""" + # Group episodes by season + seasons_map: Dict[int, Dict[int, List[Dict]]] = {} + for (season_num, episode_num), files in all_episode_files.items(): + if season_num not in seasons_map: + seasons_map[season_num] = {} + seasons_map[season_num][episode_num] = files + + seasons_data = [] + for season_num in sorted(seasons_map.keys()): + episodes_in_season = seasons_map[season_num] + + # Fetch TMDb season details if we have a TMDb ID + tmdb_season = None + tmdb_episodes: Dict[int, TMDbEpisodeInfo] = {} + + if tmdb_id: + cache_key = (tmdb_id, season_num) + if cache_key not in season_cache: + print(f" Fetching season {season_num} details for {display_title}") + season_cache[cache_key] = fetch_season_details(tmdb_id, season_num) + tmdb_season = season_cache[cache_key] + + if tmdb_season and tmdb_season.episodes: + for ep in tmdb_season.episodes: + tmdb_episodes[ep.episode_number] = ep + + # Download season poster + season_poster_path = None + if fetch_covers and tmdb_season and tmdb_season.poster_path: + season_poster_path = download_season_poster(tmdb_season.poster_path, series_folder, season_num) + + episodes_data = _build_episodes_data( + episodes_in_season, tmdb_episodes, series_folder, season_num, + generate_showreels, episode_reel_tasks, display_title, media_root + ) + + season_data = { + "season_number": season_num, + "name": tmdb_season.name if tmdb_season else None, + "overview": tmdb_season.overview if tmdb_season else None, + "air_date": tmdb_season.air_date if tmdb_season else None, + "poster_path": make_relative_path(season_poster_path, media_root) if season_poster_path else None, + "episode_count": len(episodes_data), + "episodes": episodes_data, + } + seasons_data.append(season_data) + + return seasons_data + + +def _process_movies( + categories: dict, + cover_dir: Path, + fetch_covers: bool, + generate_showreels: bool, + media_root: Optional[str] = None, +) -> Tuple[List[dict], List[Tuple[str, Path, str]]]: + """Process all movies and return (movies_list, showreel_tasks).""" + # In-memory cache for TMDb lookups + movie_tmdb_cache: Dict[str, Optional[TMDbInfo]] = {} + + def get_movie_tmdb(title: str, year: Optional[int]) -> Optional[TMDbInfo]: + cache_key = f"{title.lower()}:{year}" + if cache_key in movie_tmdb_cache: + return movie_tmdb_cache[cache_key] + tmdb_info = fetch_movie_info(title, year) + movie_tmdb_cache[cache_key] = tmdb_info + return tmdb_info + + def has_playable(item: ParsedContent) -> bool: + return find_playable_file(item.path) is not None + + # Filter movies with playable files + valid_movies = [item for item in categories[ContentType.MOVIE] if has_playable(item)] + skipped = len(categories[ContentType.MOVIE]) - len(valid_movies) + if skipped > 0: + print(f" Skipped {skipped} movie torrents with no playable video files") + + # Group by title+year + movie_groups: Dict[str, List[ParsedContent]] = {} + for item in valid_movies: + key = f"{item.title.lower()}:{item.year or 0}" + if key not in movie_groups: + movie_groups[key] = [] + movie_groups[key].append(item) + + # Re-group by TMDb ID + tmdb_movie_groups: Dict[int, Dict] = {} + no_tmdb_movie_groups: Dict[str, Dict] = {} + + print(f" Processing {len(movie_groups)} unique movies ({len(categories[ContentType.MOVIE])} total versions)...") + + for idx, (movie_key, items) in enumerate(movie_groups.items(), 1): + first_item = items[0] + print(f" [{idx}/{len(movie_groups)}] {first_item.title} ({first_item.year})\x1b[K", end="\r") + + tmdb_info = get_movie_tmdb(first_item.title, first_item.year) + + if tmdb_info and tmdb_info.tmdb_id: + if tmdb_info.tmdb_id not in tmdb_movie_groups: + tmdb_movie_groups[tmdb_info.tmdb_id] = { + "tmdb_info": tmdb_info, + "items": [], + "torrent_titles": set(), + "year": first_item.year, + } + tmdb_movie_groups[tmdb_info.tmdb_id]["items"].extend(items) + tmdb_movie_groups[tmdb_info.tmdb_id]["torrent_titles"].add(first_item.title) + else: + key = f"{first_item.title.lower()}:{first_item.year or 0}" + if key not in no_tmdb_movie_groups: + no_tmdb_movie_groups[key] = {"items": [], "title": first_item.title, "year": first_item.year} + no_tmdb_movie_groups[key]["items"].extend(items) + + print() + + movies = [] + movie_showreel_tasks: List[Tuple[str, Path, str]] = [] + + # Process movies with TMDb info + for tmdb_id, group_data in tmdb_movie_groups.items(): + tmdb_info = group_data["tmdb_info"] + items = group_data["items"] + torrent_titles = group_data["torrent_titles"] + year = group_data["year"] + + display_title = tmdb_info.title + item_id = hashlib.md5(f"movie:{tmdb_id}".encode()).hexdigest()[:12] + + # Find/download cover + cover_path = None + if fetch_covers: + cover_path = find_cover_image(display_title, year, "movie", cover_dir) + if not cover_path: + for tt in torrent_titles: + cover_path = find_cover_image(tt, year, "movie", cover_dir) + if cover_path: + break + if not cover_path and tmdb_info.poster_path: + cover_path = download_cover_image(tmdb_info.poster_path, display_title, year, "movie", cover_dir) + + versions = [_build_version_info(item, media_root) for item in items] + sort_by_quality(versions) + + # Queue showreel generation + showreel_paths = [] + if generate_showreels and versions: + best_playable = versions[0].get("playable_file") + if best_playable: + # Reconstruct absolute path from relative path + abs_playable = str(Path(media_root) / best_playable) if media_root else best_playable + media_folder = get_media_folder_path(display_title, year, "movie", cover_dir) + showreel_paths = get_expected_showreel_paths(media_folder, media_root=Path(media_root) if media_root else None) + movie_showreel_tasks.append((abs_playable, media_folder, display_title)) + + # Download backdrop + backdrop_path = None + if fetch_covers and tmdb_info.backdrop_path: + backdrop_path = download_backdrop_image(tmdb_info.backdrop_path, display_title, year, "movie", cover_dir) + + different_titles = [t for t in torrent_titles if t.lower() != display_title.lower()] + version_timestamps = [v["newest"] for v in versions if v.get("newest")] + newest = max(version_timestamps) if version_timestamps else None + + movies.append({ + "id": item_id, + "title": display_title, + "original_title": tmdb_info.original_title, + "alternative_titles": tmdb_info.alternative_titles, + "torrent_titles": different_titles if different_titles else None, + "year": year, + "newest": newest, + "cover_path": make_relative_path(cover_path, media_root), + "backdrop_path": make_relative_path(backdrop_path, media_root), + "showreel_images": showreel_paths if showreel_paths else None, + "versions": versions, + "tmdb_id": tmdb_info.tmdb_id, + "tmdb_title": tmdb_info.title, + "rating": tmdb_info.rating, + "vote_count": tmdb_info.vote_count, + "overview": tmdb_info.overview, + "genres": tmdb_info.genres, + "release_date": tmdb_info.release_date, + "runtime": tmdb_info.runtime, + "status": tmdb_info.status, + "tagline": tmdb_info.tagline, + "poster_path": tmdb_info.poster_path, + "similar": tmdb_info.similar, + "keywords": tmdb_info.keywords, + "cast": tmdb_info.cast, + "director": tmdb_info.director, + }) + + # Process movies without TMDb info + for key, group_data in no_tmdb_movie_groups.items(): + items = group_data["items"] + title = group_data["title"] + year = group_data["year"] + item_id = hashlib.md5(f"movie:{title}:{year}".encode()).hexdigest()[:12] + + cover_path = find_cover_image(title, year, "movie", cover_dir) if fetch_covers else None + + versions = [_build_version_info(item, media_root) for item in items] + sort_by_quality(versions) + + showreel_paths = [] + if generate_showreels and versions: + best_playable = versions[0].get("playable_file") + if best_playable: + # Reconstruct absolute path from relative path + abs_playable = str(Path(media_root) / best_playable) if media_root else best_playable + if not abs_playable.endswith(".bdmv"): + media_folder = get_media_folder_path(title, year, "movie", cover_dir) + showreel_paths = get_expected_showreel_paths(media_folder, media_root=Path(media_root) if media_root else None) + movie_showreel_tasks.append((abs_playable, media_folder, title)) + + version_timestamps = [v["newest"] for v in versions if v.get("newest")] + newest = max(version_timestamps) if version_timestamps else None + + movies.append({ + "id": item_id, + "title": title, + "original_title": None, + "torrent_titles": None, + "year": year, + "newest": newest, + "cover_path": make_relative_path(cover_path, media_root), + "showreel_images": showreel_paths if showreel_paths else None, + "versions": versions, + }) + + return movies, movie_showreel_tasks + + +def _process_series( + categories: dict, + cover_dir: Path, + fetch_covers: bool, + generate_showreels: bool, + media_root: Optional[str] = None, +) -> Tuple[List[dict], List[Tuple[str, Path, int, int, str]]]: + """Process all series and return (series_list, episode_reel_tasks).""" + # In-memory cache for TMDb lookups + series_tmdb_cache: Dict[str, Optional[TMDbInfo]] = {} + season_cache: Dict[Tuple[int, int], Optional[TMDbSeasonInfo]] = {} + + def get_series_tmdb(title: str) -> Optional[TMDbInfo]: + cache_key = title.lower() + if cache_key in series_tmdb_cache: + return series_tmdb_cache[cache_key] + tmdb_info = fetch_series_info(title) + series_tmdb_cache[cache_key] = tmdb_info + return tmdb_info + + def has_video_content(item: ParsedContent) -> bool: + if find_playable_file(item.path): + return True + return len(find_episode_files(item.path)) > 0 + + # Filter series with video content + valid_series = [item for item in categories[ContentType.SERIES] if has_video_content(item)] + skipped = len(categories[ContentType.SERIES]) - len(valid_series) + if skipped > 0: + print(f" Skipped {skipped} series torrents with no playable video files") + + # Group by title + series_groups: Dict[str, List[ParsedContent]] = {} + for item in valid_series: + key = item.title.lower() + if key not in series_groups: + series_groups[key] = [] + series_groups[key].append(item) + + # Re-group by TMDb ID + tmdb_groups: Dict[int, Dict] = {} + no_tmdb_groups: Dict[str, Dict] = {} + + print(f" Processing {len(series_groups)} unique series ({len(categories[ContentType.SERIES])} total entries)...") + + for idx, (series_key, items) in enumerate(series_groups.items(), 1): + first_item = items[0] + print(f" [{idx}/{len(series_groups)}] {first_item.title}\x1b[K", end="\r") + + tmdb_info = get_series_tmdb(first_item.title) + + if tmdb_info and tmdb_info.tmdb_id: + if tmdb_info.tmdb_id not in tmdb_groups: + tmdb_groups[tmdb_info.tmdb_id] = { + "tmdb_info": tmdb_info, + "items": [], + "torrent_titles": set(), + } + tmdb_groups[tmdb_info.tmdb_id]["items"].extend(items) + tmdb_groups[tmdb_info.tmdb_id]["torrent_titles"].add(first_item.title) + else: + key = first_item.title.lower() + if key not in no_tmdb_groups: + no_tmdb_groups[key] = {"items": [], "title": first_item.title} + no_tmdb_groups[key]["items"].extend(items) + + print() + + series = [] + episode_reel_tasks: List[Tuple[str, Path, int, int, str]] = [] + + # Process series with TMDb info + for series_idx, (tmdb_id, group_data) in enumerate(tmdb_groups.items(), 1): + tmdb_info = group_data["tmdb_info"] + items = group_data["items"] + torrent_titles = group_data["torrent_titles"] + + display_title = tmdb_info.title + series_id = hashlib.md5(f"series:{tmdb_id}".encode()).hexdigest()[:12] + + print(f" [{series_idx}/{len(tmdb_groups)}] {display_title}\x1b[K") + + series_folder = get_media_folder_path(display_title, None, "series", cover_dir) + + # Find/download cover + cover_path = None + if fetch_covers: + cover_path = find_cover_image(display_title, None, "series", cover_dir) + if not cover_path: + for tt in torrent_titles: + cover_path = find_cover_image(tt, None, "series", cover_dir) + if cover_path: + break + if not cover_path and tmdb_info.poster_path: + cover_path = download_cover_image(tmdb_info.poster_path, display_title, None, "series", cover_dir) + + # Download backdrop + backdrop_path = None + if fetch_covers and tmdb_info.backdrop_path: + backdrop_path = download_backdrop_image(tmdb_info.backdrop_path, display_title, None, "series", cover_dir) + + # Collect and build episode data + all_episode_files = _collect_episode_files(items) + seasons_data = _build_seasons_data( + all_episode_files, tmdb_id, series_folder, display_title, + fetch_covers, generate_showreels, season_cache, episode_reel_tasks, media_root + ) + + if not seasons_data: + print(f" Skipping {display_title} - no episodes found") + continue + + different_titles = [t for t in torrent_titles if t.lower() != display_title.lower()] + item_timestamps = [get_added_timestamp(item.path) for item in items] + item_timestamps = [t for t in item_timestamps if t is not None] + newest = max(item_timestamps) if item_timestamps else None + + series.append({ + "id": series_id, + "title": display_title, + "original_title": tmdb_info.original_title, + "torrent_titles": different_titles if different_titles else None, + "newest": newest, + "cover_path": make_relative_path(cover_path, media_root), + "backdrop_path": make_relative_path(backdrop_path, media_root), + "seasons": seasons_data, + "tmdb_id": tmdb_info.tmdb_id, + "tmdb_title": tmdb_info.title, + "rating": tmdb_info.rating, + "vote_count": tmdb_info.vote_count, + "overview": tmdb_info.overview, + "genres": tmdb_info.genres, + "release_date": tmdb_info.release_date, + "status": tmdb_info.status, + "tagline": tmdb_info.tagline, + "poster_path": tmdb_info.poster_path, + "similar": tmdb_info.similar, + "keywords": tmdb_info.keywords, + "cast": tmdb_info.cast, + "creators": tmdb_info.creators, + "number_of_seasons": tmdb_info.number_of_seasons, + "number_of_episodes": tmdb_info.number_of_episodes, + "networks": tmdb_info.networks, + }) + + # Process series without TMDb info + for key, group_data in no_tmdb_groups.items(): + items = group_data["items"] + title = group_data["title"] + series_id = hashlib.md5(f"series:{title}".encode()).hexdigest()[:12] + + cover_path = find_cover_image(title, None, "series", cover_dir) if fetch_covers else None + series_folder = get_media_folder_path(title, None, "series", cover_dir) + + all_episode_files = _collect_episode_files(items) + seasons_data = _build_seasons_data( + all_episode_files, None, series_folder, title, + fetch_covers, generate_showreels, season_cache, episode_reel_tasks, media_root + ) + + if not seasons_data: + print(f" Skipping {title} - no episodes found") + continue + + item_timestamps = [get_added_timestamp(item.path) for item in items] + item_timestamps = [t for t in item_timestamps if t is not None] + newest = max(item_timestamps) if item_timestamps else None + + series.append({ + "id": series_id, + "title": title, + "original_title": None, + "torrent_titles": None, + "newest": newest, + "cover_path": make_relative_path(cover_path, media_root), + "seasons": seasons_data, + }) + + return series, episode_reel_tasks + + +def _run_showreel_generation( + movie_tasks: List[Tuple[str, Path, str]], + episode_tasks: List[Tuple[str, Path, int, int, str]], +) -> None: + """Run showreel generation for movies and episodes.""" + pending_movie_tasks = [ + (vp, mf, t) for vp, mf, t in movie_tasks + if not movie_showreels_exist(mf) + ] + pending_episode_tasks = [ + (vp, mf, s, e, t) for vp, mf, s, e, t in episode_tasks + if not episode_reel_exists(mf, s, e) + ] + + total_units = len(pending_movie_tasks) * 5 + len(pending_episode_tasks) + skipped_movies = len(movie_tasks) - len(pending_movie_tasks) + skipped_episodes = len(episode_tasks) - len(pending_episode_tasks) + + if total_units > 0: + print(f"\nGenerating showreels: {len(pending_movie_tasks)} movies, {len(pending_episode_tasks)} episodes") + if skipped_movies > 0 or skipped_episodes > 0: + print(f" (skipping {skipped_movies} movies, {skipped_episodes} episodes already done)") + + with tqdm(total=total_units, unit="clip", dynamic_ncols=True) as pbar: + for video_path, media_folder, title in pending_movie_tasks: + pbar.set_description(f"{title[:40]}") + generate_showreel_images(video_path, media_folder, title=title, pbar=pbar) + + for video_path, media_folder, season_num, episode_num, series_title in pending_episode_tasks: + episode_code = f"S{season_num:02d}E{episode_num:02d}" + pbar.set_description(f"{series_title[:30]} {episode_code}") + generate_episode_reel(video_path, media_folder, season_num, episode_num, pbar=pbar) + + print("Showreel generation complete.") + elif movie_tasks or episode_tasks: + print(f"\nAll showreels already exist ({skipped_movies} movies, {skipped_episodes} episodes).") + + +def generate_media_index( + categories: dict[ContentType, list[ParsedContent]], + output_path: Path, + cover_dir: Path, + media_root: Optional[Path] = None, + fetch_covers: bool = True, + generate_showreels: bool = True, +) -> None: + """ + Generate a comprehensive metadata index for the media browser app. + + The index includes: + - Media metadata with versions bundled together + - Cover image paths (relative to media_root) + - Playable file paths + - TMDb data: ratings, cast, similar items, keywords, etc. + - Showreel images for movies and episodes + """ + print(f"Generating media index: {output_path}") + + # Convert media_root to string for relative path calculations + media_root_str = str(media_root) if media_root else None + + # Process movies and series + movies, movie_showreel_tasks = _process_movies(categories, cover_dir, fetch_covers, generate_showreels, media_root_str) + series, episode_reel_tasks = _process_series(categories, cover_dir, fetch_covers, generate_showreels, media_root_str) + + # Sort results + movies.sort(key=lambda x: (x["title"].lower(), x.get("year") or 0)) + series.sort(key=lambda x: x["title"].lower()) + + # Calculate totals + total_movie_versions = sum(len(m["versions"]) for m in movies) + total_series_episodes = sum( + sum(len(season.get("episodes", [])) for season in s["seasons"]) + for s in series + ) + + # Build and write the index + index = { + "version": 5, + "generated_at": datetime.now().isoformat(), + "media_root": media_root_str, + "stats": { + "total_movies": len(movies), + "total_movie_versions": total_movie_versions, + "total_series": len(series), + "total_series_episodes": total_series_episodes, + }, + "movies": movies, + "series": series, + } + + output_path.parent.mkdir(parents=True, exist_ok=True) + with open(output_path, "w", encoding="utf-8") as f: + json.dump(index, f, indent=2, ensure_ascii=False) + + print(f" Movies: {len(movies)} ({total_movie_versions} versions)") + print(f" Series: {len(series)} ({total_series_episodes} episodes)") + print(f" Output: {output_path}") + + # Generate showreels + if generate_showreels: + _run_showreel_generation(movie_showreel_tasks, episode_reel_tasks) diff --git a/hivescan/models.py b/hivescan/models.py new file mode 100644 index 0000000..18abb92 --- /dev/null +++ b/hivescan/models.py @@ -0,0 +1,62 @@ +"""Data models for the download scanner.""" + +import hashlib +from dataclasses import dataclass, field +from enum import Enum +from pathlib import Path +from typing import Optional + + +class ContentType(Enum): + """Types of content that can be identified.""" + MOVIE = "movie" + SERIES = "series" + OTHER = "other" + + +@dataclass +class ContentHash: + """Hash representing a file or directory's content based on torrent name.""" + path: Path + hash: str + _size: Optional[int] = None + + @property + def size(self) -> int: + """Get the size, computing it lazily if needed.""" + if self._size is None: + from hivescan.utils import get_directory_size + self._size = get_directory_size(self.path) + return self._size + + @size.setter + def size(self, value: int) -> None: + self._size = value + + @classmethod + def from_path(cls, path: Path) -> "ContentHash": + """Generate a content hash based on torrent name.""" + hash_val = hashlib.md5(path.name.encode()).hexdigest()[:16] + return cls(path=path, hash=hash_val) + + +@dataclass +class ParsedContent: + """Information parsed from a torrent name.""" + path: Path + name: str + content_type: ContentType + title: str + year: Optional[int] = None + resolution: Optional[str] = None + quality: Optional[str] = None + codec: Optional[str] = None + audio: Optional[str] = None + season: Optional[int] = None + episode: Optional[int] = None + episode_name: Optional[str] = None + encoder: Optional[str] = None + language: Optional[str] = None + is_directory: bool = False + raw_parsed: dict = field(default_factory=dict) + content_hash: Optional[ContentHash] = None diff --git a/hivescan/parsing.py b/hivescan/parsing.py new file mode 100644 index 0000000..1d7d4c2 --- /dev/null +++ b/hivescan/parsing.py @@ -0,0 +1,79 @@ +"""Torrent name parsing functions.""" + +import re +from pathlib import Path +from typing import Optional, Tuple + +import PTN + +from hivescan.models import ContentHash, ContentType, ParsedContent + + +def determine_content_type(parsed: dict) -> ContentType: + """Determine content type based on parsed torrent name info.""" + has_season = "season" in parsed and parsed["season"] is not None + has_episode = "episode" in parsed and parsed["episode"] is not None + has_year = "year" in parsed and parsed["year"] is not None + + if has_season or has_episode: + return ContentType.SERIES + if has_year: + return ContentType.MOVIE + return ContentType.OTHER + + +def parse_download(path: Path) -> ParsedContent: + """Parse a downloaded torrent directory/file name.""" + name = path.name + parsed = PTN.parse(name) + content_type = determine_content_type(parsed) + content_hash = ContentHash.from_path(path) + + return ParsedContent( + path=path, + name=name, + content_type=content_type, + title=parsed.get("title", name), + year=parsed.get("year"), + resolution=parsed.get("resolution"), + quality=parsed.get("quality"), + codec=parsed.get("codec"), + audio=parsed.get("audio"), + season=parsed.get("season"), + episode=parsed.get("episode"), + episode_name=parsed.get("episodeName"), + encoder=parsed.get("encoder"), + language=parsed.get("language"), + is_directory=path.is_dir(), + raw_parsed=parsed, + content_hash=content_hash, + ) + + +def parse_episode_from_filename(filename: str) -> Optional[Tuple[int, int]]: + """ + Parse season and episode numbers from a filename. + + Handles formats: S01E05, 1x05, Season 1 Episode 5 + + Returns: + Tuple of (season_number, episode_number) or None if not found + """ + name = filename.lower() + + # S01E05 format + match = re.search(r's(\d{1,2})e(\d{1,3})', name) + if match: + return int(match.group(1)), int(match.group(2)) + + # 1x05 format + match = re.search(r'(\d{1,2})x(\d{1,3})', name) + if match: + return int(match.group(1)), int(match.group(2)) + + # Season 1 Episode 5 format + match = re.search(r'season\s*(\d{1,2}).*episode\s*(\d{1,3})', name) + if match: + return int(match.group(1)), int(match.group(2)) + + return None diff --git a/hivescan/scanning.py b/hivescan/scanning.py new file mode 100644 index 0000000..d2dc9c8 --- /dev/null +++ b/hivescan/scanning.py @@ -0,0 +1,179 @@ +"""File system scanning functions.""" + +import glob +from pathlib import Path +from typing import Dict, Iterator, List, Optional, Tuple + +from hivescan.models import ContentType, ParsedContent +from hivescan.parsing import parse_download, parse_episode_from_filename +from hivescan.utils import get_media_folder_path, sanitize_filename + + +# Video file extensions +VIDEO_EXTENSIONS = {'.mkv', '.mp4', '.avi', '.m4v', '.mov', '.wmv', '.flv', '.webm', '.ts', '.m2ts'} + +# Caches for expensive operations +_episode_files_cache: Dict[str, Dict[Tuple[int, int], List[Tuple[str, int]]]] = {} +_playable_file_cache: Dict[str, Optional[str]] = {} + + +def scan_downloads(base_pattern: str) -> Iterator[ParsedContent]: + """ + Scan download directories matching the pattern. + + Args: + base_pattern: Glob pattern for finding download directories + + Yields: + ParsedContent objects for each found download + """ + exclude_patterns = [".torrents", "incomplete", ".incomplete"] + + for path_str in glob.glob(base_pattern): + path = Path(path_str) + + if path.name.startswith("."): + continue + + if any(excl.lower() in path.name.lower() for excl in exclude_patterns): + continue + + if not path.exists(): + continue + + yield parse_download(path) + + +def categorize_downloads(downloads: list[ParsedContent]) -> dict[ContentType, list[ParsedContent]]: + """Categorize downloads by content type.""" + categories: dict[ContentType, list[ParsedContent]] = { + ContentType.MOVIE: [], + ContentType.SERIES: [], + ContentType.OTHER: [], + } + + for download in downloads: + categories[download.content_type].append(download) + + return categories + + +def find_episode_files(path: Path) -> Dict[Tuple[int, int], List[Tuple[str, int]]]: + """ + Find all episode video files in a directory. + + Args: + path: Path to search (can be a season pack directory or single file) + + Returns: + Dict mapping (season_num, episode_num) to list of (file_path, file_size) tuples + """ + cache_key = str(path) + if cache_key in _episode_files_cache: + return _episode_files_cache[cache_key] + + episodes: Dict[Tuple[int, int], List[Tuple[str, int]]] = {} + + if path.is_file(): + if path.suffix.lower() in VIDEO_EXTENSIONS: + ep_info = parse_episode_from_filename(path.name) + if ep_info: + episodes[ep_info] = [(str(path), path.stat().st_size)] + _episode_files_cache[cache_key] = episodes + return episodes + + try: + for f in path.rglob("*"): + if f.is_file() and f.suffix.lower() in VIDEO_EXTENSIONS: + if 'sample' in f.name.lower(): + continue + ep_info = parse_episode_from_filename(f.name) + if ep_info: + if ep_info not in episodes: + episodes[ep_info] = [] + episodes[ep_info].append((str(f), f.stat().st_size)) + except (OSError, PermissionError): + pass + + _episode_files_cache[cache_key] = episodes + return episodes + + +def find_playable_file(path: Path) -> Optional[str]: + """ + Find the main playable media file in a directory. + + For Blu-ray discs: Returns BDMV/index.bdmv + For other content: Returns the largest video file + """ + cache_key = str(path) + if cache_key in _playable_file_cache: + return _playable_file_cache[cache_key] + + if path.is_file(): + if path.suffix.lower() in VIDEO_EXTENSIONS: + result = str(path) + _playable_file_cache[cache_key] = result + return result + _playable_file_cache[cache_key] = None + return None + + # Check for Blu-ray disc structure + bdmv_index = path / "BDMV" / "index.bdmv" + if bdmv_index.exists(): + result = str(bdmv_index) + _playable_file_cache[cache_key] = result + return result + + # Check nested Blu-ray structure (e.g., MovieName/DISC1/BDMV/) + try: + for subdir in path.iterdir(): + if subdir.is_dir(): + nested_bdmv = subdir / "BDMV" / "index.bdmv" + if nested_bdmv.exists(): + result = str(nested_bdmv) + _playable_file_cache[cache_key] = result + return result + except (OSError, PermissionError): + pass + + # Find largest video file + video_files = [] + try: + for f in path.rglob("*"): + if f.is_file() and f.suffix.lower() in VIDEO_EXTENSIONS: + if 'sample' in f.name.lower(): + continue + video_files.append((f, f.stat().st_size)) + except (OSError, PermissionError): + pass + + if not video_files: + _playable_file_cache[cache_key] = None + return None + + video_files.sort(key=lambda x: x[1], reverse=True) + result = str(video_files[0][0]) + _playable_file_cache[cache_key] = result + return result + + +def find_cover_image(title: str, year: Optional[int], media_type: str, cover_dir: Path) -> Optional[str]: + """Find a cover image for the given media item.""" + media_folder = get_media_folder_path(title, year, media_type, cover_dir) + cover_path = media_folder / "cover.jpg" + if cover_path.exists(): + return str(cover_path) + + # Legacy structure fallback + subdir = "movies" if media_type == "movie" else "series" + if media_type == "movie" and year: + legacy_path = cover_dir / subdir / f"{sanitize_filename(title)} ({year}).jpg" + if legacy_path.exists(): + return str(legacy_path) + + legacy_path = cover_dir / subdir / f"{sanitize_filename(title)}.jpg" + if legacy_path.exists(): + return str(legacy_path) + + return None diff --git a/hivescan/showreel.py b/hivescan/showreel.py new file mode 100644 index 0000000..18e0dc1 --- /dev/null +++ b/hivescan/showreel.py @@ -0,0 +1,704 @@ +""" +Showreel generation module for media preview clips. + +Generates short video clips (reels) from movies and TV episodes using ffmpeg. +Supports automatic black bar detection and removal, hardware-accelerated encoding, +and HDR passthrough. +""" + +import json +import re +import shlex +import subprocess +from collections import Counter +from pathlib import Path +from typing import Optional + +from tqdm import tqdm + + +# Showreel timestamp positions in seconds (5, 10, 15, 20, 25 minutes) +SHOWREEL_TIMESTAMPS = [5 * 60, 10 * 60, 15 * 60, 20 * 60, 25 * 60] + + +def get_expected_showreel_paths( + media_folder: Path, + timestamps: list[int] = SHOWREEL_TIMESTAMPS, + media_root: Optional[Path] = None, +) -> list[str]: + """ + Compute the expected showreel paths without generating them. + + Args: + media_folder: Folder for this specific media item + timestamps: List of timestamps (determines number of reels) + media_root: Root path for computing relative paths (optional) + + Returns: + List of relative paths where showreels will be created + """ + paths = [] + for reel_num in range(1, len(timestamps) + 1): + output_path = media_folder / f"reel{reel_num}.webm" + if media_root: + try: + paths.append(str(output_path.relative_to(media_root))) + except ValueError: + paths.append(str(output_path)) + else: + paths.append(str(output_path)) + return paths + + +def get_expected_episode_reel_path( + media_folder: Path, + season_num: int, + episode_num: int, + media_root: Optional[Path] = None, +) -> str: + """ + Compute the expected episode reel path without generating it. + + Args: + media_folder: Folder for this series + season_num: Season number + episode_num: Episode number + media_root: Root path for computing relative paths (optional) + + Returns: + Relative path where the reel will be created + """ + output_path = media_folder / f"S{season_num:02d}E{episode_num:02d}.webm" + if media_root: + try: + return str(output_path.relative_to(media_root)) + except ValueError: + return str(output_path) + return str(output_path) + + +def movie_showreels_exist(media_folder: Path, timestamps: list[int] = SHOWREEL_TIMESTAMPS) -> bool: + """Check if all showreel files for a movie already exist.""" + for reel_num in range(1, len(timestamps) + 1): + if not (media_folder / f"reel{reel_num}.webm").exists(): + return False + return True + + +def episode_reel_exists(media_folder: Path, season_num: int, episode_num: int) -> bool: + """Check if an episode reel file already exists.""" + return (media_folder / f"S{season_num:02d}E{episode_num:02d}.webm").exists() + + +def get_bluray_uri(video_path: str) -> Optional[str]: + """ + Convert a Blu-ray index.bdmv path to an ffmpeg-compatible bluray: URI. + + Args: + video_path: Path that may be a Blu-ray index.bdmv file + + Returns: + bluray: URI if this is a Blu-ray disc, None otherwise + """ + if not video_path.endswith(".bdmv"): + return None + + path = Path(video_path) + # index.bdmv is in BDMV folder, so parent's parent is the disc root + # e.g., /path/to/disc/BDMV/index.bdmv -> /path/to/disc + if path.parent.name == "BDMV": + disc_root = path.parent.parent + return f"bluray:{disc_root}" + + return None + + +# Cache for AV1 encoder availability +_av1_encoder_cache: Optional[str] = None + + +def get_av1_encoder() -> str: + """ + Detect the best available AV1 encoder. + + Prefers hardware encoders (NVIDIA av1_nvenc) over software (libsvtav1). + Falls back to libsvtav1 if no hardware encoder is available. + + Returns: + Encoder name to use with ffmpeg -c:v + """ + global _av1_encoder_cache + if _av1_encoder_cache is not None: + return _av1_encoder_cache + + # Check for NVIDIA AV1 encoder + try: + result = subprocess.run( + ["ffmpeg", "-hide_banner", "-encoders"], + capture_output=True, + text=True, + timeout=10 + ) + if "av1_nvenc" in result.stdout: + # Verify it actually works (driver support) + test_result = subprocess.run( + ["ffmpeg", "-f", "lavfi", "-i", "nullsrc=s=64x64:d=1", "-c:v", "av1_nvenc", "-f", "null", "-"], + capture_output=True, + timeout=10 + ) + if test_result.returncode == 0: + _av1_encoder_cache = "av1_nvenc" + return _av1_encoder_cache + except Exception: + pass + + # Default to libsvtav1 + _av1_encoder_cache = "libsvtav1" + return _av1_encoder_cache + + +def get_encoder_options(encoder: str) -> list[str]: + """ + Get encoder-specific options for the given AV1 encoder. + + Args: + encoder: The encoder name (av1_nvenc, libsvtav1) + + Returns: + List of ffmpeg arguments for encoder settings + """ + if encoder == "av1_nvenc": + # NVIDIA hardware encoder - use constant quality mode + return ["-cq", "35", "-preset", "p4"] + else: + # libsvtav1 software encoder + return ["-crf", "38", "-preset", "6"] + + +def detect_dovi_profile(video_path: str) -> Optional[int]: + """ + Detect Dolby Vision profile from a video file. + + Returns the DoVi profile number (5, 7, 8, etc.) or None if not DoVi. + Profile 5: Dual-layer, no HDR10 base (needs conversion) + Profile 7: Dual-layer with HDR10 base, but may have EL issues + Profile 8: Single-layer HDR10 compatible (usually OK) + """ + try: + # Check for Dolby Vision configuration record in video stream + cmd = [ + "ffprobe", "-v", "error", "-select_streams", "v:0", + "-show_entries", "stream_side_data_list", + "-of", "json", video_path + ] + print(f" $ {shlex.join(cmd)}") + result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) + + if result.returncode != 0: + return None + + data = json.loads(result.stdout) + streams = data.get("streams", []) + if not streams: + return None + + # Look for DOVI configuration in side data + side_data_list = streams[0].get("side_data_list", []) + for side_data in side_data_list: + side_data_type = side_data.get("side_data_type", "") + if "DOVI" in side_data_type or "Dolby Vision" in side_data_type: + # Try to extract profile from dv_profile field + dv_profile = side_data.get("dv_profile") + if dv_profile is not None: + return int(dv_profile) + + # Alternative: check using mediainfo-style detection via codec tag + # Some DoVi content has "dvhe" or "dvh1" codec tags + codec_cmd = [ + "ffprobe", "-v", "error", "-select_streams", "v:0", + "-show_entries", "stream=codec_tag_string,codec_name", + "-of", "csv=p=0", video_path + ] + codec_result = subprocess.run(codec_cmd, capture_output=True, text=True, timeout=30) + if codec_result.returncode == 0: + codec_info = codec_result.stdout.lower() + if "dvhe" in codec_info or "dvh1" in codec_info or "dav1" in codec_info: + # DoVi detected but profile unknown, assume needs conversion + return 7 # Conservative: treat as dual-layer + + return None + except Exception as e: + print(f" DoVi detection error: {e}") + return None + + +def get_dovi_to_hdr10_filter() -> str: + """ + Get the video filter string for converting DoVi to HDR10. + + Uses libplacebo to strip DoVi metadata while preserving HDR10 colorspace. + No tonemapping is applied - this just converts the container format. + """ + # libplacebo converts DoVi to clean HDR10 without tonemapping + # Preserves bt2020 primaries and SMPTE ST 2084 (PQ) transfer + return ( + "libplacebo=colorspace=bt2020nc:color_primaries=bt2020:" + "color_trc=smpte2084:range=tv" + ) + + +def is_hdr_video(video_path: str) -> bool: + """ + Check if a video file is HDR using ffprobe. + + Returns True if the video has HDR metadata (bt2020, SMPTE ST 2084, etc.) + """ + try: + result = subprocess.run( + [ + "ffprobe", "-v", "quiet", "-select_streams", "v:0", + "-show_entries", "stream=color_transfer,color_primaries,color_space", + "-of", "json", video_path + ], + capture_output=True, + text=True, + timeout=30 + ) + if result.returncode != 0: + return False + + data = json.loads(result.stdout) + streams = data.get("streams", []) + if not streams: + return False + + stream = streams[0] + color_transfer = stream.get("color_transfer", "") + color_primaries = stream.get("color_primaries", "") + + # HDR indicators + hdr_transfers = ["smpte2084", "arib-std-b67"] # PQ and HLG + hdr_primaries = ["bt2020"] + + return color_transfer in hdr_transfers or color_primaries in hdr_primaries + except Exception: + return False + + +def detect_crop(video_path: str) -> Optional[str]: + """ + Detect black bars in a video and return the crop filter string. + + Only runs on 16:9 (1.78:1) source videos, since other aspect ratios like + 2.35:1 or 4:3 are already correctly framed. Trusts cropping results only + when symmetric (same top/bottom OR same left/right). Final coordinates + are aligned to 8 pixels. + + Uses ffmpeg to analyze just 2 seconds of video at the 5-minute mark for speed. + + Args: + video_path: Path to the video file (or bluray: URI) + + Returns: + Crop filter string like "crop=1920:800:0:140" if black bars detected, + or None if no cropping needed or detection failed. + """ + try: + # First, get source video dimensions to check if it's 16:9 + dim_cmd = [ + "ffprobe", "-v", "error", "-select_streams", "v:0", + "-show_entries", "stream=width,height", + "-of", "csv=p=0", video_path + ] + print(f" $ {shlex.join(dim_cmd)}") + dim_result = subprocess.run(dim_cmd, capture_output=True, text=True, timeout=30) + if dim_result.returncode != 0: + print(f" Failed to get dimensions: {dim_result.stderr.strip()}") + return None + + # Parse "width,height" output + parts = dim_result.stdout.strip().split(",") + if len(parts) < 2: + return None + src_width, src_height = int(parts[0]), int(parts[1]) + + # Check if source is 16:9 (allow small tolerance for weird resolutions) + # 16:9 = 1.777..., typical: 1920x1080, 3840x2160, 1280x720 + aspect_ratio = src_width / src_height + if not (1.7 <= aspect_ratio <= 1.85): + # Not 16:9, skip crop detection (already correctly framed) + return None + + # Use ffmpeg to run cropdetect on just 2 seconds at 5-minute mark + # This is much faster than scanning 60 seconds with ffprobe lavfi + cmd = [ + "ffmpeg", "-hide_banner", "-ss", "300", "-i", video_path, + "-t", "2", "-vf", "cropdetect=limit=24:round=2:reset=0", + "-f", "null", "-" + ] + print(f" $ {shlex.join(cmd)}") + result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) + + # cropdetect outputs to stderr like: [Parsed_cropdetect_0 @ ...] x1:0 x2:1919 y1:138 y2:941 w:1920 h:800 ... + # We need to parse the crop values from stderr + crop_pattern = re.compile(r'crop=(\d+):(\d+):(\d+):(\d+)') + crop_values = [] + for line in result.stderr.split('\n'): + match = crop_pattern.search(line) + if match: + w, h, x, y = int(match.group(1)), int(match.group(2)), int(match.group(3)), int(match.group(4)) + if w > 0 and h > 0 and x >= 0 and y >= 0: + crop_values.append((w, h, x, y)) + + if not crop_values: + return None + + # Use the most common crop values (mode) for stability + most_common = Counter(crop_values).most_common(1) + if not most_common: + return None + + w, h, x, y = most_common[0][0] + + # Only crop if there's meaningful black bar removal (at least 8 pixels offset) + if x < 8 and y < 8: + return None + + # Validate symmetry: trust only if cropping is symmetric in one direction + # (same top/bottom for letterbox, OR same left/right for pillarbox) + # Allow up to 4 pixels of rounding error + left_crop = x + right_crop = src_width - (x + w) + top_crop = y + bottom_crop = src_height - (y + h) + + horizontal_symmetric = abs(left_crop - right_crop) <= 4 + vertical_symmetric = abs(top_crop - bottom_crop) <= 4 + + # Must be symmetric in at least one direction, but not require both + # (letterbox = vertical symmetric, pillarbox = horizontal symmetric) + if not (horizontal_symmetric or vertical_symmetric): + return None + + # If cropping in both directions, both must be symmetric + if x >= 8 and y >= 8: + if not (horizontal_symmetric and vertical_symmetric): + return None + + # Align all coordinates to 8 pixels (shrink content area if needed) + # x and y: round UP to next multiple of 8 + x_aligned = ((x + 7) // 8) * 8 + y_aligned = ((y + 7) // 8) * 8 + # w and h: round DOWN to multiple of 8, accounting for adjusted x/y + w_aligned = ((w - (x_aligned - x)) // 8) * 8 + h_aligned = ((h - (y_aligned - y)) // 8) * 8 + + # Ensure we still have valid dimensions + if w_aligned <= 0 or h_aligned <= 0: + return None + + crop_result = f"crop={w_aligned}:{h_aligned}:{x_aligned}:{y_aligned}" + print(f" Detected crop: {crop_result}") + return crop_result + + except Exception as e: + print(f" Crop detection error: {e}") + return None + + +def get_video_duration(video_path: str) -> Optional[float]: + """ + Get the duration of a video file in seconds using ffprobe. + """ + try: + result = subprocess.run( + [ + "ffprobe", "-v", "quiet", "-show_entries", "format=duration", + "-of", "json", video_path + ], + capture_output=True, + text=True, + timeout=30 + ) + if result.returncode != 0: + return None + + data = json.loads(result.stdout) + duration = data.get("format", {}).get("duration") + return float(duration) if duration else None + except Exception: + return None + + +def generate_showreel_images( + video_path: str, + media_folder: Path, + timestamps: list[int] = SHOWREEL_TIMESTAMPS, + title: str = None, + pbar: Optional[tqdm] = None, +) -> list[str]: + """ + Generate showreel video clips from a video file at specified timestamps. + + Saves 10-second clips in WebM format (AV1 video + Opus 2.0 audio), downscaled to max 720px width, + preserving original color metadata. Files are named reel1.webm, reel2.webm, etc. + + Args: + video_path: Path to the video file (or index.bdmv for Blu-ray discs) + media_folder: Folder for this specific media item + timestamps: List of timestamps in seconds to capture + pbar: Optional tqdm progress bar to update + + Returns: + List of relative paths to generated showreel video clips + """ + if not video_path: + return [] + + # Handle Blu-ray disc structures using bluray: protocol + bluray_uri = get_bluray_uri(video_path) + if bluray_uri: + ffmpeg_input = bluray_uri + else: + if not Path(video_path).exists(): + return [] + ffmpeg_input = video_path + + # Fast path: check if all showreel clips already exist before any ffprobe calls + existing_paths = [] + all_exist = True + for reel_num in range(1, len(timestamps) + 1): + output_filename = f"reel{reel_num}.webm" + output_path = media_folder / output_filename + if output_path.exists(): + existing_paths.append(str(output_path)) + else: + all_exist = False + break + + if all_exist and existing_paths: + return existing_paths + + media_folder.mkdir(parents=True, exist_ok=True) + + # Check video duration to avoid seeking past the end + duration = get_video_duration(ffmpeg_input) + if duration is None: + print(f" Could not get duration for: {video_path}") + return [] + + # Filter timestamps that are within the video duration (with 40s margin for 10s clips) + valid_timestamps = [t for t in timestamps if t < (duration - 40)] + if not valid_timestamps: + # If video is too short, try to get at least one clip from middle + if duration > 60: + valid_timestamps = [int(duration / 2) - 5] # Center the 10s clip + else: + return [] + + # Get the best available AV1 encoder + encoder = get_av1_encoder() + encoder_opts = get_encoder_options(encoder) + + # Detect Dolby Vision profile for tonemapping (profiles 5/7 need conversion) + dovi_profile = detect_dovi_profile(ffmpeg_input) + needs_tonemap = dovi_profile is not None and dovi_profile in (5, 7) + if needs_tonemap: + print(f" DoVi profile {dovi_profile} detected, will convert to HDR10") + + # Detect black bars once for all clips (uses same video source) + crop_filter = detect_crop(ffmpeg_input) + + generated_paths = [] + + for reel_num, timestamp in enumerate(valid_timestamps, 1): + output_filename = f"reel{reel_num}.webm" + output_path = media_folder / output_filename + + # Skip if already exists + if output_path.exists(): + generated_paths.append(str(output_path)) + if pbar: + pbar.update(1) + continue + + # Build video filter chain: + # 1. DoVi to HDR10 conversion (if needed) - must come first + # 2. Crop black bars (if detected) + # 3. Scale to max 720px width + vf_parts = [] + if needs_tonemap: + vf_parts.append(get_dovi_to_hdr10_filter()) + if crop_filter: + vf_parts.append(crop_filter) + vf_parts.append("scale='min(720,iw)':-2") + vf_filter = ",".join(vf_parts) + cmd = [ + "ffmpeg", "-y", "-ss", str(timestamp), "-i", ffmpeg_input, + "-hide_banner", "-loglevel", "warning", "-stats", + "-map", "0:v:0", "-map", "0:a:0?", # First video, first audio (optional) + "-t", "10", + "-vf", vf_filter, + "-c:v", encoder, + *encoder_opts, + "-c:a", "libopus", + "-ac", "2", + "-b:a", "128k", + str(output_path), + ] + + print(f" $ {shlex.join(cmd)}") + try: + result = subprocess.run(cmd, timeout=120) + + if result.returncode == 0 and output_path.exists(): + generated_paths.append(str(output_path)) + if pbar: + pbar.update(1) + else: + output_path.unlink(missing_ok=True) + if pbar: + # Update remaining reels as skipped + remaining = len(valid_timestamps) - reel_num + 1 + pbar.update(remaining) + pbar.refresh() + # Abort remaining reels - if first one fails, others likely will too + break + except BaseException as e: + output_path.unlink(missing_ok=True) + if isinstance(e, (KeyboardInterrupt, SystemExit)): + raise + if pbar: + pbar.clear() + print(f"\n\033[91mError generating showreel for {title or 'unknown'} at {timestamp}s: {e}\033[0m") + if pbar: + # Update remaining reels as skipped + remaining = len(valid_timestamps) - reel_num + 1 + pbar.update(remaining) + pbar.refresh() + # Abort remaining reels + break + + return generated_paths + + +def generate_episode_reel( + video_path: str, + media_folder: Path, + season_num: int, + episode_num: int, + pbar: Optional[tqdm] = None, +) -> Optional[str]: + """ + Generate a single 10-second reel video clip for a TV episode. + + Saves clip as SxxExx.webm (e.g., S01E05.webm) in the series folder. + WebM container with AV1 video + Opus 2.0 audio, downscaled to max 720px width, + preserving original color metadata. + + Args: + video_path: Path to the episode video file (or index.bdmv for Blu-ray discs) + media_folder: Folder for this series + season_num: Season number + episode_num: Episode number + pbar: Optional tqdm progress bar to update + + Returns: + Relative path to generated image, or None if failed + """ + if not video_path: + return None + + # Handle Blu-ray disc structures using bluray: protocol + bluray_uri = get_bluray_uri(video_path) + if bluray_uri: + ffmpeg_input = bluray_uri + else: + if not Path(video_path).exists(): + return None + ffmpeg_input = video_path + + media_folder.mkdir(parents=True, exist_ok=True) + + # Normalize episode code to SxxExx format + output_filename = f"S{season_num:02d}E{episode_num:02d}.webm" + output_path = media_folder / output_filename + + # Skip if already exists + if output_path.exists(): + return str(output_path) + + # Check video duration + duration = get_video_duration(ffmpeg_input) + if duration is None: + return None + + # Use 40% of total length for the clip start + actual_timestamp = int(duration * 0.4) + # Ensure we're at least 10 seconds in and have room for 10s clip + actual_timestamp = max(10, min(actual_timestamp, duration - 40)) + + # Get the best available AV1 encoder + encoder = get_av1_encoder() + encoder_opts = get_encoder_options(encoder) + + # Detect Dolby Vision profile for tonemapping (profiles 5/7 need conversion) + dovi_profile = detect_dovi_profile(ffmpeg_input) + needs_tonemap = dovi_profile is not None and dovi_profile in (5, 7) + if needs_tonemap: + print(f" DoVi profile {dovi_profile} detected, will convert to HDR10") + + # Detect black bars for cropping + crop_filter = detect_crop(ffmpeg_input) + + # Build video filter chain: + # 1. DoVi to HDR10 conversion (if needed) - must come first + # 2. Crop black bars (if detected) + # 3. Scale to max 720px width + vf_parts = [] + if needs_tonemap: + vf_parts.append(get_dovi_to_hdr10_filter()) + if crop_filter: + vf_parts.append(crop_filter) + vf_parts.append("scale='min(720,iw)':-2") + vf_filter = ",".join(vf_parts) + cmd = [ + "ffmpeg", "-y", "-ss", str(actual_timestamp), "-i", ffmpeg_input, + "-hide_banner", "-loglevel", "warning", "-stats", + "-map", "0:v:0", "-map", "0:a:0?", # First video, first audio (optional) + "-t", "10", + "-vf", vf_filter, + "-c:v", encoder, + *encoder_opts, + "-c:a", "libopus", + "-ac", "2", + "-b:a", "128k", + str(output_path), + ] + + print(f" $ {shlex.join(cmd)}") + try: + result = subprocess.run(cmd, timeout=120) + + if result.returncode == 0 and output_path.exists(): + if pbar: + pbar.update(1) + return str(output_path) + else: + output_path.unlink(missing_ok=True) + if pbar: + pbar.update(1) + pbar.refresh() + return None + except BaseException as e: + output_path.unlink(missing_ok=True) + if isinstance(e, (KeyboardInterrupt, SystemExit)): + raise + if pbar: + pbar.clear() + episode_code = f"S{season_num:02d}E{episode_num:02d}" + print(f"\n\033[91mError generating episode reel for {episode_code}: {e}\033[0m") + if pbar: + pbar.update(1) + pbar.refresh() + return None diff --git a/hivescan/tmdb_client.py b/hivescan/tmdb_client.py new file mode 100644 index 0000000..920a131 --- /dev/null +++ b/hivescan/tmdb_client.py @@ -0,0 +1,569 @@ +#!/usr/bin/env python3 +""" +TMDb Client - Fetch movie and TV series metadata from The Movie Database (TMDb). +""" + +import hashlib +import json +import os +import sys +import time +import urllib.parse +from dataclasses import dataclass +from pathlib import Path +from typing import Dict, List, Optional + +import httpx + +# TMDb API configuration +TMDB_API_KEY = os.environ.get("TMDB_API_KEY", "6bd914e6a5df1c6d1ddf622cf2dbc232") +TMDB_API_BASE = "https://api.themoviedb.org/3" + +# API response cache directory (can be overridden via set_cache_dir) +_tmdb_cache_dir: Optional[Path] = None + +# Persistent HTTP client for connection reuse +_http_client: Optional[httpx.Client] = None + + +def set_cache_dir(cache_dir: Path) -> None: + """Set the directory for TMDb API response cache.""" + global _tmdb_cache_dir + _tmdb_cache_dir = cache_dir + + +def _get_cache_dir() -> Path: + """Get the TMDb cache directory, defaulting to current directory if not set.""" + if _tmdb_cache_dir is not None: + return _tmdb_cache_dir + # Fallback to .tmdb-cache in current working directory + return Path.cwd() / ".tmdb-cache" + + +def _get_http_client() -> httpx.Client: + """Get or create a persistent HTTP client for connection reuse.""" + global _http_client + if _http_client is None: + _http_client = httpx.Client( + base_url=TMDB_API_BASE, + headers={"Accept": "application/json", "User-Agent": "TorrentManager/1.0"}, + timeout=10.0, + http2=True, # Enable HTTP/2 for better performance + ) + return _http_client + + +# Sentinel value to distinguish "cached None" from "not in cache" +_NOT_FOUND = object() + + +def _get_cache_path(endpoint: str, params: Dict[str, str]) -> Path: + """Generate a cache file path for an API request.""" + # Create a stable cache key from endpoint and sorted params + cache_key = endpoint + "?" + urllib.parse.urlencode(sorted(params.items())) + cache_hash = hashlib.sha256(cache_key.encode()).hexdigest() + return _get_cache_dir() / f"{cache_hash}.json" + + +def _load_from_cache(cache_path: Path): + """Load cached response. Returns _NOT_FOUND if not cached.""" + if not cache_path.exists(): + return _NOT_FOUND + try: + with open(cache_path, "r") as f: + data = json.load(f) + # Handle cached "no results" / errors + if data.get("_cached_none"): + return None + return data + except Exception: + return _NOT_FOUND + + +def _save_to_cache(cache_path: Path, data: Optional[Dict]): + """Save response to cache.""" + try: + _get_cache_dir().mkdir(parents=True, exist_ok=True) + with open(cache_path, "w") as f: + if data is None: + json.dump({"_cached_none": True}, f) + else: + json.dump(data, f) + except Exception: + pass # Cache write failures are not critical + + +@dataclass +class TMDbEpisodeInfo: + """Information about a TV episode from TMDb.""" + episode_number: int + season_number: int + name: Optional[str] = None + overview: Optional[str] = None + air_date: Optional[str] = None + runtime: Optional[int] = None # Minutes + still_path: Optional[str] = None # Episode screenshot + vote_average: Optional[float] = None + vote_count: Optional[int] = None + director: Optional[str] = None + + +@dataclass +class TMDbSeasonInfo: + """Information about a TV season from TMDb.""" + season_number: int + name: Optional[str] = None + overview: Optional[str] = None + air_date: Optional[str] = None + poster_path: Optional[str] = None + episode_count: Optional[int] = None + episodes: Optional[List[TMDbEpisodeInfo]] = None + + +@dataclass +class TMDbInfo: + """Information fetched from TMDb.""" + tmdb_id: int + title: Optional[str] = None # Official title from TMDb + original_title: Optional[str] = None + alternative_titles: Optional[List[str]] = None # Titles in other languages + rating: Optional[float] = None + vote_count: Optional[int] = None + overview: Optional[str] = None + genres: Optional[List[str]] = None + release_date: Optional[str] = None + runtime: Optional[int] = None # Minutes for movies + status: Optional[str] = None # Released, Ended, etc. + tagline: Optional[str] = None + poster_path: Optional[str] = None # TMDb poster path + backdrop_path: Optional[str] = None + similar: Optional[List[Dict]] = None # List of similar movies/shows + keywords: Optional[List[str]] = None + cast: Optional[List[Dict]] = None # Top cast members + director: Optional[str] = None # For movies + creators: Optional[List[str]] = None # For TV series + number_of_seasons: Optional[int] = None # For TV series + number_of_episodes: Optional[int] = None # For TV series + networks: Optional[List[str]] = None # For TV series + seasons: Optional[List[TMDbSeasonInfo]] = None # Season details for TV series + + +def tmdb_api_request(endpoint: str, params: Optional[Dict[str, str]] = None) -> Optional[Dict[str, str]]: + """Make a request to the TMDb API with disk caching and connection reuse.""" + params = params or {} + + # Check cache first (before adding API key to params for cache key) + cache_path = _get_cache_path(endpoint, params) + cached = _load_from_cache(cache_path) + if cached is not _NOT_FOUND: + return cached + + params["api_key"] = TMDB_API_KEY + + try: + client = _get_http_client() + response = client.get(endpoint, params=params) + + if response.status_code == 429: + # Rate limited - wait and retry + print(f" Rate limited, waiting...", file=sys.stderr) + time.sleep(1) + return tmdb_api_request(endpoint, {k: v for k, v in params.items() if k != "api_key"}) + + response.raise_for_status() + data = response.json() + # Cache immediately after receiving response + _save_to_cache(cache_path, data) + return data + except httpx.HTTPStatusError: + # Cache the failure (None) to avoid retrying + _save_to_cache(cache_path, None) + return None + except Exception: + # Don't cache network errors - they may be transient + return None + + +def fetch_movie_details(movie_id: int) -> Optional[Dict]: + """Fetch detailed movie info including credits, similar, keywords, and alternative titles.""" + # Use append_to_response to get multiple data in one request + data = tmdb_api_request( + f"/movie/{movie_id}", + {"append_to_response": "credits,similar,keywords,alternative_titles"} + ) + return data + + +def fetch_series_details(series_id: int) -> Optional[Dict]: + """Fetch detailed TV series info including credits, similar, and keywords.""" + # Use append_to_response to get multiple data in one request + data = tmdb_api_request( + f"/tv/{series_id}", + {"append_to_response": "credits,similar,keywords"} + ) + return data + + +def fetch_season_details(series_id: int, season_number: int) -> Optional[TMDbSeasonInfo]: + """ + Fetch detailed season info including all episodes. + + Returns season metadata with episode list including: + - Episode names, overviews, air dates + - Episode still images + - Runtime, ratings + - Directors for each episode + """ + data = tmdb_api_request( + f"/tv/{series_id}/season/{season_number}", + {"append_to_response": "images"} + ) + + if not data: + return None + + # Parse episodes + episodes = [] + for ep_data in data.get("episodes", []): + # Get director from crew + director = None + for crew_member in ep_data.get("crew", []): + if crew_member.get("job") == "Director": + director = crew_member.get("name") + break + + episode = TMDbEpisodeInfo( + episode_number=ep_data.get("episode_number", 0), + season_number=ep_data.get("season_number", season_number), + name=ep_data.get("name"), + overview=ep_data.get("overview"), + air_date=ep_data.get("air_date"), + runtime=ep_data.get("runtime"), + still_path=ep_data.get("still_path"), + vote_average=ep_data.get("vote_average"), + vote_count=ep_data.get("vote_count"), + director=director, + ) + episodes.append(episode) + + return TMDbSeasonInfo( + season_number=data.get("season_number", season_number), + name=data.get("name"), + overview=data.get("overview"), + air_date=data.get("air_date"), + poster_path=data.get("poster_path"), + episode_count=len(episodes), + episodes=episodes, + ) + + +def _generate_title_variants(words: list[str], min_words: int = 2) -> list[str]: + """ + Generate title variants by progressively removing words from both ends. + + Order: full title, then shorter from end, then shorter from start. + """ + if len(words) < min_words: + return [" ".join(words)] if words else [] + + variants = [] + + # First try full title + variants.append(" ".join(words)) + + # Then try removing from end (most common: edition names at end) + for num_words in range(len(words) - 1, min_words - 1, -1): + variants.append(" ".join(words[:num_words])) + + # Then try removing from start (garbage at beginning) + for start in range(1, len(words) - min_words + 1): + variants.append(" ".join(words[start:])) + + # Finally try middle portions (remove from both ends) + for start in range(1, len(words) - min_words): + for end in range(len(words) - 1, start + min_words - 1, -1): + variant = " ".join(words[start:end]) + if variant not in variants: + variants.append(variant) + + return variants + + +def _normalize_for_match(text: str) -> set[str]: + """Normalize text into a set of lowercase words for matching.""" + # Remove common punctuation and split + normalized = text.lower() + for char in ".:;,!?-_'\"()[]{}": + normalized = normalized.replace(char, " ") + return {w for w in normalized.split() if len(w) > 1} + + +def _titles_match(original_title: str, tmdb_title: str, search_query: str) -> bool: + """ + Check if TMDb result title reasonably matches our original title. + + Uses word overlap to verify the result is relevant, preventing + false matches from short queries like "The" or just a year. + """ + original_words = _normalize_for_match(original_title) + tmdb_words = _normalize_for_match(tmdb_title) + query_words = _normalize_for_match(search_query) + + # Remove common stop words that don't help matching + stop_words = {"the", "a", "an", "of", "and", "or", "in", "on", "at", "to", "for", "is", "it"} + original_significant = original_words - stop_words + tmdb_significant = tmdb_words - stop_words + query_significant = query_words - stop_words + + # The query words should be a subset of both original and tmdb titles + # (the search query came from the original, and should match the result) + if not query_significant: + # If query has no significant words, require direct word overlap + return bool(original_words & tmdb_words) + + # Check if significant query words appear in the TMDb title + query_in_tmdb = query_significant & tmdb_significant + if not query_in_tmdb: + return False + + # Also require some overlap between original and TMDb + # This catches cases where query matches but it's the wrong movie + overlap = original_significant & tmdb_significant + + # Either good overlap, or the TMDb title is contained in original (or vice versa) + return bool(overlap) or tmdb_significant <= original_significant or original_significant <= tmdb_significant + + +def _search_movie_with_fallbacks(title: str, year: Optional[int]) -> Optional[Dict]: + """ + Search for a movie with progressive title shortening fallbacks. + + PTN often includes edition names (THEATRICAL CUT, DIRECTOR'S CUT, etc.) + or garbage at the beginning/end of the title. + Year is always included when available as it's more reliable. + Results are validated with fuzzy matching to prevent false positives. + """ + words = title.split() + variants = _generate_title_variants(words, min_words=2) + + def _result_matches(top_result: Dict, original_title: str, search_query: str) -> bool: + """Check if result matches against either title or original_title.""" + tmdb_title = top_result.get("title", "") + tmdb_original = top_result.get("original_title", "") + return ( + _titles_match(original_title, tmdb_title, search_query) + or _titles_match(original_title, tmdb_original, search_query) + ) + + # Try all variants with year first + if year: + for search_title in variants: + params = {"query": search_title, "include_adult": "false", "year": str(year)} + data = tmdb_api_request("/search/movie", params) + if data and data.get("results"): + # Validate the top result matches our title (check both title and original_title) + top_result = data["results"][0] + if _result_matches(top_result, title, search_title): + return data + + # Then try without year + for search_title in variants: + params = {"query": search_title, "include_adult": "false"} + data = tmdb_api_request("/search/movie", params) + if data and data.get("results"): + top_result = data["results"][0] + if _result_matches(top_result, title, search_title): + return data + + return None + + +def fetch_movie_info(title: str, year: Optional[int] = None) -> Optional[TMDbInfo]: + """Fetch comprehensive movie info from TMDb.""" + data = _search_movie_with_fallbacks(title, year) + + if not data or not data.get("results"): + return None + + result = data["results"][0] + movie_id = result["id"] + + # Fetch full details with credits, similar movies, and keywords + details = fetch_movie_details(movie_id) + if not details: + # Fall back to basic info from search + return TMDbInfo( + tmdb_id=movie_id, + title=result.get("title"), + original_title=result.get("original_title"), + rating=result.get("vote_average"), + vote_count=result.get("vote_count"), + overview=result.get("overview"), + poster_path=result.get("poster_path"), + backdrop_path=result.get("backdrop_path"), + release_date=result.get("release_date"), + ) + + # Extract genres + genres = [g["name"] for g in details.get("genres", [])] + + # Extract keywords + keywords_data = details.get("keywords", {}).get("keywords", []) + keywords = [k["name"] for k in keywords_data] + + # Extract alternative titles (deduplicated) + alt_titles_data = details.get("alternative_titles", {}).get("titles", []) + alt_titles_set = set() + for t in alt_titles_data: + title_str = t.get("title", "").strip() + if title_str: + alt_titles_set.add(title_str) + # Remove the main title and original title to avoid duplicates + main_title = details.get("title", "") + orig_title = details.get("original_title", "") + alt_titles_set.discard(main_title) + alt_titles_set.discard(orig_title) + alternative_titles = sorted(alt_titles_set) if alt_titles_set else None + + # Extract top cast (limit to 10) + credits = details.get("credits", {}) + cast_data = credits.get("cast", [])[:10] + cast = [ + {"name": c["name"], "character": c.get("character", ""), "profile_path": c.get("profile_path")} + for c in cast_data + ] + + # Extract director from crew + crew = credits.get("crew", []) + directors = [c["name"] for c in crew if c.get("job") == "Director"] + director = directors[0] if directors else None + + # Extract similar movies (limit to 10) + similar_data = details.get("similar", {}).get("results", [])[:10] + similar = [ + {"id": s["id"], "title": s["title"], "poster_path": s.get("poster_path")} + for s in similar_data + ] + + return TMDbInfo( + tmdb_id=movie_id, + title=details.get("title"), + original_title=details.get("original_title"), + alternative_titles=alternative_titles, + rating=details.get("vote_average"), + vote_count=details.get("vote_count"), + overview=details.get("overview"), + genres=genres if genres else None, + release_date=details.get("release_date"), + runtime=details.get("runtime"), + status=details.get("status"), + tagline=details.get("tagline"), + poster_path=details.get("poster_path"), + backdrop_path=details.get("backdrop_path"), + similar=similar if similar else None, + keywords=keywords if keywords else None, + cast=cast if cast else None, + director=director, + ) + + +def _search_series_with_fallbacks(title: str) -> Optional[Dict]: + """ + Search for a TV series with progressive title shortening fallbacks. + + PTN often includes extra text in the title at beginning or end. + Results are validated with fuzzy matching to prevent false positives. + """ + words = title.split() + variants = _generate_title_variants(words, min_words=1) + + for search_title in variants: + params = {"query": search_title, "include_adult": "false"} + data = tmdb_api_request("/search/tv", params) + if data and data.get("results"): + # Validate the top result matches our title + top_result = data["results"][0] + tmdb_title = top_result.get("name", "") + if _titles_match(title, tmdb_title, search_title): + return data + + return None + + +def fetch_series_info(title: str) -> Optional[TMDbInfo]: + """Fetch comprehensive TV series info from TMDb.""" + data = _search_series_with_fallbacks(title) + + if not data or not data.get("results"): + return None + + result = data["results"][0] + series_id = result["id"] + + # Fetch full details with credits, similar shows, and keywords + details = fetch_series_details(series_id) + if not details: + # Fall back to basic info from search + return TMDbInfo( + tmdb_id=series_id, + title=result.get("name"), + original_title=result.get("original_name"), + rating=result.get("vote_average"), + vote_count=result.get("vote_count"), + overview=result.get("overview"), + poster_path=result.get("poster_path"), + backdrop_path=result.get("backdrop_path"), + ) + + # Extract genres + genres = [g["name"] for g in details.get("genres", [])] + + # Extract keywords (TV uses "results" instead of "keywords") + keywords_data = details.get("keywords", {}).get("results", []) + keywords = [k["name"] for k in keywords_data] + + # Extract top cast (limit to 10) + credits = details.get("credits", {}) + cast_data = credits.get("cast", [])[:10] + cast = [ + {"name": c["name"], "character": c.get("character", ""), "profile_path": c.get("profile_path")} + for c in cast_data + ] + + # Extract creators + creators = [c["name"] for c in details.get("created_by", [])] + + # Extract networks + networks = [n["name"] for n in details.get("networks", [])] + + # Extract similar series (limit to 10) + similar_data = details.get("similar", {}).get("results", [])[:10] + similar = [ + {"id": s["id"], "title": s["name"], "poster_path": s.get("poster_path")} + for s in similar_data + ] + + # Get first air date + first_air_date = details.get("first_air_date") + + return TMDbInfo( + tmdb_id=series_id, + title=details.get("name"), + original_title=details.get("original_name"), + rating=details.get("vote_average"), + vote_count=details.get("vote_count"), + overview=details.get("overview"), + genres=genres if genres else None, + release_date=first_air_date, + status=details.get("status"), + tagline=details.get("tagline"), + poster_path=details.get("poster_path"), + backdrop_path=details.get("backdrop_path"), + similar=similar if similar else None, + keywords=keywords if keywords else None, + cast=cast if cast else None, + creators=creators if creators else None, + number_of_seasons=details.get("number_of_seasons"), + number_of_episodes=details.get("number_of_episodes"), + networks=networks if networks else None, + ) diff --git a/hivescan/utils.py b/hivescan/utils.py new file mode 100644 index 0000000..830e902 --- /dev/null +++ b/hivescan/utils.py @@ -0,0 +1,174 @@ +"""Utility functions for paths, sizes, and timestamps.""" + +import os +import time +from pathlib import Path +from typing import Optional, List + + +# Default output folder name (created at common root of scanned paths) +DEFAULT_OUTPUT_FOLDER = ".mediahive" + +# Threshold for considering atime "too close" to current time (1 hour) +_ATIME_FRESHNESS_THRESHOLD = 3600 + +# Resolution priority for quality sorting (higher = better) +RESOLUTION_PRIORITY = { + "2160p": 4, "4K": 4, + "1080p": 3, "1080i": 3, + "720p": 2, + "480p": 1, +} + + +def get_added_timestamp(path: Path) -> Optional[int]: + """ + Get the timestamp when a torrent was added to the collection. + + Heuristic: + - For directories: use ctime (most accurate for torrent folder creation) + - For files: use atime unless it's too close to current time (suggesting + the filesystem updates atime on reads), otherwise use max(mtime, ctime) + + Returns: + Unix timestamp as int, or None if path doesn't exist + """ + try: + stat_info = path.stat() + except (OSError, PermissionError): + return None + + if path.is_dir(): + return int(stat_info.st_ctime) + + now = time.time() + atime = stat_info.st_atime + + if now - atime < _ATIME_FRESHNESS_THRESHOLD: + return int(max(stat_info.st_mtime, stat_info.st_ctime)) + + return int(atime) + + +def get_directory_size(path: Path) -> int: + """Calculate total size of a directory recursively.""" + total = 0 + try: + if path.is_file(): + return path.stat().st_size + for item in path.rglob("*"): + if item.is_file(): + total += item.stat().st_size + except (OSError, PermissionError): + pass + return total + + +def format_size(size_bytes: int) -> str: + """Format size in human-readable format.""" + for unit in ["B", "KB", "MB", "GB", "TB"]: + if size_bytes < 1024: + return f"{size_bytes:.2f} {unit}" + size_bytes /= 1024 + return f"{size_bytes:.2f} PB" + + +def find_common_root(paths: List[Path]) -> Optional[Path]: + """ + Find the common root directory for a list of paths. + + Returns None if paths are on different drives/mounts or have no common ancestor. + """ + if not paths: + return None + + # Resolve all paths to absolute + resolved = [p.resolve() for p in paths] + + # Check if all paths are on the same drive (relevant for Windows, but also + # catches cases where paths have completely different roots) + try: + # Get the device for each path + devices = set() + for p in resolved: + # Find the first existing parent to get device info + check_path = p + while not check_path.exists() and check_path.parent != check_path: + check_path = check_path.parent + if check_path.exists(): + devices.add(os.stat(check_path).st_dev) + + if len(devices) > 1: + # Paths are on different devices/drives + return None + except OSError: + pass + + # Find common path prefix + if len(resolved) == 1: + # Single path - use its parent as root + return resolved[0].parent if resolved[0].is_file() else resolved[0] + + # Get parts of each path + all_parts = [p.parts for p in resolved] + + # Find common prefix + common_parts = [] + for parts in zip(*all_parts): + if len(set(parts)) == 1: + common_parts.append(parts[0]) + else: + break + + if not common_parts: + return None + + return Path(*common_parts) + + +def make_relative_path(path: Optional[str], root: Optional[str] = None) -> Optional[str]: + """ + Convert an absolute path to a path relative to the given root. + + If root is None, returns the path unchanged. + """ + if path is None: + return None + if root is None: + return path + root_str = str(root).rstrip("/") + if path.startswith(root_str): + rel = path[len(root_str):] + return rel.lstrip("/") + return path + + +def sanitize_filename(name: str) -> str: + """Sanitize a string for use as a filename.""" + for char in ['/', '\\', ':', '*', '?', '"', '<', '>', '|']: + name = name.replace(char, '_') + name = name.strip('. ') + return name + + +def get_media_folder_name(title: str, year: Optional[int], media_type: str) -> str: + """Get the folder name for a media item.""" + sanitized_title = sanitize_filename(title) + if media_type == "movie" and year: + return f"{sanitized_title} ({year})" + return sanitized_title + + +def get_media_folder_path(title: str, year: Optional[int], media_type: str, cover_dir: Path) -> Path: + """Get the full path to a media item's folder.""" + subdir = "movies" if media_type == "movie" else "series" + folder_name = get_media_folder_name(title, year, media_type) + return cover_dir / subdir / folder_name + + +def sort_by_quality(items: list[dict], reverse: bool = True) -> None: + """Sort items in-place by resolution quality and size.""" + items.sort( + key=lambda v: (RESOLUTION_PRIORITY.get(v.get("resolution", ""), 0), v.get("size", 0) or 0), + reverse=reverse + ) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..31cfec3 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,25 @@ +[project] +name = "torrentmanager" +version = "0.1.0" +description = "Media torrent manager with download scanning and indexing" +readme = "README.md" +requires-python = ">=3.14" +dependencies = [ + "bencodepy>=0.9.5", + "httpx[http2]>=0.28.1", + "parse-torrent-title>=2.8.1", + "tqdm>=4.67.3", +] + +[project.scripts] +hivescan = "hivescan.__main__:main" + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["hivescan"] + +[tool.uv.sources] +parse-torrent-title = { git = "https://github.com/platelminto/parse-torrent-title.git" } diff --git a/rtorrent_client.py b/rtorrent_client.py new file mode 100644 index 0000000..1a6ea45 --- /dev/null +++ b/rtorrent_client.py @@ -0,0 +1,160 @@ +#!/usr/bin/env python3 +""" +RTorrent Client - Communicate with rtorrent via XMLRPC over SCGI socket. +""" + +import socket +import xmlrpc.client +from pathlib import Path +from typing import Dict, List, Optional + + +class SCGITransport(xmlrpc.client.Transport): + """SCGI transport for communicating with rtorrent via Unix socket.""" + + def __init__(self, socket_path: str): + super().__init__() + self.socket_path = socket_path + + def single_request(self, host, handler, request_body, verbose=False): + # Create SCGI request + headers = f"CONTENT_LENGTH\x00{len(request_body)}\x00SCGI\x001\x00" + request = f"{len(headers)}:{headers},{request_body.decode('utf-8')}" + + # Connect to socket + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.connect(self.socket_path) + sock.send(request.encode('utf-8')) + + # Read response + response = b"" + while True: + data = sock.recv(4096) + if not data: + break + response += data + sock.close() + + # Parse response - skip HTTP headers + if b"\r\n\r\n" in response: + response = response.split(b"\r\n\r\n", 1)[1] + + return self.parse_response(response) + + def parse_response(self, response_body): + p, u = xmlrpc.client.getparser() + p.feed(response_body) + p.close() + return u.close() + + +class RTorrentClient: + """Client for communicating with rtorrent via XMLRPC over SCGI socket.""" + + def __init__(self, socket_path: str = "/home/user/rtorrent/.session/rpc.socket"): + self.socket_path = socket_path + transport = SCGITransport(socket_path) + self.proxy = xmlrpc.client.ServerProxy("http://localhost/RPC2", transport=transport) + + def get_loaded_hashes(self) -> set[str]: + """Get set of info hashes for all currently loaded torrents.""" + try: + downloads = self.proxy.download_list("") + return set(h.upper() for h in downloads) + except Exception as e: + print(f"Error getting loaded torrents: {e}") + return set() + + def load_torrent(self, torrent_path: Path, download_dir: Path) -> bool: + """ + Load a torrent file and set its download directory. + Uses load.start_verbose to load and immediately start/hash-check. + + Args: + torrent_path: Path to the .torrent file + download_dir: Directory where the data already exists + + Returns: + True if successful, False otherwise + """ + try: + # load.start_verbose with d.directory.set to specify download location + # This will hash-check existing files instead of re-downloading + self.proxy.load.start_verbose( + "", + str(torrent_path), + f"d.directory.set=\"{download_dir}\"" + ) + return True + except Exception as e: + print(f"Error loading torrent {torrent_path}: {e}") + return False + + def get_torrent_info(self, info_hash: str) -> Optional[Dict]: + """Get info about a loaded torrent.""" + try: + name = self.proxy.d.name(info_hash) + message = self.proxy.d.message(info_hash) + tied_file = self.proxy.d.tied_to_file(info_hash) + directory = self.proxy.d.directory(info_hash) + base_path = self.proxy.d.base_path(info_hash) # Actual data path + is_multi_file = self.proxy.d.is_multi_file(info_hash) + return { + "hash": info_hash, + "name": name, + "message": message, + "tied_file": tied_file, + "directory": directory, + "base_path": base_path, # Full path to data (file or folder) + "is_multi_file": is_multi_file, + } + except Exception as e: + print(f"Error getting torrent info for {info_hash}: {e}") + return None + + def get_unregistered_torrents(self) -> List[Dict]: + """ + Find all torrents with 'unregistered' or 'not registered' tracker errors. + + Returns: + List of torrent info dicts for torrents with registration errors + """ + unregistered = [] + try: + hashes = self.proxy.download_list("") + for info_hash in hashes: + try: + message = self.proxy.d.message(info_hash) + if message and ("unregistered" in message.lower() or + "not registered" in message.lower()): + info = self.get_torrent_info(info_hash) + if info: + unregistered.append(info) + except Exception: + continue + except Exception as e: + print(f"Error scanning for unregistered torrents: {e}") + return unregistered + + def remove_torrent(self, info_hash: str, delete_files: bool = False) -> bool: + """ + Remove a torrent from rtorrent. + + Args: + info_hash: The info hash of the torrent to remove + delete_files: If True, also delete downloaded files (default: False) + + Returns: + True if successful, False otherwise + """ + try: + if delete_files: + # This would delete the data - NOT what we want + self.proxy.d.erase(info_hash) + else: + # Just remove from rtorrent, keep files + self.proxy.d.erase(info_hash) + return True + except Exception as e: + print(f"Error removing torrent {info_hash}: {e}") + return False \ No newline at end of file diff --git a/scripts/rtorrent-manager.py b/scripts/rtorrent-manager.py new file mode 100644 index 0000000..c34bdc7 --- /dev/null +++ b/scripts/rtorrent-manager.py @@ -0,0 +1,439 @@ +#!/usr/bin/env python3 +""" +Torrent Scanner - Scans for .torrent files and analyzes their trackers. +""" + +import argparse +import glob +import hashlib +import shutil +from pathlib import Path +from dataclasses import dataclass +from typing import Iterator + +import bencodepy +from rtorrent_client import RTorrentClient + + +@dataclass +class TorrentInfo: + """Information extracted from a torrent file.""" + path: Path + name: str + trackers: list[str] + size: int | None = None + files: list[str] | None = None + info_hash: str | None = None + is_multi_file: bool = False + + def has_tracker(self, domain: str) -> bool: + """Check if any tracker URL contains the given domain.""" + return any(domain.lower() in tracker.lower() for tracker in self.trackers) + + def get_download_directory(self) -> Path: + """Get the download directory (parent of .torrents folder). + + Assumes .torrent files are in <download_dir>/.torrents/ + so the actual downloads are one level up. + """ + return self.path.parent.parent + + def get_expected_data_path(self) -> Path: + """ + Get the expected path where downloaded data should exist. + + For multi-file torrents: download_dir/torrent_name/ (directory) + For single-file torrents: download_dir/torrent_name (file) + """ + return self.get_download_directory() / self.name + + def verify_download_exists(self) -> tuple[bool, str]: + """ + Verify that the downloaded data exists on disk. + + Returns: + Tuple of (exists: bool, message: str) + """ + expected_path = self.get_expected_data_path() + + if self.is_multi_file: + # Multi-file torrent: expect a directory + if not expected_path.exists(): + return False, f"Directory not found: {expected_path}" + if not expected_path.is_dir(): + return False, f"Expected directory but found file: {expected_path}" + # Optionally check if at least some files exist + existing_files = list(expected_path.rglob("*")) + file_count = sum(1 for f in existing_files if f.is_file()) + if file_count == 0: + return False, f"Directory exists but is empty: {expected_path}" + return True, f"Directory exists with {file_count} files" + else: + # Single-file torrent: expect a file + if not expected_path.exists(): + return False, f"File not found: {expected_path}" + if expected_path.is_dir(): + return False, f"Expected file but found directory: {expected_path}" + return True, f"File exists: {expected_path}" + + +def parse_torrent(filepath: Path) -> TorrentInfo | None: + """ + Parse a .torrent file and extract relevant information. + + Args: + filepath: Path to the .torrent file + + Returns: + TorrentInfo object or None if parsing fails + """ + try: + with open(filepath, 'rb') as f: + data = bencodepy.decode(f.read()) + except Exception as e: + print(f"Error parsing {filepath}: {e}") + return None + + # Extract trackers + trackers = [] + + # Main announce URL + if b'announce' in data: + announce = data[b'announce'] + if isinstance(announce, bytes): + trackers.append(announce.decode('utf-8', errors='replace')) + + # Announce list (multiple trackers) + if b'announce-list' in data: + for tier in data[b'announce-list']: + for tracker in tier: + if isinstance(tracker, bytes): + url = tracker.decode('utf-8', errors='replace') + if url not in trackers: + trackers.append(url) + + # Extract name + info = data.get(b'info', {}) + name = info.get(b'name', b'Unknown').decode('utf-8', errors='replace') + + # Calculate info hash + info_hash = hashlib.sha1(bencodepy.encode(info)).hexdigest().upper() + + # Extract size and files + size = None + files = None + is_multi_file = False + + if b'length' in info: + # Single file torrent + size = info[b'length'] + files = [name] + is_multi_file = False + elif b'files' in info: + # Multi-file torrent + files = [] + size = 0 + is_multi_file = True + for file_info in info[b'files']: + file_path = '/'.join( + p.decode('utf-8', errors='replace') + for p in file_info.get(b'path', []) + ) + files.append(file_path) + size += file_info.get(b'length', 0) + + return TorrentInfo( + path=filepath, + name=name, + trackers=trackers, + size=size, + files=files, + info_hash=info_hash, + is_multi_file=is_multi_file, + ) + + +def scan_torrent_directories(paths: list[str]) -> Iterator[Path]: + """ + Scan directories for .torrent files. + + Args: + paths: List of directory paths or glob patterns to scan + + Yields: + Path objects for each .torrent file found + """ + for pattern in paths: + for dir_path in glob.glob(pattern): + torrent_dir = Path(dir_path) + if torrent_dir.is_dir(): + for torrent_file in torrent_dir.glob("*.torrent"): + yield torrent_file + + +def find_torrents_with_tracker(tracker_domain: str, + paths: list[str]) -> list[TorrentInfo]: + """ + Find all torrents that have a specific tracker domain. + + Args: + tracker_domain: Domain to search for in tracker URLs (e.g., "hdbits.org") + paths: List of directory paths or glob patterns to scan + + Returns: + List of TorrentInfo objects for matching torrents + """ + matching_torrents = [] + + for torrent_path in scan_torrent_directories(paths): + info = parse_torrent(torrent_path) + if info and info.has_tracker(tracker_domain): + matching_torrents.append(info) + + return matching_torrents + + +def format_size(size_bytes: int | None) -> str: + """Format bytes as human-readable size.""" + if size_bytes is None: + return "Unknown" + + for unit in ['B', 'KB', 'MB', 'GB', 'TB']: + if size_bytes < 1024: + return f"{size_bytes:.2f} {unit}" + size_bytes /= 1024 + return f"{size_bytes:.2f} PB" + + +def main(): + """Main entry point for the torrent scanner.""" + parser = argparse.ArgumentParser( + description="Scan and manage torrent files", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + %(prog)s /path/to/torrents*/.torrents/ + %(prog)s /mnt/disk1/torrents/.torrents/ /mnt/disk2/torrents/.torrents/ + %(prog)s /torrents*/.torrents/ --tracker hdbits.org + %(prog)s /torrents*/.torrents/ --dry + """, + ) + parser.add_argument( + "paths", + nargs="+", + help="Directories or glob patterns containing .torrent files", + ) + parser.add_argument("--dry", action="store_true", help="Dry run - show what would be done without making changes") + parser.add_argument("--tracker", default="hdbits.org", + help="Tracker domain to filter by (default: hdbits.org)") + args = parser.parse_args() + + dry_run = args.dry + tracker_domain = args.tracker + + # Expand glob patterns + expanded_paths = [] + for pattern in args.paths: + matches = glob.glob(pattern) + if matches: + expanded_paths.extend(matches) + else: + expanded_paths.append(pattern) + + if dry_run: + print("=" * 60) + print("DRY RUN MODE - No changes will be made") + print("=" * 60) + + print(f"Scanning for torrents...") + print(f"Search paths: {expanded_paths}") + print("-" * 60) + + # Parse all torrents + all_torrents: list[TorrentInfo] = [] + for torrent_path in scan_torrent_directories(expanded_paths): + info = parse_torrent(torrent_path) + if info: + all_torrents.append(info) + + # Separate by tracker + with_hdbits = [t for t in all_torrents if t.has_tracker(tracker_domain)] + without_hdbits = [t for t in all_torrents if not t.has_tracker(tracker_domain)] + + # Print stats + print(f"\n{'='*60}") + print(f"SUMMARY") + print(f"{'='*60}") + print(f"Total torrents scanned: {len(all_torrents)}") + print(f"With {tracker_domain}: {len(with_hdbits)}") + print(f"Without {tracker_domain}: {len(without_hdbits)}") + + # Add hdbits torrents to rtorrent + if with_hdbits: + print(f"\n{'='*60}") + print(f"VERIFYING DOWNLOADS & ADDING TO RTORRENT") + print(f"{'='*60}") + + # First, verify which torrents have their data + verified = [] + missing_data = [] + + for torrent in with_hdbits: + exists, message = torrent.verify_download_exists() + if exists: + verified.append(torrent) + else: + missing_data.append((torrent, message)) + + print(f"\nVerification results:") + print(f" Downloads found: {len(verified)}") + print(f" Downloads missing: {len(missing_data)}") + + # Report missing downloads + if missing_data: + print(f"\n{'='*60}") + print(f"TORRENTS WITH MISSING DATA (will not add)") + print(f"{'='*60}") + for torrent, message in missing_data: + print(f"\n Name: {torrent.name}") + print(f" Torrent: {torrent.path}") + print(f" Reason: {message}") + + # Now add verified torrents to rtorrent + if verified: + print(f"\n{'='*60}") + print(f"ADDING {len(verified)} VERIFIED TORRENTS TO RTORRENT") + print(f"{'='*60}") + + client = RTorrentClient() + loaded_hashes = client.get_loaded_hashes() + print(f"Currently loaded in rtorrent: {len(loaded_hashes)} torrents") + + added = 0 + skipped = 0 + failed = 0 + + for torrent in verified: + if torrent.info_hash and torrent.info_hash in loaded_hashes: + print(f"Skipping (already loaded): {torrent.name}") + skipped += 1 + else: + download_dir = torrent.get_download_directory() + if dry_run: + print(f"Would add: {torrent.name}") + print(f" Download dir: {download_dir}") + added += 1 + else: + print(f"Adding: {torrent.name}") + print(f" Download dir: {download_dir}") + if client.load_torrent(torrent.path, download_dir): + added += 1 + else: + failed += 1 + + if dry_run: + print(f"\nDry run: {added} would be added, {skipped} already loaded") + else: + print(f"\nRtorrent results: {added} added, {skipped} skipped, {failed} failed") + + # Clean up unregistered torrents from rtorrent + print(f"\n{'='*60}") + print(f"CHECKING FOR UNREGISTERED TORRENTS") + print(f"{'='*60}") + + client = RTorrentClient() + unregistered = client.get_unregistered_torrents() + + if unregistered: + print(f"Found {len(unregistered)} unregistered torrent(s):\n") + + removed_from_rtorrent = 0 + removed_torrent_files = 0 + removed_downloads = 0 + + for torrent_info in unregistered: + # Determine the download path (base_path is the actual file/folder) + download_path = Path(torrent_info['base_path']) if torrent_info['base_path'] else None + + if dry_run: + status = "[DRY]" + if download_path: + print(f" {status} {download_path}") + else: + print(f" {status} {torrent_info['name']} (no data path)") + else: + # Remove from rtorrent (keeps downloaded files) + if client.remove_torrent(torrent_info['hash']): + removed_from_rtorrent += 1 + + # Delete the .torrent file if it exists + tied_file = torrent_info['tied_file'] + if tied_file: + torrent_file = Path(tied_file) + if torrent_file.exists(): + try: + torrent_file.unlink() + removed_torrent_files += 1 + except Exception: + pass + + # Delete the downloaded files + if download_path and download_path.exists(): + try: + if download_path.is_dir(): + shutil.rmtree(download_path) + else: + download_path.unlink() + removed_downloads += 1 + print(f" [DEL] {download_path}") + except Exception as e: + print(f" [ERR] {download_path}: {e}") + else: + print(f" [DEL] {torrent_info['name']} (no data)") + else: + print(f" [ERR] {torrent_info['name']}: failed to remove from rtorrent") + + print() + if dry_run: + print(f"Dry run: {len(unregistered)} would be removed (rtorrent + .torrent + downloads)") + else: + print(f"Cleanup: {removed_from_rtorrent} from rtorrent, {removed_torrent_files} .torrents, {removed_downloads} downloads") + else: + print("No unregistered torrents found.") + + # List torrents without hdbits.org + if without_hdbits: + print(f"\n{'='*60}") + print(f"TORRENTS WITHOUT {tracker_domain.upper()}") + print(f"{'='*60}") + for torrent in without_hdbits: + print(f"\nName: {torrent.name}") + print(f"Path: {torrent.path}") + print(f"Size: {format_size(torrent.size)}") + if torrent.trackers: + print(f"Trackers:") + for tracker in torrent.trackers: + print(f" - {tracker}") + else: + print("Trackers: (none)") + + # Remove the non-hdbits torrent files + print(f"\n{'='*60}") + if dry_run: + print(f"WOULD REMOVE {len(without_hdbits)} TORRENT FILE(S)") + print(f"{'='*60}") + for torrent in without_hdbits: + print(f"Would remove: {torrent.path}") + else: + print(f"REMOVING {len(without_hdbits)} TORRENT FILE(S)") + print(f"{'='*60}") + for torrent in without_hdbits: + try: + torrent.path.unlink() + print(f"Removed: {torrent.path}") + except Exception as e: + print(f"Failed to remove {torrent.path}: {e}") + + +if __name__ == "__main__": + main()