import asyncio import uuid from datetime import datetime, timezone from pathlib import Path from agents import run_agent, kill_process from config import ( AGENT_TIMEOUT, DEFAULT_PROVIDER, TEMPLATES_DIR, MAX_CONCURRENT_GENERATIONS, ) from database import update_guide from paths import final_html_path, project_dir _semaphore = asyncio.Semaphore(MAX_CONCURRENT_GENERATIONS) _cancelled: set[str] = set() async def cancel_guide(guide_id: str) -> bool: _cancelled.add(guide_id) kill_process(guide_id) now = datetime.now(timezone.utc).isoformat() await update_guide(guide_id, status="error", progress=None, error_msg="Abgebrochen", updated_at=now) return True async def _set_progress(guide_id: str, progress: str) -> None: now = datetime.now(timezone.utc).isoformat() await update_guide(guide_id, progress=progress, updated_at=now) # Welche Baustein-Kategorien jedes Format abdeckt. FORMAT_COVERAGE = { "OnePager": "NUR die KERN-Bausteine, maximal verdichtet", "MiniGuide": "NUR die KERN-Bausteine", "Guide": "die KERN- und WICHTIG-Bausteine", } def _prompt(name: str, **kwargs) -> str: template = (TEMPLATES_DIR / "Prompt" / f"{name}.md").read_text(encoding="utf-8") return template.format(**kwargs) def _extra(instructions: str) -> str: return f"\n\nZUSÄTZLICHE ANWEISUNGEN VOM NUTZER:\n{instructions}\n" if instructions else "" def _build_bausteine_prompt(topic: str, bausteine_path: Path, instructions: str = "", project: Path | None = None) -> str: if project: source = _prompt("Bausteine-Quelle-Projekt", project=project) else: source = _prompt("Bausteine-Quelle-Thema", topic=topic) return _prompt( "Bausteine", topic=topic, source=source, bausteine_path=bausteine_path, extra=_extra(instructions), ) def _build_guide_prompt(topic: str, format_name: str, html_path: Path, bausteine: str, instructions: str = "", project: Path | None = None) -> str: spec = (TEMPLATES_DIR / "Format" / f"{format_name}.md").read_text(encoding="utf-8") reference = (TEMPLATES_DIR / "Referenz" / f"{format_name}.md").read_text(encoding="utf-8") if project: facts = _prompt("Guide-Fakten-Projekt", project=project) else: facts = _prompt("Guide-Fakten-Thema") return _prompt( "Guide", topic=topic, format_name=format_name, html_path=html_path, bausteine=bausteine, coverage=FORMAT_COVERAGE[format_name], facts=facts, spec=spec, reference=reference, extra=_extra(instructions), ) async def generate_guide(guide_id: str, topic: str, format_name: str, instructions: str = "", provider: str = DEFAULT_PROVIDER) -> None: async with _semaphore: now = datetime.now(timezone.utc).isoformat() await update_guide(guide_id, status="generating", progress="Ermittle Bausteine…", updated_at=now) html_path = final_html_path(topic, format_name) bausteine_path = html_path.with_suffix(".bausteine.md") project = project_dir(topic) if project_dir(topic).is_dir() else None try: if guide_id in _cancelled: return # Step 1: Bausteine ermitteln (Thema: Websuche, Projekt: Dateien lesen) current_step = "Bausteine" bs_prompt = _build_bausteine_prompt(topic, bausteine_path, instructions, project) returncode, bs_out, bs_err = await run_agent( guide_id, bs_prompt, AGENT_TIMEOUT, provider=provider, role="fast", capabilities="files" if project else "full", ) if guide_id in _cancelled: return if returncode != 0: await _fail(guide_id, _claude_error("Baustein-Fehler", returncode, bs_out, bs_err)) return if not bausteine_path.exists(): await _fail(guide_id, "Baustein-Datei wurde nicht erstellt") return bausteine = bausteine_path.read_text(encoding="utf-8") # Step 2: Generator-Agent erstellt HTML nach Bausteinen await _set_progress(guide_id, "Generiere HTML…") current_step = "Generierung" gen_prompt = _build_guide_prompt(topic, format_name, html_path, bausteine, instructions, project) returncode, stdout, stderr = await run_agent(guide_id, gen_prompt, AGENT_TIMEOUT, provider=provider, role="guide", capabilities="full") if guide_id in _cancelled: return if returncode != 0: await _fail(guide_id, _claude_error("Generator-Fehler", returncode, stdout, stderr)) return if not html_path.exists(): await _fail(guide_id, "HTML-Datei wurde nicht erstellt") return now = datetime.now(timezone.utc).isoformat() await update_guide( guide_id, status="done", progress=None, updated_at=now, ) except asyncio.TimeoutError: await _fail(guide_id, f"Timeout bei {current_step} nach {AGENT_TIMEOUT}s") except Exception as e: await _fail(guide_id, str(e)[:2000]) finally: _cancelled.discard(guide_id) def _claude_error(label: str, returncode: int, stdout: str, stderr: str) -> str: stderr = (stderr or "").strip() if stderr: return f"{label}: {stderr[:1000]}" tail = (stdout or "").strip()[-500:] if tail: return f"{label} (exit {returncode}, stderr leer): …{tail}" return f"{label} (exit {returncode}, ohne Ausgabe)" async def _fail(guide_id: str, msg: str) -> None: now = datetime.now(timezone.utc).isoformat() await update_guide(guide_id, status="error", progress=None, error_msg=msg, updated_at=now) def _build_guide_chat_prompt(topic: str, format_name: str, section: str, outline: str, messages: list[dict]) -> str: transcript = "\n".join( f"{'Nutzer' if m.get('role') == 'user' else 'Assistent'}: {m.get('content', '')}" for m in messages ) return _prompt( "Chat", topic=topic, format_name=format_name, outline_block=outline.strip() or "(keine)", section_block=section.strip() or "(kein Abschnitt erkannt)", transcript=transcript, ) async def chat_with_guide(topic: str, format_name: str, section: str, outline: str, messages: list[dict], provider: str = DEFAULT_PROVIDER) -> str: try: prompt = _build_guide_chat_prompt(topic, format_name, section, outline, messages) returncode, stdout, stderr = await run_agent( "chat-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none" ) if returncode != 0: return "Entschuldigung, das hat nicht geklappt. Bitte versuche es erneut." reply = stdout.strip() return reply or "Entschuldigung, ich habe keine Antwort erhalten." except Exception: return "Entschuldigung, das hat nicht geklappt. Bitte versuche es erneut."