AegisSight-Monitor/tests/test_faktencheck_belegtiefe.py

"""Testet die Belegtiefe des Faktenchecks.

Grundlage sind die drei Fehler, die der Vergleich der Ceuta-Berichte vom
01.08.2026 in der EU-Fassung zeigte:

1. Eine sachlich zutreffende Zahlenspanne stand als WIDERLEGT, weil der Status
   "contradicted" im Auftrag Quellendivergenz meint, in der Anzeige aber als
   Widerlegung ausgegeben wurde.
2. "Der Ausloeser ist weiterhin unklar" wurde von der Nachhak-Runde auf
   BESTAETIGT hochgestuft, weil "developing" die niedrigste Prioritaet hat.
3. Eine Behauptung stand als BESTAETIGT mit einer Quelle, deren Evidenz
   woertlich mit "Nur durch Al Jazeera berichtet" begann.

Dazu kommt die Zaehlung: sechs France24-Treffer sind ein Medienhaus, nicht
sechs Belege.

Laeuft ohne Netzzugriff, ohne Kosten und ohne Datenbank.

Aufruf aus dem Projektstamm:
    venv/bin/python tests/test_faktencheck_belegtiefe.py
"""
import asyncio
import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src"))

import agents.factchecker as fc  # noqa: E402
from agents.claude_client import ClaudeUsage  # noqa: E402
import report_generator as rg  # 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. Zaehlung nach Medienhaus
# ---------------------------------------------------------------------------
print("\nA. Belege zaehlen")
france24 = [
    "https://www.france24.com/en/ceuta-spain-slams-selfish-eu-partners",
    "https://www.france24.com/en/video/20260801-ceuta-enclave-almost-back-to-normal-1",
    "https://www.france24.com/en/video/20260801-ceuta-most-migrants-have-returned-1",
    "https://www.france24.com/en/ceuta-2026-dwarfs-2021-migration-crisis",
]
pruefe("A1 vier France24-Fundstellen sind ein Haus",
       fc.zaehle_medienhaeuser(france24) == 1, fc.zaehle_medienhaeuser(france24))
gemischt = france24 + [
    "https://www.tagesschau.de/ausland/europa/ceuta-100.html",
    "https://www.theguardian.com/world/2026/aug/01/spain-ceuta",
    "https://www.bbc.co.uk/news/articles/abc",
]
pruefe("A2 drei weitere Haeuser ergeben vier",
       fc.zaehle_medienhaeuser(gemischt) == 4, fc.zaehle_medienhaeuser(gemischt))
redirects = [
    "https://news.google.com/rss/articles/AAA?oc=5",
    "https://news.google.com/rss/articles/BBB?oc=5",
]
pruefe("A3 zwei Weiterleitungen bleiben zwei, sie koennen zwei Zeitungen sein",
       fc.zaehle_medienhaeuser(redirects) == 2, fc.zaehle_medienhaeuser(redirects))
pruefe("A4 Satzzeichen am Adressende stoeren nicht",
       fc.zaehle_medienhaeuser(["https://www.zeit.de/a.html)."]) == 1)
pruefe("A5 leere Liste ergibt null", fc.zaehle_medienhaeuser([]) == 0)

# Regressionsschutz: das alte Muster zaehlte nur das Schema und lieferte
# deshalb fuer jede Evidenz 1 oder 2, egal wie viele Quellen darin standen.
evidenz = ("Bestätigt durch: tagesschau (https://www.tagesschau.de/a.html), "
           "Guardian (https://www.theguardian.com/b), BBC (https://www.bbc.co.uk/c)")
pruefe("A6 Adressen werden vollstaendig erkannt, nicht nur das Schema",
       fc.zaehle_medienhaeuser(fc._URL_RE.findall(evidenz)) == 3,
       fc._URL_RE.findall(evidenz))


# ---------------------------------------------------------------------------
# B. Nachbedingung: Status gegen Beleglage
# ---------------------------------------------------------------------------
print("\nB. Belegtiefe pruefen")
einzelquelle = {
    "claim": "Spaniens Außenminister versicherte, die Integrität sei garantiert",
    "status": "confirmed",
    "sources_count": 1,
    "evidence": "Nur durch Al Jazeera berichtet (https://www.aljazeera.com/news/x)",
}
fc.pruefe_belegtiefe(einzelquelle)
pruefe("B1 bestaetigt mit einer Quelle wird herabgestuft",
       einzelquelle["status"] == "unconfirmed", einzelquelle["status"])

zwei = {"claim": "x", "status": "confirmed", "sources_count": 2, "evidence": ""}
fc.pruefe_belegtiefe(zwei)
pruefe("B2 zwei Haeuser tragen eine Bestaetigung", zwei["status"] == "confirmed")

research_knapp = {"claim": "x", "status": "established", "sources_count": 2, "evidence": ""}
fc.pruefe_belegtiefe(research_knapp)
pruefe("B3 gesichert verlangt drei Haeuser", research_knapp["status"] == "unverified",
       research_knapp["status"])

primaer = {
    "claim": "Weber kritisiert Spanien",
    "status": "confirmed",
    "sources_count": 1,
    "evidence_type": "primary",
    "evidence": "O-Ton im ZDF-Interview (https://www.zdfheute.de/politik/x.html)",
}
fc.pruefe_belegtiefe(primaer)
pruefe("B4 Primaerbeleg traegt allein", primaer["status"] == "confirmed", primaer["status"])

ohne_zahl = {
    "claim": "x", "status": "confirmed",
    "evidence": "a (https://www.zeit.de/1), b (https://www.faz.net/2)",
}
fc.pruefe_belegtiefe(ohne_zahl)
pruefe("B5 fehlende Belegzahl wird aus der Evidenz bestimmt",
       ohne_zahl["sources_count"] == 2 and ohne_zahl["status"] == "confirmed", ohne_zahl)

widerspruch = {"claim": "x", "status": "contradicted", "sources_count": 1, "evidence": ""}
fc.pruefe_belegtiefe(widerspruch)
pruefe("B6 Widerspruchs-Status wird nicht angetastet", widerspruch["status"] == "contradicted")


# ---------------------------------------------------------------------------
# C. Anzeige: Widerspruch ist keine Widerlegung
# ---------------------------------------------------------------------------
print("\nC. Statusbezeichnungen")
labels = rg._fc_labels("de")
pruefe("C1 contradicted heisst nicht mehr Widerlegt",
       labels["contradicted"] == "Widersprüchlich", labels["contradicted"])
pruefe("C2 Widerlegt ist dem eigenen Status vorbehalten",
       labels["false"] == "Widerlegt", labels.get("false"))
pruefe("C3 englische Fassung getrennt",
       rg._fc_labels("en")["contradicted"] == "Conflicting", rg._fc_labels("en")["contradicted"])
pruefe("C4 false hat eine Prioritaet wie andere belegte Befunde",
       fc.STATUS_PRIORITY.get("false") == 4, fc.STATUS_PRIORITY.get("false"))

aufbereitet = rg._prepare_fact_checks([
    {"claim": "a", "status": "contradicted"}, {"claim": "b", "status": "false"},
], "de")
pruefe("C5 Bericht beschriftet beide Status getrennt",
       [f["status_label"] for f in aufbereitet] == ["Widersprüchlich", "Widerlegt"],
       [f["status_label"] for f in aufbereitet])


# ---------------------------------------------------------------------------
# D. Bewertungsregeln liegen jedem Auftrag bei
# ---------------------------------------------------------------------------
print("\nD. Auftragsregeln")
regeln = fc.bewertungsregeln()
for stichwort, name in (
    ("Medienhäuser", "D1 Unabhaengigkeit ueber Medienhaeuser"),
    ("news.google.com", "D2 Weiterleitungen ausgeschlossen"),
    ("Erhebungsstellen", "D3 Zahlenspannen sind kein Widerspruch"),
    ("primary", "D4 Primaerbeleg definiert"),
    ("Kenntnisstand", "D5 keine Aussagen ueber den Kenntnisstand"),
):
    pruefe(name, stichwort in regeln)
umschreibungen = ["fuer", "muessen", "koennen", "haeuser", "zaehlt", "Widerspruechlich",
                  "Auftraege", "gehoeren", "unabhaengig", "genuegt"]
gefunden = [u for u in umschreibungen if u.lower() in regeln.lower()]
pruefe("D6 Regeln nutzen echte Umlaute statt Umschreibungen", not gefunden, gefunden)


# ---------------------------------------------------------------------------
# E. Nachhak-Runde
# ---------------------------------------------------------------------------
print("\nE. Nachhak-Runde")


class FakeSuche:
    def __init__(self):
        self.block = "\n[S1] Beleg | zeit.de | https://www.zeit.de/a"
        self.usage = ClaudeUsage()
        self.queries = ["q"]
        self.quellen = []


async def fake_plan(**kwargs):
    return FakeSuche()


antwort = {"text": "[]"}


async def fake_call(prompt, **kwargs):
    return antwort["text"], ClaudeUsage()


import agents.eu_researcher as eur  # noqa: E402
import agents.claude_client as cc  # noqa: E402
eur.plan_verification_searches = fake_plan
cc.call_claude = fake_call

agent = fc.FactCheckerAgent()

# Fall 1: Die Nachrunde will eine Unklarheit bestaetigen.
unklar = {
    "claim": "Der Auslöser des massenhaften Grenzübertritts ist weiterhin unklar",
    "status": "developing", "sources_count": 4,
    "evidence": "Mehrere Quellen berichten über Unklarheit (https://www.bbc.co.uk/a)",
}
antwort["text"] = ('[{"claim": "Der Auslöser des massenhaften Grenzübertritts ist weiterhin unklar",'
                   ' "status": "confirmed", "evidence": "Vier Quellen nennen die Unklarheit"}]')
facts, _ = asyncio.run(agent._eu_nachhaken(
    facts=[unklar], offene=[unklar], title="Ceuta", output_language="Deutsch",
    incident_type="adhoc", avoid_queries=[],
))
pruefe("E1 eine Unklarheit wird nicht bestaetigt", facts[0]["status"] == "developing", facts[0]["status"])

# Fall 2: echte Hochstufung, aber nur eine Quelle im Beleg
duenn = {
    "claim": "Spaniens Außenminister versicherte, die Integrität sei garantiert",
    "status": "unconfirmed", "sources_count": 1,
    "evidence": "Nur durch Al Jazeera berichtet (https://www.aljazeera.com/a)",
}
antwort["text"] = ('[{"claim": "Spaniens Außenminister versicherte, die Integrität sei garantiert",'
                   ' "status": "confirmed", "evidence": "Auch hier (https://www.aljazeera.com/b)"}]')
facts, _ = asyncio.run(agent._eu_nachhaken(
    facts=[duenn], offene=[duenn], title="Ceuta", output_language="Deutsch",
    incident_type="adhoc", avoid_queries=[],
))
pruefe("E2 zweiter Treffer desselben Hauses traegt keine Bestaetigung",
       facts[0]["status"] == "unconfirmed", facts[0]["status"])

# Fall 3: gedeckte Hochstufung durch ein zweites Haus
tragfaehig = {
    "claim": "Italien führt Grenzkontrollen für Flüge und Schiffe aus Spanien ein",
    "status": "unconfirmed", "sources_count": 1,
    "evidence": "Kurier (https://www.kurier.at/a)",
}
antwort["text"] = ('[{"claim": "Italien führt Grenzkontrollen für Flüge und Schiffe aus Spanien ein",'
                   ' "status": "confirmed", "evidence": "Bestätigt (https://www.ansa.it/b)"}]')
facts, _ = asyncio.run(agent._eu_nachhaken(
    facts=[tragfaehig], offene=[tragfaehig], title="Ceuta", output_language="Deutsch",
    incident_type="adhoc", avoid_queries=[],
))
pruefe("E3 zweites unabhaengiges Haus stuft hoch",
       facts[0]["status"] == "confirmed", facts[0]["status"])
pruefe("E4 Belegzahl folgt der neuen Evidenz",
       facts[0]["sources_count"] == 2, facts[0]["sources_count"])

print("\nErgebnis: " + str(ok) + " bestanden, " + str(fail) + " fehlgeschlagen")
sys.exit(1 if fail else 0)