83 lines
3.2 KiB
Python
83 lines
3.2 KiB
Python
import os
|
|
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
|
|
from .. import database as db, config
|
|
|
|
router = APIRouter(prefix="/api/jobs", tags=["jobs"])
|
|
|
|
# Canonical job status vocabulary (matches the `jobs` table): queued|running|done|error.
|
|
# The frontend must use these exact strings, not invented ones like "pending"/"completed"/"failed".
|
|
|
|
|
|
@router.get("/{job_id}")
|
|
def get_job(job_id: str):
|
|
job = db.get_job(job_id)
|
|
if not job:
|
|
raise HTTPException(404, "Job not found")
|
|
return job
|
|
|
|
|
|
@router.post("/{job_id}/cancel")
|
|
async def cancel_job(job_id: str):
|
|
job = db.get_job(job_id)
|
|
if not job:
|
|
raise HTTPException(404, "Job not found")
|
|
if job["status"] in ("queued", "running"):
|
|
# The thread pool executor has no hard-kill hook, so this marks the job
|
|
# as cancelled; the pipeline itself does not currently check for
|
|
# cancellation mid-run, so a running job may still finish in the background.
|
|
db.update_job(job_id, status="error", error="Cancelled by user", error_stage="cancelled")
|
|
return {"status": "success", "message": f"Job {job_id} cancellation signal sent."}
|
|
|
|
|
|
@router.delete("/{job_id}")
|
|
async def delete_job(
|
|
job_id: str,
|
|
strategy: str = Query(..., description="One of: 'all', 'artifacts_only', 'none'"),
|
|
):
|
|
job = db.get_job(job_id)
|
|
if not job:
|
|
raise HTTPException(404, "Job not found")
|
|
|
|
session_id = job["session_id"]
|
|
session_row = db.get_conn().execute("SELECT * FROM sessions WHERE id = ?", (session_id,)).fetchone()
|
|
session = dict(session_row) if session_row else None
|
|
|
|
purged = []
|
|
|
|
if strategy in ("all", "artifacts_only") and session:
|
|
for path_key in ("transcript_path",):
|
|
path = session.get(path_key)
|
|
if path and os.path.exists(path):
|
|
os.remove(path)
|
|
purged.append(path_key)
|
|
with db.tx() as conn:
|
|
conn.execute("DELETE FROM notes WHERE session_id = ?", (session_id,))
|
|
conn.execute("DELETE FROM speakers WHERE session_id = ?", (session_id,))
|
|
purged.append("notes_and_speakers")
|
|
|
|
if strategy == "all" and session:
|
|
for path_key in ("video_path", "audio_path"):
|
|
path = session.get(path_key)
|
|
if path and os.path.exists(path):
|
|
os.remove(path)
|
|
purged.append(path_key)
|
|
with db.tx() as conn:
|
|
conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,))
|
|
purged.append("session_record")
|
|
elif strategy in ("all", "artifacts_only", "none"):
|
|
with db.tx() as conn:
|
|
conn.execute("DELETE FROM jobs WHERE id = ?", (job_id,))
|
|
purged.append("job_record")
|
|
|
|
# If a failed job left the session stuck in 'transcribing', roll it back
|
|
if job["status"] == "error" and session and session["status"] == "transcribing":
|
|
with db.tx() as conn:
|
|
conn.execute("UPDATE sessions SET status = 'uploaded' WHERE id = ?", (session_id,))
|
|
|
|
if strategy not in ("all", "artifacts_only", "none"):
|
|
raise HTTPException(400, f"Unknown strategy '{strategy}'")
|
|
|
|
return {"status": "success", "message": f"Job {job_id} cleared using strategy: {strategy}", "purged": purged}
|