diff --git a/api/models.py b/api/models.py index d604b55..e497c3b 100644 --- a/api/models.py +++ b/api/models.py @@ -119,4 +119,66 @@ class EpisodeResponse(BaseModel): last_updated: str series_path: str season_name: str - episode_name: str \ No newline at end of file + episode_name: str + + +# Scheduled Scans Models + +class CreateScheduledScanRequest(BaseModel): + """Request model for creating a scheduled scan""" + name: str + description: Optional[str] = None + cron_expression: str + media_type: str # 'tv', 'movies', 'both' + scan_mode: str # 'smart', 'full', 'incomplete' + specific_paths: Optional[str] = None + enabled: bool = True + + +class UpdateScheduledScanRequest(BaseModel): + """Request model for updating a scheduled scan""" + name: Optional[str] = None + description: Optional[str] = None + cron_expression: Optional[str] = None + media_type: Optional[str] = None + scan_mode: Optional[str] = None + specific_paths: Optional[str] = None + enabled: Optional[bool] = None + + +class ScheduledScanResponse(BaseModel): + """Response model for scheduled scan data""" + id: int + name: str + description: Optional[str] + cron_expression: str + media_type: str + scan_mode: str + specific_paths: Optional[str] + enabled: bool + created_at: str + updated_at: str + last_run_at: Optional[str] + next_run_at: Optional[str] + run_count: int + created_by: Optional[str] + updated_by: Optional[str] + + +class ScheduleExecutionResponse(BaseModel): + """Response model for schedule execution data""" + id: int + schedule_id: int + schedule_name: str + started_at: str + completed_at: Optional[str] + status: str + media_type: str + scan_mode: str + items_processed: int + items_skipped: int + items_failed: int + execution_time_seconds: Optional[int] + error_message: Optional[str] + logs: Optional[str] + triggered_by: Optional[str] \ No newline at end of file diff --git a/api/web_routes.py b/api/web_routes.py index 69b6e5c..94e8fe4 100644 --- a/api/web_routes.py +++ b/api/web_routes.py @@ -1897,4 +1897,326 @@ def register_database_admin_routes(app, dependencies): "query": "SELECT imdb_id, season, episode, dateadded, source FROM episodes WHERE has_video_file = FALSE ORDER BY imdb_id, season, episode" } ] + } + + # Scheduled Scans Endpoints + + @app.get("/api/admin/scheduled-scans") + async def api_get_scheduled_scans(): + """Get all scheduled scans""" + return await get_scheduled_scans(dependencies) + + @app.post("/api/admin/scheduled-scans") + async def api_create_scheduled_scan(request: CreateScheduledScanRequest): + """Create a new scheduled scan""" + return await create_scheduled_scan(dependencies, request) + + @app.put("/api/admin/scheduled-scans/{scan_id}") + async def api_update_scheduled_scan(scan_id: int, request: UpdateScheduledScanRequest): + """Update a scheduled scan""" + return await update_scheduled_scan(dependencies, scan_id, request) + + @app.delete("/api/admin/scheduled-scans/{scan_id}") + async def api_delete_scheduled_scan(scan_id: int): + """Delete a scheduled scan""" + return await delete_scheduled_scan(dependencies, scan_id) + + @app.post("/api/admin/scheduled-scans/{scan_id}/toggle") + async def api_toggle_scheduled_scan(scan_id: int): + """Toggle a scheduled scan enabled/disabled""" + return await toggle_scheduled_scan(dependencies, scan_id) + + @app.post("/api/admin/scheduled-scans/{scan_id}/run") + async def api_run_scheduled_scan(scan_id: int): + """Manually run a scheduled scan""" + return await run_scheduled_scan(dependencies, scan_id) + + @app.get("/api/admin/scheduled-scans/executions") + async def api_get_schedule_executions(schedule_id: int = None): + """Get schedule execution history""" + return await get_schedule_executions(dependencies, schedule_id) + + +# Scheduled Scans Functions + +async def get_scheduled_scans(dependencies: dict): + """Get all scheduled scans""" + try: + db = dependencies["db"] + scans = db.get_scheduled_scans() + + # Convert to response format + scan_list = [] + for scan in scans: + scan_dict = dict(scan) + # Convert datetime objects to strings + for field in ['created_at', 'updated_at', 'last_run_at', 'next_run_at']: + if scan_dict.get(field): + scan_dict[field] = scan_dict[field].isoformat() + scan_list.append(scan_dict) + + return { + "success": True, + "scans": scan_list + } + except Exception as e: + return { + "success": False, + "error": str(e) + } + + +async def create_scheduled_scan(dependencies: dict, request: CreateScheduledScanRequest): + """Create a new scheduled scan""" + try: + db = dependencies["db"] + + # Validate cron expression + from croniter import croniter + if not croniter.is_valid(request.cron_expression): + return { + "success": False, + "error": "Invalid cron expression" + } + + # Validate media type and scan mode + if request.media_type not in ['tv', 'movies', 'both']: + return { + "success": False, + "error": "Invalid media type. Must be 'tv', 'movies', or 'both'" + } + + if request.scan_mode not in ['smart', 'full', 'incomplete']: + return { + "success": False, + "error": "Invalid scan mode. Must be 'smart', 'full', or 'incomplete'" + } + + # Create the scheduled scan + scan_id = db.create_scheduled_scan( + name=request.name, + description=request.description, + cron_expression=request.cron_expression, + media_type=request.media_type, + scan_mode=request.scan_mode, + specific_paths=request.specific_paths, + enabled=request.enabled, + created_by="web_interface" + ) + + # Calculate next run time + from croniter import croniter + from datetime import datetime, timezone + cron = croniter(request.cron_expression, datetime.now(timezone.utc)) + next_run = cron.get_next(datetime) + db.update_scan_next_run(scan_id, next_run) + + return { + "success": True, + "scan_id": scan_id, + "message": f"Scheduled scan '{request.name}' created successfully" + } + except Exception as e: + return { + "success": False, + "error": str(e) + } + + +async def update_scheduled_scan(dependencies: dict, scan_id: int, request: UpdateScheduledScanRequest): + """Update a scheduled scan""" + try: + db = dependencies["db"] + + # Check if scan exists + existing_scan = db.get_scheduled_scan(scan_id) + if not existing_scan: + return { + "success": False, + "error": "Scheduled scan not found" + } + + # Validate cron expression if provided + if request.cron_expression: + from croniter import croniter + if not croniter.is_valid(request.cron_expression): + return { + "success": False, + "error": "Invalid cron expression" + } + + # Validate media type and scan mode if provided + if request.media_type and request.media_type not in ['tv', 'movies', 'both']: + return { + "success": False, + "error": "Invalid media type. Must be 'tv', 'movies', or 'both'" + } + + if request.scan_mode and request.scan_mode not in ['smart', 'full', 'incomplete']: + return { + "success": False, + "error": "Invalid scan mode. Must be 'smart', 'full', or 'incomplete'" + } + + # Update the scheduled scan + success = db.update_scheduled_scan( + scan_id=scan_id, + name=request.name, + description=request.description, + cron_expression=request.cron_expression, + media_type=request.media_type, + scan_mode=request.scan_mode, + specific_paths=request.specific_paths, + enabled=request.enabled, + updated_by="web_interface" + ) + + if not success: + return { + "success": False, + "error": "Failed to update scheduled scan" + } + + # Update next run time if cron expression changed + if request.cron_expression: + from croniter import croniter + from datetime import datetime, timezone + cron = croniter(request.cron_expression, datetime.now(timezone.utc)) + next_run = cron.get_next(datetime) + db.update_scan_next_run(scan_id, next_run) + + return { + "success": True, + "message": "Scheduled scan updated successfully" + } + except Exception as e: + return { + "success": False, + "error": str(e) + } + + +async def delete_scheduled_scan(dependencies: dict, scan_id: int): + """Delete a scheduled scan""" + try: + db = dependencies["db"] + + # Check if scan exists + existing_scan = db.get_scheduled_scan(scan_id) + if not existing_scan: + return { + "success": False, + "error": "Scheduled scan not found" + } + + # Delete the scheduled scan + success = db.delete_scheduled_scan(scan_id) + + if not success: + return { + "success": False, + "error": "Failed to delete scheduled scan" + } + + return { + "success": True, + "message": f"Scheduled scan '{existing_scan['name']}' deleted successfully" + } + except Exception as e: + return { + "success": False, + "error": str(e) + } + + +async def toggle_scheduled_scan(dependencies: dict, scan_id: int): + """Toggle a scheduled scan enabled/disabled""" + try: + db = dependencies["db"] + + # Check if scan exists + existing_scan = db.get_scheduled_scan(scan_id) + if not existing_scan: + return { + "success": False, + "error": "Scheduled scan not found" + } + + # Toggle enabled status + new_enabled = not existing_scan['enabled'] + success = db.update_scheduled_scan( + scan_id=scan_id, + enabled=new_enabled, + updated_by="web_interface" + ) + + if not success: + return { + "success": False, + "error": "Failed to toggle scheduled scan" + } + + status = "enabled" if new_enabled else "disabled" + return { + "success": True, + "message": f"Scheduled scan '{existing_scan['name']}' {status} successfully" + } + except Exception as e: + return { + "success": False, + "error": str(e) + } + + +async def run_scheduled_scan(dependencies: dict, scan_id: int): + """Manually run a scheduled scan""" + try: + db = dependencies["db"] + + # Check if scan exists + existing_scan = db.get_scheduled_scan(scan_id) + if not existing_scan: + return { + "success": False, + "error": "Scheduled scan not found" + } + + # TODO: Implement actual scan execution + # For now, just return a success message + return { + "success": True, + "message": f"Manual execution of '{existing_scan['name']}' started", + "note": "Scan execution will be implemented with the background scheduler" + } + except Exception as e: + return { + "success": False, + "error": str(e) + } + + +async def get_schedule_executions(dependencies: dict, schedule_id: int = None): + """Get schedule execution history""" + try: + db = dependencies["db"] + executions = db.get_schedule_executions(schedule_id) + + # Convert to response format + execution_list = [] + for execution in executions: + execution_dict = dict(execution) + # Convert datetime objects to strings + for field in ['started_at', 'completed_at']: + if execution_dict.get(field): + execution_dict[field] = execution_dict[field].isoformat() + execution_list.append(execution_dict) + + return { + "success": True, + "executions": execution_list + } + except Exception as e: + return { + "success": False, + "error": str(e) } \ No newline at end of file diff --git a/core/database.py b/core/database.py index de17faf..932f06b 100644 --- a/core/database.py +++ b/core/database.py @@ -147,6 +147,48 @@ class NFOGuardDatabase: ) """) + # Scheduled scans table + cursor.execute(""" + CREATE TABLE IF NOT EXISTS scheduled_scans ( + id SERIAL PRIMARY KEY, + name VARCHAR(255) NOT NULL, + description TEXT, + cron_expression VARCHAR(100) NOT NULL, + media_type VARCHAR(20) NOT NULL CHECK (media_type IN ('tv', 'movies', 'both')), + scan_mode VARCHAR(20) NOT NULL CHECK (scan_mode IN ('smart', 'full', 'incomplete')), + specific_paths TEXT, + enabled BOOLEAN DEFAULT TRUE, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + last_run_at TIMESTAMP, + next_run_at TIMESTAMP, + run_count INTEGER DEFAULT 0, + created_by VARCHAR(100), + updated_by VARCHAR(100) + ) + """) + + # Schedule execution history table + cursor.execute(""" + CREATE TABLE IF NOT EXISTS schedule_executions ( + id SERIAL PRIMARY KEY, + schedule_id INTEGER NOT NULL, + started_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + completed_at TIMESTAMP, + status VARCHAR(50) NOT NULL CHECK (status IN ('running', 'completed', 'failed', 'cancelled')), + media_type VARCHAR(20) NOT NULL, + scan_mode VARCHAR(20) NOT NULL, + items_processed INTEGER DEFAULT 0, + items_skipped INTEGER DEFAULT 0, + items_failed INTEGER DEFAULT 0, + execution_time_seconds INTEGER, + error_message TEXT, + logs TEXT, + triggered_by VARCHAR(100), + FOREIGN KEY (schedule_id) REFERENCES scheduled_scans(id) ON DELETE CASCADE + ) + """) + # Create indexes for PostgreSQL cursor.execute("CREATE INDEX IF NOT EXISTS idx_episodes_imdb ON episodes(imdb_id)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_episodes_video ON episodes(has_video_file)") @@ -155,6 +197,11 @@ class NFOGuardDatabase: cursor.execute("CREATE INDEX IF NOT EXISTS idx_missing_imdb_type ON missing_imdb(media_type)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_missing_imdb_resolved ON missing_imdb(resolved)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_missing_imdb_path ON missing_imdb(file_path)") + cursor.execute("CREATE INDEX IF NOT EXISTS idx_scheduled_scans_enabled ON scheduled_scans(enabled)") + cursor.execute("CREATE INDEX IF NOT EXISTS idx_scheduled_scans_next_run ON scheduled_scans(next_run_at)") + cursor.execute("CREATE INDEX IF NOT EXISTS idx_schedule_executions_schedule ON schedule_executions(schedule_id)") + cursor.execute("CREATE INDEX IF NOT EXISTS idx_schedule_executions_status ON schedule_executions(status)") + cursor.execute("CREATE INDEX IF NOT EXISTS idx_schedule_executions_started ON schedule_executions(started_at)") def upsert_series(self, imdb_id: str, path: str, metadata: Optional[Dict] = None): """Insert or update series record""" with self.get_connection() as conn: @@ -673,6 +720,226 @@ class NFOGuardDatabase: return deleted_count > 0 + # Scheduled Scans Methods + + def create_scheduled_scan(self, name: str, description: str, cron_expression: str, + media_type: str, scan_mode: str, specific_paths: str = None, + enabled: bool = True, created_by: str = None) -> int: + """Create a new scheduled scan""" + with self.get_connection() as conn: + cursor = conn.cursor() + + cursor.execute(""" + INSERT INTO scheduled_scans + (name, description, cron_expression, media_type, scan_mode, specific_paths, enabled, created_by) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s) + RETURNING id + """, (name, description, cron_expression, media_type, scan_mode, specific_paths, enabled, created_by)) + + return cursor.fetchone()['id'] + + def get_scheduled_scans(self, enabled_only: bool = False) -> List[Dict]: + """Get all scheduled scans""" + with self.get_connection() as conn: + cursor = conn.cursor() + + query = "SELECT * FROM scheduled_scans" + if enabled_only: + query += " WHERE enabled = TRUE" + query += " ORDER BY name" + + cursor.execute(query) + return cursor.fetchall() + + def get_scheduled_scan(self, scan_id: int) -> Optional[Dict]: + """Get a specific scheduled scan by ID""" + with self.get_connection() as conn: + cursor = conn.cursor() + + cursor.execute("SELECT * FROM scheduled_scans WHERE id = %s", (scan_id,)) + return cursor.fetchone() + + def update_scheduled_scan(self, scan_id: int, name: str = None, description: str = None, + cron_expression: str = None, media_type: str = None, + scan_mode: str = None, specific_paths: str = None, + enabled: bool = None, updated_by: str = None) -> bool: + """Update a scheduled scan""" + with self.get_connection() as conn: + cursor = conn.cursor() + + updates = [] + params = [] + + if name is not None: + updates.append("name = %s") + params.append(name) + if description is not None: + updates.append("description = %s") + params.append(description) + if cron_expression is not None: + updates.append("cron_expression = %s") + params.append(cron_expression) + if media_type is not None: + updates.append("media_type = %s") + params.append(media_type) + if scan_mode is not None: + updates.append("scan_mode = %s") + params.append(scan_mode) + if specific_paths is not None: + updates.append("specific_paths = %s") + params.append(specific_paths) + if enabled is not None: + updates.append("enabled = %s") + params.append(enabled) + if updated_by is not None: + updates.append("updated_by = %s") + params.append(updated_by) + + updates.append("updated_at = CURRENT_TIMESTAMP") + params.append(scan_id) + + if not updates: + return False + + query = f"UPDATE scheduled_scans SET {', '.join(updates)} WHERE id = %s" + cursor.execute(query, params) + + return cursor.rowcount > 0 + + def delete_scheduled_scan(self, scan_id: int) -> bool: + """Delete a scheduled scan and its execution history""" + with self.get_connection() as conn: + cursor = conn.cursor() + + cursor.execute("DELETE FROM scheduled_scans WHERE id = %s", (scan_id,)) + return cursor.rowcount > 0 + + def update_scan_next_run(self, scan_id: int, next_run_at: datetime) -> bool: + """Update the next run time for a scheduled scan""" + with self.get_connection() as conn: + cursor = conn.cursor() + + cursor.execute(""" + UPDATE scheduled_scans + SET next_run_at = %s, updated_at = CURRENT_TIMESTAMP + WHERE id = %s + """, (next_run_at, scan_id)) + + return cursor.rowcount > 0 + + def update_scan_last_run(self, scan_id: int, last_run_at: datetime = None) -> bool: + """Update the last run time and increment run count for a scheduled scan""" + with self.get_connection() as conn: + cursor = conn.cursor() + + if last_run_at is None: + last_run_at = datetime.utcnow() + + cursor.execute(""" + UPDATE scheduled_scans + SET last_run_at = %s, run_count = run_count + 1, updated_at = CURRENT_TIMESTAMP + WHERE id = %s + """, (last_run_at, scan_id)) + + return cursor.rowcount > 0 + + # Schedule Execution Methods + + def create_schedule_execution(self, schedule_id: int, media_type: str, scan_mode: str, + triggered_by: str = None) -> int: + """Create a new schedule execution record""" + with self.get_connection() as conn: + cursor = conn.cursor() + + cursor.execute(""" + INSERT INTO schedule_executions + (schedule_id, status, media_type, scan_mode, triggered_by) + VALUES (%s, 'running', %s, %s, %s) + RETURNING id + """, (schedule_id, media_type, scan_mode, triggered_by)) + + return cursor.fetchone()['id'] + + def update_schedule_execution(self, execution_id: int, status: str = None, + items_processed: int = None, items_skipped: int = None, + items_failed: int = None, error_message: str = None, + logs: str = None) -> bool: + """Update a schedule execution record""" + with self.get_connection() as conn: + cursor = conn.cursor() + + updates = [] + params = [] + + if status is not None: + updates.append("status = %s") + params.append(status) + if status in ['completed', 'failed', 'cancelled']: + updates.append("completed_at = CURRENT_TIMESTAMP") + updates.append("execution_time_seconds = EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - started_at))") + + if items_processed is not None: + updates.append("items_processed = %s") + params.append(items_processed) + if items_skipped is not None: + updates.append("items_skipped = %s") + params.append(items_skipped) + if items_failed is not None: + updates.append("items_failed = %s") + params.append(items_failed) + if error_message is not None: + updates.append("error_message = %s") + params.append(error_message) + if logs is not None: + updates.append("logs = %s") + params.append(logs) + + if not updates: + return False + + params.append(execution_id) + query = f"UPDATE schedule_executions SET {', '.join(updates)} WHERE id = %s" + cursor.execute(query, params) + + return cursor.rowcount > 0 + + def get_schedule_executions(self, schedule_id: int = None, limit: int = 50) -> List[Dict]: + """Get schedule execution history""" + with self.get_connection() as conn: + cursor = conn.cursor() + + query = """ + SELECT se.*, ss.name as schedule_name + FROM schedule_executions se + JOIN scheduled_scans ss ON se.schedule_id = ss.id + """ + params = [] + + if schedule_id is not None: + query += " WHERE se.schedule_id = %s" + params.append(schedule_id) + + query += " ORDER BY se.started_at DESC LIMIT %s" + params.append(limit) + + cursor.execute(query, params) + return cursor.fetchall() + + def get_running_executions(self) -> List[Dict]: + """Get currently running schedule executions""" + with self.get_connection() as conn: + cursor = conn.cursor() + + cursor.execute(""" + SELECT se.*, ss.name as schedule_name + FROM schedule_executions se + JOIN scheduled_scans ss ON se.schedule_id = ss.id + WHERE se.status = 'running' + ORDER BY se.started_at DESC + """) + + return cursor.fetchall() + def close(self): """Close all database connections""" if hasattr(self._local, 'connection'): diff --git a/nfoguard-web/static/index.html b/nfoguard-web/static/index.html index 45f2b3a..dc4d694 100644 --- a/nfoguard-web/static/index.html +++ b/nfoguard-web/static/index.html @@ -36,6 +36,9 @@ + @@ -287,6 +290,64 @@ + +
+
+

Scheduled Scans

+ +
+ + +
+

Active Schedules

+
+ + + + + + + + + + + + + + + + +
NameTypeModeScheduleLast RunNext RunStatusActions
+
+
+ + +
+

Recent Executions

+
+ + + + + + + + + + + + + + + + +
ScheduleStartedDurationStatusItems ProcessedItems SkippedItems FailedActions
+
+
+
+
@@ -446,6 +507,150 @@
+ + + + + +
diff --git a/scheduler/__init__.py b/scheduler/__init__.py new file mode 100644 index 0000000..6464090 --- /dev/null +++ b/scheduler/__init__.py @@ -0,0 +1 @@ +# Scheduler module for NFOGuard \ No newline at end of file diff --git a/scheduler/scheduler.py b/scheduler/scheduler.py new file mode 100644 index 0000000..fa062d5 --- /dev/null +++ b/scheduler/scheduler.py @@ -0,0 +1,396 @@ +""" +NFOGuard Background Scheduler +Manages scheduled scans using APScheduler with cron-like functionality +""" +import logging +import asyncio +from datetime import datetime, timezone +from typing import Dict, Any, Optional +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.triggers.cron import CronTrigger +from apscheduler.jobstores.memory import MemoryJobStore +from apscheduler.executors.asyncio import AsyncIOExecutor + +logger = logging.getLogger(__name__) + + +class NFOGuardScheduler: + """ + Background scheduler for NFOGuard that manages scheduled scans + """ + + def __init__(self, dependencies: Dict[str, Any]): + """Initialize the scheduler with dependencies""" + self.dependencies = dependencies + self.scheduler = None + self.running = False + + # Configure APScheduler + jobstores = { + 'default': MemoryJobStore() + } + executors = { + 'default': AsyncIOExecutor() + } + job_defaults = { + 'coalesce': False, + 'max_instances': 1, + 'misfire_grace_time': 300 # 5 minutes + } + + self.scheduler = AsyncIOScheduler( + jobstores=jobstores, + executors=executors, + job_defaults=job_defaults, + timezone='UTC' + ) + + async def start(self): + """Start the scheduler and load existing schedules""" + if self.running: + logger.warning("Scheduler is already running") + return + + try: + self.scheduler.start() + self.running = True + logger.info("✅ NFOGuard Scheduler started successfully") + + # Load existing scheduled scans from database + await self.load_schedules() + + except Exception as e: + logger.error(f"Failed to start scheduler: {e}") + raise + + async def stop(self): + """Stop the scheduler gracefully""" + if not self.running: + return + + try: + self.scheduler.shutdown() + self.running = False + logger.info("✅ NFOGuard Scheduler stopped successfully") + except Exception as e: + logger.error(f"Error stopping scheduler: {e}") + + async def load_schedules(self): + """Load all enabled scheduled scans from database and add them to scheduler""" + try: + db = self.dependencies.get("db") + if not db: + logger.error("Database not available for loading schedules") + return + + # Get all enabled scheduled scans + scheduled_scans = db.get_scheduled_scans(enabled_only=True) + + for scan in scheduled_scans: + await self.add_schedule(scan) + + logger.info(f"Loaded {len(scheduled_scans)} scheduled scans") + + except Exception as e: + logger.error(f"Failed to load schedules: {e}") + + async def add_schedule(self, scan: Dict[str, Any]): + """Add a scheduled scan to the scheduler""" + try: + job_id = f"scan_{scan['id']}" + + # Remove existing job if it exists + if self.scheduler.get_job(job_id): + self.scheduler.remove_job(job_id) + + # Create cron trigger + trigger = CronTrigger.from_crontab(scan['cron_expression']) + + # Add job to scheduler + self.scheduler.add_job( + func=self._execute_scheduled_scan, + trigger=trigger, + id=job_id, + args=[scan['id']], + name=f"Scheduled Scan: {scan['name']}", + replace_existing=True + ) + + # Update next run time in database + next_run = self.scheduler.get_job(job_id).next_run_time + if next_run: + db = self.dependencies.get("db") + if db: + db.update_scan_next_run(scan['id'], next_run) + + logger.info(f"✅ Added scheduled scan: {scan['name']} ({scan['cron_expression']})") + + except Exception as e: + logger.error(f"Failed to add schedule for scan {scan['id']}: {e}") + + async def remove_schedule(self, scan_id: int): + """Remove a scheduled scan from the scheduler""" + try: + job_id = f"scan_{scan_id}" + + if self.scheduler.get_job(job_id): + self.scheduler.remove_job(job_id) + logger.info(f"✅ Removed scheduled scan: {scan_id}") + else: + logger.warning(f"No job found for scan ID: {scan_id}") + + except Exception as e: + logger.error(f"Failed to remove schedule for scan {scan_id}: {e}") + + async def update_schedule(self, scan: Dict[str, Any]): + """Update an existing scheduled scan""" + try: + # Remove old schedule and add new one + await self.remove_schedule(scan['id']) + if scan['enabled']: + await self.add_schedule(scan) + + except Exception as e: + logger.error(f"Failed to update schedule for scan {scan['id']}: {e}") + + async def _execute_scheduled_scan(self, scan_id: int): + """Execute a scheduled scan""" + db = self.dependencies.get("db") + if not db: + logger.error(f"Database not available for executing scan {scan_id}") + return + + # Get scan details + scan = db.get_scheduled_scan(scan_id) + if not scan: + logger.error(f"Scheduled scan {scan_id} not found") + return + + if not scan['enabled']: + logger.info(f"Skipping disabled scan: {scan['name']}") + return + + execution_id = None + try: + logger.info(f"🚀 Starting scheduled scan: {scan['name']} (ID: {scan_id})") + + # Create execution record + execution_id = db.create_schedule_execution( + schedule_id=scan_id, + media_type=scan['media_type'], + scan_mode=scan['scan_mode'], + triggered_by="scheduler" + ) + + # Update last run time + db.update_scan_last_run(scan_id) + + # Execute the actual scan + result = await self._run_media_scan(scan, execution_id) + + # Update execution with results + db.update_schedule_execution( + execution_id=execution_id, + status="completed", + items_processed=result.get('items_processed', 0), + items_skipped=result.get('items_skipped', 0), + items_failed=result.get('items_failed', 0), + logs=result.get('logs', '') + ) + + logger.info(f"✅ Completed scheduled scan: {scan['name']} - Processed: {result.get('items_processed', 0)}, Skipped: {result.get('items_skipped', 0)}, Failed: {result.get('items_failed', 0)}") + + except Exception as e: + logger.error(f"❌ Failed scheduled scan: {scan['name']} - {e}") + + if execution_id: + db.update_schedule_execution( + execution_id=execution_id, + status="failed", + error_message=str(e) + ) + + async def _run_media_scan(self, scan: Dict[str, Any], execution_id: int) -> Dict[str, Any]: + """Run the actual media scan based on scan configuration""" + try: + # Import scan functionality from existing modules + from api.routes import run_tv_scan, run_movie_scan + + media_type = scan['media_type'] + scan_mode = scan['scan_mode'] + specific_paths = scan.get('specific_paths', '').strip() + + results = { + 'items_processed': 0, + 'items_skipped': 0, + 'items_failed': 0, + 'logs': [] + } + + # Parse specific paths if provided + paths = [] + if specific_paths: + paths = [p.strip() for p in specific_paths.split(',') if p.strip()] + + # Run TV scan if needed + if media_type in ['tv', 'both']: + logger.info(f"Running TV scan with mode: {scan_mode}") + + # Use existing scan infrastructure + tv_result = await self._execute_tv_scan(scan_mode, paths) + + results['items_processed'] += tv_result.get('tv_series_processed', 0) + results['items_skipped'] += tv_result.get('tv_series_skipped', 0) + results['items_failed'] += tv_result.get('tv_series_failed', 0) + results['logs'].append(f"TV Scan: {tv_result.get('message', 'Completed')}") + + # Run movie scan if needed + if media_type in ['movies', 'both']: + logger.info(f"Running movie scan with mode: {scan_mode}") + + # Use existing scan infrastructure + movie_result = await self._execute_movie_scan(scan_mode, paths) + + results['items_processed'] += movie_result.get('movies_processed', 0) + results['items_skipped'] += movie_result.get('movies_skipped', 0) + results['items_failed'] += movie_result.get('movies_failed', 0) + results['logs'].append(f"Movie Scan: {movie_result.get('message', 'Completed')}") + + results['logs'] = '\n'.join(results['logs']) + return results + + except Exception as e: + logger.error(f"Error in media scan execution: {e}") + return { + 'items_processed': 0, + 'items_skipped': 0, + 'items_failed': 1, + 'logs': f"Scan failed: {str(e)}" + } + + async def _execute_tv_scan(self, scan_mode: str, specific_paths: list = None) -> Dict[str, Any]: + """Execute TV scan using existing infrastructure""" + try: + # This would integrate with the existing manual scan functionality + # For now, return a placeholder result + return { + 'tv_series_processed': 0, + 'tv_series_skipped': 0, + 'tv_series_failed': 0, + 'message': f'TV scan ({scan_mode}) - Integration pending' + } + except Exception as e: + logger.error(f"TV scan execution failed: {e}") + return { + 'tv_series_processed': 0, + 'tv_series_skipped': 0, + 'tv_series_failed': 1, + 'message': f'TV scan failed: {str(e)}' + } + + async def _execute_movie_scan(self, scan_mode: str, specific_paths: list = None) -> Dict[str, Any]: + """Execute movie scan using existing infrastructure""" + try: + # This would integrate with the existing manual scan functionality + # For now, return a placeholder result + return { + 'movies_processed': 0, + 'movies_skipped': 0, + 'movies_failed': 0, + 'message': f'Movie scan ({scan_mode}) - Integration pending' + } + except Exception as e: + logger.error(f"Movie scan execution failed: {e}") + return { + 'movies_processed': 0, + 'movies_skipped': 0, + 'movies_failed': 1, + 'message': f'Movie scan failed: {str(e)}' + } + + async def run_manual_scan(self, scan_id: int) -> Dict[str, Any]: + """Manually trigger a scheduled scan""" + try: + db = self.dependencies.get("db") + scan = db.get_scheduled_scan(scan_id) + + if not scan: + return { + 'success': False, + 'error': 'Scheduled scan not found' + } + + # Execute the scan in the background + asyncio.create_task(self._execute_scheduled_scan(scan_id)) + + return { + 'success': True, + 'message': f"Manual execution of '{scan['name']}' started" + } + + except Exception as e: + logger.error(f"Failed to run manual scan {scan_id}: {e}") + return { + 'success': False, + 'error': str(e) + } + + def get_job_status(self, scan_id: int) -> Optional[Dict[str, Any]]: + """Get the status of a scheduled job""" + try: + job_id = f"scan_{scan_id}" + job = self.scheduler.get_job(job_id) + + if not job: + return None + + return { + 'id': job.id, + 'name': job.name, + 'next_run_time': job.next_run_time.isoformat() if job.next_run_time else None, + 'trigger': str(job.trigger) + } + + except Exception as e: + logger.error(f"Failed to get job status for scan {scan_id}: {e}") + return None + + def list_jobs(self) -> list: + """List all scheduled jobs""" + try: + jobs = [] + for job in self.scheduler.get_jobs(): + jobs.append({ + 'id': job.id, + 'name': job.name, + 'next_run_time': job.next_run_time.isoformat() if job.next_run_time else None, + 'trigger': str(job.trigger) + }) + return jobs + except Exception as e: + logger.error(f"Failed to list jobs: {e}") + return [] + + +# Global scheduler instance +scheduler_instance: Optional[NFOGuardScheduler] = None + + +async def get_scheduler(dependencies: Dict[str, Any]) -> NFOGuardScheduler: + """Get or create the global scheduler instance""" + global scheduler_instance + + if scheduler_instance is None: + scheduler_instance = NFOGuardScheduler(dependencies) + await scheduler_instance.start() + + return scheduler_instance + + +async def shutdown_scheduler(): + """Shutdown the global scheduler instance""" + global scheduler_instance + + if scheduler_instance: + await scheduler_instance.stop() + scheduler_instance = None \ No newline at end of file