Benjamin Admin
07cc00da11
Extends CompliancePDFGenerator with a "Quellen & Lizenzen" section
appended to every generated compliance PDF.
The footer is built from compliance.canonical_controls + control_parent_links
directly (no HTTP hop to /licenses/aggregate — same DB connection
already open in the generator). It groups by license_rule and lists
the top 8 source regulations per bucket.
For Rule-2 entries (CC-BY-SA, OECD-Public, Apache, etc.) it emits the
mandatory attribution paragraph required by the underlying licenses.
For Rule 1 a brief reference list satisfies the auditability goal
without legal obligation. Rule 3 is identifier-only by design.
Architecture decision: this is a PLATFORM-level footer (which sources
the platform draws on overall), not a per-export filter of "only the
sources actually cited in THIS document". The latter would require
control-uuid tracking across all sections (TOM/VVT/DSFA/etc.) which
the current PDF generator does not surface — that's a follow-up scope.
The platform-level footer fulfils the immediate legal mandate that
attribution be present on the work, not buried in AGB/Impressum.
Part of Attribution-Renderer Task #23. Stufe 1 (overview page) +
Stufe 3 (SourceBadge component) already shipped in commit dfac940.
Stufe 4 (tech-file appendix) remains for the IACE tech-file generator
in a separate iteration.
280 lines
13 KiB
Python
280 lines
13 KiB
Python
"""
|
|
Compliance Report PDF Generator — generates a comprehensive A4 PDF
|
|
covering all compliance modules for a project.
|
|
|
|
Uses reportlab (same as audit_pdf_generator.py).
|
|
"""
|
|
|
|
import io
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from reportlab.lib import colors
|
|
from reportlab.lib.pagesizes import A4
|
|
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
|
|
from reportlab.lib.units import mm
|
|
from reportlab.platypus import (
|
|
SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, PageBreak,
|
|
)
|
|
from sqlalchemy import text
|
|
from sqlalchemy.orm import Session
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Colors
|
|
PURPLE = colors.HexColor("#7c3aed")
|
|
LIGHT_PURPLE = colors.HexColor("#f5f3ff")
|
|
GRAY = colors.HexColor("#6b7280")
|
|
GREEN = colors.HexColor("#16a34a")
|
|
RED = colors.HexColor("#dc2626")
|
|
YELLOW = colors.HexColor("#ca8a04")
|
|
|
|
|
|
def _styles():
|
|
ss = getSampleStyleSheet()
|
|
ss.add(ParagraphStyle("Title2", parent=ss["Title"], fontSize=24, textColor=PURPLE, spaceAfter=6))
|
|
ss.add(ParagraphStyle("Section", parent=ss["Heading2"], fontSize=14, textColor=PURPLE, spaceBefore=12, spaceAfter=6))
|
|
ss.add(ParagraphStyle("Body2", parent=ss["Normal"], fontSize=10, leading=14, spaceAfter=4))
|
|
ss.add(ParagraphStyle("Small", parent=ss["Normal"], fontSize=8, textColor=GRAY))
|
|
return ss
|
|
|
|
|
|
class CompliancePDFGenerator:
|
|
"""Generates a full compliance status report as PDF."""
|
|
|
|
def __init__(self, db: Session) -> None:
|
|
self.db = db
|
|
|
|
def generate(self, tenant_id: str, project_id: str | None = None, language: str = "de") -> tuple[bytes, str]:
|
|
buf = io.BytesIO()
|
|
doc = SimpleDocTemplate(buf, pagesize=A4, leftMargin=20 * mm, rightMargin=20 * mm, topMargin=25 * mm, bottomMargin=20 * mm)
|
|
ss = _styles()
|
|
story: list = []
|
|
|
|
now = datetime.now(timezone.utc)
|
|
story.append(Paragraph("Compliance-Report", ss["Title2"]))
|
|
story.append(Paragraph(f"Stand: {now.strftime('%d.%m.%Y %H:%M')} UTC", ss["Small"]))
|
|
story.append(Spacer(1, 10 * mm))
|
|
|
|
# Company Profile
|
|
self._add_company_section(story, ss, tenant_id, project_id)
|
|
# TOM
|
|
self._add_count_section(story, ss, "TOM (Technisch-Organisatorische Massnahmen)",
|
|
"compliance_toms", tenant_id)
|
|
# VVT
|
|
self._add_count_section(story, ss, "VVT (Verarbeitungstaetigkeiten)",
|
|
"compliance_vvt_activities", tenant_id)
|
|
# DSFA
|
|
self._add_count_section(story, ss, "Datenschutz-Folgenabschaetzungen",
|
|
"compliance_dsfa_assessments", tenant_id)
|
|
# Risks
|
|
self._add_risk_section(story, ss, tenant_id)
|
|
# Vendors
|
|
self._add_count_section(story, ss, "Auftragsverarbeiter",
|
|
"compliance_vendor_assessments", tenant_id)
|
|
# Incidents
|
|
self._add_count_section(story, ss, "Datenschutz-Vorfaelle",
|
|
"compliance_notfallplan_incidents", tenant_id)
|
|
# Document Reviews
|
|
self._add_review_section(story, ss, tenant_id)
|
|
# Banner Consents
|
|
self._add_consent_section(story, ss, tenant_id)
|
|
# Org Roles
|
|
self._add_role_section(story, ss, tenant_id, project_id)
|
|
# Stufe 2 — Quellen- und Lizenz-Footer (Attribution-Renderer Task #23)
|
|
self._add_attribution_footer(story, ss)
|
|
# Footer
|
|
story.append(Spacer(1, 15 * mm))
|
|
story.append(Paragraph("Erstellt mit BreakPilot Compliance SDK", ss["Small"]))
|
|
|
|
doc.build(story)
|
|
filename = f"compliance-report-{now.strftime('%Y%m%d')}.pdf"
|
|
return buf.getvalue(), filename
|
|
|
|
def _add_company_section(self, story, ss, tid, pid):
|
|
story.append(Paragraph("Unternehmensprofil", ss["Section"]))
|
|
try:
|
|
where = "tenant_id = :tid"
|
|
params: dict[str, Any] = {"tid": tid}
|
|
if pid:
|
|
where += " AND project_id = :pid"
|
|
params["pid"] = pid
|
|
row = self.db.execute(text(f"SELECT * FROM compliance_company_profiles WHERE {where} LIMIT 1"), params).fetchone()
|
|
if row:
|
|
d = dict(row._mapping)
|
|
data = [
|
|
["Feld", "Wert"],
|
|
["Firma", d.get("company_name", "-")],
|
|
["Branche", d.get("industry", "-")],
|
|
["Rechtsform", d.get("legal_form", "-")],
|
|
["Mitarbeiter", str(d.get("employee_count", "-"))],
|
|
]
|
|
t = Table(data, colWidths=[60 * mm, 100 * mm])
|
|
t.setStyle(TableStyle([
|
|
("BACKGROUND", (0, 0), (-1, 0), LIGHT_PURPLE),
|
|
("TEXTCOLOR", (0, 0), (-1, 0), PURPLE),
|
|
("FONTSIZE", (0, 0), (-1, -1), 9),
|
|
("GRID", (0, 0), (-1, -1), 0.5, colors.lightgrey),
|
|
("VALIGN", (0, 0), (-1, -1), "TOP"),
|
|
]))
|
|
story.append(t)
|
|
else:
|
|
story.append(Paragraph("Kein Unternehmensprofil hinterlegt.", ss["Body2"]))
|
|
except Exception as e:
|
|
story.append(Paragraph(f"Fehler beim Laden: {e}", ss["Small"]))
|
|
story.append(Spacer(1, 5 * mm))
|
|
|
|
def _add_count_section(self, story, ss, title, table_name, tid):
|
|
story.append(Paragraph(title, ss["Section"]))
|
|
try:
|
|
count = self.db.execute(text(f"SELECT COUNT(*) FROM {table_name} WHERE tenant_id = :tid"), {"tid": tid}).scalar()
|
|
story.append(Paragraph(f"Eintraege: <b>{count or 0}</b>", ss["Body2"]))
|
|
except Exception:
|
|
story.append(Paragraph("Tabelle nicht vorhanden oder leer.", ss["Small"]))
|
|
story.append(Spacer(1, 3 * mm))
|
|
|
|
def _add_risk_section(self, story, ss, tid):
|
|
story.append(Paragraph("Risikobewertung", ss["Section"]))
|
|
try:
|
|
q = text("""
|
|
SELECT severity, COUNT(*) as cnt FROM compliance_risks
|
|
WHERE tenant_id = :tid GROUP BY severity ORDER BY severity
|
|
""")
|
|
rows = self.db.execute(q, {"tid": tid}).fetchall()
|
|
if rows:
|
|
data = [["Schweregrad", "Anzahl"]]
|
|
for r in rows:
|
|
data.append([r.severity or "UNKNOWN", str(r.cnt)])
|
|
t = Table(data, colWidths=[80 * mm, 40 * mm])
|
|
t.setStyle(TableStyle([
|
|
("BACKGROUND", (0, 0), (-1, 0), LIGHT_PURPLE),
|
|
("TEXTCOLOR", (0, 0), (-1, 0), PURPLE),
|
|
("FONTSIZE", (0, 0), (-1, -1), 9),
|
|
("GRID", (0, 0), (-1, -1), 0.5, colors.lightgrey),
|
|
]))
|
|
story.append(t)
|
|
else:
|
|
story.append(Paragraph("Keine Risiken erfasst.", ss["Body2"]))
|
|
except Exception:
|
|
story.append(Paragraph("Risiko-Tabelle nicht vorhanden.", ss["Small"]))
|
|
story.append(Spacer(1, 3 * mm))
|
|
|
|
def _add_review_section(self, story, ss, tid):
|
|
story.append(Paragraph("Dokumenten-Reviews", ss["Section"]))
|
|
try:
|
|
q = text("SELECT status, COUNT(*) as cnt FROM compliance_document_reviews WHERE tenant_id = :tid GROUP BY status")
|
|
rows = self.db.execute(q, {"tid": tid}).fetchall()
|
|
if rows:
|
|
data = [["Status", "Anzahl"]]
|
|
for r in rows:
|
|
data.append([r.status, str(r.cnt)])
|
|
t = Table(data, colWidths=[80 * mm, 40 * mm])
|
|
t.setStyle(TableStyle([
|
|
("BACKGROUND", (0, 0), (-1, 0), LIGHT_PURPLE),
|
|
("FONTSIZE", (0, 0), (-1, -1), 9),
|
|
("GRID", (0, 0), (-1, -1), 0.5, colors.lightgrey),
|
|
]))
|
|
story.append(t)
|
|
else:
|
|
story.append(Paragraph("Keine Reviews vorhanden.", ss["Body2"]))
|
|
except Exception:
|
|
story.append(Paragraph("Review-Tabelle nicht vorhanden.", ss["Small"]))
|
|
story.append(Spacer(1, 3 * mm))
|
|
|
|
def _add_consent_section(self, story, ss, tid):
|
|
story.append(Paragraph("Banner-Consents", ss["Section"]))
|
|
try:
|
|
count = self.db.execute(text("SELECT COUNT(*) FROM compliance_banner_consents WHERE tenant_id = :tid"), {"tid": tid}).scalar()
|
|
story.append(Paragraph(f"Gesamte Consents: <b>{count or 0}</b>", ss["Body2"]))
|
|
except Exception:
|
|
story.append(Paragraph("Banner-Tabelle nicht vorhanden.", ss["Small"]))
|
|
story.append(Spacer(1, 3 * mm))
|
|
|
|
def _add_role_section(self, story, ss, tid, pid):
|
|
story.append(Paragraph("Rollenkonzept", ss["Section"]))
|
|
try:
|
|
where = "tenant_id = :tid"
|
|
params: dict[str, Any] = {"tid": tid}
|
|
if pid:
|
|
where += " AND (project_id = :pid OR project_id IS NULL)"
|
|
params["pid"] = pid
|
|
rows = self.db.execute(text(f"SELECT role_key, role_label, person_name, person_email FROM compliance_org_roles WHERE {where} ORDER BY role_key"), params).fetchall()
|
|
if rows:
|
|
data = [["Rolle", "Name", "E-Mail"]]
|
|
for r in rows:
|
|
data.append([r.role_label or r.role_key, r.person_name or "-", r.person_email or "-"])
|
|
t = Table(data, colWidths=[60 * mm, 50 * mm, 50 * mm])
|
|
t.setStyle(TableStyle([
|
|
("BACKGROUND", (0, 0), (-1, 0), LIGHT_PURPLE),
|
|
("TEXTCOLOR", (0, 0), (-1, 0), PURPLE),
|
|
("FONTSIZE", (0, 0), (-1, -1), 9),
|
|
("GRID", (0, 0), (-1, -1), 0.5, colors.lightgrey),
|
|
]))
|
|
story.append(t)
|
|
else:
|
|
story.append(Paragraph("Keine Rollen zugewiesen.", ss["Body2"]))
|
|
except Exception:
|
|
story.append(Paragraph("Rollen-Tabelle nicht vorhanden.", ss["Small"]))
|
|
|
|
def _add_attribution_footer(self, story, ss) -> None:
|
|
"""Stufe 2 of the attribution renderer (Task #23).
|
|
|
|
Adds a "Quellen und Lizenzen" section listing the platform's
|
|
license-rule distribution and, crucially, the mandatory
|
|
attribution lines for Rule-2 sources (CC-BY-SA, OECD, Apache).
|
|
For Rule 1 sources the attribution is optional but rendered as
|
|
a brief reference list for auditability.
|
|
|
|
The section is added to every generated compliance PDF so each
|
|
export carries its own provenance footer — pauschale Hinweise
|
|
in AGB/Impressum reichen rechtlich nicht (siehe
|
|
project_attribution_strategy.md).
|
|
"""
|
|
try:
|
|
rows = self.db.execute(text("""
|
|
SELECT cc.license_rule, COUNT(*) AS n,
|
|
array_agg(DISTINCT cpl.source_regulation ORDER BY cpl.source_regulation)
|
|
FILTER (WHERE cpl.source_regulation IS NOT NULL) AS sources
|
|
FROM compliance.canonical_controls cc
|
|
LEFT JOIN compliance.control_parent_links cpl ON cpl.control_uuid = cc.id
|
|
WHERE cc.license_rule IS NOT NULL
|
|
GROUP BY cc.license_rule
|
|
ORDER BY cc.license_rule
|
|
""")).fetchall()
|
|
except Exception as e:
|
|
logger.warning("attribution footer skipped: %s", e)
|
|
return
|
|
if not rows:
|
|
return
|
|
|
|
rule_labels = {1: "Hoheitsrecht/Public Domain (woertlich)",
|
|
2: "Mit Attribution (CC-BY u.ae.)",
|
|
3: "Nur Identifier-Verweis"}
|
|
|
|
story.append(Spacer(1, 8 * mm))
|
|
story.append(Paragraph("Quellen & Lizenzen", ss["Section"]))
|
|
story.append(Paragraph(
|
|
"Dieser Bericht stuetzt sich auf klassifizierte Compliance-Controls "
|
|
"aus den folgenden Quellen. Jede Quelle ist deterministisch in eine "
|
|
"der drei Lizenzregeln (R1-R3) eingeordnet.", ss["Body2"]))
|
|
|
|
for r in rows:
|
|
rule = int(r.license_rule)
|
|
sources = (r.sources or [])[:8]
|
|
label = rule_labels.get(rule, f"Regel {rule}")
|
|
head = f"<b>R{rule} — {label}</b> ({r.n} Controls)"
|
|
story.append(Paragraph(head, ss["Body2"]))
|
|
if sources:
|
|
src_text = "; ".join(sources)
|
|
if len(r.sources or []) > 8:
|
|
src_text += f" und {len(r.sources) - 8} weitere"
|
|
story.append(Paragraph(src_text, ss["Small"]))
|
|
if rule == 2:
|
|
story.append(Paragraph(
|
|
"Pflicht-Attribution: Inhalte aus den oben genannten Quellen sind "
|
|
"unter den jeweiligen freien Lizenzen (z.B. CC-BY-SA, OECD-Public, "
|
|
"Apache-2.0) wiedergegeben. Original-Urheber bleibt in jeder "
|
|
"Weiterverwendung zu nennen.", ss["Small"]))
|
|
story.append(Spacer(1, 2 * mm))
|