#!/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. Usage: # Thema + PDF python3 article_pipeline.py --topic "Gamegenic × TCC" --pdf pm.pdf --rubric "Nachrichten" # Thema + URL python3 article_pipeline.py --topic "Neue Kennerspiel-Nominierungen" --url "https://..." # Nur Thema (autonome Recherche) python3 article_pipeline.py --topic "Spiel des Jahres 2026" --rubric "Nachrichten" # Batch (JSON) python3 article_pipeline.py --batch topics.json Output: Strukturierter Prompt nach stdout (wird vom Hermes-Cronjob als Context injiziert) Exit-Codes: 0 = Prompt bereit, 1 = Fehler in Stage 0 """ import argparse import json import os import sys import subprocess import tempfile 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 extract_url_text(url: str) -> str: """URL-Inhalt via 5-Stufen-Eskalation abrufen.""" import urllib.request import urllib.error # Stufe 1: curl direkt try: req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=15) as resp: return resp.read().decode("utf-8", errors="ignore")[:5000] except Exception: pass # Stufe 2-5 wäre hier — für Stage 0 reicht Stufe 1 return f"[URL nicht direkt abrufbar: {url}]" 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] 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:** artikel_{slug}_{now.strftime('%Y-%m-%d')}.html") 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}. Sammle Fakten aus allen Quellen, bewerte sie (AAA/AA/A), ergänze BGG-Daten (Designer, Verlag, Jahr). Erstelle Research-Brief in /tmp/research_brief.md. 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}. Nutze Research-Brief aus /tmp/research_brief.md. Datei: artikel_{slug}_{now.strftime("%Y-%m-%d")}.html in /home/hermes/workspace/. Danach: bash /home/hermes/workspace/verify_article.sh artikel_{slug}_{now.strftime("%Y-%m-%d")}.html. Bei Fehlern korrigieren, max 3 Versuche."') lines.append(f'- context: "HTML-Regeln: KEIN XML API v2, KEIN Betriebsgeheimnis, KEINE Emojis.

mit Datum+Lesezeit. Genau 1×

. Quellen als Quellen +
    mit .

    mit Quellen-Ratings. Datei endet mit . Rubrik: {rubric}."') 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" ```bash") lines.append(f" cd /home/hermes/workspace && git add artikel_{slug}_{now.strftime('%Y-%m-%d')}.html") lines.append(f' git commit -m "artikel: {topic[:50]}"') lines.append(f" ```") 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/artikel_{slug}_{now.strftime('%Y-%m-%d')}.html") 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()