voice-agent/lxc-frontend/app/services/pipeline.py

import json
import logging
from datetime import datetime, timezone
from pathlib import Path

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")


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


async def run_pipeline(job_id: int) -> None:
    """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:
        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"
        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:
            _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", [])
        )

        # 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)

        # 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)

        # 4) DOCX
        if existing_dx_path:
            log.info("[job %s] reuse docx %s", job_id, existing_dx_path.name)
        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), 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 client.export_pdf(protocol, profile=profile)
            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}")