diff --git a/core/async_nfo_manager.py b/core/async_nfo_manager.py new file mode 100644 index 0000000..372a777 --- /dev/null +++ b/core/async_nfo_manager.py @@ -0,0 +1,423 @@ +""" +Async NFO Manager for NFOGuard +High-performance async NFO file operations with concurrent processing +""" +import asyncio +from pathlib import Path +from typing import Optional, List, Dict, Any +from datetime import datetime +import xml.etree.ElementTree as ET + +from utils.logging import _log +from utils.async_file_utils import ( + async_read_nfo_file, + async_write_nfo_file, + async_file_exists, + async_set_file_mtime, + async_batch_nfo_operations, + async_concurrent_episode_processing +) +from utils.nfo_patterns import ( + create_basic_nfo_structure, + extract_imdb_from_nfo_content, + extract_dates_from_nfo, + extract_imdb_id_from_text +) +from utils.validation import validate_date_string + + +class AsyncNFOManager: + """Async NFO file manager with concurrent processing capabilities""" + + def __init__(self, manager_brand: str = "NFOGuard", debug: bool = False): + self.manager_brand = manager_brand + self.debug = debug + + async def async_parse_imdb_from_path(self, path: Path) -> Optional[str]: + """ + Async extract IMDb ID from directory path or filename + + Args: + path: Path to examine + + Returns: + IMDb ID if found, None otherwise + """ + # Use the sync version since it's just string processing + return extract_imdb_id_from_text(str(path)) + + async def async_parse_imdb_from_nfo(self, nfo_path: Path) -> Optional[str]: + """ + Async extract IMDb ID from NFO file content + + Args: + nfo_path: Path to NFO file + + Returns: + IMDb ID if found, None otherwise + """ + root = await async_read_nfo_file(nfo_path) + if root is None: + return None + + return extract_imdb_from_nfo_content(root) + + async def async_find_movie_imdb_id(self, movie_dir: Path) -> Optional[str]: + """ + Async find IMDb ID from directory name, filenames, or NFO file + + Args: + movie_dir: Path to movie directory + + Returns: + IMDb ID if found, None otherwise + """ + if self.debug: + _log("DEBUG", f"Async searching for IMDb ID in: {movie_dir.name}") + + # First try directory name + imdb_id = await self.async_parse_imdb_from_path(movie_dir) + if imdb_id: + if self.debug: + _log("DEBUG", f"Found IMDb ID in directory name: {imdb_id}") + return imdb_id + + # Try all files in the directory concurrently + if await async_file_exists(movie_dir): + try: + from utils.async_file_utils import aiofiles + entries = await aiofiles.os.listdir(movie_dir) + + # Create tasks to check all files + async def check_file(filename: str) -> Optional[str]: + file_path = movie_dir / filename + if await aiofiles.os.path.isfile(file_path): + return await self.async_parse_imdb_from_path(file_path) + return None + + # Check all files concurrently + results = await asyncio.gather( + *[check_file(filename) for filename in entries], + return_exceptions=True + ) + + # Find first valid IMDb ID + for result in results: + if isinstance(result, str) and result: + if self.debug: + _log("DEBUG", f"Found IMDb ID in filename: {result}") + return result + + except Exception as e: + _log("WARNING", f"Failed to scan directory {movie_dir}: {e}") + + # Finally, try NFO file content + nfo_path = movie_dir / "movie.nfo" + imdb_id = await self.async_parse_imdb_from_nfo(nfo_path) + if imdb_id: + if self.debug: + _log("DEBUG", f"Found IMDb ID in NFO file: {imdb_id}") + return imdb_id + + if self.debug: + _log("DEBUG", f"No IMDb ID found for: {movie_dir.name}") + return None + + async def async_create_movie_nfo( + self, + movie_dir: Path, + imdb_id: str, + dateadded: str, + premiered: Optional[str] = None, + lock_metadata: bool = True + ) -> bool: + """ + Async create movie NFO file + + Args: + movie_dir: Path to movie directory + imdb_id: IMDb ID + dateadded: Date added + premiered: Optional premiere date + lock_metadata: Whether to lock metadata + + Returns: + True if successful, False otherwise + """ + try: + nfo_path = movie_dir / "movie.nfo" + + # Prepare dates + dates = {"dateadded": dateadded} + if premiered and validate_date_string(premiered): + dates["premiered"] = premiered + + # Create NFO structure + root = create_basic_nfo_structure( + media_type="movie", + title=movie_dir.name, + imdb_id=imdb_id, + dates=dates, + additional_fields={"source": self.manager_brand} + ) + + # Write NFO file asynchronously + success = await async_write_nfo_file(nfo_path, root, lock_metadata) + + if success and self.debug: + _log("DEBUG", f"Created movie NFO: {nfo_path}") + + return success + + except Exception as e: + _log("ERROR", f"Failed to create movie NFO for {movie_dir}: {e}") + return False + + async def async_create_episode_nfo( + self, + season_dir: Path, + season: int, + episode: int, + aired: Optional[str] = None, + dateadded: Optional[str] = None, + source: str = "unknown", + lock_metadata: bool = True + ) -> bool: + """ + Async create episode NFO file + + Args: + season_dir: Path to season directory + season: Season number + episode: Episode number + aired: Optional air date + dateadded: Optional date added + source: Source of the data + lock_metadata: Whether to lock metadata + + Returns: + True if successful, False otherwise + """ + try: + nfo_filename = f"S{season:02d}E{episode:02d}.nfo" + nfo_path = season_dir / nfo_filename + + # Prepare dates + dates = {} + if aired and validate_date_string(aired): + dates["aired"] = aired + if dateadded and validate_date_string(dateadded): + dates["dateadded"] = dateadded + + # Create NFO structure + root = create_basic_nfo_structure( + media_type="episodedetails", + title=f"S{season:02d}E{episode:02d}", + dates=dates, + additional_fields={ + "season": str(season), + "episode": str(episode), + "source": source + } + ) + + # Write NFO file asynchronously + success = await async_write_nfo_file(nfo_path, root, lock_metadata) + + if success and self.debug: + _log("DEBUG", f"Created episode NFO: {nfo_path}") + + return success + + except Exception as e: + _log("ERROR", f"Failed to create episode NFO S{season:02d}E{episode:02d}: {e}") + return False + + async def async_batch_create_episode_nfos( + self, + episode_data_list: List[Dict[str, Any]], + max_concurrent: int = 5 + ) -> List[bool]: + """ + Batch create multiple episode NFOs concurrently + + Args: + episode_data_list: List of episode data dictionaries + max_concurrent: Maximum concurrent NFO operations + + Returns: + List of success/failure results + """ + async def _create_episode_nfo(episode_data: Dict[str, Any]) -> bool: + return await self.async_create_episode_nfo( + season_dir=episode_data.get('season_dir'), + season=episode_data.get('season'), + episode=episode_data.get('episode'), + aired=episode_data.get('aired'), + dateadded=episode_data.get('dateadded'), + source=episode_data.get('source', 'unknown'), + lock_metadata=episode_data.get('lock_metadata', True) + ) + + return await async_concurrent_episode_processing( + episode_data_list, + _create_episode_nfo, + max_concurrent + ) + + async def async_set_file_mtime(self, file_path: Path, date_str: str) -> bool: + """ + Async set file modification time from date string + + Args: + file_path: Path to file + date_str: Date string in ISO format + + Returns: + True if successful, False otherwise + """ + try: + if not validate_date_string(date_str): + _log("WARNING", f"Invalid date format for mtime: {date_str}") + return False + + # Parse date string to timestamp + dt = datetime.fromisoformat(date_str.replace('Z', '+00:00')) + timestamp = dt.timestamp() + + # Set file mtime asynchronously + success = await async_set_file_mtime(file_path, timestamp) + + if success and self.debug: + _log("DEBUG", f"Set mtime for {file_path}: {date_str}") + + return success + + except Exception as e: + _log("ERROR", f"Failed to set mtime for {file_path}: {e}") + return False + + async def async_batch_set_file_mtimes( + self, + file_mtime_pairs: List[Tuple[Path, str]], + max_concurrent: int = 10 + ) -> List[bool]: + """ + Batch set file modification times concurrently + + Args: + file_mtime_pairs: List of (file_path, date_str) tuples + max_concurrent: Maximum concurrent operations + + Returns: + List of success/failure results + """ + async def _set_single_mtime(file_path: Path, date_str: str) -> bool: + return await self.async_set_file_mtime(file_path, date_str) + + # Create tasks for all mtime operations + semaphore = asyncio.Semaphore(max_concurrent) + + async def _set_mtime_with_semaphore(file_path: Path, date_str: str) -> bool: + async with semaphore: + return await _set_single_mtime(file_path, date_str) + + tasks = [ + _set_mtime_with_semaphore(file_path, date_str) + for file_path, date_str in file_mtime_pairs + ] + + results = await asyncio.gather(*tasks, return_exceptions=True) + + return [result if isinstance(result, bool) else False for result in results] + + async def async_validate_nfo_integrity( + self, + nfo_paths: List[Path], + max_concurrent: int = 10 + ) -> Dict[str, Any]: + """ + Async validate integrity of multiple NFO files + + Args: + nfo_paths: List of NFO file paths to validate + max_concurrent: Maximum concurrent validations + + Returns: + Dictionary with validation results and statistics + """ + results = { + 'total_files': len(nfo_paths), + 'valid_files': 0, + 'invalid_files': 0, + 'missing_files': 0, + 'validation_errors': [], + 'file_results': {} + } + + async def _validate_single_nfo(nfo_path: Path) -> Dict[str, Any]: + file_result = { + 'path': str(nfo_path), + 'exists': False, + 'valid_xml': False, + 'has_imdb_id': False, + 'has_dates': False, + 'error': None + } + + try: + # Check if file exists + if not await async_file_exists(nfo_path): + file_result['error'] = 'File does not exist' + return file_result + + file_result['exists'] = True + + # Try to parse NFO + root = await async_read_nfo_file(nfo_path) + if root is None: + file_result['error'] = 'Failed to parse XML' + return file_result + + file_result['valid_xml'] = True + + # Check for IMDb ID + imdb_id = extract_imdb_from_nfo_content(root) + file_result['has_imdb_id'] = bool(imdb_id) + + # Check for dates + dates = extract_dates_from_nfo(root) + file_result['has_dates'] = any(dates.values()) + + except Exception as e: + file_result['error'] = str(e) + + return file_result + + # Validate all files concurrently + semaphore = asyncio.Semaphore(max_concurrent) + + async def _validate_with_semaphore(nfo_path: Path) -> Dict[str, Any]: + async with semaphore: + return await _validate_single_nfo(nfo_path) + + tasks = [_validate_with_semaphore(nfo_path) for nfo_path in nfo_paths] + file_results = await asyncio.gather(*tasks, return_exceptions=True) + + # Process results + for i, result in enumerate(file_results): + if isinstance(result, dict): + path_str = str(nfo_paths[i]) + results['file_results'][path_str] = result + + if not result['exists']: + results['missing_files'] += 1 + elif result['valid_xml']: + results['valid_files'] += 1 + else: + results['invalid_files'] += 1 + + if result.get('error'): + results['validation_errors'].append(f"{path_str}: {result['error']}") + + return results \ No newline at end of file diff --git a/processors/tv_processor.py b/processors/tv_processor.py index c30b1b2..ed351ae 100644 --- a/processors/tv_processor.py +++ b/processors/tv_processor.py @@ -1,15 +1,17 @@ """ TV Series Processor for NFOGuard -Handles TV series processing and episode management +Handles TV series processing and episode management with async I/O support """ import os import re +import asyncio from pathlib import Path from typing import Optional, Dict, List, Set, Tuple, Any from datetime import datetime from core.database import NFOGuardDatabase from core.nfo_manager import NFOManager +from core.async_nfo_manager import AsyncNFOManager from core.path_mapper import PathMapper from clients.sonarr_client import SonarrClient from clients.external_clients import ExternalClientManager @@ -20,6 +22,10 @@ from utils.file_utils import ( find_episodes_on_disk, extract_title_from_directory_name ) +from utils.async_file_utils import ( + async_find_episodes_on_disk, + async_concurrent_episode_processing +) class TVProcessor: @@ -28,6 +34,7 @@ class TVProcessor: def __init__(self, db: NFOGuardDatabase, nfo_manager: NFOManager, path_mapper: PathMapper): self.db = db self.nfo_manager = nfo_manager + self.async_nfo_manager = AsyncNFOManager(config.manager_brand, config.debug) self.path_mapper = path_mapper self.sonarr = SonarrClient( os.environ.get("SONARR_URL", ""), @@ -298,4 +305,209 @@ class TVProcessor: "aired": aired, "dateadded": dateadded, "source": source + } + + # ===== ASYNC METHODS ===== + + async def async_process_series(self, series_path: Path) -> Dict[str, Any]: + """ + Async process a TV series directory with concurrent episode processing + + Args: + series_path: Path to series directory + + Returns: + Dictionary with processing results + """ + imdb_id = self.nfo_manager.parse_imdb_from_path(series_path) + if not imdb_id: + return {"status": "error", "reason": f"No IMDb ID found in series path: {series_path}"} + + _log("INFO", f"Async processing TV series: {series_path.name}") + + # Update database + self.db.upsert_series(imdb_id, str(series_path)) + + # Find video files asynchronously + disk_episodes = await async_find_episodes_on_disk(series_path) + _log("INFO", f"Found {len(disk_episodes)} episodes on disk") + + # Get episode dates (sync for now, could be made async later) + episode_dates = self._gather_episode_dates(series_path, imdb_id, disk_episodes) + + # Prepare episode data for concurrent processing + episode_data_list = [] + mtime_operations = [] + + for (season, episode), (aired, dateadded, source) in episode_dates.items(): + if (season, episode) in disk_episodes: + season_dir = series_path / config.tv_season_dir_format.format(season=season) + + # Prepare NFO creation data + if config.manage_nfo: + episode_data_list.append({ + 'season_dir': season_dir, + 'season': season, + 'episode': episode, + 'aired': aired, + 'dateadded': dateadded, + 'source': source, + 'lock_metadata': config.lock_metadata + }) + + # Prepare mtime operations + if config.fix_dir_mtimes and dateadded: + video_files = disk_episodes[(season, episode)] + for video_file in video_files: + mtime_operations.append((video_file, dateadded)) + + # Save to database + self.db.upsert_episode_date(imdb_id, season, episode, aired, dateadded, source, True) + + # Process NFOs and mtimes concurrently + results = {} + + if episode_data_list: + _log("INFO", f"Creating {len(episode_data_list)} episode NFOs concurrently") + nfo_results = await self.async_nfo_manager.async_batch_create_episode_nfos( + episode_data_list, + max_concurrent=config.max_concurrent + ) + results['nfo_created'] = sum(nfo_results) + results['nfo_failed'] = len(nfo_results) - sum(nfo_results) + + if mtime_operations: + _log("INFO", f"Setting mtimes for {len(mtime_operations)} files concurrently") + mtime_results = await self.async_nfo_manager.async_batch_set_file_mtimes( + mtime_operations, + max_concurrent=10 + ) + results['mtime_updated'] = sum(mtime_results) + results['mtime_failed'] = len(mtime_results) - sum(mtime_results) + + _log("INFO", f"Completed async processing TV series: {series_path.name}") + + return { + "status": "success", + "imdb_id": imdb_id, + "episodes_found": len(disk_episodes), + "episodes_processed": len(episode_data_list), + "results": results + } + + async def async_process_multiple_series( + self, + series_paths: List[Path], + max_concurrent: int = 2 + ) -> List[Dict[str, Any]]: + """ + Process multiple TV series concurrently + + Args: + series_paths: List of series directory paths + max_concurrent: Maximum concurrent series processing + + Returns: + List of processing results for each series + """ + semaphore = asyncio.Semaphore(max_concurrent) + + async def _process_series_with_semaphore(series_path: Path) -> Dict[str, Any]: + async with semaphore: + try: + return await self.async_process_series(series_path) + except Exception as e: + _log("ERROR", f"Failed to process series {series_path}: {e}") + return {"status": "error", "path": str(series_path), "reason": str(e)} + + _log("INFO", f"Processing {len(series_paths)} series with max {max_concurrent} concurrent") + + tasks = [_process_series_with_semaphore(path) for path in series_paths] + results = await asyncio.gather(*tasks, return_exceptions=True) + + # Filter out exceptions and convert to proper results + processed_results = [] + for i, result in enumerate(results): + if isinstance(result, Exception): + processed_results.append({ + "status": "error", + "path": str(series_paths[i]), + "reason": str(result) + }) + else: + processed_results.append(result) + + return processed_results + + async def async_batch_episode_processing( + self, + episodes_data: List[Dict[str, Any]], + max_concurrent: int = 5 + ) -> Dict[str, Any]: + """ + Process episodes from webhook data concurrently + + Args: + episodes_data: List of episode data from webhooks + max_concurrent: Maximum concurrent episode processing + + Returns: + Processing results summary + """ + async def _process_single_episode(episode_data: Dict[str, Any]) -> Dict[str, Any]: + try: + # Extract episode information + series_path = Path(episode_data.get('series_path')) + season = episode_data.get('season') + episode = episode_data.get('episode') + aired = episode_data.get('aired') + dateadded = episode_data.get('dateadded') + + # Get IMDb ID + imdb_id = self.nfo_manager.parse_imdb_from_path(series_path) + if not imdb_id: + return {"status": "error", "reason": "No IMDb ID found"} + + # Create NFO if needed + nfo_success = True + if config.manage_nfo: + season_dir = series_path / config.tv_season_dir_format.format(season=season) + nfo_success = await self.async_nfo_manager.async_create_episode_nfo( + season_dir, season, episode, aired, dateadded, "webhook", config.lock_metadata + ) + + # Update database + self.db.upsert_episode_date(imdb_id, season, episode, aired, dateadded, "webhook", True) + + return { + "status": "success", + "season": season, + "episode": episode, + "nfo_created": nfo_success + } + + except Exception as e: + return { + "status": "error", + "episode_data": episode_data, + "reason": str(e) + } + + _log("INFO", f"Processing {len(episodes_data)} episodes concurrently") + + results = await async_concurrent_episode_processing( + episodes_data, + _process_single_episode, + max_concurrent + ) + + # Summarize results + successful = sum(1 for r in results if r and r.get("status") == "success") + failed = len(results) - successful + + return { + "total_episodes": len(episodes_data), + "successful": successful, + "failed": failed, + "results": results } \ No newline at end of file diff --git a/utils/async_file_utils.py b/utils/async_file_utils.py new file mode 100644 index 0000000..d1cea49 --- /dev/null +++ b/utils/async_file_utils.py @@ -0,0 +1,468 @@ +""" +Async file utilities for NFOGuard +High-performance async file operations with concurrent processing +""" +import asyncio +import aiofiles +import aiofiles.os +from pathlib import Path +from typing import List, Dict, Optional, Tuple, Any, Set +import xml.etree.ElementTree as ET +from datetime import datetime + +from utils.logging import _log +from utils.exceptions import FileOperationError, NFOCreationError +from utils.file_utils import VIDEO_EXTENSIONS, extract_episode_info, extract_imdb_id_from_path +from utils.nfo_patterns import parse_nfo_with_tolerance, write_nfo_file + + +async def async_read_text_file(file_path: Path, encoding: str = 'utf-8') -> Optional[str]: + """ + Async read text file with error handling + + Args: + file_path: Path to file to read + encoding: Text encoding (default: utf-8) + + Returns: + File content as string or None if error + """ + try: + async with aiofiles.open(file_path, 'r', encoding=encoding, errors='ignore') as f: + return await f.read() + except Exception as e: + _log("WARNING", f"Failed to read file {file_path}: {e}") + return None + + +async def async_write_text_file(file_path: Path, content: str, encoding: str = 'utf-8') -> bool: + """ + Async write text file with error handling + + Args: + file_path: Path to file to write + content: Content to write + encoding: Text encoding (default: utf-8) + + Returns: + True if successful, False otherwise + """ + try: + # Ensure parent directory exists + await aiofiles.os.makedirs(file_path.parent, exist_ok=True) + + async with aiofiles.open(file_path, 'w', encoding=encoding) as f: + await f.write(content) + return True + except Exception as e: + _log("ERROR", f"Failed to write file {file_path}: {e}") + return False + + +async def async_file_exists(file_path: Path) -> bool: + """ + Async check if file exists + + Args: + file_path: Path to check + + Returns: + True if file exists, False otherwise + """ + try: + return await aiofiles.os.path.exists(file_path) + except Exception: + return False + + +async def async_get_file_mtime(file_path: Path) -> Optional[float]: + """ + Async get file modification time + + Args: + file_path: Path to file + + Returns: + Modification time as timestamp or None if error + """ + try: + stat_result = await aiofiles.os.stat(file_path) + return stat_result.st_mtime + except Exception: + return None + + +async def async_set_file_mtime(file_path: Path, mtime: float) -> bool: + """ + Async set file modification time + + Args: + file_path: Path to file + mtime: New modification time as timestamp + + Returns: + True if successful, False otherwise + """ + try: + await aiofiles.os.utime(file_path, (mtime, mtime)) + return True + except Exception as e: + _log("WARNING", f"Failed to set mtime for {file_path}: {e}") + return False + + +async def async_find_video_files(directory: Path, recursive: bool = True) -> List[Path]: + """ + Async find all video files in a directory + + Args: + directory: Directory to search + recursive: Whether to search recursively + + Returns: + List of video file paths + """ + if not await async_file_exists(directory): + return [] + + video_files = [] + + try: + if recursive: + # Use os.walk equivalent for async + async def _walk_directory(path: Path): + try: + entries = await aiofiles.os.listdir(path) + for entry in entries: + entry_path = path / entry + if await aiofiles.os.path.isfile(entry_path): + if entry_path.suffix.lower() in VIDEO_EXTENSIONS: + video_files.append(entry_path) + elif await aiofiles.os.path.isdir(entry_path): + await _walk_directory(entry_path) + except Exception as e: + _log("WARNING", f"Failed to scan directory {path}: {e}") + + await _walk_directory(directory) + else: + # Non-recursive scan + try: + entries = await aiofiles.os.listdir(directory) + for entry in entries: + entry_path = directory / entry + if await aiofiles.os.path.isfile(entry_path): + if entry_path.suffix.lower() in VIDEO_EXTENSIONS: + video_files.append(entry_path) + except Exception as e: + _log("WARNING", f"Failed to scan directory {directory}: {e}") + + except Exception as e: + _log("ERROR", f"Failed to find video files in {directory}: {e}") + + return video_files + + +async def async_find_episodes_on_disk(series_path: Path) -> Dict[Tuple[int, int], List[Path]]: + """ + Async find all episodes on disk with concurrent processing + + Args: + series_path: Path to series directory + + Returns: + Dictionary mapping (season, episode) tuples to lists of video files + """ + episodes = {} + + if not await async_file_exists(series_path): + return episodes + + # Get all video files concurrently + video_files = await async_find_video_files(series_path, recursive=True) + + # Process files to extract episode information + for video_file in video_files: + episode_info = extract_episode_info(video_file.name) + if episode_info: + season, episode = episode_info["season"], episode_info["episode"] + key = (season, episode) + if key not in episodes: + episodes[key] = [] + episodes[key].append(video_file) + + return episodes + + +async def async_read_nfo_file(nfo_path: Path) -> Optional[ET.Element]: + """ + Async read and parse NFO file + + Args: + nfo_path: Path to NFO file + + Returns: + XML root element if successful, None otherwise + """ + if not await async_file_exists(nfo_path): + return None + + try: + content = await async_read_text_file(nfo_path) + if not content: + return None + + # Parse XML content + try: + root = ET.fromstring(content) + return root + except ET.ParseError: + # Try with tolerance (sync operation for now) + return parse_nfo_with_tolerance(nfo_path) + + except Exception as e: + _log("ERROR", f"Failed to read NFO file {nfo_path}: {e}") + return None + + +async def async_write_nfo_file( + nfo_path: Path, + root: ET.Element, + lock_metadata: bool = True +) -> bool: + """ + Async write NFO XML content to file + + Args: + nfo_path: Path where to write the NFO file + root: XML root element to write + lock_metadata: Whether to add file locking attributes + + Returns: + True if successful, False otherwise + """ + try: + # Ensure parent directory exists + await aiofiles.os.makedirs(nfo_path.parent, exist_ok=True) + + # Add file locking if requested + if lock_metadata: + root.set('nfoguard_managed', 'true') + root.set('last_updated', datetime.now().isoformat()) + + # Create tree and format + tree = ET.ElementTree(root) + ET.indent(tree, space=" ", level=0) # Pretty formatting + + # Convert to string + xml_str = ET.tostring(root, encoding='unicode', xml_declaration=False) + xml_content = f'\n{xml_str}' + + # Write asynchronously + success = await async_write_text_file(nfo_path, xml_content) + + if success: + _log("DEBUG", f"Successfully wrote NFO file: {nfo_path}") + + return success + + except Exception as e: + _log("ERROR", f"Failed to write NFO file {nfo_path}: {e}") + return False + + +async def async_batch_process_files( + file_paths: List[Path], + process_func, + max_concurrent: int = 10, + progress_callback: Optional[callable] = None +) -> List[Any]: + """ + Process multiple files concurrently with controlled concurrency + + Args: + file_paths: List of file paths to process + process_func: Async function to process each file + max_concurrent: Maximum number of concurrent operations + progress_callback: Optional callback for progress updates + + Returns: + List of results from processing each file + """ + semaphore = asyncio.Semaphore(max_concurrent) + results = [] + + async def _process_with_semaphore(file_path: Path, index: int) -> Any: + async with semaphore: + try: + result = await process_func(file_path) + if progress_callback: + progress_callback(index + 1, len(file_paths), file_path) + return result + except Exception as e: + _log("ERROR", f"Failed to process {file_path}: {e}") + return None + + # Create tasks for all files + tasks = [ + _process_with_semaphore(file_path, i) + for i, file_path in enumerate(file_paths) + ] + + # Execute all tasks concurrently with controlled concurrency + results = await asyncio.gather(*tasks, return_exceptions=True) + + return results + + +async def async_batch_nfo_operations( + nfo_operations: List[Dict[str, Any]], + max_concurrent: int = 5 +) -> List[bool]: + """ + Batch NFO operations (read/write) with controlled concurrency + + Args: + nfo_operations: List of operation dictionaries with 'type', 'path', and other params + max_concurrent: Maximum number of concurrent operations + + Returns: + List of success/failure results + """ + async def _execute_nfo_operation(operation: Dict[str, Any]) -> bool: + try: + op_type = operation.get('type') + path = operation.get('path') + + if op_type == 'read': + result = await async_read_nfo_file(path) + return result is not None + + elif op_type == 'write': + root = operation.get('root') + lock_metadata = operation.get('lock_metadata', True) + return await async_write_nfo_file(path, root, lock_metadata) + + else: + _log("ERROR", f"Unknown NFO operation type: {op_type}") + return False + + except Exception as e: + _log("ERROR", f"Failed to execute NFO operation: {e}") + return False + + return await async_batch_process_files( + [op.get('path') for op in nfo_operations], + lambda path: _execute_nfo_operation(next(op for op in nfo_operations if op.get('path') == path)), + max_concurrent + ) + + +async def async_concurrent_episode_processing( + episodes_data: List[Dict[str, Any]], + process_episode_func, + max_concurrent: int = 3 +) -> List[Any]: + """ + Process multiple episodes concurrently + + Args: + episodes_data: List of episode data dictionaries + process_episode_func: Async function to process each episode + max_concurrent: Maximum number of concurrent episode processes + + Returns: + List of processing results + """ + semaphore = asyncio.Semaphore(max_concurrent) + + async def _process_episode_with_semaphore(episode_data: Dict[str, Any]) -> Any: + async with semaphore: + try: + return await process_episode_func(episode_data) + except Exception as e: + _log("ERROR", f"Failed to process episode {episode_data}: {e}") + return None + + # Create tasks for all episodes + tasks = [ + _process_episode_with_semaphore(episode_data) + for episode_data in episodes_data + ] + + # Execute all tasks concurrently + results = await asyncio.gather(*tasks, return_exceptions=True) + + return results + + +async def async_directory_scan_with_stats( + directories: List[Path], + file_extensions: Optional[Set[str]] = None +) -> Dict[str, Any]: + """ + Async scan multiple directories and gather statistics + + Args: + directories: List of directories to scan + file_extensions: Optional set of file extensions to filter by + + Returns: + Dictionary with scan statistics and file lists + """ + if file_extensions is None: + file_extensions = VIDEO_EXTENSIONS + + stats = { + 'total_files': 0, + 'total_directories': len(directories), + 'files_by_directory': {}, + 'scan_errors': [], + 'total_size_bytes': 0 + } + + async def _scan_single_directory(directory: Path) -> Dict[str, Any]: + dir_stats = { + 'path': str(directory), + 'files': [], + 'file_count': 0, + 'size_bytes': 0, + 'error': None + } + + try: + if not await async_file_exists(directory): + dir_stats['error'] = 'Directory does not exist' + return dir_stats + + files = await async_find_video_files(directory, recursive=True) + dir_stats['files'] = [str(f) for f in files] + dir_stats['file_count'] = len(files) + + # Calculate total size + for file_path in files: + try: + stat_result = await aiofiles.os.stat(file_path) + dir_stats['size_bytes'] += stat_result.st_size + except Exception: + pass # Skip files we can't stat + + except Exception as e: + dir_stats['error'] = str(e) + stats['scan_errors'].append(f"{directory}: {e}") + + return dir_stats + + # Scan all directories concurrently + directory_results = await asyncio.gather( + *[_scan_single_directory(directory) for directory in directories], + return_exceptions=True + ) + + # Aggregate results + for result in directory_results: + if isinstance(result, dict) and not result.get('error'): + stats['files_by_directory'][result['path']] = result + stats['total_files'] += result['file_count'] + stats['total_size_bytes'] += result['size_bytes'] + + return stats \ No newline at end of file