"""Report-Generator: PDF und Word Berichte aus Lage-Daten.""" import base64 import io import json import logging import re import uuid from collections import defaultdict from datetime import datetime from html import escape as _html_escape from pathlib import Path import pikepdf from jinja2 import Environment, FileSystemLoader from weasyprint import HTML from docx import Document from docx.shared import Inches, Pt, Cm, RGBColor from docx.enum.text import WD_ALIGN_PARAGRAPH from docx.enum.table import WD_TABLE_ALIGNMENT from config import TIMEZONE, CLAUDE_MODEL_FAST from services import media_registry logger = logging.getLogger("osint.report") TEMPLATE_DIR = Path(__file__).parent / "report_templates" LOGO_PATH = Path(__file__).parent / "static" / "favicon.svg" FC_STATUS_LABELS_DE = { # 1:1 vom Monitor-Frontend (components.js) — konsistent zum UI. "confirmed": "Bestätigt", "unconfirmed": "Unbestätigt", # "Widerlegt" war hier falsch: Der Status meint laut Auftrag Belege, die # einander widersprechen. Im Bericht vom 01.08.2026 stand deshalb eine # sachlich zutreffende Zahlenspanne als widerlegt. Die echte Widerlegung # hat jetzt den eigenen Status "false". "contradicted": "Widersprüchlich", "developing": "Unklar", "established": "Gesichert", "disputed": "Umstritten", "unverified": "Ungeprüft", "false": "Widerlegt", } FC_STATUS_LABELS_EN = { "confirmed": "Confirmed", "unconfirmed": "Unconfirmed", "contradicted": "Conflicting", "developing": "Developing", "established": "Established", "disputed": "Disputed", "unverified": "Unverified", "false": "Disproved", } def _fc_labels(lang_iso: str = "de") -> dict: """Liefert FC-Status-Labels in der gewuenschten Sprache.""" return FC_STATUS_LABELS_EN if lang_iso == "en" else FC_STATUS_LABELS_DE # Backward-compatible alias (Default DE) -- veraltet, nutze _fc_labels(lang) FC_STATUS_LABELS = FC_STATUS_LABELS_DE def _get_logo_base64() -> str: """Logo als Base64 für HTML-Embedding.""" try: return base64.b64encode(LOGO_PATH.read_bytes()).decode() except Exception: return "" def _prepare_sources(incident: dict) -> list: """Quellenverzeichnis aus sources_json parsen. Die Nummer aus dem Lagebild (nr) ist die Identitaet des Eintrags, sie wird hier gesichert und die Liste danach sortiert. Der Anzeigename wird je Domain vereinheitlicht, damit derselbe Verlag nicht unter mehreren Namen im Verzeichnis steht. """ raw = incident.get("sources_json") if not raw: return [] try: roh = json.loads(raw) if isinstance(raw, str) else raw except (json.JSONDecodeError, TypeError): return [] if not isinstance(roh, list): return [] eintraege = [s for s in roh if isinstance(s, dict)] namen = media_registry.namen_je_domain( [(s.get("url") or "", s.get("name") or "") for s in eintraege] ) aufbereitet = [] for position, s in enumerate(eintraege, 1): kopie = dict(s) kopie["nr"] = _als_nummer(s.get("nr"), position) url = s.get("url") or "" domain = media_registry.registrable_domain(url) # Bei Weiterleitungsportalen sagt die Domain nichts ueber das Medium: # vier verschiedene Zeitungen liegen dort alle unter news.google.com. # Der gelieferte Name bleibt deshalb unangetastet. if domain and not media_registry.ist_aggregator(url): kopie["name"] = namen.get(domain) or kopie.get("name") or media_registry.name_aus_domain(domain) kopie["domain"] = domain elif domain: kopie["domain"] = domain aufbereitet.append(kopie) aufbereitet.sort(key=lambda s: s["nr"] if isinstance(s["nr"], int) else 10**6) return aufbereitet def _renumber_sources(sources: list) -> tuple[list, dict]: """Vergibt fuer die Ausgabe eine lueckenlose Nummerierung ab 1. Das Modell nummeriert alle vorgelegten Meldungen durch, zitiert aber nur einen Teil davon. Im Verzeichnis entstehen dadurch Luecken (Lage 53 sprang von 20 auf 22, 24, 26). Fachlich ist das korrekt, fuer den Leser sieht es nach einem Fehler aus. Die gespeicherten Nummern bleiben unangetastet, damit sie ueber Folge-Refreshes stabil bleiben, nur der Bericht zaehlt neu. Gibt die neu nummerierten Quellen und die Abbildung alt -> neu zurueck. """ abbildung: dict[int, int] = {} neu = [] for position, s in enumerate(sources or [], 1): kopie = dict(s) alt = s.get("nr") if isinstance(alt, int): abbildung[alt] = position kopie["nr"] = position neu.append(kopie) return neu, abbildung def _apply_citation_map(text: str, abbildung: dict) -> str: """Schreibt [alte Nummer] auf [neue Nummer] um. Verweise ohne Quelle im Verzeichnis werden entfernt statt umgeschrieben. Sie waeren nach der Umnummerierung nicht nur unaufloesbar, sondern wuerden auf einen fremden Eintrag zeigen. Ein stiller Verweis ins Leere ist der schwerere Fehler als eine fehlende Klammer. """ if not text or not abbildung: return text verwaist: list[str] = [] def ersetze(m: re.Match) -> str: roh = m.group(1) nummer = _als_nummer(roh, -1) if nummer in abbildung: return f"[{abbildung[nummer]}]" verwaist.append(roh) return "" ergebnis = re.sub(r"\[(\d{1,5}[a-z]?)\]", ersetze, text) if verwaist: logger.warning( "Bericht: %d Quellenverweise ohne Verzeichniseintrag entfernt (%s)", len(verwaist), ", ".join(sorted(set(verwaist))[:10]), ) return ergebnis def _als_nummer(wert, ersatz: int) -> int: """Quellennummer als ganze Zahl, Buchstaben-Suffixe werden abgeschnitten.""" if isinstance(wert, int): return wert text = str(wert or "").strip() m = re.match(r"^(\d+)", text) return int(m.group(1)) if m else ersatz def _prepare_source_stats(sources: list, articles: list) -> list: """Quellenstatistik ueber die im Lagebild zitierten Belege. Frueher zaehlte die Statistik alle gesammelten Artikel der Lage, waehrend das Quellenverzeichnis darunter nur die zitierten Quellen auflistete. Die beiden Tabellen widersprachen sich damit in jedem Bericht (im Ceuta-Fall 49 gezaehlte Artikel gegen 25 aufgefuehrte Quellen). Jetzt bezieht sich die Statistik auf dieselbe Grundmenge wie das Verzeichnis: die Summe der Spalte Belege entspricht genau der Zahl der Eintraege. Gezaehlt wird je Domain, nicht je Name, damit mehrere Feeds desselben Verlags eine Zeile ergeben. Die Sprache kommt aus den Artikeln derselben Domain, eine eindeutige Sprachangabe in der Adresse hat Vorrang. """ sprachen_je_domain: dict[str, set] = defaultdict(set) for art in articles or []: domain = media_registry.registrable_domain(art.get("source_url") or "") if not domain: continue aus_url = media_registry.sprache_aus_url(art.get("source_url") or "") sprachen_je_domain[domain].add(aus_url or (art.get("language") or "de").upper()) stats_map: dict[str, dict] = {} for s in sources or []: url = s.get("url") or "" domain = media_registry.registrable_domain(url) if domain and media_registry.ist_aggregator(url): # Weiterleitungsportale sind kein Medium. Zusammenfassen wuerde # hier vier Zeitungen unter einem Namen verschmelzen, also nach # Name gruppieren und die Weiterleitung offen ausweisen. anzeige = f"{s.get('name') or 'Unbekannt'} (Weiterleitung)" schluessel = f"redirect:{anzeige}" else: anzeige = s.get("name") or media_registry.name_aus_domain(domain) schluessel = domain or anzeige eintrag = stats_map.setdefault( schluessel, {"name": anzeige, "count": 0, "langs": set()}, ) eintrag["count"] += 1 aus_url = media_registry.sprache_aus_url(url) if aus_url: eintrag["langs"].add(aus_url) elif domain in sprachen_je_domain: eintrag["langs"].update(sprachen_je_domain[domain]) stats = [] for data in sorted(stats_map.values(), key=lambda d: (-d["count"], d["name"])): stats.append({ "name": data["name"], "count": data["count"], "languages": ", ".join(sorted(data["langs"])), }) return stats def _source_stats_note(sources: list, articles: list) -> str: """Erlaeuterung unter der Quellenstatistik. Haelt fest, worauf sich die Tabelle bezieht und wie viele Meldungen insgesamt ausgewertet wurden. Ohne diesen Satz wirkte die frueher groessere Artikelzahl wie eine Aufwertung der Quellenbasis. """ zitiert = len(sources or []) gesamt = len(articles or []) medien = len({ media_registry.registrable_domain(a.get("source_url") or "") for a in (articles or []) if media_registry.registrable_domain(a.get("source_url") or "") }) if not gesamt: return f"Die Tabelle zählt die {zitiert} im Lagebild zitierten Belege, gruppiert nach Medium." return ( f"Die Tabelle zählt die {zitiert} im Lagebild zitierten Belege, gruppiert nach Medium. " f"Ausgewertet wurden insgesamt {gesamt} Meldungen aus {medien} Medien." ) def _prepare_fact_checks(fact_checks: list, lang_iso: str = "de") -> list: """Faktenchecks mit Label aufbereiten.""" labels = _fc_labels(lang_iso) fallback = "Unknown" if lang_iso == "en" else "Unbekannt" result = [] for fc in fact_checks: fc_copy = dict(fc) fc_copy["status_label"] = labels.get(fc.get("status", ""), fc.get("status", fallback)) result.append(fc_copy) return result def _prepare_timeline(articles: list) -> list: """Timeline aus Artikeln: sortiert nach Datum.""" timeline = [] for art in articles: pub = art.get("published_at") or art.get("collected_at") or "" pub = str(pub) if pub else "" headline = art.get("headline_de") or art.get("headline") or "Ohne Titel" source = art.get("source") or "" if pub: try: dt = datetime.fromisoformat(pub.replace("Z", "+00:00")) date_str = dt.strftime("%d.%m.%Y %H:%M") except Exception: date_str = pub[:16] else: date_str = "" timeline.append({"date": date_str, "headline": headline, "source": source, "sort_key": pub}) timeline.sort(key=lambda x: x["sort_key"], reverse=True) return timeline[:100] # Max 100 Einträge def _markdown_to_html(text: str) -> str: """Einfache Markdown -> HTML Konvertierung für Lagebild.""" if not text: return "

Keine Zusammenfassung verfügbar.

" # Basic Markdown -> HTML html = text # Headlines html = re.sub(r'^### (.+)$', r'

\1

', html, flags=re.MULTILINE) html = re.sub(r'^## (.+)$', r'

\1

', html, flags=re.MULTILINE) # Bold html = re.sub(r'\*\*(.+?)\*\*', r'\1', html) # Links [text](url) html = re.sub(r'\[([^\]]+)\]\(([^)]+)\)', r'\1', html) # Bullet lists html = re.sub(r'^- (.+)$', r'
  • \1
    • ', html, flags=re.MULTILINE) html = re.sub(r'(
  • .*
    • \n?)+', lambda m: '', html) # Paragraphs paragraphs = html.split('\n\n') result = [] for p in paragraphs: p = p.strip() if not p: continue if p.startswith('{p}

    ') return '\n'.join(result) def _parse_developments_for_export(text: str) -> list[tuple[str, str]]: """Parst die 'Neuesten Entwicklungen' (latest_developments) fuer den Export. Eingabeformat je Eintrag: '- [DD.MM. HH:MM] Text {Quelle|URL, ...}'. Liefert (datum_label, body) je Eintrag in gespeicherter Reihenfolge. Quellen-Klammern und [N]-Zitate werden entfernt — der Export zeigt bewusst KEINE Links. Das gespeicherte Format enthaelt kein Jahr; fehlt es, wird das aktuelle Jahr ergaenzt (Live-Monitoring-Berichte sind tagesaktuell). """ if not text: return [] year2 = datetime.now(TIMEZONE).strftime("%y") bullet_re = re.compile( r"^\s*(?:[-*•]\s*)?\[\s*(\d{1,2})\.(\d{1,2})\.?(?:(\d{2,4}))?\s+(\d{1,2}:\d{2})\s*\]\s*(.+?)\s*$" ) trailing_braces = re.compile(r"\s*\{[^{}]*\}\s*\.?\s*$") citation_re = re.compile(r"\s*\[\d{1,5}[a-z]?\]") result: list[tuple[str, str]] = [] for raw in text.splitlines(): line = raw.strip() if not line: continue m = bullet_re.match(line) if not m: continue day, month, year, time = m.group(1), m.group(2), m.group(3), m.group(4) body = m.group(5).strip() # Quellen-Klammer am Ende und Inline-[N]-Zitate entfernen (keine Links) body = trailing_braces.sub("", body).strip() body = citation_re.sub("", body).strip() if not body: continue yy = year[-2:] if year else year2 label = f"{int(day):02d}.{int(month):02d}.{yy}, {time} Uhr" result.append((label, body)) return result def _format_latest_developments_html(text: str) -> str: """Rendert die 'Neuesten Entwicklungen' als HTML-Block fuer den PDF-Export. Pro Eintrag: Datum/Uhrzeit-Zeile, darunter (eigener Absatz) der Meldungstext. Keine Quellen-Links. Faellt bei nicht-parsebarem Text auf _markdown_to_html zurueck. """ pairs = _parse_developments_for_export(text) if not pairs: return _markdown_to_html(text) blocks = [] for label, body in pairs: body_html = _html_escape(body) body_html = re.sub(r'\*\*(.+?)\*\*', r'\1', body_html) blocks.append( '
    ' f'
    {_html_escape(label)}
    ' f'
    {body_html}
    ' '
    ' ) return "\n".join(blocks) def _truncate_lagebild(summary_text: str, max_chars: int = 4000) -> str: """Lagebild für den Lagebericht auf die Zusammenfassung kürzen. Nimmt nur den ersten Abschnitt (bis zur zweiten H2/H3-Überschrift) oder kürzt auf max_chars Zeichen mit sauberem Abbruch am Absatzende. """ if not summary_text or len(summary_text) <= max_chars: return summary_text lines = summary_text.split("\n") result_lines = [] heading_count = 0 char_count = 0 for line in lines: stripped = line.strip() # Zähle Überschriften (## oder ###) if stripped.startswith("## ") or stripped.startswith("### "): heading_count += 1 # Nach der 3. Überschrift abbrechen (= 2 Abschnitte) if heading_count > 3: break result_lines.append(line) char_count += len(line) + 1 # Hard-Limit bei max_chars, aber am Absatzende abbrechen if char_count > max_chars and stripped == "": break text = "\n".join(result_lines).rstrip() if len(text) < len(summary_text) - 100: text += "\n\n*[Vollständige Zusammenfassung im Vollständigen Bericht]*" return text def _strip_citation_numbers(text: str) -> str: """Entfernt [1234]-Quellenreferenzen aus dem Text.""" # Einzelne Referenzen: [1302] text = re.sub(r"\s*\[\d{1,5}\]", "", text) # Mehrfach-Referenzen: [725][765][768] text = re.sub(r"(\[\d{1,5}\]){2,}", "", text) # Aufräumen: Doppelte Leerzeichen text = re.sub(r" +", " ", text) return text def _find_source_for_citation(num: str, sources: list) -> dict | None: """Sucht eine Quelle anhand der Zitat-Nummer (inkl. Suffix-Fallback wie 1383a -> 1383).""" if not sources: return None for s in sources: try: if str(s.get("nr")) == num: return s except Exception: continue # Suffix-Fallback: 1383a -> 1383 if re.search(r"[a-z]$", num): base = re.sub(r"[a-z]$", "", num) for s in sources: if str(s.get("nr")) == base: return s return None def _linkify_citations_html(text: str, sources: list) -> str: """Ersetzt [1234]-Zitate durch HTML-Links zur jeweiligen Quelle. Nummern ohne zugeordnete Quelle bleiben als sichtbare Zahl erhalten. """ if not text: return text if not sources: return text def repl(match: re.Match) -> str: num = match.group(1) src = _find_source_for_citation(num, sources) if src and src.get("url"): url = src["url"].replace('"', """) name = (src.get("name") or "").replace('"', """) return f'[{num}]' return match.group(0) return re.sub(r"\[(\d{1,5}[a-z]?)\]", repl, text) def _add_docx_hyperlink(paragraph, url: str, text: str): """Fügt einen klickbaren Hyperlink in ein python-docx-Paragraph-Objekt ein.""" from docx.oxml.shared import OxmlElement, qn part = paragraph.part r_id = part.relate_to( url, "http://schemas.openxmlformats.org/officeDocument/2006/relationships/hyperlink", is_external=True, ) hyperlink = OxmlElement("w:hyperlink") hyperlink.set(qn("r:id"), r_id) new_run = OxmlElement("w:r") rPr = OxmlElement("w:rPr") color = OxmlElement("w:color") color.set(qn("w:val"), "0066CC") rPr.append(color) u = OxmlElement("w:u") u.set(qn("w:val"), "single") rPr.append(u) sz = OxmlElement("w:sz") sz.set(qn("w:val"), "20") rPr.append(sz) new_run.append(rPr) t = OxmlElement("w:t") t.text = text t.set(qn("xml:space"), "preserve") new_run.append(t) hyperlink.append(new_run) paragraph._p.append(hyperlink) return hyperlink def _add_docx_paragraph_with_citations(doc_or_para, text: str, sources: list, style: str | None = None): """Fügt ein Paragraph hinzu, bei dem [1234]-Zitate als Hyperlink-Runs eingefügt werden. doc_or_para darf ein Document sein (neues Paragraph wird angelegt) oder bereits ein Paragraph. """ if hasattr(doc_or_para, "add_paragraph"): para = doc_or_para.add_paragraph(style=style) if style else doc_or_para.add_paragraph() else: para = doc_or_para pattern = re.compile(r"\[(\d{1,5}[a-z]?)\]") pos = 0 for m in pattern.finditer(text): if m.start() > pos: para.add_run(text[pos:m.start()]) num = m.group(1) src = _find_source_for_citation(num, sources) if src and src.get("url"): _add_docx_hyperlink(para, src["url"], f"[{num}]") else: para.add_run(m.group(0)) pos = m.end() if pos < len(text): para.add_run(text[pos:]) return para def _extract_zusammenfassung_lines(summary_text: str) -> tuple[list[str], str]: """Extrahiert die ZUSAMMENFASSUNG-Sektion als Liste von Rohzeilen (ohne Zitatbearbeitung). Returns: (lines, remaining_summary) """ if not summary_text: return [], summary_text pattern = r"(## (?:ZUSAMMENFASSUNG|ÜBERBLICK)\s*\n)(.*?)(?=\n## |\Z)" match = re.search(pattern, summary_text, re.DOTALL) if not match: return [], summary_text zusammenfassung_raw = match.group(2).strip() remaining = summary_text[:match.start()] + summary_text[match.end():] remaining = remaining.strip() lines: list[str] = [] for line in zusammenfassung_raw.split("\n"): stripped = line.strip() if stripped.startswith("- ") or stripped.startswith("* "): content = stripped[2:].strip() if content: lines.append(content) elif stripped and not stripped.startswith("#"): lines.append(stripped) return lines, remaining def _extract_zusammenfassung(summary_text: str, sources: list | None = None) -> tuple[str, str]: """Extrahiert die ZUSAMMENFASSUNG-Sektion und liefert sie als HTML mit verlinkten Zitaten.""" lines, remaining = _extract_zusammenfassung_lines(summary_text) if not lines: return "", summary_text src_list = sources or [] html_lines = [f"
  • {_linkify_citations_html(line, src_list)}
    • " for line in lines] html = "" return html, remaining async def generate_executive_summary(summary_text: str) -> str: """KI-verdichtetes Executive Summary aus dem Lagebild.""" if not summary_text or len(summary_text.strip()) < 50: return "" from agents.claude_client import call_claude prompt = f"""Du bist ein Intelligence-Analyst für ein OSINT-Lagemonitoring-System. Verdichte das folgende Lagebild auf genau 3-5 Kernpunkte. REGELN: - Jeder Punkt: 1-2 Sätze, faktenbasiert - Fokus: Was ist passiert? Was bedeutet es? Was ist die aktuelle Dynamik? - Sprache: Deutsch, sachlich, prägnant - Format: Gib NUR die Bullet Points aus, einen pro Zeile, mit "- " am Anfang - KEINE Einleitung, KEINE Überschrift, NUR die Punkte LAGEBILD: {summary_text}""" try: result, usage = await call_claude(prompt, tools=None, model=CLAUDE_MODEL_FAST) # Robuster Parser: Akzeptiert JSON, Markdown-Listen oder Freitext lines = [] text = result.strip() # Code-Fences entfernen (```json ... ```) if text.startswith("```"): text = re.sub(r"^```\w*\n?", "", text) text = re.sub(r"\n?```$", "", text) text = text.strip() # Fall 1: JSON-Antwort (Haiku gibt manchmal JSON zurück) if text.startswith("{"): try: data = json.loads(text) for key in data: if isinstance(data[key], list): for item in data[key]: clean = str(item).strip().lstrip("- ").lstrip("* ") if clean: lines.append(clean) break except json.JSONDecodeError: pass # Fall 2: Markdown Bullet Points if not lines: for line in text.split("\n"): stripped = line.strip() if stripped.startswith(("- ", "* ")): clean = stripped.lstrip("- ").lstrip("* ").strip() if clean: lines.append(clean) # Fall 3: Nummerierte Liste (1. 2. 3.) if not lines: for line in text.split("\n"): m = re.match(r"^\d+\.\s+(.+)", line.strip()) if m: lines.append(m.group(1).strip()) # Fallback: Ganzen Text als einen Punkt if not lines: lines = [text[:500]] html = "" return html except Exception as e: logger.error(f"Executive Summary Generierung fehlgeschlagen: {e}") return "" def _parse_db_timestamp(value) -> datetime | None: """SQLite-Timestamp robust als datetime parsen (ISO oder 'YYYY-MM-DD HH:MM:SS').""" if not value: return None if isinstance(value, datetime): return value try: text = str(value).replace("T", " ").replace("Z", "") # Sekundenbruchteile und Timezone-Offset abschneiden (python-docx mag nur naive dt) text = text.split(".")[0].split("+")[0].strip() return datetime.strptime(text, "%Y-%m-%d %H:%M:%S") except (ValueError, TypeError): try: return datetime.strptime(str(value)[:10], "%Y-%m-%d") except (ValueError, TypeError): return None def _slug_scope_label(scope: str, sections: set[str] | None) -> str: """Scope-Label fuer Metadaten und Dateinamen.""" if sections: if sections == {"zusammenfassung"}: return "Zusammenfassung" if "timeline" in sections: return "Vollständiger Bericht" return "Lagebericht" return {"summary": "Zusammenfassung", "report": "Lagebericht", "full": "Vollständiger Bericht"}.get( scope, "Lagebericht" ) def _build_export_metadata( incident: dict, articles: list, fact_checks: list, sources: list, creator: str, scope: str, sections: set[str] | None, organization_name: str | None, top_locations: list[str] | None, snapshot_count: int = 0, include_branding: bool = True, ) -> dict: """Einheitlicher Metadaten-Dict fuer PDF (HTML-Meta-Tags) und DOCX (core_properties). include_branding=False neutralisiert alle AegisSight-Firmenbezeichnungen (White-Label-Export). """ is_research = incident.get("type") == "research" type_label = "Hintergrundrecherche" if is_research else "Live-Monitoring" category = "OSINT-Hintergrundrecherche" if is_research else "OSINT-Lagebericht" scope_label = _slug_scope_label(scope, sections) title_raw = (incident.get("title") or "Unbenannte Lage").strip() title = f"{title_raw} — {type_label}" subject = (incident.get("description") or "").strip() if not subject: subject = f"{type_label} zu: {title_raw}" # DOCX-Core-Property "subject" erzwingt ein 255-Zeichen-Limit; laengere # Beschreibungen wuerden den Word-Export sonst mit ValueError abbrechen. if len(subject) > 255: subject = subject[:255] # Keywords sammeln (Reihenfolge relevant für Anzeige, Dedup mit dict.fromkeys) keywords: list[str] = ["OSINT", type_label] if organization_name: keywords.append(organization_name) # category_labels: kann JSON-Dict (Karte primary/secondary/...), JSON-Liste # oder ein Komma-getrennter String sein. Nur die Label-Werte extrahieren. cat_labels_raw = (incident.get("category_labels") or "").strip() if cat_labels_raw: cat_values: list[str] = [] try: parsed = json.loads(cat_labels_raw) if isinstance(parsed, dict): cat_values = [str(v).strip() for v in parsed.values() if isinstance(v, str) and v.strip()] elif isinstance(parsed, list): cat_values = [str(v).strip() for v in parsed if isinstance(v, str) and v.strip()] except (json.JSONDecodeError, TypeError): cat_values = [lbl.strip() for lbl in cat_labels_raw.split(",") if lbl.strip()] # Keine JSON-Fragmente (geschweifte/eckige Klammern) als Keyword zulassen for lbl in cat_values: if lbl and not any(c in lbl for c in "{}[]"): keywords.append(lbl) if top_locations: keywords.extend([loc for loc in top_locations if loc]) # Sanitize: Zeilenumbrueche/Tabs weg, Sonderzeichen mit PDF-Sonderbedeutung filtern def _sanitize_keyword(kw: str) -> str: if not kw: return "" # Whitespace normalisieren cleaned = re.sub(r"\s+", " ", kw).strip() # PDF-Dict/Array-Klammern und Backslash raus (WeasyPrint escaped () bei Strings, # { und [ koennen aber den Keywords-Stream abschneiden) cleaned = re.sub(r"[{}\[\]\\]", "", cleaned) return cleaned.strip(" ,;:") # Dedup (case-insensitive) mit Reihenfolge erhalten, max 15 seen = set() unique_keywords: list[str] = [] for kw in keywords: clean_kw = _sanitize_keyword(kw) if not clean_kw: continue key = clean_kw.lower() if key not in seen: seen.add(key) unique_keywords.append(clean_kw) if len(unique_keywords) >= 15: break now = datetime.now(TIMEZONE) created = _parse_db_timestamp(incident.get("created_at")) or now.replace(tzinfo=None) modified = _parse_db_timestamp(incident.get("updated_at")) or created # Strukturierter Comments-Block (wird in DOCX angezeigt, kompakt) stand = now.strftime("%d.%m.%Y") comments_lines = [ f"Incident-ID: {incident.get('id', '?')} | Typ: {incident.get('type', 'adhoc')} | Scope: {scope_label}", f"Stand: {stand}", ] if organization_name: comments_lines.append(f"Organisation: {organization_name}") comments_lines.append( f"Umfang: {len(articles)} Artikel, {len(fact_checks)} Faktenchecks, {len(sources)} Quellen" ) if top_locations: comments_lines.append("Orte: " + ", ".join(top_locations[:5])) comments = "\n".join(comments_lines) # Branding-abhaengige Felder: bei include_branding=False neutralisiert (White-Label-Export) if include_branding: publisher = organization_name or "AegisSight" author = creator or "AegisSight Monitor" creator_app = "AegisSight Monitor" producer = "WeasyPrint + AegisSight Monitor" urn_ns = "aegissight" rights = ( "Vertrauliche Lageanalyse — AegisSight Monitor. " "Weitergabe nur an autorisierte Empfänger." ) else: publisher = organization_name or "" author = creator or "Unbekannt" creator_app = "" producer = "WeasyPrint" urn_ns = "report" rights = "Vertrauliche Lageanalyse. Weitergabe nur an autorisierte Empfänger." identifier = f"urn:{urn_ns}:incident:{incident.get('id', '0')}:{now.strftime('%Y%m%dT%H%M%S')}" return { "title": title, "author": author, "subject": subject, "keywords": unique_keywords, "keywords_comma": ", ".join(unique_keywords), "keywords_semicolon": "; ".join(unique_keywords), "category": category, "comments": comments, "creator_app": creator_app, "producer": producer, "language": "de-DE", "created": created, "modified": modified, "created_iso": created.strftime("%Y-%m-%dT%H:%M:%S"), "modified_iso": modified.strftime("%Y-%m-%dT%H:%M:%S"), "type_label": type_label, "scope_label": scope_label, "publisher": publisher, "identifier": identifier, "rights": rights, "doc_type": "Report", "version_id": str(max(1, snapshot_count)), } def _format_pdf_date(dt: datetime) -> str: """PDF-Datumsformat: D:YYYYMMDDHHmmSS+HH'mm' (mit Zeitzone) oder Z (UTC).""" if dt.tzinfo is None: # Naive dt — als lokale TIMEZONE interpretieren dt = dt.replace(tzinfo=TIMEZONE) base = dt.strftime("D:%Y%m%d%H%M%S") offset = dt.utcoffset() if offset is None: return base + "Z" total_minutes = int(offset.total_seconds() // 60) sign = "+" if total_minutes >= 0 else "-" total_minutes = abs(total_minutes) return f"{base}{sign}{total_minutes // 60:02d}'{total_minutes % 60:02d}'" def _enrich_pdf_metadata(pdf_bytes: bytes, meta: dict) -> bytes: """PDF-Ausgabe um XMP-Metadaten und CreationDate/ModDate erweitern (post-process via pikepdf).""" try: buf_in = io.BytesIO(pdf_bytes) with pikepdf.Pdf.open(buf_in) as pdf: created: datetime = meta.get("created") modified: datetime = meta.get("modified") if created and created.tzinfo is None: created = created.replace(tzinfo=TIMEZONE) if modified and modified.tzinfo is None: modified = modified.replace(tzinfo=TIMEZONE) # Klassisches Info-Dict: CreationDate + ModDate nachziehen if created: pdf.docinfo["/CreationDate"] = pikepdf.String(_format_pdf_date(created)) if modified: pdf.docinfo["/ModDate"] = pikepdf.String(_format_pdf_date(modified)) # Document-/Instance-ID fuer DMS-Versionierung (frisch pro Export) doc_uuid = f"uuid:{uuid.uuid4()}" instance_uuid = f"uuid:{uuid.uuid4()}" # XMP-Metadatenblock schreiben (Dublin Core + XMP + PDF + xmpRights + xmpMM) with pdf.open_metadata(set_pikepdf_as_editor=False) as xmp: # Dublin Core xmp["dc:title"] = meta.get("title", "") xmp["dc:creator"] = [meta.get("author", "")] xmp["dc:description"] = meta.get("subject", "") if meta.get("keywords"): xmp["dc:subject"] = list(meta["keywords"]) xmp["dc:language"] = [meta.get("language", "de-DE")] xmp["dc:publisher"] = [meta.get("publisher", "AegisSight")] xmp["dc:identifier"] = meta.get("identifier", "") xmp["dc:format"] = "application/pdf" xmp["dc:type"] = [meta.get("doc_type", "Report")] xmp["dc:rights"] = meta.get("rights", "") if created: xmp["dc:date"] = [created.strftime("%Y-%m-%dT%H:%M:%S%z")] # PDF Namespace xmp["pdf:Keywords"] = meta.get("keywords_comma", "") xmp["pdf:Producer"] = meta.get("producer", "WeasyPrint + AegisSight Monitor") # XMP Namespace xmp["xmp:CreatorTool"] = meta.get("creator_app", "AegisSight Monitor") if created: xmp["xmp:CreateDate"] = created.strftime("%Y-%m-%dT%H:%M:%S%z") if modified: xmp["xmp:ModifyDate"] = modified.strftime("%Y-%m-%dT%H:%M:%S%z") xmp["xmp:MetadataDate"] = modified.strftime("%Y-%m-%dT%H:%M:%S%z") # xmpRights: Rechte- und Vertraulichkeitshinweis (XMP erwartet String "True") xmp["xmpRights:Marked"] = "True" if meta.get("rights"): # String: pikepdf wrapped das automatisch als LangAlt mit x-default xmp["xmpRights:UsageTerms"] = meta["rights"] # xmpMM: Document- und Instance-ID fuer DMS-Versionierung xmp["xmpMM:DocumentID"] = doc_uuid xmp["xmpMM:InstanceID"] = instance_uuid xmp["xmpMM:VersionID"] = meta.get("version_id", "1") # xmpMM:History — Audit-Event fuer diesen Export (einzeiliger Eintrag je Seq-Item) history_when = (modified or datetime.now(TIMEZONE)).strftime("%Y-%m-%dT%H:%M:%S%z") history_entry = ( f"action=published; when={history_when}; " f"softwareAgent={meta.get('creator_app', 'AegisSight Monitor')}; " f"instanceID={instance_uuid}; " f"scope={meta.get('scope_label', '')}; " f"version={meta.get('version_id', '1')}" ) xmp["xmpMM:History"] = [history_entry] buf_out = io.BytesIO() pdf.save(buf_out) return buf_out.getvalue() except Exception as e: logger.warning(f"PDF-Metadaten-Anreicherung (XMP/Dates) fehlgeschlagen: {e}") return pdf_bytes async def generate_pdf( incident: dict, articles: list, fact_checks: list, snapshots: list, scope: str, creator: str, executive_summary_html: str, sections: set[str] | None = None, organization_name: str | None = None, top_locations: list[str] | None = None, snapshot_count: int = 0, include_branding: bool = True, ) -> bytes: """PDF-Report via WeasyPrint generieren.""" # Sections aus scope ableiten wenn nicht explizit angegeben if sections is None: if scope == "summary": sections = {"zusammenfassung"} elif scope == "report": sections = {"zusammenfassung", "bericht", "faktencheck", "quellen"} else: # full sections = {"zusammenfassung", "bericht", "faktencheck", "quellen", "timeline"} # Zusammenfassungs-Quelle bestimmen: # - Research: ZUSAMMENFASSUNG/UEBERBLICK aus dem Bericht extrahieren. # - Live-Monitoring (adhoc): "Neueste Entwicklungen" aus latest_developments, # ohne Quellen-Links, Datum/Uhrzeit als eigene Zeile. # - sonst: KI-Executive-Summary (executive_summary_html). is_research = incident.get("type") == "research" all_sources, citation_map = _renumber_sources(_prepare_sources(incident)) latest_dev = (incident.get("latest_developments") or "").strip() zusammenfassung_html = _apply_citation_map(executive_summary_html, citation_map) bericht_summary = _apply_citation_map(incident.get("summary", ""), citation_map) zusammenfassung_title = "Zusammenfassung" summary_has_links = True if is_research and bericht_summary: extracted_html, remaining = _extract_zusammenfassung(bericht_summary, all_sources) if extracted_html: zusammenfassung_html = extracted_html bericht_summary = remaining elif not is_research and latest_dev: dev_html = _format_latest_developments_html(latest_dev) if dev_html: zusammenfassung_html = dev_html zusammenfassung_title = "Neueste Entwicklungen" summary_has_links = False # Quellen bewusst entfernt # KI-/Research-Zusammenfassung linkifizieren; Developments bleiben linkfrei if not is_research and summary_has_links and zusammenfassung_html: zusammenfassung_html = _linkify_citations_html(zusammenfassung_html, all_sources) meta = _build_export_metadata( incident, articles, fact_checks, all_sources, creator, scope, sections, organization_name, top_locations, snapshot_count=snapshot_count, include_branding=include_branding, ) env = Environment(loader=FileSystemLoader(str(TEMPLATE_DIR))) template = env.get_template("report.html") now = datetime.now(TIMEZONE) incident_type_label = "Hintergrundrecherche" if incident.get("type") == "research" else "Live-Monitoring" html_content = template.render( incident=incident, incident_type_label=incident_type_label, report_date=now.strftime("%d.%m.%Y, %H:%M Uhr"), creator=creator, logo_base64=_get_logo_base64(), executive_summary=zusammenfassung_html, zusammenfassung_title=zusammenfassung_title, sections=sections, scope=scope, lagebild_html=_linkify_citations_html( _markdown_to_html(bericht_summary), all_sources ), lagebild_timestamp=(incident.get("updated_at") or "")[:16].replace("T", " "), # Keine Kappung mehr. Ein bei 30 abgeschnittenes Verzeichnis machte # jede hoehere Referenz im Text unaufloesbar, der Bericht wies also # Belege aus, die er selbst nicht zeigte. sources=all_sources, fact_checks=_prepare_fact_checks(fact_checks), source_stats=_prepare_source_stats(all_sources, articles), source_stats_note=_source_stats_note(all_sources, articles), timeline=_prepare_timeline(articles) if scope == "full" else [], articles=articles if scope == "full" else [], meta=meta, include_branding=include_branding, ) # Artikel pub_date aufbereiten for art in articles: pub = str(art.get("published_at") or art.get("collected_at") or "") try: dt = datetime.fromisoformat(pub.replace("Z", "+00:00")) art["pub_date"] = dt.strftime("%d.%m.%Y") except Exception: art["pub_date"] = pub[:10] if pub else "" pdf_bytes = HTML(string=html_content).write_pdf() pdf_bytes = _enrich_pdf_metadata(pdf_bytes, meta) return pdf_bytes async def generate_docx( incident: dict, articles: list, fact_checks: list, snapshots: list, scope: str, creator: str, executive_summary_text: str, sections: set[str] | None = None, organization_name: str | None = None, top_locations: list[str] | None = None, snapshot_count: int = 0, include_branding: bool = True, ) -> bytes: """Word-Report via python-docx generieren.""" doc = Document() # Sections aus scope ableiten wenn nicht explizit angegeben if sections is None: if scope == "summary": sections = {"zusammenfassung"} elif scope == "report": sections = {"zusammenfassung", "bericht", "faktencheck", "quellen"} else: # full sections = {"zusammenfassung", "bericht", "faktencheck", "quellen", "timeline"} # Zusammenfassungs-Quelle bestimmen (analog generate_pdf): # Research -> Bericht-Extrakt, Live-Monitoring -> "Neueste Entwicklungen", sonst KI. is_research = incident.get("type") == "research" all_sources, citation_map = _renumber_sources(_prepare_sources(incident)) latest_dev = (incident.get("latest_developments") or "").strip() zusammenfassung_text = _apply_citation_map(executive_summary_text, citation_map) bericht_summary = _apply_citation_map( incident.get("summary") or "Keine Zusammenfassung verfügbar.", citation_map ) zusammenfassung_title = "Zusammenfassung" zusammenfassung_lines: list[str] = [] zusammenfassung_developments: list[tuple[str, str]] = [] if is_research and bericht_summary: extracted_lines, remaining = _extract_zusammenfassung_lines(bericht_summary) if extracted_lines: zusammenfassung_lines = extracted_lines bericht_summary = remaining elif not is_research and latest_dev: dev_pairs = _parse_developments_for_export(latest_dev) if dev_pairs: zusammenfassung_developments = dev_pairs zusammenfassung_title = "Neueste Entwicklungen" meta = _build_export_metadata( incident, articles, fact_checks, all_sources, creator, scope, sections, organization_name, top_locations, snapshot_count=snapshot_count, include_branding=include_branding, ) # Dateimetadaten setzen (sichtbar in Explorer/Finder, DMS-Systemen) cp = doc.core_properties cp.title = meta["title"] cp.author = meta["author"] cp.subject = meta["subject"] cp.keywords = meta["keywords_semicolon"] cp.comments = meta["comments"] cp.category = meta["category"] cp.last_modified_by = meta["author"] cp.language = meta["language"] cp.content_status = "Final" try: cp.created = meta["created"] cp.modified = meta["modified"] except (ValueError, TypeError) as e: logger.warning(f"DOCX created/modified konnte nicht gesetzt werden: {e}") # Styles style = doc.styles['Normal'] style.font.size = Pt(10) style.font.name = 'Calibri' # --- Deckblatt --- for _ in range(6): doc.add_paragraph() # Firmenname-Zeile nur im gebrandeten Export if include_branding: title_para = doc.add_paragraph() title_para.alignment = WD_ALIGN_PARAGRAPH.CENTER run = title_para.add_run("AegisSight Monitor") run.font.size = Pt(12) run.font.color.rgb = RGBColor(0x0a, 0x18, 0x32) doc.add_paragraph() type_label = "Hintergrundrecherche" if incident.get("type") == "research" else "Live-Monitoring" type_para = doc.add_paragraph() type_para.alignment = WD_ALIGN_PARAGRAPH.CENTER run = type_para.add_run(type_label) run.font.size = Pt(10) run.font.color.rgb = RGBColor(0x0a, 0x18, 0x32) title_para2 = doc.add_paragraph() title_para2.alignment = WD_ALIGN_PARAGRAPH.CENTER run = title_para2.add_run(incident.get("title", "")) run.font.size = Pt(24) run.font.bold = True run.font.color.rgb = RGBColor(0x0a, 0x18, 0x32) if incident.get("description"): desc_para = doc.add_paragraph() desc_para.alignment = WD_ALIGN_PARAGRAPH.CENTER run = desc_para.add_run(incident["description"]) run.font.size = Pt(11) run.font.color.rgb = RGBColor(0x66, 0x66, 0x66) doc.add_paragraph() for _ in range(3): doc.add_paragraph() now = datetime.now(TIMEZONE) meta_para = doc.add_paragraph() meta_para.alignment = WD_ALIGN_PARAGRAPH.CENTER run = meta_para.add_run(f"Stand: {now.strftime('%d.%m.%Y, %H:%M Uhr')}\nErstellt von: {creator}") run.font.size = Pt(9) run.font.color.rgb = RGBColor(0x0a, 0x18, 0x32) doc.add_page_break() # --- Zusammenfassung / Neueste Entwicklungen --- if "zusammenfassung" in sections: doc.add_heading(zusammenfassung_title, level=1) if zusammenfassung_developments: # Live-Monitoring: pro Eintrag Datum/Uhrzeit-Zeile + Absatz mit Text, ohne Links for label, body in zusammenfassung_developments: date_para = doc.add_paragraph() date_para.paragraph_format.space_after = Pt(1) run = date_para.add_run(label) run.bold = True run.font.size = Pt(9) run.font.color.rgb = RGBColor(0x0a, 0x18, 0x32) body_para = doc.add_paragraph() body_para.paragraph_format.space_after = Pt(8) body_para.add_run(re.sub(r'\*\*(.+?)\*\*', r'\1', body)) elif zusammenfassung_lines: for line in zusammenfassung_lines: _add_docx_paragraph_with_citations(doc, line, all_sources, style='List Bullet') else: # Fallback: HTML-Tags aus executive_summary_text strippen, dann Bullets bilden clean_text = re.sub(r'<[^>]+>', '', zusammenfassung_text or '') lines = [line.strip().lstrip("- ").lstrip("* ") for line in clean_text.strip().split("\n") if line.strip()] for line in lines: if line: _add_docx_paragraph_with_citations(doc, line, all_sources, style='List Bullet') if "bericht" in sections: # --- Lagebild / Recherchebericht --- doc.add_heading("Recherchebericht" if is_research else "Lagebild", level=1) # Markdown-Formatierung entfernen, Zitate aber als [NNN] beibehalten und als Hyperlinks rendern clean_summary = re.sub(r'\*\*(.+?)\*\*', r'\1', bericht_summary) clean_summary = re.sub(r'\[([^\]]+)\]\(([^)]+)\)', r'\1', clean_summary) clean_summary = re.sub(r'^#{1,3}\s+', '', clean_summary, flags=re.MULTILINE) for para_text in clean_summary.split("\n\n"): para_text = para_text.strip() if not para_text: continue if para_text.startswith("- "): for bullet in para_text.split("\n"): bullet = bullet.lstrip("- ").strip() if bullet: _add_docx_paragraph_with_citations(doc, bullet, all_sources, style='List Bullet') else: _add_docx_paragraph_with_citations(doc, para_text, all_sources) if "faktencheck" in sections: # --- Faktencheck --- report_fcs = fact_checks if report_fcs: doc.add_heading("Faktencheck", level=1) table = doc.add_table(rows=1, cols=3) table.style = 'Table Grid' table.alignment = WD_TABLE_ALIGNMENT.CENTER hdr = table.rows[0].cells hdr[0].text = "Behauptung" hdr[1].text = "Status" hdr[2].text = "Quellen" for cell in hdr: for p in cell.paragraphs: p.runs[0].font.bold = True p.runs[0].font.size = Pt(9) for fc in report_fcs: row = table.add_row().cells row[0].text = fc.get("claim", "") row[1].text = FC_STATUS_LABELS.get(fc.get("status", ""), fc.get("status", "")) row[2].text = str(fc.get("sources_count", 0)) if "quellen" in sections: # --- Quellenstatistik --- source_stats = _prepare_source_stats(all_sources, articles) if source_stats: doc.add_heading("Quellenstatistik", level=1) table = doc.add_table(rows=1, cols=3) table.style = 'Table Grid' table.alignment = WD_TABLE_ALIGNMENT.CENTER hdr = table.rows[0].cells hdr[0].text = "Quelle" hdr[1].text = "Belege" hdr[2].text = "Sprache" for cell in hdr: for p in cell.paragraphs: p.runs[0].font.bold = True p.runs[0].font.size = Pt(9) for stat in source_stats: row = table.add_row().cells row[0].text = stat["name"] row[1].text = str(stat["count"]) row[2].text = stat["languages"] hinweis = doc.add_paragraph() hinweis_run = hinweis.add_run(_source_stats_note(all_sources, articles)) hinweis_run.font.size = Pt(8) hinweis_run.font.color.rgb = RGBColor(0x66, 0x66, 0x66) # --- Quellenverzeichnis --- # Fehlte in der Word-Ausgabe komplett: Der Text verwies auf [1], [2], # ohne dass die Datei die Nummern irgendwo aufloeste. Die Nummer ist # die aus dem Lagebild (nr), NICHT die laufende Zeilennummer. if all_sources: doc.add_heading("Quellenverzeichnis", level=1) table = doc.add_table(rows=1, cols=3) table.style = 'Table Grid' table.alignment = WD_TABLE_ALIGNMENT.CENTER hdr = table.rows[0].cells hdr[0].text = "#" hdr[1].text = "Quelle" hdr[2].text = "URL" for cell in hdr: for p in cell.paragraphs: p.runs[0].font.bold = True p.runs[0].font.size = Pt(9) for src in all_sources: row = table.add_row().cells row[0].text = str(src.get("nr", "")) row[1].text = src.get("name") or src.get("title") or "" url = src.get("url") or "" if url: zelle = row[2].paragraphs[0] _add_docx_hyperlink(zelle, url, url) for cell in row: for p in cell.paragraphs: for run in p.runs: run.font.size = Pt(8) if "timeline" in sections: # --- Artikelverzeichnis --- if articles: doc.add_page_break() doc.add_heading(f"Artikelverzeichnis ({len(articles)} Artikel)", level=1) table = doc.add_table(rows=1, cols=4) table.style = 'Table Grid' table.alignment = WD_TABLE_ALIGNMENT.CENTER hdr = table.rows[0].cells for i, txt in enumerate(["Headline", "Quelle", "Sprache", "Datum"]): hdr[i].text = txt for p in hdr[i].paragraphs: p.runs[0].font.bold = True p.runs[0].font.size = Pt(8) for art in articles: row = table.add_row().cells row[0].text = art.get("headline_de") or art.get("headline") or "Ohne Titel" row[1].text = art.get("source") or "" row[2].text = (art.get("language") or "de").upper() pub = str(art.get("published_at") or art.get("collected_at") or "") try: dt = datetime.fromisoformat(pub.replace("Z", "+00:00")) row[3].text = dt.strftime("%d.%m.%Y") except Exception: row[3].text = pub[:10] if pub else "" # Schriftgröße reduzieren for cell in row: for p in cell.paragraphs: for run in p.runs: run.font.size = Pt(8) # --- Footer --- doc.add_paragraph() footer = doc.add_paragraph() footer.alignment = WD_ALIGN_PARAGRAPH.CENTER if include_branding: footer_text = f"Erstellt mit AegisSight Monitor — aegis-sight.de — {now.strftime('%d.%m.%Y')}" else: footer_text = f"Stand: {now.strftime('%d.%m.%Y')}" run = footer.add_run(footer_text) run.font.size = Pt(8) run.font.color.rgb = RGBColor(0x0a, 0x18, 0x32) buf = io.BytesIO() doc.save(buf) return buf.getvalue()