Spaces:
Paused
Paused
| from __future__ import annotations | |
| import asyncio | |
| import threading | |
| import uuid | |
| from typing import Any | |
| from app.config import settings | |
| _pool = None | |
| _local_jobs: dict[str, dict[str, Any]] = {} | |
| _jobs_lock = threading.Lock() | |
| def _set_local_job(job_id: str, status: str, result: Any = None, error: str | None = None) -> None: | |
| with _jobs_lock: | |
| _local_jobs[job_id] = { | |
| "status": status, | |
| "result": result, | |
| "error": error, | |
| } | |
| # Bound memory to most recent 500 local jobs | |
| if len(_local_jobs) > 500: | |
| oldest_key = next(iter(_local_jobs)) | |
| _local_jobs.pop(oldest_key, None) | |
| def _run_coro_in_thread(job_id: str, coro_fn, *args: Any) -> None: | |
| def runner() -> None: | |
| _set_local_job(job_id, "in_progress") | |
| try: | |
| res = asyncio.run(coro_fn({}, *args)) | |
| if isinstance(res, dict) and res.get("status") == "failed": | |
| _set_local_job(job_id, "failed", result=None, error=str(res.get("error") or "Job failed")) | |
| else: | |
| _set_local_job(job_id, "complete", result=res, error=None) | |
| except Exception as exc: # pragma: no cover - defensive thread wrapper | |
| _set_local_job(job_id, "failed", result=None, error=str(exc)) | |
| thread = threading.Thread(target=runner, daemon=True) | |
| thread.start() | |
| async def get_arq_pool(): | |
| global _pool | |
| if _pool is None: | |
| from arq.connections import RedisSettings, create_pool | |
| _pool = await create_pool( | |
| RedisSettings(host=settings.redis_host, port=settings.redis_port, conn_timeout=1, conn_retries=0) | |
| ) | |
| return _pool | |
| async def enqueue_transcription( | |
| session_id: int, | |
| user_id: int, | |
| model_name: str | None, | |
| backend_id: str | None, | |
| ) -> str: | |
| try: | |
| pool = await get_arq_pool() | |
| job = await pool.enqueue_job( | |
| "transcribe_recording", | |
| session_id, | |
| user_id, | |
| model_name, | |
| backend_id, | |
| ) | |
| if job is not None: | |
| return job.job_id | |
| except Exception: | |
| if not settings.use_local_worker_fallback: | |
| raise | |
| from app.worker.jobs import transcribe_recording | |
| job_id = f"local-transcribe-{uuid.uuid4().hex[:12]}" | |
| _set_local_job(job_id, "pending") | |
| _run_coro_in_thread(job_id, transcribe_recording, session_id, user_id, model_name, backend_id) | |
| return job_id | |
| async def enqueue_coaching(session_id: int, user_id: int) -> str: | |
| try: | |
| pool = await get_arq_pool() | |
| job = await pool.enqueue_job("generate_session_coaching", session_id, user_id) | |
| if job is not None: | |
| return job.job_id | |
| except Exception: | |
| if not settings.use_local_worker_fallback: | |
| raise | |
| from app.worker.jobs import generate_session_coaching | |
| job_id = f"local-coaching-{uuid.uuid4().hex[:12]}" | |
| _set_local_job(job_id, "pending") | |
| _run_coro_in_thread(job_id, generate_session_coaching, session_id, user_id) | |
| return job_id | |
| async def get_job_status(job_id: str) -> dict: | |
| with _jobs_lock: | |
| if job_id in _local_jobs: | |
| return dict(_local_jobs[job_id]) | |
| try: | |
| pool = await get_arq_pool() | |
| from arq.jobs import Job | |
| job = Job(job_id, pool) | |
| info = await job.info() | |
| if info is None: | |
| return {"status": "pending", "result": None, "error": None} | |
| status_value = info.status.value if info.status else "unknown" | |
| return { | |
| "status": status_value, | |
| "result": info.result if info.success else None, | |
| "error": str(info.result) if not info.success and info.result else None, | |
| } | |
| except Exception: | |
| return {"status": "not_found", "result": None, "error": None} | |