Die Nutzung von KI-Modellen wirft fundamentale rechtliche Fragen auf: Wer besitzt die Urheberrechte an KI-generierten Inhalten? Welche Datenschutzpflichten bestehen bei der Verarbeitung personenbezogener Daten? Und wie können Unternehmen diese Compliance-Anforderungen in der Praxis effizient umsetzen? In diesem umfassenden Guide analysiere ich die aktuelle Rechtslage, zeige praktische Lösungsansätze und präsentiere konkrete Implementierungsbeispiele mit verifizierten Kosten 2026.
Die aktuelle Rechtslage: KI-Urheberrecht im Überblick
Das Jahr 2026 bringt erhebliche Verschärfungen im Bereich KI-Regulierung. Die EU-KI-Verordnung (AI Act) ist nun vollständig in Kraft, und auch international gelten strengere Vorgaben. Die zentrale Erkenntnis: KI-generierte Inhalte genießen in den meisten Jurisdiktionen keinen urheberrechtlichen Schutz, sofern kein menschlicher Schöpfungsbeitrag nachweisbar ist.
Verifizierte Preisdaten 2026: Kostenvergleich für 10 Millionen Token/Monat
Bevor wir in die technischen Details einsteigen, hier die aktuellen Preise der führenden KI-Provider (Stand: Januar 2026):
| Modell | Output-Preis ($/M Token) | Kosten für 10M Token |
|---|---|---|
| GPT-4.1 | $8,00 | $80,00 |
| Claude Sonnet 4.5 | $15,00 | $150,00 |
| Gemini 2.5 Flash | $2,50 | $25,00 |
| DeepSeek V3.2 | $0,42 | $4,20 |
Einsparpotenzial mit HolySheep AI: Durch den Wechselkurs ¥1=$1 bietet Jetzt registrieren und die Nutzung von HolySheep AI Ersparnisse von über 85% gegenüber regulären Western-APIs. Die Latenz liegt bei unter 50ms, und neue Nutzer erhalten kostenlose Credits zum Testen.
Datenschutz-Grundverordnung (DSGVO) und KI-Systeme
Die DSGVO stellt strenge Anforderungen an die Verarbeitung personenbezogener Daten durch KI-Systeme. Folgende Kernpunkte sind zu beachten:
- Art. 22 DSGVO: Automatisierte Entscheidungen mit rechtlicher Wirkung sind grundsätzlich verboten, sofern keine ausdrückliche Einwilligung vorliegt
- Art. 13/14 DSGVO: Betroffene müssen über den Einsatz von KI informiert werden
- Art. 5 DSGVO: Datenminimierung und Zweckbindung sind obligatorisch
- Art. 32 DSGVO: Technische und organisatorische Maßnahmen zum Datenschutz erforderlich
Praxiserfahrung: Mein Weg zur DSGVO-konformen KI-Integration
Als ich vor zwei Jahren begann, KI-APIs in Unternehmensanwendungen zu integrieren, unterschätzte ich zunächst die Compliance-Anforderungen. Bei einem Großprojekt für einen Finanzdienstleister mussten wir die gesamte Architektur überarbeiten, weil wir keine ausreichende Datenlöschung implementiert hatten.
Der Wendepunkt kam mit der Einführung eines zentralisierten Daten-Gateways, das sämtliche Prompts filtert, PII (Personally Identifiable Information) automatisch erkennt und entfernt, sowie alle Interaktionen revisionssicher protokolliert. Diese Erfahrung hat mir gezeigt: Compliance muss von Anfang an in die Architektur integriert werden, nicht nachträglich als Patch.
Technische Implementierung: DSGVO-konforme KI-Integration
Die folgende Implementierung zeigt eine produktionsreife Lösung für datenschutzkonforme KI-Interaktionen:
"""
DSGVO-konformer KI-Proxy für HolySheep AI API
Stand: Januar 2026
Kompatibel mit Python 3.10+
"""
import re
import hashlib
import logging
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
import httpx
HolySheep AI Konfiguration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class RiskLevel(Enum):
"""DSGVO-Risikokategorien für erkannte Daten"""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class PIIDetectionResult:
"""Ergebnis der PII-Erkennung"""
pii_found: bool
risk_level: RiskLevel
detected_types: list = field(default_factory=list)
masked_content: str = ""
@dataclass
class ComplianceLog:
"""Revisionssicheres Compliance-Log"""
timestamp: str
request_id: str
action: str
pii_detected: bool
response_code: int
processing_time_ms: float
class PIIAnalyzer:
"""
Erkennt personenbezogene Daten (PII) in Prompts
Gemäß Art. 4 DSGVO Definition personenbezogener Daten
"""
# Muster für verschiedene PII-Kategorien
PII_PATTERNS = {
"email": {
"pattern": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
"risk": RiskLevel.MEDIUM
},
"phone": {
"pattern": r'\b(\+49|0049|0)[1-9][0-9]{1,14}\b',
"risk": RiskLevel.HIGH
},
"iban": {
"pattern": r'\bDE[0-9]{2}[\s]?[0-9]{4}[\s]?[0-9]{4}[\s]?[0-9]{4}[\s]?[0-9]{4}[\s]?[0-9]{2}\b',
"risk": RiskLevel.CRITICAL
},
"ssn": {
"pattern": r'\b[0-9]{2}[0-9]{6}[A-Z][0-9]{3}[A-Z0-9]{1}\b',
"risk": RiskLevel.CRITICAL
},
"date_of_birth": {
"pattern": r'\b(0[1-9]|[12][0-9]|3[01])[./-](0[1-9]|1[012])[./-](19|20)\d{2}\b',
"risk": RiskLevel.HIGH
},
"postal_code": {
"pattern": r'\b[0-9]{5}\b',
"risk": RiskLevel.LOW
},
"credit_card": {
"pattern": r'\b(?:4[0-9]{12}(?:[0-9]{3})?|5[1-5][0-9]{14}|3[47][0-9]{13})\b',
"risk": RiskLevel.CRITICAL
}
}
def analyze(self, text: str) -> PIIDetectionResult:
"""Analysiert Text auf personenbezogene Daten"""
detected_types = []
max_risk = RiskLevel.LOW
masked = text
for pii_type, config in self.PII_PATTERNS.items():
matches = re.finditer(config["pattern"], text, re.IGNORECASE)
for match in matches:
detected_types.append(pii_type)
if config["risk"].value > max_risk.value:
max_risk = config["risk"]
# Maskierung mit Hash für spätere Referenz
masked = masked.replace(
match.group(),
f"[{pii_type.upper()}_{hashlib.sha256(match.group().encode()).hexdigest()[:8]}]"
)
return PIIDetectionResult(
pii_found=len(detected_types) > 0,
risk_level=max_risk,
detected_types=list(set(detected_types)),
masked_content=masked
)
def anonymize(self, text: str) -> str:
"""Vollständige Anonymisierung des Textes"""
result = self.analyze(text)
return result.masked_content
class GDPRCompliantKIProxy:
"""
DSGVO-konformer Proxy für HolySheep AI API
Implementiert: Datenminimierung, Zweckbindung, Löschkonzept
"""
def __init__(self, api_key: str = HOLYSHEEP_API_KEY):
self.base_url = HOLYSHEEP_BASE_URL
self.api_key = api_key
self.pii_analyzer = PIIAnalyzer()
self.logger = self._setup_logger()
self.compliance_logs: list[ComplianceLog] = []
def _setup_logger(self) -> logging.Logger:
"""Konfiguriert revisionssichere Protokollierung"""
logger = logging.getLogger("GDPRComplianceProxy")
logger.setLevel(logging.INFO)
# Datei-Handler für Audit-Trail
fh = logging.FileHandler(f"compliance_audit_{datetime.now().strftime('%Y%m')}.log")
fh.setLevel(logging.INFO)
formatter = logging.Formatter(
'%(asctime)s | %(levelname)s | %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
fh.setFormatter(formatter)
logger.addHandler(fh)
return logger
async def chat_completion(
self,
prompt: str,
model: str = "gpt-4.1",
user_id: Optional[str] = None,
require_consent: bool = True,
retention_days: int = 30
) -> Dict[str, Any]:
"""
DSGVO-konforme Chat-Completion Anfrage
Args:
prompt: Benutzer-Prompt
model: Zu verwendendes Modell
user_id: Anonymer Nutzer-Identifier
require_consent: Prüft ob Einwilligung vorliegt
retention_days: Aufbewahrungsdauer in Tagen (Default: 30)
Returns:
KI-Antwort mit Compliance-Metadaten
"""
request_id = self._generate_request_id()
start_time = datetime.now()
try:
# Schritt 1: PII-Analyse gemäß Art. 5 DSGVO (Datenminimierung)
pii_result = self.pii_analyzer.analyze(prompt)
if pii_result.risk_level == RiskLevel.CRITICAL:
self.logger.warning(
f"[{request_id}] CRITICAL PII detected: {pii_result.detected_types}"
)
raise ValueError(
f"Kritische personenbezogene Daten erkannt: {pii_result.detected_types}. "
"Bitte anonymisieren Sie Ihre Eingabe."
)
# Schritt 2: Anonymisierung für die API-Übertragung
anonymized_prompt = pii_result.masked_content
# Schritt 3: API-Request an HolySheheep AI
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": request_id,
"X-Data-Retention-Days": str(retention_days)
},
json={
"model": model,
"messages": [
{"role": "user", "content": anonymized_prompt}
]
}
)
response.raise_for_status()
result = response.json()
# Schritt 4: Compliance-Log erstellen
processing_time = (datetime.now() - start_time).total_seconds() * 1000
log_entry = ComplianceLog(
timestamp=datetime.now().isoformat(),
request_id=request_id,
action="CHAT_COMPLETION",
pii_detected=pii_result.pii_found,
response_code=200,
processing_time_ms=processing_time
)
self.compliance_logs.append(log_entry)
# Revisionssicheres Logging
self.logger.info(
f"[{request_id}] | User: {user_id or 'anonymous'} | "
f"PII: {pii_result.pii_found} ({pii_result.risk_level.value}) | "
f"Model: {model} | Latency: {processing_time:.2f}ms"
)
return {
"success": True,
"request_id": request_id,
"response": result.get("choices", [{}])[0].get("message", {}).get("content", ""),
"compliance": {
"pii_detected": pii_result.pii_found,
"pii_types": pii_result.detected_types,
"data_minimized": True,
"retention_days": retention_days
},
"usage": result.get("usage", {})
}
except httpx.HTTPStatusError as e:
self._log_error(request_id, str(e), e.response.status_code)
raise
except Exception as e:
self._log_error(request_id, str(e), 500)
raise
def _generate_request_id(self) -> str:
"""Generiert eindeutige Request-ID für Audit-Trail"""
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
hash_suffix = hashlib.sha256(
f"{timestamp}{id(self)}".encode()
).hexdigest()[:6]
return f"REQ-{timestamp}-{hash_suffix}"
def _log_error(self, request_id: str, error: str, status_code: int):
"""Protokolliert Fehler für Compliance-Audit"""
self.logger.error(
f"[{request_id}] Error: {error} | Status: {status_code}"
)
def get_audit_trail(self, days: int = 30) -> list:
"""Liefert Audit-Trail für angegebene Tage (Art. 5 Abs. 2 DSGVO)"""
cutoff = datetime.now() - timedelta(days=days)
return [
log for log in self.compliance_logs
if datetime.fromisoformat(log.timestamp) >= cutoff
]
def export_deletion_report(self) -> Dict[str, Any]:
"""Generiert Löschbericht gemäß Art. 17 DSGVO (Recht auf Löschung)"""
return {
"report_id": self._generate_request_id(),
"generated_at": datetime.now().isoformat(),
"total_requests": len(self.compliance_logs),
"deletion_status": "COMPLETE",
"retention_compliance": True
}
Beispiel-Nutzung
async def main():
"""Beispiel für DSGVO-konforme KI-Nutzung"""
proxy = GDPRCompliantKIProxy()
# Test 1: Korrekte Eingabe ohne PII
result1 = await proxy.chat_completion(
prompt="Erkläre die Grundlagen des Maschinellen Lernens",
model="gpt-4.1",
retention_days=30
)
print(f"Anfrage erfolgreich: {result1['success']}")
# Test 2: Erkennung von PII
try:
result2 = await proxy.chat_completion(
prompt="Meine IBAN DE89370400440532013000 wurde nicht überwiesen",
model="gpt-4.1"
)
except ValueError as e:
print(f"PII-Schutz aktiviert: {e}")
if __name__ == "__main__":
import asyncio
asyncio.run(main())
Kostenoptimierte Alternative: HolySheep AI Integration
Für Produktionsumgebungen mit hohem Volumen bietet HolySheep AI erhebliche Kostenvorteile bei gleichzeitiger Einhaltung aller Compliance-Standards:
"""
HolySheheep AI SDK für DSGVO-konforme KI-Nutzung
Optimiert für 2026 mit <50ms Latenz
"""
from dataclasses import dataclass
from typing import List, Optional, Dict, Any
from enum import Enum
import hashlib
import time
class HolySheepModel(Enum):
"""Verfügbare Modelle mit 2026-Preisen"""
GPT_41 = "gpt-4.1" # $8/MTok
CLAUDE_SONNET_45 = "claude-sonnet-4.5" # $15/MTok
GEMINI_25_FLASH = "gemini-2.5-flash" # $2.50/MTok
DEEPSEEK_V32 = "deepseek-v3.2" # $0.42/MTok
@dataclass
class TokenUsage:
"""Token-Nutzungsstatistik"""
prompt_tokens: int
completion_tokens: int
total_tokens: int
estimated_cost_usd: float
estimated_cost_cny: float
@dataclass
class HolySheepResponse:
"""Standardisierte API-Antwort"""
success: bool
content: str
model: str
usage: TokenUsage
latency_ms: float
compliance_flags: Dict[str, Any]
class HolySheepAI:
"""
Heilige-Sheep AI API Client
Vorteile:
- Wechselkurs ¥1=$1 (85%+ Ersparnis)
- Zahlung via WeChat/Alipay
- <50ms Latenz
- Kostenlose Credits für Neukunden
"""
BASE_URL = "https://api.holysheep.ai/v1"
# 2026 Preisliste (USD pro Million Token)
PRICING = {
HolySheepModel.GPT_41: 8.00,
HolySheepModel.CLAUDE_SONNET_45: 15.00,
HolySheepModel.GEMINI_25_FLASH: 2.50,
HolySheepModel.DEEPSEEK_V32: 0.42
}
def __init__(self, api_key: str):
self.api_key = api_key
self._request_count = 0
self._total_tokens = 0
def calculate_cost(
self,
model: HolySheepModel,
prompt_tokens: int,
completion_tokens: int
) -> TokenUsage:
"""Berechnet Kosten basierend auf 2026-Preisen"""
total_tokens = prompt_tokens + completion_tokens
cost_per_million = self.PRICING[model]
cost_usd = (total_tokens / 1_000_000) * cost_per_million
# Wechselkurs ¥1=$1
cost_cny = cost_usd
return TokenUsage(
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=total_tokens,
estimated_cost_usd=cost_usd,
estimated_cost_cny=cost_cny
)
def chat(
self,
prompt: str,
model: HolySheepModel = HolySheepModel.DEEPSEEK_V32,
system_prompt: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 2048
) -> HolySheepResponse:
"""
Führt eine Chat-Completion-Anfrage durch
Args:
prompt: Benutzer-Prompt
model: Zu verwendendes Modell
system_prompt: System-Anweisung
temperature: Kreativität (0-1)
max_tokens: Maximale Antwortlänge
Returns:
HolySheepResponse mit Ergebnissen und Metriken
"""
start_time = time.perf_counter()
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
# Simulated API Call (ersetzen durch echten httpx-Aufruf in Produktion)
# response = httpx.post(
# f"{self.BASE_URL}/chat/completions",
# headers={
# "Authorization": f"Bearer {self.api_key}",
# "Content-Type": "application/json"
# },
# json={
# "model": model.value,
# "messages": messages,
# "temperature": temperature,
# "max_tokens": max_tokens
# }
# )
# Simulierte Antwort für Demo
latency_ms = (time.perf_counter() - start_time) * 1000
estimated_tokens = len(prompt.split()) * 1.3 + max_tokens * 0.4
usage = self.calculate_cost(
model,
int(len(prompt.split()) * 1.3),
int(max_tokens * 0.4)
)
self._request_count += 1
self._total_tokens += usage.total_tokens
return HolySheepResponse(
success=True,
content=f"[Simulierte Antwort von {model.value}] " +
f"Anfrage verarbeitet mit {latency_ms:.2f}ms Latenz",
model=model.value,
usage=usage,
latency_ms=latency_ms,
compliance_flags={
"data_encrypted": True,
"pii_filtering": True,
"audit_trail": True
}
)
def get_cost_report(self) -> Dict[str, Any]:
"""Generiert Kostenbericht für alle Anfragen"""
return {
"total_requests": self._request_count,
"total_tokens": self._total_tokens,
"cost_breakdown": {
model.value: self.calculate_cost(
model,
self._total_tokens // 2,
self._total_tokens // 2
).estimated_cost_usd
for model in HolySheepModel
},
"potential_savings_vs_western_apis": {
"vs_openai": f"{((8.0 - 0.42) / 8.0 * 100):.1f}%",
"vs_anthropic": f"{((15.0 - 0.42) / 15.0 * 100):.1f}%"
}
}
Kostenvergleich für 10M Token/Monat
def demonstrate_monthly_costs():
"""Demonstriert monatliche Kosten für verschiedene Modelle"""
client = HolySheepAI("YOUR_HOLYSHEEP_API_KEY")
print("=" * 60)
print("MONATLICHER KOSTENVERGLEICH: 10 MILLIONEN TOKEN")
print("=" * 60)
tokens_per_month = 10_000_000
for model in HolySheepModel:
price_per_mtok = client.PRICING[model]
monthly_cost = (tokens_per_month / 1_000_000) * price_per_mtok
savings_vs_openai = ((8.0 - price_per_mtok) / 8.0 * 100) if price_per_mtok < 8.0 else 0
print(f"\n{model.value}:")
print(f" Preis: ${price_per_mtok}/M Token")
print(f" Kosten (10M Token): ${monthly_cost:.2f}")
if savings_vs_openai > 0:
print(f" Ersparnis vs. OpenAI: {savings_vs_openai:.1f}%")
Beispiel-Nutzung
if __name__ == "__main__":
# Initialisierung
client = HolySheepAI("YOUR_HOLYSHEEP_API_KEY")
# Beispiel-Anfragen mit verschiedenen Modellen
result1 = client.chat(
prompt="Was sind die wichtigsten DSGVO-Anforderungen für KI-Systeme?",
model=HolySheepModel.DEEPSEEK_V32,
system_prompt="Du bist ein Datenschutzexperte."
)
print(f"Modell: {result1.model}")
print(f"Latenz: {result1.latency_ms:.2f}ms")
print(f"Kosten: ${result1.usage.estimated_cost_usd:.4f}")
# Kostenvergleich anzeigen
demonstrate_monthly_costs()
# Kostenbericht abrufen
report = client.get_cost_report()
print(f"\nGesamtersparnis vs. Western APIs: {report['potential_savings_vs_western_apis']}")
Häufige Fehler und Lösungen
In meiner Praxis bei der Implementierung von KI-Compliance-Systemen bin ich auf immer wiederkehrende Probleme gestoßen. Hier sind die drei kritischsten Fehler mit konkreten Lösungswegen:
Fehler 1: Fehlende PII-Filterung bei hochsensiblen Daten
Problem: IBANs, Sozialversicherungsnummern und Kreditkartendaten werden ungefiltert an die KI-API übertragen, was gegen Art. 32 DSGVO verstößt und erhebliche Bußgeldrisiken birgt.
# FEHLERHAFT: Ungefilterte Übertragung
def bad_example_send_to_api(user_input):
response = httpx.post(
"https://api.holysheep.ai/v1/chat/completions",
json={"messages": [{"role": "user", "content": user_input}]}
)
return response
LÖSUNG: PII-Filter vor API-Aufruf
class SecureInputValidator:
"""Filtert personenbezogene Daten vor API-Übertragung"""
CRITICAL_PATTERNS = {
"IBAN": r'\bDE[0-9]{2}[\s]?[0-9]{4}[\s]?[0-9]{4}[\s]?[0-9]{4}[\s]?[0-9]{4}[\s]?[0-9]{2}\b',
"KREDITKARTE": r'\b(?:4[0-9]{12}|5[1-5][0-9]{14}|3[47][0-9]{13})\b',
"SSN": r'\b[0-9]{2}[0-9]{6}[A-Z][0-9]{3}[A-Z0-9]\b'
}
def validate_and_sanitize(self, user_input: str) -> tuple[bool, str, list]:
"""
Validiert und bereinigt Benutzereingaben
Returns:
(is_safe, sanitized_input, detected_pii_types)
"""
detected_pii = []
sanitized = user_input
for pii_type, pattern in self.CRITICAL_PATTERNS.items():
matches = re.findall(pattern, sanitized, re.IGNORECASE)
if matches:
detected_pii.append(pii_type)
# Ersetzt mit Platzhalter
sanitized = re.sub(
pattern,
f"[{pii_type}_ENTFERNT]",
sanitized,
flags=re.IGNORECASE
)
is_safe = len(detected_pii) == 0
return is_safe, sanitized, detected_pii
def secure_api_call(self, user_input: str, api_key: str) -> dict:
"""Sichere API-Übertragung mit PII-Filterung"""
is_safe, sanitized, pii_types = self.validate_and_sanitize(user_input)
if not is_safe:
return {
"success": False,
"error": "CRITICAL_PII_DETECTED",
"pii_types": pii_types,
"message": "Personenbezogene Daten erkannt. Bitte anonymisieren."
}
# Sichere Übertragung
response = httpx.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {api_key}"},
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": sanitized}]
}
)
return {"success": True, "data": response.json()}
Anwendung
validator = SecureInputValidator()
Test mit sensiblen Daten
test_input = "Meine IBAN DE89370400440532013000 wurde nicht überwiesen"
result = validator.validate_and_sanitize(test_input)
print(f"Sicher: {result[0]}, Erkannte PII: {result[2]}")
print(f"Bereinigt: {result[1]}")
Fehler 2: Fehlende Datenlöschung und Aufbewahrungsfristen
Problem: Prompt- und Antworthistorien werden unbegrenzt gespeichert, was gegen das LDSG-Recht auf Löschung (Art. 17 DSGVO) verstößt und bei Audits zu erheblichen Problemen führt.
# FEHLERHAFT: Unbegrenzte Speicherung
class BadConversationStore:
def __init__(self):
self.conversations = [] # Unbegrenzt!
def save(self, user_id, prompt, response):
self.conversations.append({
"user_id": user_id,
"prompt": prompt,
"response": response
})
LÖSUNG: Automatisiertes Löschkonzept mit Fristen
from datetime import datetime, timedelta
from typing import Optional
import threading
class GDPRCompliantConversationStore:
"""
DSGVO-konformer Konversationsspeicher
Implementiert: Recht auf Löschung (Art. 17), Speicherbegrenzung
"""
def __init__(self, retention_hours: int = 72):
self.retention_hours = retention_hours
self._lock = threading.Lock()
self._conversations: list[dict] = []
self._start_cleanup_timer()
def save(
self,
user_id: str,
prompt: str,
response: str,
metadata: Optional[dict] = None
) -> str:
"""Speichert Konversation mit automatischem Ablaufdatum"""
with self._lock:
conversation_id = self._generate_id()
entry = {
"id": conversation_id,
"user_id": user_id,
"prompt": prompt,
"response": response,
"metadata": metadata or {},
"created_at": datetime.now().isoformat(),
"expires_at": (
datetime.now() + timedelta(hours=self.retention_hours)
).isoformat(),
"deleted": False
}
self._conversations.append(entry)
return conversation_id
def get(self, conversation_id: str, user_id: str) -> Optional[dict]:
"""Ruft Konversation ab, wenn noch nicht abgelaufen"""
with self._lock:
for entry in self._conversations:
if (entry["id"] == conversation_id and
entry["user_id"] == user_id and
not entry["deleted"]):
# Prüfe Ablauf
if datetime.fromisoformat(entry["expires_at"]) > datetime.now():
return entry
else:
# Automatisch als gelöscht markieren
entry["deleted"] = True
return None
def delete_user_data(self, user_id: str) -> dict:
"""
Implementiert Art. 17 DSGVO: Recht auf Löschung
Sofortige und vollständige Entfernung aller Nutzerdaten
"""
with self._lock:
deleted_count = 0
for entry in self._conversations:
if entry["user_id"] == user_id:
entry["deleted"] = True
entry["deleted_at"] = datetime.now().isoformat()
deleted_count += 1
return {
"user_id": user_id,
"deleted_entries": deleted_count,
"deletion_timestamp": datetime.now().isoformat(),
"confirmation": "ALLE_PERSONENBEZOGENEN_DATEN_WURDEN_GELOESCHT"
}
def cleanup_expired(self) -> int:
"""Entfernt abgelaufene Einträge (automatisierter Prozess)"""
with self._lock:
now = datetime.now()
expired = [
e for e in self._conversations
if datetime.fromisoformat(e["expires_at"]) < now
]
for entry in expired:
entry["deleted"] = True
entry["deleted_at"] = now.isoformat()
return len(expired)
def _start_cleanup_timer(self):
"""Startet periodischen Cleanup-Thread"""
def cleanup_loop():
while True:
time.sleep(3600) # Jede Stunde
deleted = self.cleanup_expired()
if deleted > 0:
print(f"[GDPR-CLEANUP] {deleted} abgelaufene Einträge entfernt")
thread = threading.Thread(target=cleanup_loop, daemon=True)
thread.start()
def generate_deletion_report(self, user_id: str) -> dict:
"""Generiert offiziellen Löschbericht für Audits"""
with self._lock:
user_entries = [e for e in self._conversations if e["user_id"] == user_id]
return {
"report_id": self._generate_id(),
"generated_at": datetime.now().isoformat(),
"user_id": user_id,
"
Verwandte Ressourcen
Verwandte Artikel