""" Progress tracking service for job processing. """ from datetime import datetime from enum import Enum from typing import Any, ClassVar from app.core.logging import get_logger from app.models.queue import AIFeedbackQueue logger = get_logger("progress_tracker") class ProcessingStep(str, Enum): """Processing step enumeration.""" INITIALIZING = "initializing" DOWNLOADING_FILE = "downloading_file" EXTRACTING_AUDIO = "extracting_audio" TRANSCRIBING_AUDIO = "transcribing_audio" ANALYZING_CONTENT = "analyzing_content" ANALYZING_VISUAL = "analyzing_visual" GAZE_TRACKING = "gaze_tracking" FACE_ANALYSIS = "face_analysis" SAVING_RESULTS = "saving_results" COMPLETING = "completing" COMPLETED = "completed" FAILED = "failed" class ProgressTracker: """Service for tracking job processing progress.""" # Step progress percentages STEP_PROGRESS: ClassVar[dict[ProcessingStep, int]] = { ProcessingStep.INITIALIZING: 5, ProcessingStep.DOWNLOADING_FILE: 15, ProcessingStep.EXTRACTING_AUDIO: 25, ProcessingStep.TRANSCRIBING_AUDIO: 45, ProcessingStep.ANALYZING_CONTENT: 60, ProcessingStep.ANALYZING_VISUAL: 75, ProcessingStep.GAZE_TRACKING: 85, ProcessingStep.FACE_ANALYSIS: 90, ProcessingStep.SAVING_RESULTS: 95, ProcessingStep.COMPLETING: 98, ProcessingStep.COMPLETED: 100, ProcessingStep.FAILED: 0, } @classmethod async def update_progress( cls, job_id: str | int, step: ProcessingStep, progress_percentage: int | None = None, heartbeat: bool = True, ) -> bool: """ Update job progress. Args: job_id: Job ID step: Current processing step progress_percentage: Optional custom progress percentage heartbeat: Whether to update last_heartbeat timestamp Returns: True if update was successful, False otherwise """ try: # Use predefined progress or custom value if progress_percentage is None: progress_percentage = cls.STEP_PROGRESS.get(step, 0) update_data = { "current_step": step.value, "progress_percentage": progress_percentage, "step_start_time": datetime.now(), } if heartbeat: update_data["last_heartbeat"] = datetime.now() # Update the job record updated_count = await AIFeedbackQueue.filter(id=job_id).update( **update_data ) if updated_count > 0: logger.info( f"[JOB {job_id}] Progress updated: {step.value} ({progress_percentage}%)" ) return True else: logger.warning(f"[JOB {job_id}] No job found to update progress") return False except Exception as e: logger.error(f"[JOB {job_id}] Failed to update progress: {e}") return False @classmethod async def heartbeat(cls, job_id: str | int) -> bool: """ Update job heartbeat to indicate it's still alive. Args: job_id: Job ID Returns: True if update was successful, False otherwise """ try: updated_count = await AIFeedbackQueue.filter(id=job_id).update( last_heartbeat=datetime.now() ) if updated_count > 0: logger.debug(f"[JOB {job_id}] Heartbeat updated") return True else: logger.warning(f"[JOB {job_id}] No job found for heartbeat update") return False except Exception as e: logger.error(f"[JOB {job_id}] Failed to update heartbeat: {e}") return False @classmethod async def get_job_progress(cls, job_id: str | int) -> dict[str, Any] | None: """ Get current job progress. Args: job_id: Job ID Returns: Progress information or None if job not found """ try: job = await AIFeedbackQueue.filter(id=job_id).first() if not job: return None return { "job_id": job.id, "status": job.status, "current_step": job.current_step, "progress_percentage": job.progress_percentage, "step_start_time": job.step_start_time.isoformat() if job.step_start_time else None, "last_heartbeat": job.last_heartbeat.isoformat() if job.last_heartbeat else None, "process_start_time": job.process_start_time.isoformat() if job.process_start_time else None, "video_name": job.video_name, "user_id": job.user_id, "interview_id": job.interview_id, "question_id": job.question_id, } except Exception as e: logger.error(f"[JOB {job_id}] Failed to get progress: {e}") return None @classmethod async def get_stuck_jobs( cls, step_timeout_minutes: int = 30 ) -> list[dict[str, Any]]: """ Find jobs that appear to be stuck. Args: step_timeout_minutes: Minutes after which a step is considered stuck Returns: List of stuck job information """ try: from datetime import timedelta cutoff_time = datetime.now() - timedelta(minutes=step_timeout_minutes) # Find processing jobs where last heartbeat or step start is too old stuck_jobs = await AIFeedbackQueue.filter( status="1", # Processing step_start_time__lt=cutoff_time, ).all() # Also check for jobs with no heartbeat no_heartbeat_jobs = await AIFeedbackQueue.filter( status="1", # Processing last_heartbeat__isnull=True, process_start_time__lt=cutoff_time, ).all() # Combine and deduplicate all_stuck = {job.id: job for job in stuck_jobs + no_heartbeat_jobs} stuck_info = [] for job in all_stuck.values(): time_since_step = None time_since_heartbeat = None if job.step_start_time: time_since_step = ( datetime.now() - job.step_start_time ).total_seconds() / 60 if job.last_heartbeat: time_since_heartbeat = ( datetime.now() - job.last_heartbeat ).total_seconds() / 60 stuck_info.append( { "job_id": job.id, "video_name": job.video_name, "current_step": job.current_step, "progress_percentage": job.progress_percentage, "time_since_step_minutes": time_since_step, "time_since_heartbeat_minutes": time_since_heartbeat, "process_start_time": job.process_start_time.isoformat() if job.process_start_time else None, "step_start_time": job.step_start_time.isoformat() if job.step_start_time else None, "last_heartbeat": job.last_heartbeat.isoformat() if job.last_heartbeat else None, } ) if stuck_info: logger.warning(f"Found {len(stuck_info)} potentially stuck jobs") return stuck_info except Exception as e: logger.error(f"Failed to get stuck jobs: {e}") return [] @classmethod async def reset_stuck_job( cls, job_id: str | int, reason: str = "Stuck job reset" ) -> bool: """ Reset a stuck job back to pending status. Args: job_id: Job ID to reset reason: Reason for reset Returns: True if reset was successful, False otherwise """ try: updated_count = await AIFeedbackQueue.filter(id=job_id).update( status="0", # Back to pending current_step=None, progress_percentage=0, step_start_time=None, last_heartbeat=None, process_start_time=None, exception=f"Reset: {reason}", tries=0, # Reset tries counter ) if updated_count > 0: logger.info(f"[JOB {job_id}] Reset stuck job: {reason}") return True else: logger.warning(f"[JOB {job_id}] No job found to reset") return False except Exception as e: logger.error(f"[JOB {job_id}] Failed to reset stuck job: {e}") return False