"""Betriebsvereinbarung template generator — creates BV draft from UCCA assessment. Generates a modular works council agreement (Betriebsvereinbarung) based on: - UCCA Assessment result (triggered rules, risk score, obligations) - Company profile (name, location, works council) - System details (name, type, modules) Sections A-M follow the template in migration 006. """ from typing import Optional # -- Default verbotene Nutzungen nach BAG-Rechtsprechung -------------------- DEFAULT_VERBOTENE_NUTZUNGEN = [ "Verdeckte Leistungs- oder Verhaltenskontrolle einzelner Beschaeftigter", "Erstellung individueller Persoenlichkeitsprofile oder Verhaltensanalysen", "Nutzung von Nutzungshistorien zu disziplinarischen Zwecken", "Automatisierte Personalentscheidungen ohne menschliche Ueberpruefung (Art. 22 DSGVO)", "Personenbezogene Rankings oder Leistungsvergleiche ohne gesonderte Mitbestimmung", "Korrelation von Systemnutzungsdaten mit Leistungsbeurteilungen", ] AI_VERBOTENE_NUTZUNGEN = [ "Einsatz von KI-Funktionen zur biometrischen Echtzeit-Identifizierung am Arbeitsplatz", "KI-gestuetztes Social Scoring von Beschaeftigten", "Nutzung von KI-generierten Bewertungen als alleinige Grundlage fuer Personalentscheidungen", ] # -- Standard-TOM Massnahmen ------------------------------------------------ DEFAULT_TOM = [ "Rollen- und Rechtekonzept mit Least-Privilege-Prinzip", "Verschluesselung der Daten bei Uebertragung (TLS 1.2+) und Speicherung (AES-256)", "Protokollierung aller administrativen Zugriffe", "Pseudonymisierung personenbezogener Daten, wo technisch moeglich", "Deaktivierung nicht benoetigter Telemetrie- und Diagnosefunktionen", "Getrennte Umgebungen fuer Test und Produktion", "Regelmaessige Sicherheitsupdates und Patch-Management", "Zugangsschutz durch Multi-Faktor-Authentifizierung fuer Administratoren", ] # -- Standard erlaubte Reports ---------------------------------------------- DEFAULT_ERLAUBTE_REPORTS = [ "Systemgesundheit und Verfuegbarkeit (ohne Personenbezug)", "Lizenznutzung auf aggregierter Ebene (Abteilung/Standort, nicht Person)", "Sicherheitsereignisse und Anomalien", "Speicherplatznutzung (ohne Personenbezug)", "Fehlerstatistiken (technisch, nicht personenbezogen)", ] # -- Standard Datenarten bei IT/KI-Systemen --------------------------------- DATENARTEN_MAP = { "email": "E-Mail-Metadaten (Absender, Empfaenger, Zeitstempel — NICHT Inhalte)", "chat": "Chat-/Messaging-Metadaten (Teilnehmer, Zeitstempel)", "document": "Dokumentenmetadaten (Ersteller, Aenderungsdatum, Dateiname)", "login": "Anmeldedaten (Benutzername, Zeitstempel, IP-Adresse)", "usage": "Nutzungsdaten (aufgerufene Funktionen, Nutzungsdauer — aggregiert)", "prompt": "KI-Eingaben und -Ausgaben (Prompts, Antworten)", "calendar": "Kalendereintraege (Betreff, Teilnehmer, Zeiten)", "hr": "Personalstammdaten (Name, Abteilung, Position, Eintrittsdatum)", "performance": "Leistungsdaten (Kennzahlen, Bewertungen, Zielvereinbarungen)", "video": "Videoaufnahmen (Arbeitsplatz, Zugangsbereiche)", "location": "Standortdaten (GPS, WLAN-basierte Ortung, Gebaeudezutritt)", } def generate_betriebsvereinbarung_draft(ctx: dict) -> dict: """Generate a Betriebsvereinbarung draft from company + assessment context. Args: ctx: Dict with keys: Required: - company_name: str - system_name: str - system_description: str Optional: - company_address: str - employer_representative: str - works_council_chair: str - system_vendor: str - locations: list[str] - departments: list[str] - modules: list[str] - purposes: list[str] - data_types: list[str] — keys from DATENARTEN_MAP - is_ai_system: bool - has_employee_monitoring: bool - has_hr_features: bool - has_video: bool - dpo_name: str - dpo_contact: str - audit_interval: str — e.g. "12 Monate" - duration: str — e.g. "unbefristet" - notice_period: str — e.g. "3 Monate" - retention_audit_logs: str — e.g. "90 Tage" - retention_usage_data: str — e.g. "30 Tage" - retention_prompts: str — e.g. "deaktiviert" - additional_forbidden: list[str] - additional_tom: list[str] - additional_reports: list[str] - betrvg_conflict_score: int — 0-100 Returns: Dict with placeholder values ready for template substitution. """ result = {} # Basic info result["UNTERNEHMEN_NAME"] = ctx.get("company_name", "{{UNTERNEHMEN_NAME}}") result["UNTERNEHMEN_SITZ"] = ctx.get("company_address", "{{UNTERNEHMEN_SITZ}}") result["ARBEITGEBER_VERTRETER"] = ctx.get("employer_representative", "{{ARBEITGEBER_VERTRETER}}") result["BETRIEBSRAT_VORSITZ"] = ctx.get("works_council_chair", "{{BETRIEBSRAT_VORSITZ}}") result["SYSTEM_NAME"] = ctx.get("system_name", "{{SYSTEM_NAME}}") result["SYSTEM_BESCHREIBUNG"] = ctx.get("system_description", "{{SYSTEM_BESCHREIBUNG}}") result["SYSTEM_HERSTELLER"] = ctx.get("system_vendor", "") result["DSB_NAME"] = ctx.get("dpo_name", "{{DSB_NAME}}") result["DSB_KONTAKT"] = ctx.get("dpo_contact", "{{DSB_KONTAKT}}") # B. Geltungsbereich locations = ctx.get("locations", []) result["GELTUNGSBEREICH_STANDORTE"] = _bullet_list(locations) if locations else "Alle Standorte der {{UNTERNEHMEN_NAME}}" departments = ctx.get("departments", []) result["GELTUNGSBEREICH_BEREICHE"] = _bullet_list(departments) if departments else "Alle Beschaeftigten" modules = ctx.get("modules", []) result["GELTUNGSBEREICH_MODULE"] = _bullet_list(modules) if modules else "Alle Module und Dienste von {{SYSTEM_NAME}}" # C. Zweck purposes = ctx.get("purposes", []) result["ZWECK_BESCHREIBUNG"] = _bullet_list(purposes) if purposes else "{{ZWECK_BESCHREIBUNG}}" # C.2 Verbotene Nutzungen forbidden = list(DEFAULT_VERBOTENE_NUTZUNGEN) if ctx.get("is_ai_system"): forbidden.extend(AI_VERBOTENE_NUTZUNGEN) forbidden.extend(ctx.get("additional_forbidden", [])) result["VERBOTENE_NUTZUNGEN"] = _bullet_list(forbidden) # D. Datenarten data_type_keys = ctx.get("data_types", []) datenarten = [] for key in data_type_keys: if key in DATENARTEN_MAP: datenarten.append(DATENARTEN_MAP[key]) else: datenarten.append(key) result["DATENARTEN_LISTE"] = _bullet_list(datenarten) if datenarten else "{{DATENARTEN_LISTE}}" # E. Rollen result["ROLLEN_ADMIN"] = ctx.get("roles_admin", "IT-Administration: Systemkonfiguration, Benutzerverwaltung, Sicherheitsupdates") result["ROLLEN_FUEHRUNGSKRAFT"] = ctx.get("roles_manager", "Fuehrungskraefte: Nur aggregierte, nicht-personenbezogene Reports") result["ROLLEN_REPORTING"] = ctx.get("roles_reporting", "Controlling/Reporting: Nur freigegebene Standardreports (siehe Abschnitt G)") # F. Transparenz result["TRANSPARENZ_INFO"] = ctx.get("transparency_info", "Die Information erfolgt schriftlich und in einer Informationsveranstaltung vor Einfuehrung des Systems.") # G. Reports reports = list(DEFAULT_ERLAUBTE_REPORTS) reports.extend(ctx.get("additional_reports", [])) result["ERLAUBTE_REPORTS"] = _bullet_list(reports) # H. Speicherfristen result["SPEICHERFRIST_AUDIT_LOGS"] = ctx.get("retention_audit_logs", "90 Tage") result["SPEICHERFRIST_NUTZUNGSDATEN"] = ctx.get("retention_usage_data", "30 Tage") result["SPEICHERFRIST_CHAT_PROMPTS"] = ctx.get("retention_prompts", "deaktiviert") # I. TOM tom = list(DEFAULT_TOM) tom.extend(ctx.get("additional_tom", [])) # Intensivere Schutzmassnahmen bei hohem Konflikt-Score conflict_score = ctx.get("betrvg_conflict_score", 0) if conflict_score >= 50: tom.append("Automatische Anomalie-Erkennung bei ungewoehnlichen Admin-Zugriffen") tom.append("Quartalsweise Datenschutz-Audit durch externen Prueer") if conflict_score >= 75: tom.append("Betriebsrat erhaelt Leserechte auf Audit-Log-Dashboard") tom.append("Jede Sonderauswertung wird dem Betriebsrat innerhalb von 24h gemeldet") result["TOM_MASSNAHMEN"] = _bullet_list(tom) # J. Change-Management result["CHANGE_MANAGEMENT_PROZESS"] = ctx.get("change_process", "Die Arbeitgeberin informiert den Betriebsrat schriftlich ueber geplante Aenderungen " "mindestens 14 Kalendertage vor Umsetzung. Bei sicherheitskritischen Updates kann die " "Frist auf 3 Werktage verkuerzt werden.") # K. Audit result["AUDIT_INTERVALL"] = ctx.get("audit_interval", "12 Monate") # L. Beschwerde result["BESCHWERDE_ANSPRECHPARTNER"] = ctx.get("complaint_contacts", "- Direkter Vorgesetzter\n- Betriebsrat ({{BETRIEBSRAT_VORSITZ}})\n" "- Datenschutzbeauftragter ({{DSB_NAME}}, {{DSB_KONTAKT}})") # M. Schluss result["LAUFZEIT"] = ctx.get("duration", "unbefristet") result["KUENDIGUNGSFRIST"] = ctx.get("notice_period", "3 Monate") result["DATUM_UNTERZEICHNUNG"] = ctx.get("signing_date", "{{DATUM_UNTERZEICHNUNG}}") # Conditional flags result["AI_SYSTEM"] = ctx.get("is_ai_system", False) result["VIDEO_UEBERWACHUNG"] = ctx.get("has_video", False) result["HR_SYSTEM"] = ctx.get("has_hr_features", False) return result def _bullet_list(items: list) -> str: """Format a list as markdown bullet points.""" return "\n".join(f"- {item}" for item in items)