fegger
293ec8bdec
Reproduced on the 68-min meeting: the timestamped transcript filled the whole 32k window (prompt_eval 32451), gemma4 hit the length limit and returned an EMPTY answer (done_reason=length) — which the server wrote as an empty summary.md and marked "done". Fixes: - num_ctx 32768 -> 65536 and transcript cap 90k -> 100k chars (env OLLAMA_NUM_CTX / TRANSCRIPT_MAX_CHARS); ~33k tokens now fit with plenty of room for the answer - _ollama_chat treats an empty response as an ERROR (with done_reason and prompt_eval in the message) instead of producing a done-with-empty summary - truncated transcripts get a note appended to the summary - startup recovery: a container restart resets stale "pending" statuses so the UI can never stick on "wird erstellt" from dead threads - phone: the summary/agenda poll survives transient fetch errors (previously one network hiccup stopped the poll forever)
378 lines
13 KiB
Python
378 lines
13 KiB
Python
"""meetrec-server: storage + orchestration API for meetrec recordings.
|
|
|
|
Single-user service for a private tailnet: stores the recording bundles
|
|
(wav/txt/srt/json + metadata) uploaded by the meetrec apps and, later in
|
|
M3c, generates meeting summaries and agenda coverage via Ollama.
|
|
|
|
Storage layout (no database):
|
|
data/
|
|
index.json # list of recording metadata
|
|
recordings/<id>/
|
|
meeting.wav / .txt / .srt / .json
|
|
summary.md # (M3c)
|
|
agenda.json # (M3c)
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import threading
|
|
import time
|
|
import urllib.request
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from fastapi import FastAPI, File, Form, HTTPException, UploadFile
|
|
from fastapi.responses import FileResponse
|
|
|
|
DATA_DIR = Path("/data")
|
|
REC_DIR = DATA_DIR / "recordings"
|
|
INDEX_PATH = DATA_DIR / "index.json"
|
|
|
|
ALLOWED_FILES = ("wav", "txt", "srt", "json")
|
|
RID_RE = re.compile(r"^[A-Za-z0-9-]+$")
|
|
|
|
OLLAMA_URL = os.environ.get("OLLAMA_URL", "").rstrip("/")
|
|
OLLAMA_MODEL = os.environ.get("OLLAMA_MODEL", "gemma4:12b")
|
|
OLLAMA_NUM_CTX = int(os.environ.get("OLLAMA_NUM_CTX", "65536"))
|
|
OLLAMA_TIMEOUT = int(os.environ.get("OLLAMA_TIMEOUT", "600"))
|
|
|
|
# char guard so the prompt leaves room for the answer: ~3 chars/token for
|
|
# German + per-line timestamps, so 100k chars ~ 33k tokens in a 64k window
|
|
TRANSCRIPT_MAX_CHARS = int(os.environ.get("TRANSCRIPT_MAX_CHARS", "100000"))
|
|
|
|
app = FastAPI(title="meetrec-server", version="0.1.0")
|
|
_lock = threading.Lock()
|
|
|
|
|
|
def _load_index() -> list:
|
|
if not INDEX_PATH.exists():
|
|
return []
|
|
with INDEX_PATH.open() as f:
|
|
return json.load(f)
|
|
|
|
|
|
def _save_index(items: list) -> None:
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
tmp = INDEX_PATH.with_suffix(".json.tmp")
|
|
with tmp.open("w") as f:
|
|
json.dump(items, f, ensure_ascii=False, indent=2)
|
|
tmp.replace(INDEX_PATH)
|
|
|
|
|
|
def _find(items: list, rid: str) -> Optional[dict]:
|
|
return next((it for it in items if it["id"] == rid), None)
|
|
|
|
|
|
def _update_item(rid: str, **fields) -> Optional[dict]:
|
|
"""Thread-safe index field update for one recording."""
|
|
with _lock:
|
|
items = _load_index()
|
|
item = _find(items, rid)
|
|
if item is None:
|
|
return None
|
|
item.update(fields)
|
|
_save_index(items)
|
|
return dict(item)
|
|
|
|
|
|
# ------------------------------------------------------------------ Ollama
|
|
|
|
|
|
def _ollama_chat(prompt: str) -> str:
|
|
body = json.dumps({
|
|
"model": OLLAMA_MODEL,
|
|
"messages": [{"role": "user", "content": prompt}],
|
|
"stream": False,
|
|
"options": {"num_ctx": OLLAMA_NUM_CTX, "temperature": 0.2},
|
|
}).encode()
|
|
req = urllib.request.Request(
|
|
OLLAMA_URL + "/api/chat", data=body,
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=OLLAMA_TIMEOUT) as r:
|
|
data = json.load(r)
|
|
content = data["message"]["content"].strip()
|
|
if not content:
|
|
# e.g. a prompt that fills the context window yields done_reason
|
|
# "length" with an empty answer — treat as failure, never as "done"
|
|
raise RuntimeError(
|
|
"empty model response (done_reason=%s, prompt_eval=%s) — "
|
|
"transcript too long for the context window?"
|
|
% (data.get("done_reason"), data.get("prompt_eval_count")),
|
|
)
|
|
return content
|
|
|
|
|
|
def _timestamped_transcript(rec_dir: Path) -> str:
|
|
"""'[h:]mm:ss text' lines from meeting.json, meeting.txt as fallback."""
|
|
js = rec_dir / "meeting.json"
|
|
if js.is_file():
|
|
try:
|
|
with js.open() as f:
|
|
segments = json.load(f).get("segments", [])
|
|
lines = []
|
|
for s in segments:
|
|
t = int(float(s.get("start", 0)))
|
|
stamp = f"{t // 3600}:{t % 3600 // 60:02d}:{t % 60:02d}" if t >= 3600 \
|
|
else f"{t // 60:02d}:{t % 60:02d}"
|
|
lines.append(f"[{stamp}] {s.get('text', '').strip()}")
|
|
if lines:
|
|
return "\n".join(lines)
|
|
except (json.JSONDecodeError, OSError, ValueError):
|
|
pass
|
|
txt = rec_dir / "meeting.txt"
|
|
return txt.read_text() if txt.is_file() else ""
|
|
|
|
|
|
_SUMMARY_PROMPT = """Du erstellst ein Meetingprotokoll auf Deutsch.\n\
|
|
Fasse das folgende Transkript in Markdown zusammen, mit den Abschnitten:
|
|
|
|
- **Thema** — das Thema des Meetings in einem Satz
|
|
- **Wichtigste Punkte** — Stichpunkte
|
|
- **Entscheidungen** — falls erkennbar
|
|
- **Aufgaben / To-dos** — wer macht was, falls erkennbar
|
|
|
|
Transkript:
|
|
{transcript}
|
|
|
|
Antworte nur mit der Zusammenfassung, ohne Vorwort.
|
|
"""
|
|
|
|
_AGENDA_PROMPT = """Prüfe für jeden Agendapunkt, ob er im folgenden Transkript\n\
|
|
besprochen wurde.
|
|
|
|
Antworte AUSSCHLIESSLICH mit JSON — kein Markdown, keine Erklärung:
|
|
[{{"item": "<Agendapunkt>", "covered": true oder false, "time": "<mm:ss des ersten Belegs, sonst leer>", "evidence": "<kurzes Zitat, maximal 15 Wörter, sonst leer>"}}]
|
|
|
|
Agenda:
|
|
{agenda}
|
|
|
|
Transkript:
|
|
{transcript}
|
|
"""
|
|
|
|
|
|
def _generate_summary(rid: str) -> None:
|
|
rec_dir = REC_DIR / rid
|
|
try:
|
|
transcript = _timestamped_transcript(rec_dir)
|
|
if not transcript.strip():
|
|
raise RuntimeError("no transcript available")
|
|
truncated = len(transcript) > TRANSCRIPT_MAX_CHARS
|
|
prompt = _SUMMARY_PROMPT.format(transcript=transcript[:TRANSCRIPT_MAX_CHARS])
|
|
summary = _ollama_chat(prompt)
|
|
if truncated:
|
|
summary += (
|
|
"\n\n---\n\n*(Hinweis: sehr langes Meeting — das Transkript "
|
|
"wurde für die Zusammenfassung gekürzt.)*"
|
|
)
|
|
(rec_dir / "summary.md").write_text(summary, encoding="utf-8")
|
|
_update_item(rid, summary="done")
|
|
except Exception as e: # noqa: BLE001 — report in index, never crash the thread
|
|
_update_item(rid, summary=f"error: {str(e)[:160]}")
|
|
|
|
|
|
def _check_agenda(rid: str) -> None:
|
|
item = _update_item(rid) # read-only pass
|
|
items = item.get("agenda") if item else None
|
|
if not items:
|
|
return
|
|
rec_dir = REC_DIR / rid
|
|
try:
|
|
transcript = _timestamped_transcript(rec_dir)
|
|
if not transcript.strip():
|
|
raise RuntimeError("no transcript available")
|
|
agenda = "\n".join(f"- {a}" for a in items)
|
|
raw = _ollama_chat(
|
|
_AGENDA_PROMPT.format(agenda=agenda,
|
|
transcript=transcript[:TRANSCRIPT_MAX_CHARS]),
|
|
)
|
|
start, end = raw.find("["), raw.rfind("]")
|
|
if start < 0 or end <= start:
|
|
raise RuntimeError("model returned no JSON")
|
|
results = json.loads(raw[start:end + 1])
|
|
if not isinstance(results, list):
|
|
raise RuntimeError("model returned unexpected JSON")
|
|
clean = [
|
|
{
|
|
"item": str(r.get("item", ""))[:300],
|
|
"covered": bool(r.get("covered", False)),
|
|
"time": str(r.get("time", ""))[:20],
|
|
"evidence": str(r.get("evidence", ""))[:200],
|
|
}
|
|
for r in results if isinstance(r, dict)
|
|
]
|
|
(rec_dir / "agenda.json").write_text(
|
|
json.dumps(clean, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
_update_item(rid, agenda_results=clean, agenda_status="done")
|
|
except Exception as e: # noqa: BLE001
|
|
_update_item(rid, agenda_results=[{"error": str(e)[:120]}],
|
|
agenda_status=f"error: {str(e)[:120]}")
|
|
|
|
|
|
def _postprocess(rid: str) -> None:
|
|
"""Background pipeline after an upload: summary, then agenda check."""
|
|
_generate_summary(rid)
|
|
_check_agenda(rid)
|
|
|
|
|
|
@app.on_event("startup")
|
|
def startup() -> None:
|
|
"""Recover stale state: a container restart kills in-flight Ollama
|
|
threads, which would leave recordings "pending" forever."""
|
|
with _lock:
|
|
items = _load_index()
|
|
dirty = False
|
|
for it in items:
|
|
if it.get("summary") == "pending":
|
|
it["summary"] = "error: interrupted by restart"
|
|
dirty = True
|
|
if it.get("agenda_status") == "pending":
|
|
it["agenda_status"] = "error: interrupted by restart"
|
|
dirty = True
|
|
if dirty:
|
|
_save_index(items)
|
|
|
|
|
|
@app.get("/api/health")
|
|
def health() -> dict:
|
|
with _lock:
|
|
return {"ok": True, "recordings": len(_load_index())}
|
|
|
|
|
|
@app.get("/api/recordings")
|
|
def list_recordings() -> list:
|
|
with _lock:
|
|
items = _load_index()
|
|
return list(reversed(items)) # newest first
|
|
|
|
|
|
@app.post("/api/recordings", status_code=201)
|
|
async def upload(
|
|
wav: UploadFile = File(...),
|
|
txt: Optional[UploadFile] = File(None),
|
|
srt: Optional[UploadFile] = File(None),
|
|
json_file: Optional[UploadFile] = File(None, alias="json"),
|
|
started_at: str = Form(""),
|
|
duration_ms: int = Form(0),
|
|
language: str = Form(""),
|
|
device: str = Form(""),
|
|
agenda: str = Form("[]"),
|
|
) -> dict:
|
|
uploads = []
|
|
for name, f in (("wav", wav), ("txt", txt), ("srt", srt), ("json", json_file)):
|
|
if f is not None and f.filename:
|
|
uploads.append((name, f))
|
|
if not any(name == "wav" for name, _ in uploads):
|
|
raise HTTPException(400, "a 'wav' file part is required")
|
|
|
|
try:
|
|
agenda_items = json.loads(agenda or "[]")
|
|
except json.JSONDecodeError:
|
|
raise HTTPException(400, "agenda must be a JSON list of strings")
|
|
if not isinstance(agenda_items, list) or not all(isinstance(a, str) for a in agenda_items):
|
|
raise HTTPException(400, "agenda must be a JSON list of strings")
|
|
|
|
rid = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") + "-" + uuid.uuid4().hex[:4]
|
|
rec_dir = REC_DIR / rid
|
|
rec_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
stored = []
|
|
for name, f in uploads:
|
|
dest = rec_dir / f"meeting.{name}"
|
|
with dest.open("wb") as out:
|
|
while chunk := await f.read(1 << 20):
|
|
out.write(chunk)
|
|
stored.append(dest.name)
|
|
|
|
item = {
|
|
"id": rid,
|
|
"started_at": started_at or datetime.now(timezone.utc).isoformat(timespec="seconds"),
|
|
"duration_ms": duration_ms,
|
|
"language": language,
|
|
"device": device,
|
|
"agenda": agenda_items,
|
|
"files": sorted(stored),
|
|
"summary": None, # none | pending | done | error: ...
|
|
"agenda_results": None,
|
|
"agenda_status": None,
|
|
"uploaded_at": time.time(),
|
|
}
|
|
with _lock:
|
|
items = _load_index()
|
|
items.append(item)
|
|
_save_index(items)
|
|
|
|
if OLLAMA_URL:
|
|
threading.Thread(target=_postprocess, args=(rid,), daemon=True).start()
|
|
return {"id": rid, "files": sorted(stored)}
|
|
|
|
|
|
@app.get("/api/recordings/{rid}")
|
|
def get_recording(rid: str) -> dict:
|
|
with _lock:
|
|
item = _find(_load_index(), rid)
|
|
if item is None:
|
|
raise HTTPException(404, "unknown recording id")
|
|
return item
|
|
|
|
|
|
@app.get("/api/recordings/{rid}/files/{name}")
|
|
def get_file(rid: str, name: str) -> FileResponse:
|
|
if not RID_RE.match(rid) or "/" in name or ".." in name:
|
|
raise HTTPException(400, "invalid id or file name")
|
|
with _lock:
|
|
item = _find(_load_index(), rid)
|
|
if item is None:
|
|
raise HTTPException(404, "unknown recording id")
|
|
if name not in item["files"] and name not in ("summary.md", "agenda.json"):
|
|
raise HTTPException(404, f"no such file: {name}")
|
|
path = REC_DIR / rid / name
|
|
if not path.is_file():
|
|
raise HTTPException(404, f"no such file: {name}")
|
|
return FileResponse(path)
|
|
|
|
|
|
@app.post("/api/recordings/{rid}/summary", status_code=202)
|
|
def trigger_summary(rid: str, force: bool = False) -> dict:
|
|
with _lock:
|
|
item = _find(_load_index(), rid)
|
|
if item is None:
|
|
raise HTTPException(404, "unknown recording id")
|
|
if item.get("summary") == "pending" and not force:
|
|
return {"status": "pending"}
|
|
_update_item(rid, summary="pending")
|
|
threading.Thread(target=_generate_summary, args=(rid,), daemon=True).start()
|
|
return {"status": "started"}
|
|
|
|
|
|
@app.post("/api/recordings/{rid}/agenda", status_code=202)
|
|
def trigger_agenda(rid: str, force: bool = False) -> dict:
|
|
with _lock:
|
|
item = _find(_load_index(), rid)
|
|
if item is None:
|
|
raise HTTPException(404, "unknown recording id")
|
|
if not item.get("agenda"):
|
|
raise HTTPException(400, "recording has no agenda items")
|
|
if item.get("agenda_status") == "pending" and not force:
|
|
return {"status": "pending"}
|
|
_update_item(rid, agenda_status="pending")
|
|
threading.Thread(target=_check_agenda, args=(rid,), daemon=True).start()
|
|
return {"status": "started"}
|
|
|
|
|
|
@app.delete("/api/recordings/{rid}", status_code=204)
|
|
def delete_recording(rid: str) -> None:
|
|
if not RID_RE.match(rid):
|
|
raise HTTPException(400, "invalid id")
|
|
with _lock:
|
|
items = _load_index()
|
|
item = _find(items, rid)
|
|
if item is None:
|
|
raise HTTPException(404, "unknown recording id")
|
|
_save_index([it for it in items if it["id"] != rid])
|
|
shutil.rmtree(REC_DIR / rid, ignore_errors=True) |