diff --git a/CLAUDE.md b/CLAUDE.md index f0153db..9d54f0f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -115,6 +115,7 @@ src/: source_suggester.py: "KI-Quellen-Vorschlaege via Haiku + Karteileichen-Heuristik" pdf_ingest.py: "Minutenjob: hochgeladene PDFs einlesen (pdfplumber + OCR-Fallback), uebersetzen, als Pool-Artikel ablegen" org_settings.py: "Key-Value-Einstellungen je Organisation (output_language etc.) mit 60s-Cache" + media_registry.py: "Kanonische Medien-Identitaet je Domain (registrable_domain, namen_je_domain, Aggregator-Liste, Sprache aus der Adresse). Sorgt dafuer, dass ein Verlag im Bericht unter einem Namen gefuehrt und einmal gezaehlt wird" staan_client.py: "staan.ai Such-Client (EU-Umbau Phase 2): Suche + Volltexte (full_content=markdown) ueber den europaeischen Index, Pflicht-Domain-Ausschlussliste je Anfrage, max 10 Ausschluss-Domains" license_service.py: "Lizenz-Pruefung, Credits-Buchung (charge_usage_to_tenant), Periodenwechsel, Budget-Warnung. expire_licenses() existiert, hat aber KEINEN Scheduler-Job" @@ -163,6 +164,7 @@ tests/: test_eu_belegsuche.py: "Gezielte Belegsuche ueber staan, Planung, Grenzen, Entdopplung, Ausfallverhalten" test_qc_und_runden.py: "Absicherung der Duplikatpruefung und Abbruch der Suchrunden bei erreichter Treffergrenze" test_mehrfachantwort.py: "Antworten, in denen sich das Modell selbst korrigiert und mehrere Fassungen liefert, die ausfuehrlichste muss gewinnen" + test_quellenausgabe.py: "Ausgabetreue des Quellenverzeichnisses: Nummern aus dem Lagebild statt Zeilennummern, keine Kappung, ein Verlag ein Name, Statistik deckungsgleich mit dem Verzeichnis" ``` ## Architektur diff --git a/src/report_generator.py b/src/report_generator.py index af4a7ca..55cf49f 100644 --- a/src/report_generator.py +++ b/src/report_generator.py @@ -19,6 +19,7 @@ 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") @@ -68,29 +69,121 @@ def _get_logo_base64() -> str: def _prepare_sources(incident: dict) -> list: - """Quellenverzeichnis aus sources_json parsen.""" + """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: - return json.loads(raw) if isinstance(raw, str) else raw + 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) + domain = media_registry.registrable_domain(s.get("url") or "") + if domain: + kopie["name"] = namen.get(domain) or kopie.get("name") or media_registry.name_aus_domain(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 _prepare_source_stats(articles: list) -> list: - """Quellenstatistik: Artikel pro Quelle + Sprachen.""" - source_map = defaultdict(lambda: {"count": 0, "langs": set()}) - for art in articles: - name = art.get("source") or "Unbekannt" - source_map[name]["count"] += 1 - source_map[name]["langs"].add((art.get("language") or "de").upper()) +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) + schluessel = domain or (s.get("name") or "Unbekannt") + eintrag = stats_map.setdefault( + schluessel, + {"name": s.get("name") or media_registry.name_aus_domain(domain), "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 name, data in sorted(source_map.items(), key=lambda x: -x[1]["count"]): - stats.append({"name": name, "count": data["count"], "languages": ", ".join(sorted(data["langs"]))}) + 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) @@ -831,9 +924,13 @@ async def generate_pdf( _markdown_to_html(bericht_summary), all_sources ), lagebild_timestamp=(incident.get("updated_at") or "")[:16].replace("T", " "), - sources=_prepare_sources(incident)[:30] if scope == "report" else _prepare_sources(incident), + # 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(articles)[:20] if scope == "report" else _prepare_source_stats(articles), + 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, @@ -1043,7 +1140,7 @@ async def generate_docx( if "quellen" in sections: # --- Quellenstatistik --- - source_stats = _prepare_source_stats(articles) + 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) @@ -1051,7 +1148,7 @@ async def generate_docx( table.alignment = WD_TABLE_ALIGNMENT.CENTER hdr = table.rows[0].cells hdr[0].text = "Quelle" - hdr[1].text = "Artikel" + hdr[1].text = "Belege" hdr[2].text = "Sprache" for cell in hdr: for p in cell.paragraphs: @@ -1062,6 +1159,40 @@ async def generate_docx( 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 --- diff --git a/src/report_templates/report.html b/src/report_templates/report.html index cf32d7a..6a1aad5 100644 --- a/src/report_templates/report.html +++ b/src/report_templates/report.html @@ -159,20 +159,21 @@ tr:nth-child(even) { background: #f8f9fa; } {% if source_stats %}

Quellenstatistik

- + {% for stat in source_stats %} {% endfor %}
QuelleArtikelSprache
QuelleBelegeSprache
{{ stat.name }}{{ stat.count }}{{ stat.languages }}
+ {% if source_stats_note %}

{{ source_stats_note }}

{% endif %} {% endif %}

Quellen

{% for src in sources %} - + {% endfor %}
#QuelleURL
{{ loop.index }}{{ src.name or src.title or '' }}{{ src.url or '' }}
{{ src.nr if src.nr is not none else loop.index }}{{ src.name or src.title or '' }}{{ src.url or '' }}
diff --git a/src/services/media_registry.py b/src/services/media_registry.py new file mode 100644 index 0000000..a56bee0 --- /dev/null +++ b/src/services/media_registry.py @@ -0,0 +1,180 @@ +"""Kanonische Medien-Identitaet je Domain. + +Hintergrund: Quellennamen kommen aus drei Richtungen in den Bestand, aus dem +Feed-Namen (RSS), aus dem Hostnamen (staan-Treffer im EU-Weg) und aus der freien +Benennung durch das Modell in der Abschlussauswahl. Dadurch fuehrt derselbe +Verlag mehrere Identitaeten, etwa "Guardian UK", "Guardian World" und +"The Guardian" oder "tagesschau" und "tagesschau.de". Jede Aussage ueber +Quellenvielfalt oder Belegtiefe wird damit falsch, weil dieselbe Redaktion +mehrfach gezaehlt wird. + +Dieses Modul bestimmt die Identitaet ueber die Domain, nicht ueber den Namen. +Der Anzeigename wird datengetrieben gewaehlt: der haeufigste Name, der fuer +diese Domain vorkommt. Die Tabelle unten deckt nur die Faelle ab, in denen gar +kein brauchbarer Name vorliegt oder der Hostname als Name durchschlaegt. +""" +from collections import Counter +from urllib.parse import urlparse + +# Zusammengesetzte Endungen, bei denen die registrierbare Domain drei Labels +# hat (bbc.co.uk statt co.uk). +_MEHRTEILIGE_ENDUNGEN = frozenset({ + "co.uk", "org.uk", "ac.uk", "gov.uk", "me.uk", "net.uk", + "com.au", "net.au", "org.au", "gov.au", + "co.jp", "or.jp", "ne.jp", "go.jp", + "com.br", "com.mx", "com.ar", "com.tr", "com.cn", "com.hk", + "co.za", "co.nz", "co.kr", "co.in", "co.il", + "or.at", "co.at", "gv.at", +}) + +# Praefixe, die dieselbe Redaktion nur anders ausliefern. +_HOST_PRAEFIXE = ("www.", "m.", "amp.", "rss.", "feeds.", "news.", "english.", "de.", "en.") + +# Anzeigename, wenn kein brauchbarer Name mitgeliefert wurde. Bewusst kurz +# gehalten, die Liste ist keine Pflegepflicht: unbekannte Domains bekommen +# ihren ersten Domain-Bestandteil in Grossschreibung. +_NAMEN_JE_DOMAIN = { + "tagesschau.de": "tagesschau", + "zdfheute.de": "ZDF heute", + "zdf.de": "ZDF heute", + "faz.net": "FAZ", + "sueddeutsche.de": "Süddeutsche Zeitung", + "spiegel.de": "Spiegel", + "zeit.de": "Zeit", + "welt.de": "Welt", + "n-tv.de": "n-tv", + "dw.com": "Deutsche Welle", + "deutschlandfunk.de": "Deutschlandfunk", + "bundesregierung.de": "Bundesregierung", + "nzz.ch": "NZZ", + "srf.ch": "SRF", + "kurier.at": "Kurier", + "diepresse.com": "Die Presse", + "theguardian.com": "The Guardian", + "nytimes.com": "New York Times", + "ft.com": "Financial Times", + "bbc.co.uk": "BBC", + "bbc.com": "BBC", + "aljazeera.com": "Al Jazeera", + "france24.com": "France24", + "reuters.com": "Reuters", + "apnews.com": "AP News", + "elpais.com": "El País", + "elmundo.es": "El Mundo", + "abc.es": "ABC", + "infobae.com": "Infobae", + "politico.eu": "Politico Europe", + "euronews.com": "Euronews", +} + +# Weiterleitungs- und Sammelportale. Sie sind kein eigenstaendiges Medium und +# duerfen in keiner Vielfaltsstatistik als Redaktion durchgehen. +AGGREGATOR_DOMAINS = frozenset({ + "news.google.com", "google.com", "msn.com", "headtopics.com", + "newsbreak.com", "yahoo.com", "flipboard.com", "smartnews.com", + "reutersconnect.com", "pressreader.com", "onvista.de", "finanznachrichten.de", + "boersennews.de", "aol.com", +}) + + +def registrable_domain(url: str) -> str: + """Registrierbare Domain einer URL, klein geschrieben. + + 'https://www.theguardian.com/world/...' -> 'theguardian.com' + 'https://www.bbc.co.uk/news/...' -> 'bbc.co.uk' + Leere oder unparsebare Eingaben ergeben einen leeren String. + """ + host = "" + text = (url or "").strip() + if not text: + return "" + try: + host = (urlparse(text if "//" in text else "//" + text).hostname or "").lower() + except ValueError: + return "" + if not host: + return "" + for praefix in _HOST_PRAEFIXE: + if host.startswith(praefix) and host.count(".") >= 2: + host = host[len(praefix):] + teile = host.split(".") + if len(teile) <= 2: + return host + if ".".join(teile[-2:]) in _MEHRTEILIGE_ENDUNGEN: + return ".".join(teile[-3:]) + return ".".join(teile[-2:]) + + +def ist_aggregator(url: str) -> bool: + """True, wenn die URL auf ein Weiterleitungs- oder Sammelportal zeigt.""" + return registrable_domain(url) in AGGREGATOR_DOMAINS + + +def name_aus_domain(domain: str) -> str: + """Anzeigename allein aus der Domain, wenn kein Name vorliegt.""" + if not domain: + return "Unbekannt" + if domain in _NAMEN_JE_DOMAIN: + return _NAMEN_JE_DOMAIN[domain] + erstes = domain.split(".")[0] + return erstes if any(c.isupper() for c in erstes) else erstes.capitalize() + + +def _ist_hostname_als_name(name: str, domain: str) -> bool: + """True, wenn der Name nur der Hostname ist ('www.france24.com').""" + schlicht = (name or "").strip().lower() + return bool(schlicht) and (schlicht == domain or schlicht.endswith(domain)) + + +def namen_je_domain(eintraege: list[tuple[str, str]]) -> dict[str, str]: + """Waehlt je Domain einen Anzeigenamen aus allen gelieferten Varianten. + + eintraege ist eine Liste aus (url, name). Gewinner ist der haeufigste Name, + bei Gleichstand der kuerzeste. Hostnamen und leere Namen zaehlen nicht mit, + fuer sie greift die Tabelle beziehungsweise die Domain selbst. + """ + kandidaten: dict[str, Counter] = {} + for url, name in eintraege: + domain = registrable_domain(url) + if not domain: + continue + zaehler = kandidaten.setdefault(domain, Counter()) + sauber = (name or "").strip() + if sauber and not _ist_hostname_als_name(sauber, domain): + zaehler[sauber] += 1 + + ergebnis: dict[str, str] = {} + for domain, zaehler in kandidaten.items(): + if not zaehler: + ergebnis[domain] = name_aus_domain(domain) + continue + hoechste = max(zaehler.values()) + beste = sorted((n for n, c in zaehler.items() if c == hoechste), key=lambda n: (len(n), n)) + ergebnis[domain] = beste[0] + return ergebnis + + +def sprache_aus_url(url: str) -> str: + """Sprachkennung, die sich eindeutig aus der Adresse ergibt, sonst ''. + + Deckt die Faelle ab, in denen das Modell die Sprache falsch angibt, etwa + 'english.elpais.com' als DE. Nur eindeutige Muster, keine Raterei. + """ + text = (url or "").strip().lower() + if not text: + return "" + try: + zerlegt = urlparse(text if "//" in text else "//" + text) + except ValueError: + return "" + host = zerlegt.hostname or "" + pfad = zerlegt.path or "" + if host.startswith("english.") or pfad.startswith("/en/") or "/english/" in pfad: + return "EN" + if host.startswith("de.") or pfad.startswith("/de/"): + return "DE" + if host.startswith("fr.") or pfad.startswith("/fr/"): + return "FR" + if host.startswith("es.") or pfad.startswith("/es/"): + return "ES" + return "" diff --git a/tests/test_quellenausgabe.py b/tests/test_quellenausgabe.py new file mode 100644 index 0000000..3a474b5 --- /dev/null +++ b/tests/test_quellenausgabe.py @@ -0,0 +1,184 @@ +"""Testet die Ausgabetreue des Quellenverzeichnisses. + +Hintergrund ist der Vergleich zweier Berichte zur selben Lage am 01.08.2026 +(Ceuta). Beide Fassungen verwiesen im Text auf Quellennummern, die im +gedruckten Verzeichnis nicht auftauchten. Ursache war nicht die Analyse, in der +Datenbank passten Text und Liste zusammen, sondern die Ausgabe: Das Template +druckte die laufende Zeilennummer statt der Quellennummer aus dem Lagebild. +Sobald die Nummern Luecken haben, verschiebt sich damit jeder Beleg. + +Laeuft ohne Netzzugriff, ohne Kosten und ohne Datenbank. + +Aufruf aus dem Projektstamm: + venv/bin/python tests/test_quellenausgabe.py +""" +import json +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src")) + +from jinja2 import Environment, FileSystemLoader # noqa: E402 + +import report_generator as rg # noqa: E402 +from services import media_registry as mr # noqa: E402 + +ok = 0 +fail = 0 + + +def pruefe(name, bedingung, extra=""): + global ok, fail + if bedingung: + ok += 1 + print(" OK " + name) + else: + fail += 1 + print(" FEHL " + name + " " + str(extra)) + + +# --------------------------------------------------------------------------- +# A. Domain-Erkennung +# --------------------------------------------------------------------------- +print("\nA. Registrierbare Domain") +pruefe("A1 www wird abgeschnitten", + mr.registrable_domain("https://www.theguardian.com/world/x") == "theguardian.com", + mr.registrable_domain("https://www.theguardian.com/world/x")) +pruefe("A2 mehrteilige Endung bleibt erhalten", + mr.registrable_domain("https://www.bbc.co.uk/news/articles/y") == "bbc.co.uk", + mr.registrable_domain("https://www.bbc.co.uk/news/articles/y")) +pruefe("A3 Sprachsubdomain zaehlt zur selben Domain", + mr.registrable_domain("https://english.elpais.com/spain/a.html") == "elpais.com", + mr.registrable_domain("https://english.elpais.com/spain/a.html")) +pruefe("A4 leere Eingabe bleibt leer", mr.registrable_domain("") == "") +pruefe("A5 Weiterleitungsportal wird erkannt", + mr.ist_aggregator("https://news.google.com/rss/articles/CBMi123?oc=5")) +pruefe("A6 echtes Medium ist kein Aggregator", + not mr.ist_aggregator("https://www.tagesschau.de/ausland/europa/x-100.html")) + + +# --------------------------------------------------------------------------- +# B. Ein Verlag, ein Name +# --------------------------------------------------------------------------- +print("\nB. Namensvereinheitlichung") +eintraege = [ + ("https://www.theguardian.com/world/a", "Guardian World"), + ("https://www.theguardian.com/world/b", "Guardian UK"), + ("https://www.theguardian.com/live/c", "The Guardian"), + ("https://www.theguardian.com/live/d", "The Guardian"), + ("https://www.france24.com/en/x", "www.france24.com"), +] +namen = mr.namen_je_domain(eintraege) +pruefe("B1 drei Guardian-Varianten ergeben einen Namen", + namen.get("theguardian.com") == "The Guardian", namen.get("theguardian.com")) +pruefe("B2 Hostname als Name faellt auf die Tabelle zurueck", + namen.get("france24.com") == "France24", namen.get("france24.com")) +pruefe("B3 unbekannte Domain bekommt einen lesbaren Namen", + mr.name_aus_domain("beispielzeitung.de") == "Beispielzeitung", + mr.name_aus_domain("beispielzeitung.de")) + + +# --------------------------------------------------------------------------- +# C. Quellenverzeichnis behaelt die Nummern aus dem Lagebild +# --------------------------------------------------------------------------- +print("\nC. Quellenaufbereitung") +# Nachgebaut nach Lage 53 (EU-Fassung): 25 Eintraege, aber Nummern bis 28. +quellen_roh = [{"nr": n, "name": f"Medium {n}", "url": f"https://medium{n}.de/artikel"} + for n in [1, 2, 3, 20, 22, 24, 26, 27, 28]] +incident = {"sources_json": json.dumps(quellen_roh, ensure_ascii=False)} +sources = rg._prepare_sources(incident) +nummern = [s["nr"] for s in sources] +pruefe("C1 Luecken in der Nummerierung bleiben erhalten", + nummern == [1, 2, 3, 20, 22, 24, 26, 27, 28], nummern) +pruefe("C2 hoechste Nummer ueberlebt die Aufbereitung", max(nummern) == 28, max(nummern)) + +# Buchstaben-Suffixe und kaputte Nummern duerfen nichts umwerfen +gemischt = {"sources_json": json.dumps([ + {"nr": "7a", "name": "A", "url": "https://a.de/1"}, + {"nr": None, "name": "B", "url": "https://b.de/1"}, + {"nr": 3, "name": "C", "url": "https://c.de/1"}, +], ensure_ascii=False)} +gemischte_nrn = [s["nr"] for s in rg._prepare_sources(gemischt)] +pruefe("C3 Suffix wird zur Zahl, fehlende Nummer bekommt die Position", + all(isinstance(n, int) for n in gemischte_nrn) and 7 in gemischte_nrn, + gemischte_nrn) + +# Namen im Verzeichnis vereinheitlicht +mehrfach = {"sources_json": json.dumps([ + {"nr": 1, "name": "Guardian World", "url": "https://www.theguardian.com/a"}, + {"nr": 2, "name": "The Guardian", "url": "https://www.theguardian.com/b"}, + {"nr": 3, "name": "The Guardian", "url": "https://www.theguardian.com/c"}, +], ensure_ascii=False)} +namen_liste = {s["name"] for s in rg._prepare_sources(mehrfach)} +pruefe("C4 Verzeichnis fuehrt den Verlag unter einem Namen", + namen_liste == {"The Guardian"}, namen_liste) + + +# --------------------------------------------------------------------------- +# D. Statistik und Verzeichnis widersprechen sich nicht mehr +# --------------------------------------------------------------------------- +print("\nD. Quellenstatistik") +artikel = [ + {"source": "France24", "source_url": "https://www.france24.com/en/1", "language": "en"}, + {"source": "France24", "source_url": "https://www.france24.com/en/2", "language": "en"}, + {"source": "El País", "source_url": "https://english.elpais.com/spain/1.html", "language": "de"}, + {"source": "tagesschau", "source_url": "https://www.tagesschau.de/1.html", "language": "de"}, + {"source": "n-tv", "source_url": "https://www.n-tv.de/1.html", "language": "de"}, +] +zitiert = rg._prepare_sources({"sources_json": json.dumps([ + {"nr": 1, "name": "France24", "url": "https://www.france24.com/en/1"}, + {"nr": 2, "name": "France24", "url": "https://www.france24.com/en/2"}, + {"nr": 3, "name": "El País", "url": "https://english.elpais.com/spain/1.html"}, + {"nr": 4, "name": "tagesschau", "url": "https://www.tagesschau.de/1.html"}, +], ensure_ascii=False)}) +stats = rg._prepare_source_stats(zitiert, artikel) +summe = sum(s["count"] for s in stats) +pruefe("D1 Summe der Belege entspricht dem Verzeichnis", summe == len(zitiert), (summe, len(zitiert))) +pruefe("D2 zwei France24-Belege ergeben eine Zeile", + [s for s in stats if s["name"] == "France24"][0]["count"] == 2, stats) +elpais = [s for s in stats if "Pa" in s["name"] or "País" in s["name"]] +pruefe("D3 englische Ausgabe wird nicht als DE gefuehrt", + elpais and elpais[0]["languages"] == "EN", elpais) +notiz = rg._source_stats_note(zitiert, artikel) +pruefe("D4 Hinweis nennt zitierte Belege und ausgewertete Meldungen", + "4 im Lagebild zitierten Belege" in notiz and "5 Meldungen" in notiz, notiz) +pruefe("D5 Hinweis nutzt echte Umlaute", "zaehlt" not in notiz and "zählt" in notiz, notiz) + + +# --------------------------------------------------------------------------- +# E. Das Template druckt die Quellennummer, nicht die Zeilennummer +# --------------------------------------------------------------------------- +print("\nE. Ausgabe im Bericht") +env = Environment(loader=FileSystemLoader(str(rg.TEMPLATE_DIR))) +template = env.get_template("report.html") +html = template.render( + incident={"title": "Testlage", "type": "adhoc"}, + incident_type_label="Live-Monitoring", + report_date="01.08.2026, 20:39 Uhr", + creator="test@aegis-sight.de", + logo_base64="", + executive_summary="", + zusammenfassung_title="Neueste Entwicklungen", + sections={"quellen"}, + scope="report", + lagebild_html="", + lagebild_timestamp="", + sources=sources, + fact_checks=[], + source_stats=[{"name": "Medium 28", "count": 1, "languages": "DE"}], + source_stats_note="Hinweistext zur Prüfung", + timeline=[], + articles=[], + meta={}, + include_branding=True, +) +zeilen = [z for z in html.splitlines() if "medium28.de" in z] +pruefe("E1 letzter Eintrag wird als 28 gedruckt, nicht als 9", + zeilen and ">28<" in zeilen[0] and ">9<" not in zeilen[0], zeilen[:1]) +pruefe("E2 Statistikspalte heisst Belege", "Belege" in html) +pruefe("E3 Hinweis unter der Statistik erscheint", "Hinweistext zur Prüfung" in html) +pruefe("E4 alle Eintraege erscheinen, nichts wird gekappt", + all(f"medium{n}.de" in html for n in [1, 2, 3, 20, 22, 24, 26, 27, 28])) + +print("\nErgebnis: " + str(ok) + " bestanden, " + str(fail) + " fehlgeschlagen") +sys.exit(1 if fail else 0)