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

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)

        # 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),
            )
            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),
            )
            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}")