1e8a057615
Neuer Agent entscheidet aus Quellen und Leitfrage, welche Themen ins Buch kommen und wie tief (herleiten, erklären, nennen, weggelassen), mit Begründung in buch/STOFF.md. Die Gliederung baut nur daraus, das Gewicht folgt der Tiefe. Autoren und Prüfer halten sich an STOFF.md. Nachgetragene Stufen gelten als fertig, wenn eine spätere schon fertig ist. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
281 lines
11 KiB
Python
281 lines
11 KiB
Python
"""Zustand je Themenversion. Ablage: storage/<Thema>/thema.json und storage/<Thema>/v<N>/{state.json, deutung.json, quellen/, buch/, …}.
|
||
Intern heißt eine Version „<Thema>/v<N>“ (Schlüssel); alle Funktionen mit `key` erwarten diesen Schlüssel. Alles im Speicher, bei Änderung auf Platte."""
|
||
import json
|
||
import re
|
||
import shutil
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
from . import config
|
||
|
||
_cache: dict[str, dict] = {}
|
||
|
||
|
||
def jetzt() -> str:
|
||
return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
|
||
|
||
|
||
def name_ok(name: str) -> bool:
|
||
return bool(name) and name.strip() == name and "/" not in name and ".." not in name and len(name) < 120
|
||
|
||
|
||
def verzeichnis(key: str) -> Path:
|
||
return config.STORAGE / key
|
||
|
||
|
||
def thema_von(key: str) -> str:
|
||
"""„Godot 4/v2“ → „Godot 4“; ein reiner Themenname bleibt unverändert."""
|
||
return key.rsplit("/", 1)[0] if "/" in key else key
|
||
|
||
|
||
def versionen(name: str) -> list[int]:
|
||
d = config.STORAGE / name
|
||
return sorted(int(x.name[1:]) for x in d.glob("v[0-9]*") if x.is_dir() and x.name[1:].isdigit())
|
||
|
||
|
||
def schluessel(name: str, v: int | None = None) -> str | None:
|
||
"""Schlüssel einer Version; ohne v die neueste. None, wenn Thema oder Version fehlen."""
|
||
vs = versionen(name)
|
||
if not vs or (v is not None and v not in vs):
|
||
return None
|
||
return f"{name}/v{vs[-1] if v is None else v}"
|
||
|
||
|
||
def slug_frei(name: str) -> str:
|
||
"""Eindeutiger Slug für die Adresse: „Godot 4“ → „godot-4“; bei Kollision „godot-4-2“."""
|
||
basis = slug(name)
|
||
vergeben = {t.get("slug") for t in themen()}
|
||
k, i = basis, 2
|
||
while k in vergeben:
|
||
k, i = f"{basis}-{i}", i + 1
|
||
return k
|
||
|
||
|
||
def name_von_slug(sl: str) -> str | None:
|
||
return next((t["name"] for t in themen() if t.get("slug") == sl), None)
|
||
|
||
|
||
def version_anlegen(name: str, hinweise: str | None = None, stopp_gliederung: bool = False) -> str:
|
||
"""Neue Version; ohne Angabe übernimmt sie die Hinweise der neuesten Version (bzw. des Themas).
|
||
stopp_gliederung: der Lauf hält nach der Gliederung an (günstiger Testlauf), „Fortsetzen“ schreibt weiter."""
|
||
vs = versionen(name)
|
||
alt = hinweise_von(f"{name}/v{vs[-1]}") if vs else (thema(name) or {}).get("hinweise", "")
|
||
key = f"{name}/v{(vs or [0])[-1] + 1}"
|
||
v = verzeichnis(key)
|
||
for u in ("quellen", "buch", "berichte", "arbeit"):
|
||
(v / u).mkdir(parents=True)
|
||
_schreiben(v / "state.json", {**leerer_zustand(), "hinweise": (alt if hinweise is None else hinweise).strip(),
|
||
"stopp_gliederung": stopp_gliederung})
|
||
return key
|
||
|
||
|
||
def hinweise_von(key: str) -> str:
|
||
"""Hinweise gelten je Version; alte Versionen ohne eigene nehmen die des Themas."""
|
||
st = zustand(key)
|
||
return st["hinweise"] if "hinweise" in st else (thema(key) or {}).get("hinweise", "")
|
||
|
||
|
||
def spezialisierung_von(key: str) -> str:
|
||
"""Die Spezialisierung gehört zum Thema: eine andere Spezialisierung ist ein anderes Buch."""
|
||
return (thema(key) or {}).get("spezialisierung", "")
|
||
|
||
|
||
def bezeichnung(name: str, spezialisierung: str) -> str:
|
||
"""Ordner- und Anzeigename: „Game Design – digitale Sammelkartenspiele“."""
|
||
return f"{name} – {spezialisierung}" if spezialisierung else name
|
||
|
||
|
||
def version_loeschen(key: str) -> None:
|
||
_cache.pop(key, None)
|
||
shutil.rmtree(verzeichnis(key), ignore_errors=True)
|
||
name = thema_von(key)
|
||
if not versionen(name):
|
||
shutil.rmtree(config.STORAGE / name, ignore_errors=True)
|
||
|
||
|
||
def migrieren() -> None:
|
||
"""Alte Ablage (alles direkt unter storage/<Thema>/) einmalig nach v1/ verschieben; Slugs nachtragen."""
|
||
for d in config.STORAGE.glob("*/"):
|
||
t = _lesen(d / "thema.json", None)
|
||
if t and not t.get("slug"):
|
||
t["slug"] = slug_frei(t["name"])
|
||
_schreiben(d / "thema.json", t)
|
||
if not (d / "thema.json").exists() or not (d / "state.json").exists() or (d / "v1").exists():
|
||
continue
|
||
v1 = d / "v1"
|
||
v1.mkdir()
|
||
for kind in list(d.iterdir()):
|
||
if kind.name not in ("thema.json", "v1"):
|
||
shutil.move(str(kind), str(v1 / kind.name))
|
||
|
||
|
||
def _lesen(pfad: Path, standard):
|
||
try:
|
||
return json.loads(pfad.read_text())
|
||
except (OSError, ValueError):
|
||
return standard
|
||
|
||
|
||
def _schreiben(pfad: Path, daten) -> None:
|
||
pfad.parent.mkdir(parents=True, exist_ok=True)
|
||
tmp = pfad.with_suffix(pfad.suffix + ".tmp")
|
||
tmp.write_text(json.dumps(daten, ensure_ascii=False, indent=1))
|
||
tmp.replace(pfad)
|
||
|
||
|
||
def leerer_zustand() -> dict:
|
||
return {"status": "neu", "grund": "", "stufe": "", "deutung": None, "angelegt": jetzt(),
|
||
"stufen": [{"id": s, "titel": t, "status": "offen", "fertig": 0, "gesamt": 0,
|
||
"kosten_usd": 0.0, "tokens": 0, "gestartet": "", "beendet": ""} for s, t in config.STUFEN],
|
||
"kosten_usd": 0.0, "tokens": 0, "agenten": [], "ereignisse": []}
|
||
|
||
|
||
def themen() -> list[dict]:
|
||
aus = []
|
||
for d in sorted(config.STORAGE.glob("*/thema.json")):
|
||
t = _lesen(d, None)
|
||
if t:
|
||
aus.append(t)
|
||
return aus
|
||
|
||
|
||
def anlegen(name: str, hinweise: str, spezialisierung: str = "") -> dict:
|
||
"""name ist das Hauptthema („Game Design“), die Spezialisierung grenzt es ein; beides zusammen bildet die Bezeichnung."""
|
||
spezialisierung = spezialisierung.strip()
|
||
bez = bezeichnung(name, spezialisierung)
|
||
d = config.STORAGE / bez
|
||
if d.exists():
|
||
raise ValueError(f"„{bez}“ gibt es schon – anderes Thema oder andere Spezialisierung wählen")
|
||
d.mkdir(parents=True)
|
||
t = {"name": bez, "thema": name, "spezialisierung": spezialisierung, "slug": slug_frei(bez), "titel": bez,
|
||
"hinweise": hinweise.strip(), "angelegt": jetzt()}
|
||
_schreiben(d / "thema.json", t)
|
||
version_anlegen(bez)
|
||
return t
|
||
|
||
|
||
def loeschen(name: str) -> None:
|
||
for k in [k for k in _cache if thema_von(k) == name]:
|
||
_cache.pop(k, None)
|
||
shutil.rmtree(config.STORAGE / name, ignore_errors=True)
|
||
|
||
|
||
def zuruecksetzen(key: str) -> None:
|
||
v = verzeichnis(key)
|
||
h, stopp = hinweise_von(key), bool(zustand(key).get("stopp_gliederung"))
|
||
_cache.pop(key, None)
|
||
for u in ("quellen", "buch", "berichte", "arbeit"):
|
||
shutil.rmtree(v / u, ignore_errors=True)
|
||
(v / u).mkdir()
|
||
for f in ("deutung.json", "log.jsonl", "state.json", "position.json"):
|
||
(v / f).unlink(missing_ok=True)
|
||
_schreiben(v / "state.json", {**leerer_zustand(), "hinweise": h, "stopp_gliederung": stopp})
|
||
|
||
|
||
def thema(key: str) -> dict | None:
|
||
"""thema.json des Themas – nimmt Schlüssel oder Themenname."""
|
||
return _lesen(config.STORAGE / thema_von(key) / "thema.json", None)
|
||
|
||
|
||
def zustand(name: str) -> dict:
|
||
if name not in _cache:
|
||
st = _lesen(verzeichnis(name) / "state.json", None) or leerer_zustand()
|
||
# Ein Lauf, der beim Serverstart noch als „laeuft“ gespeichert ist, ist tot.
|
||
if st.get("status") == "laeuft":
|
||
st["status"] = "pausiert"
|
||
st["grund"] = "Server wurde beendet – mit „Fortsetzen“ weitermachen"
|
||
for s in st["stufen"]:
|
||
if s["status"] == "laeuft":
|
||
s["status"] = "offen"
|
||
for a in st["agenten"]:
|
||
if a["status"] == "laeuft":
|
||
a["status"] = "abgebrochen"
|
||
# Später hinzugekommene Stufen (z. B. Lektorat, Stoff) in alte Zustände einfügen, in der Reihenfolge der Konfiguration;
|
||
# eine nachgetragene Stufe gilt als fertig, wenn eine spätere Stufe schon fertig ist (der Lauf ist darüber hinaus).
|
||
vorhanden = {x["id"] for x in st["stufen"]}
|
||
if vorhanden != {sid for sid, _ in config.STUFEN}:
|
||
alt = {x["id"]: x for x in st["stufen"]}
|
||
reihe = [sid for sid, _ in config.STUFEN]
|
||
fertig_bis = max((reihe.index(x["id"]) for x in st["stufen"] if x["status"] == "fertig" and x["id"] in reihe), default=-1)
|
||
st["stufen"] = [alt.get(sid) or {"id": sid, "titel": t, "status": "fertig" if i <= fertig_bis else "offen", "fertig": 0, "gesamt": 0,
|
||
"kosten_usd": 0.0, "tokens": 0, "gestartet": "", "beendet": ""} for i, (sid, t) in enumerate(config.STUFEN)]
|
||
_cache[name] = st
|
||
return _cache[name]
|
||
|
||
|
||
def speichern(name: str) -> None:
|
||
if name in _cache:
|
||
_schreiben(verzeichnis(name) / "state.json", _cache[name])
|
||
|
||
|
||
def deutung(name: str) -> dict | None:
|
||
return _lesen(verzeichnis(name) / "deutung.json", None)
|
||
|
||
|
||
def stufe(name: str, sid: str) -> dict:
|
||
return next(s for s in zustand(name)["stufen"] if s["id"] == sid)
|
||
|
||
|
||
def ereignis(name: str, agent: str, art: str, text: str) -> None:
|
||
st = zustand(name)
|
||
e = {"zeit": jetzt(), "agent": agent, "art": art, "text": text[:300]}
|
||
st["ereignisse"].append(e)
|
||
del st["ereignisse"][:-200]
|
||
try:
|
||
with open(verzeichnis(name) / "log.jsonl", "a") as f:
|
||
f.write(json.dumps(e, ensure_ascii=False) + "\n")
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def agent_eintrag(name: str, agent_id: str, stufe_id: str, titel: str) -> dict:
|
||
st = zustand(name)
|
||
st["agenten"] = [a for a in st["agenten"] if a["id"] != agent_id]
|
||
a = {"id": agent_id, "stufe": stufe_id, "name": titel, "status": "laeuft", "kosten_usd": 0.0, "tokens": 0,
|
||
"werkzeuge": 0, "gestartet": jetzt(), "beendet": "", "bericht": ""}
|
||
st["agenten"].append(a)
|
||
speichern(name)
|
||
return a
|
||
|
||
|
||
def agent_fertig(name: str, a: dict, status: str, bericht: str, kosten: float, tokens: int) -> None:
|
||
st = zustand(name)
|
||
a.update(status=status, bericht=bericht, kosten_usd=round(kosten, 4), tokens=tokens, beendet=jetzt())
|
||
s = stufe(name, a["stufe"])
|
||
s["kosten_usd"] = round(s["kosten_usd"] + kosten, 4)
|
||
s["tokens"] += tokens
|
||
st["kosten_usd"] = round(st["kosten_usd"] + kosten, 4)
|
||
st["tokens"] += tokens
|
||
speichern(name)
|
||
|
||
|
||
def bericht_schreiben(name: str, agent_id: str, text: str) -> None:
|
||
p = verzeichnis(name) / "berichte" / f"{agent_id}.md"
|
||
p.parent.mkdir(exist_ok=True)
|
||
p.write_text(text.strip() + "\n")
|
||
|
||
|
||
def bericht_lesen(name: str, agent_id: str) -> str:
|
||
p = verzeichnis(name) / "berichte" / f"{agent_id}.md"
|
||
return p.read_text() if p.exists() else ""
|
||
|
||
|
||
def slug(text: str) -> str:
|
||
t = text.lower()
|
||
for a, b in (("ä", "ae"), ("ö", "oe"), ("ü", "ue"), ("ß", "ss")):
|
||
t = t.replace(a, b)
|
||
t = re.sub(r"[^a-z0-9]+", "-", t).strip("-")
|
||
return t[:60] or "thema"
|
||
|
||
|
||
def position(name: str) -> dict:
|
||
"""Leseposition (Abschnitt der Buchansicht, Bildschirm des Lesemodus) – geräteübergreifend auf dem Server."""
|
||
return _lesen(verzeichnis(name) / "position.json", {})
|
||
|
||
|
||
def position_setzen(name: str, werte: dict) -> dict:
|
||
p = position(name)
|
||
p.update(werte)
|
||
_schreiben(verzeichnis(name) / "position.json", p)
|
||
return p
|