diff --git a/backend-compliance/compliance/api/agent_scan_helpers.py b/backend-compliance/compliance/api/agent_scan_helpers.py
new file mode 100644
index 0000000..5adcb43
--- /dev/null
+++ b/backend-compliance/compliance/api/agent_scan_helpers.py
@@ -0,0 +1,109 @@
+"""
+Agent scan helpers — summary builder and correction generator.
+Extracted from agent_scan_routes.py to keep route file under 500 LOC.
+"""
+
+import logging
+import os
+import re
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+
+async def add_corrections(findings: list, dse_text: str) -> None:
+ """Add correction suggestions for pre-launch mode via LLM."""
+ for finding in findings:
+ if finding.severity in ("HIGH", "MEDIUM") and "MISSING" in finding.code:
+ service_name = finding.code.replace("DSE-MISSING-", "").replace("_", " ").title()
+ try:
+ ollama_url = os.environ.get("OLLAMA_URL", "http://host.docker.internal:11434")
+ ollama_model = os.environ.get("OLLAMA_MODEL", "qwen3.5:35b-a3b")
+ async with httpx.AsyncClient(timeout=120.0) as client:
+ resp = await client.post(f"{ollama_url}/api/generate", json={
+ "model": ollama_model,
+ "prompt": (
+ f"Erstelle einen einbaufertigen Textbaustein fuer eine deutsche "
+ f"Datenschutzerklaerung fuer den Dienst '{service_name}'. "
+ f"Enthalte: Ueberschrift, Anbietername mit Sitz, Zweck der Verarbeitung, "
+ f"Rechtsgrundlage nach DSGVO, Drittlandtransfer-Hinweis wenn noetig, "
+ f"Widerspruchsmoeglichkeit. Max 150 Woerter. "
+ f"Antworte NUR mit dem fertigen Textbaustein."
+ ),
+ "stream": False,
+ })
+ data = resp.json()
+ raw = data.get("response", "").strip()
+ raw = re.sub(r".*?", "", raw, flags=re.DOTALL).strip()
+ if raw and len(raw) > 50:
+ finding.correction = raw
+ except Exception as e:
+ logger.warning("Correction generation failed for %s: %s", service_name, e)
+
+
+def build_scan_summary(
+ url: str, scan, comparison: dict, findings: list, is_live: bool,
+ discovered_docs: list | None = None,
+) -> str:
+ """Build German scan summary including DSI document results."""
+ mode = "PRUEFUNG LIVE-WEBSITE" if is_live else "INTERNE PRUEFUNG"
+ n_undoc = len(comparison["undocumented"])
+ n_ok = len(comparison["documented"])
+ n_outdated = len(comparison["outdated"])
+ n_findings = len(findings)
+ high = sum(1 for f in findings if f.severity == "HIGH")
+
+ parts = [
+ f"{mode} — Website-Scan",
+ f"URL: {url}",
+ f"Seiten gescannt: {len(scan.pages_scanned)}",
+ ]
+ for page in scan.pages_scanned:
+ status = scan.missing_pages.get(page, 200)
+ marker = "\u2717" if status >= 400 else "\u2713"
+ parts.append(f" {marker} {page}" + (f" (HTTP {status})" if status >= 400 else ""))
+ parts.extend([
+ "",
+ "Dienstleister-Abgleich (DSE vs. Website):",
+ f" Korrekt dokumentiert: {n_ok}",
+ f" NICHT in DSE (Verstoss): {n_undoc}",
+ f" Veraltet in DSE: {n_outdated}",
+ "",
+ f"Findings: {n_findings} ({high} mit hoher Prioritaet)",
+ ])
+
+ # DSI Documents section
+ if discovered_docs:
+ parts.extend([
+ "",
+ f"Rechtliche Dokumente gefunden: {len(discovered_docs)}",
+ ])
+ for doc in discovered_docs:
+ pct = doc.completeness_pct if hasattr(doc, 'completeness_pct') else 0
+ fc = doc.findings_count if hasattr(doc, 'findings_count') else 0
+ wc = doc.word_count if hasattr(doc, 'word_count') else 0
+ status = "OK" if pct >= 80 else "LUECKENHAFT" if pct >= 50 else "MANGELHAFT"
+ dt = doc.doc_type if hasattr(doc, 'doc_type') else "unknown"
+ title = doc.title if hasattr(doc, 'title') else "?"
+ parts.append(
+ f" [{status}] {title} ({dt}, {wc} Woerter, "
+ f"{pct}% vollstaendig, {fc} Maengel)"
+ )
+
+ if findings:
+ parts.append("")
+ for f in findings[:20]:
+ sev = f.severity if hasattr(f, 'severity') else "?"
+ txt = f.text if hasattr(f, 'text') else str(f)
+ marker = "!!" if sev == "HIGH" else "!" if sev == "MEDIUM" else "i"
+ parts.append(f" [{marker}] {txt}")
+
+ if is_live and high > 0:
+ parts.extend([
+ "",
+ "ACHTUNG: Verstoesse auf einer bereits veroeffentlichten Website. "
+ "Sofortige Korrektur empfohlen.",
+ ])
+
+ return "\n".join(parts)
diff --git a/backend-compliance/compliance/api/agent_scan_routes.py b/backend-compliance/compliance/api/agent_scan_routes.py
index 7b00bfc..154b776 100644
--- a/backend-compliance/compliance/api/agent_scan_routes.py
+++ b/backend-compliance/compliance/api/agent_scan_routes.py
@@ -16,6 +16,13 @@ from pydantic import BaseModel
from compliance.services.website_scanner import scan_website, DetectedService
from compliance.services.dse_service_extractor import extract_dse_services, compare_services
from compliance.services.smtp_sender import send_email
+from compliance.services.dse_parser import parse_dse
+from compliance.services.dse_matcher import build_text_references, TextReference
+from compliance.services.mandatory_content_checker import (
+ check_mandatory_documents, check_dse_mandatory_content, MandatoryFinding,
+)
+from compliance.services.legal_basis_validator import validate_legal_bases
+from compliance.api.agent_scan_helpers import add_corrections, build_scan_summary
logger = logging.getLogger(__name__)
@@ -49,11 +56,37 @@ class ServiceInfo(BaseModel):
status: str # "ok", "undocumented", "outdated"
+class TextReferenceModel(BaseModel):
+ found: bool = False
+ source_url: str = ""
+ document_type: str = "Datenschutzerklaerung"
+ section_heading: str = ""
+ section_number: str = ""
+ parent_section: str = ""
+ paragraph_index: int = 0
+ original_text: str = ""
+ issue: str = ""
+ correction_type: str = ""
+ correction_text: str = ""
+ insert_after: str = ""
+
+
class ScanFinding(BaseModel):
code: str
severity: str
text: str
correction: str = ""
+ text_reference: TextReferenceModel | None = None
+
+
+class DiscoveredDocument(BaseModel):
+ title: str
+ url: str
+ doc_type: str
+ language: str = ""
+ word_count: int = 0
+ completeness_pct: int = 0
+ findings_count: int = 0
class ScanResponse(BaseModel):
@@ -62,6 +95,7 @@ class ScanResponse(BaseModel):
pages_list: list[str] = []
services: list[ServiceInfo]
findings: list[ScanFinding]
+ discovered_documents: list[DiscoveredDocument] = []
ai_detected: bool
chatbot_detected: bool
chatbot_provider: str
@@ -76,30 +110,178 @@ async def scan_website_endpoint(req: ScanRequest):
"""Deep website scan: multi-page crawl + SOLL/IST service comparison."""
is_live = req.mode == "post_launch"
- # Step 1: Scan website (5-10 pages)
- scan = await scan_website(req.url)
+ # Step 1: Scan website — try Playwright first (JS-rendered), fallback to httpx
+ playwright_htmls: dict[str, str] = {}
+ try:
+ async with httpx.AsyncClient(timeout=120.0) as pw_client:
+ pw_resp = await pw_client.post(
+ "http://bp-compliance-consent-tester:8094/website-scan",
+ json={"url": req.url, "max_pages": 15, "click_nav": True},
+ )
+ if pw_resp.status_code == 200:
+ pw_data = pw_resp.json()
+ playwright_htmls = pw_data.get("page_htmls", {})
+ logger.info("Playwright scan: %d pages, %d scripts",
+ pw_data.get("pages_count", 0), len(pw_data.get("external_scripts", [])))
+ except Exception as e:
+ logger.warning("Playwright scanner unavailable, falling back to httpx: %s", e)
+
+ # Use Playwright results if available, otherwise fall back to httpx scanner
+ if playwright_htmls:
+ # Build ScanResult from Playwright data
+ from compliance.services.website_scanner import ScanResult, DetectedService, _detect_services, _detect_ai_mentions
+ from compliance.services.service_registry import SERVICE_REGISTRY
+ scan = ScanResult()
+ scan.pages_scanned = list(playwright_htmls.keys())
+ for page_url, html in playwright_htmls.items():
+ _detect_services(html, page_url, scan)
+ _detect_ai_mentions(html, page_url, scan)
+ # Deduplicate
+ seen = set()
+ unique = []
+ for svc in scan.detected_services:
+ if svc.id not in seen:
+ seen.add(svc.id)
+ unique.append(svc)
+ scan.detected_services = unique
+ scan.chatbot_detected = any(s.category == "chatbot" for s in scan.detected_services)
+ if scan.chatbot_detected:
+ scan.chatbot_provider = next(s.name for s in scan.detected_services if s.category == "chatbot")
+ else:
+ scan = await scan_website(req.url)
+
logger.info("Scanned %d pages, found %d services", len(scan.pages_scanned), len(scan.detected_services))
- # Step 2: Fetch privacy policy text for SOLL extraction
- dse_text = await _fetch_dse_text(req.url, scan.pages_scanned)
+ # Step 1b: DSI Discovery — find all legal documents on the website
+ discovered_docs: list[DiscoveredDocument] = []
+ dsi_findings: list[ScanFinding] = []
+ try:
+ async with httpx.AsyncClient(timeout=180.0) as dsi_client:
+ dsi_resp = await dsi_client.post(
+ "http://bp-compliance-consent-tester:8094/dsi-discovery",
+ json={"url": req.url, "max_documents": 20},
+ )
+ if dsi_resp.status_code == 200:
+ dsi_data = dsi_resp.json()
+ logger.info("DSI discovery: %d documents found", dsi_data.get("total_found", 0))
- # Step 3: Extract services mentioned in DSE via LLM
+ # Check each document against its legal requirements
+ from compliance.services.dsi_document_checker import (
+ check_document_completeness, classify_document_type,
+ )
+ for doc in dsi_data.get("documents", []):
+ doc_type = classify_document_type(doc["title"], doc["url"])
+ doc_findings = check_document_completeness(
+ doc.get("text_preview", ""), doc_type, doc["title"], doc["url"],
+ )
+ # Count completeness
+ score_finding = next((f for f in doc_findings if "SCORE" in f.get("code", "")), None)
+ completeness = 0
+ if score_finding:
+ import re as _re2
+ pct_match = _re2.search(r"(\d+)%", score_finding.get("text", ""))
+ if pct_match:
+ completeness = int(pct_match.group(1))
+
+ discovered_docs.append(DiscoveredDocument(
+ title=doc["title"], url=doc["url"],
+ doc_type=doc_type, language=doc.get("language", ""),
+ word_count=doc.get("word_count", 0),
+ completeness_pct=completeness,
+ findings_count=len([f for f in doc_findings if "SCORE" not in f.get("code", "")]),
+ ))
+ for df in doc_findings:
+ if "SCORE" not in df.get("code", ""):
+ dsi_findings.append(ScanFinding(
+ code=df["code"], severity=df["severity"], text=df["text"],
+ ))
+ except Exception as e:
+ logger.warning("DSI discovery failed: %s", e)
+
+ # Step 2: Fetch privacy policy text (from Playwright HTMLs or httpx)
+ dse_text = ""
+ for page_url, html in playwright_htmls.items():
+ if re.search(r"datenschutz|privacy|dsgvo", page_url, re.IGNORECASE):
+ import re as _re
+ clean = _re.sub(r"<(script|style)[^>]*>.*?", "", html, flags=_re.DOTALL | _re.IGNORECASE)
+ clean = _re.sub(r"<[^>]+>", " ", clean)
+ clean = _re.sub(r"\s+", " ", clean).strip()
+ dse_text = clean[:4000]
+ break
+ if not dse_text:
+ dse_text = await _fetch_dse_text(req.url, scan.pages_scanned)
+
+ # Step 3: Extract services mentioned in DSE via LLM + text fallback
dse_services = await extract_dse_services(dse_text) if dse_text else []
- logger.info("DSE mentions %d services", len(dse_services))
+ logger.info("DSE mentions %d services (LLM)", len(dse_services))
- # Step 4: SOLL/IST comparison
+ # Fallback: if LLM extraction failed, search DSE text directly for service names
+ if not dse_services and dse_text:
+ dse_lower = dse_text.lower()
+ detected_dicts_for_check = [_service_to_dict(s) for s in scan.detected_services]
+ for svc in detected_dicts_for_check:
+ name = svc.get("name", "").lower()
+ # Check if service name appears in DSE text
+ if name and len(name) > 3 and name in dse_lower:
+ dse_services.append({"name": svc["name"], "purpose": "", "country": svc.get("country", ""), "legal_basis": ""})
+ if dse_services:
+ logger.info("DSE text fallback found %d services", len(dse_services))
+
+ # Step 4: Parse DSE into structured sections (prefer Playwright HTML)
+ dse_html = ""
+ for page_url, html in playwright_htmls.items():
+ if re.search(r"datenschutz|privacy|dsgvo", page_url, re.IGNORECASE):
+ dse_html = html
+ break
+ if not dse_html:
+ dse_html = await _fetch_dse_html(req.url, scan.pages_scanned)
+ dse_sections = parse_dse(dse_html, req.url) if dse_html else []
+ logger.info("Parsed %d DSE sections", len(dse_sections))
+
+ # Step 5: SOLL/IST comparison
detected_dicts = [_service_to_dict(s) for s in scan.detected_services]
comparison = compare_services(detected_dicts, dse_services)
- # Step 5: Generate findings
- services_info, findings = _build_findings(comparison, scan, is_live)
+ # Step 6: Build TextReferences for each detected service
+ text_refs = build_text_references(detected_dicts, dse_services, dse_sections, req.url)
- # Step 6: Generate corrections for pre-launch mode
+ # Step 7: Generate findings with text references
+ services_info, findings = _build_findings(comparison, scan, is_live, text_refs)
+
+ # Step 8: Check mandatory content (documents + DSE sections)
+ mandatory_findings = check_mandatory_documents(scan.pages_scanned, scan.missing_pages)
+ mandatory_findings += check_dse_mandatory_content(dse_sections, dse_text)
+ for mf in mandatory_findings:
+ findings.append(ScanFinding(
+ code=mf.code, severity=mf.severity,
+ text=f"{mf.text}" + (f" — {mf.suggestion}" if mf.suggestion else ""),
+ ))
+
+ # Step 8b: Validate legal bases (lit. a-f) in DSE
+ if dse_text:
+ lit_findings = validate_legal_bases(dse_text)
+ for lf in lit_findings:
+ findings.append(ScanFinding(
+ code=f"LIT-{lf.purpose.upper()}",
+ severity=lf.severity,
+ text=lf.text,
+ text_reference=TextReferenceModel(
+ found=True, source_url=req.url,
+ original_text=lf.original_text,
+ issue="incorrect", correction_type="replace",
+ correction_text=f"Korrekte Rechtsgrundlage: {lf.correct_basis} ({lf.legal_ref})",
+ ) if lf.original_text else None,
+ ))
+
+ # Step 8c: Add DSI document findings
+ findings.extend(dsi_findings)
+
+ # Step 9: Generate corrections for pre-launch mode
if not is_live and findings:
- await _add_corrections(findings, dse_text)
+ await add_corrections(findings, dse_text)
# Step 7: Build summary
- summary = _build_scan_summary(req.url, scan, comparison, findings, is_live)
+ summary = build_scan_summary(req.url, scan, comparison, findings, is_live, discovered_docs)
# Step 8: Send notification
mode_label = "INTERNE PRUEFUNG" if not is_live else "LIVE-WEBSITE"
@@ -115,6 +297,7 @@ async def scan_website_endpoint(req: ScanRequest):
pages_list=scan.pages_scanned,
services=services_info,
findings=findings,
+ discovered_documents=discovered_docs,
ai_detected=len(scan.ai_mentions) > 0,
chatbot_detected=scan.chatbot_detected,
chatbot_provider=scan.chatbot_provider,
@@ -149,6 +332,24 @@ async def _fetch_dse_text(url: str, scanned_pages: list[str]) -> str:
return ""
+async def _fetch_dse_html(url: str, scanned_pages: list[str]) -> str:
+ """Fetch the raw HTML of the privacy policy page (for structured parsing)."""
+ import re
+ dse_url = None
+ for page in scanned_pages:
+ if re.search(r"datenschutz|privacy|dsgvo", page, re.IGNORECASE):
+ dse_url = page
+ break
+ if not dse_url:
+ dse_url = url
+ try:
+ async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
+ resp = await client.get(dse_url, headers={"User-Agent": "BreakPilot-Compliance-Agent/1.0"})
+ return resp.text
+ except Exception:
+ return ""
+
+
def _service_to_dict(svc: DetectedService) -> dict:
return {
"id": svc.id, "name": svc.name, "category": svc.category,
@@ -159,11 +360,25 @@ def _service_to_dict(svc: DetectedService) -> dict:
def _build_findings(
- comparison: dict, scan, is_live: bool,
+ comparison: dict, scan, is_live: bool, text_refs: dict | None = None,
) -> tuple[list[ServiceInfo], list[ScanFinding]]:
"""Build service info list and findings from comparison."""
services = []
findings = []
+ text_refs = text_refs or {}
+
+ def _get_ref(svc_id: str) -> TextReferenceModel | None:
+ ref = text_refs.get(svc_id)
+ if not ref:
+ return None
+ return TextReferenceModel(
+ found=ref.found, source_url=ref.source_url,
+ document_type=ref.document_type, section_heading=ref.section_heading,
+ section_number=ref.section_number, parent_section=ref.parent_section,
+ paragraph_index=ref.paragraph_index, original_text=ref.original_text,
+ issue=ref.issue, correction_type=ref.correction_type,
+ correction_text=ref.correction_text, insert_after=ref.insert_after,
+ )
# Undocumented services (on website, NOT in DSE)
for svc in comparison["undocumented"]:
@@ -175,12 +390,14 @@ def _build_findings(
legal_ref=svc.get("legal_ref", ""), in_dse=False, status="undocumented",
))
severity = "HIGH" if is_live else "MEDIUM"
+ ref = _get_ref(svc.get("id", ""))
findings.append(ScanFinding(
code=f"DSE-MISSING-{svc['id'].upper()}",
severity=severity,
text=f"{svc['name']} ({svc.get('provider', '')}, {svc.get('country', '')}) "
f"ist auf der Website eingebunden aber NICHT in der Datenschutzerklaerung "
f"dokumentiert (Art. 13 DSGVO).",
+ text_reference=ref,
))
# Documented services (OK)
@@ -229,79 +446,3 @@ def _build_findings(
return services, findings
-async def _add_corrections(findings: list[ScanFinding], dse_text: str) -> None:
- """Add correction suggestions for pre-launch mode via LLM."""
- for finding in findings:
- if finding.severity in ("HIGH", "MEDIUM") and "MISSING" in finding.code:
- service_name = finding.code.replace("DSE-MISSING-", "").replace("_", " ").title()
- try:
- # Call Ollama directly (bypasses SDK RBAC + Think-mode issues)
- ollama_url = os.environ.get("OLLAMA_URL", "http://host.docker.internal:11434")
- ollama_model = os.environ.get("OLLAMA_MODEL", "qwen3.5:35b-a3b")
- async with httpx.AsyncClient(timeout=120.0) as client:
- resp = await client.post(f"{ollama_url}/api/generate", json={
- "model": ollama_model,
- "prompt": (
- f"Erstelle einen einbaufertigen Textbaustein fuer eine deutsche "
- f"Datenschutzerklaerung fuer den Dienst '{service_name}'. "
- f"Enthalte: Ueberschrift, Anbietername mit Sitz, Zweck der Verarbeitung, "
- f"Rechtsgrundlage nach DSGVO, Drittlandtransfer-Hinweis wenn noetig, "
- f"Widerspruchsmoeglichkeit. Max 150 Woerter. "
- f"Antworte NUR mit dem fertigen Textbaustein."
- ),
- "stream": False,
- })
- data = resp.json()
- import re
- raw = data.get("response", "").strip()
- raw = re.sub(r".*?", "", raw, flags=re.DOTALL).strip()
- if raw and len(raw) > 50:
- finding.correction = raw
- except Exception as e:
- logger.warning("Correction generation failed for %s: %s", service_name, e)
-
-
-def _build_scan_summary(
- url: str, scan, comparison: dict, findings: list[ScanFinding], is_live: bool,
-) -> str:
- """Build German scan summary."""
- mode = "PRUEFUNG LIVE-WEBSITE" if is_live else "INTERNE PRUEFUNG"
- n_undoc = len(comparison["undocumented"])
- n_ok = len(comparison["documented"])
- n_outdated = len(comparison["outdated"])
- n_findings = len(findings)
- high = sum(1 for f in findings if f.severity == "HIGH")
-
- parts = [
- f"{mode} — Website-Scan",
- f"URL: {url}",
- f"Seiten gescannt: {len(scan.pages_scanned)}",
- ]
- for page in scan.pages_scanned:
- status = scan.missing_pages.get(page, 200)
- marker = "✗" if status >= 400 else "✓"
- parts.append(f" {marker} {page}" + (f" (HTTP {status})" if status >= 400 else ""))
- parts.extend([
- "",
- f"Dienstleister-Abgleich (DSE vs. Website):",
- f" Korrekt dokumentiert: {n_ok}",
- f" NICHT in DSE (Verstoss): {n_undoc}",
- f" Veraltet in DSE: {n_outdated}",
- "",
- f"Findings: {n_findings} ({high} mit hoher Prioritaet)",
- ])
-
- if findings:
- parts.append("")
- for f in findings[:10]:
- marker = "!!" if f.severity == "HIGH" else "!" if f.severity == "MEDIUM" else "i"
- parts.append(f" [{marker}] {f.text}")
-
- if is_live and high > 0:
- parts.extend([
- "",
- "ACHTUNG: Verstoesse auf einer bereits veroeffentlichten Website. "
- "Sofortige Korrektur empfohlen.",
- ])
-
- return "\n".join(parts)
diff --git a/backend-compliance/compliance/services/dsi_document_checker.py b/backend-compliance/compliance/services/dsi_document_checker.py
new file mode 100644
index 0000000..70eb56f
--- /dev/null
+++ b/backend-compliance/compliance/services/dsi_document_checker.py
@@ -0,0 +1,229 @@
+"""
+DSI Document Checker — validates discovered legal documents against
+mandatory content requirements.
+
+Checks each document type against its specific legal requirements:
+- Datenschutzinformation: Art. 13/14 DSGVO (9 Pflichtangaben)
+- AGB: §305ff BGB
+- Widerrufsbelehrung: §355, §312g BGB
+- Cookie-Richtlinie: §25 TDDDG
+- Impressum: §5 TMG / §18 MStV
+"""
+
+import logging
+import re
+
+logger = logging.getLogger(__name__)
+
+
+# Art. 13 DSGVO mandatory fields for privacy policies
+ART13_CHECKLIST = [
+ {
+ "id": "controller",
+ "label": "Verantwortlicher (Art. 13(1)(a))",
+ "patterns": [
+ r"verantwortlich\w*\s+(?:ist|im sinne|fuer)",
+ r"controller", r"verantwortliche\s+stelle",
+ r"responsible\s+(?:party|for)",
+ ],
+ "severity": "HIGH",
+ },
+ {
+ "id": "dpo",
+ "label": "Datenschutzbeauftragter (Art. 13(1)(b))",
+ "patterns": [
+ r"datenschutzbeauftragt", r"data\s+protection\s+officer",
+ r"dsb", r"dpo",
+ ],
+ "severity": "MEDIUM",
+ },
+ {
+ "id": "purposes",
+ "label": "Zwecke der Verarbeitung (Art. 13(1)(c))",
+ "patterns": [
+ r"zweck\w*\s+(?:der|die)\s+(?:verarbeitung|datenerhebung|datenverarbeitung)",
+ r"purpose\w*\s+(?:of|for)\s+(?:processing|data)",
+ r"zu\s+welch\w+\s+zweck",
+ ],
+ "severity": "HIGH",
+ },
+ {
+ "id": "legal_basis",
+ "label": "Rechtsgrundlage (Art. 13(1)(c))",
+ "patterns": [
+ r"rechtsgrundlage", r"art\.\s*6\s*(?:abs|absatz)?\s*\.?\s*1",
+ r"legal\s+basis", r"berechtigtes\s+interesse",
+ ],
+ "severity": "HIGH",
+ },
+ {
+ "id": "recipients",
+ "label": "Empfaenger (Art. 13(1)(e))",
+ "patterns": [
+ r"empf(?:ae|ä)nger", r"(?:ueber|weiter)mitt(?:el|l)ung",
+ r"recipient", r"weitergabe\s+(?:an|von)\s+daten",
+ r"dritte", r"third\s+part",
+ ],
+ "severity": "MEDIUM",
+ },
+ {
+ "id": "third_country",
+ "label": "Drittlandtransfer (Art. 13(1)(f))",
+ "patterns": [
+ r"drittland", r"dritt\s*staat", r"drittl(?:ae|ä)nder",
+ r"third\s+countr", r"angemessenheitsbeschluss",
+ r"standard\s*vertragsklausel", r"scc",
+ ],
+ "severity": "MEDIUM",
+ },
+ {
+ "id": "retention",
+ "label": "Speicherdauer (Art. 13(2)(a))",
+ "patterns": [
+ r"speicherdauer", r"aufbewahrungsfrist",
+ r"(?:wie\s+lange|dauer)\s+(?:werden|gespeichert)",
+ r"retention\s+period", r"l(?:oe|ö)sch(?:ung|frist|konzept)",
+ ],
+ "severity": "HIGH",
+ },
+ {
+ "id": "rights",
+ "label": "Betroffenenrechte (Art. 13(2)(b))",
+ "patterns": [
+ r"recht\s+auf\s+auskunft", r"recht\s+auf\s+l(?:oe|ö)schung",
+ r"recht\s+auf\s+berichtigung", r"widerspruchsrecht",
+ r"art\.\s*1[5-9]", r"art\.\s*2[0-2]",
+ r"right\s+to\s+(?:access|erasure|rectification|object)",
+ ],
+ "severity": "HIGH",
+ },
+ {
+ "id": "complaint",
+ "label": "Beschwerderecht (Art. 13(2)(d))",
+ "patterns": [
+ r"beschwerderecht", r"aufsichtsbeh(?:oe|ö)rde",
+ r"right\s+to\s+lodge\s+a\s+complaint",
+ r"supervisory\s+authority", r"datenschutzbeh(?:oe|ö)rde",
+ ],
+ "severity": "MEDIUM",
+ },
+]
+
+# §355 BGB requirements for cancellation/withdrawal policies
+WIDERRUF_CHECKLIST = [
+ {"id": "right_info", "label": "Belehrung ueber Widerrufsrecht",
+ "patterns": [r"widerrufsrecht", r"right\s+of\s+withdrawal", r"recht\s+(?:zum|auf)\s+widerruf"]},
+ {"id": "deadline", "label": "Widerrufsfrist (14 Tage)",
+ "patterns": [r"14\s+tage", r"vierzehn\s+tage", r"14\s+days", r"fourteen\s+days"]},
+ {"id": "form", "label": "Form des Widerrufs",
+ "patterns": [r"widerrufsformular", r"muster.?widerruf", r"withdrawal\s+form", r"formular"]},
+ {"id": "consequences", "label": "Folgen des Widerrufs",
+ "patterns": [r"folgen\s+des\s+widerrufs", r"consequences\s+of\s+withdrawal", r"rueckerstattung"]},
+]
+
+# AGB minimal requirements
+AGB_CHECKLIST = [
+ {"id": "scope", "label": "Geltungsbereich",
+ "patterns": [r"geltungsbereich", r"geltung", r"scope", r"diese\s+(?:agb|bedingungen)\s+gelten"]},
+ {"id": "contract", "label": "Vertragsschluss",
+ "patterns": [r"vertragsschluss", r"zustandekommen", r"contract\s+formation", r"angebot\s+und\s+annahme"]},
+ {"id": "liability", "label": "Haftung",
+ "patterns": [r"haftung", r"liability", r"schadensersatz", r"haftungsbeschr(?:ae|ä)nkung"]},
+ {"id": "jurisdiction", "label": "Gerichtsstand / Anwendbares Recht",
+ "patterns": [r"gerichtsstand", r"anwendbares\s+recht", r"jurisdiction", r"governing\s+law"]},
+]
+
+
+def check_document_completeness(
+ text: str,
+ doc_type: str,
+ doc_title: str,
+ doc_url: str,
+) -> list[dict]:
+ """Check a legal document against its type-specific requirements.
+
+ Returns a list of findings (missing/present fields).
+ """
+ findings = []
+ text_lower = text.lower()
+
+ if not text or len(text) < 50:
+ findings.append({
+ "code": f"DSI-EMPTY-{doc_type.upper()}",
+ "severity": "HIGH",
+ "text": f"Dokument '{doc_title}' ist leer oder zu kurz fuer eine Pruefung.",
+ "doc_title": doc_title,
+ "doc_url": doc_url,
+ "doc_type": doc_type,
+ })
+ return findings
+
+ # Select checklist based on document type
+ if doc_type in ("dse", "datenschutz", "privacy"):
+ checklist = ART13_CHECKLIST
+ label = "Art. 13 DSGVO"
+ elif doc_type in ("widerruf", "withdrawal", "cancellation"):
+ checklist = WIDERRUF_CHECKLIST
+ label = "§355 BGB"
+ elif doc_type in ("agb", "terms", "nutzungsbedingungen"):
+ checklist = AGB_CHECKLIST
+ label = "§305ff BGB"
+ else:
+ checklist = ART13_CHECKLIST # Default: check as DSE
+ label = "Art. 13 DSGVO"
+
+ present = 0
+ total = len(checklist)
+ for check in checklist:
+ found = any(re.search(p, text_lower) for p in check["patterns"])
+ if not found:
+ findings.append({
+ "code": f"DSI-MISSING-{check['id'].upper()}",
+ "severity": check.get("severity", "MEDIUM"),
+ "text": (
+ f"'{doc_title}': Pflichtangabe '{check['label']}' nicht gefunden. "
+ f"Erforderlich nach {label}."
+ ),
+ "doc_title": doc_title,
+ "doc_url": doc_url,
+ "doc_type": doc_type,
+ "check_id": check["id"],
+ })
+ else:
+ present += 1
+
+ # Add summary finding
+ if total > 0:
+ pct = round(present / total * 100)
+ if pct < 100:
+ findings.insert(0, {
+ "code": f"DSI-SCORE-{doc_type.upper()}",
+ "severity": "LOW" if pct >= 80 else "MEDIUM" if pct >= 50 else "HIGH",
+ "text": (
+ f"'{doc_title}': {present}/{total} Pflichtangaben vorhanden ({pct}%). "
+ f"Fehlend: {total - present} Angaben nach {label}."
+ ),
+ "doc_title": doc_title,
+ "doc_url": doc_url,
+ "doc_type": doc_type,
+ })
+
+ return findings
+
+
+def classify_document_type(title: str, url: str) -> str:
+ """Classify a document by its title/URL into a legal document type."""
+ combined = f"{title} {url}".lower()
+
+ if any(kw in combined for kw in ["datenschutz", "privacy", "dsgvo", "data protection", "données"]):
+ return "dse"
+ if any(kw in combined for kw in ["widerruf", "withdrawal", "rétractation", "desistimiento"]):
+ return "widerruf"
+ if any(kw in combined for kw in ["agb", "allgemeine geschäftsbedingungen", "terms",
+ "nutzungsbedingungen", "conditions"]):
+ return "agb"
+ if any(kw in combined for kw in ["cookie", "slapuk", "evästeet", "kakor"]):
+ return "cookie"
+ if any(kw in combined for kw in ["impressum", "imprint", "legal notice", "mentions légales"]):
+ return "impressum"
+ return "other"