5f0b41161e
LLM komprimiert das Transkript bisher auf wenige Sätze + 20 Excerpt-Einträge — der Rest fiel aus dem Dokument. Pipeline reicht Whisper-Segmente jetzt an build_docx/build_pdf durch; Profile mit `docx.include_full_transcript: true` hängen das komplette Transkript (9 pt) am Ende an. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
208 lines
8.1 KiB
Python
208 lines
8.1 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Awaitable, TypeVar
|
|
|
|
from sqlmodel import Session
|
|
|
|
from app.core.config import settings
|
|
from app.core.db import engine
|
|
from app.core.mac_client import MacClient
|
|
from app.models.job import Job, JobStatus
|
|
|
|
log = logging.getLogger("voice-agent.pipeline")
|
|
|
|
T = TypeVar("T")
|
|
|
|
HEARTBEAT_INTERVAL = 3.0 # Sekunden
|
|
|
|
|
|
def _update(job_id: int, **fields) -> None:
|
|
with Session(engine) as s:
|
|
job = s.get(Job, job_id)
|
|
if not job:
|
|
return
|
|
for k, v in fields.items():
|
|
setattr(job, k, v)
|
|
job.updated_at = datetime.now(timezone.utc)
|
|
s.add(job)
|
|
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
|
|
|
|
|
|
def _start_step(job_id: int, status: str, progress: int) -> None:
|
|
"""Setzt status/progress + step_started_at + initialen Heartbeat."""
|
|
now = datetime.now(timezone.utc)
|
|
_update(
|
|
job_id,
|
|
status=status,
|
|
progress=progress,
|
|
step_started_at=now,
|
|
last_heartbeat_at=now,
|
|
error="",
|
|
)
|
|
|
|
|
|
async def _with_heartbeat(job_id: int, coro: Awaitable[T]) -> T:
|
|
"""Führt `coro` aus und pingt parallel alle paar Sekunden last_heartbeat_at.
|
|
|
|
Damit sieht die UI live, ob ein Schritt noch lebt — auch wenn der Mac
|
|
minutenlang an einer Ollama-Generierung arbeitet.
|
|
"""
|
|
stop = asyncio.Event()
|
|
|
|
async def beat() -> None:
|
|
while not stop.is_set():
|
|
try:
|
|
_update(job_id, last_heartbeat_at=datetime.now(timezone.utc))
|
|
except Exception: # noqa: BLE001
|
|
pass # Heartbeat darf den Job nie killen
|
|
try:
|
|
await asyncio.wait_for(stop.wait(), timeout=HEARTBEAT_INTERVAL)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
|
|
beat_task = asyncio.create_task(beat())
|
|
try:
|
|
return await coro
|
|
finally:
|
|
stop.set()
|
|
try:
|
|
await beat_task
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
|
|
async def run_pipeline(job_id: int) -> None:
|
|
"""Pipeline: transcribe → preload → summarize → protocol → DOCX → PDF.
|
|
|
|
Idempotent: bereits vorhandene Zwischenergebnisse werden weiterverwendet,
|
|
damit ein 'Erneut versuchen' nicht die teure Transkription wiederholt.
|
|
"""
|
|
client = MacClient()
|
|
|
|
with Session(engine) as s:
|
|
job = s.get(Job, job_id)
|
|
if not job:
|
|
log.error("Job %s not found", job_id)
|
|
return
|
|
audio_path = settings.upload_dir / job.filename
|
|
title = job.title or audio_path.stem
|
|
profile = job.profile or "meeting"
|
|
# Leeres `model` → Mac-Worker nutzt seinen .env-Default.
|
|
model = (job.model or "").strip() or None
|
|
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)
|
|
existing_pdf_path = _existing(job.pdf_path)
|
|
|
|
job_dir = settings.result_dir / str(job_id)
|
|
job_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
try:
|
|
# 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:
|
|
_start_step(job_id, JobStatus.TRANSCRIBING, 10)
|
|
log.info("[job %s] transcribe (profile=%s)", job_id, profile)
|
|
tx = await _with_heartbeat(job_id, client.transcribe(audio_path, language="de", job_id=job_id))
|
|
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", [])
|
|
)
|
|
|
|
# 2) SUMMARY — vorher Preload, damit Ollama nicht mitten drin lädt
|
|
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:
|
|
_start_step(job_id, JobStatus.SUMMARIZING, 50)
|
|
log.info("[job %s] preload model=%s", job_id, model or "(worker default)")
|
|
try:
|
|
await _with_heartbeat(job_id, client.preload(profile=profile, model=model, job_id=job_id))
|
|
except Exception as e: # noqa: BLE001
|
|
log.warning("[job %s] preload failed (non-fatal): %s", job_id, e)
|
|
_update(job_id, progress=55)
|
|
log.info("[job %s] summarize", job_id)
|
|
summary = await _with_heartbeat(
|
|
job_id,
|
|
client.summarize(transcript_text, title=title, profile=profile, model=model,
|
|
job_id=job_id),
|
|
)
|
|
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)
|
|
|
|
# 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:
|
|
_start_step(job_id, JobStatus.PROTOCOLLING, 75)
|
|
log.info("[job %s] protocol", job_id)
|
|
protocol = await _with_heartbeat(
|
|
job_id,
|
|
client.protocol(transcript_text, summary, title=title, profile=profile, model=model,
|
|
job_id=job_id),
|
|
)
|
|
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)
|
|
|
|
# Volltranskript-Segmente: optionaler Anhang ans DOCX/PDF, falls das
|
|
# Profil `docx.include_full_transcript` setzt. Renderer ignoriert die
|
|
# Liste sonst — wir reichen sie trotzdem immer durch.
|
|
full_segments = tx.get("segments", []) or []
|
|
|
|
# 4) DOCX
|
|
if existing_dx_path:
|
|
log.info("[job %s] reuse docx %s", job_id, existing_dx_path.name)
|
|
else:
|
|
_start_step(job_id, JobStatus.EXPORTING, 90)
|
|
log.info("[job %s] export docx", job_id)
|
|
docx_bytes = await _with_heartbeat(
|
|
job_id,
|
|
client.export_docx(protocol, profile=profile, job_id=job_id,
|
|
full_transcript=full_segments),
|
|
)
|
|
docx_path = job_dir / "protocol.docx"
|
|
docx_path.write_bytes(docx_bytes)
|
|
_update(job_id, docx_path=str(docx_path), progress=95)
|
|
|
|
# 5) PDF
|
|
if existing_pdf_path:
|
|
log.info("[job %s] reuse pdf %s", job_id, existing_pdf_path.name)
|
|
_update(job_id, status=JobStatus.DONE, progress=100, error="")
|
|
else:
|
|
_update(job_id, status=JobStatus.EXPORTING, progress=96)
|
|
log.info("[job %s] export pdf", job_id)
|
|
pdf_bytes = await _with_heartbeat(
|
|
job_id,
|
|
client.export_pdf(protocol, profile=profile, job_id=job_id,
|
|
full_transcript=full_segments),
|
|
)
|
|
pdf_path = job_dir / "protocol.pdf"
|
|
pdf_path.write_bytes(pdf_bytes)
|
|
_update(job_id, pdf_path=str(pdf_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)
|
|
_update(job_id, status=JobStatus.FAILED, error=f"{type(e).__name__}: {e}")
|