import json
import os
import sqlite3
from datetime import datetime
from typing import Optional
import requests
from fastapi import FastAPI, File, Form, HTTPException, UploadFile
from fastapi.responses import HTMLResponse, PlainTextResponse, Response
API_BASE = os.getenv("API_BASE", "http://gx10.aquantico.lan:8093").rstrip("/")
OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://gx10.aquantico.lan:11434").rstrip("/")
OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "qwen3.5:9b")
DB_PATH = os.getenv("DB_PATH", "/data/ui.db")
app = FastAPI(title="Diarization UI")
def db():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
return conn
def now_iso() -> str:
return datetime.utcnow().isoformat()
def init_db():
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
with db() as c:
c.execute(
"""
CREATE TABLE IF NOT EXISTS projects (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL,
created_at TEXT NOT NULL
)
"""
)
c.execute(
"""
CREATE TABLE IF NOT EXISTS prompts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL,
prompt TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
"""
)
c.execute(
"""
CREATE TABLE IF NOT EXISTS documents (
id INTEGER PRIMARY KEY AUTOINCREMENT,
project_id INTEGER NOT NULL,
kind TEXT NOT NULL, -- transcript|analysis
title TEXT NOT NULL,
content_md TEXT NOT NULL,
source_document_id INTEGER,
prompt_id INTEGER,
raw_json TEXT,
created_at TEXT NOT NULL,
FOREIGN KEY(project_id) REFERENCES projects(id),
FOREIGN KEY(source_document_id) REFERENCES documents(id),
FOREIGN KEY(prompt_id) REFERENCES prompts(id)
)
"""
)
# defaults
c.execute("INSERT OR IGNORE INTO projects(name, created_at) VALUES (?,?)", ("Default", now_iso()))
c.execute(
"INSERT OR IGNORE INTO prompts(name, prompt, created_at, updated_at) VALUES (?,?,?,?)",
(
"Zusammenfassung",
"Erstelle eine prägnante Zusammenfassung des Gesprächs in Stichpunkten.",
now_iso(),
now_iso(),
),
)
c.execute(
"INSERT OR IGNORE INTO prompts(name, prompt, created_at, updated_at) VALUES (?,?,?,?)",
(
"Aufgaben",
"Extrahiere alle Aufgaben. Gib pro Aufgabe: Verantwortlich, Aufgabe, Deadline (falls vorhanden), Priorität.",
now_iso(),
now_iso(),
),
)
def layout(title: str, body: str) -> str:
return f"""
{title}{body}
"""
def get_projects():
with db() as c:
return c.execute("SELECT id,name FROM projects ORDER BY name").fetchall()
def get_project_name(project_id: int) -> str:
with db() as c:
r = c.execute("SELECT name FROM projects WHERE id=?", (project_id,)).fetchone()
return r[0] if r else ""
def get_prompts():
with db() as c:
return c.execute("SELECT id,name,prompt FROM prompts ORDER BY name").fetchall()
@app.on_event("startup")
def startup():
init_db()
@app.get("/manifest.webmanifest")
def manifest():
return {
"name": "Audio Copilot",
"short_name": "Copilot",
"start_url": "/",
"display": "standalone",
"background_color": "#020617",
"theme_color": "#0f172a",
"icons": [
{"src": "/icon.svg", "sizes": "any", "type": "image/svg+xml", "purpose": "any maskable"}
],
}
@app.get("/icon.svg")
def icon_svg():
svg = """"""
return Response(content=svg, media_type="image/svg+xml")
@app.get("/sw.js")
def sw_js():
js = """
self.addEventListener('install', (event) => { self.skipWaiting(); });
self.addEventListener('activate', (event) => { event.waitUntil(clients.claim()); });
self.addEventListener('fetch', (event) => {
if (event.request.method !== 'GET') return;
event.respondWith(fetch(event.request).catch(() => new Response('offline', {status: 503})));
});
"""
return Response(content=js, media_type="application/javascript")
@app.get("/healthz")
def healthz():
return {"ok": True, "api_base": API_BASE, "ollama_base_url": OLLAMA_BASE_URL, "ollama_model": OLLAMA_MODEL, "db_path": DB_PATH}
@app.get("/", response_class=HTMLResponse)
def upload_page(msg: str = ""):
projects = get_projects()
opts = "".join([f"" for p in projects])
body = f"""
Audio Upload
Audio wird transkribiert + mit Sprechern angereichert und als Dokument gespeichert.
{f"
{msg}
" if msg else ""}
"""
return layout("Upload", body)
@app.post("/projects", response_class=HTMLResponse)
def add_project(name: str = Form(...)):
with db() as c:
c.execute("INSERT INTO projects(name, created_at) VALUES (?,?)", (name.strip(), now_iso()))
return HTMLResponse("")
@app.post("/projects/update", response_class=HTMLResponse)
def rename_project(id: int = Form(...), name: str = Form(...)):
with db() as c:
c.execute("UPDATE projects SET name=? WHERE id=?", (name.strip(), id))
return HTMLResponse("")
@app.post("/projects/{project_id}/delete", response_class=HTMLResponse)
def delete_project(project_id: int):
with db() as c:
default = c.execute("SELECT id FROM projects WHERE name='Default'").fetchone()
if not default:
c.execute("INSERT INTO projects(name, created_at) VALUES (?,?)", ("Default", now_iso()))
default_id = c.execute("SELECT id FROM projects WHERE name='Default'").fetchone()[0]
else:
default_id = default[0]
if project_id == default_id:
return HTMLResponse("")
c.execute("UPDATE documents SET project_id=? WHERE project_id=?", (default_id, project_id))
c.execute("DELETE FROM projects WHERE id=?", (project_id,))
return HTMLResponse("")
@app.post("/upload", response_class=HTMLResponse)
async def upload(project_id: int = Form(...), title: str = Form(""), file: UploadFile = File(...)):
data = await file.read()
if not data:
raise HTTPException(400, "Leere Datei")
files = {"file": (file.filename or "audio.bin", data, file.content_type or "application/octet-stream")}
r = requests.post(f"{API_BASE}/transcribe-diarize", files=files, timeout=1800)
if r.status_code >= 400:
raise HTTPException(r.status_code, r.text)
payload = r.json()
content_md = payload.get("formatted_text", "")
doc_title = (title or "").strip() or (file.filename or "Transkript")
with db() as c:
cur = c.execute(
"INSERT INTO documents(project_id, kind, title, content_md, raw_json, created_at) VALUES (?,?,?,?,?,?)",
(project_id, "transcript", doc_title, content_md, json.dumps(payload, ensure_ascii=False), now_iso()),
)
doc_id = cur.lastrowid
return HTMLResponse(f"")
@app.get("/library", response_class=HTMLResponse)
def library(project_id: Optional[int] = None):
with db() as c:
projects = c.execute("SELECT id,name FROM projects ORDER BY name").fetchall()
if project_id:
docs = c.execute(
"""
SELECT d.id,d.kind,d.title,d.created_at,p.name AS project
FROM documents d JOIN projects p ON p.id=d.project_id
WHERE d.project_id=? ORDER BY d.id DESC
""",
(project_id,),
).fetchall()
else:
docs = c.execute(
"""
SELECT d.id,d.kind,d.title,d.created_at,p.name AS project
FROM documents d JOIN projects p ON p.id=d.project_id
ORDER BY d.id DESC LIMIT 200
"""
).fetchall()
p_opts = "" + "".join(
[f"" for p in projects]
)
proj_opts = "".join([f"" for p in projects])
items = "".join(
[
f"
'}
"""
return layout("Library", body)
@app.get("/document/{doc_id}", response_class=HTMLResponse)
def view_document(doc_id: int):
with db() as c:
d = c.execute(
"""
SELECT d.*, p.name AS project, pr.name AS prompt_name
FROM documents d
JOIN projects p ON p.id=d.project_id
LEFT JOIN prompts pr ON pr.id=d.prompt_id
WHERE d.id=?
""",
(doc_id,),
).fetchone()
if not d:
raise HTTPException(404, "not found")
body = f"""
"""
return layout("Dokument", body)
@app.get("/document/{doc_id}/download.md", response_class=PlainTextResponse)
def download_md(doc_id: int):
with db() as c:
d = c.execute("SELECT title,content_md FROM documents WHERE id=?", (doc_id,)).fetchone()
if not d:
raise HTTPException(404, "not found")
base = (d["title"] or f"document_{doc_id}").strip()
safe = "".join(ch if ch.isalnum() or ch in ("-", "_", " ") else "_" for ch in base).strip()
safe = safe.replace(" ", "_") or f"document_{doc_id}"
filename = f"{safe}.md"
return PlainTextResponse(
d["content_md"],
headers={"Content-Disposition": f"attachment; filename={filename}"},
)
@app.post("/document/{doc_id}/rename", response_class=HTMLResponse)
def rename_document(doc_id: int, title: str = Form(...)):
with db() as c:
c.execute("UPDATE documents SET title=? WHERE id=?", (title.strip(), doc_id))
return HTMLResponse("")
@app.post("/document/{doc_id}/move", response_class=HTMLResponse)
def move_document(doc_id: int, project_id: int = Form(...)):
with db() as c:
c.execute("UPDATE documents SET project_id=? WHERE id=?", (project_id, doc_id))
return HTMLResponse("")
@app.post("/document/{doc_id}/delete", response_class=HTMLResponse)
def delete_document(doc_id: int):
with db() as c:
c.execute("DELETE FROM documents WHERE id=?", (doc_id,))
return HTMLResponse("")
@app.get("/prompts", response_class=HTMLResponse)
def prompts_page():
with db() as c:
prompts = c.execute("SELECT * FROM prompts ORDER BY name").fetchall()
projects = c.execute("SELECT id,name FROM projects ORDER BY name").fetchall()
p_list = "".join(
[
f"
{p['name']}
{(p['prompt'] or '').replace('<','<')}
"
f""
f""
f"
"
for p in prompts
]
)
project_opts = "".join([f"" for p in projects])
project_list = "".join([
f"
{p['name']}"
f""
f""
f"
" for p in projects
])
body = f"""
Prompt-Konfiguration
Projekte
{project_list}
Prompts
{p_list}
"""
return layout("Prompts", body)
@app.post("/prompts/add", response_class=HTMLResponse)
def prompt_add(name: str = Form(...), prompt: str = Form(...)):
with db() as c:
c.execute(
"INSERT INTO prompts(name,prompt,created_at,updated_at) VALUES (?,?,?,?)",
(name.strip(), prompt.strip(), now_iso(), now_iso()),
)
return HTMLResponse("")
@app.post("/prompts/update", response_class=HTMLResponse)
def prompt_update(id: int = Form(...), name: str = Form(...), prompt: str = Form(...)):
with db() as c:
c.execute("UPDATE prompts SET name=?, prompt=?, updated_at=? WHERE id=?", (name.strip(), prompt.strip(), now_iso(), id))
return HTMLResponse("")
@app.post("/prompts/{prompt_id}/delete", response_class=HTMLResponse)
def prompt_delete(prompt_id: int):
with db() as c:
c.execute("DELETE FROM prompts WHERE id=?", (prompt_id,))
return HTMLResponse("")
@app.get("/run", response_class=HTMLResponse)
def run_page():
with db() as c:
docs = c.execute("SELECT id,title,kind,created_at FROM documents ORDER BY id DESC LIMIT 200").fetchall()
prompts = c.execute("SELECT id,name FROM prompts ORDER BY name").fetchall()
d_opts = "".join([f"" for d in docs])
p_opts = "".join([f"" for p in prompts])
body = f"""
Prompt ausführen
"""
return layout("Run", body)
@app.post("/run", response_class=HTMLResponse)
def run_prompt(document_id: int = Form(...), prompt_id: int = Form(...)):
with db() as c:
doc = c.execute("SELECT * FROM documents WHERE id=?", (document_id,)).fetchone()
prm = c.execute("SELECT * FROM prompts WHERE id=?", (prompt_id,)).fetchone()
if not doc or not prm:
raise HTTPException(404, "Dokument oder Prompt nicht gefunden")
llm_prompt = (
"Du bist ein präziser Assistent. Antworte auf Deutsch.\n"
f"AUFTRAG:\n{prm['prompt']}\n\n"
f"TEXT:\n{doc['content_md']}\n"
)
r = requests.post(
f"{OLLAMA_BASE_URL}/api/generate",
json={"model": OLLAMA_MODEL, "prompt": llm_prompt, "stream": False},
timeout=1200,
)
if r.status_code >= 400:
raise HTTPException(r.status_code, r.text)
answer = r.json().get("response", "")
with db() as c:
cur = c.execute(
"""
INSERT INTO documents(project_id, kind, title, content_md, source_document_id, prompt_id, raw_json, created_at)
VALUES (?,?,?,?,?,?,?,?)
""",
(
doc["project_id"],
"analysis",
f"Analyse: {prm['name']} · {doc['title']}",
answer,
doc["id"],
prm["id"],
json.dumps({"ollama_response": r.json()}, ensure_ascii=False),
now_iso(),
),
)
new_id = cur.lastrowid
return HTMLResponse(f"")