#!/usr/bin/env python3 """ Example script demonstrating how to monitor job progress and detect stuck jobs. """ import asyncio from datetime import datetime from typing import Any import httpx class ProgressMonitor: """Simple progress monitoring client.""" def __init__(self, base_url: str = "http://localhost:9001"): self.base_url = base_url self.client = httpx.AsyncClient() async def get_job_progress(self, job_id: int) -> dict[str, Any] | None: """Get progress for a specific job.""" try: response = await self.client.get( f"{self.base_url}/api/v1/queue/progress/{job_id}" ) if response.status_code == 200: return response.json() elif response.status_code == 404: print(f"Job {job_id} not found") return None else: print(f"Error getting job progress: {response.status_code}") return None except Exception as e: print(f"Failed to get job progress: {e}") return None async def get_all_processing_jobs(self) -> list[dict[str, Any]]: """Get all currently processing jobs.""" try: response = await self.client.get( f"{self.base_url}/api/v1/queue/processing-jobs" ) if response.status_code == 200: data = response.json() return data.get("processing_jobs", []) else: print(f"Error getting processing jobs: {response.status_code}") return [] except Exception as e: print(f"Failed to get processing jobs: {e}") return [] async def get_stuck_jobs(self, timeout_minutes: int = 30) -> list[dict[str, Any]]: """Get jobs that appear to be stuck.""" try: response = await self.client.get( f"{self.base_url}/api/v1/queue/stuck-jobs?timeout_minutes={timeout_minutes}" ) if response.status_code == 200: data = response.json() return data.get("stuck_jobs", []) else: print(f"Error getting stuck jobs: {response.status_code}") return [] except Exception as e: print(f"Failed to get stuck jobs: {e}") return [] async def reset_stuck_job(self, job_id: int, reason: str = "Manual reset") -> bool: """Reset a stuck job.""" try: response = await self.client.post( f"{self.base_url}/api/v1/queue/reset-stuck-job/{job_id}?reason={reason}" ) if response.status_code == 200: print(f"Successfully reset job {job_id}") return True else: print(f"Error resetting job {job_id}: {response.status_code}") return False except Exception as e: print(f"Failed to reset job {job_id}: {e}") return False async def monitor_job(self, job_id: int, check_interval: int = 5) -> None: """Monitor a specific job until completion.""" print(f"Monitoring job {job_id}...") while True: progress = await self.get_job_progress(job_id) if not progress: print(f"Job {job_id} not found or completed") break status = progress.get("status") current_step = progress.get("current_step", "unknown") progress_pct = progress.get("progress_percentage", 0) print(f"Job {job_id}: {current_step} ({progress_pct}%) - Status: {status}") if status in ["3", "2"]: # Completed or failed print(f"Job {job_id} finished with status: {status}") break await asyncio.sleep(check_interval) async def dashboard(self) -> None: """Display a simple dashboard of all processing jobs.""" jobs = await self.get_all_processing_jobs() if not jobs: print("No jobs currently processing") return print("\n" + "=" * 80) print("PROCESSING JOBS DASHBOARD") print("=" * 80) print( f"{'Job ID':<8} {'Video':<20} {'Step':<20} {'Progress':<10} {'Duration':<10}" ) print("-" * 80) for job in jobs: job_id = job.get("job_id", "N/A") video_name = job.get("video_name", "N/A")[:18] current_step = job.get("current_step", "N/A")[:18] progress = f"{job.get('progress_percentage', 0)}%" # Calculate duration if start time available start_time_str = job.get("process_start_time") duration = "N/A" if start_time_str: try: start_time = datetime.fromisoformat( start_time_str.replace("Z", "+00:00") ) duration_seconds = ( datetime.now().astimezone() - start_time ).total_seconds() duration = f"{duration_seconds / 60:.1f}m" except Exception: pass print( f"{job_id:<8} {video_name:<20} {current_step:<20} {progress:<10} {duration:<10}" ) print("=" * 80) async def check_stuck_jobs( self, timeout_minutes: int = 30, auto_reset: bool = False ) -> None: """Check for and optionally reset stuck jobs.""" stuck_jobs = await self.get_stuck_jobs(timeout_minutes) if not stuck_jobs: print(f"No stuck jobs found (timeout: {timeout_minutes} minutes)") return print(f"\nFound {len(stuck_jobs)} stuck jobs:") print("-" * 60) for job in stuck_jobs: job_id = job.get("job_id") video_name = job.get("video_name", "N/A") current_step = job.get("current_step", "N/A") time_stuck = job.get("time_since_step_minutes", 0) print(f"Job {job_id}: {video_name}") print(f" Stuck in: {current_step}") print(f" Duration: {time_stuck:.1f} minutes") if auto_reset and time_stuck > timeout_minutes: success = await self.reset_stuck_job( job_id, f"Auto-reset after {time_stuck:.1f} minutes" ) if success: print(" ✓ Reset successfully") else: print(" ✗ Reset failed") print() async def close(self): """Close the HTTP client.""" await self.client.aclose() async def main(): """Example usage of the progress monitoring system.""" monitor = ProgressMonitor() try: print("AI Feedback Progress Monitor") print("=" * 40) # Show dashboard await monitor.dashboard() # Check for stuck jobs await monitor.check_stuck_jobs(timeout_minutes=30, auto_reset=False) # Example: Monitor a specific job (uncomment and set job_id) # await monitor.monitor_job(job_id=12345) finally: await monitor.close() if __name__ == "__main__": asyncio.run(main())