"""Ein Thema als JSON exportieren/importieren (make server-push/-pull TOPIC=…). Ein Thema lebt in ~15 verzahnten Tabellen mit Integer-PKs plus Dateien. Export schreibt alle Zeilen (Snapshot-Pfade relativ); Import LÖSCHT das Thema auf der Zielseite und fügt mit neuen IDs ein — alle Verweise werden über alte→neue Maps nachgezogen. Die Dateien (storage/korpus/, topics/) trägt das Makefile per rsync NACH dem Import (topic_loeschen räumt den Korpus-Ordner).""" import json import re import sys from pathlib import Path import config import db import pipeline from config import PROJECT_ROOT def export(topic: str) -> dict: if not db.one("SELECT name FROM topics WHERE name=?", (topic,)): sys.exit(f"unbekanntes Topic: {topic}") q = db.query d = { "topic": topic, "topics": q("SELECT * FROM topics WHERE name=?", (topic,)), "quellen": q("SELECT * FROM quellen WHERE topic=?", (topic,)), "soll": q("SELECT * FROM soll WHERE topic=?", (topic,)), "themen": q("SELECT * FROM themen WHERE topic=?", (topic,)), "lernziele": q("SELECT * FROM lernziele WHERE topic=?", (topic,)), "kapitel": q("SELECT * FROM kapitel WHERE topic=?", (topic,)), "bausteine": q("SELECT * FROM bausteine WHERE topic=?", (topic,)), "atome": q("SELECT * FROM atome WHERE topic=?", (topic,)), "anker": q("SELECT k.* FROM anker k JOIN atome a ON a.id=k.atom_id" " WHERE a.topic=?", (topic,)), "kanten": q("SELECT * FROM kanten WHERE topic=?", (topic,)), "artefakte": q("SELECT ar.* FROM artefakte ar JOIN atome a ON a.id=ar.atom_id" " WHERE a.topic=?", (topic,)), "leitner": q("SELECT l.* FROM leitner l JOIN artefakte ar ON ar.id=l.artefakt_id" " JOIN atome a ON a.id=ar.atom_id WHERE a.topic=?", (topic,)), "sections": q("SELECT s.* FROM sections s JOIN bausteine b ON b.id=s.baustein_id" " WHERE b.topic=?", (topic,)), "auftraege": q("SELECT au.* FROM auftraege au JOIN bausteine b" " ON b.id=au.baustein_id WHERE b.topic=?", (topic,)), "runs": q("SELECT * FROM runs WHERE topic=? ORDER BY id", (topic,)), "befunde": q("SELECT b.* FROM befunde b JOIN runs r ON r.id=b.run_id" " WHERE r.topic=?", (topic,)), "events": q("SELECT e.* FROM events e JOIN runs r ON r.id=e.run_id" " WHERE r.topic=? ORDER BY e.id", (topic,)), } for zeile in d["quellen"]: # absolute Pfade überleben den Rechnerwechsel nicht if zeile.get("snapshot"): try: zeile["snapshot"] = str(Path(zeile["snapshot"]).relative_to(PROJECT_ROOT)) except ValueError: pass return d def _einfuegen(tabelle: str, zeilen: list[dict], **ersetzungen) -> dict: """Zeilen ohne id einfügen; Verweisfelder via ersetzungen={feld: map} remappen. → Map alte id → neue id.""" id_map: dict = {} for z in zeilen: z = dict(z) alt = z.pop("id", None) for feld, m in ersetzungen.items(): if z.get(feld) is not None: z[feld] = m.get(z[feld]) id_map[alt] = db.insert(tabelle, **z) return id_map def importieren(d: dict) -> None: topic = d["topic"] # Vor dem destruktiven topic_loeschen (rmtree) den Namen validieren — ein # manipulierter Export mit topic="../evil" würde sonst außerhalb von storage # löschen/schreiben. if not config.topic_name_ok(topic): raise ValueError(f"ungültiger Topic-Name: {topic!r}") if db.one("SELECT name FROM topics WHERE name=?", (topic,)): pipeline.topic_loeschen(topic) # Zielseite wird überschrieben t = dict(d["topics"][0]) resets_alt = db.uj(t.get("resets") or "{}", {}) t["resets"] = db.j({}) # wird nach dem Event-Import übersetzt db.insert("topics", **t) q_map = _einfuegen("quellen", [ {**z, "snapshot": str(PROJECT_ROOT / z["snapshot"]) if z.get("snapshot") and not Path(z["snapshot"]).is_absolute() else z.get("snapshot")} for z in d["quellen"]]) soll_zeilen = [] for z in d["soll"]: z = dict(z) belege = db.uj(z.get("belege") or "[]") for b in belege: if b.get("quelle") in q_map: b["quelle"] = q_map[b["quelle"]] z["belege"] = db.j(belege) z["geprueft"] = db.j([q_map[q] for q in db.uj(z.get("geprueft") or "[]") if q in q_map]) soll_zeilen.append(z) t_map = _einfuegen("themen", d.get("themen", [])) # Alt-Exporte: kein Key → leer s_map = _einfuegen("soll", soll_zeilen, thema_id=t_map) z_map = _einfuegen("lernziele", d["lernziele"], soll_id=s_map) k_map = _einfuegen("kapitel", d["kapitel"]) b_map = _einfuegen("bausteine", d["bausteine"], ziel_id=z_map, kapitel_id=k_map, thema_id=t_map) # merged_into zeigt auf atome selbst → zweiphasig; braucht-Key aus # Alt-Exporten verwerfen (Spalte existiert nicht mehr) a_map = _einfuegen("atome", [{**{k: v for k, v in z.items() if k != "braucht"}, "merged_into": None} for z in d["atome"]], soll_id=s_map, ziel_id=z_map, baustein_id=b_map) for z in d["atome"]: if z.get("merged_into") is not None: db.update("atome", "id", a_map[z["id"]], merged_into=a_map.get(z["merged_into"])) _einfuegen("anker", d["anker"], atom_id=a_map, quelle_id=q_map) _einfuegen("kanten", [z for z in d["kanten"] if z.get("art") != "braucht"], von_atom=a_map, zu_atom=a_map) ar_map = _einfuegen("artefakte", d["artefakte"], atom_id=a_map) for z in d["leitner"]: # PK ist artefakt_id — kein eigenes id-Feld z = dict(z) z["artefakt_id"] = ar_map.get(z["artefakt_id"]) db.insert("leitner", **z) # Atom-Marker im Text remappen — ohne Rewrite zeigten nach dem Import ALLE # Marker auf nicht existierende Atom-IDs (aak: 421/421 tot; Beispiel-/ # Fragen-Verknüpfung lief ins Leere) def marker_remap(m: re.Match) -> str: alt = int(m.group(1)) return m.group(0).replace(m.group(1), str(a_map[alt]), 1) if alt in a_map else m.group(0) def beispiel_remap(m: re.Match) -> str: # Beispiel-Marker tragen ARTEFAKT-ids — ohne Rewrite zeigt der Marker nach # dem Import auf eine fremde Zeile (latent: fiel erst auf, als zusätzliche # Artefakt-Typen die id-Sequenz verschoben) alt = int(m.group(1)) return m.group(0).replace(m.group(1), str(ar_map[alt]), 1) if alt in ar_map else m.group(0) def auftrag_remap(m: re.Match) -> str: alt = int(m.group(2)) return f"{m.group(1)}{a_map[alt]}" if alt in a_map else m.group(0) for z in d["sections"]: # PK ist baustein_id z = dict(z) z["baustein_id"] = b_map.get(z["baustein_id"]) for feld in ("text_lang", "text_kompakt"): if z.get(feld): z[feld] = re.sub(r"", beispiel_remap, z[feld]) # Alt-Exporte tragen Freitext-Aufträge in der (gedroppten) befunde-Spalte — # KRITISCH-Arten in auftraege-Zeilen wandeln, Rest verwerfen (wie Migration) for alt in db.uj(z.pop("befunde", None) or "[]"): p = db.alt_auftrag(re.sub(r"(Atom |Marker )(\d+)", auftrag_remap, str(alt))) if p and z["baustein_id"] is not None: db.insert("auftraege", baustein_id=z["baustein_id"], art=p[0], detail=p[1], quelle=p[2]) db.insert("sections", **z) # Aufträge nennen Atom-IDs im Klartext („Atom 123") — mitremappen, sonst # urteilt der Prüfer gegen tote IDs for z in d.get("auftraege", []): z = dict(z) z.pop("id", None) z["baustein_id"] = b_map.get(z["baustein_id"]) if z["baustein_id"] is None: continue z["detail"] = re.sub(r"(Atom |Marker )(\d+)", auftrag_remap, z["detail"]) db.insert("auftraege", **z) r_map = _einfuegen("runs", d["runs"]) _einfuegen("befunde", d["befunde"], run_id=r_map) e_map = _einfuegen("events", d["events"], run_id=r_map) resets_neu = {} for ebene, grenze in resets_alt.items(): # Marker sind Event-ids → übersetzen neue = [neu for alt, neu in e_map.items() if alt is not None and alt <= grenze] resets_neu[ebene] = max(neue, default=0) db.update("topics", "name", topic, resets=db.j(resets_neu)) print(f"importiert: {topic} — {len(d['atome'])} Atome, {len(d['sections'])} Sections," f" {len(d['events'])} Events", file=sys.stderr) if __name__ == "__main__": if len(sys.argv) >= 3 and sys.argv[1] == "export": print(json.dumps(export(sys.argv[2]), ensure_ascii=False)) elif len(sys.argv) >= 2 and sys.argv[1] == "import": importieren(json.load(sys.stdin)) else: sys.exit("Nutzung: transfer.py export | transfer.py import < datei.json")