Merge pull request 'feat: implement comprehensive async I/O capabilities' (#17) from async into dev

Reviewed-on: #17

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

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
         }

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'<?xml version="1.0" encoding="utf-8"?>\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