From 0e5251ab9111e838d80f762003d2886cc670339c Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Mon, 25 May 2026 01:57:17 +0000 Subject: [PATCH] Ruff automatic fixes and formatting. --- mediahive/hivescan/__init__.py | 11 +- mediahive/hivescan/images.py | 22 ++-- mediahive/hivescan/indexer.py | 196 +++++++++++++++--------------- mediahive/hivescan/models.py | 27 ++-- mediahive/hivescan/parsing.py | 10 +- mediahive/hivescan/scanignore.py | 19 ++- mediahive/hivescan/scanner.py | 75 +++++++----- mediahive/hivescan/scanning.py | 55 ++++----- mediahive/hivescan/showreel.py | 91 +++++++------- mediahive/hivescan/tmdb_client.py | 72 +++++------ mediahive/hivescan/utils.py | 27 ++-- mediahive/index_store.py | 13 +- mediahive/models/data.py | 3 +- mediahive/models/events.py | 3 +- mediahive/models/protocol.py | 1 - mediahive/models/tmdb.py | 4 +- mediahive/root_registry.py | 46 ++++--- mediahive/server.py | 17 ++- mediahive/winmain.py | 37 +++--- rtorrent_client.py | 21 ++-- scripts/fastapi-vue/devutil.py | 7 +- scripts/guibuild.py | 4 +- scripts/release.py | 4 +- scripts/rtorrent-manager.py | 100 +++++++-------- 24 files changed, 417 insertions(+), 448 deletions(-) diff --git a/mediahive/hivescan/__init__.py b/mediahive/hivescan/__init__.py index ed05e1f..a187073 100644 --- a/mediahive/hivescan/__init__.py +++ b/mediahive/hivescan/__init__.py @@ -1,5 +1,4 @@ -""" -Hivescan - Continuous media scanning with live WebSocket updates. +"""Hivescan - Continuous media scanning with live WebSocket updates. Import from submodules directly: from mediahive.hivescan.scanner import start, stop @@ -9,13 +8,13 @@ Import from submodules directly: """ # Minimal public API - prefer importing from submodules directly -from mediahive.hivescan.models import ContentType, ContentHash, ParsedContent +from mediahive.hivescan.models import ContentHash, ContentType, ParsedContent from mediahive.hivescan.utils import DEFAULT_OUTPUT_FOLDER, find_common_root __all__ = [ - "ContentType", - "ContentHash", - "ParsedContent", "DEFAULT_OUTPUT_FOLDER", + "ContentHash", + "ContentType", + "ParsedContent", "find_common_root", ] diff --git a/mediahive/hivescan/images.py b/mediahive/hivescan/images.py index 6bebb6e..1f0aed4 100644 --- a/mediahive/hivescan/images.py +++ b/mediahive/hivescan/images.py @@ -1,14 +1,12 @@ """TMDb image downloading functions.""" -import httpx from pathlib import Path -from typing import Optional +import httpx from aiopathlib import AsyncPath from mediahive.hivescan.utils import get_media_folder_path, sanitize_filename - # TMDb image configuration TMDB_IMAGE_BASE = "https://image.tmdb.org/t/p" DEFAULT_POSTER_SIZE = "w500" @@ -16,7 +14,7 @@ DEFAULT_BACKDROP_SIZE = "w1280" DEFAULT_PROFILE_SIZE = "w185" # Shared async HTTP client (created lazily) -_image_client: Optional[httpx.AsyncClient] = None +_image_client: httpx.AsyncClient | None = None def _get_image_client() -> httpx.AsyncClient: @@ -31,9 +29,7 @@ def _get_image_client() -> httpx.AsyncClient: return _image_client -async def _download_image( - url: str, output_path: Path, description: str -) -> Optional[str]: +async def _download_image(url: str, output_path: Path, description: str) -> str | None: """Download an image from URL to output path.""" ap = AsyncPath(output_path) if await ap.exists(): @@ -54,11 +50,11 @@ async def _download_image( async def download_cover_image( poster_path: str, title: str, - year: Optional[int], + year: int | None, media_type: str, cover_dir: Path, size: str = DEFAULT_POSTER_SIZE, -) -> Optional[str]: +) -> str | None: """Download a cover image from TMDb.""" if not poster_path: return None @@ -77,11 +73,11 @@ async def download_cover_image( async def download_backdrop_image( backdrop_path: str, title: str, - year: Optional[int], + year: int | None, media_type: str, cover_dir: Path, size: str = DEFAULT_BACKDROP_SIZE, -) -> Optional[str]: +) -> str | None: """Download a backdrop image from TMDb.""" if not backdrop_path: return None @@ -101,7 +97,7 @@ async def download_season_poster( poster_path: str, media_folder: Path, season_num: int, -) -> Optional[str]: +) -> str | None: """Download a season poster image from TMDb.""" if not poster_path: return None @@ -122,7 +118,7 @@ async def download_cast_profile( cast_name: str, cast_index: int, size: str = DEFAULT_PROFILE_SIZE, -) -> Optional[str]: +) -> str | None: """Download a cached cast profile image from TMDb.""" if not profile_path: return None diff --git a/mediahive/hivescan/indexer.py b/mediahive/hivescan/indexer.py index 03effe6..b5d8836 100644 --- a/mediahive/hivescan/indexer.py +++ b/mediahive/hivescan/indexer.py @@ -3,8 +3,8 @@ import asyncio import hashlib import logging +from collections.abc import AsyncIterator from pathlib import Path -from typing import AsyncIterator, Dict, List, Optional, Tuple from mediahive.hivescan.images import ( download_backdrop_image, @@ -52,7 +52,7 @@ logger = logging.getLogger("hivescan.indexer") async def _build_torrent_info( - item: ParsedContent, media_root: Optional[str] = None + item: ParsedContent, media_root: str | None = None ) -> Torrent: """Build torrent info for a single torrent.""" playable_file = await find_playable_file(item.path) @@ -86,10 +86,10 @@ async def _build_torrent_info( async def _cache_cast_profiles( - info: Optional[Info], + info: Info | None, media_folder: Path, - media_root: Optional[str] = None, -) -> Optional[Info]: + media_root: str | None = None, +) -> Info | None: """Replace TMDb cast profile paths with cached local image paths.""" if not info or not info.cast: return info @@ -121,15 +121,14 @@ async def _cache_cast_profiles( async def _collect_episode_files( - items: List[ParsedContent], -) -> Dict[Tuple[int, int], List[Dict]]: - """ - Collect all episode files from a list of torrent items. + 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]] = {} - probe_cache: Dict[str, object] = {} + all_episode_files: dict[tuple[int, int], list[dict]] = {} + probe_cache: dict[str, object] = {} async def get_probe(path: str): cached = probe_cache.get(path) @@ -148,26 +147,24 @@ async def _collect_episode_files( all_episode_files[key] = [] for file_path, file_size in files: probe = await get_probe(file_path) - all_episode_files[key].append( - { - "path": file_path, - "size": file_size, - "probed_resolution": probe.resolution, - "audio_languages": probe.audio_languages, - "subtitle_languages": probe.subtitle_languages, - "is_hdr": probe.is_hdr, - "has_dolby_vision": probe.has_dolby_vision, - "has_dolby_atmos": probe.has_dolby_atmos, - "resolution": item.resolution, - "quality": item.quality, - "network": item.network, - "codec": item.codec, - "audio": item.audio, - "encoder": item.encoder, - "torrent_path": item.path.as_posix(), - "torrent_title": item.title, - } - ) + all_episode_files[key].append({ + "path": file_path, + "size": file_size, + "probed_resolution": probe.resolution, + "audio_languages": probe.audio_languages, + "subtitle_languages": probe.subtitle_languages, + "is_hdr": probe.is_hdr, + "has_dolby_vision": probe.has_dolby_vision, + "has_dolby_atmos": probe.has_dolby_atmos, + "resolution": item.resolution, + "quality": item.quality, + "network": item.network, + "codec": item.codec, + "audio": item.audio, + "encoder": item.encoder, + "torrent_path": item.path.as_posix(), + "torrent_title": item.title, + }) # Handle individual episodes from PTN parsing if item.episode is not None and item.season is not None: @@ -196,40 +193,38 @@ async def _collect_episode_files( item.content_hash.path ) size = item.content_hash.size if item.content_hash else 0 - all_episode_files[key].append( - { - "path": playable, - "size": size, - "probed_resolution": probe.resolution, - "audio_languages": probe.audio_languages, - "subtitle_languages": probe.subtitle_languages, - "is_hdr": probe.is_hdr, - "has_dolby_vision": probe.has_dolby_vision, - "has_dolby_atmos": probe.has_dolby_atmos, - "resolution": item.resolution, - "quality": item.quality, - "network": item.network, - "codec": item.codec, - "audio": item.audio, - "encoder": item.encoder, - "torrent_path": item.path.as_posix(), - "torrent_title": item.title, - } - ) + all_episode_files[key].append({ + "path": playable, + "size": size, + "probed_resolution": probe.resolution, + "audio_languages": probe.audio_languages, + "subtitle_languages": probe.subtitle_languages, + "is_hdr": probe.is_hdr, + "has_dolby_vision": probe.has_dolby_vision, + "has_dolby_atmos": probe.has_dolby_atmos, + "resolution": item.resolution, + "quality": item.quality, + "network": item.network, + "codec": item.codec, + "audio": item.audio, + "encoder": item.encoder, + "torrent_path": item.path.as_posix(), + "torrent_title": item.title, + }) return all_episode_files def _build_episodes_data( - episodes_in_season: Dict[int, List[Dict]], - tmdb_episodes: Dict[int, EpisodeInfo], + episodes_in_season: dict[int, list[dict]], + tmdb_episodes: dict[int, EpisodeInfo], series_folder: Path, season_num: int, generate_showreels: bool, - episode_reel_tasks: List, + episode_reel_tasks: list, series_title: str, - media_root: Optional[str] = None, -) -> List[Episode]: + media_root: str | None = None, +) -> list[Episode]: """Build episode data list for a season.""" episodes_data = [] @@ -256,9 +251,13 @@ def _build_episodes_data( 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) - ) + episode_reel_tasks.append(( + best_file, + series_folder, + season_num, + episode_num, + series_title, + )) torrents = {} for f in episode_files: @@ -290,7 +289,7 @@ def _build_episodes_data( rating=tmdb_ep.vote_average if tmdb_ep else None, director=tmdb_ep.director if tmdb_ep else None, reel_image=reel_path, - reel_sources=reel_sources if reel_sources else None, + reel_sources=reel_sources or None, torrents=torrents, ) episodes_data.append(episode_data) @@ -299,19 +298,19 @@ def _build_episodes_data( async def _build_seasons_data( - all_episode_files: Dict[Tuple[int, int], List[Dict]], - tmdb_id: Optional[int], + all_episode_files: dict[tuple[int, int], list[dict]], + tmdb_id: int | None, 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[Season]: + season_cache: dict, + episode_reel_tasks: list, + media_root: str | None = None, +) -> list[Season]: """Build seasons data structure for a series.""" # Group episodes by season - seasons_map: Dict[int, Dict[int, List[Dict]]] = {} + 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] = {} @@ -323,7 +322,7 @@ async def _build_seasons_data( # Fetch TMDb season details if we have a TMDb ID tmdb_season = None - tmdb_episodes: Dict[int, EpisodeInfo] = {} + tmdb_episodes: dict[int, EpisodeInfo] = {} if tmdb_id: cache_key = (tmdb_id, season_num) @@ -379,18 +378,17 @@ async def _process_movies( cover_dir: Path, fetch_covers: bool, generate_showreels: bool, - media_root: Optional[str] = None, - root_id: Optional[str] = None, -) -> AsyncIterator[Tuple[Movie, Optional[Tuple[str, Path, str]]]]: - """ - Async generator that processes all movies. + media_root: str | None = None, + root_id: str | None = None, +) -> AsyncIterator[tuple[Movie, tuple[str, Path, str] | None]]: + """Async generator that processes all movies. Yields (Movie, showreel_task_or_None) for each movie as it is processed. """ # In-memory cache for TMDb lookups - movie_tmdb_cache: Dict[str, Optional[Info]] = {} + movie_tmdb_cache: dict[str, Info | None] = {} - async def get_movie_tmdb(title: str, year: Optional[int]) -> Optional[Info]: + async def get_movie_tmdb(title: str, year: int | None) -> Info | None: cache_key = f"{title.lower()}:{year}" if cache_key in movie_tmdb_cache: return movie_tmdb_cache[cache_key] @@ -413,7 +411,7 @@ async def _process_movies( ) # Group by title+year - movie_groups: Dict[str, List[ParsedContent]] = {} + 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: @@ -421,8 +419,8 @@ async def _process_movies( movie_groups[key].append(item) # Re-group by TMDb ID - tmdb_movie_groups: Dict[int, Dict] = {} - no_tmdb_movie_groups: Dict[str, Dict] = {} + tmdb_movie_groups: dict[int, dict] = {} + no_tmdb_movie_groups: dict[str, dict] = {} if movie_groups: logger.info( @@ -550,8 +548,8 @@ async def _process_movies( 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, - showreel_source_sets=showreel_source_sets if showreel_source_sets else None, + showreel_images=showreel_paths or None, + showreel_source_sets=showreel_source_sets or None, torrents=torrents, root_id=root_id, ) @@ -593,9 +591,10 @@ async def _process_movies( ), ) best_version = torrents[best_relpath] - if best_version.playable_file and not best_version.playable_file.endswith( - (".bdmv", ".ifo") - ): + if best_version.playable_file and not best_version.playable_file.endswith(( + ".bdmv", + ".ifo", + )): abs_playable = ( (Path(media_root) / best_version.playable_file).as_posix() if media_root @@ -621,8 +620,8 @@ async def _process_movies( year=year, newest=newest, cover_path=make_relative_path(cover_path, media_root), - showreel_images=showreel_paths if showreel_paths else None, - showreel_source_sets=showreel_source_sets if showreel_source_sets else None, + showreel_images=showreel_paths or None, + showreel_source_sets=showreel_source_sets or None, torrents=torrents, root_id=root_id, ) @@ -634,19 +633,18 @@ async def _process_series( cover_dir: Path, fetch_covers: bool, generate_showreels: bool, - media_root: Optional[str] = None, - root_id: Optional[str] = None, -) -> AsyncIterator[Tuple[Series, List[Tuple[str, Path, int, int, str]]]]: - """ - Async generator that processes all series. + media_root: str | None = None, + root_id: str | None = None, +) -> AsyncIterator[tuple[Series, list[tuple[str, Path, int, int, str]]]]: + """Async generator that processes all series. Yields (Series, episode_reel_tasks) for each series as it is processed. """ # In-memory cache for TMDb lookups - series_tmdb_cache: Dict[str, Optional[Info]] = {} - season_cache: Dict[Tuple[int, int], Optional[SeasonInfo]] = {} + series_tmdb_cache: dict[str, Info | None] = {} + season_cache: dict[tuple[int, int], SeasonInfo | None] = {} - async def get_series_tmdb(title: str) -> Optional[Info]: + async def get_series_tmdb(title: str) -> Info | None: cache_key = title.lower() if cache_key in series_tmdb_cache: return series_tmdb_cache[cache_key] @@ -671,7 +669,7 @@ async def _process_series( ) # Group by title - series_groups: Dict[str, List[ParsedContent]] = {} + series_groups: dict[str, list[ParsedContent]] = {} for item in valid_series: key = item.title.lower() if key not in series_groups: @@ -679,8 +677,8 @@ async def _process_series( series_groups[key].append(item) # Re-group by TMDb ID - tmdb_groups: Dict[int, Dict] = {} - no_tmdb_groups: Dict[str, Dict] = {} + tmdb_groups: dict[int, dict] = {} + no_tmdb_groups: dict[str, dict] = {} if series_groups: logger.info( @@ -753,7 +751,7 @@ async def _process_series( # Collect and build episode data all_episode_files = await _collect_episode_files(items) - ep_reel_tasks: List[Tuple[str, Path, int, int, str]] = [] + ep_reel_tasks: list[tuple[str, Path, int, int, str]] = [] seasons_data = await _build_seasons_data( all_episode_files, tmdb_id, @@ -781,7 +779,7 @@ async def _process_series( id=series_id, title=display_title, info=tmdb_info, - alternative_titles=different_titles if different_titles else None, + alternative_titles=different_titles or None, newest=newest, cover_path=make_relative_path(cover_path, media_root), backdrop_path=make_relative_path(backdrop_path, media_root), @@ -805,7 +803,7 @@ async def _process_series( series_folder = get_media_folder_path(title, None, "series", cover_dir) all_episode_files = await _collect_episode_files(items) - ep_reel_tasks: List[Tuple[str, Path, int, int, str]] = [] + ep_reel_tasks: list[tuple[str, Path, int, int, str]] = [] seasons_data = await _build_seasons_data( all_episode_files, None, diff --git a/mediahive/hivescan/models.py b/mediahive/hivescan/models.py index a3140aa..ad1ffc3 100644 --- a/mediahive/hivescan/models.py +++ b/mediahive/hivescan/models.py @@ -4,7 +4,6 @@ import hashlib from dataclasses import dataclass, field from enum import Enum from pathlib import Path -from typing import Optional class ContentType(Enum): @@ -24,7 +23,7 @@ class ContentHash: size: int = 0 @classmethod - def from_path(cls, path: Path) -> "ContentHash": + 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) @@ -38,17 +37,17 @@ class ParsedContent: name: str content_type: ContentType title: str - year: Optional[int] = None - resolution: Optional[str] = None - quality: Optional[str] = None - network: 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 + year: int | None = None + resolution: str | None = None + quality: str | None = None + network: str | None = None + codec: str | None = None + audio: str | None = None + season: int | None = None + episode: int | None = None + episode_name: str | None = None + encoder: str | None = None + language: str | None = None is_directory: bool = False raw_parsed: dict = field(default_factory=dict) - content_hash: Optional[ContentHash] = None + content_hash: ContentHash | None = None diff --git a/mediahive/hivescan/parsing.py b/mediahive/hivescan/parsing.py index e71f8d1..8e495b0 100644 --- a/mediahive/hivescan/parsing.py +++ b/mediahive/hivescan/parsing.py @@ -2,7 +2,6 @@ import re from pathlib import Path -from typing import Optional, Tuple import PTN from aiopathlib import AsyncPath @@ -10,11 +9,10 @@ from aiopathlib import AsyncPath from mediahive.hivescan.models import ContentHash, ContentType, ParsedContent from mediahive.hivescan.utils import normalize_resolution_label - _EDGE_NON_ALPHANUMERICS_RE = re.compile(r"^[^0-9A-Za-z]+|[^0-9A-Za-z]+$") -def strip_edge_non_alphanumerics(value: Optional[str]) -> Optional[str]: +def strip_edge_non_alphanumerics(value: str | None) -> str | None: """Remove punctuation from the start and end of PTN scene tags.""" if not value: return None @@ -64,14 +62,14 @@ async def parse_download(path: Path) -> ParsedContent: ) -def parse_episode_from_filename(filename: str) -> Optional[Tuple[int, int]]: - """ - Parse season and episode numbers from a filename. +def parse_episode_from_filename(filename: str) -> tuple[int, int] | None: + """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() diff --git a/mediahive/hivescan/scanignore.py b/mediahive/hivescan/scanignore.py index bb1d4ce..ce69751 100644 --- a/mediahive/hivescan/scanignore.py +++ b/mediahive/hivescan/scanignore.py @@ -1,5 +1,4 @@ -""" -Gitignore-style path matcher for controlling which directories the scanner visits. +"""Gitignore-style path matcher for controlling which directories the scanner visits. Reads patterns from ``/.mediahive/scanignore``. The file uses the same syntax as ``.gitignore``: @@ -30,8 +29,6 @@ from __future__ import annotations import re from pathlib import Path -from typing import List, Tuple - # Built-in patterns that are always excluded (before user file) _BUILTIN_EXCLUDES: list[str] = [ @@ -75,13 +72,11 @@ def _pattern_to_regex(pattern: str) -> re.Pattern[str]: parts.append("(?:.+/)?") i += 3 continue - else: - parts.append(".*") - i += 2 - continue - else: - parts.append("[^/]*") - i += 1 + parts.append(".*") + i += 2 + continue + parts.append("[^/]*") + i += 1 elif c == "?": parts.append("[^/]") i += 1 @@ -112,7 +107,7 @@ class ScanIgnore: def __init__(self, media_root: Path) -> None: self.media_root = media_root.resolve() - self._rules: List[Tuple[bool, re.Pattern[str]]] = [] # (negated, regex) + self._rules: list[tuple[bool, re.Pattern[str]]] = [] # (negated, regex) self._load_builtins() self._load_file() diff --git a/mediahive/hivescan/scanner.py b/mediahive/hivescan/scanner.py index 486a125..8d0527c 100644 --- a/mediahive/hivescan/scanner.py +++ b/mediahive/hivescan/scanner.py @@ -1,5 +1,4 @@ -""" -Scan orchestration — background tasks for continuous media scanning. +"""Scan orchestration — background tasks for continuous media scanning. All scanning logic lives here in hivescan. Communication with the mediahive server happens exclusively through an async ``send`` callable that pushes @@ -14,10 +13,9 @@ from __future__ import annotations import asyncio import logging -import os import uuid +from collections.abc import Awaitable, Callable from pathlib import Path -from typing import Awaitable, Callable, List, Optional from aiopathlib import AsyncPath @@ -69,10 +67,10 @@ class RootScanner: self._scanignore = ScanIgnore(media_root) # Runtime state - self._scan_task: Optional[asyncio.Task] = None + self._scan_task: asyncio.Task | None = None self._showreel_queue: asyncio.Queue = asyncio.Queue() - self._showreel_worker_task: Optional[asyncio.Task] = None - self._rescan_worker_task: Optional[asyncio.Task] = None + self._showreel_worker_task: asyncio.Task | None = None + self._rescan_worker_task: asyncio.Task | None = None self._seen_mtimes: dict[str, int] = {} # ------------------------------------------------------------------ @@ -96,15 +94,23 @@ class RootScanner: async def stop(self) -> None: """Cancel all background tasks.""" - for task in (self._scan_task, self._showreel_worker_task, self._rescan_worker_task): + for task in ( + self._scan_task, + self._showreel_worker_task, + self._rescan_worker_task, + ): if task and not task.done(): task.cancel() # Wait briefly for graceful shutdown - for task in (self._scan_task, self._showreel_worker_task, self._rescan_worker_task): + for task in ( + self._scan_task, + self._showreel_worker_task, + self._rescan_worker_task, + ): if task and not task.done(): try: await asyncio.wait_for(task, timeout=2.0) - except (asyncio.TimeoutError, asyncio.CancelledError): + except TimeoutError, asyncio.CancelledError: pass def is_scanning(self) -> bool: @@ -139,9 +145,9 @@ class RootScanner: except Exception: logger.exception("Rescan loop error") - async def _discover_downloads(self, task_id: str) -> List[ParsedContent]: + async def _discover_downloads(self, task_id: str) -> list[ParsedContent]: """Recursively walk the media root, respecting scanignore rules.""" - downloads: List[ParsedContent] = [] + downloads: list[ParsedContent] = [] media_root_str = self.media_root.as_posix() dirs_visited = 0 @@ -196,7 +202,7 @@ class RootScanner: child_dirs.append(item) else: child_files.append(item) - except (OSError, PermissionError): + except OSError, PermissionError: logger.debug("Cannot list directory: %s", directory) return @@ -207,14 +213,19 @@ class RootScanner: mtime = int(stat_info.st_mtime) except OSError: return - if relpath not in self._seen_mtimes or self._seen_mtimes[relpath] != mtime: + if ( + relpath not in self._seen_mtimes + or self._seen_mtimes[relpath] != mtime + ): self._seen_mtimes[relpath] = mtime downloads.append(await parse_download(directory)) return if child_dirs: dirs_visited += 1 - rel = make_relative_path(str(directory), media_root_str) or str(directory) + rel = make_relative_path(str(directory), media_root_str) or str( + directory + ) if dirs_visited % 5 == 1: await _report(f"Scanning: {rel} ({len(downloads)} found)") logger.info("Scanning: %s (%d found so far)", rel, len(downloads)) @@ -229,7 +240,10 @@ class RootScanner: mtime = int(stat_info.st_mtime) except OSError: continue - if relpath in self._seen_mtimes and self._seen_mtimes[relpath] == mtime: + if ( + relpath in self._seen_mtimes + and self._seen_mtimes[relpath] == mtime + ): continue self._seen_mtimes[relpath] = mtime downloads.append(await parse_download(child_file)) @@ -251,7 +265,7 @@ class RootScanner: root_ap = AsyncPath(self.media_root) try: root_children = await asyncio.to_thread(lambda: list(root_ap.iterdir())) - except (OSError, PermissionError): + except OSError, PermissionError: logger.error("Cannot list media root: %s", self.media_root) return downloads @@ -284,12 +298,11 @@ class RootScanner: return downloads async def _run_scan(self) -> None: - """ - Full scan pipeline: - 1. Discover downloads - 2. Categorise → movies / series - 3. Iterate async generators, send each item as Upsert - 4. Queue showreel tasks + """Full scan pipeline: + 1. Discover downloads + 2. Categorise → movies / series + 3. Iterate async generators, send each item as Upsert + 4. Queue showreel tasks """ task_id = f"scan-{uuid.uuid4().hex[:8]}" media_root_str = self.media_root.as_posix() @@ -538,8 +551,8 @@ class RootScanner: media_folder, media_root=media_root_path ) paths = [sources[0] for sources in source_sets if sources] - movie.showreel_images = paths if paths else None - movie.showreel_source_sets = source_sets if source_sets else None + movie.showreel_images = paths or None + movie.showreel_source_sets = source_sets or None await self._send(Upsert(kind="movie", item=movie)) await self._send( Task( @@ -552,7 +565,9 @@ class RootScanner: ) ) else: - logger.warning("Showreel generation returned nothing: %s", title) + logger.warning( + "Showreel generation returned nothing: %s", title + ) await self._send( Task( data=TaskInfo( @@ -620,9 +635,7 @@ class RootScanner: media_root_str, ) ) - episode.reel_sources = ( - reel_sources if reel_sources else None - ) + episode.reel_sources = reel_sources or None await self._send(Upsert(kind="series", item=series)) await self._send( Task( @@ -658,7 +671,8 @@ class RootScanner: return except Exception: logger.exception( - "Showreel worker error (queue size=%d)", self._showreel_queue.qsize() + "Showreel worker error (queue size=%d)", + self._showreel_queue.qsize(), ) try: self._showreel_queue.task_done() @@ -666,7 +680,6 @@ class RootScanner: pass - # --------------------------------------------------------------------------- # Legacy module-level API removed — use RootScanner per root instead. # --------------------------------------------------------------------------- diff --git a/mediahive/hivescan/scanning.py b/mediahive/hivescan/scanning.py index 70d7183..e63c929 100644 --- a/mediahive/hivescan/scanning.py +++ b/mediahive/hivescan/scanning.py @@ -4,7 +4,6 @@ import asyncio import glob from collections import defaultdict from pathlib import Path -from typing import Dict, List, Optional, Tuple from aiopathlib import AsyncPath @@ -27,23 +26,23 @@ VIDEO_EXTENSIONS = { } # Caches for expensive operations -_episode_files_cache: Dict[str, Dict[Tuple[int, int], List[Tuple[str, int]]]] = {} -_playable_file_cache: Dict[str, Optional[str]] = {} -_bluray_probe_file_cache: Dict[str, Optional[str]] = {} +_episode_files_cache: dict[str, dict[tuple[int, int], list[tuple[str, int]]]] = {} +_playable_file_cache: dict[str, str | None] = {} +_bluray_probe_file_cache: dict[str, str | None] = {} -async def scan_downloads(base_pattern: str) -> List[ParsedContent]: - """ - Scan download directories matching the pattern. +async def scan_downloads(base_pattern: str) -> list[ParsedContent]: + """Scan download directories matching the pattern. Args: base_pattern: Glob pattern for finding download directories Returns: List of ParsedContent objects for each found download + """ exclude_patterns = [".torrents", "incomplete", ".incomplete"] - results: List[ParsedContent] = [] + results: list[ParsedContent] = [] paths = await asyncio.to_thread(glob.glob, base_pattern) for path_str in paths: @@ -82,21 +81,21 @@ def categorize_downloads( async def find_episode_files( path: Path, -) -> Dict[Tuple[int, int], List[Tuple[str, int]]]: - """ - Find all episode video files in a directory. +) -> 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 = path.as_posix() if cache_key in _episode_files_cache: return _episode_files_cache[cache_key] - episodes: Dict[Tuple[int, int], List[Tuple[str, int]]] = {} + episodes: dict[tuple[int, int], list[tuple[str, int]]] = {} ap = AsyncPath(path) if await ap.is_file(): @@ -117,19 +116,19 @@ async def find_episode_files( if ep_info: if ep_info not in episodes: episodes[ep_info] = [] - episodes[ep_info].append( - (Path(f).as_posix(), (await af.stat()).st_size) - ) - except (OSError, PermissionError): + episodes[ep_info].append(( + Path(f).as_posix(), + (await af.stat()).st_size, + )) + except OSError, PermissionError: pass _episode_files_cache[cache_key] = episodes return episodes -async def find_playable_file(path: Path) -> Optional[str]: - """ - Find the main playable media file in a directory. +async def find_playable_file(path: Path) -> str | None: + """Find the main playable media file in a directory. For Blu-ray discs: Returns BDMV/MovieObject.bdmv (fallback: BDMV/index.bdmv) For other content: Returns the largest video file @@ -198,7 +197,7 @@ async def find_playable_file(path: Path) -> Optional[str]: result = nested_video_ts_ifo.as_posix() _playable_file_cache[cache_key] = result return result - except (OSError, PermissionError): + except OSError, PermissionError: pass # Find largest video file @@ -210,7 +209,7 @@ async def find_playable_file(path: Path) -> Optional[str]: if "sample" in Path(f).name.lower(): continue video_files.append((Path(f).as_posix(), (await af.stat()).st_size)) - except (OSError, PermissionError): + except OSError, PermissionError: pass if not video_files: @@ -223,7 +222,7 @@ async def find_playable_file(path: Path) -> Optional[str]: return result -async def find_metadata_probe_file(playable_path: Optional[str]) -> Optional[str]: +async def find_metadata_probe_file(playable_path: str | None) -> str | None: """Resolve a path suitable for ffmpeg stream metadata probing. For regular files, returns ``playable_path`` unchanged. @@ -252,7 +251,7 @@ async def find_metadata_probe_file(playable_path: Optional[str]) -> Optional[str ) # Group VOBs by title set (VTS_XX_Y.VOB) - title_sets: Dict[str, List[Tuple[str, int]]] = defaultdict(list) + title_sets: dict[str, list[tuple[str, int]]] = defaultdict(list) try: for f in AsyncPath(video_ts_dir).glob("*.vob"): af = AsyncPath(f) @@ -263,7 +262,7 @@ async def find_metadata_probe_file(playable_path: Optional[str]) -> Optional[str ts_num = name[4:6] size = (await af.stat()).st_size title_sets[ts_num].append((Path(f).as_posix(), size)) - except (OSError, PermissionError): + except OSError, PermissionError: _bluray_probe_file_cache[cache_key] = None return None @@ -298,14 +297,14 @@ async def find_metadata_probe_file(playable_path: Optional[str]) -> Optional[str _bluray_probe_file_cache[cache_key] = None return None - candidates: List[Tuple[str, int]] = [] + candidates: list[tuple[str, int]] = [] try: for f in ap_stream.rglob("*.m2ts"): af = AsyncPath(f) if not await af.is_file(): continue candidates.append((Path(f).as_posix(), (await af.stat()).st_size)) - except (OSError, PermissionError): + except OSError, PermissionError: _bluray_probe_file_cache[cache_key] = None return None @@ -320,8 +319,8 @@ async def find_metadata_probe_file(playable_path: Optional[str]) -> Optional[str async def find_cover_image( - title: str, year: Optional[int], media_type: str, cover_dir: Path -) -> Optional[str]: + title: str, year: int | None, media_type: str, cover_dir: Path +) -> str | None: """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" diff --git a/mediahive/hivescan/showreel.py b/mediahive/hivescan/showreel.py index 0dc0f10..0496b08 100644 --- a/mediahive/hivescan/showreel.py +++ b/mediahive/hivescan/showreel.py @@ -1,5 +1,4 @@ -""" -Showreel generation module for media preview clips. +"""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, @@ -15,7 +14,6 @@ import sys from collections import Counter from dataclasses import dataclass from pathlib import Path -from typing import Optional from aiopathlib import AsyncPath @@ -71,7 +69,7 @@ async def _run_ffmpeg( ) try: stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) - except asyncio.TimeoutError: + except TimeoutError: await _kill_proc(proc) try: stdout, stderr = await proc.communicate() @@ -133,7 +131,7 @@ def get_reel_source_extensions() -> list[str]: return REEL_SOURCE_EXTENSIONS.copy() -def _to_media_path(path: Path, media_root: Optional[Path] = None) -> str: +def _to_media_path(path: Path, media_root: Path | None = None) -> str: """Convert an absolute reel file path to a media-root-relative path when possible.""" if media_root: try: @@ -180,10 +178,9 @@ def get_reel_audio_options() -> list[str]: def get_expected_showreel_paths( media_folder: Path, timestamps: list[int] = SHOWREEL_TIMESTAMPS, - media_root: Optional[Path] = None, + media_root: Path | None = None, ) -> list[str]: - """ - Compute the expected showreel paths without generating them. + """Compute the expected showreel paths without generating them. Args: media_folder: Folder for this specific media item @@ -192,6 +189,7 @@ def get_expected_showreel_paths( Returns: List of relative paths where showreels will be created for this platform + """ paths = [] extension = get_reel_extension() @@ -211,10 +209,9 @@ def get_expected_episode_reel_path( media_folder: Path, season_num: int, episode_num: int, - media_root: Optional[Path] = None, + media_root: Path | None = None, ) -> str: - """ - Compute the expected episode reel path without generating it. + """Compute the expected episode reel path without generating it. Args: media_folder: Folder for this series @@ -224,6 +221,7 @@ def get_expected_episode_reel_path( Returns: Relative path where the reel will be created for this platform + """ output_path = ( media_folder / f"S{season_num:02d}E{episode_num:02d}{get_reel_extension()}" @@ -239,7 +237,7 @@ def get_expected_episode_reel_path( def get_existing_showreel_paths( media_folder: Path, timestamps: list[int] = SHOWREEL_TIMESTAMPS, - media_root: Optional[Path] = None, + media_root: Path | None = None, ) -> list[str]: """Return preferred existing showreel paths, one per reel slot, in AV1-first order.""" source_sets = get_existing_showreel_source_sets( @@ -253,7 +251,7 @@ def get_existing_showreel_paths( def get_existing_showreel_source_sets( media_folder: Path, timestamps: list[int] = SHOWREEL_TIMESTAMPS, - media_root: Optional[Path] = None, + media_root: Path | None = None, ) -> list[list[str]]: """Return all existing showreel source files for each reel slot in AV1-first order.""" source_sets: list[list[str]] = [] @@ -272,7 +270,7 @@ def get_existing_episode_reel_path( media_folder: Path, season_num: int, episode_num: int, - media_root: Optional[Path] = None, + media_root: Path | None = None, ) -> str | None: """Return the preferred existing episode reel path in AV1-first order.""" sources = get_existing_episode_reel_sources( @@ -288,7 +286,7 @@ def get_existing_episode_reel_sources( media_folder: Path, season_num: int, episode_num: int, - media_root: Optional[Path] = None, + media_root: Path | None = None, ) -> list[str]: """Return all existing episode reel source files in AV1-first order.""" ep_code = f"S{season_num:02d}E{episode_num:02d}" @@ -319,15 +317,15 @@ async def episode_reel_exists( ).exists() -def get_bluray_uri(video_path: str) -> Optional[str]: - """ - Convert a Blu-ray index.bdmv path to an ffmpeg-compatible bluray: URI. +def get_bluray_uri(video_path: str) -> str | None: + """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 @@ -343,18 +341,18 @@ def get_bluray_uri(video_path: str) -> Optional[str]: # Cache for AV1 encoder availability -_av1_encoder_cache: Optional[str] = None +_av1_encoder_cache: str | None = None async def get_av1_encoder() -> str: - """ - Detect the best available AV1 encoder. + """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: @@ -389,21 +387,20 @@ async def get_av1_encoder() -> str: def get_encoder_options(encoder: str) -> list[str]: - """ - Get encoder-specific options for the given AV1 encoder. + """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"] + # libsvtav1 software encoder + return ["-crf", "38", "-preset", "6"] @dataclass @@ -423,7 +420,9 @@ class MediaProbeInfo: _media_probe_cache: dict[str, MediaProbeInfo] = {} _duration_re = re.compile(r"Duration:\s*(\d+):(\d+):(\d+(?:\.\d+)?)") _dimension_re = re.compile(r"(\d{2,5})x(\d{2,5})") -_dovi_profile_re = re.compile(r"DOVI configuration record:.*?profile:\s*(\d+)", re.I) +_dovi_profile_re = re.compile( + r"DOVI configuration record:.*?profile:\s*(\d+)", re.IGNORECASE +) _audio_stream_re = re.compile(r"Stream #\d+:\d+(?:\(([^)]+)\))?:\s+Audio:") _subtitle_stream_re = re.compile(r"Stream #\d+:\d+(?:\(([^)]+)\))?:\s+Subtitle:") @@ -511,9 +510,8 @@ async def probe_media_info(video_path: str) -> MediaProbeInfo: return info -async def detect_dovi_profile(video_path: str) -> Optional[int]: - """ - Detect Dolby Vision profile from a video file. +async def detect_dovi_profile(video_path: str) -> int | None: + """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) @@ -528,8 +526,7 @@ async def detect_dovi_profile(video_path: str) -> Optional[int]: def get_dovi_to_hdr10_filter() -> str: - """ - Get the video filter string for converting DoVi to HDR10. + """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. @@ -543,8 +540,7 @@ def get_dovi_to_hdr10_filter() -> str: async def is_hdr_video(video_path: str) -> bool: - """ - Check if a video file is HDR using ffmpeg probe output. + """Check if a video file is HDR using ffmpeg probe output. Returns True if the video has HDR metadata (bt2020, SMPTE ST 2084, etc.) """ @@ -554,9 +550,8 @@ async def is_hdr_video(video_path: str) -> bool: return False -async def detect_crop(video_path: str) -> Optional[str]: - """ - Detect black bars in a video and return the crop filter string. +async def detect_crop(video_path: str) -> str | None: + """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 @@ -571,6 +566,7 @@ async def detect_crop(video_path: str) -> Optional[str]: Returns: Crop filter string like "crop=1920:800:0:140" if black bars detected, or None if no cropping needed or detection failed. + """ # First, get source video dimensions to check if it's 16:9 probe_info = await probe_media_info(video_path) @@ -655,9 +651,8 @@ async def detect_crop(video_path: str) -> Optional[str]: 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 + if x >= 8 and y >= 8 and 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 @@ -676,10 +671,8 @@ async def detect_crop(video_path: str) -> Optional[str]: return crop_result -async def get_video_duration(video_path: str) -> Optional[float]: - """ - Get the duration of a video file in seconds using ffmpeg probe output. - """ +async def get_video_duration(video_path: str) -> float | None: + """Get the duration of a video file in seconds using ffmpeg probe output.""" try: return (await probe_media_info(video_path)).duration except Exception: @@ -693,8 +686,7 @@ async def generate_showreel_images( title: str | None = None, on_progress=None, ) -> list[str]: - """ - Generate showreel video clips from a video file at specified timestamps. + """Generate showreel video clips from a video file at specified timestamps. Saves 10-second clips in a platform-native format, downscaled to max 720px width, preserving original color metadata. macOS emits MP4/H.265; other platforms emit WebM/AV1. @@ -708,6 +700,7 @@ async def generate_showreel_images( Returns: List of relative paths to generated showreel video clips + """ if not video_path: logger.warning( @@ -869,9 +862,8 @@ async def generate_episode_reel( media_folder: Path, season_num: int, episode_num: int, -) -> Optional[str]: - """ - Generate a single 10-second reel video clip for a TV episode. +) -> str | None: + """Generate a single 10-second reel video clip for a TV episode. Saves a platform-native clip such as S01E05.mp4 on macOS or S01E05.webm elsewhere. The clip is downscaled to max 720px width while preserving original color metadata. @@ -884,6 +876,7 @@ async def generate_episode_reel( Returns: Relative path to generated image, or None if failed + """ ep_code = f"S{season_num:02d}E{episode_num:02d}" if not video_path: diff --git a/mediahive/hivescan/tmdb_client.py b/mediahive/hivescan/tmdb_client.py index daa6bf3..95db49c 100644 --- a/mediahive/hivescan/tmdb_client.py +++ b/mediahive/hivescan/tmdb_client.py @@ -1,7 +1,5 @@ #!/usr/bin/env python3 -""" -TMDb Client - Fetch movie and TV series metadata from The Movie Database (TMDb). -""" +"""TMDb Client - Fetch movie and TV series metadata from The Movie Database (TMDb).""" import asyncio import hashlib @@ -10,7 +8,6 @@ import os import sys import urllib.parse from pathlib import Path -from typing import Dict, Optional import httpx from aiopathlib import AsyncPath @@ -28,10 +25,10 @@ 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 +_tmdb_cache_dir: Path | None = None # Persistent async HTTP client for connection reuse -_http_client: Optional[httpx.AsyncClient] = None +_http_client: httpx.AsyncClient | None = None def set_cache_dir(cache_dir: Path) -> None: @@ -65,7 +62,7 @@ def _get_http_client() -> httpx.AsyncClient: _NOT_FOUND = object() -def _get_cache_path(endpoint: str, params: Dict[str, str]) -> Path: +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())) @@ -88,7 +85,7 @@ async def _load_from_cache(cache_path: Path): return _NOT_FOUND -async def _save_to_cache(cache_path: Path, data: Optional[Dict]): +async def _save_to_cache(cache_path: Path, data: dict | None): """Save response to cache.""" try: await AsyncPath(_get_cache_dir()).mkdir(parents=True, exist_ok=True) @@ -105,8 +102,8 @@ async def _save_to_cache(cache_path: Path, data: Optional[Dict]): async def tmdb_api_request( - endpoint: str, params: Optional[Dict[str, str]] = None -) -> Optional[Dict[str, str]]: + endpoint: str, params: dict[str, str] | None = None +) -> dict[str, str] | None: """Make a request to the TMDb API with disk caching and connection reuse.""" params = params or {} @@ -144,7 +141,7 @@ async def tmdb_api_request( return None -async def fetch_movie_details(movie_id: int) -> Optional[Dict]: +async def fetch_movie_details(movie_id: int) -> dict | None: """Fetch detailed movie info including credits, similar, keywords, and alternative titles.""" # Use append_to_response to get multiple data in one request data = await tmdb_api_request( @@ -154,7 +151,7 @@ async def fetch_movie_details(movie_id: int) -> Optional[Dict]: return data -async def fetch_series_details(series_id: int) -> Optional[Dict]: +async def fetch_series_details(series_id: int) -> dict | None: """Fetch detailed TV series info including credits, similar, and keywords.""" # Use append_to_response to get multiple data in one request data = await tmdb_api_request( @@ -163,11 +160,8 @@ async def fetch_series_details(series_id: int) -> Optional[Dict]: return data -async def fetch_season_details( - series_id: int, season_number: int -) -> Optional[SeasonInfo]: - """ - Fetch detailed season info including all episodes. +async def fetch_season_details(series_id: int, season_number: int) -> SeasonInfo | None: + """Fetch detailed season info including all episodes. Returns season metadata with episode list including: - Episode names, overviews, air dates @@ -231,8 +225,7 @@ def _map_person_gender(value: object) -> str | None: def _generate_title_variants(words: list[str], min_words: int = 2) -> list[str]: - """ - Generate title variants by progressively removing words from both ends. + """Generate title variants by progressively removing words from both ends. Order: full title, then shorter from end, then shorter from start. """ @@ -272,8 +265,7 @@ def _normalize_for_match(text: str) -> set[str]: def _titles_match(original_title: str, tmdb_title: str, search_query: str) -> bool: - """ - Check if TMDb result title reasonably matches our original title. + """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. @@ -325,11 +317,8 @@ def _titles_match(original_title: str, tmdb_title: str, search_query: str) -> bo ) -async def _search_movie_with_fallbacks( - title: str, year: Optional[int] -) -> Optional[Dict]: - """ - Search for a movie with progressive title shortening fallbacks. +async def _search_movie_with_fallbacks(title: str, year: int | None) -> dict | None: + """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. @@ -340,7 +329,7 @@ async def _search_movie_with_fallbacks( variants = _generate_title_variants(words, min_words=2) def _result_matches( - top_result: Dict, original_title: str, search_query: str + 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", "") @@ -376,7 +365,7 @@ async def _search_movie_with_fallbacks( return None -async def fetch_movie_info(title: str, year: Optional[int] = None) -> Optional[Info]: +async def fetch_movie_info(title: str, year: int | None = None) -> Info | None: """Fetch comprehensive movie info from TMDb.""" data = await _search_movie_with_fallbacks(title, year) @@ -456,23 +445,22 @@ async def fetch_movie_info(title: str, year: Optional[int] = None) -> Optional[I rating=details.get("vote_average"), vote_count=details.get("vote_count"), overview=details.get("overview"), - genres=genres if genres else None, + genres=genres or 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, + similar=similar or None, + keywords=keywords or None, + cast=cast or None, director=director, ) -async def _search_series_with_fallbacks(title: str) -> Optional[Dict]: - """ - Search for a TV series with progressive title shortening fallbacks. +async def _search_series_with_fallbacks(title: str) -> dict | None: + """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. @@ -493,7 +481,7 @@ async def _search_series_with_fallbacks(title: str) -> Optional[Dict]: return None -async def fetch_series_info(title: str) -> Optional[Info]: +async def fetch_series_info(title: str) -> Info | None: """Fetch comprehensive TV series info from TMDb.""" data = await _search_series_with_fallbacks(title) @@ -561,17 +549,17 @@ async def fetch_series_info(title: str) -> Optional[Info]: rating=details.get("vote_average"), vote_count=details.get("vote_count"), overview=details.get("overview"), - genres=genres if genres else None, + genres=genres or 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, + similar=similar or None, + keywords=keywords or None, + cast=cast or None, + creators=creators or None, number_of_seasons=details.get("number_of_seasons"), number_of_episodes=details.get("number_of_episodes"), - networks=networks if networks else None, + networks=networks or None, ) diff --git a/mediahive/hivescan/utils.py b/mediahive/hivescan/utils.py index 90453a0..c5dd9ca 100644 --- a/mediahive/hivescan/utils.py +++ b/mediahive/hivescan/utils.py @@ -2,7 +2,6 @@ import time from pathlib import Path -from typing import List, Optional from aiopathlib import AsyncPath @@ -84,9 +83,8 @@ def normalize_resolution_label(value: str | None) -> str | None: return mapping.get(normalized) -async def get_added_timestamp(path: Path) -> Optional[int]: - """ - Get the timestamp when a torrent was added to the collection. +async def get_added_timestamp(path: Path) -> int | None: + """Get the timestamp when a torrent was added to the collection. Heuristic: - For directories: use ctime (most accurate for torrent folder creation) @@ -95,11 +93,12 @@ async def get_added_timestamp(path: Path) -> Optional[int]: Returns: Unix timestamp as int, or None if path doesn't exist + """ ap = AsyncPath(path) try: stat_info = await ap.stat() - except (OSError, PermissionError): + except OSError, PermissionError: return None if await ap.is_dir(): @@ -124,7 +123,7 @@ async def get_directory_size(path: Path) -> int: for item in ap.rglob("*"): if await AsyncPath(item).is_file(): total += (await AsyncPath(item).stat()).st_size - except (OSError, PermissionError): + except OSError, PermissionError: pass return total @@ -138,9 +137,8 @@ def format_size(size_bytes: int) -> str: return f"{size_bytes:.2f} PB" -async def find_common_root(paths: List[Path]) -> Optional[Path]: - """ - Find the common root directory for a list of paths. +async def find_common_root(paths: list[Path]) -> Path | None: + """Find the common root directory for a list of paths. Returns None if paths are on different drives/mounts or have no common ancestor. """ @@ -198,11 +196,8 @@ async def find_common_root(paths: List[Path]) -> Optional[Path]: return Path(*common_parts) -def make_relative_path( - path: Optional[str], root: Optional[str] = None -) -> Optional[str]: - """ - Convert an absolute path to a posix-style path relative to the given root. +def make_relative_path(path: str | None, root: str | None = None) -> str | None: + """Convert an absolute path to a posix-style path relative to the given root. If root is None, returns the path as a posix string unchanged. """ @@ -225,7 +220,7 @@ def sanitize_filename(name: str) -> str: return name -def get_media_folder_name(title: str, year: Optional[int], media_type: str) -> str: +def get_media_folder_name(title: str, year: int | None, media_type: str) -> str: """Get the folder name for a media item.""" sanitized_title = sanitize_filename(title) if media_type == "movie" and year: @@ -234,7 +229,7 @@ def get_media_folder_name(title: str, year: Optional[int], media_type: str) -> s def get_media_folder_path( - title: str, year: Optional[int], media_type: str, cover_dir: Path + title: str, year: int | None, 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" diff --git a/mediahive/index_store.py b/mediahive/index_store.py index b84af98..c4d6971 100644 --- a/mediahive/index_store.py +++ b/mediahive/index_store.py @@ -1,5 +1,4 @@ -""" -In-memory index store with disk snapshot and WebSocket broadcast. +"""In-memory index store with disk snapshot and WebSocket broadcast. The IndexStore is the single source of truth for the media index. All mutations happen synchronously in the asyncio event loop — no locks needed. @@ -12,7 +11,6 @@ import logging import os from datetime import datetime from pathlib import Path -from typing import Optional import msgspec from aiopathlib import AsyncPath @@ -48,7 +46,12 @@ class IndexStore: # This would make IndexStore testable without FastAPI's WebSocket. """ - def __init__(self, snapshot_path: Path, media_root: Optional[str] = None, root_id: Optional[str] = None): + def __init__( + self, + snapshot_path: Path, + media_root: str | None = None, + root_id: str | None = None, + ): self.snapshot_path = snapshot_path self.media_root = media_root self.root_id = root_id @@ -62,7 +65,7 @@ class IndexStore: # Snapshot debounce state self._snapshot_dirty = False - self._snapshot_task: Optional[asyncio.Task] = None + self._snapshot_task: asyncio.Task | None = None # ------------------------------------------------------------------ # Persistence diff --git a/mediahive/models/data.py b/mediahive/models/data.py index a46b7f1..e3a8cb3 100644 --- a/mediahive/models/data.py +++ b/mediahive/models/data.py @@ -1,5 +1,4 @@ -""" -Data structures for mediahive and hivescan. +"""Data structures for mediahive and hivescan. All types are msgspec.Structs for fast serialization. """ diff --git a/mediahive/models/events.py b/mediahive/models/events.py index d5fa4d3..8ef5dc9 100644 --- a/mediahive/models/events.py +++ b/mediahive/models/events.py @@ -1,5 +1,4 @@ -""" -Event types shared between scanner and WebSocket. +"""Event types shared between scanner and WebSocket. These types are used as: - Internal scan events (scanner → server queue) diff --git a/mediahive/models/protocol.py b/mediahive/models/protocol.py index 2781a00..b7b3aae 100644 --- a/mediahive/models/protocol.py +++ b/mediahive/models/protocol.py @@ -11,7 +11,6 @@ from fastapi.responses import Response from .data import Movie, Series from .events import Remove, ScanEvent, Task, Upsert - # --------------------------------------------------------------------------- # WebSocket message types # --------------------------------------------------------------------------- diff --git a/mediahive/models/tmdb.py b/mediahive/models/tmdb.py index 24f50a8..f292813 100644 --- a/mediahive/models/tmdb.py +++ b/mediahive/models/tmdb.py @@ -1,5 +1,4 @@ -""" -TMDb data structures. +"""TMDb data structures. All types are msgspec.Structs for fast serialization. """ @@ -8,7 +7,6 @@ from __future__ import annotations import msgspec - # --------------------------------------------------------------------------- # Sub-types (shared by TMDb results and index items) # --------------------------------------------------------------------------- diff --git a/mediahive/root_registry.py b/mediahive/root_registry.py index 1ec50dd..d4e46c7 100644 --- a/mediahive/root_registry.py +++ b/mediahive/root_registry.py @@ -5,16 +5,13 @@ from __future__ import annotations import asyncio import hashlib import logging -import os from pathlib import Path -from typing import Optional import msgspec from mediahive.config import load_config, save_config from mediahive.index_store import IndexStore from mediahive.models.events import ScanEvent, Task, Upsert -from mediahive.models.data import TaskInfo logger = logging.getLogger("mediahive.root_registry") @@ -40,8 +37,10 @@ def _normalize_path(path: str) -> str: if len(posix) >= 2 and posix[1] == ":": posix = posix[0].lower() + posix[1:] # Strip trailing slash (except root "/") - while len(posix) > 1 and posix.endswith("/") and not ( - len(posix) == 3 and posix[1] == ":" and posix[2] == "/" + while ( + len(posix) > 1 + and posix.endswith("/") + and not (len(posix) == 3 and posix[1] == ":" and posix[2] == "/") ): posix = posix[:-1] return posix @@ -92,17 +91,19 @@ class RootContext: self.name = name or root_id self.root_path = root_path self.status = "loading" - self.error: Optional[str] = None + self.error: str | None = None snapshot_path = root_path / ".mediahive" / "index.json" - self.store = IndexStore(snapshot_path, media_root=root_path.as_posix(), root_id=root_id) + self.store = IndexStore( + snapshot_path, media_root=root_path.as_posix(), root_id=root_id + ) # Scanner is injected later by the supervisor - self.scanner: Optional[object] = None + self.scanner: object | None = None # Event queue and consumer self._events: asyncio.Queue[ScanEvent] = asyncio.Queue() - self._consumer_task: Optional[asyncio.Task] = None + self._consumer_task: asyncio.Task | None = None async def start(self) -> None: """Load snapshot and start event consumer.""" @@ -161,7 +162,9 @@ class RootContext: except asyncio.CancelledError: return except Exception: - logger.exception("Error processing scan event for root %s", self.root_id) + logger.exception( + "Error processing scan event for root %s", self.root_id + ) # --------------------------------------------------------------------------- @@ -181,7 +184,7 @@ class Supervisor: # Read helpers # ------------------------------------------------------------------ - def get(self, root_id: str) -> Optional[RootContext]: + def get(self, root_id: str) -> RootContext | None: return self._contexts.get(root_id) def all_contexts(self) -> dict[str, RootContext]: @@ -212,14 +215,15 @@ class Supervisor: continue movies.extend(ctx.store.movies.values()) series.extend(ctx.store.series.values()) - total_movie_versions += sum(len(m.torrents) for m in ctx.store.movies.values()) + total_movie_versions += sum( + len(m.torrents) for m in ctx.store.movies.values() + ) total_series_episodes += sum( sum(len(season.episodes) for season in s.seasons) for s in ctx.store.series.values() ) from datetime import datetime - from mediahive.models.data import IndexSnapshot, MediaStats return { "version": 7, @@ -238,7 +242,9 @@ class Supervisor: # Atomic replacement # ------------------------------------------------------------------ - async def replace_roots(self, roots: dict[str, str]) -> tuple[list[RootEntry], list[dict]]: + async def replace_roots( + self, roots: dict[str, str] + ) -> tuple[list[RootEntry], list[dict]]: """Atomically replace the active root set. Returns (accepted_entries, failed_entries_with_reason). @@ -253,11 +259,19 @@ class Supervisor: for requested_name, path_str in roots.items(): p = Path(path_str).expanduser() if not p.exists() or not p.is_dir(): - failed.append({"name": requested_name, "path": path_str, "reason": "not a directory"}) + failed.append({ + "name": requested_name, + "path": path_str, + "reason": "not a directory", + }) continue norm = _normalize_path(p.as_posix()) if norm in seen_paths: - failed.append({"name": requested_name, "path": path_str, "reason": "duplicate path"}) + failed.append({ + "name": requested_name, + "path": path_str, + "reason": "duplicate path", + }) continue seen_paths.add(norm) rid = compute_root_id(str(p)) diff --git a/mediahive/server.py b/mediahive/server.py index d2cab3f..73ca413 100644 --- a/mediahive/server.py +++ b/mediahive/server.py @@ -1,5 +1,4 @@ -""" -FastAPI server for MediaHive. +"""FastAPI server for MediaHive. Serves media files, the Vue frontend, and runs the continuous scanning pipeline with live WebSocket updates. Excluded paths are controlled by @@ -17,7 +16,6 @@ import os import re import subprocess import sys -import tempfile import urllib.error import urllib.request from contextlib import asynccontextmanager @@ -39,7 +37,7 @@ from mediahive.models.protocol import ( PlayMediaRequest, RootsRequest, ) -from mediahive.root_registry import Supervisor, compute_root_id +from mediahive.root_registry import Supervisor logger = logging.getLogger("mediahive.server") @@ -339,7 +337,10 @@ async def _activate_all_roots() -> None: return await _attach_scanners() - logger.info("Background root activation complete; %d root(s) active", len(supervisor.all_contexts())) + logger.info( + "Background root activation complete; %d root(s) active", + len(supervisor.all_contexts()), + ) # --------------------------------------------------------------------------- @@ -518,9 +519,7 @@ async def open_folder(root_id: str, request: Request): if target_path.is_file(): if not _select_file_in_windows_explorer(target_path): select_arg = f'/n,/select,"{native_path}"' - subprocess.Popen( - ["explorer.exe", select_arg], **_POPEN_KWARGS - ) + subprocess.Popen(["explorer.exe", select_arg], **_POPEN_KWARGS) else: subprocess.Popen(["explorer.exe", native_path], **_POPEN_KWARGS) elif sys.platform == "darwin": @@ -569,7 +568,7 @@ def _mpcbe_request(path: str, timeout: float = 0.75) -> bool: try: with urllib.request.urlopen(req, timeout=timeout) as resp: return 200 <= resp.status < 300 - except (urllib.error.URLError, TimeoutError, OSError): + except urllib.error.URLError, TimeoutError, OSError: return False diff --git a/mediahive/winmain.py b/mediahive/winmain.py index b1b0b93..4541673 100644 --- a/mediahive/winmain.py +++ b/mediahive/winmain.py @@ -155,7 +155,9 @@ def _save_playback_state(path: Path, state: dict[str, object]) -> None: tmp_path.replace(path) -def _media_key_for_filepath(filepath: str, roots: list[Path]) -> tuple[str, Path] | None: +def _media_key_for_filepath( + filepath: str, roots: list[Path] +) -> tuple[str, Path] | None: """Resolve a filepath to a (relative_key, matched_root) tuple.""" for root in roots: try: @@ -195,7 +197,7 @@ def _mpcbe_request(path: str, timeout: float = MPC_BE_REQUEST_TIMEOUT) -> bool: try: with urllib.request.urlopen(req, timeout=timeout) as resp: return 200 <= resp.status < 300 - except (urllib.error.URLError, TimeoutError, OSError): + except urllib.error.URLError, TimeoutError, OSError: return False @@ -211,12 +213,10 @@ def _format_mpcbe_position(position_ms: int) -> str: def _seek_mpcbe_to_position(position_ms: int) -> bool: - query = urllib.parse.urlencode( - { - "wm_command": -1, - "position": _format_mpcbe_position(position_ms), - } - ) + query = urllib.parse.urlencode({ + "wm_command": -1, + "position": _format_mpcbe_position(position_ms), + }) return _mpcbe_request(f"/command.html?{query}") @@ -226,7 +226,7 @@ def _mpcbe_fetch_status() -> tuple[str, int, int, int] | None: try: with urllib.request.urlopen(req, timeout=MPC_BE_REQUEST_TIMEOUT) as resp: response_html = resp.read().decode("utf-8", errors="replace") - except (urllib.error.URLError, TimeoutError, OSError): + except urllib.error.URLError, TimeoutError, OSError: return None state_match = _STATE_RE.search(response_html) @@ -255,13 +255,9 @@ def _start_gamepad_remote( seek_begin_hold_started_at: list[float | None] = [None, None, None, None] seek_begin_fired = [False, False, False, False] last_repeat_at = [ - { - mask: 0.0 - for mask in ( - *_MPC_BE_COMMANDS.keys(), - *_MPC_BE_SEEK_MASK_TO_COMMANDS.keys(), - ) - } + dict.fromkeys( + (*_MPC_BE_COMMANDS.keys(), *_MPC_BE_SEEK_MASK_TO_COMMANDS.keys()), 0.0 + ) for _ in range(4) ] request_pool = ThreadPoolExecutor( @@ -405,13 +401,12 @@ def _start_gamepad_remote( player_filepath, \ player_position_ms, \ player_duration_ms, \ - player_state - nonlocal \ + player_state, \ status_updated_at, \ status_miss_count, \ tracked_media_key, \ - tracked_filepath - nonlocal resume_applied_for_key + tracked_filepath, \ + resume_applied_for_key if status_future is None or not status_future.done(): return @@ -603,7 +598,7 @@ def _setup_logging() -> Path: prev.unlink() log_path.rename(prev) - log_file = open(log_path, "w", encoding="utf-8", buffering=1) # line-buffered + log_file = Path(log_path).open("w", encoding="utf-8", buffering=1) # line-buffered # Redirect raw stdout/stderr so print() and tracebacks go to the file sys.stdout = log_file diff --git a/rtorrent_client.py b/rtorrent_client.py index 153d833..1e6d865 100644 --- a/rtorrent_client.py +++ b/rtorrent_client.py @@ -1,12 +1,9 @@ #!/usr/bin/env python3 -""" -RTorrent Client - Communicate with rtorrent via XMLRPC over SCGI socket. -""" +"""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): @@ -68,8 +65,7 @@ class RTorrentClient: return set() def load_torrent(self, torrent_path: Path, download_dir: Path) -> bool: - """ - Load a torrent file and set its download directory. + """Load a torrent file and set its download directory. Uses load.start_verbose to load and immediately start/hash-check. Args: @@ -78,6 +74,7 @@ class RTorrentClient: Returns: True if successful, False otherwise + """ try: # load.start_verbose with d.directory.set to specify download location @@ -90,7 +87,7 @@ class RTorrentClient: print(f"Error loading torrent {torrent_path}: {e}") return False - def get_torrent_info(self, info_hash: str) -> Optional[Dict]: + def get_torrent_info(self, info_hash: str) -> dict | None: """Get info about a loaded torrent.""" try: name = self.proxy.d.name(info_hash) @@ -112,12 +109,12 @@ class RTorrentClient: 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. + 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: @@ -139,8 +136,7 @@ class RTorrentClient: return unregistered def remove_torrent(self, info_hash: str, delete_files: bool = False) -> bool: - """ - Remove a torrent from rtorrent. + """Remove a torrent from rtorrent. Args: info_hash: The info hash of the torrent to remove @@ -148,6 +144,7 @@ class RTorrentClient: Returns: True if successful, False otherwise + """ try: if delete_files: diff --git a/scripts/fastapi-vue/devutil.py b/scripts/fastapi-vue/devutil.py index 9a82d25..9e7d559 100644 --- a/scripts/fastapi-vue/devutil.py +++ b/scripts/fastapi-vue/devutil.py @@ -9,9 +9,8 @@ from pathlib import Path from typing import Any import httpx -from fastapi_vue.hostutil import parse_endpoint - from buildutil import find_dev_tool, find_install_tool, logger +from fastapi_vue.hostutil import parse_endpoint class ProcessGroup: @@ -33,7 +32,7 @@ class ProcessGroup: return proc async def wait( - self, *waitables: "asyncio.subprocess.Process | Coroutine[Any, Any, Any]" + self, *waitables: asyncio.subprocess.Process | Coroutine[Any, Any, Any] ) -> None: """Wait for processes/coroutines to complete, raise SystemExit on failure.""" @@ -175,7 +174,7 @@ def setup_fastapi( host = endpoints[0]["host"] port = endpoints[0]["port"] - reload_dir = module.split(".")[0] # Don't reload on frontend changes + reload_dir = module.split(".", maxsplit=1)[0] # Don't reload on frontend changes cmd = [ sys.executable, diff --git a/scripts/guibuild.py b/scripts/guibuild.py index b21cded..fe07903 100644 --- a/scripts/guibuild.py +++ b/scripts/guibuild.py @@ -73,7 +73,7 @@ def fetch_ffmpeg() -> Path: ffmpeg_entry = next( name for name in zf.namelist() if name.endswith("/bin/ffmpeg.exe") ) - with zf.open(ffmpeg_entry) as src, open(dest, "wb") as out: + with zf.open(ffmpeg_entry) as src, Path(dest).open("wb") as out: out.write(src.read()) print(f"ffmpeg staged at {dest} ({dest.stat().st_size // 1024 // 1024} MB)") @@ -108,7 +108,7 @@ def fetch_macos_arm64_binaries() -> dict[str, Path]: for name in zf.namelist() if Path(name).name == tool_name and not name.endswith("/") ) - with zf.open(entry_name) as src, open(dest, "wb") as out: + with zf.open(entry_name) as src, Path(dest).open("wb") as out: out.write(src.read()) dest.chmod(dest.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) diff --git a/scripts/release.py b/scripts/release.py index 88a8c36..048e998 100644 --- a/scripts/release.py +++ b/scripts/release.py @@ -34,7 +34,7 @@ REPO_ROOT = Path(__file__).parent.parent def load_gitea_config() -> dict: pyproject = REPO_ROOT / "pyproject.toml" - with open(pyproject, "rb") as f: + with Path(pyproject).open("rb") as f: data = tomllib.load(f) repo_url = data.get("project", {}).get("urls", {}).get("Repository") if not repo_url: @@ -156,7 +156,7 @@ def upload_asset( size_mb = path.stat().st_size / (1024 * 1024) mime = "application/zip" if path.suffix == ".zip" else "application/octet-stream" print(f"Uploading {path.name} ({size_mb:.1f} MB) ...") - with open(path, "rb") as fh: + with Path(path).open("rb") as fh: resp = client.post( url, files={"attachment": (path.name, fh, mime)}, diff --git a/scripts/rtorrent-manager.py b/scripts/rtorrent-manager.py index f44db8e..7c559a2 100644 --- a/scripts/rtorrent-manager.py +++ b/scripts/rtorrent-manager.py @@ -1,17 +1,16 @@ #!/usr/bin/env python3 -""" -Torrent Scanner - Scans for .torrent files and analyzes their trackers. -""" +"""Torrent Scanner - Scans for .torrent files and analyzes their trackers.""" import argparse import glob import hashlib import shutil -from pathlib import Path +from collections.abc import Iterator from dataclasses import dataclass -from typing import Iterator +from pathlib import Path import bencodepy + from rtorrent_client import RTorrentClient @@ -40,8 +39,7 @@ class TorrentInfo: return self.path.parent.parent def get_expected_data_path(self) -> Path: - """ - Get the expected path where downloaded data should exist. + """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) @@ -49,11 +47,11 @@ class TorrentInfo: return self.get_download_directory() / self.name def verify_download_exists(self) -> tuple[bool, str]: - """ - Verify that the downloaded data exists on disk. + """Verify that the downloaded data exists on disk. Returns: Tuple of (exists: bool, message: str) + """ expected_path = self.get_expected_data_path() @@ -69,27 +67,26 @@ class TorrentInfo: 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}" + # 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. + """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: + with Path(filepath).open("rb") as f: data = bencodepy.decode(f.read()) except Exception as e: print(f"Error parsing {filepath}: {e}") @@ -154,14 +151,14 @@ def parse_torrent(filepath: Path) -> TorrentInfo | None: def scan_torrent_directories(paths: list[str]) -> Iterator[Path]: - """ - Scan directories for .torrent files. + """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): @@ -174,8 +171,7 @@ def scan_torrent_directories(paths: list[str]) -> Iterator[Path]: def find_torrents_with_tracker( tracker_domain: str, paths: list[str] ) -> list[TorrentInfo]: - """ - Find all torrents that have a specific tracker domain. + """Find all torrents that have a specific tracker domain. Args: tracker_domain: Domain to search for in tracker URLs (e.g., "hdbits.org") @@ -183,6 +179,7 @@ def find_torrents_with_tracker( Returns: List of TorrentInfo objects for matching torrents + """ matching_torrents = [] @@ -373,39 +370,36 @@ Examples: 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 + # Remove from rtorrent (keeps downloaded files) + elif 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(): + # 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: - 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)") + 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" [ERR] {torrent_info['name']}: failed to remove from rtorrent" - ) + print(f" [DEL] {torrent_info['name']} (no data)") + else: + print(f" [ERR] {torrent_info['name']}: failed to remove from rtorrent") print() if dry_run: