Benjamin Admin
7258744107
- 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>
101 lines
4.0 KiB
Python
101 lines
4.0 KiB
Python
"""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")
|