feat: Retry-Funktion + idempotente Pipeline (Transkript bleibt erhalten)
Backend:
- POST /api/jobs/{id}/retry — löscht summary/protocol/docx, queued den Job
neu und startet Pipeline. Transkript bleibt erhalten (teurer Schritt).
- Pipeline überspringt Steps, deren Output-Datei schon existiert →
retry rennt direkt in summarize ohne erneute Transkription.
Frontend:
- Kleiner gelber '↻ Erneut versuchen'-Button in jeder Job-Zeile/-Karte
wenn Status != done
- Confirm-Dialog erklärt was passiert
- Event-Delegation, funktioniert für Tabelle (Desktop) und Cards (Mobile)
This commit is contained in:
root
@@ -25,8 +25,19 @@ def _update(job_id: int, **fields) -> None:
|
||||
s.commit()
|
||||
|
||||
|
||||
def _existing(path_str: str) -> Path | None:
|
||||
if not path_str:
|
||||
return None
|
||||
p = Path(path_str)
|
||||
return p if p.exists() and p.stat().st_size > 0 else None
|
||||
|
||||
|
||||
async def run_pipeline(job_id: int) -> None:
|
||||
"""Run the full pipeline: transcribe → summarize → protocol → DOCX."""
|
||||
"""Pipeline: transcribe → summarize → protocol → DOCX.
|
||||
|
||||
Idempotent: bereits vorhandene Zwischenergebnisse werden weiterverwendet,
|
||||
damit ein 'Erneut versuchen' nicht die teure Transkription wiederholt.
|
||||
"""
|
||||
client = MacClient()
|
||||
|
||||
with Session(engine) as s:
|
||||
@@ -37,40 +48,69 @@ async def run_pipeline(job_id: int) -> None:
|
||||
audio_path = settings.upload_dir / job.filename
|
||||
title = job.title or audio_path.stem
|
||||
profile = job.profile or "meeting"
|
||||
existing_tx_path = _existing(job.transcript_path)
|
||||
existing_sum_path = _existing(job.summary_path)
|
||||
existing_pro_path = _existing(job.protocol_path)
|
||||
existing_dx_path = _existing(job.docx_path)
|
||||
|
||||
job_dir = settings.result_dir / str(job_id)
|
||||
job_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
try:
|
||||
_update(job_id, status=JobStatus.TRANSCRIBING, progress=10)
|
||||
log.info("[job %s] transcribe (profile=%s)", job_id, profile)
|
||||
tx = await client.transcribe(audio_path, language="de")
|
||||
transcript_path = job_dir / "transcript.json"
|
||||
transcript_path.write_text(json.dumps(tx, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
_update(job_id, transcript_path=str(transcript_path), progress=45)
|
||||
# 1) TRANSKRIPTION
|
||||
if existing_tx_path:
|
||||
log.info("[job %s] reuse transcript %s", job_id, existing_tx_path.name)
|
||||
tx = json.loads(existing_tx_path.read_text(encoding="utf-8"))
|
||||
_update(job_id, status=JobStatus.TRANSCRIBING, progress=45)
|
||||
else:
|
||||
_update(job_id, status=JobStatus.TRANSCRIBING, progress=10, error="")
|
||||
log.info("[job %s] transcribe (profile=%s)", job_id, profile)
|
||||
tx = await client.transcribe(audio_path, language="de")
|
||||
transcript_path = job_dir / "transcript.json"
|
||||
transcript_path.write_text(json.dumps(tx, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
_update(job_id, transcript_path=str(transcript_path), progress=45)
|
||||
|
||||
transcript_text = tx.get("text") or "\n".join(s.get("text", "") for s in tx.get("segments", []))
|
||||
transcript_text = tx.get("text") or "\n".join(
|
||||
s.get("text", "") for s in tx.get("segments", [])
|
||||
)
|
||||
|
||||
_update(job_id, status=JobStatus.SUMMARIZING, progress=55)
|
||||
log.info("[job %s] summarize", job_id)
|
||||
summary = await client.summarize(transcript_text, title=title, profile=profile)
|
||||
summary_path = job_dir / "summary.json"
|
||||
summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
_update(job_id, summary_path=str(summary_path), progress=70)
|
||||
# 2) SUMMARY
|
||||
if existing_sum_path:
|
||||
log.info("[job %s] reuse summary %s", job_id, existing_sum_path.name)
|
||||
summary = json.loads(existing_sum_path.read_text(encoding="utf-8"))
|
||||
_update(job_id, status=JobStatus.SUMMARIZING, progress=70)
|
||||
else:
|
||||
_update(job_id, status=JobStatus.SUMMARIZING, progress=55)
|
||||
log.info("[job %s] summarize", job_id)
|
||||
summary = await client.summarize(transcript_text, title=title, profile=profile)
|
||||
summary_path = job_dir / "summary.json"
|
||||
summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
_update(job_id, summary_path=str(summary_path), progress=70)
|
||||
|
||||
_update(job_id, status=JobStatus.PROTOCOLLING, progress=75)
|
||||
log.info("[job %s] protocol", job_id)
|
||||
protocol = await client.protocol(transcript_text, summary, title=title, profile=profile)
|
||||
protocol_path = job_dir / "protocol.json"
|
||||
protocol_path.write_text(json.dumps(protocol, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
_update(job_id, protocol_path=str(protocol_path), progress=85)
|
||||
# 3) PROTOKOLL
|
||||
if existing_pro_path:
|
||||
log.info("[job %s] reuse protocol %s", job_id, existing_pro_path.name)
|
||||
protocol = json.loads(existing_pro_path.read_text(encoding="utf-8"))
|
||||
_update(job_id, status=JobStatus.PROTOCOLLING, progress=85)
|
||||
else:
|
||||
_update(job_id, status=JobStatus.PROTOCOLLING, progress=75)
|
||||
log.info("[job %s] protocol", job_id)
|
||||
protocol = await client.protocol(transcript_text, summary, title=title, profile=profile)
|
||||
protocol_path = job_dir / "protocol.json"
|
||||
protocol_path.write_text(json.dumps(protocol, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
_update(job_id, protocol_path=str(protocol_path), progress=85)
|
||||
|
||||
_update(job_id, status=JobStatus.EXPORTING, progress=90)
|
||||
log.info("[job %s] export docx", job_id)
|
||||
docx_bytes = await client.export_docx(protocol, profile=profile)
|
||||
docx_path = job_dir / "protocol.docx"
|
||||
docx_path.write_bytes(docx_bytes)
|
||||
_update(job_id, docx_path=str(docx_path), status=JobStatus.DONE, progress=100)
|
||||
# 4) DOCX
|
||||
if existing_dx_path:
|
||||
log.info("[job %s] reuse docx %s", job_id, existing_dx_path.name)
|
||||
_update(job_id, status=JobStatus.DONE, progress=100, error="")
|
||||
else:
|
||||
_update(job_id, status=JobStatus.EXPORTING, progress=90)
|
||||
log.info("[job %s] export docx", job_id)
|
||||
docx_bytes = await client.export_docx(protocol, profile=profile)
|
||||
docx_path = job_dir / "protocol.docx"
|
||||
docx_path.write_bytes(docx_bytes)
|
||||
_update(job_id, docx_path=str(docx_path), status=JobStatus.DONE, progress=100, error="")
|
||||
log.info("[job %s] done", job_id)
|
||||
except Exception as e: # noqa: BLE001
|
||||
log.exception("[job %s] failed", job_id)
|
||||
|
||||
Reference in New Issue
Block a user