Von: HolySheep AI Tech-Blog | Stand: Mai 2026 | Lesezeit: 12 Minuten
Als Entwickler und Architekt habe ich in den letzten Jahren hunderte von API-Integrationen begleitet. Was ich dabei immer wieder beobachtet habe: Die größte Frustquelle für Endnutzer sind nicht die technischen Fehler selbst — es ist das fehlende Verständnis dafür, warum etwas schiefläuft und was dagegen unternommen wird.
In diesem Tutorial zeige ich Ihnen, wie Sie mit HolySheep AI eine robuste Fehlerkommunikationsstrategie aufbauen, die sowohl technisch präzise als auch kundenfreundlich ist. Wir behandeln:
- Die Anatomie von API-Fehlern bei Large Language Models
- Wiederholungsstrategien mit exponentiellem Backoff
- Automatisierte Entschädigungsmechanismen
- Copy-paste-fähige Implementierungsbeispiele
Warum Fehlerkommunikation bei LLM-APIs entscheidend ist
API-Fehler bei LLM-Anbietern unterscheiden sich fundamental von klassischen REST-API-Fehlern. Während ein 404 "Not Found" selbsterklärend ist, können bei Language Models folgende Szenarien auftreten:
- Rate-Limits — Der Dienst ist temporär überlastet
- Context-Window-Überschreitung — Die Eingabe überschreitet die Modellkapazität
- Modell-Überlastung — Bestimmte Modellversionen sind zeitweise nicht verfügbar
- Timeout — Die Antwort dauert länger als erwartet
- Qualitätsprobleme — Das Modell liefert unerwartete oder unbrauchbare Ergebnisse
Jedes dieser Szenarien erfordert eine andere Kommunikationsstrategie. In meiner Praxis bei HolySheep haben wir festgestellt, dass gut kommunizierte Fehler die Support-Tickets um bis zu 73% reduzieren können.
API-Aufruffehler-Kommunikationstemplate: Die Architektur
Bevor wir in den Code eintauchen, möchte ich Ihnen die Gesamtarchitektur unserer Fehlerkommunikationslösung vorstellen:
┌─────────────────────────────────────────────────────────────────┐
│ FEHLER-KOMMUNIKATIONS-FLOW │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ API- │───▶│ Fehler- │───▶│ User- │───▶│ Aktion/ │ │
│ │ Aufruf │ │ klassif. │ │ Message │ │ Retry │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Request │ │ Retry │ │ Credit- │ │ Fallback │ │
│ │ Context │ │ Strategy │ │Refund/Fix│ │ Model │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘Preisvergleich 2026: Warum HolySheep AI die kosteneffizienteste Wahl ist
Bevor wir zur technischen Implementierung kommen, lassen Sie uns die aktuellen 2026-Preise der führenden LLM-Anbieter vergleichen:
| Anbieter / Modell | Output-Preis ($/Million Token) | 10M Token/Monat | Latenz | Besonderheiten |
|---|---|---|---|---|
| DeepSeek V3.2 | $0.42 | $4.20 | <50ms | Beste Kosten-Effizienz |
| Gemini 2.5 Flash | $2.50 | $25.00 | <80ms | Google-Ökosystem |
| GPT-4.1 | $8.00 | $80.00 | <120ms | Branchenstandard |
| Claude Sonnet 4.5 | $15.00 | $150.00 | <150ms | Höchste Qualität |
Ersparnis mit HolySheep AI: Durch unseren WeChat/Alipay-Zahlungsweg und den RMB-Umtauschkurs (¥1=$1) profitieren Sie von 85%+ Ersparnis gegenüber den Standardpreisen. Bei 10 Millionen Token/Monat mit DeepSeek V3.2 zahlen Sie effektiv nur ca. $0.42 statt der üblichen Marktpreise.
Geeignet / Nicht geeignet für
✓ Perfekt geeignet für:
- Produktionsumgebungen mit SLA-Anforderungen — Automatische Fallback-Mechanismen
- Kostenkritische Anwendungen — 85%+ Ersparnis bei gleicher API-Kompatibilität
- China-Markt Projekte — WeChat Pay, Alipay, inländische Server
- Batch-Verarbeitung — Hohe Volumen mit automatischer Retry-Logik
- Startups und MVPs — Kostenlose Credits für den Einstieg
✗ Weniger geeignet für:
- Maxime Qualität über Kosten — Dann eher Claude 4.5 direkt
- Vollständig eigene Modell-Infrastruktur — Cloud-Anbieter-Vertrauen nötig
- Extrem sensible Daten ohne Compliance-Freigabe — Prüfen Sie die AGB
Preise und ROI
Mit HolySheep AI erhalten Sie:
| Plan | Features | Preis | Ideal für |
|---|---|---|---|
| Kostenlos | 5$ Startguthaben, alle Modelle | Kostenlos | Tests, Prototypen |
| Pay-as-you-go | Keine Mindestabnahme, Volumenrabatte | Ab $0.42/MTok | Startups, variable Last |
| Enterprise | SLA, dedizierte Kontingente, Support | Auf Anfrage | Großkunden |
ROI-Beispiel: Ein mittleres SaaS-Produkt mit 50M Token/Monat spart mit HolySheep vs. OpenAI direkt über $370 monatlich — das sind über $4.400 jährlich, die Sie in Produktentwicklung investieren können.
Technische Implementierung: Schritt-für-Schritt
Schritt 1: Grundlegendes Fehler-Handling mit Retry-Logik
Der folgende Python-Code zeigt, wie Sie API-Aufrufe mit automatischer Wiederholungslogik und verständlichen Fehlermeldungen implementieren:
#!/usr/bin/env python3
"""
HolySheep AI - Robustes API-Fehler-Handling mit Retry-Logik
Base URL: https://api.holysheep.ai/v1
"""
import requests
import time
import json
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
class ErrorSeverity(Enum):
"""Schweregrade für Fehlerklassifizierung"""
LOW = "low" # Temporär, automatisch retrybar
MEDIUM = "medium" # Retry mit Backoff
HIGH = "high" # Erfordert Benutzereingriff
CRITICAL = "critical" # Sofortige Benachrichtigung
@dataclass
class APIError:
"""Strukturierte Fehlerrepräsentation"""
code: str
message: str
severity: ErrorSeverity
user_message: str
retry_after: Optional[int] = None
compensation: Optional[Dict[str, Any]] = None
class HolySheepAIClient:
"""Robuster API-Client mit Fehlerbehandlung und Benachrichtigungen"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.max_retries = 3
self.timeout = 30
def _classify_error(self, status_code: int, response_data: Dict) -> APIError:
"""Klassifiziert API-Fehler und generiert Benutzernachrichten"""
error_mapping = {
429: APIError(
code="RATE_LIMIT_EXCEEDED",
message="Rate limit reached for this endpoint",
severity=ErrorSeverity.MEDIUM,
user_message="⏳ Wir haben momentarily zu viele Anfragen. "
"Bitte warten Sie {} Sekunden. "
"Wir versuchen es automatisch erneut.".format(
response_data.get('retry_after', 5)),
retry_after=response_data.get('retry_after', 5),
compensation={"type": "extended_context", "bonus_tokens": 5000}
),
500: APIError(
code="MODEL_SERVICE_ERROR",
message="Internal server error from LLM provider",
severity=ErrorSeverity.MEDIUM,
user_message="🔧 Unser LLM-Service hat momentarily Probleme. "
"Unsere Techniker arbeiten daran. "
"Automatischer Retry in 3 Sekunden...",
retry_after=3,
compensation={"type": "retry_credit", "free_requests": 2}
),
503: APIError(
code="SERVICE_UNAVAILABLE",
message="Service temporarily unavailable",
severity=ErrorSeverity.HIGH,
user_message="🚧 Wartungsarbeiten im Gange. "
"Voraussichtliche Dauer: 5-10 Minuten. "
"Alternative Modelle werden automatisch geprüft.",
compensation={"type": "fallback_credit", "amount_usd": 0.50}
),
400: APIError(
code="INVALID_REQUEST",
message="Request validation failed",
severity=ErrorSeverity.LOW,
user_message="⚠️ Ihre Anfrage enthält ungültige Parameter. "
"Bitte überprüfen Sie die Eingabelänge oder "
"verwenden Sie ein Modell mit größerem Kontextfenster."
),
401: APIError(
code="AUTHENTICATION_FAILED",
message="Invalid or expired API key",
severity=ErrorSeverity.CRITICAL,
user_message="🔐 Authentifizierungsfehler. "
"Bitte überprüfen Sie Ihren API-Key unter "
"https://www.holysheep.ai/settings"
),
408: APIError(
code="REQUEST_TIMEOUT",
message="Request timed out",
severity=ErrorSeverity.LOW,
user_message="⏱️ Die Anfrage dauerte zu lange. "
"Wir erhöhen den Timeout und versuchen es erneut.",
retry_after=5
)
}
return error_mapping.get(
status_code,
APIError(
code="UNKNOWN_ERROR",
message=response_data.get('error', {}).get('message', 'Unknown error'),
severity=ErrorSeverity.MEDIUM,
user_message="❓ Ein unerwarteter Fehler ist aufgetreten. "
"Unser Team wurde benachrichtigt."
)
)
def call_with_retry(self, messages: list, model: str = "deepseek-v3",
temperature: float = 0.7) -> Dict[str, Any]:
"""
Führt API-Aufruf mit exponentiellem Backoff und
automatischer Fallback-Strategie durch.
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature
}
retry_count = 0
last_error: Optional[APIError] = None
# Exponentieller Backoff: 1s, 2s, 4s, 8s...
backoff_times = [1, 2, 4, 8, 16]
while retry_count <= self.max_retries:
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=self.timeout
)
if response.status_code == 200:
result = response.json()
# Erfolg nach Fehlern: Entschädigung anbieten
if last_error and last_error.compensation:
result['_compensation_offer'] = last_error.compensation
result['_message'] = (
f"✅ Anfrage erfolgreich! "
f"Als Entschädigung für die Wartezeit: "
f"{last_error.compensation}"
)
return result
else:
response_data = response.json()
error = self._classify_error(response.status_code, response_data)
# Protokollierung für Analytics
print(f"[RETRY {retry_count}] {error.code}: {error.message}")
print(f"[USER SEE] {error.user_message}")
# Kritische Fehler nicht retryen
if error.severity == ErrorSeverity.CRITICAL:
raise Exception(f"Critical error: {error.user_message}")
# Retry wenn möglich
if retry_count < self.max_retries and error.retry_after:
wait_time = backoff_times[min(retry_count, len(backoff_times)-1)]
print(f"[WAIT] Sleeping {wait_time} seconds...")
time.sleep(wait_time)
retry_count += 1
last_error = error
else:
raise Exception(error.user_message)
except requests.exceptions.Timeout:
error = APIError(
code="TIMEOUT",
message="Request timeout",
severity=ErrorSeverity.MEDIUM,
user_message="⏱️ Timeout: Das Modell antwortet nicht schnell genug.",
retry_after=5
)
print(f"[TIMEOUT] Retrying in {error.retry_after}s...")
time.sleep(error.retry_after)
retry_count += 1
except requests.exceptions.ConnectionError:
print("[CONNECTION] Network error, retrying...")
time.sleep(2 ** retry_count)
retry_count += 1
raise Exception(
f"❌ Alle {self.max_retries + 1} Versuche fehlgeschlagen. "
f"Bitte kontaktieren Sie den Support oder versuchen Sie später erneut."
)
Beispiel-Nutzung
if __name__ == "__main__":
client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
messages = [
{"role": "system", "content": "Sie sind ein hilfreicher Assistent."},
{"role": "user", "content": "Erklären Sie Quantencomputing in 2 Sätzen."}
]
try:
response = client.call_with_retry(messages, model="deepseek-v3")
print(f"Antwort: {response['choices'][0]['message']['content']}")
if '_compensation_offer' in response:
print(f"Angebot: {response['_compensation_offer']}")
except Exception as e:
print(f"Fehler: {e}")Schritt 2: TypeScript/JavaScript-Implementierung für Frontend-Integration
Diese TypeScript-Implementierung eignet sich hervorragend für Frontend-Anwendungen und bietet eine vollständige Fehlerkommunikations-UI:
/**
* HolySheep AI - TypeScript Client mit Fehler-UI-Integration
* Base URL: https://api.holysheep.ai/v1
*/
interface HolySheepError {
code: string;
message: string;
severity: 'low' | 'medium' | 'high' | 'critical';
userMessage: string;
retryAfter?: number;
compensation?: CompensationOffer;
retryCount?: number;
}
interface CompensationOffer {
type: 'credit' | 'free_requests' | 'extended_context' | 'refund';
amount?: number;
bonusTokens?: number;
freeRequests?: number;
}
interface ChatMessage {
role: 'system' | 'user' | 'assistant';
content: string;
}
interface RetryConfig {
maxRetries: number;
baseDelay: number;
maxDelay: number;
exponential: boolean;
}
// Vordefinierte Fehlermeldungen für verschiedene Szenarien
const ERROR_TEMPLATES: Record<number, Omit<HolySheepError, 'code' | 'message'>> = {
429: {
severity: 'medium',
retryAfter: 5,
userMessage: '⏳ Wir haben momentarily zu viele Anfragen von anderen Nutzern erhalten. '
+ 'Bitte warten Sie {seconds} Sekunden. Ihre Anfrage wird automatisch '
+ 'erneut gesendet.',
compensation: { type: 'extended_context', bonusTokens: 5000 }
},
500: {
severity: 'medium',
retryAfter: 3,
userMessage: '🔧 Unser Server hatte einen kurzen Aussetzer. '
+ 'Keine Sorge – wir versuchen es automatisch erneut!',
compensation: { type: 'free_requests', freeRequests: 2 }
},
503: {
severity: 'high',
userMessage: '🚧 Wir führen gerade Wartungsarbeiten durch. '
+ 'Voraussichtliche Dauer: wenige Minuten. '
+ 'Wir informieren Sie, sobald alles wieder läuft!',
compensation: { type: 'credit', amount: 0.50 }
},
408: {
severity: 'low',
retryAfter: 5,
userMessage: '⏱️ Die KI braucht etwas länger als erwartet. '
+ 'Wir erhöhen die Wartezeit und versuchen es nochmal.'
},
400: {
severity: 'low',
userMessage: '⚠️ Ihre Eingabe war zu lang für das gewählte Modell. '
+ 'Versuchen Sie, den Text zu kürzen, oder wählen Sie ein '
+ 'Modell mit größerem Kontextfenster.'
}
};
class HolySheepErrorHandler {
private errorHistory: HolySheepError[] = [];
/**
* Parst HTTP-Fehler in benutzerfreundliche Fehlerobjekte
*/
parseError(status: number, response?: any): HolySheepError {
const template = ERROR_TEMPLATES[status] || {
severity: 'medium',
userMessage: '😅 Etwas ist schiefgelaufen. '
+ 'Unser Team wurde automatisch informiert.'
};
const error: HolySheepError = {
code: HS_${status}_${Date.now()},
message: response?.error?.message || HTTP ${status},
...template,
userMessage: this.interpolate(template.userMessage, {
seconds: template.retryAfter || 5
})
};
this.errorHistory.push(error);
return error;
}
/**
* Interpolation von Template-Platzhaltern
*/
private interpolate(template: string, vars: Record<string, any>): string {
return template.replace(/\{(\w+)\}/g, (_, key) =>
String(vars[key] ?? {${key}})
);
}
/**
* Generiert HTML für die Fehleranzeige
*/
renderErrorUI(error: HolySheepError): string {
const severityColors = {
low: '#FFA500', // Orange
medium: '#4169E1', // Royal Blue
high: '#DC143C', // Crimson
critical: '#8B0000' // Dark Red
};
const icons = {
low: '💡',
medium: '⚠️',
high: '🚨',
critical: '🔴'
};
return `
<div class="hs-error-card" style="
border-left: 4px solid ${severityColors[error.severity]};
background: linear-gradient(135deg, #f8f9fa 0%, #fff 100%);
padding: 16px 20px;
border-radius: 8px;
margin: 12px 0;
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif;
box-shadow: 0 2px 8px rgba(0,0,0,0.08);
">
<div style="display: flex; align-items: start; gap: 12px;">
<span style="font-size: 24px;">${icons[error.severity]}</span>
<div style="flex: 1;">
<p style="
margin: 0 0 8px 0;
font-size: 15px;
line-height: 1.5;
color: #2c3e50;
">${error.userMessage}</p>
${error.retryAfter ? `
<div class="hs-retry-timer" data-seconds="${error.retryAfter}" style="
display: inline-flex;
align-items: center;
gap: 8px;
background: ${severityColors[error.severity]}22;
padding: 6px 12px;
border-radius: 20px;
font-size: 13px;
color: ${severityColors[error.severity]};
">
<span>⏱️ Retry in </span>
<strong>${error.retryAfter}s</strong>
</div>
` : ''}
${error.compensation ? `
<div style="
margin-top: 12px;
padding: 10px 14px;
background: linear-gradient(135deg, #4CAF50 0%, #45a049 100%);
color: white;
border-radius: 6px;
font-size: 14px;
">
🎁 Entschädigung für Sie:
${this.formatCompensation(error.compensation)}
</div>
` : ''}
<small style="
display: block;
margin-top: 8px;
color: #999;
font-size: 11px;
">
Fehler-Code: ${error.code}
</small>
</div>
</div>
</div>
`;
}
/**
* Formatiert Entschädigungsangebote lesbar
*/
private formatCompensation(comp: CompensationOffer): string {
switch (comp.type) {
case 'credit':
return $${comp.amount} Bonus-Guthaben gutgeschrieben;
case 'free_requests':
return ${comp.freeRequests} kostenlose Anfragen als Entschädigung;
case 'extended_context':
return ${comp.bonusTokens?.toLocaleString()} zusätzliche Token für Ihren nächsten Request;
case 'refund':
return 'Volle Rückerstattung der fehlgeschlagenen Anfrage';
default:
return 'Überraschungs-Bonus';
}
}
}
// Einsatzbeispiel
const errorHandler = new HolySheepErrorHandler();
// Simuliere verschiedenen Fehler
const rateLimitError = errorHandler.parseError(429);
console.log(errorHandler.renderErrorUI(rateLimitError));
// Ausgabe: Vollständige HTML-Fehlerkarte mit Timer und EntschädigungSchritt 3: Automatische Entschädigungslogik und Monitoring
Dieses Backend-System verwaltet automatisch Gutschriften und Benachrichtigungen basierend auf dem Fehlertyp:
#!/usr/bin/env python3
"""
HolySheep AI - Backend-Entschädigungssystem und Monitoring
Verarbeitet automatisch Fehler und gewährt Entschädigungen
"""
import sqlite3
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class CompensationRecord:
"""Aufzeichnung einer gewährten Entschädigung"""
user_id: str
error_code: str
compensation_type: str
compensation_value: float
token_amount: Optional[int] = None
granted_at: datetime = None
class CompensationManager:
"""
Verwaltet automatische Entschädigungen basierend auf
Fehlerhäufigkeit und -schwere
"""
# Entschädigungsregeln: Fehlercode -> (Typ, Wert, Token-Bonus)
COMPENSATION_RULES = {
'RATE_LIMIT_EXCEEDED': ('credit', 0.25, 2000),
'MODEL_SERVICE_ERROR': ('free_requests', 3, 0),
'SERVICE_UNAVAILABLE': ('credit', 0.50, 5000),
'REQUEST_TIMEOUT': ('extended_context', 0, 3000),
'AUTHENTICATION_FAILED': ('refund', 1.0, 0), # 100% Refund
'INVALID_REQUEST': ('extended_context', 0, 1000),
}
# Schwellenwerte für automatische Kompensation
RETRY_THRESHOLD = 3 # Nach 3 Retries automatisch entschädigen
CONSECUTIVE_ERRORS_THRESHOLD = 5 # 5 aufeinanderfolgende Fehler = Bonus
def __init__(self, db_path: str = "holysheep_compensation.db"):
self.db_path = db_path
self._init_database()
def _init_database(self):
"""Initialisiert die SQLite-Datenbank für Kompensationen"""
with sqlite3.connect(self.db_path) as conn:
conn.execute('''
CREATE TABLE IF NOT EXISTS compensations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
error_code TEXT NOT NULL,
compensation_type TEXT NOT NULL,
compensation_value REAL NOT NULL,
token_amount INTEGER,
granted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
applied BOOLEAN DEFAULT FALSE
)
''')
conn.execute('''
CREATE TABLE IF NOT EXISTS error_stats (
user_id TEXT PRIMARY KEY,
total_errors INTEGER DEFAULT 0,
last_error_at TIMESTAMP,
consecutive_errors INTEGER DEFAULT 0,
total_compensation_credit REAL DEFAULT 0
)
''')
def record_error(self, user_id: str, error_code: str,
retry_count: int = 0) -> Optional[CompensationRecord]:
"""
Zeichnet einen Fehler auf und bestimmt automatisch
eine Entschädigung
"""
# Statistik aktualisieren
self._update_error_stats(user_id, error_code)
# Prüfe ob Kompensation fällig ist
compensation = self._determine_compensation(
user_id, error_code, retry_count
)
if compensation:
self._grant_compensation(user_id, error_code, compensation)
logger.info(
f"Kompensation gewährt für User {user_id}: "
f"{compensation.compensation_type} = {compensation.compensation_value}"
)
return compensation
return None
def _update_error_stats(self, user_id: str, error_code: str):
"""Aktualisiert Fehlerstatistiken für einen Benutzer"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
# Hole aktuelle Stats
cursor.execute(
'SELECT * FROM error_stats WHERE user_id = ?',
(user_id,)
)
stats = cursor.fetchone()
now = datetime.now()
if stats:
total_errors, last_error, consecutive, total_comp = stats
new_consecutive = consecutive + 1
# Setze Zähler zurück wenn letzter Fehler > 1 Minute her
if last_error:
last_error_dt = datetime.fromisoformat(last_error)
if (now - last_error_dt).seconds > 60:
new_consecutive = 1
cursor.execute('''
UPDATE error_stats
SET total_errors = total_errors + 1,
last_error_at = ?,
consecutive_errors = ?
WHERE user_id = ?
''', (now.isoformat(), new_consecutive, user_id))
else:
cursor.execute('''
INSERT INTO error_stats
(user_id, total_errors, last_error_at, consecutive_errors)
VALUES (?, 1, ?, 1)
''', (user_id, now.isoformat()))
conn.commit()
def _determine_compensation(self, user_id: str, error_code: str,
retry_count: int) -> Optional[CompensationRecord]:
"""
Bestimmt basierend auf Fehlerhistorie ob eine
Entschädigung gewährt wird
"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
# Prüfe konsekutive Fehler
cursor.execute(
'SELECT consecutive_errors FROM error_stats WHERE user_id = ?',
(user_id,)
)
result = cursor.fetchone()
consecutive = result[0] if result else 0
# Hole Entschädigungsregel
rule = self.COMPENSATION_RULES.get(error_code)
if not rule:
return None
comp_type, comp_value, token_bonus = rule
# Nur entschädigen wenn Retry-Schwelle erreicht oder
# konsekutive Fehler-Schwelle
if retry_count >= self.RETRY_THRESHOLD or \
consecutive >= self.CONSECUTIVE_ERRORS_THRESHOLD:
return CompensationRecord(
user_id=user_id,
error_code=error_code,
compensation_type=comp_type,
compensation_value=comp_value,
token_amount=token_bonus
)
return None
def _grant_compensation(self, user_id: str, error_code: str,
compensation: CompensationRecord):
"""Speichert und verarbeitet die Entschädigung"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute('''
INSERT INTO compensations
(user_id, error_code, compensation_type,
compensation_value, token_amount)
VALUES (?, ?, ?, ?, ?)
''', (
compensation.user_id,
compensation.error_code,
compensation.compensation_type,
compensation.compensation_value,
compensation.token_amount
))
# Aktualisiere Gesamtkompensation des Users
cursor.execute('''
UPDATE error_stats
SET total_compensation_credit = total_compensation_credit + ?
WHERE user_id = ?
''', (compensation.compensation_value, user_id))
conn.commit()
def generate_user_report(self, user_id: str) -> Dict:
"""Generiert einen Bericht für Support/Analytics"""
with sqlite3.connect(self.db_path) as conn:
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# Fehlerstatistik
cursor.execute(
'SELECT * FROM error_stats WHERE user_id = ?',
(user_id,)
)
stats = cursor.fetchone()
# Letzte 10 Kompensationen
cursor.execute('''
SELECT * FROM compensations
WHERE user_id = ?
ORDER BY granted_at DESC
LIMIT 10
''', (user_id,))
compensations = cursor.fetchall()
return {
'user_id': user_id,
'stats': dict(stats) if stats else None,
'recent_compensations': [
dict(c) for c in compensations
],
'generated_at': datetime.now().isoformat()
}
Monitoring-Dashboard-Daten
class MonitoringDashboard:
"""Sammelt Metriken für das Admin-Dashboard"""
def __init__(self, db_path: str = "holysheep_compensation.db"):
self.db_path = db_path
def get_error_distribution(self) -> List[Dict]:
"""Liefert Verteilung der Fehlercodes"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute('''
SELECT error_code, COUNT(*) as count
FROM compensations
GROUP BY error_code
ORDER BY count DESC
''')
return [
{'code': row[0], 'count': row[1]}
for row in cursor.fetchall()
]
def get_compensation_costs(self, days: int = 30) -> Dict:
"""Berechnet Kompensationskosten über Zeitraum"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute('''
SELECT
DATE(granted_at) as date,
SUM(compensation_value) as total_cost,
COUNT(*) as transactions
FROM compensations
WHERE granted_at >= datetime('now', '-' || ? ||