refactor+feat: Snapshot-Router-Split + generischer ChecklistAgent + AGB-Modul

- Item 2: Snapshot-Doc-Checks (cookie/impressum/dse/agb) in snapshot_check_routes.py
  (agent_compliance_check_routes.py 464→365 Z.); gleiche Pfade, in main.py registriert.
- ChecklistAgent-Basis: DSE-Logik generalisiert (L1/L2, kurze Titel, _severity_
  override-Hook). DSEAgent + AGBAgent sind jetzt Thin-Subclasses → künftige
  Doc-Agenten (widerruf/avv/…) trivial.
- Item 4: AGBAgent (§§ 305 ff. BGB, AGB_CHECKLIST) + agb-check + AGB-Tab via
  AgentModuleTab. Kein Library-Firehose.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-06-11 14:23:29 +02:00
parent b40edd6d33
commit 7258744107
11 changed files with 391 additions and 262 deletions
@@ -223,105 +223,6 @@ async def get_snapshot(snapshot_id: str):
db.close()
@router.get("/snapshots/{snapshot_id}/cookie-check")
async def snapshot_cookie_check(snapshot_id: str):
"""Pro-Cookie-Abgleich der Snapshot-Vendors gegen cookie_knowledge_db."""
from fastapi import HTTPException
from database import SessionLocal
from compliance.services.check_snapshot import load_snapshot
from compliance.services.cookie_library_check import (
analyze_cookies, load_big_library,
)
from compliance.services.cookie_storage_inventory import (
build_storage_inventory, storage_transparency_finding,
)
from compliance.services.cookie_compliance_audit import (
audit_cookie_compliance,
)
db = SessionLocal()
try:
snap = load_snapshot(db, snapshot_id)
if not snap:
raise HTTPException(status_code=404, detail="snapshot not found")
vendors = snap.get("cmp_vendors") or []
names = [c.get("name", "")
for v in vendors for c in (v.get("cookies") or [])]
big = load_big_library(db, names)
out = analyze_cookies(vendors, big)
inv = build_storage_inventory(vendors)
tf = storage_transparency_finding(inv)
if tf:
out["findings"].insert(0, tf)
out["summary"]["findings"] = len(out["findings"])
out["storage_inventory"] = inv
# ② Documentation Drift: Cookie-Richtlinie (Text) vs. Browser-Realität.
docs = snap.get("doc_entries") or []
cookie_text = next(
(e.get("text") or e.get("content") or "" for e in docs
if e.get("doc_type") in ("cookie", "cookie_richtlinie", "cookies")),
"",
)
out["drift"] = audit_cookie_compliance(
db, cookie_text, snap.get("banner_result"))
return out
finally:
db.close()
@router.get("/snapshots/{snapshot_id}/impressum-check")
async def snapshot_impressum_check(snapshot_id: str):
"""Impressum-Analyse aus dem Snapshot (kein Re-Crawl): laeuft den v3
ImpressumAgent auf dem gespeicherten Impressum-Text + Profil/Scope und
liefert den AgentOutput (Findings/Massnahmen/MC-Coverage) fuer den Tab."""
from fastapi import HTTPException
from database import SessionLocal
from compliance.services.check_snapshot import load_snapshot
from compliance.services.specialist_agents import REGISTRY, AgentInput
from compliance.api.agent_check._agent_outputs import (
impressum_input_from_snapshot,
)
db = SessionLocal()
try:
snap = load_snapshot(db, snapshot_id)
if not snap:
raise HTTPException(status_code=404, detail="snapshot not found")
agent_input = impressum_input_from_snapshot(snap)
if not agent_input:
return {"findings": [], "recommendations": [], "mc_coverage": [],
"notes": "kein Impressum-Text im Snapshot", "confidence": 0.0}
out = await REGISTRY.get("impressum").evaluate(AgentInput(**agent_input))
return out.model_dump(mode="json")
finally:
db.close()
@router.get("/snapshots/{snapshot_id}/dse-check")
async def snapshot_dse_check(snapshot_id: str):
"""DSE-Analyse aus dem Snapshot (kein Re-Crawl): laeuft den kuratierten
DSEAgent (Art. 13/14, ART13_CHECKLIST — KEIN Library-Firehose) auf dem
gespeicherten DSE-Text und liefert den AgentOutput fuer den Tab."""
from fastapi import HTTPException
from database import SessionLocal
from compliance.services.check_snapshot import load_snapshot
from compliance.services.specialist_agents import REGISTRY, AgentInput
from compliance.api.agent_check._agent_outputs import (
doc_input_from_snapshot,
)
db = SessionLocal()
try:
snap = load_snapshot(db, snapshot_id)
if not snap:
raise HTTPException(status_code=404, detail="snapshot not found")
agent_input = doc_input_from_snapshot(snap, "dse")
if not agent_input:
return {"findings": [], "recommendations": [], "mc_coverage": [],
"notes": "kein DSE-Text im Snapshot", "confidence": 0.0}
out = await REGISTRY.get("dse").evaluate(AgentInput(**agent_input))
return out.model_dump(mode="json")
finally:
db.close()
@router.get("/admin/benchmark")
async def benchmark(
industry: str = "",
@@ -0,0 +1,100 @@
"""Snapshot-getriebene Doc-Check-Endpoints (kein Re-Crawl).
Cookie-Library-Abgleich + v3-Doc-Agenten (Impressum/DSE/AGB …) laufen auf den
gespeicherten Snapshot-Texten. Ausgelagert aus agent_compliance_check_routes.py
(LOC-Budget). Gleicher Router-Prefix → identische Pfade, keine Contract-Änderung.
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, HTTPException
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/compliance/agent", tags=["agent-snapshots"])
async def _run_doc_agent(snapshot_id: str, doc_type: str, agent_id: str) -> dict:
"""Lädt den Snapshot, baut den AgentInput für doc_type und läuft den
registrierten v3-Doc-Agenten. Geteilt von impressum/dse/agb (kein Re-Crawl)."""
from database import SessionLocal
from compliance.services.check_snapshot import load_snapshot
from compliance.services.specialist_agents import REGISTRY, AgentInput
from compliance.api.agent_check._agent_outputs import doc_input_from_snapshot
db = SessionLocal()
try:
snap = load_snapshot(db, snapshot_id)
if not snap:
raise HTTPException(status_code=404, detail="snapshot not found")
agent_input = doc_input_from_snapshot(snap, doc_type)
if not agent_input:
return {"findings": [], "recommendations": [], "mc_coverage": [],
"notes": f"kein {doc_type}-Text im Snapshot", "confidence": 0.0}
out = await REGISTRY.get(agent_id).evaluate(AgentInput(**agent_input))
return out.model_dump(mode="json")
finally:
db.close()
@router.get("/snapshots/{snapshot_id}/cookie-check")
async def snapshot_cookie_check(snapshot_id: str):
"""Pro-Cookie-Abgleich der Snapshot-Vendors gegen cookie_knowledge_db."""
from database import SessionLocal
from compliance.services.check_snapshot import load_snapshot
from compliance.services.cookie_library_check import (
analyze_cookies, load_big_library,
)
from compliance.services.cookie_storage_inventory import (
build_storage_inventory, storage_transparency_finding,
)
from compliance.services.cookie_compliance_audit import (
audit_cookie_compliance,
)
db = SessionLocal()
try:
snap = load_snapshot(db, snapshot_id)
if not snap:
raise HTTPException(status_code=404, detail="snapshot not found")
vendors = snap.get("cmp_vendors") or []
names = [c.get("name", "")
for v in vendors for c in (v.get("cookies") or [])]
big = load_big_library(db, names)
out = analyze_cookies(vendors, big)
inv = build_storage_inventory(vendors)
tf = storage_transparency_finding(inv)
if tf:
out["findings"].insert(0, tf)
out["summary"]["findings"] = len(out["findings"])
out["storage_inventory"] = inv
# ② Documentation Drift: Cookie-Richtlinie (Text) vs. Browser-Realität.
docs = snap.get("doc_entries") or []
cookie_text = next(
(e.get("text") or e.get("content") or "" for e in docs
if e.get("doc_type") in ("cookie", "cookie_richtlinie", "cookies")),
"",
)
out["drift"] = audit_cookie_compliance(
db, cookie_text, snap.get("banner_result"))
return out
finally:
db.close()
@router.get("/snapshots/{snapshot_id}/impressum-check")
async def snapshot_impressum_check(snapshot_id: str):
"""Impressum-Analyse (v3 ImpressumAgent) auf dem gespeicherten Text."""
return await _run_doc_agent(snapshot_id, "impressum", "impressum")
@router.get("/snapshots/{snapshot_id}/dse-check")
async def snapshot_dse_check(snapshot_id: str):
"""DSE-Analyse (kuratierter DSEAgent, Art. 13/14) auf dem gespeicherten Text."""
return await _run_doc_agent(snapshot_id, "dse", "dse")
@router.get("/snapshots/{snapshot_id}/agb-check")
async def snapshot_agb_check(snapshot_id: str):
"""AGB-Analyse (kuratierter AGBAgent, §§ 305 ff. BGB) auf dem gespeicherten Text."""
return await _run_doc_agent(snapshot_id, "agb", "agb")