import logging import socket import os from datetime import datetime, UTC from celery.exceptions import SoftTimeLimitExceeded from app.worker.celery_app import celery_app from app.core.processor import process_file from app.db.database import SessionLocal from app.models.user_model import User from app.models.job_model import Job logger = logging.getLogger(__name__) @celery_app.task( bind=True, max_retries=3, soft_time_limit=300, time_limit=360, retry_backoff=True, retry_backoff_max=600, acks_late=True, reject_on_worker_lost=True ) def process_file_task(self, file_id: int): worker_id = socket.gethostname() logger.info(f"Starting Celery worker process for file_id: {file_id}") db = SessionLocal() job = db.query(Job).filter(Job.job_id == self.request.id).first() if not job: job = Job( job_id=self.request.id, file_id=file_id, status="QUEUED", worker_id=worker_id ) db.add(job) job.status = "PROCESSING" job.started_at = datetime.now(UTC) db.commit() logger.info(f"Starting Celery worker process for file_id: {file_id}") try: from app.models.file_model import File db_file = db.query(File).filter(File.id == file_id).first() if not db_file: logger.info(f"File {file_id} not found (possibly user deleted account). Aborting task.") job.status = "CANCELLED" job.finished_at = datetime.now(UTC) db.commit() return if db_file.filetype and db_file.filetype.startswith('video/'): from app.services.video_processor import process_video_pipeline process_video_pipeline(file_id, db_file.filepath, db) else: process_file(file_id, db) job.status = "COMPLETED" job.finished_at = datetime.now(UTC) logger.info(f"Successfully processed file_id: {file_id}") db.commit() except SoftTimeLimitExceeded as exc: logger.error(f"Task for file_id {file_id} timed out. Requeuing...") job.status = "FAILED" job.error_message = "SoftTimeLimitExceeded" job.finished_at = datetime.now(UTC) db.commit() db.close() raise self.retry(exc=exc) except Exception as exc: logger.error(f"Error processing file_id {file_id}: {exc}. Retrying...") job.status = "FAILED" job.error_message = str(exc) job.finished_at = datetime.now(UTC) db.commit() db.close() raise self.retry(exc=exc) except BaseException as exc: job.status = "FAILED" job.error_message = f"CRITICAL WORKER DEATH {type(exc).__name__}" db.commit() raise finally: db.close() @celery_app.task def delete_guest_file_task(file_id: int): db = SessionLocal() try: from app.models.file_model import File db_file = db.query(File).filter(File.id == file_id).first() from app.core.storage import active_storage if db_file: if db_file.filepath: active_storage.delete(db_file.filepath) if db_file.heatmap_path: active_storage.delete(db_file.heatmap_path) db.delete(db_file) db.commit() logger.info(f"Guest file {file_id} wiped from servers after 15 minutes.") except Exception as e: logger.error(f"Failed to auto-delete guest file {file_id}: {e}") finally: db.close()