Files
mathiasandClaude Opus 5 489b4bb2a1
CI / Lint / Test / Vet (push) Successful in 37s
CI / Build & Import (push) Failing after 5s
CI / Deploy via GitOps (push) Skipped
chore(spike): recover the /bygge toolchain from tmpfs (#28)
analyze_srt.py, build_page.py and transcode.yaml produced the /bygge
prototype. They were sitting in a session scratchpad on tmpfs, one reboot
from gone, while spike issues #29-#31 were written as if they had to be
built from scratch.

Committed as found, defects documented rather than fixed: max_tokens=6000
truncates anything past ~6 minutes of audio, the page title is hardcoded,
and the schema still carries fields no consumer renders.

The four-check validator is the part worth keeping — the coverage-gap and
Swedish-number-grounding checks catch errors that timestamp and citation
checks structurally cannot.

Real transcripts and analyses stay out: this repo is public and the
recordings are a named person discussing a client's project. Only the
synthetic fixture is committed, which exercises all four checks with no
model call.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BeAp5LTscnz2W7Ubt4eJ6m
2026-08-13 11:37:03 +02:00

501 lines
18 KiB
Python
Raw Permalink Blame History

This file contains invisible Unicode characters
This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""Analysera en svensk SRT-transkription av ett byggprojekt.
Steg 1 av 2: producerar tidsstämplad, källbelagd JSON som steg 2
(ffmpeg-bildutdrag + markdown-montering) konsumerar.
Varje påstående måste bära sitt ordagranna citat och sin tidsstämpel.
Det gör hallucination synlig i stället för trolig, och ger samtidigt
bildtiderna gratis.
Körning:
export DMABE_LLMAPI_KEY=... # redan satt på koala
python3 analyze_srt.py IMG_1233.sv.srt > analys.json
"""
from __future__ import annotations
import argparse
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, asdict
GATEWAY = os.environ.get("LITELLM_BASE", "http://koala:30401/v1")
DEFAULT_MODEL = "berget/mistral-medium"
# ---------------------------------------------------------------- SRT-parsning
TS = re.compile(
r"(\d{2}):(\d{2}):(\d{2})[,.](\d{3})\s*-->\s*(\d{2}):(\d{2}):(\d{2})[,.](\d{3})"
)
@dataclass
class Cue:
idx: int
start: float # sekunder
end: float
text: str
@property
def start_hms(self) -> str:
return secs_to_hms(self.start)
def secs_to_hms(s: float) -> str:
h, rem = divmod(int(s), 3600)
m, sec = divmod(rem, 60)
return f"{h:02d}:{m:02d}:{sec:02d}"
def hms_to_secs(v: str) -> float:
"""Tål '00:01:23', '01:23', '83' och '00:01:23,450'."""
v = v.strip().replace(",", ".")
parts = v.split(":")
try:
nums = [float(p) for p in parts]
except ValueError as e:
raise ValueError(f"ogiltig tidsstämpel: {v!r}") from e
if len(nums) == 3:
return nums[0] * 3600 + nums[1] * 60 + nums[2]
if len(nums) == 2:
return nums[0] * 60 + nums[1]
if len(nums) == 1:
return nums[0]
raise ValueError(f"ogiltig tidsstämpel: {v!r}")
def parse_srt(raw: str) -> list[Cue]:
"""Blockbaserad parsning. Tål saknade index och CRLF."""
raw = raw.replace("\r\n", "\n").replace("", "")
cues: list[Cue] = []
for block in re.split(r"\n\s*\n", raw.strip()):
lines = [ln for ln in block.split("\n") if ln.strip()]
if not lines:
continue
m = None
text_from = 0
for i, ln in enumerate(lines[:2]):
m = TS.search(ln)
if m:
text_from = i + 1
break
if not m:
continue
g = [int(x) for x in m.groups()]
start = g[0] * 3600 + g[1] * 60 + g[2] + g[3] / 1000
end = g[4] * 3600 + g[5] * 60 + g[6] + g[7] / 1000
text = " ".join(lines[text_from:]).strip()
if text:
cues.append(Cue(len(cues) + 1, start, end, text))
return cues
def as_numbered_transcript(cues: list[Cue]) -> str:
"""Tidsstämplad text som modellen kan citera exakt ur."""
return "\n".join(f"[{c.start_hms}] {c.text}" for c in cues)
# ------------------------------------------------------------------ LLM-anrop
SYSTEM = """Du är byggteknisk analytiker. Du analyserar en transkription från ett \
platsbesök där Jonas beskriver ett PLANERAT byggprojekt för minst en annan person.
Detta är ett SAMTAL, inte en monolog. Korta inpass som "Hela vägen?", "Vilken höjd?" \
eller "Måste det bli det?" är motpartens frågor - de driver samtalet men är inte \
Jonas påståenden. Attribuera inte en fråga som ett beslut.
Det mesta som sägs är FÖRSLAG under diskussion, inte färdiga beslut. Signalord som \
"vi tänker", "alternativt", "kanske", "förmodligen", "eventuellt", "om möjligt", \
"troligast" och "beror på" markerar något ÖPPET. Blanda aldrig ihop ett övervägt \
alternativ med ett fattat beslut - den skillnaden är hela poängen med analysen.
ABSOLUTA KRAV:
- Rapportera ENDAST det som faktiskt sägs. Dra inga egna slutsatser, lägg inte till \
byggmoment, material eller mått som inte nämns explicit.
- Varje påstående MÅSTE bära ett ordagrant citat ur transkriptionen samt dess tidsstämpel.
- Mått revideras under samtalets gång ("nio meter" -> "9,40"). Återge alla varianter \
som nämns, med citat, i stället för att välja en.
- Är något oklart, ohörbart eller avbrutet: skriv det i fältet "oklart" i stället för \
att gissa.
- Svara ENDAST med giltig JSON. Ingen markdown, inga kodstaket, ingen förklarande text."""
USER_TMPL = """Dela in transkriptionen i tematiska avsnitt (typiskt 4-8 stycken) som \
följer hur Jonas rör sig genom projektet.
TACKNINGSKRAV: avsnitten maste tillsammans tacka HELA klippet, fran 00:00:00 till \
{slut}. Forsta avsnittet borjar 00:00:00, sista avsnittet slutar {slut}, och varje \
avsnitts "start" ar lika med foregaende avsnitts "slut". Lamna inga luckor - ocksa \
korta avslutande kommentarer (tidplan, kostnader, nasta steg) ska tillhora ett avsnitt.
Svara med exakt denna JSON-struktur:
{{
"projekt": {{
"sammanfattning": "2-3 meningar om vad projektet är",
"typ": "t.ex. nybyggnad villa / tillbyggnad / renovering - eller null om det inte framgår",
"skede": "vilket byggskede som visas - eller null"
}},
"avsnitt": [
{{
"rubrik": "kort beskrivande rubrik",
"start": "HH:MM:SS",
"slut": "HH:MM:SS",
"sammanfattning": "vad Jonas beskriver i detta avsnitt, 1-3 meningar",
"nyckelpunkter": [
{{"pastaende": "vad som sags", "citat": "ordagrant citat", "tid": "HH:MM:SS"}}
],
"bildtid": "HH:MM:SS - ogonblicket dar det Jonas beskriver syns tydligast",
"bildmotiv": "vad som bor synas i bilden vid bildtid"
}}
],
"beslut": [{{"vad": "nagot som uttalas som bestamt", "citat": "ordagrant citat", "tid": "HH:MM:SS"}}],
"alternativ": [{{
"fraga": "vad som vags mot varandra",
"optioner": ["alternativ A", "alternativ B"],
"lutar_at": "det alternativ som forespraakas - eller null om oavgjort",
"citat": "ordagrant citat", "tid": "HH:MM:SS"
}}],
"oppna_fragor": [{{
"fraga": "det som inte ar avgjort",
"beror_pa": "vad avgorandet hanger pa - eller null",
"citat": "ordagrant citat", "tid": "HH:MM:SS"
}}],
"matt": [{{"vad": "matt/dimension", "citat": "ordagrant citat", "tid": "HH:MM:SS"}}],
"material": [{{"vad": "material/produkt", "citat": "ordagrant citat", "tid": "HH:MM:SS"}}],
"oklart": ["saker som ar ohorbara, avbrutna eller tvetydiga i transkriptionen"]
}}
Valj "bildtid" omsorgsfullt. I detta samtal pekar Jonas standigt pa saker med ord som \
"har", "hit", "den typ sa" - orden ar obegripliga utan bilden, sa bildtiden ska ligga \
dar foremalet han syftar pa syns, inte dar han byter amne. Tomma listor ar helt i sin \
ordning om inget relevant namns.
TRANSKRIPTION:
{transcript}"""
def call_llm(
model: str, transcript: str, slut: str, timeout: int = 300, retries: int = 2
) -> str:
key = os.environ.get("DMABE_LLMAPI_KEY") or os.environ.get("LITELLM_API_KEY")
if not key:
sys.exit("DMABE_LLMAPI_KEY (eller LITELLM_API_KEY) saknas i miljon.")
body = json.dumps(
{
"model": model,
"temperature": 0,
"max_tokens": 6000,
"messages": [
{"role": "system", "content": SYSTEM},
{"role": "user", "content": USER_TMPL.format(transcript=transcript, slut=slut)},
],
}
).encode()
# Gatewayen begransar burst (limit_req burst=5) och berget-anropet kan
# spika. TimeoutError arvs INTE av URLError, sa den maste fangas separat -
# annars blir felet en rasptrace efter flera minuters vantan.
last = ""
for attempt in range(retries + 1):
req = urllib.request.Request(
f"{GATEWAY}/chat/completions",
data=body,
headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
)
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
payload = json.load(r)
break
except urllib.error.HTTPError as e:
detail = e.read()[:400].decode(errors="replace")
if e.code == 429 and attempt < retries:
last = f"HTTP 429 (rate limit)"
elif e.code >= 500 and attempt < retries:
last = f"HTTP {e.code}"
else:
sys.exit(f"gateway HTTP {e.code}: {detail}")
except (TimeoutError, urllib.error.URLError, OSError) as e:
reason = getattr(e, "reason", e)
if attempt < retries:
last = f"{type(e).__name__}: {reason}"
else:
sys.exit(
f"nadde inte gateway {GATEWAY} efter {retries + 1} forsok "
f"({type(e).__name__}: {reason}). Kontrollera att koala:30401 svarar."
)
backoff = 3 * (attempt + 1)
print(f"[!] {last} - forsok {attempt + 2}/{retries + 1} om {backoff}s", file=sys.stderr)
time.sleep(backoff)
else:
sys.exit(f"gav upp efter {retries + 1} forsok: {last}")
choice = payload["choices"][0]
content = (choice.get("message") or {}).get("content") or ""
if not content.strip():
# Kant lage for thinking-modeller: resonemanget ater hela budgeten.
sys.exit(
f"tomt svar (finish_reason={choice.get('finish_reason')}). "
"Valj en icke-resonerande modell for strukturerad utdata."
)
return content
def extract_json(s: str) -> dict:
s = s.strip()
if s.startswith("```"):
s = re.sub(r"^```(?:json)?\s*", "", s)
s = re.sub(r"\s*```$", "", s)
try:
return json.loads(s)
except json.JSONDecodeError:
# Sista utvag: forsta {...sista }
i, j = s.find("{"), s.rfind("}")
if i >= 0 and j > i:
return json.loads(s[i : j + 1])
raise
# -------------------------------------------------------------- Verifiering
def norm(s: str) -> str:
return re.sub(r"[^\wåäöÅÄÖ]+", " ", (s or "").lower()).strip()
# --- svenska talord -> siffror -------------------------------------------
# Byggmatt sags i klartext ("fyrtiofem ganger tvahundratjugo") men refereras
# i siffror. Utan denna oversattning kan ett felaktigt matt bara ett giltigt
# ordagrant citat och passera oupptackt - den farligaste feltypen i ett
# byggdokument.
_SW_UNITS = {
"noll": 0, "en": 1, "ett": 1, "två": 2, "tva": 2, "tre": 3, "fyra": 4,
"fem": 5, "sex": 6, "sju": 7, "åtta": 8, "atta": 8, "nio": 9, "tio": 10,
"elva": 11, "tolv": 12, "tretton": 13, "fjorton": 14, "femton": 15,
"sexton": 16, "sjutton": 17, "arton": 18, "nitton": 19,
}
_SW_TENS = {
"tjugo": 20, "trettio": 30, "fyrtio": 40, "femtio": 50, "sextio": 60,
"sjuttio": 70, "åttio": 80, "attio": 80, "nittio": 90,
}
_SW_ATOMS = {**_SW_UNITS, **_SW_TENS, "hundra": 100, "tusen": 1000}
# Langsta forst sa att "sexton" vinner over "sex", "attio" over "atta".
_SW_ORDER = sorted(_SW_ATOMS, key=len, reverse=True)
def _sw_word_to_int(word: str) -> int | None:
"""'tvåhundratjugo' -> 220. None om ordet inte ar ett rent talord."""
w = word.lower()
toks: list[str] = []
while w:
for atom in _SW_ORDER:
if w.startswith(atom):
toks.append(atom)
w = w[len(atom) :]
break
else:
return None
if not toks:
return None
total = current = 0
for t in toks:
v = _SW_ATOMS[t]
if t == "hundra":
current = (current or 1) * 100
elif t == "tusen":
total += (current or 1) * 1000
current = 0
else:
current += v
return total + current
def numbers_in(text: str) -> set[float]:
"""Alla tal i texten, som siffror och som svenska talord.
Svenska anvander decimalkomma ("9,40" = 9.4). Naiv \\d+-matchning skulle
lasa det som tva separata heltal 9 och 40 - och just har revideras matten
live i samtalet ("nio meter" -> "9,40"), sa den skillnaden ar betydelsebarande.
"""
found: set[float] = set()
for m in re.finditer(r"\d+(?:[.,]\d+)?", text or ""):
found.add(float(m.group().replace(",", ".")))
for w in re.findall(r"[a-zåäöA-ZÅÄÖ]+", text or ""):
# "en"/"ett" ar obestamd artikel vida oftare an raknetalet 1
# ("ett sedumtak"), sa fristaende traffar ger falsklarm. Sammansatta
# former ("etthundra") fangas anda av den giriga tokeniseringen.
if len(w) < 3 or w.lower() in {"en", "ett"}:
continue
v = _sw_word_to_int(w)
if v is not None and v > 0:
found.add(float(v))
return found
def fmt_num(v: float) -> str:
return str(int(v)) if v == int(v) else f"{v:g}"
def unsupported(claimed: set[float], supported: set[float]) -> list[float]:
"""Tal i pastaendet som inte styrks av citatet (med liten tolerans)."""
return sorted(
n for n in claimed if not any(abs(n - s) < 0.01 for s in supported)
)
def verify(doc: dict, cues: list[Cue], duration: float) -> list[str]:
"""Varje citat ska ga att aterfinna; varje tid ska ligga inom klippet.
Detta ar sjalva poangen med strukturen - ett pastaende utan verifierbar
kalla ar en hallucination, och den ska synas har och inte i dokumentet.
"""
warnings: list[str] = []
haystack = norm(" ".join(c.text for c in cues))
def check_quote(where: str, quote: str) -> None:
n = norm(quote)
if not n:
warnings.append(f"{where}: tomt citat")
return
if n in haystack:
return
# Delvis traff: minst 60 % av orden i foljd nagonstans
words = n.split()
if len(words) >= 4:
for size in (len(words), max(4, int(len(words) * 0.6))):
for i in range(0, len(words) - size + 1):
if " ".join(words[i : i + size]) in haystack:
warnings.append(f"{where}: citat matchar bara delvis: {quote!r}")
return
warnings.append(f"{where}: CITAT SAKNAS I TRANSKRIPTION: {quote!r}")
def check_time(where: str, t: str) -> None:
try:
v = hms_to_secs(t)
except ValueError:
warnings.append(f"{where}: ogiltig tid {t!r}")
return
if not (0 <= v <= duration + 1):
warnings.append(f"{where}: tid {t} utanfor klippet (0-{secs_to_hms(duration)})")
for i, sec in enumerate(doc.get("avsnitt") or [], 1):
tag = f"avsnitt {i} ({sec.get('rubrik','?')})"
for f in ("start", "slut", "bildtid"):
if sec.get(f):
check_time(f"{tag}.{f}", sec[f])
for j, kp in enumerate(sec.get("nyckelpunkter") or [], 1):
check_quote(f"{tag}.nyckelpunkt {j}", kp.get("citat", ""))
if kp.get("tid"):
check_time(f"{tag}.nyckelpunkt {j}.tid", kp["tid"])
def check_numbers(where: str, claim: str, quote: str) -> None:
"""Varje tal i pastaendet maste finnas i dess eget citat."""
claimed = numbers_in(claim)
if not claimed:
return
supported = numbers_in(quote)
for n in unsupported(claimed, supported):
stod = ", ".join(fmt_num(s) for s in sorted(supported)) or "inga tal"
warnings.append(
f"{where}: TALET {fmt_num(n)} FINNS INTE I CITATET "
f"(citat stoder {stod}): {claim!r}"
)
for i, sec in enumerate(doc.get("avsnitt") or [], 1):
tag = f"avsnitt {i} ({sec.get('rubrik','?')})"
for j, kp in enumerate(sec.get("nyckelpunkter") or [], 1):
check_numbers(f"{tag}.nyckelpunkt {j}", kp.get("pastaende", ""), kp.get("citat", ""))
for key in ("material", "matt", "beslut"):
for j, it in enumerate(doc.get(key) or [], 1):
check_quote(f"{key}[{j}]", it.get("citat", ""))
if it.get("tid"):
check_time(f"{key}[{j}].tid", it["tid"])
check_numbers(f"{key}[{j}]", it.get("vad", ""), it.get("citat", ""))
# Tackningskontroll: avsnitten ska tacka hela klippet utan hal, annars
# tappas material tyst (och inget bildutdrag gors for den luckan).
spans: list[tuple[float, float]] = []
for sec in doc.get("avsnitt") or []:
try:
spans.append((hms_to_secs(sec["start"]), hms_to_secs(sec["slut"])))
except (KeyError, ValueError):
continue
spans.sort()
cursor = 0.0
for s, e in spans:
if s - cursor > 2.0:
gap_text = " ".join(
c.text for c in cues if c.start >= cursor - 0.5 and c.end <= s + 0.5
)
warnings.append(
f"TACKNINGSLUCKA {secs_to_hms(cursor)}-{secs_to_hms(s)} "
f"({s - cursor:.0f}s utan avsnitt): {gap_text[:90]!r}"
)
cursor = max(cursor, e)
if duration - cursor > 2.0:
warnings.append(
f"TACKNINGSLUCKA {secs_to_hms(cursor)}-{secs_to_hms(duration)} "
f"({duration - cursor:.0f}s i slutet utan avsnitt)"
)
return warnings
# ------------------------------------------------------------------- main
def main() -> None:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("srt")
ap.add_argument("--model", default=DEFAULT_MODEL)
ap.add_argument("--raw", action="store_true", help="skriv aven modellens rasvar till stderr")
args = ap.parse_args()
with open(args.srt, encoding="utf-8") as fh:
cues = parse_srt(fh.read())
if not cues:
sys.exit(f"inga cues hittades i {args.srt} - fel format?")
duration = cues[-1].end
print(
f"[i] {len(cues)} cues, langd {secs_to_hms(duration)}, modell {args.model}",
file=sys.stderr,
)
content = call_llm(args.model, as_numbered_transcript(cues), secs_to_hms(duration))
if args.raw:
print(content, file=sys.stderr)
try:
doc = extract_json(content)
except json.JSONDecodeError as e:
sys.exit(f"modellen returnerade ogiltig JSON: {e}\n---\n{content[:800]}")
warnings = verify(doc, cues, duration)
doc["_meta"] = {
"modell": args.model,
"kallfil": os.path.basename(args.srt),
"antal_cues": len(cues),
"langd": secs_to_hms(duration),
"varningar": warnings,
}
print(json.dumps(doc, ensure_ascii=False, indent=2))
if warnings:
print(f"\n[!] {len(warnings)} verifieringsvarningar:", file=sys.stderr)
for w in warnings:
print(f" - {w}", file=sys.stderr)
else:
print("[ok] alla citat aterfunna, alla tider inom klippet", file=sys.stderr)
if __name__ == "__main__":
main()