feat(jobs): "Mit anderem Modell verarbeiten" + Verlauf pro Job
Neuer Endpoint POST /api/jobs/{id}/reprocess: erlaubt einen LLM-Lauf
auch auf done/failed-Jobs, sichert Summary/Protokoll/DOCX/PDF in
results/<id>/runs/<ts>-<modell>/ als Snapshot, leert die DB-Pfade,
startet die Pipeline mit dem aktuell auf dem Mac konfigurierten
Ollama-Modell. Transkript bleibt erhalten.
GET /api/jobs/{id}/runs liefert die Snapshot-Liste,
GET /api/jobs/{id}/runs/{run_id}/{file} den Einzeldownload.
UI: zwei neue Buttons auf done/failed-Jobs — "↻ Mit anderem Modell"
und "Verlauf ▾". Reprocess-Dialog zeigt das aktuell konfigurierte
Mac-Modell an, bevor er startet, damit kein Versehen passiert.
Verlauf-Panel listet pro Lauf Modellname + Zeitstempel + Download-
Links für die Snapshot-Dateien.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -1,4 +1,7 @@
|
||||
import json
|
||||
import re
|
||||
import secrets
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import aiofiles
|
||||
@@ -17,6 +20,31 @@ router = APIRouter(prefix="/api/jobs", tags=["jobs"])
|
||||
|
||||
MAX_BYTES = lambda: settings.max_upload_mb * 1024 * 1024
|
||||
|
||||
# Status, in denen der Job aktiv läuft — Reprocess wird dann verweigert,
|
||||
# damit kein Schreibwettlauf mit der laufenden Pipeline entsteht.
|
||||
_ACTIVE_STATUSES = {
|
||||
JobStatus.QUEUED,
|
||||
JobStatus.TRANSCRIBING,
|
||||
JobStatus.SUMMARIZING,
|
||||
JobStatus.PROTOCOLLING,
|
||||
JobStatus.EXPORTING,
|
||||
}
|
||||
|
||||
|
||||
def _slug(s: str) -> str:
|
||||
"""Dateisystem-sichere Variante eines Modellnamens (z. B. 'gpt-oss:latest' → 'gpt-oss_latest')."""
|
||||
return re.sub(r"[^A-Za-z0-9._-]+", "_", s).strip("_") or "unknown"
|
||||
|
||||
|
||||
async def _current_mac_model() -> str:
|
||||
"""Aktuelles Ollama-Modell vom Mac-Worker — fällt auf 'unknown' zurück, wenn unerreichbar."""
|
||||
try:
|
||||
from app.core.mac_client import MacClient
|
||||
h = await MacClient().health()
|
||||
return h.get("ollama_model") or "unknown"
|
||||
except Exception: # noqa: BLE001
|
||||
return "unknown"
|
||||
|
||||
|
||||
def _require_access(job: Job | None, user: User) -> Job:
|
||||
"""404 wenn nicht gefunden oder nicht zugänglich (kein Hinweis darauf, dass es existiert)."""
|
||||
@@ -130,6 +158,132 @@ async def retry_job(
|
||||
return _job_dict(job, user)
|
||||
|
||||
|
||||
@router.post("/{job_id}/reprocess")
|
||||
async def reprocess_job(
|
||||
job_id: int,
|
||||
background: BackgroundTasks,
|
||||
session: Session = Depends(get_session),
|
||||
user: User = Depends(current_user),
|
||||
):
|
||||
"""Erlaubt einen erneuten LLM-Lauf auf einem bereits abgeschlossenen Job.
|
||||
|
||||
Anders als /retry blockiert dieser Endpoint nicht bei status=done.
|
||||
Sichert Summary/Protokoll/DOCX/PDF in runs/<ts>-<modell>/ als Snapshot
|
||||
(Transkript bleibt im Hauptordner), startet die Pipeline neu — die
|
||||
Mac-aktuell konfigurierte Ollama-Variante wird dabei verwendet.
|
||||
"""
|
||||
job = _require_access(session.get(Job, job_id), user)
|
||||
if job.status in _ACTIVE_STATUSES:
|
||||
raise HTTPException(409, "Job läuft gerade — bitte warten oder Diagnose prüfen")
|
||||
|
||||
model = await _current_mac_model()
|
||||
ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
||||
run_id = f"{ts}-{_slug(model)}"
|
||||
|
||||
job_dir = settings.result_dir / str(job.id)
|
||||
runs_dir = job_dir / "runs" / run_id
|
||||
runs_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
moved: list[str] = []
|
||||
for attr in ("summary_path", "protocol_path", "docx_path", "pdf_path"):
|
||||
path_str = getattr(job, attr)
|
||||
if path_str:
|
||||
src = Path(path_str)
|
||||
if src.exists():
|
||||
try:
|
||||
src.replace(runs_dir / src.name)
|
||||
moved.append(src.name)
|
||||
except OSError:
|
||||
pass
|
||||
setattr(job, attr, "")
|
||||
|
||||
(runs_dir / "meta.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"run_id": run_id,
|
||||
"model": model,
|
||||
"profile": job.profile,
|
||||
"snapshot_at": ts,
|
||||
"previous_status": job.status,
|
||||
"files": moved,
|
||||
},
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
job.status = JobStatus.QUEUED
|
||||
job.progress = 0
|
||||
job.error = ""
|
||||
job.step_started_at = None
|
||||
job.last_heartbeat_at = None
|
||||
session.add(job)
|
||||
session.commit()
|
||||
session.refresh(job)
|
||||
|
||||
background.add_task(run_pipeline, job.id)
|
||||
return {"job": _job_dict(job, user), "snapshot": run_id, "model_for_new_run": model}
|
||||
|
||||
|
||||
@router.get("/{job_id}/runs")
|
||||
def list_runs(
|
||||
job_id: int,
|
||||
session: Session = Depends(get_session),
|
||||
user: User = Depends(current_user),
|
||||
):
|
||||
"""Listet vergangene LLM-Läufe (Snapshots) eines Jobs."""
|
||||
job = _require_access(session.get(Job, job_id), user)
|
||||
runs_dir = settings.result_dir / str(job.id) / "runs"
|
||||
out: list[dict] = []
|
||||
if runs_dir.exists():
|
||||
for d in sorted(runs_dir.iterdir(), reverse=True):
|
||||
if not d.is_dir():
|
||||
continue
|
||||
meta: dict = {}
|
||||
try:
|
||||
meta = json.loads((d / "meta.json").read_text(encoding="utf-8"))
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
out.append({
|
||||
"run_id": d.name,
|
||||
"model": meta.get("model", "unknown"),
|
||||
"snapshot_at": meta.get("snapshot_at", ""),
|
||||
"profile": meta.get("profile", job.profile),
|
||||
"files": [
|
||||
{"name": f.name, "size": f.stat().st_size}
|
||||
for f in sorted(d.iterdir())
|
||||
if f.is_file() and f.name != "meta.json"
|
||||
],
|
||||
})
|
||||
return {"runs": out}
|
||||
|
||||
|
||||
@router.get("/{job_id}/runs/{run_id}/{filename}")
|
||||
def download_run_artifact(
|
||||
job_id: int,
|
||||
run_id: str,
|
||||
filename: str,
|
||||
session: Session = Depends(get_session),
|
||||
user: User = Depends(current_user),
|
||||
):
|
||||
"""Liefert eine einzelne Datei aus einem Snapshot-Lauf zum Download."""
|
||||
job = _require_access(session.get(Job, job_id), user)
|
||||
if "/" in run_id or ".." in run_id or "/" in filename or ".." in filename:
|
||||
raise HTTPException(400, "Ungültige Pfad-Komponenten")
|
||||
target = settings.result_dir / str(job.id) / "runs" / run_id / filename
|
||||
if not target.exists():
|
||||
raise HTTPException(404, "Snapshot-Datei nicht gefunden")
|
||||
media_map = {
|
||||
".json": "application/json",
|
||||
".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
||||
".pdf": "application/pdf",
|
||||
}
|
||||
media = media_map.get(target.suffix.lower(), "application/octet-stream")
|
||||
base = (job.title or Path(job.original_name).stem).replace("/", "_")
|
||||
return FileResponse(target, media_type=media, filename=f"{base}-{run_id}-{filename}")
|
||||
|
||||
|
||||
@router.get("/{job_id}/diag")
|
||||
async def diag(job_id: int, session: Session = Depends(get_session), user: User = Depends(current_user)):
|
||||
"""Diagnose-Bündel: Job-Status + letzte Mac-Worker-Logzeilen für genau diesen Job."""
|
||||
|
||||
Reference in New Issue
Block a user