""" job_store.py — Thread-Safe Persistent Job Storage Files stored in _jobs/ dir → survives browser refresh/close. Cleared only on Docker container restart. Thread safety: atomic write (temp file → rename) prevents corruption when background threads update jobs concurrently. """ import json import os import time import uuid import tempfile import threading from typing import Optional JOBS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "_jobs") os.makedirs(JOBS_DIR, exist_ok=True) # Per-job locks so concurrent updates to the same job don't corrupt _locks: dict[str, threading.Lock] = {} _locks_mutex = threading.Lock() def _job_lock(job_id: str) -> threading.Lock: with _locks_mutex: if job_id not in _locks: _locks[job_id] = threading.Lock() return _locks[job_id] # ───────────────────────────────────────────────────────────────────────────── # Internal atomic write # ───────────────────────────────────────────────────────────────────────────── def _save_job(job: dict) -> None: """Atomic write: write to temp → rename. Thread-safe.""" target = os.path.join(JOBS_DIR, f"{job['id']}.json") tmp_fd, tmp_path = tempfile.mkstemp(dir=JOBS_DIR, suffix=".tmp") try: with os.fdopen(tmp_fd, "w", encoding="utf-8") as f: json.dump(job, f, ensure_ascii=False, indent=2) os.replace(tmp_path, target) # atomic on POSIX except Exception: try: os.unlink(tmp_path) except Exception: pass raise # ───────────────────────────────────────────────────────────────────────────── # Job creation # ───────────────────────────────────────────────────────────────────────────── def create_job( job_type: str, filename: str, language: str = "", model: str = "", provider: str = "", extra: dict | None = None, ) -> str: job_id = str(uuid.uuid4())[:12] now = time.time() job = { "id": job_id, "type": job_type, # "audio_srt" | "translate" "filename": filename, "language": language, "model": model, "provider": provider, "status": "queued", # queued|running|paused|completed|failed "progress": 0.0, "status_msg": "", "created_at": now, "updated_at": now, "error": None, "checkpoint": {}, "extra": extra or {}, "has_result": False, } _save_job(job) return job_id # ───────────────────────────────────────────────────────────────────────────── # Job updates (thread-safe) # ───────────────────────────────────────────────────────────────────────────── def update_job( job_id: str, status: str | None = None, progress: float | None = None, status_msg: str | None = None, error: str | None = None, checkpoint: dict | None = None, extra_update: dict | None = None, ) -> None: with _job_lock(job_id): job = load_job(job_id) if job is None: return if status is not None: job["status"] = status if progress is not None: job["progress"] = round(float(progress), 4) if status_msg is not None: job["status_msg"] = str(status_msg)[:300] if error is not None: job["error"] = str(error)[:500] if checkpoint is not None: job["checkpoint"] = checkpoint if extra_update: job.setdefault("extra", {}).update(extra_update) job["updated_at"] = time.time() _save_job(job) def save_result(job_id: str, data: bytes) -> None: result_path = os.path.join(JOBS_DIR, f"{job_id}.result") # Atomic write for result too tmp_fd, tmp_path = tempfile.mkstemp(dir=JOBS_DIR, suffix=".tmp") try: os.write(tmp_fd, data) os.close(tmp_fd) os.replace(tmp_path, result_path) except Exception: try: os.close(tmp_fd) except Exception: pass try: os.unlink(tmp_path) except Exception: pass raise with _job_lock(job_id): job = load_job(job_id) if job: job["has_result"] = True job["status"] = "completed" job["progress"] = 1.0 job["status_msg"] = "ပြီးစီးသည် ✓" job["updated_at"] = time.time() _save_job(job) def load_result(job_id: str) -> bytes | None: path = os.path.join(JOBS_DIR, f"{job_id}.result") if not os.path.exists(path): return None with open(path, "rb") as f: return f.read() # ───────────────────────────────────────────────────────────────────────────── # Reading # ───────────────────────────────────────────────────────────────────────────── def load_job(job_id: str) -> dict | None: path = os.path.join(JOBS_DIR, f"{job_id}.json") if not os.path.exists(path): return None try: with open(path, "r", encoding="utf-8") as f: return json.load(f) except Exception: return None def list_jobs(job_type: str | None = None) -> list[dict]: jobs = [] for fname in os.listdir(JOBS_DIR): if not fname.endswith(".json"): continue job = load_job(fname[:-5]) if job is None: continue if job_type and job.get("type") != job_type: continue jobs.append(job) jobs.sort(key=lambda j: j.get("created_at", 0), reverse=True) return jobs def delete_job(job_id: str) -> None: for ext in (".json", ".result"): path = os.path.join(JOBS_DIR, f"{job_id}{ext}") if os.path.exists(path): try: os.remove(path) except Exception: pass def clear_all_jobs() -> int: count = 0 for fname in os.listdir(JOBS_DIR): if fname.endswith(".tmp"): continue try: os.remove(os.path.join(JOBS_DIR, fname)) count += 1 except Exception: pass return count # ───────────────────────────────────────────────────────────────────────────── # Glossary persistence # Glossary is saved independently of checkpoint so it survives # model/provider changes. Resume always reuses the approved glossary. # ───────────────────────────────────────────────────────────────────────────── def save_glossary(job_id: str, glossary: dict) -> None: """ Persist the approved glossary for a translation job. Stored inside the job JSON under "approved_glossary". Thread-safe (uses update_job which is already atomic). """ update_job(job_id, extra_update={"approved_glossary": glossary}) def load_glossary(job_id: str) -> dict: """ Load the saved glossary for a job. Returns empty dict if not found. """ job = load_job(job_id) if job is None: return {} return job.get("extra", {}).get("approved_glossary", {}) def get_latest_translate_job_with_glossary() -> dict | None: """ Return the most recent translate job that has a saved glossary. Used to offer glossary reuse when starting a new translation. """ jobs = list_jobs(job_type="translate") for job in jobs: g = job.get("extra", {}).get("approved_glossary", {}) if g: return job return None # ───────────────────────────────────────────────────────────────────────────── # Display helpers # ───────────────────────────────────────────────────────────────────────────── def job_age_str(job: dict) -> str: delta = time.time() - job.get("updated_at", job.get("created_at", 0)) if delta < 60: return "just now" if delta < 3600: return f"{int(delta/60)} min ago" if delta < 86400: return f"{int(delta/3600)} hr ago" return f"{int(delta/86400)} days ago" def status_icon(status: str) -> str: return {"queued":"⏳","running":"🔄","paused":"⏸","completed":"✅","failed":"❌"}.get(status,"❓") def status_color(status: str) -> str: return {"queued":"#4b5563","running":"#3b82f6","paused":"#f59e0b", "completed":"#22c55e","failed":"#ef4444"}.get(status,"#4b5563")