695983245d
Zwei Befunde aus dem Bericht zu Lage 60 vom 02.08.2026.
1. Doppelte Verweise, ein Folgefehler des Dubletten-Zusammenfuehrens. Zeigten
zwei Nummern auf dieselbe Adresse, wurden beide Verweise auf dieselbe neue
Nummer umgeschrieben und standen dann nebeneinander: "[24][24]". Im Bericht
kam das dreimal vor, in der gespeicherten Fassung nicht. Aufeinanderfolgende
Verweise auf dieselbe Nummer werden jetzt zusammengezogen.
2. Platzhalter ohne Adresse. Das Modell hatte einen Eintrag {"nr": 12, "name":
"Quelle", "url": ""} erzeugt. Der Bericht druckte ihn als "12 Quelle" ohne
Adresse und die Statistik fuehrte ein Medium namens "Quelle". Ein Eintrag
ohne Adresse ist kein pruefbarer Beleg und kommt nicht mehr ins
Verzeichnis; der Verweis darauf wird wie ein verwaister behandelt und
entfernt.
Gegenprobe an Lage 60: aus 30 Eintraegen werden 26 (drei doppelte Adressen
zusammengefuehrt, ein Platzhalter entfernt), die Zaehlung bleibt lueckenlos,
kein doppelter Verweis mehr im Text.
Neu sind 4 Pruefungen, insgesamt laufen 220 ohne Netzzugriff und ohne Kosten.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1380 Zeilen
54 KiB
Python
1380 Zeilen
54 KiB
Python
"""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 _dubletten_zusammenfuehren(aufbereitet)
|
|
|
|
|
|
def _url_schluessel(url: str) -> str:
|
|
"""Vergleichsform einer Adresse: ohne Schema, ohne www, ohne Endschraegstrich."""
|
|
text = (url or "").strip().lower()
|
|
for praefix in ("https://", "http://"):
|
|
if text.startswith(praefix):
|
|
text = text[len(praefix):]
|
|
if text.startswith("www."):
|
|
text = text[4:]
|
|
return text.rstrip("/")
|
|
|
|
|
|
def _dubletten_zusammenfuehren(sources: list) -> list:
|
|
"""Fuehrt Eintraege mit identischer Adresse auf eine Nummer zusammen.
|
|
|
|
Im Bericht vom 02.08.2026 standen 30 Eintraege fuer 28 verschiedene
|
|
Adressen: zweimal dieselbe tagesschau-Meldung unter je zwei Nummern. Eine
|
|
Behauptung wirkte dadurch mit "3 Quellen" belegt, obwohl zwei davon
|
|
derselbe Artikel waren. Die kleinere Nummer gewinnt, die groessere wird
|
|
unter "alias_nrs" vermerkt, damit die Verweise im Text auf sie umgebogen
|
|
werden koennen.
|
|
"""
|
|
behalten: dict[str, dict] = {}
|
|
reihenfolge: list[str] = []
|
|
ohne_adresse = 0
|
|
for s in sources:
|
|
schluessel = _url_schluessel(s.get("url") or "")
|
|
if not schluessel:
|
|
# Eintraege ohne Adresse sind keine pruefbaren Belege. Im Bericht
|
|
# vom 02.08.2026 stand so ein Platzhalter als "12 Quelle" ohne
|
|
# Adresse im Verzeichnis und als "Quelle 1" in der Statistik.
|
|
ohne_adresse += 1
|
|
continue
|
|
if schluessel in behalten:
|
|
behalten[schluessel].setdefault("alias_nrs", []).append(s.get("nr"))
|
|
continue
|
|
behalten[schluessel] = s
|
|
reihenfolge.append(schluessel)
|
|
|
|
ergebnis = [behalten[k] for k in reihenfolge]
|
|
entfernt = len(sources) - len(ergebnis) - ohne_adresse
|
|
if entfernt:
|
|
logger.info("Quellenverzeichnis: %d doppelte Adresse(n) zusammengefuehrt", entfernt)
|
|
if ohne_adresse:
|
|
logger.info("Quellenverzeichnis: %d Eintrag/Eintraege ohne Adresse entfernt", ohne_adresse)
|
|
return ergebnis
|
|
|
|
|
|
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
|
|
# Zusammengefuehrte Dubletten zeigen auf denselben Eintrag, damit ein
|
|
# Verweis auf die entfernte Nummer nicht ins Leere laeuft.
|
|
for alias in kopie.pop("alias_nrs", []) or []:
|
|
if isinstance(alias, int):
|
|
abbildung[alias] = 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)
|
|
# Zeigten zwei Nummern auf dieselbe Adresse, landen ihre Verweise nach der
|
|
# Umschreibung auf derselben Zahl und stehen dann doppelt nebeneinander:
|
|
# "[24][24]". Im Bericht vom 02.08.2026 kam das dreimal vor.
|
|
ergebnis = re.sub(r"(\[(\d{1,5})\])(?:\s*\[\2\])+", r"\1", ergebnis)
|
|
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 "<p><em>Keine Zusammenfassung verfügbar.</em></p>"
|
|
# Basic Markdown -> HTML
|
|
html = text
|
|
# Headlines
|
|
html = re.sub(r'^### (.+)$', r'<h3>\1</h3>', html, flags=re.MULTILINE)
|
|
html = re.sub(r'^## (.+)$', r'<h3>\1</h3>', html, flags=re.MULTILINE)
|
|
# Bold
|
|
html = re.sub(r'\*\*(.+?)\*\*', r'<strong>\1</strong>', html)
|
|
# Links [text](url)
|
|
html = re.sub(r'\[([^\]]+)\]\(([^)]+)\)', r'<a href="\2">\1</a>', html)
|
|
# Bullet lists
|
|
html = re.sub(r'^- (.+)$', r'<li>\1</li>', html, flags=re.MULTILINE)
|
|
html = re.sub(r'(<li>.*</li>\n?)+', lambda m: '<ul>' + m.group(0) + '</ul>', html)
|
|
# Paragraphs
|
|
paragraphs = html.split('\n\n')
|
|
result = []
|
|
for p in paragraphs:
|
|
p = p.strip()
|
|
if not p:
|
|
continue
|
|
if p.startswith('<h') or p.startswith('<ul') or p.startswith('<ol'):
|
|
result.append(p)
|
|
else:
|
|
result.append(f'<p>{p}</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"
|
|
stunde, minute = (time.split(":") + ["0"])[:2]
|
|
sortierwert = (int(yy), int(month), int(day), int(stunde), int(minute))
|
|
result.append((label, body, sortierwert))
|
|
|
|
# Neueste zuerst. Der Auftrag verlangt diese Reihenfolge zwar schon vom
|
|
# Modell, im Bericht vom 02.08.2026 sprang der letzte Eintrag aber von
|
|
# 12:44 zurueck auf 17:47. Im prominentesten Kapitel darf das nicht von
|
|
# der Sorgfalt des Modells abhaengen, deshalb hier deterministisch.
|
|
result.sort(key=lambda e: e[2], reverse=True)
|
|
return [(label, body) for label, body, _ in 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'<strong>\1</strong>', body_html)
|
|
blocks.append(
|
|
'<div class="dev-entry">'
|
|
f'<div class="dev-entry-date">{_html_escape(label)}</div>'
|
|
f'<div class="dev-entry-body">{body_html}</div>'
|
|
'</div>'
|
|
)
|
|
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'<a href="{url}" class="citation" title="{name}">[{num}]</a>'
|
|
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"<li>{_linkify_citations_html(line, src_list)}</li>" for line in lines]
|
|
html = "<ul>\n" + "\n".join(html_lines) + "\n</ul>"
|
|
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 "<ul><li>Kein Lagebild verfügbar. Zusammenfassung kann nicht erstellt werden.</li></ul>"
|
|
|
|
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 = "<ul>\n" + "\n".join(f"<li>{line}</li>" for line in lines if line) + "\n</ul>"
|
|
return html
|
|
except Exception as e:
|
|
logger.error(f"Executive Summary Generierung fehlgeschlagen: {e}")
|
|
return "<ul><li>Zusammenfassung konnte nicht generiert werden.</li></ul>"
|
|
|
|
|
|
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()
|