"""Scheduler service for automatic library rescans. Uses APScheduler's AsyncIOScheduler with CronTrigger for precise cron-based scheduling. The legacy interval-based loop has been removed in favour of the cron approach. """ from __future__ import annotations from datetime import datetime, timezone from typing import List, Optional import structlog from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger from src.server.models.config import SchedulerConfig from src.server.services.config_service import ConfigServiceError, get_config_service logger = structlog.get_logger(__name__) _JOB_ID = "scheduled_rescan" class SchedulerServiceError(Exception): """Service-level exception for scheduler operations.""" class SchedulerService: """Manages automatic library rescans on a cron-based schedule. Uses APScheduler's AsyncIOScheduler so scheduling integrates cleanly with the running asyncio event loop. Supports: - Cron-based scheduling (time of day + days of week) - Immediate manual trigger - Live config reloading without app restart - Auto-queueing downloads of missing episodes after rescan """ def __init__(self) -> None: """Initialise the scheduler service.""" self._is_running: bool = False self._scheduler: Optional[AsyncIOScheduler] = None self._config: Optional[SchedulerConfig] = None self._last_scan_time: Optional[datetime] = None self._scan_in_progress: bool = False logger.info("SchedulerService initialised") # ------------------------------------------------------------------ # Public lifecycle methods # ------------------------------------------------------------------ async def start(self) -> None: """Start the APScheduler with the configured cron trigger. Raises: SchedulerServiceError: If the scheduler is already running or config cannot be loaded. """ if self._is_running: raise SchedulerServiceError("Scheduler is already running") try: config_service = get_config_service() config = config_service.load_config() self._config = config.scheduler except ConfigServiceError as exc: logger.error("Failed to load scheduler configuration", error=str(exc)) raise SchedulerServiceError(f"Failed to load config: {exc}") from exc self._scheduler = AsyncIOScheduler() if not self._config.enabled: logger.info("Scheduler is disabled in configuration — not adding jobs") self._is_running = True return trigger = self._build_cron_trigger() if trigger is None: logger.warning( "schedule_days is empty — scheduler started but no job scheduled" ) else: self._scheduler.add_job( self._perform_rescan, trigger=trigger, id=_JOB_ID, replace_existing=True, misfire_grace_time=300, ) logger.info( "Scheduler started with cron trigger", schedule_time=self._config.schedule_time, schedule_days=self._config.schedule_days, ) self._scheduler.start() self._is_running = True async def stop(self) -> None: """Stop the APScheduler gracefully.""" if not self._is_running: logger.debug("Scheduler stop called but not running") return if self._scheduler and self._scheduler.running: self._scheduler.shutdown(wait=False) logger.info("Scheduler stopped") self._is_running = False async def trigger_rescan(self) -> bool: """Manually trigger a library rescan. Returns: True if rescan was started; False if a scan is already running. Raises: SchedulerServiceError: If the scheduler service is not started. """ if not self._is_running: raise SchedulerServiceError("Scheduler is not running") if self._scan_in_progress: logger.warning("Cannot trigger rescan: scan already in progress") return False logger.info("Manual rescan triggered") await self._perform_rescan() return True def reload_config(self, config: SchedulerConfig) -> None: """Apply a new SchedulerConfig immediately. If the scheduler is already running the job is rescheduled (or removed) without stopping the scheduler. Args: config: New scheduler configuration to apply. """ self._config = config logger.info( "Scheduler config reloaded", enabled=config.enabled, schedule_time=config.schedule_time, schedule_days=config.schedule_days, auto_download=config.auto_download_after_rescan, ) if not self._scheduler or not self._scheduler.running: return if not config.enabled: if self._scheduler.get_job(_JOB_ID): self._scheduler.remove_job(_JOB_ID) logger.info("Scheduler job removed (disabled)") return trigger = self._build_cron_trigger() if trigger is None: if self._scheduler.get_job(_JOB_ID): self._scheduler.remove_job(_JOB_ID) logger.warning("Scheduler job removed — schedule_days is empty") else: if self._scheduler.get_job(_JOB_ID): self._scheduler.reschedule_job(_JOB_ID, trigger=trigger) logger.info( "Scheduler rescheduled with cron trigger", schedule_time=config.schedule_time, schedule_days=config.schedule_days, ) else: self._scheduler.add_job( self._perform_rescan, trigger=trigger, id=_JOB_ID, replace_existing=True, misfire_grace_time=300, ) logger.info( "Scheduler job added with cron trigger", schedule_time=config.schedule_time, schedule_days=config.schedule_days, ) def get_status(self) -> dict: """Return current scheduler status including cron configuration. Returns: Dict containing scheduler state and config fields. """ next_run: Optional[str] = None if self._scheduler and self._scheduler.running: job = self._scheduler.get_job(_JOB_ID) if job and job.next_run_time: next_run = job.next_run_time.isoformat() return { "is_running": self._is_running, "enabled": self._config.enabled if self._config else False, "interval_minutes": self._config.interval_minutes if self._config else None, "schedule_time": self._config.schedule_time if self._config else None, "schedule_days": self._config.schedule_days if self._config else [], "auto_download_after_rescan": ( self._config.auto_download_after_rescan if self._config else False ), "last_run": self._last_scan_time.isoformat() if self._last_scan_time else None, "next_run": next_run, "scan_in_progress": self._scan_in_progress, } # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ def _build_cron_trigger(self) -> Optional[CronTrigger]: """Convert config fields into an APScheduler CronTrigger. Returns: CronTrigger instance or None if schedule_days is empty. """ if not self._config or not self._config.schedule_days: return None hour_str, minute_str = self._config.schedule_time.split(":") day_of_week = ",".join(self._config.schedule_days) trigger = CronTrigger( hour=int(hour_str), minute=int(minute_str), day_of_week=day_of_week, ) logger.debug( "CronTrigger built", hour=hour_str, minute=minute_str, day_of_week=day_of_week, ) return trigger async def _broadcast(self, event_type: str, data: dict) -> None: """Broadcast a WebSocket event to all connected clients.""" try: from src.server.services.websocket_service import ( # noqa: PLC0415 get_websocket_service, ) ws_service = get_websocket_service() await ws_service.manager.broadcast({"type": event_type, "data": data}) except Exception as exc: # pylint: disable=broad-exception-caught logger.warning("WebSocket broadcast failed", event=event_type, error=str(exc)) async def _auto_download_missing(self) -> None: """Queue and start downloads for all series with missing episodes.""" from src.server.models.download import EpisodeIdentifier # noqa: PLC0415 from src.server.utils.dependencies import ( # noqa: PLC0415 get_anime_service, get_download_service, ) anime_service = get_anime_service() download_service = get_download_service() series_list = anime_service._cached_list_missing() queued_count = 0 for series in series_list: episode_dict: dict = series.get("episodeDict") or {} if not episode_dict: continue episodes: List[EpisodeIdentifier] = [] for season_str, ep_numbers in episode_dict.items(): for ep_num in ep_numbers: episodes.append( EpisodeIdentifier(season=int(season_str), episode=int(ep_num)) ) if not episodes: continue await download_service.add_to_queue( serie_id=series.get("key", ""), serie_folder=series.get("folder", series.get("name", "")), serie_name=series.get("name", ""), episodes=episodes, ) queued_count += len(episodes) logger.info( "Auto-download queued episodes", series=series.get("key"), count=len(episodes), ) if queued_count: await download_service.start_queue_processing() logger.info("Auto-download queue processing started", queued=queued_count) await self._broadcast("auto_download_started", {"queued_count": queued_count}) logger.info("Auto-download completed", queued_count=queued_count) async def _perform_rescan(self) -> None: """Execute a library rescan and optionally trigger auto-download.""" if self._scan_in_progress: logger.warning("Skipping rescan: previous scan still in progress") return self._scan_in_progress = True scan_start = datetime.now(timezone.utc) try: logger.info("Starting scheduled library rescan") from src.server.utils.dependencies import get_anime_service # noqa: PLC0415 anime_service = get_anime_service() await self._broadcast( "scheduled_rescan_started", {"timestamp": scan_start.isoformat()}, ) await anime_service.rescan() self._last_scan_time = datetime.now(timezone.utc) duration = (self._last_scan_time - scan_start).total_seconds() logger.info("Scheduled library rescan completed", duration_seconds=duration) await self._broadcast( "scheduled_rescan_completed", { "timestamp": self._last_scan_time.isoformat(), "duration_seconds": duration, }, ) # Auto-download after rescan if self._config and self._config.auto_download_after_rescan: logger.info("Auto-download after rescan is enabled — starting") try: await self._auto_download_missing() except Exception as dl_exc: # pylint: disable=broad-exception-caught logger.error( "Auto-download after rescan failed", error=str(dl_exc), exc_info=True, ) await self._broadcast( "auto_download_error", {"error": str(dl_exc)} ) else: logger.debug("Auto-download after rescan is disabled — skipping") except Exception as exc: # pylint: disable=broad-exception-caught logger.error("Scheduled rescan failed", error=str(exc), exc_info=True) await self._broadcast( "scheduled_rescan_error", {"error": str(exc), "timestamp": datetime.now(timezone.utc).isoformat()}, ) finally: self._scan_in_progress = False # --------------------------------------------------------------------------- # Module-level singleton # --------------------------------------------------------------------------- _scheduler_service: Optional[SchedulerService] = None def get_scheduler_service() -> SchedulerService: """Return the singleton SchedulerService instance.""" global _scheduler_service if _scheduler_service is None: _scheduler_service = SchedulerService() return _scheduler_service def reset_scheduler_service() -> None: """Reset the singleton (used in tests).""" global _scheduler_service _scheduler_service = None