tueit_Transkriptor/llm.py

import httpx

IDENTIFY_SPEAKERS_PROMPT = """Du bekommst den Anfang eines Gesprächstranskripts mit Sprecher-Labels (SPEAKER_00, SPEAKER_01, ...).
Ermittle, welche echten Namen den Sprechern zugeordnet werden können — z.B. durch direkte Anrede ("Herr Möller", "Frank").
Antworte NUR mit einem JSON-Objekt: {"SPEAKER_00": "Name oder null", "SPEAKER_01": "Name oder null"}
Kein weiterer Text, keine Erklärung."""

TITLE_TLDR_PROMPT = """Du bekommst einen aufbereiteten Transkript-Text.
Gib NUR ein JSON-Objekt zurück mit zwei Feldern:
- "title": ein prägnanter, aussagekräftiger Titel (max. 8 Wörter, kein Datum, kein "Diktat")
- "tldr": 2-3 Sätze, die den Inhalt des Transkripts konkret zusammenfassen

Kein weiterer Text, kein Kommentar, kein Markdown-Block."""

SUMMARIZE_PROMPT = """Du bist ein präziser Assistent für Business-Kommunikation.
Du bekommst ein Gesprächstranskript mit Sprecher-Labels.
Erstelle eine strukturierte Zusammenfassung auf Deutsch mit:
1. Einem passenden H1-Titel
2. ## Wichtigste Punkte (Aufzählung)
3. ## Offene Fragen (Aufzählung, falls vorhanden)
4. ## Nächste Schritte / Ideen (Aufzählung, falls vorhanden)
Antworte NUR mit dem fertigen Markdown."""

SYSTEM_PROMPT = """Du bist ein präziser Schreibassistent.
Du bekommst einen rohen Sprachtranskript und optionale Instruktionen des Nutzers.
Deine Aufgabe:
1. Bereinige den Text (Füllwörter, Wiederholungen, Tippfehler)
2. Gliedere den Text in sinnvolle Absätze — trenne Gedanken durch Leerzeilen
3. Verwende Markdown-Überschriften (##) wenn der Text mehrere Themen hat
4. Verwende Aufzählungslisten (- ) für Aufzählungen oder Handlungsschritte
5. Erzeuge einen passenden deutschen Titel als H1
6. Beachte Instruktionen des Nutzers wenn vorhanden
7. Antworte NUR mit dem fertigen Markdown — kein Kommentar, keine Erklärung

Format:
# Titel

Erster Absatz...

Zweiter Absatz...

## Abschnitt (nur wenn sinnvoll)

- Punkt 1
- Punkt 2
"""


PUNCTUATE_PROMPT = """Du bekommst einen rohen deutschen Sprachtranskript ohne Großschreibung und Satzzeichen.
Füge AUSSCHLIESSLICH Satzzeichen (Punkt, Komma, Fragezeichen, Ausrufezeichen) und Großschreibung am Satzanfang hinzu.
Verändere KEINE Wörter, kürze NICHTS, füge NICHTS hinzu.
Wenn Sprecher-Labels vorhanden sind (z.B. **Thomas:**), behalte sie exakt so bei.
Antworte NUR mit dem korrigierten Text, ohne Kommentar."""


def _strip_code_fences(text: str) -> str:
    """Remove markdown code fences (```json ... ```) from LLM responses."""
    import re
    m = re.search(r"```(?:json)?\s*([\s\S]*?)\s*```", text)
    if m:
        return m.group(1)
    return text


class OllamaClient:
    def __init__(self, base_url: str = "http://localhost:11434"):
        self.base_url = base_url

    async def list_models(self) -> list[str]:
        async with httpx.AsyncClient() as client:
            r = await client.get(f"{self.base_url}/api/tags")
            r.raise_for_status()
            return [m["name"] for m in r.json().get("models", [])]

    async def refine(
        self,
        raw_text: str,
        instructions: str = "",
        model: str = "gemma3:12b",
    ) -> str:
        prompt = f"Transkript:\n{raw_text}"
        if instructions.strip():
            prompt += f"\n\nInstruktionen:\n{instructions.strip()}"
        async with httpx.AsyncClient(timeout=120) as client:
            r = await client.post(
                f"{self.base_url}/api/generate",
                json={"model": model, "prompt": prompt, "system": SYSTEM_PROMPT, "stream": False},
            )
            r.raise_for_status()
            return r.json()["response"]

    async def generate_title_and_tldr(
        self,
        text: str,
        model: str = "gemma3:12b",
    ) -> tuple[str, str]:
        """Return (title, tldr) for the given text. Falls back to defaults on error."""
        import json
        async with httpx.AsyncClient(timeout=60) as client:
            r = await client.post(
                f"{self.base_url}/api/generate",
                json={
                    "model": model,
                    "prompt": f"Text:\n{text[:3000]}",
                    "system": TITLE_TLDR_PROMPT,
                    "stream": False,
                },
            )
            r.raise_for_status()
            raw = _strip_code_fences(r.json()["response"].strip())
        try:
            data = json.loads(raw)
            title = str(data.get("title", "")).strip() or "Diktat"
            tldr = str(data.get("tldr", "")).strip() or "Kein TL;DR verfügbar."
            return title, tldr
        except Exception:
            return "Diktat", "Kein TL;DR verfügbar."

    async def punctuate(
        self,
        text: str,
        model: str = "gemma3:12b",
    ) -> str:
        """Add punctuation and capitalisation to raw whisper output without changing words."""
        async with httpx.AsyncClient(timeout=120) as client:
            r = await client.post(
                f"{self.base_url}/api/generate",
                json={"model": model, "prompt": text, "system": PUNCTUATE_PROMPT, "stream": False},
            )
            r.raise_for_status()
            result = r.json()["response"].strip()
        return result if result else text

    async def identify_speakers(
        self,
        transcript_excerpt: str,
        model: str = "gemma3:12b",
    ) -> dict[str, str]:
        """Try to map SPEAKER_XX labels to real names. Returns {} on failure."""
        import json
        async with httpx.AsyncClient(timeout=60) as client:
            r = await client.post(
                f"{self.base_url}/api/generate",
                json={
                    "model": model,
                    "prompt": f"Transkript-Anfang:\n{transcript_excerpt[:2000]}",
                    "system": IDENTIFY_SPEAKERS_PROMPT,
                    "stream": False,
                },
            )
            r.raise_for_status()
            raw = r.json()["response"].strip()
        try:
            data = json.loads(raw)
            if not isinstance(data, dict):
                return {}
            return {k: v for k, v in data.items() if v and str(v).lower() != "null"}
        except Exception:
            return {}

    async def summarize(
        self,
        annotated_transcript: str,
        model: str = "gemma3:12b",
    ) -> str:
        async with httpx.AsyncClient(timeout=180) as client:
            r = await client.post(
                f"{self.base_url}/api/generate",
                json={
                    "model": model,
                    "prompt": f"Transkript:\n{annotated_transcript}",
                    "system": SUMMARIZE_PROMPT,
                    "stream": False,
                },
            )
            r.raise_for_status()
            return r.json()["response"].strip()