#!/usr/bin/env python3 """ article_pipeline.py — Stage 0: Input-Preparation für den BSN-Artikel-Workflow. Liest Topic, PDFs, URLs und baut einen strukturierten Prompt für den Hermes-Agenten, der dann via delegate_task Research (Stage 1) und Write+QC (Stage 2+3) durchführt. Stage 4 (Git, Verify) macht der Agent selbst. """ import argparse import json import os import sys from datetime import datetime WORKSPACE = "/home/hermes/workspace" DOC_CACHE = os.path.expanduser("~/.hermes/cache/documents") def ocr_pdf(pdf_path: str) -> str: """PDF mit pymupdf OCRen und Text extrahieren.""" import pymupdf doc = pymupdf.open(pdf_path) texts = [] for page in doc: texts.append(page.get_text()) doc.close() return "\n\n".join(texts) def build_prompt(topic: str, sources: list[str], rubric: str, category_id: int, image_urls: list[str], deadline: str | None, pdf_texts: dict[str, str]) -> str: """Direkten Agent-Prompt für den Artikel-Workflow bauen.""" now = datetime.now() date_str = now.strftime("%d. %B %Y") slug = topic.lower().replace(" ", "_").replace(":", "").replace("×", "x")[:40] date_slug = now.strftime("%Y-%m-%d") filename = f"artikel_{slug}_{date_slug}.html" # --- WRITE+QC CONTEXT (most complex part) --- write_context = ( "STRUKTUR (exakt einhalten, nummeriert):\n" "1.

TITEL

\n" '2. DACHZEILE:
' 'Zusammenfassung in 1-2 Sätzen
\n' '3.

' 'DATUM · Lesezeit: ca. N Minuten

\n' '4. Einleitungs-Absatz

\n' '5. INFOBOX (PFLICHT bei Unternehmen/Produkt-News): ' '

' '

' 'NAME — Fakten

' '
' 'LabelWert
\n' '6. ALLE Inhalts-Abschnitte (

+

)\n' '7.


<-- NACH allen Inhalten, VOR Quellen!\n' '8. Quellen-Überschrift: Quellen\n' '9. JEDE Quelle als:
  • Name — Beschreibung
  • \n' '10.

    \n' '11.

    Recherchestand: DATUM. ' 'Basierend auf QUELLEN.

    \n' '\n' 'KRITISCHE REGELN (QC-Fail bei Verstoß):\n' '- Read-More NACH allen Inhalten, VOR Quellen (NICHT nach Absatz 1!)\n' '- JEDE Quelle als mit target="_blank" — KEINE unverlinkten Quellen!\n' '- Dachzeile + Infobox sind PFLICHT\n' '- Quellen-Überschrift: font-size:1.17em (NICHT 0.9rem)\n' '- Kein "XML API v2", kein "Betriebsgeheimnis", keine Emojis\n' '- Datei endet mit \n' f'- Rubrik: {rubric}' ) # --- BUILD OUTPUT --- lines = [] lines.append(f"Führe den BSN-Artikel-Workflow für folgendes Thema aus:") lines.append(f"**Thema:** {topic}") lines.append(f"**Datum:** {date_str}") lines.append(f"**Rubrik:** {rubric} (Joomla catid={category_id})") if deadline: lines.append(f"**Deadline:** {deadline}") lines.append(f"**Dateiname:** {filename}") lines.append("") # PDF-OCR-Ergebnisse if pdf_texts: lines.append("## PDF-Quellen (OCR-extrahiert, bereits im Context)") for path, text in pdf_texts.items(): fname = os.path.basename(path) truncated = text[:3000] if len(text) > 3000: truncated += f"\n... (gekürzt, Original: {len(text)} Zeichen)" lines.append(f"\n### {fname}\n{truncated}\n") lines.append("") # URL-Quellen if sources: lines.append("## URL-Quellen") for s in sources: lines.append(f"- {s}") lines.append("") # STAGE 1: Research lines.append("## STAGE 1: Research") lines.append("") lines.append("Nutze **delegate_task** um einen Research-Subagent zu starten:") lines.append("") lines.append( f'- goal: "Recherchiere zum Thema: {topic}. ' f'Sammle Fakten aus allen Quellen, bewerte sie (AAA/AA/A), ' f'ergänze BGG-Daten (Designer, Verlag, Jahr). ' f'Erstelle Research-Brief in /tmp/research_brief.md. ' f'Bei unvollständigen Kern-Fakten → Exit 1."' ) lines.append( '- context: "Enthält PDF-Texte und URLs (siehe oben). ' 'Nutze 5-Stufen-Eskalation für unerreichbare URLs. ' 'web_search site:boardgamegeek.com für BGG-Daten."' ) lines.append('- toolsets: ["web", "terminal", "file", "search", "skills"]') lines.append("") lines.append( "**WICHTIG:** Warte bis Sub-Agent #1 fertig ist. " "Prüfe /tmp/research_brief.md. Bei Exit-Code 1 → MELDUNG an Basti, NICHT fortfahren." ) lines.append("") # STAGE 2+3: Write+QC lines.append("## STAGE 2+3: Write + QC") lines.append("") lines.append("Nutze **delegate_task** für einen Write+QC-Subagent:") lines.append("") lines.append( f'- goal: "Schreibe HTML-Artikel zum Thema: {topic}. ' f'Nutze Research-Brief aus /tmp/research_brief.md. ' f'Datei: /home/hermes/workspace/{filename}. ' f'Danach: bash /home/hermes/workspace/verify_article.sh ' f'/home/hermes/workspace/{filename}. ' f'Bei Fehlern korrigieren, max 3 Versuche."' ) lines.append(f'- context: """\n{write_context}\n"""') lines.append('- toolsets: ["web", "terminal", "file", "search", "skills"]') lines.append("") lines.append("**WICHTIG:** Warte bis Sub-Agent #2 fertig. Prüfe Exit-Code und Datei.") lines.append("") # STAGE 4: Deliver lines.append("## STAGE 4: Deliver") lines.append("") lines.append("1. Git-Commit:") lines.append(f" cd /home/hermes/workspace && git add {filename}") lines.append(f' git commit -m "artikel: {topic[:50]}"') lines.append("2. Prüfen ob Article-Server erreichbar:") lines.append(" curl -s 'http://185.162.249.159:8890/?key=br3ttsp1el-n3ws-2026' | head -3") lines.append("3. Basti benachrichtigen per send_message (Telegram):") lines.append(f" ✅ Artikel fertig: {topic}") lines.append(f" 📄 /home/hermes/workspace/{filename}") lines.append(" 📊 verify_article.sh: 0 Fehler") return "\n".join(lines) def main(): parser = argparse.ArgumentParser( description="BSN Article Pipeline — Stage 0 Prompt Builder" ) parser.add_argument("--topic", help="Artikel-Thema") parser.add_argument("--sources", nargs="*", default=[], help="URLs oder PDF-Pfade") parser.add_argument("--rubric", default="Korrektur", help="Rubrik-Name (default: Korrektur)") parser.add_argument("--category-id", type=int, default=8, help="Joomla Category-ID (default: 8)") parser.add_argument("--image-urls", nargs="*", default=[], help="Bild-URLs aus PM") parser.add_argument("--deadline", default=None, help="ISO-Timestamp für Deadline") parser.add_argument("--batch", help="JSON-Datei mit Batch-Inputs") args = parser.parse_args() # Batch-Modus if args.batch: with open(args.batch) as f: batch_inputs = json.load(f) outputs = [] for inp in batch_inputs: prompt = build_prompt( topic=inp.get("topic", ""), sources=inp.get("sources", []), rubric=inp.get("rubric", "Korrektur"), category_id=inp.get("category_id", 8), image_urls=inp.get("image_urls", []), deadline=inp.get("deadline"), pdf_texts={} ) outputs.append({ "topic": inp.get("topic"), "prompt": prompt }) print(json.dumps(outputs, indent=2, ensure_ascii=False)) return # Einzel-Modus if not args.topic: print("ERROR: --topic erforderlich (oder --batch)", file=sys.stderr) sys.exit(1) # PDFs OCRen pdf_texts = {} for src in args.sources: if src.lower().endswith(".pdf"): full_path = src if not os.path.isabs(full_path): full_path = os.path.join(DOC_CACHE, src) if os.path.exists(full_path): try: pdf_texts[full_path] = ocr_pdf(full_path) print(f"[Stage 0] PDF OCR: {full_path} " f"({len(pdf_texts[full_path])} Zeichen)", file=sys.stderr) except Exception as e: print(f"[Stage 0] PDF-Fehler: {e}", file=sys.stderr) # Prompt bauen prompt = build_prompt( topic=args.topic, sources=args.sources, rubric=args.rubric, category_id=args.category_id, image_urls=args.image_urls, deadline=args.deadline, pdf_texts=pdf_texts, ) print(prompt) if __name__ == "__main__": main()