"""Researcher-Agent: Sucht nach Informationen via Claude WebSearch.""" import json import logging import re from agents.claude_client import call_claude, ClaudeUsage from config import CLAUDE_MODEL_FAST logger = logging.getLogger("osint.researcher") RESEARCH_PROMPT_TEMPLATE = """Du bist ein OSINT-Recherche-Agent für ein Lagemonitoring-System. AUSGABESPRACHE: {output_language} WICHTIG: Verwende IMMER echte UTF-8-Umlaute (ä, ö, ü, ß) — NIEMALS Umschreibungen (ae, oe, ue, ss). AUFTRAG: Suche nach aktuellen Informationen zu folgendem Vorfall: Titel: {title} Kontext: {description} REGELN: - Suche nur bei seriösen Nachrichtenquellen (Nachrichtenagenturen, Qualitätszeitungen, öffentlich-rechtliche Medien, Behörden) - KEIN Social Media (Twitter/X, Facebook, Instagram, TikTok, Reddit) - KEINE Boulevardmedien (Bild, Sun, Daily Mail etc.) {language_instruction} - Faktenbasiert und neutral - keine Spekulationen - Nutze removepaywalls.com für Paywall-geschützte Artikel (z.B. Spiegel+, Zeit+, SZ+): https://www.removepaywalls.com/search?url=ARTIKEL_URL - Nutze WebFetch um die 3-5 wichtigsten Artikel vollständig abzurufen und zusammenzufassen Gib die Ergebnisse AUSSCHLIESSLICH als JSON-Array zurück, ohne Erklärungen davor oder danach. Jedes Element hat diese Felder: - "headline": Originale Überschrift - "headline_de": Übersetzung in Ausgabesprache (falls Originalsprache abweicht) - "source": Name der Quelle (z.B. "Reuters", "tagesschau") - "source_url": URL des Artikels - "content_summary": Zusammenfassung des Inhalts (3-5 Sätze, in Ausgabesprache) - "language": Sprache des Originals (z.B. "de", "en", "fr") - "published_at": Veröffentlichungsdatum falls bekannt (ISO-Format) Antworte NUR mit dem JSON-Array. Keine Einleitung, keine Erklärung.""" DEEP_RESEARCH_PROMPT_TEMPLATE = """Du bist ein OSINT-Tiefenrecherche-Agent für ein Lagemonitoring-System. AUSGABESPRACHE: {output_language} WICHTIG: Verwende IMMER echte UTF-8-Umlaute (ä, ö, ü, ß) — NIEMALS Umschreibungen (ae, oe, ue, ss). AUFTRAG: Führe eine umfassende Hintergrundrecherche durch zu: Titel: {title} Kontext: {description} RECHERCHE-STRATEGIE: - Breite Suche: Hintergrundberichte, Analysen, Expertenmeinungen, Think-Tank-Publikationen - Suche nach: Akteuren, Zusammenhängen, historischem Kontext, rechtlichen Rahmenbedingungen - Akademische und Fachquellen zusätzlich zu Nachrichtenquellen - Nutze removepaywalls.com für Paywall-geschützte Artikel (z.B. https://www.removepaywalls.com/search?url=ARTIKEL_URL) - Nutze WebFetch um die 3-5 wichtigsten Artikel vollständig abzurufen und zusammenzufassen {language_instruction} - Ziel: 8-15 hochwertige Quellen QUELLENTYPEN (priorisiert): 1. Fachzeitschriften und Branchenmedien 2. Qualitätszeitungen (Hintergrundberichte, Dossiers) 3. Think Tanks und Forschungsinstitute 4. Offizielle Dokumente und Pressemitteilungen 5. Nachrichtenagenturen (für Faktengrundlage) AUSSCHLUSS: - KEIN Social Media (Twitter/X, Facebook, Instagram, TikTok, Reddit) - KEINE Boulevardmedien - KEINE Meinungsblogs ohne Quellenbelege Gib die Ergebnisse AUSSCHLIESSLICH als JSON-Array zurück, ohne Erklärungen davor oder danach. Jedes Element hat diese Felder: - "headline": Originale Überschrift - "headline_de": Übersetzung in Ausgabesprache (falls Originalsprache abweicht) - "source": Name der Quelle (z.B. "netzpolitik.org", "Handelsblatt") - "source_url": URL des Artikels - "content_summary": Ausführliche Zusammenfassung des Inhalts (5-8 Sätze, in Ausgabesprache) - "language": Sprache des Originals (z.B. "de", "en", "fr") - "published_at": Veröffentlichungsdatum falls bekannt (ISO-Format) Antworte NUR mit dem JSON-Array. Keine Einleitung, keine Erklärung.""" # Sprach-Anweisungen LANG_INTERNATIONAL = "- Suche in Deutsch UND Englisch für internationale Abdeckung" LANG_GERMAN_ONLY = "- Suche NUR auf Deutsch bei deutschsprachigen Quellen (Deutschland, Österreich, Schweiz)\n- KEINE englischsprachigen oder anderssprachigen Quellen" LANG_DEEP_INTERNATIONAL = "- Suche in Deutsch, Englisch und weiteren relevanten Sprachen" LANG_DEEP_GERMAN_ONLY = "- Suche NUR auf Deutsch bei deutschsprachigen Quellen (Deutschland, Österreich, Schweiz)\n- KEINE englischsprachigen oder anderssprachigen Quellen" FEED_SELECTION_PROMPT_TEMPLATE = """Du bist ein OSINT-Analyst. Wähle aus dieser Feed-Liste die Feeds aus, die für die Lage relevant sein könnten. Generiere außerdem optimierte Suchbegriffe für das RSS-Matching. LAGE: {title} KONTEXT: {description} INTERNATIONALE QUELLEN: {international} FEEDS: {feed_list} REGELN: - Wähle alle Feeds die thematisch oder regional relevant sein könnten - Lieber einen Feed zu viel als zu wenig auswählen - Bei "Internationale Quellen: Nein": Keine internationalen Feeds auswählen - Allgemeine Nachrichtenfeeds (tagesschau, Spiegel etc.) sind fast immer relevant - QUELLENVIELFALT: Wähle pro Domain maximal 2-3 Feeds. Bevorzuge eine breite Mischung aus verschiedenen Quellen statt vieler Feeds derselben Domain. KEYWORDS-REGELN: - Generiere 5-10 thematisch relevante Suchbegriffe für das RSS-Matching - Nur inhaltlich relevante Begriffe (Personen, Orte, Themen, Organisationen) - KEINE Jahreszahlen (2024, 2025, 2026 etc.) - KEINE Monatsnamen (Januar, Februar, März etc.) - KEINE generischen Wörter (aktuell, news, update etc.) - Begriffe in Kleinbuchstaben - Sowohl deutsche als auch englische Begriffe wo sinnvoll Antworte NUR mit einem JSON-Objekt in diesem Format: {{"feeds": [1, 2, 5, 12], "keywords": ["begriff1", "begriff2", "begriff3"]}}""" class ResearcherAgent: """Führt OSINT-Recherchen über Claude CLI WebSearch durch.""" async def select_relevant_feeds( self, title: str, description: str, international: bool, feeds_metadata: list[dict], ) -> tuple[list[dict], list[str] | None, ClaudeUsage | None]: """Lässt Claude die relevanten Feeds für eine Lage vorauswählen. Nutzt Haiku (CLAUDE_MODEL_FAST) für diese einfache Aufgabe. Returns: (ausgewählte Feeds, keywords, usage) — Bei Fehler: (alle Feeds, None, None) """ # Feed-Liste als nummerierte Übersicht formatieren feed_lines = [] for i, feed in enumerate(feeds_metadata, 1): feed_lines.append( f"{i}. {feed['name']} ({feed['domain']}) [{feed['category']}]" ) prompt = FEED_SELECTION_PROMPT_TEMPLATE.format( title=title, description=description or "Keine weitere Beschreibung", international="Ja" if international else "Nein", feed_list="\n".join(feed_lines), ) try: result, usage = await call_claude(prompt, tools=None, model=CLAUDE_MODEL_FAST) # Neues Format: JSON-Objekt mit "feeds" und "keywords" keywords = None indices = None # Versuche JSON-Objekt zu parsen obj_match = re.search(r'\{[^{}]*"feeds"\s*:\s*\[[\d\s,]+\][^{}]*\}', result, re.DOTALL) if obj_match: try: obj = json.loads(obj_match.group()) indices = obj.get("feeds", []) raw_keywords = obj.get("keywords", []) if isinstance(raw_keywords, list) and raw_keywords: keywords = [str(k).lower().strip() for k in raw_keywords if k] logger.info(f"Feed-Selektion Keywords: {keywords}") except (json.JSONDecodeError, ValueError): pass # Fallback: altes Array-Format if indices is None: arr_match = re.search(r'\[[\d\s,]+\]', result) if not arr_match: logger.warning("Feed-Selektion: Kein JSON in Antwort, nutze alle Feeds") return feeds_metadata, None, usage indices = json.loads(arr_match.group()) selected = [] for idx in indices: if isinstance(idx, int) and 1 <= idx <= len(feeds_metadata): selected.append(feeds_metadata[idx - 1]) if not selected: logger.warning("Feed-Selektion: Keine gültigen Indizes, nutze alle Feeds") return feeds_metadata, keywords, usage logger.info( f"Feed-Selektion: {len(selected)} von {len(feeds_metadata)} Feeds ausgewählt" ) return selected, keywords, usage except Exception as e: logger.warning(f"Feed-Selektion fehlgeschlagen ({e}), nutze alle Feeds") return feeds_metadata, None, None async def search(self, title: str, description: str = "", incident_type: str = "adhoc", international: bool = True) -> tuple[list[dict], ClaudeUsage | None]: """Sucht nach Informationen zu einem Vorfall.""" from config import OUTPUT_LANGUAGE if incident_type == "research": lang_instruction = LANG_DEEP_INTERNATIONAL if international else LANG_DEEP_GERMAN_ONLY prompt = DEEP_RESEARCH_PROMPT_TEMPLATE.format( title=title, description=description, language_instruction=lang_instruction, output_language=OUTPUT_LANGUAGE, ) else: lang_instruction = LANG_INTERNATIONAL if international else LANG_GERMAN_ONLY prompt = RESEARCH_PROMPT_TEMPLATE.format( title=title, description=description, language_instruction=lang_instruction, output_language=OUTPUT_LANGUAGE, ) try: result, usage = await call_claude(prompt) articles = self._parse_response(result) # Ausgeschlossene Quellen dynamisch aus DB laden excluded_sources = await self._get_excluded_sources() # Ausgeschlossene Quellen filtern filtered = [] for article in articles: source = article.get("source", "").lower() source_url = article.get("source_url", "").lower() excluded = False for excl in excluded_sources: if excl in source or excl in source_url: excluded = True break if not excluded: # Bei nur-deutsch: nicht-deutsche Ergebnisse nachfiltern if not international and article.get("language", "de") != "de": continue filtered.append(article) logger.info(f"Recherche ergab {len(filtered)} Artikel (von {len(articles)} gefundenen, international={international})") return filtered, usage except TimeoutError: raise # Timeout nach oben durchreichen fuer Retry im Orchestrator except Exception as e: logger.error(f"Recherche-Fehler: {e}") return [], None async def _get_excluded_sources(self) -> list[str]: """Lädt ausgeschlossene Quellen aus der Datenbank.""" try: from source_rules import get_source_rules rules = await get_source_rules() return rules.get("excluded_domains", []) except Exception as e: logger.warning(f"Fallback auf config.py für Excluded Sources: {e}") from config import EXCLUDED_SOURCES return list(EXCLUDED_SOURCES) def _parse_response(self, response: str) -> list[dict]: """Parst die Claude-Antwort als JSON-Array.""" # Versuche JSON direkt zu parsen try: data = json.loads(response) if isinstance(data, list): return data if isinstance(data, dict) and "articles" in data: return data["articles"] except json.JSONDecodeError: pass # JSON-Code-Block extrahieren code_pat = r'`{3}(?:json)?\s*\n?(\[.*?\])\s*`{3}' code_match = re.search(code_pat, response, re.DOTALL) if code_match: try: data = json.loads(code_match.group(1)) if isinstance(data, list): return data except json.JSONDecodeError: pass # Versuche JSON aus der Antwort zu extrahieren (zwischen [ und ]) arr_pat = r'\[\s*\{.*\}\s*\]' match = re.search(arr_pat, response, re.DOTALL) if match: try: data = json.loads(match.group()) if isinstance(data, list): return data except json.JSONDecodeError: pass # Letzter Versuch: einzelne JSON-Objekte mit headline objects = re.findall(r'\{[^{}]*"headline"[^{}]*\}', response) if objects: results = [] for obj_str in objects: try: obj = json.loads(obj_str) if "headline" in obj: results.append(obj) except json.JSONDecodeError: continue if results: logger.info(f"JSON-Recovery: {len(results)} Artikel aus Einzelobjekten extrahiert") return results logger.warning(f"Konnte Claude-Antwort nicht als JSON parsen (Laenge: {len(response)})") return []