From cae3dbb98524be492ecfec6b4abc8406422c5356 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dieter=20Schl=C3=BCter?= Date: Sat, 13 Jun 2026 00:04:33 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20KI-gest=C3=BCtzte=20Inhaltsanalyse=20vi?= =?UTF-8?q?a=20OpenRouter=20(hash-gegated)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Optionale semantische Prüfung von Text und Bildern (inkl. OCR) auf problematische Inhalte ohne Link-Signal: Pornografie, Propaganda, diffamierende/strafbare Texte, versteckter Spam, themenfremde Werbung, widersprüchliche Aussagen. Bewertet die thematische Passung zum deklarierten site_context — eingestreute Heimat-Begriffe täuschen die Erkennung nicht. Kern: - scanner/ai_analyzer.py: run_ai_analysis + score_ai_findings. Hash-Gate über data/ai_ledger.json → unveränderte Inhalte = Cache- Treffer = kein API-Call. Nur neue/geänderte Inhalte kosten etwas. - Modell-Kette mit zweifacher Eskalation (Stufe 1+2 free, Stufe 3 günstig bezahlt); eskaliert bei Fehler ODER Timeout (attempt_timeout). Erfolgreiches Modell wird im Ledger vermerkt. - KI-Funde sind auf GELB gedeckelt — ROT bleibt harten Integritäts- Signalen vorbehalten. Graceful degradation: ohne Key/bei Fehler wird übersprungen, Scan läuft unverändert weiter. Integration: - baseline.py: load/save_ai_ledger, dismiss_ai_entries. - config.py: ai_analysis-Block + ai_* Scoring-Schlüssel. - __main__.py: Einhängung in cmd_scan/cmd_check, ai-dismiss-Subcommand, approve quittiert zugehörige Funde, status-Anzeige, diff-only + report. - alerter.py + __main__.py: beanstandete Dateien erscheinen mit URL, Begründung und Quittier-Fingerprint in E-Mail UND Markdown-Report. - plain.py: laienverständliche KI-Sätze. API-Key nur aus Umgebungsvariable (OPENROUTER_API_KEY). Audio/Video als abschaltbare Hooks vorbereitet (default aus). Modelle live gegen OpenRouter verifiziert; Demo auf bredelar.info zeigte korrekte Erkennung (echter Inhalt clean, eingeschleuster Casino-Spam mit Heimat-Begriffen als hidden_spam erkannt). Tests: 227 grün (+26 für ai_analyzer, +1 für E-Mail-KI-Abschnitt). Co-Authored-By: Claude Opus 4.8 --- BEDIENUNGSANLEITUNG.md | 129 +++++++++++ README.md | 79 +++++++ meine-seite.de/config.yaml | 39 ++++ scanner/__main__.py | 120 +++++++++- scanner/ai_analyzer.py | 452 +++++++++++++++++++++++++++++++++++++ scanner/alerter.py | 18 +- scanner/baseline.py | 49 ++++ scanner/config.py | 51 +++++ scanner/plain.py | 33 +++ tests/test_ai_analyzer.py | 321 ++++++++++++++++++++++++++ tests/test_alerter.py | 17 ++ 11 files changed, 1301 insertions(+), 7 deletions(-) create mode 100644 scanner/ai_analyzer.py create mode 100644 tests/test_ai_analyzer.py diff --git a/BEDIENUNGSANLEITUNG.md b/BEDIENUNGSANLEITUNG.md index 2ae4b37..3422bd0 100644 --- a/BEDIENUNGSANLEITUNG.md +++ b/BEDIENUNGSANLEITUNG.md @@ -71,6 +71,7 @@ Alles Weitere in dieser Anleitung richtet sich an **Fortgeschrittene** und an 17. [Cloaking-Erkennung](#17-cloaking-erkennung) 18. [Externe Links auf Erreichbarkeit prüfen](#18-externe-links-auf-erreichbarkeit-prüfen) 19. [Binärdateien und Assets prüfen](#19-binärdateien-und-assets-prüfen) +20. [KI-gestützte Inhaltsanalyse](#20-ki-gestützte-inhaltsanalyse) --- @@ -220,6 +221,7 @@ Globale Optionen: `--config path/to/config.yaml`, `--verbose` (`-v`). | `check-assets` | Binärdateien (JS/CSS/Bilder) auf Inhaltsänderung prüfen | | `approve-assets` | Aktuelle Asset-Hashes als neue Baseline setzen | | `test-alert` | Test-E-Mail senden ohne echten Scan — prüft SMTP-Konfiguration | +| `ai-dismiss --all` | KI-Funde als geprüft/akzeptiert quittieren (Fehlalarm) | ### Beispiele @@ -630,6 +632,8 @@ angewendet — falsch-positive Alarme durch dynamische CMS-Inhalte werden so unt | Kaputte **externe** Links (4xx/5xx) | Ja, mit `check-ext-links` | Kein Score, nur Report | | UA-basiertes Cloaking (Googlebot sieht anderen Inhalt) | Ja, mit `cloak-check` | Score 40–60 | | Veränderungen in JS/CSS/Bildern (Inhalt) | Ja, mit `check-assets` | Score 20–40 | +| Problematischer Text (Pornografie, Propaganda, Spam, Off-Topic) | Ja, mit KI-Analyse | Score 20–50, auf Gelb gedeckelt | +| Problematische Bildinhalte + Text im Bild (OCR) | Ja, mit KI-Analyse | Score ≥ 40, auf Gelb gedeckelt | | IP-basiertes Cloaking | Nein | Nur Google Search Console kann das erkennen | | Serverseitige Code-Änderungen | Nein | Nur sichtbare HTML-Ausgabe | @@ -822,3 +826,128 @@ Der Report liegt unter `reports/_assets/report.md` und zeigt pro geänderter `periodic_checks` in der `config.yaml`). Die Datei-Vergleichsgrundlage wird zudem beim ersten `approve --all` automatisch angelegt — `approve-assets` ist nur für den manuellen Sonderfall gedacht. + +--- + +## 20. KI-gestützte Inhaltsanalyse + +Eine **optionale** Ergänzung zum Integritäts-Kern. Sie prüft Text und Bilder **semantisch** +auf problematische Inhalte, die kein strukturelles Signal hinterlassen: + +- sexuell anstößige Inhalte / Pornografie +- politische Propaganda +- diffamierende oder strafbare Aussagen +- versteckter Werbe-/Spam-Text +- Werbung, die thematisch nicht zur Website passt +- in sich widersprüchliche Aussagen +- problematische Bildinhalte **und** Text in Bildern (OCR) + +Die KI ist **Ergänzung, nicht Kern**: Ist sie deaktiviert (Standard), verhält sich der +Scanner exakt wie bisher. + +### Das Kostengate (warum das günstig ist) + +Jeder Text und jedes Bild bekommt einen Fingerabdruck (Hash). Das Ergebnis der KI-Prüfung +wird zu diesem Fingerabdruck in `data/ai_ledger.json` gespeichert. Bei jedem weiteren Scan +gilt: + +- **Inhalt unverändert** → gleicher Hash → Ergebnis kommt aus dem Cache → **keine** Abfrage, + keine Kosten. +- **Inhalt neu oder geändert** → eine günstige OpenRouter-Abfrage, Ergebnis wird gespeichert. + +Dadurch kann die Analyse bei **jedem** Scan mitlaufen und kostet an den meisten Tagen 0 €. + +### Wirkung auf die Ampel + +KI-Funde sind **auf Gelb gedeckelt** — sie lösen niemals allein ROT aus. ROT bleibt den +harten Integritäts-Signalen (Webshell, versteckte Inhalte) vorbehalten. Nur Funde mit +ausreichender Sicherheit (Konfidenz ≥ `ai_confidence_min`, Standard 0,7) und mindestens +mittlerer Schwere werden gewertet — das hält Fehlalarme niedrig. + +Fehlt der API-Key oder schlägt eine Abfrage fehl, wird die Analyse **übersprungen** und der +Scan läuft unverändert weiter. Die KI kann den normalen Betrieb nie blockieren. + +### Aktivieren + +```yaml +ai_analysis: + enabled: true + api_key_env: "OPENROUTER_API_KEY" + site_context: "Heimat- und Vereinswebsite über die Bergbaugeschichte in Bredelar." + ai_confidence_min: 0.7 + attempt_timeout: 30 # Zeitlimit je Modell-Versuch → eskaliert bei Langsamkeit + text: + enabled: true + models: # Kette: Stufe 1+2 free, Stufe 3 günstig bezahlt + - "qwen/qwen3-next-80b-a3b-instruct:free" + - "meta-llama/llama-3.3-70b-instruct:free" + - "google/gemini-2.5-flash-lite" + min_chars: 200 # nur Seitentexte ab dieser Größe + max_pages_per_scan: 20 # Kostendeckel pro Lauf + image: + enabled: true + models: # multimodal: Inhalt + OCR + - "google/gemma-4-31b-it:free" + - "nvidia/nemotron-nano-12b-v2-vl:free" + - "google/gemini-2.5-flash-lite" + max_images_per_scan: 15 +``` + +**Wichtig — `site_context`:** Die KI bewertet die **thematische Passung** zur hier +beschriebenen Website, nicht einzelne Schlüsselwörter. Ein Angreifer kann Heimat- oder +Ortsbegriffe einstreuen, um simple Wortlisten zu täuschen — gegen die semantische +Themenprüfung hilft das nicht. Je präziser der `site_context`, desto besser die Trefferquote. + +### API-Key setzen + +Das Passwort/Token **niemals** in `config.yaml` eintragen, nur als Umgebungsvariable: + +```bash +# Dauerhaft in ~/.bashrc +echo 'export OPENROUTER_API_KEY="sk-or-..."' >> ~/.bashrc +``` + +Für den Cron-Betrieb die Variable wie bei `SCANNER_SMTP_PASSWORD` in der Crontab oder in +`/etc/environment` hinterlegen. + +### Fehlalarm quittieren + +Wurde legitimer Inhalt fälschlich als auffällig markiert: + +```bash +python -m scanner https://ihre-website.de ai-dismiss --all # alle offenen Funde +python -m scanner https://ihre-website.de ai-dismiss --hash # gezielt einen Fund +``` + +Der Fingerabdruck `` steht im Report bei jedem KI-Fund. Ein `approve --all` oder +`approve --url` quittiert die zugehörigen KI-Funde automatisch mit (freigegebener Inhalt +gilt als geprüft). + +### Modelle (OpenRouter) und Eskalation + +Pro Modalität wird eine **Modell-Kette mit zweifacher Eskalation** abgearbeitet. Schlägt ein +Modell fehl (Fehler, Rate-Limit) oder antwortet es langsamer als `attempt_timeout` Sekunden, +wird automatisch zur nächsten Stufe gewechselt. Stufe 1 und 2 sind kostenlose Modelle, Stufe 3 +ist ein günstiges Bezahlmodell — so bleibt die Prüfung in der Regel kostenlos und ist trotzdem +zuverlässig. Welche Stufe das Verdikt geliefert hat, wird im Ledger und Report festgehalten. + +| Zweck | Stufe 1 (free) | Stufe 2 (free) | Stufe 3 (bezahlt) | +|---|---|---|---| +| Text/Semantik | `qwen3-next-80b-a3b-instruct:free` | `llama-3.3-70b-instruct:free` | `gemini-2.5-flash-lite` | +| Bild + OCR | `gemma-4-31b-it:free` | `nemotron-nano-12b-v2-vl:free` | `gemini-2.5-flash-lite` | + +Hinweis: Kostenlose Modelle liefern nicht immer ein gültiges Ergebnis (sie unterstützen das +strikte JSON-Format nicht durchgehend). In dem Fall greift automatisch die nächste Stufe — das +zuverlässige Bezahlmodell auf Stufe 3 fängt solche Fälle ab. + +Audio- und Video-Prüfung sind als abschaltbare Erweiterungspunkte vorbereitet +(`ai_analysis.audio` / `ai_analysis.video`, default deaktiviert) und können aktiviert werden, +sobald eine Website solche Mediendateien direkt einbindet. + +### Status prüfen + +```bash +python -m scanner https://ihre-website.de status +``` + +zeigt bei aktivierter KI, wie viele Inhalte im Cache liegen und wie viele offene Funde es gibt. diff --git a/README.md b/README.md index 6ddf6d4..1209ef7 100644 --- a/README.md +++ b/README.md @@ -59,6 +59,7 @@ dafür **nichts** extra einzurichten — ein täglicher Aufruf genügt für alle | `report --diff-only` | Nur Textdiffs geänderter Seiten (farbig, kompakt) | | `status` | Baseline-Datum, letzter Scan, Risiko-Level | | `test-alert` | Test-E-Mail senden, ohne echten Scan (SMTP-Check) | +| `ai-dismiss --all` | KI-Funde als geprüft/akzeptiert quittieren (Fehlalarm) | Alle Befehle akzeptieren `--config path/to/config.yaml` und `--verbose`. @@ -219,6 +220,84 @@ Nach dem ersten `init` alle legitimen externen Domains in --- +## KI-Inhaltsanalyse (optional) + +Der Integritäts-Kern erkennt strukturelle Manipulationen (fremde Links, Hidden Content, +Webshells). Die optionale KI-Analyse ergänzt ihn um **semantische** Prüfung von Text und +Bildern: Pornografie, Propaganda, diffamierende/strafbare Inhalte, versteckter Spam, +thematisch unpassende Werbung, widersprüchliche Aussagen — auch ohne Link-Signal und +inklusive Text in Bildern (OCR). + +**Kostengate:** Jeder Inhalt bekommt einen Hash. Unveränderte Inhalte liegen mit ihrem +Verdikt im Cache (`data/ai_ledger.json`) → **kein API-Call**. Nur neue oder geänderte +Inhalte lösen eine (günstige) OpenRouter-Abfrage aus. Dadurch läuft die Analyse bei jedem +Scan mit und kostet an den meisten Tagen 0 €. + +KI-Funde sind **auf Gelb gedeckelt** — sie lösen nie allein ROT aus (das bleibt harten +Integritäts-Signalen vorbehalten). Fehlt der API-Key oder schlägt eine Abfrage fehl, wird +die Analyse übersprungen und der Scan läuft unverändert weiter. + +### Aktivieren + +In `config.yaml`: +```yaml +ai_analysis: + enabled: true + api_key_env: "OPENROUTER_API_KEY" + site_context: "Heimat- und Vereinswebsite über die Bergbaugeschichte in Bredelar." + ai_confidence_min: 0.7 + attempt_timeout: 30 # Zeitlimit je Modell-Versuch → eskaliert bei Langsamkeit + text: + enabled: true + models: # Kette: Stufe 1+2 free, Stufe 3 günstig bezahlt + - "qwen/qwen3-next-80b-a3b-instruct:free" + - "meta-llama/llama-3.3-70b-instruct:free" + - "google/gemini-2.5-flash-lite" + image: + enabled: true + models: # multimodal: Inhalt + OCR + - "google/gemma-4-31b-it:free" + - "nvidia/nemotron-nano-12b-v2-vl:free" + - "google/gemini-2.5-flash-lite" +``` + +Die Modelle werden als **Kette mit zweifacher Eskalation** abgearbeitet: Stufe 1 und 2 sind +kostenlose Modelle, Stufe 3 ein günstiges Bezahlmodell. Schlägt ein Modell fehl oder antwortet +es langsamer als `attempt_timeout`, wird automatisch zur nächsten Stufe eskaliert. Im Report +und Ledger wird festgehalten, welche Stufe das Verdikt geliefert hat. + +`site_context` ist entscheidend: Die KI bewertet die **thematische Passung** zur +deklarierten Beschreibung, nicht einzelne Schlüsselwörter — eingestreute Heimat-Begriffe +machen Spam so nicht unauffällig. + +API-Key als Umgebungsvariable setzen (nie in config.yaml): +```bash +export OPENROUTER_API_KEY="sk-or-..." +``` + +### Fehlalarm quittieren + +Stuft die KI legitimen Inhalt fälschlich als auffällig ein: +```bash +python -m scanner https://ihre-website.de ai-dismiss --all # alle offenen Funde +python -m scanner https://ihre-website.de ai-dismiss --hash +``` +Ein `approve --all` / `--url` quittiert die zugehörigen KI-Funde automatisch mit. + +### Modelle (OpenRouter) + +Pro Modalität eine Kette mit zweifacher Eskalation (Stufe 1+2 free, Stufe 3 bezahlt): + +| Zweck | Stufe 1 (free) | Stufe 2 (free) | Stufe 3 (bezahlt) | +|---|---|---|---| +| Text/Semantik | `qwen3-next-80b-a3b-instruct:free` | `llama-3.3-70b-instruct:free` | `gemini-2.5-flash-lite` | +| Bild + OCR | `gemma-4-31b-it:free` | `nemotron-nano-12b-v2-vl:free` | `gemini-2.5-flash-lite` | + +Audio/Video sind als abschaltbare Hooks vorbereitet (`ai_analysis.audio/video`, +default aus). + +--- + ## Verzeichnisstruktur ``` diff --git a/meine-seite.de/config.yaml b/meine-seite.de/config.yaml index baed84e..cea9079 100644 --- a/meine-seite.de/config.yaml +++ b/meine-seite.de/config.yaml @@ -127,6 +127,45 @@ alerting: enabled: false url: "" # z.B. Slack-Webhook oder Telegram-Bot-URL +# --- KI-gestützte Inhaltsanalyse (optional, OpenRouter) --- +# Prüft Text und Bilder semantisch auf problematische Inhalte (Pornografie, +# Propaganda, Spam, themenfremde Werbung, strafbare/widersprüchliche Aussagen). +# Kostengate: nur NEUE/GEÄNDERTE Inhalte lösen einen API-Call aus (Hash-Cache). +# KI-Funde sind auf GELB gedeckelt. API-Key NUR als Umgebungsvariable setzen: +# export OPENROUTER_API_KEY="sk-or-..." +ai_analysis: + enabled: false + api_key_env: "OPENROUTER_API_KEY" + # Themenbeschreibung der Website — die KI bewertet die Passung dazu, nicht + # einzelne Schlüsselwörter. Bitte ausfüllen, wenn enabled: true. + site_context: "" + ai_confidence_min: 0.7 + attempt_timeout: 30 # Zeitlimit je Modell-Versuch (Sek.) → Auslöser der Eskalation + # Modell-Kette mit zweifacher Eskalation: Stufe 1+2 free, Stufe 3 günstig bezahlt. + # Schlägt ein Modell fehl oder antwortet zu langsam, wird zur nächsten Stufe eskaliert. + text: + enabled: true + models: + - "qwen/qwen3-next-80b-a3b-instruct:free" # Stufe 1: free + - "meta-llama/llama-3.3-70b-instruct:free" # Stufe 2: free + - "google/gemini-2.5-flash-lite" # Stufe 3: günstig bezahlt, zuverlässig + min_chars: 200 + max_pages_per_scan: 20 + image: + enabled: true + models: + - "google/gemma-4-31b-it:free" # Stufe 1: free, multimodal + OCR + - "nvidia/nemotron-nano-12b-v2-vl:free" # Stufe 2: free, Vision + - "google/gemini-2.5-flash-lite" # Stufe 3: günstig bezahlt, Vision + max_images_per_scan: 15 + # Audio/Video: vorbereitet, default aus (Medien auf statischen Seiten selten). + audio: + enabled: false + models: ["google/gemini-2.5-flash"] + video: + enabled: false + models: ["google/gemini-2.5-flash"] + # --- Normalisierung --- # Elemente, die vor dem Vergleich aus dem HTML entfernt werden # (z. B. dynamische Datums-/Zeitanzeigen die sich täglich ändern). diff --git a/scanner/__main__.py b/scanner/__main__.py index 071d5a0..10d5c80 100644 --- a/scanner/__main__.py +++ b/scanner/__main__.py @@ -24,6 +24,7 @@ from datetime import datetime, timezone from pathlib import Path from urllib.parse import urlparse +from .ai_analyzer import run_ai_analysis, score_ai_findings from .alerter import send_alert from .baseline import BaselineManager from .checker import ( @@ -149,6 +150,7 @@ def _build_report( cloak_diff: dict | None = None, asset_diff: dict | None = None, ext_links: dict | None = None, + ai_result: dict | None = None, ) -> dict: report = { "generated_at": datetime.now(timezone.utc).isoformat(), @@ -167,6 +169,8 @@ def _build_report( report["asset_diff"] = asset_diff if ext_links is not None: report["ext_links"] = ext_links + if ai_result is not None: + report["ai_result"] = ai_result return report @@ -400,6 +404,29 @@ def _render_markdown(r: dict) -> str: for page in entry.get("pages", [])[:3]: lines.append(f" - gefunden auf: {page}") + ai = r.get("ai_result") + if ai is not None: + lines += ["", "## KI-Inhaltsanalyse", ""] + if ai.get("skipped"): + lines.append(f"- Übersprungen: {ai['skipped']}") + else: + lines.append( + f"- {ai.get('checked', 0)} Inhalt(e) geprüft, " + f"{ai.get('cache_hits', 0)} aus Cache, {ai.get('api_calls', 0)} API-Abfrage(n)." + ) + for f in ai.get("findings", []): + asset = f.get("asset_url") + ziel = f"`{asset}` (auf {f['url']})" if asset else f"`{f['url']}`" + lines.append( + f"- [{f['kind']}] {ziel}: **{f['category']}** " + f"(Schwere {f['severity']}, Konfidenz {f['confidence']:.0%})" + ) + if f.get("explanation"): + lines.append(f" - Begründung: {f['explanation']}") + if f.get("fingerprint"): + lines.append(f" - Quittieren bei Fehlalarm: " + f"`python -m scanner ai-dismiss --hash {f['fingerprint']}`") + lines += [ "", "---", @@ -941,12 +968,21 @@ def cmd_check(args: argparse.Namespace, cfg: dict) -> int: baseline = bm.load_baseline() diff, assessment, wl_violations, missing_sec = _run_content_check(cfg, snap, baseline) + # --- KI-Inhaltsanalyse (hash-gegated) --- + ai_result = None + if cfg.get("ai_analysis", {}).get("enabled"): + ai_result = run_ai_analysis(cfg, bm, snap, diff) + ai_assess = score_ai_findings(ai_result, cfg) + overall = _worst_level([assessment["level"], ai_assess["level"]]) + assessment = {**assessment, "level": overall, "exit_code": _LEVEL_ORDER[overall], + "reasons": assessment.get("reasons", []) + ai_assess["reasons"]} + report = _build_report( cfg["target"], diff, assessment, wl_violations, missing_sec, - snap.get("errors", []), snap["dir"], + snap.get("errors", []), snap["dir"], ai_result=ai_result, ) _write_report(report, cfg["reports_dir"]) - _print_summary(assessment, diff) + _print_summary(assessment, diff, ai_result=ai_result) send_alert(report, cfg) return assessment["exit_code"] @@ -989,7 +1025,18 @@ def cmd_scan(args: argparse.Namespace, cfg: dict) -> int: ext_links = _run_ext_links_core(cfg, snap) # informativ, hebt Level nicht an bm.mark_check_run("ext_links") - # Gesamt-Level = schlechtestes aus Inhalt + Cloaking + Datei-Check + # --- KI-Inhaltsanalyse (hash-gegated, läuft bei jedem Scan; meist 0 API-Calls) --- + ai_result = None + if cfg.get("ai_analysis", {}).get("enabled"): + ai_result = run_ai_analysis(cfg, bm, snap, diff) + if ai_result.get("api_calls"): + print(f" KI-Inhaltsanalyse: {ai_result['api_calls']} neue Abfrage(n), " + f"{ai_result['cache_hits']} aus Cache.") + ai_assess = score_ai_findings(ai_result, cfg) + levels.append(ai_assess["level"]) + assessment["reasons"] = assessment.get("reasons", []) + ai_assess["reasons"] + + # Gesamt-Level = schlechtestes aus Inhalt + Cloaking + Datei-Check + KI overall = _worst_level(levels) assessment = {**assessment, "level": overall, "exit_code": _LEVEL_ORDER[overall]} @@ -998,9 +1045,10 @@ def cmd_scan(args: argparse.Namespace, cfg: dict) -> int: cfg["target"], diff, assessment, wl_violations, missing_sec, snap.get("errors", []), snap["dir"], cloak_diff=cloak_diff, asset_diff=asset_diff, ext_links=ext_links, + ai_result=ai_result, ) _write_report(report, cfg["reports_dir"]) - _print_summary(assessment, diff, cloak_diff, asset_diff, ext_links) + _print_summary(assessment, diff, cloak_diff, asset_diff, ext_links, ai_result) send_alert(report, cfg) return assessment["exit_code"] @@ -1036,6 +1084,7 @@ def cmd_approve(args: argparse.Namespace, cfg: dict) -> int: print(f"Ihre Website wurde als neuer Vergleichsstand gespeichert ({len(approved)} Seiten).") if not skip_assets: _establish_asset_baseline(cfg, bm, snap_dir) + _dismiss_ai_after_approve(bm) return 0 if getattr(args, "all", False): @@ -1044,6 +1093,7 @@ def cmd_approve(args: argparse.Namespace, cfg: dict) -> int: print("Ab jetzt meldet das Programm jede spätere Veränderung.") if not skip_assets: _establish_asset_baseline(cfg, bm, snap_dir) + _dismiss_ai_after_approve(bm) return 0 url = getattr(args, "url", None) @@ -1051,6 +1101,7 @@ def cmd_approve(args: argparse.Namespace, cfg: dict) -> int: ok = bm.approve_url(url, snap_dir=snap_dir, note=note) if ok: print(f"Diese Seite wurde als in Ordnung bestätigt: {url}") + _dismiss_ai_after_approve(bm, url=url) return 0 else: print(f"Diese Adresse wurde im letzten Stand nicht gefunden: {url}", file=sys.stderr) @@ -1061,6 +1112,35 @@ def cmd_approve(args: argparse.Namespace, cfg: dict) -> int: return 1 +def _dismiss_ai_after_approve(bm: BaselineManager, url: str | None = None) -> None: + """Freigegebene Inhalte gelten als geprüft → zugehörige KI-Funde quittieren.""" + if url: + ledger = bm.load_ai_ledger() + fps = [fp for fp, e in ledger.get("entries", {}).items() if e.get("url") == url] + n = bm.dismiss_ai_entries(fps) if fps else 0 + else: + n = bm.dismiss_ai_entries(None) + if n: + print(f" ({n} offene(r) KI-Fund(e) als geprüft mitquittiert.)") + + +def cmd_ai_dismiss(args: argparse.Namespace, cfg: dict) -> int: + """Markiert KI-Funde als geprüft/akzeptiert (Fehlalarm-Quittung).""" + target_hash = getattr(args, "hash", None) + if not target_hash and not getattr(args, "all", False): + print("Bitte angeben, was quittiert werden soll: --hash oder --all.", + file=sys.stderr) + return 1 + bm = BaselineManager(cfg["data_dir"]) + hashes = [target_hash] if target_hash else None + n = bm.dismiss_ai_entries(hashes) + if target_hash and n == 0: + print(f"Kein offener KI-Fund mit Fingerprint {target_hash} gefunden.") + return 1 + print(f"{n} KI-Fund(e) als geprüft/akzeptiert markiert.") + return 0 + + def _print_diff_only(report: dict) -> None: """Print a compact colored diff view of the latest report.""" a = report.get("assessment", {}) @@ -1126,12 +1206,23 @@ def _print_diff_only(report: dict) -> None: for cl in new_comments: print(f"{C['del']}! Kommentar-Link: {cl}{C['rst']}") + ai = report.get("ai_result") or {} + ai_findings = ai.get("findings", []) + for f in ai_findings: + asset = f.get("asset_url") + ziel = f"{asset} (auf {f['url']})" if asset else f["url"] + print(f"{C['del']}! KI [{f['kind']}] {f['category']} " + f"({f['confidence']:.0%}): {ziel}{C['rst']}") + if f.get("explanation"): + print(f" {f['explanation']}") + nothing = ( not d.get("new_external_domains") and not d.get("new_internal_urls") and not d.get("missing_internal_urls") and not d.get("new_broken_urls") and not has_page_diffs + and not ai_findings ) if nothing: print("Keine Änderungen.") @@ -1217,6 +1308,17 @@ def cmd_status(args: argparse.Namespace, cfg: dict) -> int: for key, label in labels.items(): ts = state.get(key, "noch nie") print(f" {label}: {ts[:10] if ts != 'noch nie' else ts}") + + # KI-Inhaltsanalyse + if cfg.get("ai_analysis", {}).get("enabled"): + ledger = bm.load_ai_ledger().get("entries", {}) + offen = sum( + 1 for e in ledger.values() + if e.get("category", "clean") != "clean" and not e.get("dismissed") + ) + print(f"KI-Inhaltsanalyse: aktiv — {len(ledger)} Inhalt(e) im Cache, {offen} offene(r) Fund(e)") + else: + print("KI-Inhaltsanalyse: deaktiviert") return 0 @@ -1367,12 +1469,13 @@ def _print_kein_baseline_hinweis() -> None: def _print_summary(assessment: dict, diff: dict, cloak_diff: dict | None = None, asset_diff: dict | None = None, - ext_links: dict | None = None) -> None: + ext_links: dict | None = None, + ai_result: dict | None = None) -> None: level = assessment.get("level", "green") print() print(ampel_zeile(level)) print() - for satz in klartext_befunde(diff, cloak_diff, asset_diff, ext_links): + for satz in klartext_befunde(diff, cloak_diff, asset_diff, ext_links, ai_result): print(f" • {satz}") print() for schritt in was_tun(level): @@ -1480,6 +1583,10 @@ def _build_parser() -> argparse.ArgumentParser: help="Simulierter Alarm-Level (default: yellow)", ) + ad = sub.add_parser("ai-dismiss", help="KI-Funde als geprüft/akzeptiert markieren (Fehlalarm-Quittung)") + ad.add_argument("--hash", help="Nur diesen Fingerprint quittieren") + ad.add_argument("--all", action="store_true", help="Alle offenen KI-Funde quittieren") + return p @@ -1567,6 +1674,7 @@ def main() -> None: "check-assets": cmd_check_assets, "approve-assets": cmd_approve_assets, "test-alert": cmd_test_alert, + "ai-dismiss": cmd_ai_dismiss, } handler = commands.get(args.command) if not handler: diff --git a/scanner/ai_analyzer.py b/scanner/ai_analyzer.py new file mode 100644 index 0000000..6c97543 --- /dev/null +++ b/scanner/ai_analyzer.py @@ -0,0 +1,452 @@ +""" +KI-gestützte semantische Inhaltsanalyse via OpenRouter. + +Ergänzung zum Integritäts-Kern, niemals dessen Ersatz. Erkennt problematische +*Inhalte ohne Link-Signal*: Pornografie, Propaganda, diffamierende/strafbare +Texte, versteckten Spam, thematisch unpassende Werbung, widersprüchliche +Aussagen — in Text und in Bildern (inkl. eingebettetem Text via OCR). + +Kostengate: Jeder Kandidat bekommt einen Fingerprint (SHA-256). Ein bereits +geprüfter Fingerprint liegt mit seinem Verdikt im Ledger (data/ai_ledger.json) +→ Cache-Treffer → KEIN API-Call. Nur neue/geänderte Inhalte kosten etwas. + +Robustheit: Fehlt der API-Key oder schlägt ein Call fehl, wird die Analyse +übersprungen und der Scan läuft unverändert weiter (graceful degradation). +KI darf den Kern-Scan nie brechen. +""" +import json +import logging +import os +from datetime import datetime, timezone + +import requests + +from .checker import fetch_asset_hashes +from .differ import normalize_text + +logger = logging.getLogger(__name__) + +_OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions" + +# Erlaubte Kategorien, die das Modell zurückgeben darf. +_CATEGORIES = [ + "clean", "pornography", "propaganda", "defamation_illegal", + "hidden_spam", "off_topic_commercial", "contradiction", +] + +# Kategorie → Scoring-Schlüssel (Fallback-Punkte, falls cfg sie nicht liefert). +_CATEGORY_SCORE_KEY = { + "pornography": ("ai_pornography", 50), + "defamation_illegal": ("ai_defamation_illegal", 50), + "propaganda": ("ai_propaganda", 40), + "hidden_spam": ("ai_hidden_spam", 40), + "off_topic_commercial": ("ai_off_topic_commercial", 30), + "contradiction": ("ai_contradiction", 20), +} + +_SEVERITY_ORDER = {"none": 0, "low": 1, "medium": 2, "high": 3} + +# JSON-Schema für die strukturierte Modell-Antwort. +_RESPONSE_SCHEMA = { + "type": "json_schema", + "json_schema": { + "name": "content_verdict", + "strict": True, + "schema": { + "type": "object", + "properties": { + "category": {"type": "string", "enum": _CATEGORIES}, + "severity": {"type": "string", "enum": ["none", "low", "medium", "high"]}, + "confidence": {"type": "number"}, + "explanation": {"type": "string"}, + }, + "required": ["category", "severity", "confidence", "explanation"], + "additionalProperties": False, + }, + }, +} + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +def run_ai_analysis(cfg: dict, bm, snap: dict, diff: dict) -> dict: + """ + Hash-gegate KI-Analyse über neue/geänderte Texte und Bilder. + + Returns {"findings": [...], "checked": int, "cache_hits": int, + "api_calls": int, "skipped": str|None}. + Bricht nie mit einer Exception — bei Problemen wird "skipped" gesetzt. + """ + result = {"findings": [], "checked": 0, "cache_hits": 0, "api_calls": 0, "skipped": None} + ai_cfg = cfg.get("ai_analysis", {}) + + if not ai_cfg.get("enabled"): + result["skipped"] = "deaktiviert" + return result + + api_key = os.environ.get(ai_cfg.get("api_key_env", "OPENROUTER_API_KEY"), "") + if not api_key: + result["skipped"] = f"kein API-Key ({ai_cfg.get('api_key_env', 'OPENROUTER_API_KEY')})" + logger.warning("KI-Analyse übersprungen: %s", result["skipped"]) + return result + + try: + ledger = bm.load_ai_ledger() + entries = ledger.setdefault("entries", {}) + dirty = False + + if ai_cfg.get("text", {}).get("enabled", True): + dirty |= _analyze_text(cfg, ai_cfg, api_key, snap, diff, entries, result) + if ai_cfg.get("image", {}).get("enabled", True): + dirty |= _analyze_images(cfg, ai_cfg, api_key, snap, diff, entries, result) + # Audio/Video: vorbereitet, default aus (siehe _collect_audio/video_candidates). + + if dirty: + bm.save_ai_ledger(ledger) + except Exception as exc: # pragma: no cover - Schutzschirm, darf Scan nie brechen + logger.error("KI-Analyse mit unerwartetem Fehler abgebrochen: %s", exc) + result["skipped"] = f"interner Fehler: {exc}" + + return result + + +def score_ai_findings(ai_result: dict, cfg: dict) -> dict: + """ + Bewertet KI-Funde additiv. Das Level ist hart auf 'yellow' gedeckelt — + KI allein löst niemals ROT aus (ROT bleibt harten Integritäts-Signalen). + + Returns {score, level, reasons, exit_code}. + """ + sc = cfg.get("scoring", {}) + thr = cfg.get("thresholds", {"yellow": 20, "red": 60}) + yellow = thr.get("yellow", 20) + + score = 0 + reasons: list[str] = [] + for f in ai_result.get("findings", []): + score += (pts := _finding_points(f, sc)) + conf = f.get("confidence", 0.0) + reasons.append( + f"KI [{f.get('kind', '?')}] {f.get('url', '?')}: " + f"{f.get('category', '?')} ({f.get('severity', '?')}, {conf:.0%}) (+{pts})" + ) + + # Deckelung auf gelb: auch bei Score ≥ rot-Schwelle bleibt es gelb. + if score >= yellow: + level, exit_code = "yellow", 1 + else: + level, exit_code = "green", 0 + + return {"score": score, "level": level, "reasons": reasons, "exit_code": exit_code} + + +# --------------------------------------------------------------------------- +# Text analysis +# --------------------------------------------------------------------------- + +def _analyze_text(cfg, ai_cfg, api_key, snap, diff, entries, result) -> bool: + """Prüft neue/geänderte Seitentexte. Returns True wenn Ledger verändert.""" + text_cfg = ai_cfg.get("text", {}) + min_chars = text_cfg.get("min_chars", 200) + models = _models_for(text_cfg) + site_context = ai_cfg.get("site_context", "") + + # Fingerprint je Seite (lokal, gratis). Geänderte/neue Seiten zuerst (Risiko). + changed = {pd["url"] for pd in diff.get("page_diffs", [])} | set(diff.get("new_internal_urls", [])) + candidates: list[tuple[str, str, str]] = [] + for url, page in snap.get("pages", {}).items(): + if page.get("status") != 200: + continue + text = page.get("text", "") or "" + if len(text) < min_chars: + continue + candidates.append((url, text, _text_fingerprint(text, cfg))) + candidates.sort(key=lambda c: c[0] not in changed) # changed first (False < True) + + budget = text_cfg.get("max_pages_per_scan", 20) + dirty = False + for url, text, fp in candidates: + result["checked"] += 1 + entry = entries.get(fp) + if entry is None: + if budget <= 0: + continue # Kostendeckel erreicht — nächster Scan holt den Rest nach + classified = _classify_text(text, site_context, models, ai_cfg, api_key) + if classified is None: + result["skipped"] = result["skipped"] or "API-Fehler (Text)" + continue + verdict, used_model = classified + entry = _make_entry("text", url, verdict, used_model) + entries[fp] = entry + result["api_calls"] += 1 + budget -= 1 + dirty = True + else: + result["cache_hits"] += 1 + _maybe_finding(result, entry, fp, url, ai_cfg) + return dirty + + +# --------------------------------------------------------------------------- +# Image analysis +# --------------------------------------------------------------------------- + +def _analyze_images(cfg, ai_cfg, api_key, snap, diff, entries, result) -> bool: + """Prüft neue/geänderte Bilder. Returns True wenn Ledger verändert.""" + img_cfg = ai_cfg.get("image", {}) + models = _models_for(img_cfg) + site_context = ai_cfg.get("site_context", "") + + current_img_urls = _all_image_urls(snap) + if not current_img_urls: + return False + + has_image_entries = any(e.get("kind") == "image" for e in entries.values()) + changed = {pd["url"] for pd in diff.get("page_diffs", [])} | set(diff.get("new_internal_urls", [])) + changed_img_urls = _all_image_urls(snap, only_pages=changed) + # Markierte Bilder erneut holen: ersetztes Bild → neue Bytes → Neubewertung + # (verhindert Dauer-Fehlalarm nach Bereinigung); unverändert → Cache-Treffer. + flagged_img_urls = { + e["url"] for e in entries.values() + if e.get("kind") == "image" and e.get("category", "clean") != "clean" + and not e.get("dismissed") and e.get("url") in current_img_urls + } + + if has_image_entries: + fetch_urls = sorted((changed_img_urls | flagged_img_urls) & current_img_urls) + else: + fetch_urls = sorted(current_img_urls) # Erstlauf: vollständiger Sweep + + fetch_urls = fetch_urls[: img_cfg.get("max_images_per_scan", 15)] + if not fetch_urls: + return False + + # url → Seite(n) für den Report + url_to_page = _image_url_sources(snap) + hashes = fetch_asset_hashes(fetch_urls, timeout=cfg.get("request_timeout", 15)) + + dirty = False + for url in fetch_urls: + h = hashes.get(url, {}) + fp = h.get("sha256") + if not fp: + continue # nicht abrufbar — überspringen + result["checked"] += 1 + page_url = url_to_page.get(url, url) + entry = entries.get(fp) + if entry is None: + classified = _classify_image(url, site_context, models, ai_cfg, api_key) + if classified is None: + result["skipped"] = result["skipped"] or "API-Fehler (Bild)" + continue + verdict, used_model = classified + entry = _make_entry("image", url, verdict, used_model) + entries[fp] = entry + result["api_calls"] += 1 + dirty = True + else: + result["cache_hits"] += 1 + _maybe_finding(result, entry, fp, page_url, ai_cfg, asset_url=url) + return dirty + + +# --------------------------------------------------------------------------- +# Audio / Video — vorbereitete Erweiterungspunkte (default deaktiviert) +# --------------------------------------------------------------------------- + +def _collect_audio_candidates(snap: dict, cfg: dict) -> list[str]: + """TODO: Audio-Quellen sammeln. Erfordert