root
1c1962e1d8
Browseren streamer 16 kHz PCM direkte til ElevenLabs' realtime-STT med et engangs-token fra serveren (/api/stt-token); teksten vokser under optagelsen og sendes med kladden, så serveren slipper for at skrive af. 'Spørgsmål og svar' bruger ét audio-element (iOS blokerer ellers det andet play()) og laver spørgsmålets lyd on demand for ældre klip. Admin kan rense favorit-klip med ElevenLabs Audio Isolation (<id>-ren.mp3), aldrig som standard. Claude-Session: https://claude.ai/code/session_01JmhJG6JnhXfzNiKv8f2vHu
689 lines
28 KiB
Python
689 lines
28 KiB
Python
"""Erindringer: familien fortæller sine minder (eller fører dagbog) til en AI-interviewer.
|
|
|
|
Interviewer = Claude (Anthropic Messages API via httpx; ingen SDK), der stiller ét spørgsmål ad
|
|
gangen og følger op på det, personen lige har sagt. Spørgsmålet læses op af ElevenLabs (cachet pr.
|
|
tekst), svaret optages i browseren, transskriberes af ElevenLabs STT og gemmes som mp3 + tekst.
|
|
Klippene bruges som råstof til podcasten (Slægtens Stemmer) og som personlig dagbog.
|
|
|
|
Data (ikke i git, persondata):
|
|
DATA/erindringer/projekter.json slug -> titel, beskrivelse, kontekst (til Claude), start (spørgsmål)
|
|
DATA/erindringer/personer.json pid -> navn, bruger | noegle, projekter, oprettet, sidst
|
|
DATA/erindringer/klip/<pid>/<id>.mp3 svaret (mono 64 kbit)
|
|
DATA/erindringer/klip/<pid>/<id>.json id, tid, projekt, samtale, spoergsmaal, tekst, varighed, favorit
|
|
DATA/erindringer/tts/<sha>.mp3 oplæste spørgsmål
|
|
|
|
Personer: indloggede brugere har pid "u:<username>" og kan altid føre dagbog; gæster (familie uden
|
|
login) får pid "g:<slug>" og en hemmelig nøgle i adressen, som podcasten gør det. Admin bestemmer,
|
|
hvilke projekter en person inviteres til; projektet "dagbog" er altid åbent for indloggede.
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
import config
|
|
|
|
ROOT = config.DATA / "erindringer"
|
|
PROJEKTER = ROOT / "projekter.json"
|
|
PERSONER = ROOT / "personer.json"
|
|
KLIP = ROOT / "klip"
|
|
TTS = ROOT / "tts"
|
|
SLUG = re.compile(r"^[a-z0-9][a-z0-9-]{0,40}$")
|
|
KEY = re.compile(r"^[A-Za-z0-9_-]{32,64}$")
|
|
KLIPID = re.compile(r"^[0-9]{8}-[0-9]{6}-[a-f0-9]{6}$")
|
|
PID = re.compile(r"^[ug]:[A-Za-z0-9_.@-]{1,64}$")
|
|
|
|
ANTHROPIC_KEY = os.environ.get("ANTHROPIC_API_KEY", "")
|
|
ANTHROPIC_WORKSPACE = os.environ.get("ANTHROPIC_WORKSPACE_ID", "")
|
|
MODEL = os.environ.get("ERINDRINGER_MODEL", "claude-sonnet-5")
|
|
ELEVEN_KEY = os.environ.get("ELEVEN_API_KEY", "")
|
|
VOICE = os.environ.get("ERINDRINGER_VOICE", "tJDFCHyviItsYF1qqToS") # Leif, podcastens fortæller
|
|
MAX_SVAR_SEK = 15 * 60
|
|
MAX_UPLOAD = 40 * 1024 * 1024
|
|
|
|
DAGBOG = {
|
|
"titel": "Dagbog",
|
|
"beskrivelse": "Hvad der fylder lige nu. Et par minutter, når du har lyst; interviewerens spørgsmål hjælper dig i gang.",
|
|
"kontekst": "Dette er personens private dagbog. Spørg til hverdagen, til det der fylder, til små og store ting, "
|
|
"og følg op på det, personen selv bringer på bane. Vær nysgerrig uden at være nærgående.",
|
|
"start": ["Hvad har fyldt mest for dig i dag?", "Hvad har du tænkt på i denne uge?",
|
|
"Er der sket noget siden sidst, du gerne vil huske?"],
|
|
}
|
|
|
|
|
|
# ---------------- fil-hjælpere ----------------
|
|
|
|
def _læs(p: Path, standard):
|
|
try:
|
|
return json.loads(p.read_text())
|
|
except (OSError, ValueError):
|
|
return standard
|
|
|
|
|
|
def _skriv(p: Path, data) -> None:
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = p.with_name("." + p.name + ".tmp")
|
|
tmp.write_text(json.dumps(data, ensure_ascii=False, indent=1))
|
|
os.replace(tmp, p)
|
|
|
|
|
|
def _nu() -> str:
|
|
return time.strftime("%Y-%m-%dT%H:%M:%S%z")
|
|
|
|
|
|
# ---------------- projekter ----------------
|
|
|
|
def projekter() -> dict:
|
|
p = _læs(PROJEKTER, {})
|
|
p.setdefault("dagbog", dict(DAGBOG))
|
|
return p
|
|
|
|
|
|
def projekt(slug: str) -> dict | None:
|
|
if not SLUG.match(slug or ""):
|
|
return None
|
|
p = projekter().get(slug)
|
|
return {**p, "slug": slug} if p else None
|
|
|
|
|
|
def gem_projekt(slug: str, titel: str, beskrivelse: str, kontekst: str, start: list[str], opfoelg: int = 2) -> bool:
|
|
"""start = de forberedte spørgsmål, som intervieweren skal nå igennem; opfoelg = højst så mange
|
|
opfølgende spørgsmål pr. forberedt spørgsmål, før den går videre."""
|
|
if not SLUG.match(slug or "") or not titel.strip():
|
|
return False
|
|
p = projekter()
|
|
gammel = p.get(slug, {})
|
|
p[slug] = {"titel": titel.strip(), "beskrivelse": beskrivelse.strip(), "kontekst": kontekst.strip(),
|
|
"start": [s.strip() for s in start if s.strip()], "opfoelg": max(0, min(6, int(opfoelg or 0))),
|
|
"oprettet": gammel.get("oprettet") or _nu()}
|
|
_skriv(PROJEKTER, p)
|
|
return True
|
|
|
|
|
|
def slet_projekt(slug: str) -> bool:
|
|
p = projekter()
|
|
if slug == "dagbog" or slug not in p:
|
|
return False
|
|
del p[slug]
|
|
_skriv(PROJEKTER, p)
|
|
ps = personer()
|
|
for v in ps.values():
|
|
if slug in v.get("projekter", []):
|
|
v["projekter"].remove(slug)
|
|
_skriv(PERSONER, ps)
|
|
return True
|
|
|
|
|
|
# ---------------- personer ----------------
|
|
|
|
def personer() -> dict:
|
|
return _læs(PERSONER, {})
|
|
|
|
|
|
def person(pid: str) -> dict | None:
|
|
if not PID.match(pid or ""):
|
|
return None
|
|
v = personer().get(pid)
|
|
return {**v, "pid": pid} if v else None
|
|
|
|
|
|
def person_for_bruger(username: str, navn: str) -> dict:
|
|
"""Indlogget bruger: oprettes første gang, dagbog er altid med."""
|
|
pid = f"u:{username}"
|
|
ps = personer()
|
|
v = ps.get(pid)
|
|
if v is None:
|
|
v = {"navn": navn or username, "bruger": username, "projekter": ["dagbog"], "oprettet": _nu()}
|
|
ps[pid] = v
|
|
else:
|
|
if navn and v.get("navn") != navn:
|
|
v["navn"] = navn
|
|
if "dagbog" not in v.setdefault("projekter", []):
|
|
v["projekter"].insert(0, "dagbog")
|
|
v["sidst"] = _nu()
|
|
_skriv(PERSONER, ps)
|
|
return {**v, "pid": pid}
|
|
|
|
|
|
def person_for_noegle(key: str) -> dict | None:
|
|
if not KEY.match(key or ""):
|
|
return None
|
|
for pid, v in personer().items():
|
|
if v.get("noegle") == key and not v.get("lukket"):
|
|
v["sidst"] = _nu()
|
|
ps = personer(); ps[pid]["sidst"] = v["sidst"]; _skriv(PERSONER, ps)
|
|
return {**v, "pid": pid}
|
|
return None
|
|
|
|
|
|
def ny_gaest(navn: str, projekt_slugs: list[str]) -> dict:
|
|
base = re.sub(r"[^a-z0-9]+", "-", navn.lower().replace("æ", "ae").replace("ø", "oe").replace("å", "aa")).strip("-")[:30] or "gaest"
|
|
ps = personer()
|
|
slug, n = base, 2
|
|
while f"g:{slug}" in ps:
|
|
slug, n = f"{base}-{n}", n + 1
|
|
pid = f"g:{slug}"
|
|
ps[pid] = {"navn": navn.strip(), "noegle": secrets.token_urlsafe(32), "projekter": list(projekt_slugs),
|
|
"oprettet": _nu()}
|
|
_skriv(PERSONER, ps)
|
|
return {**ps[pid], "pid": pid}
|
|
|
|
|
|
def saet_projekter(pid: str, projekt_slugs: list[str]) -> bool:
|
|
ps = personer()
|
|
if pid not in ps:
|
|
return False
|
|
ps[pid]["projekter"] = [s for s in projekt_slugs if s in projekter()]
|
|
if pid.startswith("u:") and "dagbog" not in ps[pid]["projekter"]:
|
|
ps[pid]["projekter"].insert(0, "dagbog")
|
|
_skriv(PERSONER, ps)
|
|
return True
|
|
|
|
|
|
def ny_noegle(pid: str) -> str | None:
|
|
ps = personer()
|
|
if pid not in ps or not pid.startswith("g:"):
|
|
return None
|
|
ps[pid]["noegle"] = secrets.token_urlsafe(32)
|
|
ps[pid].pop("lukket", None)
|
|
_skriv(PERSONER, ps)
|
|
return ps[pid]["noegle"]
|
|
|
|
|
|
def luk_gaest(pid: str, lukket: bool) -> bool:
|
|
ps = personer()
|
|
if pid not in ps or not pid.startswith("g:"):
|
|
return False
|
|
if lukket:
|
|
ps[pid]["lukket"] = _nu()
|
|
else:
|
|
ps[pid].pop("lukket", None)
|
|
_skriv(PERSONER, ps)
|
|
return True
|
|
|
|
|
|
def alle_personer() -> list[dict]:
|
|
ud = []
|
|
for pid, v in personer().items():
|
|
ud.append({**v, "pid": pid, "antal": len(klip_for(pid))})
|
|
return sorted(ud, key=lambda v: (v.get("sidst") or v.get("oprettet") or ""), reverse=True)
|
|
|
|
|
|
def maa(p: dict, slug: str) -> bool:
|
|
return slug in (p.get("projekter") or []) and projekt(slug) is not None
|
|
|
|
|
|
# ---------------- klip ----------------
|
|
|
|
def _klipmappe(pid: str) -> Path:
|
|
return KLIP / pid.replace(":", "_")
|
|
|
|
|
|
def klip_for(pid: str, projekt_slug: str | None = None, samtale: str | None = None, med_kladder: bool = False) -> list[dict]:
|
|
"""Klip for en person, ældste først. Kladder (optagelser, der ikke er sendt endnu) udelades."""
|
|
d = _klipmappe(pid)
|
|
if not PID.match(pid or "") or not d.exists():
|
|
return []
|
|
ud = []
|
|
for meta in d.glob("*.json"):
|
|
if not KLIPID.match(meta.stem):
|
|
continue
|
|
k = _læs(meta, None)
|
|
if not k or not (d / f"{meta.stem}.mp3").exists():
|
|
continue
|
|
if projekt_slug and k.get("projekt") != projekt_slug:
|
|
continue
|
|
if samtale and k.get("samtale") != samtale:
|
|
continue
|
|
if k.get("kladde") and not med_kladder:
|
|
continue
|
|
ud.append(k)
|
|
return sorted(ud, key=lambda k: (k.get("tid") or "", k["id"]))
|
|
|
|
|
|
def klip(pid: str, kid: str) -> dict | None:
|
|
if not PID.match(pid or "") or not KLIPID.match(kid or ""):
|
|
return None
|
|
k = _læs(_klipmappe(pid) / f"{kid}.json", None)
|
|
return k if k and k.get("id") == kid else None
|
|
|
|
|
|
def klip_fil(pid: str, kid: str) -> Path | None:
|
|
if not PID.match(pid or "") or not KLIPID.match(kid or ""):
|
|
return None
|
|
p = _klipmappe(pid) / f"{kid}.mp3"
|
|
return p if p.exists() else None
|
|
|
|
|
|
def _varighed(p: Path) -> int:
|
|
try:
|
|
ud = subprocess.run(["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", str(p)],
|
|
capture_output=True, text=True, timeout=30).stdout.strip()
|
|
return int(float(ud))
|
|
except (OSError, ValueError, subprocess.SubprocessError):
|
|
return 0
|
|
|
|
|
|
def gem_svar(pid: str, projekt_slug: str, samtale: str, spoergsmaal: str, lyd: bytes | None, filnavn: str,
|
|
tekst: str | None, spoergsmaal_talt: str = "", lyd_sha: str = "", kilde: str = "",
|
|
kladde: bool = False) -> dict:
|
|
"""Gem et svar: lyd konverteres til mp3 og transskriberes; ren tekst gemmes som det er.
|
|
Spørgsmålets oplæsning (lyd_sha fra tts-cachen) kopieres med som <id>-sp.mp3, så spørgsmål og
|
|
svar kan klippes sammen i en podcast. Returnerer klip-metadata. Kaster ValueError med besked."""
|
|
if not PID.match(pid or "") or not SLUG.match(projekt_slug or ""):
|
|
raise ValueError("Ugyldig person eller projekt.")
|
|
d = _klipmappe(pid)
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
kid = time.strftime("%Y%m%d-%H%M%S") + "-" + secrets.token_hex(3)
|
|
mp3 = d / f"{kid}.mp3"
|
|
sp = tts_fil(lyd_sha) if lyd_sha else None
|
|
if sp is None and spoergsmaal_talt:
|
|
sp = tts_sti(spoergsmaal_talt)
|
|
if sp:
|
|
(d / f"{kid}-sp.mp3").write_bytes(sp.read_bytes())
|
|
varighed = 0
|
|
if lyd:
|
|
if len(lyd) > MAX_UPLOAD:
|
|
raise ValueError("Optagelsen er for stor. Prøv at dele den op i to svar.")
|
|
suffix = "." + (re.sub(r"[^a-z0-9]", "", (filnavn or "").rsplit(".", 1)[-1].lower()) or "webm")
|
|
with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
|
|
tmp.write(lyd)
|
|
raa = Path(tmp.name)
|
|
try:
|
|
r = subprocess.run(["ffmpeg", "-y", "-v", "error", "-i", str(raa), "-vn", "-ac", "1", "-ar", "44100",
|
|
"-c:a", "libmp3lame", "-b:a", "64k", str(mp3)], capture_output=True, text=True, timeout=300)
|
|
if r.returncode or not mp3.exists():
|
|
raise ValueError("Optagelsen kunne ikke læses. Prøv igen, eller skriv svaret i stedet.")
|
|
varighed = _varighed(mp3)
|
|
if varighed > MAX_SVAR_SEK:
|
|
mp3.unlink(missing_ok=True)
|
|
raise ValueError("Svaret er længere end et kvarter. Del det op, så bliver det også nemmere at bruge.")
|
|
# Transskriptionen laves bagefter (transskriber_klip), så svaret gemmes med det samme
|
|
finally:
|
|
raa.unlink(missing_ok=True)
|
|
else:
|
|
if not (tekst or "").strip():
|
|
raise ValueError("Der kom hverken lyd eller tekst med.")
|
|
# Tekstsvar: en stille mp3 på 1 s, så alle klip har samme form (afspiller + tekst).
|
|
subprocess.run(["ffmpeg", "-y", "-v", "error", "-f", "lavfi", "-i", "anullsrc=r=44100:cl=mono", "-t", "1",
|
|
"-c:a", "libmp3lame", "-b:a", "32k", str(mp3)], capture_output=True, timeout=60)
|
|
meta = {"id": kid, "tid": _nu(), "projekt": projekt_slug, "samtale": samtale, "spoergsmaal": spoergsmaal.strip()[:600],
|
|
"spoergsmaal_talt": (spoergsmaal_talt or "").strip()[:800], "spoergsmaal_lyd": bool(sp),
|
|
"kilde": kilde if kilde in ("forberedt", "opfoelgning", "afrunding") else "",
|
|
"tekst": (tekst or "").strip(), "varighed": varighed, "favorit": False, "skrevet": not lyd,
|
|
"kladde": bool(kladde), "status": "transskriberes" if (lyd and not (tekst or "").strip()) else "klar"}
|
|
_skriv(d / f"{kid}.json", meta)
|
|
return meta
|
|
|
|
|
|
def transskriber_klip(pid: str, kid: str) -> dict | None:
|
|
"""Transskribér et gemt lydsvar og skriv teksten ind; idempotent."""
|
|
k = klip(pid, kid)
|
|
if not k:
|
|
return None
|
|
if k.get("status") != "transskriberes":
|
|
return k
|
|
f = klip_fil(pid, kid)
|
|
tekst = transskriber(f) if f else ""
|
|
k = klip(pid, kid) or k # genlæs: brugeren kan have bekræftet imens
|
|
k["tekst"] = k["tekst"] or tekst
|
|
k["status"] = "klar" if tekst else "fejl"
|
|
_skriv(_klipmappe(pid) / f"{kid}.json", k)
|
|
return k
|
|
|
|
|
|
def bekraeft(pid: str, kid: str) -> dict | None:
|
|
k = klip(pid, kid)
|
|
if not k:
|
|
return None
|
|
k.pop("kladde", None)
|
|
_skriv(_klipmappe(pid) / f"{kid}.json", k)
|
|
return k
|
|
|
|
|
|
# ---------------- næste spørgsmål, forberedt i baggrunden ----------------
|
|
|
|
def _naeste_sti(pid: str, samtale: str) -> Path:
|
|
return _klipmappe(pid) / f"_naeste_{samtale}.json"
|
|
|
|
|
|
def gem_forberedt(pid: str, samtale: str, basis: str, q: dict, lyd_sha: str) -> None:
|
|
_skriv(_naeste_sti(pid, samtale), {"basis": basis, "q": q, "lyd_sha": lyd_sha, "tid": _nu()})
|
|
|
|
|
|
def hent_forberedt(pid: str, samtale: str, basis: str) -> dict | None:
|
|
"""Det forudberegnede næste spørgsmål, hvis det bygger på netop det seneste svar."""
|
|
p = _naeste_sti(pid, samtale)
|
|
v = _læs(p, None)
|
|
if v and v.get("basis") == basis:
|
|
p.unlink(missing_ok=True)
|
|
return v
|
|
return None
|
|
|
|
|
|
def ret_tekst(pid: str, kid: str, tekst: str) -> bool:
|
|
k = klip(pid, kid)
|
|
if not k:
|
|
return False
|
|
k["tekst"] = tekst.strip()[:20000]
|
|
k["redigeret"] = _nu()
|
|
_skriv(_klipmappe(pid) / f"{kid}.json", k)
|
|
return True
|
|
|
|
|
|
def saet_favorit(pid: str, kid: str, favorit: bool) -> bool:
|
|
k = klip(pid, kid)
|
|
if not k:
|
|
return False
|
|
k["favorit"] = favorit
|
|
_skriv(_klipmappe(pid) / f"{kid}.json", k)
|
|
return True
|
|
|
|
|
|
def slet_klip(pid: str, kid: str) -> bool:
|
|
k = klip(pid, kid)
|
|
if not k:
|
|
return False
|
|
d = _klipmappe(pid)
|
|
for navn in (f"{kid}.json", f"{kid}.mp3", f"{kid}-sp.mp3", f"{kid}-ren.mp3"):
|
|
(d / navn).unlink(missing_ok=True)
|
|
return True
|
|
|
|
|
|
def spoergsmaal_fil(pid: str, kid: str, lav: bool = True) -> Path | None:
|
|
"""Spørgsmålets oplæsning til et svar. Mangler den (ældre klip), laves den fra den talte tekst;
|
|
teksten er den samme, så tts-cachen rammer oftest."""
|
|
if not PID.match(pid or "") or not KLIPID.match(kid or ""):
|
|
return None
|
|
p = _klipmappe(pid) / f"{kid}-sp.mp3"
|
|
if p.exists():
|
|
return p
|
|
if not lav:
|
|
return None
|
|
k = klip(pid, kid)
|
|
if not k:
|
|
return None
|
|
sp = tts_sti(k.get("spoergsmaal_talt") or k.get("spoergsmaal") or "")
|
|
if not sp:
|
|
return None
|
|
p.write_bytes(sp.read_bytes())
|
|
k["spoergsmaal_lyd"] = True
|
|
_skriv(_klipmappe(pid) / f"{kid}.json", k)
|
|
return p
|
|
|
|
|
|
def renset_fil(pid: str, kid: str) -> Path | None:
|
|
if not PID.match(pid or "") or not KLIPID.match(kid or ""):
|
|
return None
|
|
p = _klipmappe(pid) / f"{kid}-ren.mp3"
|
|
return p if p.exists() else None
|
|
|
|
|
|
def isoler_stemme(pid: str, kid: str) -> bool:
|
|
"""ElevenLabs Audio Isolation: fjern baggrundsstøj fra svaret -> <id>-ren.mp3 (valgfrit, pr. klip;
|
|
koster ca. 1000 credits pr. minut, så det er en admin-knap, ikke standard)."""
|
|
f = klip_fil(pid, kid)
|
|
if not f or not ELEVEN_KEY:
|
|
return False
|
|
try:
|
|
with open(f, "rb") as fh:
|
|
r = httpx.post("https://api.elevenlabs.io/v1/audio-isolation", headers={"xi-api-key": ELEVEN_KEY},
|
|
files={"audio": (f.name, fh, "audio/mpeg")}, timeout=600)
|
|
if r.status_code != 200 or len(r.content) < 1000:
|
|
return False
|
|
except httpx.HTTPError:
|
|
return False
|
|
ud = _klipmappe(pid) / f"{kid}-ren.mp3"
|
|
tmp = ud.with_name("." + ud.name + ".tmp")
|
|
tmp.write_bytes(r.content)
|
|
os.replace(tmp, ud)
|
|
k = klip(pid, kid)
|
|
if k:
|
|
k["renset"] = True
|
|
_skriv(_klipmappe(pid) / f"{kid}.json", k)
|
|
return True
|
|
|
|
|
|
def stt_token() -> dict | None:
|
|
"""Engangs-token til browserens direkte forbindelse til ElevenLabs' realtime-STT (15 min, kun STT),
|
|
så API-nøglen aldrig forlader serveren."""
|
|
if not ELEVEN_KEY:
|
|
return None
|
|
try:
|
|
r = httpx.post("https://api.elevenlabs.io/v1/single-use-token/realtime_scribe",
|
|
headers={"xi-api-key": ELEVEN_KEY}, timeout=30)
|
|
if r.status_code == 200 and r.json().get("token"):
|
|
t = r.json()["token"]
|
|
return {"token": t, "url": "wss://api.elevenlabs.io/v1/speech-to-text/realtime?model_id=scribe_v2_realtime"
|
|
f"&token={t}&audio_format=pcm_16000&sample_rate=16000&commit_strategy=vad&language_code=da"}
|
|
except (httpx.HTTPError, ValueError):
|
|
pass
|
|
return None
|
|
|
|
|
|
# ---------------- ElevenLabs: tale ----------------
|
|
|
|
def transskriber(mp3: Path) -> str:
|
|
if not ELEVEN_KEY:
|
|
return ""
|
|
try:
|
|
with open(mp3, "rb") as f:
|
|
r = httpx.post("https://api.elevenlabs.io/v1/speech-to-text", headers={"xi-api-key": ELEVEN_KEY},
|
|
data={"model_id": "scribe_v2", "language_code": "da"},
|
|
files={"file": (mp3.name, f, "audio/mpeg")}, timeout=600)
|
|
if r.status_code == 200:
|
|
return (r.json().get("text") or "").strip()
|
|
except httpx.HTTPError:
|
|
pass
|
|
return ""
|
|
|
|
|
|
def tts_sti(tekst: str) -> Path | None:
|
|
"""Oplæst spørgsmål, cachet på tekstens hash; None hvis det ikke kunne laves."""
|
|
tekst = (tekst or "").strip()
|
|
if not tekst or not ELEVEN_KEY:
|
|
return None
|
|
sha = hashlib.sha256((VOICE + "|" + tekst).encode()).hexdigest()[:32]
|
|
p = TTS / f"{sha}.mp3"
|
|
if p.exists():
|
|
return p
|
|
TTS.mkdir(parents=True, exist_ok=True)
|
|
body = {"text": tekst, "model_id": "eleven_v4", "language_code": "da",
|
|
"voice_settings": {"stability": 0.5, "similarity_boost": 0.8, "style": 0.2, "use_speaker_boost": True}}
|
|
try:
|
|
r = httpx.post(f"https://api.elevenlabs.io/v1/text-to-speech/{VOICE}?output_format=mp3_44100_64",
|
|
headers={"xi-api-key": ELEVEN_KEY}, json=body, timeout=120)
|
|
if r.status_code == 200 and r.content[:3] in (b"ID3", b"\xff\xfb", b"\xff\xf3"):
|
|
tmp = p.with_name("." + p.name + ".tmp")
|
|
tmp.write_bytes(r.content)
|
|
os.replace(tmp, p)
|
|
return p
|
|
except httpx.HTTPError:
|
|
pass
|
|
return None
|
|
|
|
|
|
def tts_fil(sha: str) -> Path | None:
|
|
if not re.match(r"^[a-f0-9]{32}$", sha or ""):
|
|
return None
|
|
p = TTS / f"{sha}.mp3"
|
|
return p if p.exists() else None
|
|
|
|
|
|
# ---------------- Claude: interviewer ----------------
|
|
|
|
# Lyd-tags, som ElevenLabs v4 udfører i stedet for at læse dem op. Alt andet i [klammer] fjernes.
|
|
AUDIO_TAGS = {"warmly", "softly", "curious", "gently", "chuckles", "laughs", "pause", "thoughtful", "amused",
|
|
"sighs", "whispers", "excited", "surprised", "sincere", "calm", "slowly"}
|
|
_TAG = re.compile(r"\[([a-zA-Z ]+)\]")
|
|
|
|
|
|
def keep_tags(text: str) -> str:
|
|
return _TAG.sub(lambda m: m.group(0) if m.group(1).strip().lower() in AUDIO_TAGS else "", text or "")
|
|
|
|
|
|
def untag(text: str) -> str:
|
|
return re.sub(r"\s{2,}", " ", _TAG.sub("", text or "")).strip()
|
|
|
|
|
|
SYSTEM = """Du er en varm, rolig interviewer for en dansk familie. Du hjælper et familiemedlem med at fortælle
|
|
sine minder (eller føre dagbog) ved at stille ÉT spørgsmål ad gangen på dansk. Samtalen skal føles som en
|
|
stille snak over kaffen, ikke som et forhør.
|
|
|
|
Sådan styrer du samtalen:
|
|
- Emnet har en liste FORBEREDTE spørgsmål. Dem skal du nå igennem i løbet af samtalen, i den rækkefølge der
|
|
passer. "tilbage" viser, hvilke der endnu ikke er stillet (eller er sprunget over; dem lader du ligge).
|
|
- Når personen lige har svaret, må du stille højst "opfoelg_max" OPFØLGENDE spørgsmål til det svar, og kun hvis
|
|
der er noget konkret at spørge ind til: et navn, et sted, en ting, en pudsig detalje, noget personen selv
|
|
griner af eller skynder sig forbi. Det pudsige er guld; spørg ind til det med et smil. "opfoelg_brugt" fortæller,
|
|
hvor mange du allerede har brugt på det aktuelle forberedte spørgsmål.
|
|
- Er der ikke noget konkret at følge op på, eller er opfølgningerne brugt, så gå videre til næste forberedte
|
|
spørgsmål. Kobl gerne overgangen til noget, personen har sagt ("Du nævnte …, det minder mig om …").
|
|
- Hvis svaret stikker af fra emnet, så tag venligt tråden op igen med næste forberedte spørgsmål, uden at
|
|
påpege afvigelsen.
|
|
- Når alle forberedte spørgsmål er stillet og opfølgningerne brugt, stiller du ét let, afrundende spørgsmål
|
|
("Er der noget om …, vi ikke har fået med?") og sætter "afslut" til true. Det samme hvis personen virker
|
|
træt, svarer meget kort flere gange eller siger de vil stoppe.
|
|
- Er "denne_samtale" tom, åbner du med det første forberedte spørgsmål i "tilbage", ordret.
|
|
|
|
Tone og form:
|
|
- Konkret og hverdagsnært: steder, personer, lyde, lugte, hvad man lavede. Højst to sætninger pr. spørgsmål, ét spørgsmål ad gangen.
|
|
- Brug personens egne ord. Foreslå ikke svar, ret ikke personen; strider noget mod baggrunden, så spørg nysgerrigt.
|
|
- Ingen smiger, ingen udråbstegn, ingen "fantastisk", ingen tankestreger. Skriv som man taler. Navnet bruges sjældent.
|
|
- Gentag aldrig et spørgsmål, der allerede er stillet.
|
|
|
|
Du svarer KUN med JSON med disse felter:
|
|
- "spoergsmaal": spørgsmålet som ren tekst, til skærmen.
|
|
- "talt": samme spørgsmål som det skal læses op, med lydtags i klammer, der gør stemmen levende: [warmly] [softly]
|
|
[curious] [gently] [chuckles] [laughs] [pause] [thoughtful] [amused] [sighs] [slowly]. Højst to-tre tags, lagt
|
|
naturligt ind, fx "[warmly] Du nævnte motorcyklen. [chuckles] Hvor stod den egentlig henne?". Ingen andre tags.
|
|
- "kilde": "forberedt", "opfoelgning" eller "afrunding".
|
|
- "afslut": true/false."""
|
|
|
|
|
|
def _historik(pid: str, projekt_slug: str, samtale: str) -> tuple[list[dict], list[dict]]:
|
|
"""(denne samtale, tidligere svar i projektet) som korte dicts til prompten."""
|
|
alle = klip_for(pid, projekt_slug)
|
|
denne = [k for k in alle if k.get("samtale") == samtale]
|
|
tidligere = [k for k in alle if k.get("samtale") != samtale][-30:]
|
|
return denne, tidligere
|
|
|
|
|
|
def _kort(k: dict, n: int = 700) -> dict:
|
|
return {"dato": (k.get("tid") or "")[:10], "spoergsmaal": k.get("spoergsmaal", ""), "kilde": k.get("kilde", ""),
|
|
"svar": (k.get("tekst") or "")[:n]}
|
|
|
|
|
|
def _opfoelg_brugt(denne: list[dict], forberedte: list[str]) -> int:
|
|
"""Hvor mange opfølgende spørgsmål er stillet siden sidste forberedte spørgsmål i denne samtale."""
|
|
n = 0
|
|
for k in reversed(denne):
|
|
if k.get("spoergsmaal") in forberedte or k.get("kilde") == "forberedt":
|
|
break
|
|
n += 1
|
|
return n
|
|
|
|
|
|
async def naeste_spoergsmaal(p: dict, pr: dict, samtale: str, sprunget: list[str] | None = None) -> dict:
|
|
"""Næste spørgsmål: {"spoergsmaal", "talt", "afslut", "kilde"}. Claude styrer samtalen ud fra de forberedte
|
|
spørgsmål; uden nøgle (eller ved fejl) tages de forberedte i rækkefølge."""
|
|
denne, tidligere = _historik(p["pid"], pr["slug"], samtale)
|
|
forberedte = list(pr.get("start") or DAGBOG["start"])
|
|
if pr["slug"] == "dagbog" and forberedte:
|
|
# Én dags dagbog: ét åbningsspørgsmål (skifter fra dag til dag), opfølgninger, afrunding.
|
|
forberedte = [forberedte[int(samtale[:8]) % len(forberedte)]]
|
|
stillede = {k.get("spoergsmaal") for k in denne + tidligere} | set(sprunget or [])
|
|
tilbage = [s for s in forberedte if s not in stillede]
|
|
opfoelg_max = int(pr.get("opfoelg", 2) if pr.get("opfoelg") is not None else 2)
|
|
brugt = _opfoelg_brugt(denne, forberedte)
|
|
|
|
def fallback() -> dict:
|
|
if tilbage:
|
|
return {"spoergsmaal": tilbage[0], "talt": tilbage[0], "afslut": False, "kilde": "forberedt"}
|
|
q = "Er der noget, vi ikke har fået med, som du gerne vil have med?"
|
|
return {"spoergsmaal": q, "talt": "[warmly] " + q, "afslut": True, "kilde": "afrunding"}
|
|
|
|
if not ANTHROPIC_KEY:
|
|
return fallback()
|
|
payload = {
|
|
"person": {"navn": p.get("navn")},
|
|
"projekt": {"titel": pr.get("titel"), "beskrivelse": pr.get("beskrivelse"), "baggrund": pr.get("kontekst", "")[:6000]},
|
|
"forberedte": forberedte, "tilbage": tilbage, "sprunget_over": sprunget or [],
|
|
"opfoelg_max": opfoelg_max, "opfoelg_brugt": brugt,
|
|
"tidligere_samtaler": [_kort(k, 300) for k in tidligere],
|
|
"denne_samtale": [_kort(k) for k in denne],
|
|
"antal_svar_i_dag": len(denne),
|
|
}
|
|
headers = {"x-api-key": ANTHROPIC_KEY, "anthropic-version": "2023-06-01", "content-type": "application/json"}
|
|
if ANTHROPIC_WORKSPACE:
|
|
headers["anthropic-workspace-id"] = ANTHROPIC_WORKSPACE
|
|
body = {"model": MODEL, "max_tokens": 400, "system": SYSTEM,
|
|
"messages": [{"role": "user", "content": "Her er samtalen indtil nu som JSON. Stil næste spørgsmål.\n\n"
|
|
+ json.dumps(payload, ensure_ascii=False)}]}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=60) as c:
|
|
r = await c.post("https://api.anthropic.com/v1/messages", headers=headers, json=body)
|
|
if r.status_code == 200:
|
|
tekst = "".join(b.get("text", "") for b in r.json().get("content", []) if b.get("type") == "text")
|
|
m = re.search(r"\{.*\}", tekst, re.S)
|
|
if m:
|
|
svar = json.loads(m.group(0))
|
|
q = untag(svar.get("spoergsmaal") or "").replace("—", ",").replace(" - ", ", ")
|
|
talt = keep_tags((svar.get("talt") or q).replace("—", ",").replace(" - ", ", "))
|
|
if q:
|
|
if untag(talt) != q and len(untag(talt)) < 5:
|
|
talt = q
|
|
kilde = svar.get("kilde") if svar.get("kilde") in ("forberedt", "opfoelgning", "afrunding") else ("forberedt" if q in forberedte else "opfoelgning")
|
|
return {"spoergsmaal": q[:400], "talt": talt[:600], "afslut": bool(svar.get("afslut")), "kilde": kilde}
|
|
except (httpx.HTTPError, ValueError):
|
|
pass
|
|
return fallback()
|
|
|
|
|
|
def ny_samtale() -> str:
|
|
return time.strftime("%Y%m%d") + "-" + secrets.token_hex(3)
|
|
|
|
|
|
SAMTALE = re.compile(r"^[0-9]{8}(-[a-f0-9]{6})?$")
|
|
|
|
|
|
def dagbog_samtale(dato: str = "") -> str:
|
|
"""Dagbogen føres én dag ad gangen: samtalen ER datoen (YYYYMMDD), i dag medmindre andet vælges."""
|
|
d = re.sub(r"[^0-9]", "", dato or "")
|
|
if len(d) == 8 and "19000101" <= d <= time.strftime("%Y%m%d"):
|
|
return d
|
|
return time.strftime("%Y%m%d")
|
|
|
|
|
|
def eksport(pid: str) -> bytes:
|
|
"""Zip med alle klip (mp3 + json) og en samlet tekstfil, til podcast-produktionen."""
|
|
import io
|
|
import zipfile
|
|
k = klip_for(pid)
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as z:
|
|
linjer = []
|
|
for x in k:
|
|
f = klip_fil(pid, x["id"])
|
|
if f:
|
|
z.write(f, f"{x['id']}.mp3")
|
|
sp = spoergsmaal_fil(pid, x["id"])
|
|
if sp:
|
|
z.write(sp, f"{x['id']}-sp.mp3")
|
|
ren = renset_fil(pid, x["id"])
|
|
if ren:
|
|
z.write(ren, f"{x['id']}-ren.mp3")
|
|
z.writestr(f"{x['id']}.json", json.dumps(x, ensure_ascii=False, indent=1))
|
|
linjer.append(f"## {x['id']} · {x.get('tid', '')[:16]} · {x.get('projekt')}{' · favorit' if x.get('favorit') else ''}\n"
|
|
f"Spørgsmål ({x.get('kilde') or 'ukendt'}{', oplæsning i ' + x['id'] + '-sp.mp3' if sp else ''}): {x.get('spoergsmaal', '')}\n\n"
|
|
f"{x.get('tekst', '')}\n")
|
|
z.writestr("alle_svar.md", "\n".join(linjer))
|
|
return buf.getvalue()
|