"""OnePager-Pipeline: Recherche → Recherche-Prüfung → Bauen → Prüfung (7 Karten im 3×3-Raster).""" from pathlib import Path from fsutil import atomic_write_json from jsonio import read_json_file as _json_datei from pipeline import ( CANCELLED, FAILED, GenContext, _check_then_fix, _extra, _fail, _prompt, _set_step, _timeout, is_guide_cancelled, run_single_slot, ) async def _generate_onepager( guide_id: str, topic: str, instructions: str, provider: str, project: Path | None, content_path: Path, ) -> list[dict] | None: ctx = GenContext(topic=topic, provider=provider, is_cancelled=lambda: is_guide_cancelled(guide_id), guide_id=guide_id) # 3×3-Raster: 7 Karten mit festen Schlüsseln (Reihenfolge = Lesereihenfolge mobil) KARTEN_KEYS = ("info", "eigenschaften", "beispiel", "zusammenhaenge", "voraussetzungen", "modern", "veraltet") def karten_schema(data): """{"karten": {key: {titel, md}}} → Liste · sonst None.""" if not isinstance(data, dict): return None karten = data.get("karten") if not isinstance(karten, dict): return None out = [] for key in KARTEN_KEYS: k = karten.get(key) if not isinstance(k, dict) or not isinstance(k.get("titel"), str) or not isinstance(k.get("md"), str): return None titel, md = k["titel"].strip(), k["md"].strip() if not titel or len(md) < 5: # abgebrochene/leere Karten sind ungültig return None out.append({"key": key, "titel": titel, "md": md}) return out d, stem = content_path.parent, content_path.stem recherche_path = d / f"{stem}.recherche.md" recherche_check_path = d / f"{stem}.recherche-check.json" karten_path = d / f"{stem}.karten.json" check_path = d / f"{stem}.onepager-check.json" # Projekte bekommen eigene Recherche-Dimensionen — Produkt-Fragen # (Version, Lizenz, Alternativen) laufen dort ins Leere. if project: source = _prompt("OnePager-Quelle-Projekt", project=project) recherche_template = "OnePager-Recherche-Projekt" recherche_check_template = "OnePager-Recherche-Check-Projekt" else: source = _prompt("OnePager-Quelle-Thema", topic=topic) recherche_template = "OnePager-Recherche" recherche_check_template = "OnePager-Recherche-Check" def recherche_payload(result=None): if not recherche_path.exists(): return None text = recherche_path.read_text(encoding="utf-8").strip() return text or None # Schritt 1: Recherche — vorhandene Datei wird übernommen (Resume) recherche = recherche_payload() if recherche is None: await _set_step(guide_id, 0, "Recherchiere…") status, recherche = await run_single_slot( ctx, "OnePager-Recherche", key=f"{guide_id}-recherche", prompt=_prompt(recherche_template, topic=topic, source=source, out_path=recherche_path, extra=_extra(instructions)), role="quick", capabilities="files" if project else "full", payload=recherche_payload, timeout=_timeout("onepager_recherche"), ) if status == CANCELLED: return None if status == FAILED: await _fail(guide_id, "OnePager-Recherche fehlgeschlagen") return None # Schritt 2: Recherche-Prüfung — notiert Probleme; Anpassung macht ein Recherche-Agent status, fixed = await _check_then_fix( ctx, name="Recherche", step=1, check_key=f"{guide_id}-recherche-check", check_prompt=_prompt(recherche_check_template, topic=topic, recherche=recherche, out_path=recherche_check_path), check_path=recherche_check_path, check_timeout=_timeout("onepager_verify"), fix_key=f"{guide_id}-recherche-fix", build_fix_prompt=lambda probleme: _prompt( "OnePager-Recherche-Fix", topic=topic, source=source, recherche=recherche, probleme="\n".join(f"- {p}" for p in probleme), out_path=recherche_path, extra=_extra(instructions), ), fix_payload=recherche_payload, fix_timeout=_timeout("onepager_recherche"), fix_role="quick", fix_caps="files" if project else "full", ) if status == CANCELLED: return None if status == FAILED: await _fail(guide_id, "Recherche-Prüfung fehlgeschlagen") return None if fixed is not None: recherche = fixed # Schritt 3: Bauen — Karten nur aus der Faktenbasis (Resume: gültige Datei wird übernommen) karten = karten_schema(_json_datei(karten_path)) if karten is None: await _set_step(guide_id, 2, "Baue OnePager…") karten_path.unlink(missing_ok=True) status, karten = await run_single_slot( ctx, "OnePager-Bauen", key=f"{guide_id}-bauen", prompt=_prompt("OnePager-Bauen", topic=topic, recherche=recherche, out_path=karten_path, extra=_extra(instructions)), role="fast", capabilities="files", payload=lambda result: karten_schema(_json_datei(karten_path)), timeout=_timeout("onepager_bauen"), ) if status == CANCELLED: return None if status == FAILED: await _fail(guide_id, "OnePager-Bau fehlgeschlagen") return None def karten_block() -> str: return "\n\n".join(f"### {k['titel']} [{k['key']}]\n{k['md']}" for k in karten) # Schritt 4: Prüfung — notiert Probleme; Anpassung macht ein Bauen-Agent status, fixed = await _check_then_fix( ctx, name="OnePager", step=3, check_key=f"{guide_id}-verify", check_prompt=_prompt("OnePager-Verifikation", topic=topic, recherche=recherche, karten=karten_block(), out_path=check_path), check_path=check_path, check_timeout=_timeout("onepager_verify"), fix_key=f"{guide_id}-karten-fix", build_fix_prompt=lambda probleme: _prompt( "OnePager-Fix", topic=topic, recherche=recherche, karten=karten_block(), probleme="\n".join(f"- {p}" for p in probleme), out_path=karten_path, extra=_extra(instructions), ), fix_payload=lambda result: karten_schema(_json_datei(karten_path)), fix_timeout=_timeout("onepager_bauen"), on_fix_invalid=lambda: atomic_write_json( karten_path, {"karten": {k["key"]: {"titel": k["titel"], "md": k["md"]} for k in karten}}, ), ) if status == CANCELLED: return None if status == FAILED: await _fail(guide_id, "OnePager-Prüfung fehlgeschlagen") return None if fixed is not None: karten = fixed sections = [ {"num": i, "title": k["titel"], "md": k["md"], "key": k["key"]} for i, k in enumerate(karten, 1) ] return [{"title": topic, "sections": sections}]