From 989f5a933fd4d1fb6a9a7257a4f116f15d2c74d3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dieter=20Schl=C3=BCter?= Date: Sat, 13 Jun 2026 05:12:19 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20Inhaltsinventar=20in=20SQLite=20?= =?UTF-8?q?=E2=80=94=20KI-Inhaltsangabe=20je=20Objekt=20(URL=20+=20Hash)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Vereinheitlicht den bisherigen JSON-Ledger zu EINEM abfragbaren SQLite-Store pro Site (data/content_inventory.db): jedes geprüfte Objekt mit Inhalts-Hash, URL(s), Sicherheits-Verdikt UND neutraler KI-Inhaltsangabe. - baseline.py: load/save_ai_ledger jetzt SQLite-gestützt (Dict-Schnittstelle bleibt → Analyzer unverändert). Tabellen objects + object_urls. Einmalige Migration eines vorhandenen ai_ledger.json → SQLite (.migrated). UPSERT mit first_seen-Erhalt; object_urls transaktional ersetzt. Neu: query_inventory, export_inventory_csv (beide migrieren failsafe). - ai_analyzer.py: Antwort-Schema + Prompt um neutrales Feld 'description' erweitert (im selben Call, keine Extrakosten); _make_entry speichert es. - __main__.py: neues Kommando 'inventory' (Übersicht, --search, --kind, --csv). - Doku: README + Bedienungsanleitung. Audio/Video sind im Schema (kind) vorbereitet (Phase 2: Extractor + Modalitäten). bredelar.info real migriert: 245 Objekte (200 Bilder, 45 Texte). Tests: 266 grün (+6: Round-Trip mit description, Migration, first_seen-Erhalt, invalidate, Suche/CSV-Export, description landet im Ledger). Co-Authored-By: Claude Opus 4.8 --- BEDIENUNGSANLEITUNG.md | 24 +++++++ README.md | 1 + scanner/__main__.py | 33 +++++++++ scanner/ai_analyzer.py | 8 ++- scanner/baseline.py | 143 ++++++++++++++++++++++++++++++++------ tests/test_ai_analyzer.py | 11 +++ tests/test_baseline.py | 64 +++++++++++++++++ 7 files changed, 262 insertions(+), 22 deletions(-) diff --git a/BEDIENUNGSANLEITUNG.md b/BEDIENUNGSANLEITUNG.md index 4752c27..b2df3e0 100644 --- a/BEDIENUNGSANLEITUNG.md +++ b/BEDIENUNGSANLEITUNG.md @@ -215,6 +215,9 @@ Globale Optionen: `--config path/to/config.yaml`, `--verbose` (`-v`). | `report` | Letzten Report im Terminal anzeigen | | `report --diff-only` | Nur Textdiffs geänderter Seiten — farbig, kompakt | | `report --format json` | Letzten Report als JSON anzeigen | +| `inventory` | KI-Inhaltsinventar (jedes Objekt: URL, Hash, Inhaltsangabe) anzeigen | +| `inventory --search ` | Inventar nach Inhaltsangabe/URL/Kategorie durchsuchen | +| `inventory --csv ` | Inventar als CSV exportieren | | `status` | Baseline-Datum, letzter Scan, aktueller Risk-Level | | `cloak-check` | Doppel-Crawl: Browser-UA vs. Googlebot (Cloaking-Erkennung) | | `check-ext-links` | Externe Links auf Erreichbarkeit prüfen (4xx/5xx) | @@ -1050,3 +1053,24 @@ 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. + +### Inhaltsinventar (SQLite) + +Jedes geprüfte Objekt (Seitentext, Bild; später Audio/Video) wird mit **Inhalts-Hash, URL(s), +Sicherheits-Verdikt und einer neutralen KI-Inhaltsangabe** in `data/content_inventory.db` +(SQLite) gespeichert — eine durchsuchbare Quelle der Wahrheit pro Site. + +```bash +python -m scanner https://ihre-website.de inventory # Übersicht +python -m scanner https://ihre-website.de inventory --search kloster # Volltextsuche +python -m scanner https://ihre-website.de inventory --kind image # nur Bilder +python -m scanner https://ihre-website.de inventory --csv inventar.csv +``` + +Die Datenbank lässt sich auch direkt mit jedem SQLite-Werkzeug abfragen (Tabellen `objects` +und `object_urls`). Ein bereits vorhandenes `ai_ledger.json` (Vorgängerformat) wird beim ersten +Zugriff automatisch nach SQLite migriert und als `ai_ledger.json.migrated` abgelegt. + +Hinweis: Das `description`-Feld füllt sich erst, wenn ein Objekt (neu/geändert) analysiert wird. +Objekte, die vor Einführung der Inhaltsangabe geprüft wurden, haben zunächst eine leere +Beschreibung; sie wird beim nächsten Re-Check des jeweiligen Inhalts ergänzt. diff --git a/README.md b/README.md index 095237f..8cacf02 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,7 @@ dafür **nichts** extra einzurichten — ein täglicher Aufruf genügt für alle | `approve --rebuild` | Gesamte Baseline ersetzen (nach großer Änderung) | | `report` | Letzten Report anzeigen | | `report --diff-only` | Nur Textdiffs geänderter Seiten (farbig, kompakt) | +| `inventory` | KI-Inhaltsinventar (Objekt → URL, Hash, Inhaltsangabe) anzeigen/suchen/`--csv` exportieren | | `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) | diff --git a/scanner/__main__.py b/scanner/__main__.py index 91831c8..6f635fc 100644 --- a/scanner/__main__.py +++ b/scanner/__main__.py @@ -991,6 +991,32 @@ def _print_diff_only(report: dict) -> None: print("Keine Änderungen.") +def cmd_inventory(args: argparse.Namespace, cfg: dict) -> int: + """KI-Inhaltsinventar (Objekt → URL, Hash, Inhaltsangabe) anzeigen/suchen/exportieren.""" + bm = BaselineManager(cfg["data_dir"]) + if getattr(args, "csv", None): + n = bm.export_inventory_csv(args.csv) + print(f"{n} Objekt(e) nach {args.csv} exportiert.") + return 0 + rows = bm.query_inventory(search=getattr(args, "search", None), kind=getattr(args, "kind", None)) + if not rows: + print("Inventar ist leer (noch keine KI-Analyse gelaufen oder kein Treffer).") + return 0 + from collections import Counter + kinds = Counter(r["kind"] for r in rows) + print(f"Inventar: {len(rows)} Objekt(e) — " + ", ".join(f"{k}: {v}" for k, v in sorted(kinds.items()))) + print() + for r in rows: + desc = (r.get("description") or "").replace("\n", " ").strip() + if len(desc) > 90: + desc = desc[:87] + "…" + flag = "" if (r.get("category") or "clean") == "clean" else f" [{r['category']}]" + print(f" [{r['kind']:5s}] {(r.get('hash') or '')[:12]} {r.get('url', '')}{flag}") + if desc: + print(f" {desc}") + return 0 + + def cmd_report(args: argparse.Namespace, cfg: dict) -> int: rd = Path(cfg["reports_dir"]) if not rd.exists(): @@ -1352,6 +1378,12 @@ def _build_parser() -> argparse.ArgumentParser: ad.add_argument("--hash", help="Nur diesen Fingerprint quittieren") ad.add_argument("--all", action="store_true", help="Alle offenen KI-Funde quittieren") + iv = sub.add_parser("inventory", help="KI-Inhaltsinventar anzeigen/durchsuchen/exportieren") + iv.add_argument("--search", help="Volltextsuche über Inhaltsangabe/URL/Kategorie") + iv.add_argument("--kind", choices=["text", "image", "audio", "video"], + help="Nur diesen Objekttyp anzeigen") + iv.add_argument("--csv", help="Inventar als CSV-Datei exportieren (Pfad)") + return p @@ -1433,6 +1465,7 @@ def main() -> None: "scan": cmd_scan, "approve": cmd_approve, "report": cmd_report, + "inventory": cmd_inventory, "status": cmd_status, "cloak-check": cmd_cloak_check, "check-ext-links": cmd_check_ext_links, diff --git a/scanner/ai_analyzer.py b/scanner/ai_analyzer.py index 23cf142..f8b7f48 100644 --- a/scanner/ai_analyzer.py +++ b/scanner/ai_analyzer.py @@ -58,12 +58,13 @@ _RESPONSE_SCHEMA = { "schema": { "type": "object", "properties": { + "description": {"type": "string"}, "category": {"type": "string", "enum": _CATEGORIES}, "severity": {"type": "string", "enum": ["none", "low", "medium", "high"]}, "confidence": {"type": "number"}, "explanation": {"type": "string"}, }, - "required": ["category", "severity", "confidence", "explanation"], + "required": ["description", "category", "severity", "confidence", "explanation"], "additionalProperties": False, }, }, @@ -483,6 +484,7 @@ def _make_entry(kind: str, url: str, verdict: dict, model: str) -> dict: return { "kind": kind, "url": url, + "description": verdict.get("description", ""), "category": verdict.get("category", "clean"), "severity": verdict.get("severity", "none"), "confidence": float(verdict.get("confidence", 0.0)), @@ -566,7 +568,9 @@ _SYSTEM_PROMPT = ( "einen Werbe- oder Spam-Text nicht unproblematisch — ein Angreifer kann sie " "gezielt einstreuen. Bewerte den Gesamtcharakter des Inhalts.\n" "Gib 'clean' zurück, wenn der Inhalt zum Thema passt und unproblematisch ist. " - "Antworte ausschließlich im vorgegebenen JSON-Format mit deutscher Begründung." + "Gib im Feld 'description' eine knappe, neutrale Inhaltsangabe (1–2 Sätze, deutsch): " + "was ist auf dem Bild zu sehen bzw. worum geht es im Text — sachlich, unabhängig vom " + "Sicherheitsurteil. Antworte ausschließlich im vorgegebenen JSON-Format." ) diff --git a/scanner/baseline.py b/scanner/baseline.py index f51da96..2c108e5 100644 --- a/scanner/baseline.py +++ b/scanner/baseline.py @@ -5,9 +5,11 @@ Snapshots are timestamped crawl results. The baseline is a manually approved reference state. Compromised content must never flow into the baseline automatically — every update requires an explicit `approve` call. """ +import csv import hashlib import json import logging +import sqlite3 from datetime import datetime, timezone from pathlib import Path from typing import Optional @@ -234,35 +236,136 @@ class BaselineManager: logger.info("Asset hashes saved: %d entries", len(hashes)) # ------------------------------------------------------------------ - # AI verdict ledger (cache keyed by content fingerprint) + # AI content inventory (SQLite store: verdict + content description + url→hash) # ------------------------------------------------------------------ # - # Unlike the baseline, the ledger is a *cache*: it records the AI verdict - # for every fingerprint already analysed — including "clean" ones — so that - # unchanged content is never re-sent to the API. It lives in data_dir - # (not baseline_dir): it is operational state, not an approved reference. + # EINE Quelle der Wahrheit pro Site: jedes geprüfte Objekt steht mit Inhalts-Hash, + # URL(s), Sicherheits-Verdikt UND neutraler KI-Inhaltsangabe in der DB. Der Analyzer + # arbeitet weiterhin mit dem In-Memory-Dict {"entries": {hash:…}, "url_hashes": {url:hash}}; + # nur die Persistenz liegt in SQLite (abfragbar, skaliert, CSV-Export möglich). @property - def _ai_ledger_path(self) -> Path: - return self.data_dir / "ai_ledger.json" + def _content_db_path(self) -> Path: + return self.data_dir / "content_inventory.db" + + def _connect_inventory(self) -> sqlite3.Connection: + self.data_dir.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(self._content_db_path) + conn.row_factory = sqlite3.Row + conn.execute(""" + CREATE TABLE IF NOT EXISTS objects ( + hash TEXT PRIMARY KEY, + kind TEXT, url TEXT, description TEXT, + category TEXT, severity TEXT, confidence REAL, explanation TEXT, + model TEXT, dismissed INTEGER DEFAULT 0, checked_at TEXT, + first_seen TEXT, last_seen TEXT + )""") + conn.execute(""" + CREATE TABLE IF NOT EXISTS object_urls ( + url TEXT PRIMARY KEY, hash TEXT NOT NULL, last_seen TEXT + )""") + conn.execute("CREATE INDEX IF NOT EXISTS idx_object_urls_hash ON object_urls(hash)") + return conn def load_ai_ledger(self) -> dict: - """Load the AI verdict ledger. Returns {'entries': {}} if none/corrupt.""" - if not self._ai_ledger_path.exists(): - return {"entries": {}} + """Load the AI store into the in-memory dict the analyzer uses. + Returns {"entries": {hash: {...}}, "url_hashes": {url: hash}}.""" + self._migrate_json_ledger() + conn = self._connect_inventory() try: - data = json.loads(self._ai_ledger_path.read_text(encoding="utf-8")) - except (json.JSONDecodeError, ValueError): - return {"entries": {}} - data.setdefault("entries", {}) - return data + entries: dict = {} + for row in conn.execute("SELECT * FROM objects"): + d = dict(row) + h = d.pop("hash") + d.pop("first_seen", None) + d.pop("last_seen", None) + d["dismissed"] = bool(d.get("dismissed")) + entries[h] = d + url_hashes = {r["url"]: r["hash"] + for r in conn.execute("SELECT url, hash FROM object_urls")} + finally: + conn.close() + return {"entries": entries, "url_hashes": url_hashes} def save_ai_ledger(self, ledger: dict) -> None: - """Persist the AI verdict ledger.""" - self.data_dir.mkdir(parents=True, exist_ok=True) - self._ai_ledger_path.write_text( - json.dumps(ledger, indent=2, ensure_ascii=False), encoding="utf-8" - ) + """Persist the in-memory dict back to SQLite. Objekte werden geupsertet + (first_seen bleibt erhalten); object_urls wird ersetzt (damit Entfernungen wirken).""" + now = datetime.now(timezone.utc).isoformat() + conn = self._connect_inventory() + try: + for h, e in ledger.get("entries", {}).items(): + conn.execute(""" + INSERT INTO objects (hash, kind, url, description, category, severity, + confidence, explanation, model, dismissed, checked_at, first_seen, last_seen) + VALUES (:hash,:kind,:url,:description,:category,:severity,:confidence, + :explanation,:model,:dismissed,:checked_at,:now,:now) + ON CONFLICT(hash) DO UPDATE SET + kind=excluded.kind, url=excluded.url, description=excluded.description, + category=excluded.category, severity=excluded.severity, + confidence=excluded.confidence, explanation=excluded.explanation, + model=excluded.model, dismissed=excluded.dismissed, + checked_at=excluded.checked_at, last_seen=excluded.last_seen + """, { + "hash": h, "kind": e.get("kind"), "url": e.get("url"), + "description": e.get("description", ""), "category": e.get("category"), + "severity": e.get("severity"), "confidence": e.get("confidence"), + "explanation": e.get("explanation", ""), "model": e.get("model"), + "dismissed": 1 if e.get("dismissed") else 0, + "checked_at": e.get("checked_at"), "now": now, + }) + conn.execute("DELETE FROM object_urls") + conn.executemany( + "INSERT INTO object_urls (url, hash, last_seen) VALUES (?,?,?)", + [(u, h, now) for u, h in ledger.get("url_hashes", {}).items()]) + conn.commit() + finally: + conn.close() + + def _migrate_json_ledger(self) -> None: + """Einmalige Migration eines alten ai_ledger.json nach SQLite.""" + old = self.data_dir / "ai_ledger.json" + if not old.exists(): + return + try: + data = json.loads(old.read_text(encoding="utf-8")) + except (json.JSONDecodeError, ValueError): + old.rename(old.with_suffix(".json.corrupt")) + return + self.save_ai_ledger({"entries": data.get("entries", {}), + "url_hashes": data.get("url_hashes", {})}) + old.rename(old.with_suffix(".json.migrated")) + logger.info("ai_ledger.json nach SQLite migriert (%d Objekte).", + len(data.get("entries", {}))) + + def query_inventory(self, search: str | None = None, kind: str | None = None) -> list[dict]: + """Inventar abfragen (für das `inventory`-Kommando).""" + self._migrate_json_ledger() + conn = self._connect_inventory() + try: + sql, params, cond = "SELECT * FROM objects", [], [] + if kind: + cond.append("kind = ?"); params.append(kind) + if search: + cond.append("(description LIKE ? OR url LIKE ? OR category LIKE ?)") + params += [f"%{search}%"] * 3 + if cond: + sql += " WHERE " + " AND ".join(cond) + sql += " ORDER BY kind, url" + return [dict(r) for r in conn.execute(sql, params)] + finally: + conn.close() + + def export_inventory_csv(self, path: str | Path) -> int: + """Inventar als CSV exportieren. Returns Anzahl Zeilen.""" + cols = ["hash", "kind", "url", "description", "category", "severity", "confidence", + "explanation", "model", "dismissed", "checked_at", "first_seen", "last_seen"] + rows = self.query_inventory() + with Path(path).open("w", newline="", encoding="utf-8") as f: + w = csv.DictWriter(f, fieldnames=cols, extrasaction="ignore") + w.writeheader() + for r in rows: + w.writerow(r) + return len(rows) # ------------------------------------------------------------------ # AI model-router state (circuit-breaker telemetry + /models catalog) diff --git a/tests/test_ai_analyzer.py b/tests/test_ai_analyzer.py index a0e7d28..46861f9 100644 --- a/tests/test_ai_analyzer.py +++ b/tests/test_ai_analyzer.py @@ -506,6 +506,17 @@ class TestParallel: assert len(res["findings"]) == 2 # aber Fund für beide URLs assert {f["url"] for f in res["findings"]} == {"https://x.de/a", "https://x.de/b"} + def test_description_stored_in_ledger(self, tmp_path, monkeypatch): + """Die neutrale Inhaltsangabe aus dem Verdikt landet im Inventar.""" + monkeypatch.setenv("OPENROUTER_API_KEY", "test") + bm = BaselineManager(tmp_path) + v = {"description": "Eine Seite über die Dorfgeschichte.", "category": "clean", + "severity": "none", "confidence": 0.9, "explanation": "ok"} + with patch("scanner.ai_analyzer._openrouter_chat", return_value=v): + run_ai_analysis(_cfg(), bm, _snap(_LONG), {}) + entry = next(iter(bm.load_ai_ledger()["entries"].values())) + assert entry["description"] == "Eine Seite über die Dorfgeschichte." + def test_changed_page_always_analyzed_despite_backlog_limit(self, tmp_path, monkeypatch): """Geänderte Seiten werden IMMER geprüft — die Bestands-Drossel gilt nicht für sie.""" monkeypatch.setenv("OPENROUTER_API_KEY", "test") diff --git a/tests/test_baseline.py b/tests/test_baseline.py index 73bbfe3..79f495e 100644 --- a/tests/test_baseline.py +++ b/tests/test_baseline.py @@ -159,3 +159,67 @@ class TestRebuildBaseline: baseline = bm.load_baseline() assert baseline["pages"]["https://b.info/"]["text"] == "Neu" + + +# --------------------------------------------------------------------------- +# AI content inventory (SQLite store: verdict + description + url→hash) +# --------------------------------------------------------------------------- + +def _entry(kind="image", url="https://b.info/x.jpg", category="clean", desc="Ein Logo"): + return {"kind": kind, "url": url, "description": desc, "category": category, + "severity": "none", "confidence": 0.9, "explanation": "ok", + "model": "m", "dismissed": False, "checked_at": "2026-01-01T00:00:00"} + + +class TestAiInventory: + def test_roundtrip_preserves_entries_and_url_hashes(self, tmp_path): + bm = BaselineManager(tmp_path) + bm.save_ai_ledger({"entries": {"H1": _entry(desc="Schützenfest-Plakat")}, + "url_hashes": {"https://b.info/x.jpg": "H1"}}) + led = bm.load_ai_ledger() + assert led["entries"]["H1"]["description"] == "Schützenfest-Plakat" + assert led["entries"]["H1"]["dismissed"] is False + assert led["url_hashes"] == {"https://b.info/x.jpg": "H1"} + + def test_migrates_json_ledger_once(self, tmp_path): + (tmp_path).mkdir(parents=True, exist_ok=True) + old = tmp_path / "ai_ledger.json" + old.write_text(json.dumps({"entries": {"H1": _entry()}, "url_hashes": {"u": "H1"}}), + encoding="utf-8") + bm = BaselineManager(tmp_path) + led = bm.load_ai_ledger() + assert "H1" in led["entries"] + assert not old.exists() # umbenannt + assert (tmp_path / "ai_ledger.json.migrated").exists() + assert (tmp_path / "content_inventory.db").exists() + + def test_first_seen_preserved_last_seen_updated(self, tmp_path): + bm = BaselineManager(tmp_path) + bm.save_ai_ledger({"entries": {"H1": _entry()}, "url_hashes": {}}) + rows1 = bm.query_inventory() + first1, last1 = rows1[0]["first_seen"], rows1[0]["last_seen"] + import time; time.sleep(0.01) + bm.save_ai_ledger({"entries": {"H1": _entry(category="propaganda")}, "url_hashes": {}}) + rows2 = bm.query_inventory() + assert rows2[0]["first_seen"] == first1 # bleibt + assert rows2[0]["last_seen"] >= last1 # aktualisiert + assert rows2[0]["category"] == "propaganda" # Feld aktualisiert + + def test_invalidate_removes_url(self, tmp_path): + bm = BaselineManager(tmp_path) + bm.save_ai_ledger({"entries": {}, "url_hashes": {"a": "H1", "b": "H2"}}) + assert bm.invalidate_ai_url_hashes(["a"]) == 1 + assert bm.load_ai_ledger()["url_hashes"] == {"b": "H2"} + + def test_query_search_and_csv_export(self, tmp_path): + bm = BaselineManager(tmp_path) + bm.save_ai_ledger({"entries": { + "H1": _entry(url="https://b.info/kloster.jpg", desc="Foto des Klosters Bredelar"), + "H2": _entry(url="https://b.info/auto.jpg", desc="Ein Auto"), + }, "url_hashes": {}}) + hits = bm.query_inventory(search="kloster") + assert len(hits) == 1 and "Kloster" in hits[0]["description"] + out = tmp_path / "inv.csv" + assert bm.export_inventory_csv(out) == 2 + content = out.read_text(encoding="utf-8") + assert "description" in content and "Kloster" in content