added video feature

api.py

@@ -11,6 +11,8 @@ import asyncio
 from backend import RegexClassifier
 from email_service import send_welcome_email
 import evaluator_api
+import video_router
+import video_job_queue
 
 # Initialize FastAPI app
 app = FastAPI(
@@ -41,6 +43,16 @@ classifier = RegexClassifier()
 evaluator_api.setup(classifier)
 app.include_router(evaluator_api.router)
 
+# Wire video API — async job queue (startup worker launched below)
+video_job_queue.setup(classifier)
+app.include_router(video_router.router)
+
+
+@app.on_event("startup")
+async def _start_video_worker():
+    """Launch the background asyncio video processing worker."""
+    await video_job_queue.startup()
+
 # Maximum file size (1GB)
 MAX_FILE_SIZE = 1024 * 1024 * 1024  # 1GB in bytes
 

file_handlers/universal_parser.py

@@ -26,6 +26,12 @@ CATEGORY_FILE_TYPES = {
         {"ext": "eml",  "label": "Email (.eml)"},
         {"ext": "epub", "label": "eBook (.epub)"},
         {"ext": "pptx", "label": "PowerPoint (.pptx)"},
+        # Video formats — processed via async job queue, not universal_parser
+        {"ext": "mp4",  "label": "🎬 MP4 Video (.mp4)",  "async": True},
+        {"ext": "mkv",  "label": "🎬 MKV Video (.mkv)",  "async": True},
+        {"ext": "avi",  "label": "🎬 AVI Video (.avi)",  "async": True},
+        {"ext": "mov",  "label": "🎬 MOV Video (.mov)",  "async": True},
+        {"ext": "webm", "label": "🎬 WebM Video (.webm)", "async": True},
     ],
     "semi_structured": [
         {"ext": "json",    "label": "JSON (.json)"},

file_handlers/video_pipeline.py

@@ -0,0 +1,266 @@
+"""
+Video Pipeline — 3-channel text extraction for PII detection.
+
+Channels:
+  1. Metadata   — ffprobe JSON → flat key: value text
+  2. Subtitles  — embedded SRT/VTT tracks demuxed via ffmpeg
+  3. Audio      — ffmpeg WAV + faster-whisper (base) transcription
+
+Returns merged plain text with source attribution headers so the
+downstream NLP models receive clearly labelled, scannable content.
+
+All external calls are guarded with timeouts so this never hangs forever.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+import subprocess
+import tempfile
+from typing import Callable, Optional
+
+# ── Lazy Whisper loader ──────────────────────────────────────────────────────
+
+WHISPER_MODEL_SIZE = "base"
+_whisper_model = None
+
+
+def _get_whisper():
+    global _whisper_model
+    if _whisper_model is None:
+        try:
+            from faster_whisper import WhisperModel
+            _whisper_model = WhisperModel(
+                WHISPER_MODEL_SIZE,
+                device="cpu",
+                compute_type="int8",  # quantised — runs on CPU without VRAM
+            )
+        except ImportError:
+            raise RuntimeError(
+                "faster-whisper is not installed. "
+                "Run: pip install faster-whisper"
+            )
+    return _whisper_model
+
+
+# ── ffmpeg availability ───────────────────────────────────────────────────────
+
+def _check_ffmpeg() -> bool:
+    """Return True if ffmpeg/ffprobe are available in PATH."""
+    try:
+        subprocess.run(
+            ["ffmpeg", "-version"],
+            capture_output=True, check=True, timeout=5,
+        )
+        return True
+    except Exception:
+        return False
+
+
+# ── Channel 1: Metadata ───────────────────────────────────────────────────────
+
+def _extract_metadata(video_path: str) -> str:
+    """Use ffprobe to extract all metadata tags as flat key: value text."""
+    try:
+        result = subprocess.run(
+            [
+                "ffprobe", "-v", "quiet",
+                "-print_format", "json",
+                "-show_format", "-show_streams",
+                video_path,
+            ],
+            capture_output=True, text=True, timeout=30, check=True,
+        )
+        data = json.loads(result.stdout)
+        lines = ["[METADATA]"]
+
+        fmt_tags = data.get("format", {}).get("tags", {})
+        for key, val in fmt_tags.items():
+            lines.append(f"{key}: {val}")
+
+        for idx, stream in enumerate(data.get("streams", [])):
+            stream_tags = stream.get("tags", {})
+            for key, val in stream_tags.items():
+                lines.append(f"stream{idx}_{key}: {val}")
+
+        return "\n".join(lines) if len(lines) > 1 else ""
+    except Exception as exc:
+        return f"[METADATA]\n[Could not extract: {exc}]"
+
+
+# ── Channel 2: Subtitles ──────────────────────────────────────────────────────
+
+_TS_PATTERN = re.compile(r"(\d{2}:\d{2}:\d{2}),\d{3} --> ")
+_TS_CAPTURE = re.compile(r"(\d{2}:\d{2}:\d{2})")
+
+
+def _parse_srt(raw: str) -> list[str]:
+    """Convert raw SRT content to clean timestamped lines."""
+    lines = raw.splitlines()
+    result = []
+    i = 0
+    while i < len(lines):
+        line = lines[i].strip()
+        if _TS_PATTERN.match(line):
+            ts_match = _TS_CAPTURE.match(line)
+            label = ts_match.group(1) if ts_match else ""
+            i += 1
+            parts: list[str] = []
+            while i < len(lines) and lines[i].strip():
+                parts.append(lines[i].strip())
+                i += 1
+            if parts:
+                result.append(f"[{label}] {' '.join(parts)}")
+        i += 1
+    return result
+
+
+def _extract_subtitles(video_path: str, tmp_dir: str) -> str:
+    """Demux the first embedded subtitle track; parse and return clean text."""
+    srt_path = os.path.join(tmp_dir, "subs.srt")
+    try:
+        subprocess.run(
+            ["ffmpeg", "-y", "-i", video_path, "-map", "0:s:0", srt_path],
+            capture_output=True, timeout=60, check=True,
+        )
+    except Exception:
+        return ""  # No subtitles is normal — not an error
+
+    if not os.path.exists(srt_path) or os.path.getsize(srt_path) == 0:
+        return ""
+
+    with open(srt_path, encoding="utf-8", errors="replace") as fh:
+        raw = fh.read()
+
+    lines = _parse_srt(raw)
+    if not lines:
+        return ""
+    return "[SUBTITLES]\n" + "\n".join(lines)
+
+
+# ── Channel 3: Audio Transcript ────────────────────────────────────���──────────
+
+def _extract_audio_transcript(
+    video_path: str,
+    tmp_dir: str,
+    progress_cb: Optional[Callable[[int, str], None]] = None,
+) -> str:
+    """
+    Extract audio track with ffmpeg → transcribe with faster-whisper (base).
+    progress_cb(percent: int, detail: str) is called throughout transcription.
+    """
+    wav_path = os.path.join(tmp_dir, "audio.wav")
+
+    # Step A: Extract audio as 16 kHz mono WAV (Whisper requirement)
+    try:
+        subprocess.run(
+            [
+                "ffmpeg", "-y", "-i", video_path,
+                "-vn",                        # no video
+                "-acodec", "pcm_s16le",       # 16-bit PCM
+                "-ar", "16000",               # 16 kHz sample rate
+                "-ac", "1",                   # mono
+                wav_path,
+            ],
+            capture_output=True, timeout=180, check=True,
+        )
+    except subprocess.TimeoutExpired:
+        return "[AUDIO TRANSCRIPT]\n[Audio extraction timed out after 3 minutes]"
+    except Exception as exc:
+        return f"[AUDIO TRANSCRIPT]\n[Audio extraction failed: {exc}]"
+
+    if not os.path.exists(wav_path) or os.path.getsize(wav_path) < 1024:
+        return "[AUDIO TRANSCRIPT]\n[No audio track found in this video]"
+
+    # Step B: Transcribe
+    try:
+        model = _get_whisper()
+        if progress_cb:
+            progress_cb(0, "Starting Whisper transcription…")
+
+        segments_iter, info = model.transcribe(
+            wav_path,
+            beam_size=5,
+            language=None,  # auto-detect
+            vad_filter=True,  # skip silent sections — faster
+        )
+        duration = float(info.duration) if info.duration else 1.0
+
+        transcript_lines = ["[AUDIO TRANSCRIPT]"]
+        for seg in segments_iter:
+            start_s = int(seg.start)
+            end_s   = int(seg.end)
+            ts = f"{start_s // 60:02d}:{start_s % 60:02d} → {end_s // 60:02d}:{end_s % 60:02d}"
+            transcript_lines.append(f"[{ts}] {seg.text.strip()}")
+
+            if progress_cb:
+                pct = min(int((seg.end / duration) * 100), 99)
+                dur_fmt = f"{int(duration // 60):02d}:{int(duration % 60):02d}"
+                progress_cb(pct, f"Transcribing audio… {ts} / {dur_fmt}")
+
+        return "\n".join(transcript_lines)
+
+    except Exception as exc:
+        return f"[AUDIO TRANSCRIPT]\n[Transcription failed: {exc}]"
+
+
+# ── Public entry point ────────────────────────────────────────────────────────
+
+def process_video(
+    video_path: str,
+    progress_cb: Optional[Callable[[int, str], None]] = None,
+) -> str:
+    """
+    Full 3-channel extraction pipeline.
+    Returns merged text with source-attribution headers.
+
+    progress_cb(percent: int, detail: str)
+      — called throughout; percent maps 0 → 95 (final 5% is model scanning)
+    """
+    if not _check_ffmpeg():
+        return (
+            "[ERROR] ffmpeg not found in PATH.\n"
+            "On HuggingFace Spaces add 'ffmpeg' to packages.txt.\n"
+            "Locally: https://ffmpeg.org/download.html"
+        )
+
+    parts: list[str] = []
+
+    with tempfile.TemporaryDirectory() as tmp_dir:
+
+        # ── Channel 1: Metadata (instant, ~1 s) ─────────────────────────────
+        if progress_cb:
+            progress_cb(3, "Extracting video metadata…")
+        meta = _extract_metadata(video_path)
+        if meta:
+            parts.append(meta)
+
+        # ── Channel 2: Subtitles (fast, ~2–5 s) ─────────────────────────────
+        if progress_cb:
+            progress_cb(8, "Demuxing embedded subtitles…")
+        subs = _extract_subtitles(video_path, tmp_dir)
+        if subs:
+            parts.append(subs)
+
+        # ── Channel 3: Audio Transcript (slow — Whisper) ─────────────────────
+        if progress_cb:
+            progress_cb(13, "Initialising Whisper model…")
+
+        def _audio_progress(pct: int, detail: str) -> None:
+            # Remap audio progress: 13 % → 93 %
+            mapped = 13 + int(pct * 0.80)
+            if progress_cb:
+                progress_cb(mapped, detail)
+
+        transcript = _extract_audio_transcript(video_path, tmp_dir, _audio_progress)
+        if transcript:
+            parts.append(transcript)
+
+    if progress_cb:
+        progress_cb(95, "Extraction complete — handing off to PII models…")
+
+    if not parts:
+        return "[No extractable text found in this video file]"
+
+    return "\n\n".join(parts)

packages.txt

@@ -1,4 +1,5 @@
 tesseract-ocr
 tesseract-ocr-eng
 libgl1-mesa-glx
-libglib2.0-0
+libglib2.0-0
+ffmpeg

requirements.txt

@@ -49,3 +49,6 @@ pydifact
 openpyxl
 xlrd
 lxml
+# ── Video PII processing ────────────────────────────────────
+faster-whisper
+opencv-python-headless

video_job_queue.py

@@ -0,0 +1,279 @@
+"""
+Video Job Queue — async job infrastructure for Segmento Sense.
+
+Pattern: in-process asyncio.Queue + ThreadPoolExecutor.
+  - No Redis, no Celery, no external dependencies.
+  - Single background asyncio Task consumes jobs one at a time.
+  - Job state lives in _job_store (dict); results also persisted to /tmp/
+    so they survive a dict reset (not a full process restart).
+
+Public API (called from video_router.py and api.py):
+  setup(classifier)  — inject shared classifier
+  create_job(...)    — register a new job, return job_id
+  enqueue(job_id)    — push job_id onto the asyncio.Queue
+  get_job(job_id)    — read job status / result
+  cancel_job(job_id) — cancel if still queued
+  startup()          — launch background worker (call from FastAPI startup)
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import time
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any, Dict, List, Optional
+
+from file_handlers.video_pipeline import process_video
+
+# ── Constants ─────────────────────────────────────────────────────────────────
+
+_JOB_DIR = "/tmp/video_jobs"
+_MAX_WORKERS = 2          # max concurrent video jobs in thread pool
+
+# ── Module-level state ────────────────────────────────────────────────────────
+
+_job_store: Dict[str, Dict[str, Any]] = {}
+_queue: asyncio.Queue = asyncio.Queue()
+_executor = ThreadPoolExecutor(max_workers=_MAX_WORKERS, thread_name_prefix="video_worker")
+_classifier = None       # injected by setup()
+
+
+# ── Dependency injection ──────────────────────────────────────────────────────
+
+def setup(classifier_instance) -> None:
+    """Inject the shared RegexClassifier from api.py."""
+    global _classifier
+    _classifier = classifier_instance
+
+
+# ── Disk persistence (best-effort) ────────────────────────────────────────────
+
+def _persist(job_id: str) -> None:
+    """Write completed/errored job to disk. Skips large 'parsed_text' field to keep files small."""
+    os.makedirs(_JOB_DIR, exist_ok=True)
+    entry = _job_store.get(job_id)
+    if not entry:
+        return
+    try:
+        # Shallow copy — omit internal path and large text from disk store
+        safe = {k: v for k, v in entry.items() if k not in ("video_path",)}
+        if safe.get("result") and isinstance(safe["result"], dict):
+            # Store parsed_text only truncated to 2000 chars on disk (full lives in memory)
+            result_copy = dict(safe["result"])
+            result_copy["parsed_text"] = result_copy.get("parsed_text", "")[:2000]
+            safe["result"] = result_copy
+        path = os.path.join(_JOB_DIR, f"{job_id}.json")
+        with open(path, "w", encoding="utf-8") as fh:
+            json.dump(safe, fh, ensure_ascii=False)
+    except Exception:
+        pass  # persistence is best-effort
+
+
+def _load_from_disk(job_id: str) -> Optional[Dict[str, Any]]:
+    """Try to recover a job result from disk (for jobs not in memory)."""
+    path = os.path.join(_JOB_DIR, f"{job_id}.json")
+    if os.path.exists(path):
+        try:
+            with open(path, encoding="utf-8") as fh:
+                return json.load(fh)
+        except Exception:
+            pass
+    return None
+
+
+# ── Job lifecycle ─────────────────────────────────────────────────────────────
+
+def create_job(video_path: str, model_keys: List[str]) -> str:
+    """Create a job entry in _job_store, return the new job_id."""
+    job_id = str(uuid.uuid4())
+    _job_store[job_id] = {
+        "status": "queued",
+        "progress": 0,
+        "stage_detail": "Waiting in queue…",
+        "result": None,
+        "error": None,
+        "created_at": time.time(),
+        "video_path": video_path,   # internal — stripped before sending to client
+        "model_keys": model_keys,
+    }
+    return job_id
+
+
+def get_job(job_id: str) -> Optional[Dict[str, Any]]:
+    """Return job status dict, checking disk if not found in memory."""
+    if job_id in _job_store:
+        entry = _job_store[job_id]
+        # Return safe copy without internal file path
+        return {k: v for k, v in entry.items() if k != "video_path"}
+    return _load_from_disk(job_id)
+
+
+def cancel_job(job_id: str) -> bool:
+    """
+    Cancel a queued job. Returns True on success.
+    Cannot cancel a job that is already extracting/transcribing/scanning.
+    """
+    entry = _job_store.get(job_id)
+    if entry and entry["status"] == "queued":
+        entry["status"] = "cancelled"
+        entry["stage_detail"] = "Cancelled by user."
+        _persist(job_id)
+        return True
+    return False
+
+
+async def enqueue(job_id: str) -> None:
+    """Put an already-created job onto the async queue."""
+    await _queue.put(job_id)
+
+
+# ── Core job processor ────────────────────────────────────────────────────────
+
+def _process_job_sync(job_id: str) -> None:
+    """
+    Synchronous heavy-lifting — runs inside ThreadPoolExecutor so it never
+    blocks the FastAPI event loop.
+
+    Stages:
+      1. Video extraction  (ffmpeg + Whisper)   → 0 % … 95 %
+      2. NLP model scan                          → 90 % … 100 %
+    """
+    entry = _job_store.get(job_id)
+    if not entry or entry["status"] in ("cancelled", "error"):
+        return
+
+    video_path  = entry["video_path"]
+    model_keys  = entry["model_keys"]
+
+    # ── progress callback (thread-safe dict write) ────────────────────────────
+    def _progress(pct: int, detail: str) -> None:
+        if entry.get("status") == "cancelled":
+            raise InterruptedError("Job cancelled by user")
+        entry["progress"] = pct
+        entry["stage_detail"] = detail
+
+    try:
+        # ── Stage 1: Extract text from video ─────────────────────────────────
+        entry["status"] = "extracting"
+        _progress(1, "Starting video extraction…")
+
+        extracted_text = process_video(video_path, _progress)
+
+        if not extracted_text.strip() or extracted_text.startswith("[ERROR]"):
+            entry["status"] = "error"
+            entry["error"] = extracted_text or "No extractable text found."
+            _persist(job_id)
+            return
+
+        # ── Stage 2: Run PII models ───────────────────────────────────────────
+        entry["status"] = "scanning"
+        _progress(90, f"Running {len(model_keys)} PII model(s)…")
+
+        if _classifier is None:
+            raise RuntimeError("Classifier not initialised — call video_job_queue.setup() first.")
+
+        all_detections: Dict[str, List[dict]] = {}
+        for i, key in enumerate(model_keys):
+            _progress(
+                90 + int((i / len(model_keys)) * 8),
+                f"Scanning with {key} ({i + 1}/{len(model_keys)})…",
+            )
+            dets = _classifier.analyze_text_hybrid(extracted_text[:100_000], [key])
+            all_detections[key] = dets
+
+        # ── Stage 3: Build showdown result ────────────────────────────────────
+        # Re-use the exact same logic as evaluator_api so the frontend
+        # gets identical shaped data — zero frontend changes needed.
+        from evaluator_api import _build_showdown_result
+
+        union_keys: set = set()
+        for dets in all_detections.values():
+            for m in dets:
+                union_keys.add((m["start"], m["end"], m["text"]))
+
+        per_model: Dict[str, dict] = {}
+        for key in model_keys:
+            per_model[key] = _build_showdown_result(
+                key,
+                all_detections[key],
+                union_keys,
+                all_detections,
+            )
+
+        ranked = sorted(
+            [
+                {"model_key": k, "pii_count": v["pii_count"], "accuracy": v["accuracy"]}
+                for k, v in per_model.items()
+            ],
+            key=lambda x: x["pii_count"],
+            reverse=True,
+        )
+        for i, r in enumerate(ranked):
+            r["rank"] = i + 1
+
+        # ── Done ──────────────────────────────────────────────────────────────
+        entry["status"]       = "done"
+        entry["progress"]     = 100
+        entry["stage_detail"] = "Complete — PII scan finished."
+        entry["result"] = {
+            "per_model":   per_model,
+            "has_gt":      False,
+            "elapsed":     round(time.time() - entry["created_at"], 2),
+            "union_total": len(union_keys),
+            "ranked":      ranked,
+            "parsed_text": extracted_text[:100_000],
+        }
+        _persist(job_id)
+
+    except InterruptedError:
+        entry["status"]       = "cancelled"
+        entry["stage_detail"] = "Cancelled."
+        _persist(job_id)
+
+    except Exception as exc:
+        entry["status"] = "error"
+        entry["error"]  = str(exc)
+        entry["stage_detail"] = f"Error: {exc}"
+        _persist(job_id)
+
+    finally:
+        # Always clean up the temp video file
+        try:
+            if video_path and os.path.exists(video_path):
+                os.remove(video_path)
+        except Exception:
+            pass
+
+
+# ── Background asyncio worker ─────────────────────────────────────────────────
+
+async def _worker() -> None:
+    """
+    Runs forever as a background asyncio Task.
+    Pulls job_ids from the queue and processes them in the thread pool.
+    """
+    loop = asyncio.get_event_loop()
+    while True:
+        job_id: str = await _queue.get()
+        entry = _job_store.get(job_id, {})
+        if entry.get("status") == "cancelled":
+            _queue.task_done()
+            continue
+        try:
+            await loop.run_in_executor(_executor, _process_job_sync, job_id)
+        except Exception as exc:
+            if job_id in _job_store:
+                _job_store[job_id]["status"] = "error"
+                _job_store[job_id]["error"]  = str(exc)
+        finally:
+            _queue.task_done()
+
+
+async def startup() -> None:
+    """
+    Launch the background worker task.
+    Must be called once from FastAPI's startup event.
+    """
+    asyncio.create_task(_worker())

video_router.py

@@ -0,0 +1,122 @@
+"""
+Video Router — FastAPI router for async video PII processing.
+
+Endpoints:
+  POST   /api/video/upload          — accept video, start async job, return job_id
+  GET    /api/video/status/{job_id} — poll job status + progress + result
+  DELETE /api/video/cancel/{job_id} — cancel a queued job
+"""
+from __future__ import annotations
+
+import os
+import tempfile
+
+from fastapi import APIRouter, File, Form, HTTPException, UploadFile
+from fastapi.responses import JSONResponse
+
+import video_job_queue as _queue
+
+router = APIRouter(prefix="/api/video", tags=["video"])
+
+ALLOWED_EXTENSIONS = {"mp4", "mkv", "avi", "mov", "webm"}
+MAX_VIDEO_BYTES    = 500 * 1024 * 1024   # 500 MB hard limit
+UPLOAD_DIR         = "/tmp/video_uploads"
+
+
+@router.post("/upload")
+async def video_upload(
+    file: UploadFile = File(...),
+    model_keys: str = Form("regex,spacy,deberta"),
+):
+    """
+    Accept a video file upload.
+    Saves bytes to /tmp/, creates an async job, returns {job_id} immediately.
+    The client should poll GET /api/video/status/{job_id} every ~3 seconds.
+    """
+    # Validate extension
+    filename = file.filename or ""
+    ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
+    if ext not in ALLOWED_EXTENSIONS:
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"Unsupported video format '{ext}'. "
+                f"Supported: {', '.join(sorted(ALLOWED_EXTENSIONS))}"
+            ),
+        )
+
+    # Read and size-check
+    content = await file.read()
+    if len(content) > MAX_VIDEO_BYTES:
+        raise HTTPException(
+            status_code=413,
+            detail=f"File is {len(content) // (1024*1024)} MB — limit is 500 MB.",
+        )
+    if len(content) < 1024:
+        raise HTTPException(status_code=400, detail="File appears to be empty.")
+
+    # Persist to temp dir
+    os.makedirs(UPLOAD_DIR, exist_ok=True)
+    fd, tmp_path = tempfile.mkstemp(suffix=f".{ext}", dir=UPLOAD_DIR)
+    with os.fdopen(fd, "wb") as fh:
+        fh.write(content)
+
+    # Parse model keys
+    keys = [k.strip() for k in model_keys.split(",") if k.strip()]
+    if not keys:
+        keys = ["regex", "spacy", "deberta"]
+
+    # Create and enqueue job
+    job_id = _queue.create_job(tmp_path, keys)
+    await _queue.enqueue(job_id)
+
+    return JSONResponse(
+        content={
+            "job_id": job_id,
+            "status": "queued",
+            "file_size_mb": round(len(content) / (1024 * 1024), 2),
+            "model_keys": keys,
+        }
+    )
+
+
+@router.get("/status/{job_id}")
+async def video_status(job_id: str):
+    """
+    Return the current job status.
+
+    Response shape:
+    {
+      "status":       "queued" | "extracting" | "scanning" | "done" | "error" | "cancelled",
+      "progress":     0-100,
+      "stage_detail": "Human-readable current stage",
+      "result":       null | { per_model, ranked, union_total, elapsed, parsed_text },
+      "error":        null | "error message string",
+      "created_at":   unix timestamp
+    }
+    """
+    entry = _queue.get_job(job_id)
+    if entry is None:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Job '{job_id}' not found. It may have expired.",
+        )
+    return JSONResponse(content=entry)
+
+
+@router.delete("/cancel/{job_id}")
+async def video_cancel(job_id: str):
+    """
+    Cancel a queued job.
+    Returns 409 if the job is already running or completed.
+    """
+    success = _queue.cancel_job(job_id)
+    if not success:
+        raise HTTPException(
+            status_code=409,
+            detail=(
+                "Cannot cancel: job is already running, completed, "
+                "or does not exist in queue."
+            ),
+        )
+    return JSONResponse(content={"cancelled": True, "job_id": job_id})