feat: implement comprehensive async I/O capabilities
Local Docker Build (Dev) / build-dev (pull_request) Successful in 20s
Local Docker Build (Dev) / build-dev (pull_request) Successful in 20s
- Add utils/async_file_utils.py with aiofiles-based operations - Create core/async_nfo_manager.py for concurrent NFO processing - Add async methods to TVProcessor for concurrent episode handling - Implement async_find_episodes_on_disk() for faster scanning - Add async_batch_create_episode_nfos() for concurrent NFO creation - Support async_process_multiple_series() for parallel processing - Optimize I/O performance for large media libraries - Enable controlled concurrency with semaphores and rate limiting Performance improvements: - Concurrent file operations reduce I/O wait times - Batch processing minimizes individual file access overhead - Configurable concurrency limits prevent system overload - Async directory scanning for faster episode discovery
This commit is contained in:
+213
-1
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user