Files
weircon-front/erindringer.py
T
root 4005860364
CD / deploy (push) Successful in 15s
CI / check (push) Canceled after 17s
feat(erindringer): familien fortæller minder og dagbog til en AI-interviewer
Nyt modul erindringer.py: emner (projekter) med baggrund og startspørgsmål,
personer (indloggede + gæstelinks), Claude stiller opfølgende spørgsmål,
ElevenLabs læser op (cachet) og transskriberer, ffmpeg laver mp3, admin med
klip, favoritter og zip-eksport. Forsidekort og admin-modul. Røgtest i tests/.

Claude-Session: https://claude.ai/code/session_01JmhJG6JnhXfzNiKv8f2vHu
2026-10-06 22:04:20 +02:00

478 lines
18 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]) -> bool:
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()], "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) -> list[dict]:
"""Klip for en person, ældste først."""
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
ud.append(k)
return sorted(ud, key=lambda k: 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) -> dict:
"""Gem et svar: lyd konverteres til mp3 og transskriberes; ren tekst gemmes som det er.
Returnerer klip-metadata. Kaster ValueError med besked til brugeren."""
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"
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.")
if tekst is None or not tekst.strip():
tekst = transskriber(mp3)
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],
"tekst": (tekst or "").strip(), "varighed": varighed, "favorit": False, "skrevet": not lyd}
_skriv(d / f"{kid}.json", meta)
return meta
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)
(d / f"{kid}.json").unlink(missing_ok=True)
(d / f"{kid}.mp3").unlink(missing_ok=True)
return True
# ---------------- 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 ----------------
SYSTEM = """Du er en varm, nysgerrig interviewer for en dansk familie. Du hjælper et familiemedlem med at
fortælle sine minder (eller føre dagbog) ved at stille ÉT kort spørgsmål ad gangen på dansk.
Regler:
- Spørg konkret og hverdagsnært: steder, personer, lyde, lugte, hvad man lavede, hvad man følte. Et spørgsmål er på højst to sætninger.
- Følg op på det, personen lige har fortalt (navne, steder, ting de kun nævnte kort), før du skifter emne. Brug personens egne ord.
- Led ikke vidnet: foreslå ikke svar, og ret ikke personen. Hvis noget i det sagte strider mod baggrunden, så spørg nysgerrigt i stedet for at påpege det.
- Hold en rolig, respektfuld tone, ingen smiger, ingen udråbstegn, ingen "fantastisk". Skriv som man taler. Brug aldrig tankestreger.
- Hvis personen virker træt, skriver kort eller siger de vil stoppe, så stil et let, afrundende spørgsmål og sæt "afslut" til true.
- Gentag ikke et spørgsmål, der allerede er stillet i samtalen. Vend tilbage til startspørgsmålene, hvis samtalen går i stå.
- Svar KUN med JSON: {"spoergsmaal": "...", "afslut": 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", ""),
"svar": (k.get("tekst") or "")[:n]}
async def naeste_spoergsmaal(p: dict, pr: dict, samtale: str) -> tuple[str, bool]:
"""Næste spørgsmål til personen i projektet. Falder tilbage til startspørgsmålene uden nøgle/ved fejl."""
denne, tidligere = _historik(p["pid"], pr["slug"], samtale)
stillede = {k.get("spoergsmaal") for k in denne + tidligere}
start = [s for s in (pr.get("start") or DAGBOG["start"]) if s not in stillede] or list(pr.get("start") or DAGBOG["start"])
if not denne:
return start[0], False # samtalen åbner altid med et af projektets egne spørgsmål
if not ANTHROPIC_KEY:
return start[len(denne) % len(start)], False
payload = {
"person": {"navn": p.get("navn")},
"projekt": {"titel": pr.get("titel"), "beskrivelse": pr.get("beskrivelse"), "baggrund": pr.get("kontekst", "")[:6000],
"startspoergsmaal": pr.get("start") or []},
"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": 300, "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 = (svar.get("spoergsmaal") or "").strip().replace("—", ",").replace(" - ", ", ")
if q:
return q[:400], bool(svar.get("afslut"))
except (httpx.HTTPError, ValueError):
pass
return start[len(denne) % len(start)], False
def ny_samtale() -> str:
return time.strftime("%Y%m%d") + "-" + secrets.token_hex(3)
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")
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('spoergsmaal', '')}\n\n{x.get('tekst', '')}\n")
z.writestr("alle_svar.md", "\n".join(linjer))
return buf.getvalue()