Lukas
b1726968e5
- Replace callback system with events library in SerieScanner - Update SeriesApp to subscribe to loader and scanner events - Refactor ScanService to use Events instead of CallbackManager - Remove CallbackManager imports and callback classes - Add safe event calling with error handling in SerieScanner - Update AniworldLoader to use Events for download progress - Remove progress_callback parameter from download methods - Update all affected tests for Events pattern - Fix test_series_app.py for new event subscription model - Comment out obsolete callback tests in test_scan_service.py All core tests passing. Events provide cleaner event-driven architecture.
522 lines
17 KiB
Python
522 lines
17 KiB
Python
"""Scan service for managing anime library scan operations.
|
|
|
|
This module provides a service layer for scanning the anime library directory,
|
|
identifying missing episodes, and broadcasting scan progress updates.
|
|
All scan operations use 'key' as the primary series identifier.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Callable, Dict, List, Optional
|
|
|
|
import structlog
|
|
|
|
from src.server.services.progress_service import (
|
|
ProgressService,
|
|
ProgressType,
|
|
get_progress_service,
|
|
)
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
|
|
class ScanServiceError(Exception):
|
|
"""Service-level exception for scan operations."""
|
|
|
|
|
|
class ScanProgress:
|
|
"""Represents the current state of a scan operation.
|
|
|
|
Attributes:
|
|
scan_id: Unique identifier for this scan operation
|
|
status: Current status (started, in_progress, completed, failed)
|
|
current: Number of folders processed
|
|
total: Total number of folders to process
|
|
percentage: Completion percentage
|
|
message: Human-readable progress message
|
|
key: Current series key being scanned (if applicable)
|
|
folder: Current folder being scanned (metadata only)
|
|
started_at: When the scan started
|
|
updated_at: When the progress was last updated
|
|
series_found: Number of series found with missing episodes
|
|
errors: List of error messages encountered
|
|
"""
|
|
|
|
def __init__(self, scan_id: str):
|
|
"""Initialize scan progress.
|
|
|
|
Args:
|
|
scan_id: Unique identifier for this scan
|
|
"""
|
|
self.scan_id = scan_id
|
|
self.status = "started"
|
|
self.current = 0
|
|
self.total = 0
|
|
self.percentage = 0.0
|
|
self.message = "Initializing scan..."
|
|
self.key: Optional[str] = None
|
|
self.folder: Optional[str] = None
|
|
self.started_at = datetime.now(timezone.utc)
|
|
self.updated_at = datetime.now(timezone.utc)
|
|
self.series_found = 0
|
|
self.errors: List[str] = []
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert to dictionary for serialization.
|
|
|
|
Returns:
|
|
Dictionary representation with 'key' as primary identifier
|
|
and 'folder' as metadata only.
|
|
"""
|
|
result = {
|
|
"scan_id": self.scan_id,
|
|
"status": self.status,
|
|
"current": self.current,
|
|
"total": self.total,
|
|
"percentage": round(self.percentage, 2),
|
|
"message": self.message,
|
|
"started_at": self.started_at.isoformat(),
|
|
"updated_at": self.updated_at.isoformat(),
|
|
"series_found": self.series_found,
|
|
"errors": self.errors,
|
|
}
|
|
|
|
# Include optional series identifiers
|
|
if self.key is not None:
|
|
result["key"] = self.key
|
|
if self.folder is not None:
|
|
result["folder"] = self.folder
|
|
|
|
return result
|
|
|
|
|
|
class ScanService:
|
|
"""Manages anime library scan operations.
|
|
|
|
Features:
|
|
- Trigger library scans
|
|
- Track scan progress in real-time
|
|
- Use 'key' as primary series identifier
|
|
- Broadcast scan progress via WebSocket
|
|
- Handle scan errors gracefully
|
|
- Provide scan history and statistics
|
|
|
|
All operations use 'key' (provider-assigned, URL-safe identifier)
|
|
as the primary series identifier. 'folder' is used only as metadata
|
|
for display and filesystem operations.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
progress_service: Optional[ProgressService] = None,
|
|
):
|
|
"""Initialize the scan service.
|
|
|
|
Args:
|
|
progress_service: Optional progress service for tracking
|
|
"""
|
|
self._progress_service = progress_service or get_progress_service()
|
|
|
|
# Current scan state
|
|
self._current_scan: Optional[ScanProgress] = None
|
|
self._is_scanning = False
|
|
|
|
# Scan history (limited size)
|
|
self._scan_history: List[ScanProgress] = []
|
|
self._max_history_size = 10
|
|
|
|
# Event handlers for scan events
|
|
self._scan_event_handlers: List[
|
|
Callable[[Dict[str, Any]], None]
|
|
] = []
|
|
|
|
# Lock for thread-safe operations
|
|
self._lock = asyncio.Lock()
|
|
|
|
logger.info("ScanService initialized")
|
|
|
|
def subscribe_to_scan_events(
|
|
self,
|
|
handler: Callable[[Dict[str, Any]], None],
|
|
) -> None:
|
|
"""Subscribe to scan events.
|
|
|
|
Args:
|
|
handler: Function to call when scan events occur.
|
|
Receives a dictionary with event data including
|
|
'key' as the primary identifier.
|
|
"""
|
|
self._scan_event_handlers.append(handler)
|
|
logger.debug("Scan event handler subscribed")
|
|
|
|
def unsubscribe_from_scan_events(
|
|
self,
|
|
handler: Callable[[Dict[str, Any]], None],
|
|
) -> None:
|
|
"""Unsubscribe from scan events.
|
|
|
|
Args:
|
|
handler: Handler function to remove
|
|
"""
|
|
try:
|
|
self._scan_event_handlers.remove(handler)
|
|
logger.debug("Scan event handler unsubscribed")
|
|
except ValueError:
|
|
logger.warning("Handler not found for unsubscribe")
|
|
|
|
async def _emit_scan_event(self, event_data: Dict[str, Any]) -> None:
|
|
"""Emit scan event to all subscribers.
|
|
|
|
Args:
|
|
event_data: Event data to broadcast, includes 'key' as
|
|
primary identifier and 'folder' as metadata
|
|
"""
|
|
for handler in self._scan_event_handlers:
|
|
try:
|
|
if asyncio.iscoroutinefunction(handler):
|
|
await handler(event_data)
|
|
else:
|
|
handler(event_data)
|
|
except Exception as e:
|
|
logger.error(
|
|
"Scan event handler error",
|
|
error=str(e),
|
|
)
|
|
|
|
@property
|
|
def is_scanning(self) -> bool:
|
|
"""Check if a scan is currently in progress."""
|
|
return self._is_scanning
|
|
|
|
@property
|
|
def current_scan(self) -> Optional[ScanProgress]:
|
|
"""Get the current scan progress."""
|
|
return self._current_scan
|
|
|
|
async def start_scan(
|
|
self,
|
|
scanner: Any, # SerieScanner instance
|
|
) -> str:
|
|
"""Start a new library scan.
|
|
|
|
Args:
|
|
scanner: SerieScanner instance to use for scanning.
|
|
The service will subscribe to its events.
|
|
|
|
Returns:
|
|
Scan ID for tracking
|
|
|
|
Raises:
|
|
ScanServiceError: If a scan is already in progress
|
|
|
|
Note:
|
|
The scan uses 'key' as the primary identifier for all series.
|
|
The 'folder' field is included only as metadata for display.
|
|
"""
|
|
async with self._lock:
|
|
if self._is_scanning:
|
|
raise ScanServiceError("A scan is already in progress")
|
|
|
|
self._is_scanning = True
|
|
|
|
scan_id = str(uuid.uuid4())
|
|
scan_progress = ScanProgress(scan_id)
|
|
self._current_scan = scan_progress
|
|
|
|
logger.info("Starting library scan", scan_id=scan_id)
|
|
|
|
# Start progress tracking
|
|
try:
|
|
await self._progress_service.start_progress(
|
|
progress_id=f"scan_{scan_id}",
|
|
progress_type=ProgressType.SCAN,
|
|
title="Library Scan",
|
|
message="Initializing scan...",
|
|
)
|
|
except Exception as e:
|
|
logger.error("Failed to start progress tracking: %s", e)
|
|
|
|
# Emit scan started event
|
|
await self._emit_scan_event({
|
|
"type": "scan_started",
|
|
"scan_id": scan_id,
|
|
"message": "Library scan started",
|
|
})
|
|
|
|
# Create event handlers for the scanner
|
|
def on_progress_handler(progress_data: Dict[str, Any]) -> None:
|
|
"""Handle progress events from scanner."""
|
|
scan_progress.current = progress_data.get('current', 0)
|
|
scan_progress.total = progress_data.get('total', 0)
|
|
scan_progress.percentage = progress_data.get('percentage', 0.0)
|
|
scan_progress.message = progress_data.get('message', '')
|
|
scan_progress.updated_at = datetime.now(timezone.utc)
|
|
|
|
phase = progress_data.get('phase', '')
|
|
if phase == 'STARTING':
|
|
scan_progress.status = "started"
|
|
elif phase == 'IN_PROGRESS':
|
|
scan_progress.status = "in_progress"
|
|
|
|
# Schedule the progress update on the event loop
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
asyncio.run_coroutine_threadsafe(
|
|
self._handle_progress_update(scan_progress),
|
|
loop
|
|
)
|
|
except RuntimeError:
|
|
pass
|
|
|
|
def on_error_handler(error_data: Dict[str, Any]) -> None:
|
|
"""Handle error events from scanner."""
|
|
error_msg = error_data.get('message', 'Unknown error')
|
|
scan_progress.errors.append(error_msg)
|
|
scan_progress.updated_at = datetime.now(timezone.utc)
|
|
|
|
logger.warning(
|
|
"Scan error",
|
|
error=str(error_data.get('error')),
|
|
recoverable=error_data.get('recoverable', True),
|
|
)
|
|
|
|
# Schedule the error handling on the event loop
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
asyncio.run_coroutine_threadsafe(
|
|
self._handle_scan_error(scan_progress, error_data),
|
|
loop
|
|
)
|
|
except RuntimeError:
|
|
pass
|
|
|
|
def on_completion_handler(completion_data: Dict[str, Any]) -> None:
|
|
"""Handle completion events from scanner."""
|
|
success = completion_data.get('success', False)
|
|
scan_progress.status = "completed" if success else "failed"
|
|
scan_progress.message = completion_data.get('message', '')
|
|
scan_progress.updated_at = datetime.now(timezone.utc)
|
|
|
|
if 'statistics' in completion_data:
|
|
stats = completion_data['statistics']
|
|
scan_progress.series_found = stats.get('series_found', 0)
|
|
|
|
# Schedule the completion handling on the event loop
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
asyncio.run_coroutine_threadsafe(
|
|
self._handle_scan_completion(scan_progress, completion_data),
|
|
loop
|
|
)
|
|
except RuntimeError:
|
|
pass
|
|
|
|
# Subscribe to scanner events
|
|
scanner.subscribe_on_progress(on_progress_handler)
|
|
scanner.subscribe_on_error(on_error_handler)
|
|
scanner.subscribe_on_completion(on_completion_handler)
|
|
|
|
return scan_id
|
|
|
|
async def _handle_progress_update(
|
|
self,
|
|
scan_progress: ScanProgress,
|
|
) -> None:
|
|
"""Handle a scan progress update.
|
|
|
|
Args:
|
|
scan_progress: Updated scan progress with 'key' as identifier
|
|
"""
|
|
# Update progress service
|
|
try:
|
|
await self._progress_service.update_progress(
|
|
progress_id=f"scan_{scan_progress.scan_id}",
|
|
current=scan_progress.current,
|
|
total=scan_progress.total,
|
|
message=scan_progress.message,
|
|
)
|
|
except Exception as e:
|
|
logger.debug("Progress update skipped: %s", e)
|
|
|
|
# Emit progress event with key as primary identifier
|
|
await self._emit_scan_event({
|
|
"type": "scan_progress",
|
|
"data": scan_progress.to_dict(),
|
|
})
|
|
|
|
async def _handle_scan_error(
|
|
self,
|
|
scan_progress: ScanProgress,
|
|
error_data: Dict[str, Any],
|
|
) -> None:
|
|
"""Handle a scan error.
|
|
|
|
Args:
|
|
scan_progress: Current scan progress
|
|
error_data: Error data dictionary with error info
|
|
"""
|
|
# Emit error event with key as primary identifier
|
|
await self._emit_scan_event({
|
|
"type": "scan_error",
|
|
"scan_id": scan_progress.scan_id,
|
|
"error": str(error_data.get('error')),
|
|
"message": error_data.get('message', 'Unknown error'),
|
|
"recoverable": error_data.get('recoverable', True),
|
|
})
|
|
|
|
async def _handle_scan_completion(
|
|
self,
|
|
scan_progress: ScanProgress,
|
|
completion_data: Dict[str, Any],
|
|
) -> None:
|
|
"""Handle scan completion.
|
|
|
|
Args:
|
|
scan_progress: Final scan progress
|
|
completion_data: Completion data dictionary with statistics
|
|
"""
|
|
success = completion_data.get('success', False)
|
|
message = completion_data.get('message', '')
|
|
statistics = completion_data.get('statistics', {})
|
|
|
|
async with self._lock:
|
|
self._is_scanning = False
|
|
|
|
# Add to history
|
|
self._scan_history.append(scan_progress)
|
|
if len(self._scan_history) > self._max_history_size:
|
|
self._scan_history.pop(0)
|
|
|
|
# Complete progress tracking
|
|
try:
|
|
if success:
|
|
await self._progress_service.complete_progress(
|
|
progress_id=f"scan_{scan_progress.scan_id}",
|
|
message=message,
|
|
)
|
|
else:
|
|
await self._progress_service.fail_progress(
|
|
progress_id=f"scan_{scan_progress.scan_id}",
|
|
error_message=message,
|
|
)
|
|
except Exception as e:
|
|
logger.debug("Progress completion skipped: %s", e)
|
|
|
|
# Emit completion event
|
|
await self._emit_scan_event({
|
|
"type": "scan_completed" if success else "scan_failed",
|
|
"scan_id": scan_progress.scan_id,
|
|
"success": success,
|
|
"message": message,
|
|
"statistics": statistics,
|
|
"data": scan_progress.to_dict(),
|
|
})
|
|
|
|
logger.info(
|
|
"Scan completed",
|
|
scan_id=scan_progress.scan_id,
|
|
success=success,
|
|
series_found=scan_progress.series_found,
|
|
errors_count=len(scan_progress.errors),
|
|
)
|
|
|
|
async def cancel_scan(self) -> bool:
|
|
"""Cancel the current scan if one is in progress.
|
|
|
|
Returns:
|
|
True if scan was cancelled, False if no scan in progress
|
|
"""
|
|
async with self._lock:
|
|
if not self._is_scanning:
|
|
return False
|
|
|
|
self._is_scanning = False
|
|
|
|
if self._current_scan:
|
|
self._current_scan.status = "cancelled"
|
|
self._current_scan.message = "Scan cancelled by user"
|
|
self._current_scan.updated_at = datetime.now(timezone.utc)
|
|
|
|
# Add to history
|
|
self._scan_history.append(self._current_scan)
|
|
if len(self._scan_history) > self._max_history_size:
|
|
self._scan_history.pop(0)
|
|
|
|
# Emit cancellation event
|
|
if self._current_scan:
|
|
await self._emit_scan_event({
|
|
"type": "scan_cancelled",
|
|
"scan_id": self._current_scan.scan_id,
|
|
"message": "Scan cancelled by user",
|
|
})
|
|
|
|
# Update progress service
|
|
try:
|
|
await self._progress_service.fail_progress(
|
|
progress_id=f"scan_{self._current_scan.scan_id}",
|
|
error_message="Scan cancelled by user",
|
|
)
|
|
except Exception as e:
|
|
logger.debug("Progress cancellation skipped: %s", e)
|
|
|
|
logger.info("Scan cancelled")
|
|
return True
|
|
|
|
async def get_scan_status(self) -> Dict[str, Any]:
|
|
"""Get the current scan status.
|
|
|
|
Returns:
|
|
Dictionary with scan status information, including 'key'
|
|
as the primary series identifier for any current scan.
|
|
"""
|
|
return {
|
|
"is_scanning": self._is_scanning,
|
|
"current_scan": (
|
|
self._current_scan.to_dict() if self._current_scan else None
|
|
),
|
|
}
|
|
|
|
async def get_scan_history(
|
|
self,
|
|
limit: int = 10,
|
|
) -> List[Dict[str, Any]]:
|
|
"""Get scan history.
|
|
|
|
Args:
|
|
limit: Maximum number of history entries to return
|
|
|
|
Returns:
|
|
List of scan history entries, newest first.
|
|
Each entry includes 'key' as the primary identifier.
|
|
"""
|
|
history = self._scan_history[-limit:]
|
|
history.reverse() # Newest first
|
|
return [scan.to_dict() for scan in history]
|
|
|
|
|
|
# Module-level singleton instance
|
|
_scan_service: Optional[ScanService] = None
|
|
|
|
|
|
def get_scan_service() -> ScanService:
|
|
"""Get the singleton ScanService instance.
|
|
|
|
Returns:
|
|
The ScanService singleton
|
|
"""
|
|
global _scan_service
|
|
if _scan_service is None:
|
|
_scan_service = ScanService()
|
|
return _scan_service
|
|
|
|
|
|
def reset_scan_service() -> None:
|
|
"""Reset the singleton ScanService instance.
|
|
|
|
Primarily used for testing to ensure clean state.
|
|
"""
|
|
global _scan_service
|
|
_scan_service = None
|