""" Queue processor service using Tortoise ORM with streaming S3 processing. """ import asyncio from typing import Any from app.core.config import settings from app.core.logging import get_logger from app.repositories.interview_repository import InterviewRepository from app.repositories.queue_repository import QueueRepository from app.services.audio_processor import AudioProcessor from app.services.aws_service import AWSService from app.services.notification_service import NotificationService from app.services.progress_tracker import ProcessingStep, ProgressTracker from app.services.video_processor import VideoProcessor logger = get_logger("queue_processor") class QueueProcessor: """Service for processing the AI feedback queue with streaming S3 processing.""" def __init__(self, max_concurrent_jobs: int | None = None): # Use setting if not provided, with fallback to default # With 8GB RAM, we can handle more concurrent jobs self.max_concurrent_jobs = max_concurrent_jobs or getattr( settings, "MAX_CONCURRENT_JOBS", 3, # Conservative default for stability ) self.active_jobs: dict[str, asyncio.Task] = {} self.video_processor = VideoProcessor() self.audio_processor = AudioProcessor() self.notification_service = NotificationService() self.aws_service = AWSService() async def process_queue(self) -> dict[str, Any]: """Process the queue with streaming S3 processing and immediate pickup of new records.""" try: logger.info( "Queue processing pass started with streaming S3 processing and immediate pickup", ) queue_repo = QueueRepository() # Clean up completed tasks before calculating available slots completed_job_ids = [ job_id for job_id, task in self.active_jobs.items() if task.done() ] for job_id in completed_job_ids: logger.info(f"Cleaning up completed task for job {job_id}") del self.active_jobs[job_id] if completed_job_ids: logger.info(f"Cleaned up {len(completed_job_ids)} completed tasks") # Check if we can process more jobs (respecting max concurrent limit) available_slots = self.max_concurrent_jobs - len(self.active_jobs) if available_slots <= 0: # Log details about active jobs for debugging active_job_details = { job_id: { "done": task.done(), "cancelled": task.cancelled() if hasattr(task, "cancelled") else False, } for job_id, task in self.active_jobs.items() } logger.info( f"No available slots for new jobs. Active: {len(self.active_jobs)}, Max: {self.max_concurrent_jobs}" ) logger.info(f"Active job details: {active_job_details}") return { "status": "at_capacity", "message": "All concurrent slots are occupied", "active_jobs": len(self.active_jobs), "max_concurrent": self.max_concurrent_jobs, } # Get pending jobs up to available slots jobs = await queue_repo.get_pending_jobs(available_slots) if not jobs: logger.info("No pending jobs to process (status='0', data_process='1')") return { "status": "no_jobs", "message": "Queue is empty", } logger.info( f"Starting streaming processing of {len(jobs)} jobs (available slots: {available_slots})" ) # Extract job data early to avoid potential session issues job_data = [] for job in jobs: try: job_data.append({"id": job.id, "video_name": job.video_name}) except Exception as e: logger.warning(f"Could not extract data from job: {e}") job_data.append({"id": "unknown", "video_name": "unknown"}) logger.info(f"Job IDs: {[job_info['id'] for job_info in job_data]}") logger.info( f"Job files: {[job_info['video_name'] for job_info in job_data]}" ) # Start processing jobs concurrently without waiting for completion # NOTE: Jobs are already atomically claimed by get_pending_jobs() for i, job in enumerate(jobs): job_id = job_data[i]["id"] logger.info(f"Starting background task for job {job_id}") # Start job processing as a background task # Job is already marked as processing by get_pending_jobs() task = asyncio.create_task(self._process_job_with_cleanup(job)) self.active_jobs[str(job_id)] = task logger.info( f"Started {len(jobs)} jobs. Active jobs: {len(self.active_jobs)}" ) return { "status": "jobs_started", "jobs_started": len(jobs), "active_jobs": len(self.active_jobs), "max_concurrent": self.max_concurrent_jobs, "processing_method": "immediate_streaming_s3", } except Exception as e: logger.error(f"Failed to process queue: {e}") return {"status": "error", "message": str(e)} async def _process_job(self, job) -> dict[str, Any]: """Process a single job using the updated streaming processors.""" import time start_time = time.time() try: # Extract job data early to avoid potential session issues job_id = job.id video_name = job.video_name user_id = job.user_id interview_id = job.interview_id question_id = job.question_id interview_type = job.interview_type status = job.status data_process = job.data_process logger.info( f"[JOB {job_id}] STEP 1/5: Job claimed and starting processing | " f"user_id={user_id} interview_id={interview_id} question_id={question_id} " f"type={interview_type} file={video_name} status={status} data_process={data_process}" ) # Update progress tracking await ProgressTracker.update_progress(job_id, ProcessingStep.INITIALIZING) # NOTE: Job is already marked as processing by get_pending_jobs() atomic claim # No need to call mark_job_processing() again here - it would be a duplicate! # Determine file type file_type = ( "video" if video_name.lower().endswith((".mp4", ".avi", ".mov", ".mkv")) else "audio" ) logger.info( f"[JOB {job_id}] STEP 2/5: Starting {file_type} processing | file={video_name}" ) # Update progress for file processing start await ProgressTracker.update_progress( job_id, ProcessingStep.DOWNLOADING_FILE ) step_start = time.time() # Process based on file type using the updated processors if video_name.lower().endswith((".mp4", ".avi", ".mov", ".mkv")): result = await self.video_processor.process_video( job_id=str(job_id), user_id=user_id, interview_id=interview_id, question_id=question_id, interview_type=interview_type, video_path=video_name, ) elif video_name.lower().endswith((".mp3", ".wav", ".m4a", ".aac")): result = await self.audio_processor.process_audio( job_id=str(job_id), user_id=user_id, interview_id=interview_id, question_id=question_id, interview_type=interview_type, audio_path=video_name, ) else: raise Exception(f"Unsupported file type: {video_name}") step_duration = time.time() - step_start logger.info( f"[JOB {job_id}] STEP 3/5: {file_type.capitalize()} processing completed | " f"duration={step_duration:.2f}s" ) # Update progress for analysis completion await ProgressTracker.update_progress(job_id, ProcessingStep.SAVING_RESULTS) # Handle job completion (includes interview status updates and notifications) logger.info( f"[JOB {job_id}] STEP 4/5: Saving results and updating job status" ) await ProgressTracker.update_progress(job_id, ProcessingStep.COMPLETING) await self._handle_job_completion(job, result) total_duration = time.time() - start_time logger.info( f"[JOB {job_id}] STEP 5/5: Job completed successfully | " f"total_duration={total_duration:.2f}s" ) # Mark as completed await ProgressTracker.update_progress(job_id, ProcessingStep.COMPLETED) return result except Exception as e: job_id = getattr(job, "id", "unknown") video_name = getattr(job, "video_name", "unknown") elapsed = time.time() - start_time if "start_time" in locals() else 0 logger.error( f"[JOB {job_id}] FAILED after {elapsed:.2f}s | " f"error={type(e).__name__}: {e} | file={video_name}" ) # Mark as failed in progress tracker await ProgressTracker.update_progress(job_id, ProcessingStep.FAILED) # Handle job failure (includes notifications) await self._handle_job_failure(job, str(e)) return {"status": "error", "message": str(e)} async def _process_job_with_cleanup(self, job) -> None: """Process a job and clean up from active jobs when done.""" job_id = str(job.id) try: result = await self._process_job(job) logger.info(f"Job {job_id} completed with status: {result.get('status')}") except Exception as e: logger.error(f"Job {job_id} failed with exception: {e}") finally: # Always clean up from active jobs if job_id in self.active_jobs: del self.active_jobs[job_id] logger.info( f"Cleaned up job {job_id} from active jobs. Remaining active: {len(self.active_jobs)}" ) else: logger.warning( f"Job {job_id} not found in active_jobs during cleanup - may have been cleaned up already" ) async def check_timeout_jobs(self, timeout_hours: int = 5) -> dict[str, Any]: """Check for jobs that have been processing for too long and handle them.""" try: queue_repo = QueueRepository() timeout_jobs = await queue_repo.check_timeout_jobs( timeout_hours=timeout_hours ) if not timeout_jobs: return { "status": "no_timeout_jobs", "timeout_jobs": [], "jobs_reset": 0, } logger.info(f"Found {len(timeout_jobs)} timeout jobs") jobs_reset = 0 for job in timeout_jobs: try: # Skip jobs with None IDs if not job or not job.id: logger.warning(f"Skipping invalid timeout job: {job}") continue job_id = job.id logger.info(f"Handling timeout job {job_id}") # Reset the timeout job success = await queue_repo.reset_timeout_job(job) if success: jobs_reset += 1 logger.info(f"Successfully reset timeout job {job_id}") except Exception as e: job_id = getattr(job, "id", "unknown") if job else "unknown" logger.error(f"Error handling timeout job {job_id}: {e}") logger.info(f"Timeout job processing complete: {jobs_reset} reset") return { "status": "timeout_jobs_handled", "timeout_jobs": [ job.to_dict() for job in timeout_jobs if hasattr(job, "to_dict") ], "jobs_reset": jobs_reset, } except Exception as e: logger.error(f"Failed to check timeout jobs: {e}") return { "status": "error", "message": str(e), "timeout_jobs": [], "jobs_reset": 0, } async def _handle_job_completion(self, job, result: dict[str, Any]) -> None: """Handle job completion including interview status updates and notifications.""" try: logger.info(f"Starting job completion handling for job {job.id}") # Mark job as completed queue_repo = QueueRepository() completion_success = await queue_repo.mark_job_completed( job.id, result.get("duration", "0") ) if completion_success: logger.info(f"Job {job.id} marked as completed successfully") else: logger.error(f"Failed to mark job {job.id} as completed") # Check if all questions for this interview are completed interview_repo = InterviewRepository() remaining_jobs = await queue_repo.get_jobs_by_interview( job.interview_id, job.interview_type ) logger.info( f"Found {len(remaining_jobs)} total jobs for interview {job.interview_id}" ) # Filter for jobs that are still pending or processing active_jobs = [j for j in remaining_jobs if j.status in ["0", "1"]] logger.info( f"Found {len(active_jobs)} active jobs for interview {job.interview_id}" ) if not active_jobs: logger.info( f"All jobs completed for interview {job.interview_id}, updating interview status" ) # All questions completed, update interview status success = await interview_repo.update_interview_ai_feedback_status( interview_id=job.interview_id, interview_type=job.interview_type, status="2", # Completed ) if success: logger.info( f"Interview {job.interview_id} status updated to completed" ) else: logger.warning( f"Failed to update interview {job.interview_id} status, but job completed successfully" ) # Send completion notification regardless of status update success try: await self.notification_service.notify_ai_feedback_completion( interview_id=job.interview_id, interview_type=job.interview_type, ) logger.info( f"Interview {job.interview_id} completion notification sent" ) except Exception as notification_error: logger.error( f"Failed to send completion notification for interview {job.interview_id}: {notification_error}" ) else: logger.info( f"Interview {job.interview_id} still has {len(active_jobs)} active jobs, skipping completion notification" ) except Exception as e: logger.error( f"Error handling job completion for job {getattr(job, 'id', 'unknown')}: {e}" ) import traceback logger.error(f"Full traceback: {traceback.format_exc()}") async def _handle_job_failure(self, job, error: str) -> None: """Handle job failure including notifications.""" try: logger.info(f"Starting job failure handling for job {job.id}") # Mark job as failed queue_repo = QueueRepository() failure_success = await queue_repo.mark_job_failed(job.id, error) if failure_success: logger.info(f"Job {job.id} marked as failed successfully") else: logger.error(f"Failed to mark job {job.id} as failed") # Send failure notification try: await self.notification_service.notify_ai_feedback_failure( interview_id=job.interview_id, interview_type=job.interview_type, error_message=error, ) logger.info(f"Job {job.id} failure handled and notification sent") except Exception as notification_error: logger.error( f"Failed to send failure notification for job {job.id}: {notification_error}" ) except Exception as e: logger.error( f"Error handling job failure for job {getattr(job, 'id', 'unknown')}: {e}" ) import traceback logger.error(f"Full traceback: {traceback.format_exc()}") async def get_queue_status(self) -> dict[str, Any]: """Get current queue status.""" try: queue_repo = QueueRepository() stats = await queue_repo.get_queue_stats() return { "total_jobs": sum(stats.values()), "pending_jobs": stats.get("0", 0), "processing_jobs": stats.get("1", 0), "failed_jobs": stats.get("2", 0), "completed_jobs": stats.get("3", 0), "active_jobs_count": len(self.active_jobs), "processing_method": "immediate_streaming_s3", "max_concurrent_jobs": self.max_concurrent_jobs, "available_slots": max( 0, self.max_concurrent_jobs - len(self.active_jobs) ), } except Exception as e: logger.error(f"Failed to get queue status: {e}") return {"error": str(e)} async def check_s3_data_availability(self) -> dict[str, Any]: """Check jobs with data_process=0 and mark as 1 if data exists in S3.""" try: logger.info("Starting S3 data availability check") queue_repo = QueueRepository() # Get jobs where data is not yet processed jobs = await queue_repo.get_unprocessed_data_jobs(limit=50) if not jobs: logger.info("No jobs with data_process=0 found") return { "status": "no_jobs", "message": "No unprocessed jobs to check", "jobs_checked": 0, "jobs_marked_ready": 0, } logger.info(f"Checking S3 availability for {len(jobs)} jobs") jobs_marked_ready = 0 for job in jobs: try: # Check if video file exists in S3 file_exists = await self.aws_service.check_file_exists( file_name=job.video_name ) if file_exists: # Mark as data_processed = 1 so it can be picked up for processing success = await queue_repo.mark_data_processed(str(job.id)) if success: jobs_marked_ready += 1 logger.info( f"Job {job.id} ({job.video_name}) marked as ready - file exists in S3" ) else: logger.warning( f"Failed to mark job {job.id} as ready despite file existing in S3" ) else: logger.debug( f"Job {job.id} ({job.video_name}) - file not yet in S3" ) except Exception as e: logger.error( f"Error checking S3 for job {job.id} ({job.video_name}): {e}" ) logger.info( f"S3 data check completed: {jobs_marked_ready}/{len(jobs)} jobs marked as ready" ) return { "status": "completed", "jobs_checked": len(jobs), "jobs_marked_ready": jobs_marked_ready, "message": f"Checked {len(jobs)} jobs, marked {jobs_marked_ready} as ready for processing", } except Exception as e: logger.error(f"Failed to check S3 data availability: {e}") return { "status": "error", "message": str(e), "jobs_checked": 0, "jobs_marked_ready": 0, } async def stop_processing(self) -> dict[str, Any]: """Stop all active processing.""" try: active_count = len(self.active_jobs) self.active_jobs.clear() return { "status": "stopped", "active_jobs_cleared": active_count, "message": "Streaming processing stopped successfully", } except Exception as e: logger.error(f"Failed to stop processing: {e}") return {"status": "error", "message": str(e)}