Introduction
En tant qu'architecte senior ayant déployé des systèmes d'IA en production dans des environnements hautement régulés (finance, santé, secteur public), je peux vous confirmer que la conformité légale des API IA est devenue un impératif non négociable. Les réglementations comme le RGPD, la directive IA européenne, et les lois sectorielles exigent désormais une traçabilité complète et une auditabilité en temps réel de chaque interaction avec les modèles d'intelligence artificielle.
Dans ce tutoriel, je vais vous présenter une architecture complète d'un outil d'audit de conformité pour les API IA, construite autour de l'API HolySheep qui offre des avantages significatifs : un taux de change optimal avec ¥1=$1 soit une économie de 85% par rapport aux fournisseurs traditionnels, une latence inférieure à 50ms qui permet un audit en temps réel sans créer de goulot d'étranglement, et la possibilité de payer via WeChat et Alipay pour les équipes asiatiques.
Vous pouvez vous inscrire ici pour obtenir vos crédits gratuits et commencer à tester cette architecture.
Architecture Globale du Système d'Audit
Le système d'audit de conformité se compose de quatre couches distinctes mais interdépendantes :
- Couche de Capture : Interception et journalisation de toutes les requêtes/réponses API
- Couche de Traitement : Analyse en temps réel des contenus pour détection de PII et违规内容
- Couche de Stockage : Persistance sécurisée avec cryptage bout-en-bout et rétention configurable
- Couche de Reporting : Génération de rapports d'audit et alertes en cas de non-conformité
Implémentation du Module Principal
Commençons par l'implémentation du client d'audit complet. Ce module intègre nativement les appels à l'API HolySheep pour l'analyse de contenu, tout en maintenant un audit trail complet.
#!/usr/bin/env python3
"""
Compliance Audit Client pour API IA
Auteur: HolySheep AI Technical Team
Version: 2.1.0
License: MIT
Ce module implémente un système complet d'audit de conformité
pour les interactions avec les API d'IA, incluant:
- Journalisation des prompts et réponses
- Détection de PII (Personally Identifiable Information)
- Analyse de contenu réglementé
- Calcul de coûts en temps réel
"""
import asyncio
import hashlib
import hmac
import json
import logging
import time
from dataclasses import dataclass, field, asdict
from datetime import datetime, timedelta, timezone
from enum import Enum
from typing import Any, Optional, Dict, List, Callable, Union
from contextlib import asynccontextmanager
import traceback
Dépendances optionnelles
try:
import aiohttp
AIOHTTP_AVAILABLE = True
except ImportError:
AIOHTTP_AVAILABLE = False
try:
import cryptography
from cryptography.fernet import Fernet
CRYPTO_AVAILABLE = True
except ImportError:
CRYPTO_AVAILABLE = False
Configuration HolySheep API
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Remplacez par votre clé
Configuration des tarifs HolySheep 2026 (USD par million de tokens)
HOLYSHEEP_PRICING = {
"gpt-4.1": {"input": 8.00, "output": 8.00},
"claude-sonnet-4.5": {"input": 15.00, "output": 15.00},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50},
"deepseek-v3.2": {"input": 0.42, "output": 0.42},
}
Mapping des modèles
MODEL_MAPPING = {
"gpt-4": "gpt-4.1",
"claude": "claude-sonnet-4.5",
"gemini-flash": "gemini-2.5-flash",
"deepseek": "deepseek-v3.2",
}
class AuditLevel(Enum):
"""Niveaux d'audit disponibles"""
MINIMAL = "minimal" # Timestamps et IDs uniquement
STANDARD = "standard" # Prompts et réponses hashés
FULL = "full" # Contenu complet avec métadonnées
COMPLIANCE = "compliance" # Audit complet pour conformité réglementaire
class ContentCategory(Enum):
"""Catégories de contenu pour classification"""
CLEAN = "clean"
PII_DETECTED = "pii_detected"
SENSITIVE_DATA = "sensitive_data"
POLICY_VIOLATION = "policy_violation"
SUSPICIOUS = "suspicious"
COMPLIANT = "compliant"
class ViolationType(Enum):
"""Types de violations potentielles"""
NONE = "none"
GDPR_PII = "gdpr_pii" # Données personnelles non protégées
FINANCIAL_PCI = "financial_pci" # Données de carte bancaire
HEALTH_HIPAA = "health_hipaa" # Informations de santé
AGE_RESTRICTION = "age_restriction" # Contenu inadapté aux mineurs
PROHIBITED_CONTENT = "prohibited" # Contenu interdit par la politique
@dataclass
class AuditEntry:
"""Entrée d'audit individuelle"""
entry_id: str
timestamp: str
request_id: str
user_id: Optional[str]
session_id: str
ip_address: Optional[str]
model: str
prompt_tokens: int
completion_tokens: int
total_tokens: int
cost_usd: float
cost_cny: float
latency_ms: float
prompt_hash: str
response_hash: str
prompt_preview: str
response_preview: str
content_category: str
violations: List[str]
compliance_flags: Dict[str, bool]
metadata: Dict[str, Any]
def to_dict(self) -> Dict[str, Any]:
"""Convertit l'entrée en dictionnaire sérialisable"""
data = asdict(self)
data['timestamp'] = self.timestamp.isoformat() if isinstance(self.timestamp, datetime) else self.timestamp
return data
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'AuditEntry':
"""Crée une entrée depuis un dictionnaire"""
if isinstance(data.get('timestamp'), str):
data['timestamp'] = datetime.fromisoformat(data['timestamp'])
return cls(**data)
class CostCalculator:
"""Calculateur de coûts pour l'API HolySheep"""
def __init__(self, pricing: Optional[Dict[str, Dict[str, float]]] = None):
self.pricing = pricing or HOLYSHEEP_PRICING
def calculate_cost(
self,
model: str,
prompt_tokens: int,
completion_tokens: int
) -> Dict[str, Union[float, Dict[str, float]]]:
"""
Calcule le coût exact en USD et CNY
Args:
model: Nom du modèle (ex: "deepseek-v3.2")
prompt_tokens: Nombre de tokens d'entrée
completion_tokens: Nombre de tokens de sortie
Returns:
Dict contenant les coûts détaillés
"""
# Normalisation du nom du modèle
normalized_model = MODEL_MAPPING.get(model, model)
if normalized_model not in self.pricing:
# Prix par défaut si modèle non reconnu
logging.warning(f"Modèle {model} non reconnu, utilisation du prix DeepSeek")
normalized_model = "deepseek-v3.2"
pricing = self.pricing[normalized_model]
# Calcul des coûts
input_cost = (prompt_tokens / 1_000_000) * pricing["input"]
output_cost = (completion_tokens / 1_000_000) * pricing["output"]
total_usd = input_cost + output_cost
# Conversion CNY (taux HolySheep: ¥1 = $1)
total_cny = total_usd
return {
"input_tokens": prompt_tokens,
"output_tokens": completion_tokens,
"total_tokens": prompt_tokens + completion_tokens,
"input_cost_usd": round(input_cost, 6),
"output_cost_usd": round(output_cost, 6),
"total_usd": round(total_usd, 6),
"total_cny": round(total_cny, 6),
"model": normalized_model,
"pricing_tier": pricing
}
class PIIAnalyzer:
"""Analyseur de PII intégré à HolySheep API"""
PII_PATTERNS = {
"email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
"phone_fr": r'\b(?:\+33|0033|0)[1-9](?:[.\-\s]?\d{2}){4}\b',
"phone_cn": r'\b1[3-9]\d{9}\b',
"ssn_us": r'\b\d{3}-\d{2}-\d{4}\b',
"credit_card": r'\b(?:\d{4}[-\s]?){3}\d{4}\b',
"iban": r'\b[A-Z]{2}\d{2}[A-Z0-9]{4,30}\b',
"passport": r'\b[A-Z]{1,2}\d{6,9}\b',
}
def __init__(self, api_key: str, base_url: str = HOLYSHEEP_BASE_URL):
self.api_key = api_key
self.base_url = base_url
self._session = None
async def analyze_content(self, text: str) -> Dict[str, Any]:
"""
Analyse le contenu pour détecter les PII
Returns:
Dict avec les types de PII détectés et leurslocalisations
"""
import re
findings = {
"has_pii": False,
"pii_types": [],
"occurrences": [],
"risk_score": 0.0
}
for pii_type, pattern in self.PII_PATTERNS.items():
matches = list(re.finditer(pattern, text, re.IGNORECASE))
if matches:
findings["has_pii"] = True
findings["pii_types"].append(pii_type)
for match in matches:
findings["occurrences"].append({
"type": pii_type,
"value": match.group(),
"start": match.start(),
"end": match.end()
})
# Calcul du score de risque
if findings["has_pii"]:
findings["risk_score"] = min(1.0, len(findings["occurrences"]) * 0.2)
return findings
async def analyze_with_ai(self, text: str) -> Dict[str, Any]:
"""
Utilise HolySheep API pour une analyse sémantique plus approfondie
Cette méthode est plus précise pour détecter des PII dans des contextes complexes
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
prompt = f"""Analyse ce texte pour les données personnelles (PII) et conforms au RGPD:
Texte à analyser:
{text[:2000]} # Limite de caractères
Réponds en JSON avec:
- pii_detected: boolean
- pii_types: array des types trouvés (email, phone, address, name, etc.)
- risk_level: low/medium/high/critical
- gdpr_violations: array des articles RGPD potentiellement violés
- recommendations: array de recommandations de remediation"""
payload = {
"model": "deepseek-v3.2", # Modèle le plus économique
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.1,
"max_tokens": 500
}
if not AIOHTTP_AVAILABLE:
return {"error": "aiohttp requis pour cette fonctionnalité"}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=10)
) as response:
if response.status != 200:
return {"error": f"API error: {response.status}"}
result = await response.json()
content = result["choices"][0]["message"]["content"]
# Parsing JSON de la réponse
try:
import json as json_module
return json_module.loads(content)
except:
return {"raw_analysis": content}
class ComplianceAuditClient:
"""
Client principal d'audit de conformité pour les API IA
Fonctionnalités:
- Journalisation complète de toutes les interactions
- Détection automatique de PII
- Calcul de coûts en temps réel
- Génération de rapports d'audit
- Conformité RGPD et autres réglementations
"""
def __init__(
self,
api_key: str = HOLYSHEEP_API_KEY,
base_url: str = HOLYSHEEP_BASE_URL,
audit_level: AuditLevel = AuditLevel.COMPLIANCE,
encryption_key: Optional[bytes] = None,
storage_backend: Optional[Any] = None
):
self.api_key = api_key
self.base_url = base_url
self.audit_level = audit_level
self.audit_log: List[AuditEntry] = []
self.cost_calculator = CostCalculator()
self.pii_analyzer = PIIAnalyzer(api_key, base_url)
# Configuration du chiffrement si disponible
if encryption_key and CRYPTO_AVAILABLE:
self.cipher = Fernet(encryption_key)
else:
self.cipher = None
# Backend de stockage
self.storage = storage_backend or InMemoryAuditStorage()
# Configuration du logging
self.logger = logging.getLogger("ComplianceAudit")
self.logger.setLevel(logging.INFO)
# Compteurs de statistiques
self.stats = {
"total_requests": 0,
"total_tokens": 0,
"total_cost_usd": 0.0,
"total_cost_cny": 0.0,
"violations_detected": 0,
"pii_detected": 0,
"avg_latency_ms": 0.0
}
def _generate_entry_id(self, session_id: str, index: int) -> str:
"""Génère un ID unique pour l'entrée d'audit"""
timestamp = datetime.now(timezone.utc).isoformat()
data = f"{session_id}:{index}:{timestamp}"
return hashlib.sha256(data.encode()).hexdigest()[:16]
def _hash_content(self, content: str) -> str:
"""Génère un hash SHA-256 du contenu"""
return hashlib.sha256(content.encode()).hexdigest()
def _create_preview(self, content: str, max_length: int = 200) -> str:
"""Crée un aperçu tronqué du contenu"""
if len(content) <= max_length:
return content
return content[:max_length] + "..."
@asynccontextmanager
async def audit_request(
self,
model: str,
messages: List[Dict[str, str]],
user_id: Optional[str] = None,
session_id: Optional[str] = None,
ip_address: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None
):
"""
Context manager pour auditer une requête API
Usage:
async with client.audit_request("deepseek-v3.2", messages) as result:
response = result["response"]
"""
session_id = session_id or hashlib.md5(str(time.time()).encode()).hexdigest()
request_id = hashlib.uuid4().hex
start_time = time.time()
# Construire le prompt complet pour l'analyse
prompt_text = "\n".join([m.get("content", "") for m in messages])
entry = AuditEntry(
entry_id=self._generate_entry_id(session_id, self.stats["total_requests"]),
timestamp=datetime.now(timezone.utc),
request_id=request_id,
user_id=user_id,
session_id=session_id,
ip_address=ip_address,
model=model,
prompt_tokens=0,
completion_tokens=0,
total_tokens=0,
cost_usd=0.0,
cost_cny=0.0,
latency_ms=0.0,
prompt_hash=self._hash_content(prompt_text),
response_hash="",
prompt_preview=self._create_preview(prompt_text),
response_preview="",
content_category=ContentCategory.CLEAN.value,
violations=[],
compliance_flags={
"gdpr_compliant": True,
"pii_analyzed": False,
"encrypted": self.cipher is not None
},
metadata=metadata or {}
)
result_data = {"entry": entry, "response": None, "error": None}
try:
# Effectuer la requête API
response = await self._make_request(model, messages)
# Extraire les informations de la réponse
usage = response.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
# Calculer les coûts
costs = self.cost_calculator.calculate_cost(model, prompt_tokens, completion_tokens)
# Mettre à jour l'entrée d'audit
response_content = response.get("choices", [{}])[0].get("message", {}).get("content", "")
entry.prompt_tokens = prompt_tokens
entry.completion_tokens = completion_tokens
entry.total_tokens = costs["total_tokens"]
entry.cost_usd = costs["total_usd"]
entry.cost_cny = costs["total_cny"]
entry.latency_ms = (time.time() - start_time) * 1000
entry.response_hash = self._hash_content(response_content)
entry.response_preview = self._create_preview(response_content)
# Analyser le contenu pour PII
if self.audit_level in [AuditLevel.FULL, AuditLevel.COMPLIANCE]:
pii_analysis = await self.pii_analyzer.analyze_content(prompt_text)
if pii_analysis["has_pii"]:
entry.content_category = ContentCategory.PII_DETECTED.value
entry.compliance_flags["pii_analyzed"] = True
entry.violations.append(f"PII_DETECTED:{','.join(pii_analysis['pii_types'])}")
# Analyser la réponse également
if response_content:
response_pii = await self.pii_analyzer.analyze_content(response_content)
if response_pii["has_pii"]:
entry.content_category = ContentCategory.SENSITIVE_DATA.value
entry.violations.append(f"PII_IN_RESPONSE:{','.join(response_pii['pii_types'])}")
result_data["response"] = response
except Exception as e:
entry.violations.append(f"ERROR:{type(e).__name__}")
result_data["error"] = str(e)
self.logger.error(f"Erreur lors de la requête: {traceback.format_exc()}")
finally:
# Calculer la latence finale
entry.latency_ms = (time.time() - start_time) * 1000
# Mettre à jour les statistiques
self._update_stats(entry)
# Stocker l'entrée d'audit
await self.storage.store(entry)
self.audit_log.append(entry)
yield result_data
async def _make_request(
self,
model: str,
messages: List[Dict[str, str]]
) -> Dict[str, Any]:
"""Effectue la requête vers l'API HolySheep"""
if not AIOHTTP_AVAILABLE:
raise RuntimeError("aiohttp est requis. Installez-le: pip install aiohttp")
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 4096
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
if response.status != 200:
error_text = await response.text()
raise RuntimeError(f"API request failed: {response.status} - {error_text}")
return await response.json()
def _update_stats(self, entry: AuditEntry):
"""Met à jour les statistiques globales"""
self.stats["total_requests"] += 1
self.stats["total_tokens"] += entry.total_tokens
self.stats["total_cost_usd"] += entry.cost_usd
self.stats["total_cost_cny"] += entry.cost_cny
if entry.violations:
self.stats["violations_detected"] += len(entry.violations)
if entry.content_category in [ContentCategory.PII_DETECTED.value, ContentCategory.SENSITIVE_DATA.value]:
self.stats["pii_detected"] += 1
# Moyenne mobile de la latence
n = self.stats["total_requests"]
current_avg = self.stats["avg_latency_ms"]
self.stats["avg_latency_ms"] = ((current_avg * (n - 1)) + entry.latency_ms) / n
async def generate_compliance_report(
self,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None,
user_id: Optional[str] = None
) -> Dict[str, Any]:
"""
Génère un rapport de conformité complet
Args:
start_date: Date de début de la période
end_date: Date de fin de la période
user_id: Filtrer par utilisateur spécifique
Returns:
Rapport complet de conformité
"""
entries = await self.storage.query(
start_date=start_date,
end_date=end_date,
user_id=user_id
)
# Calcul des métriques
total_entries = len(entries)
violation_count = sum(1 for e in entries if e.violations)
pii_count = sum(1 for e in entries if e.content_category in [
ContentCategory.PII_DETECTED.value,
ContentCategory.SENSITIVE_DATA.value
])
# Répartition par catégorie
category_breakdown = {}
for entry in entries:
cat = entry.content_category
category_breakdown[cat] = category_breakdown.get(cat, 0) + 1
# Calcul des coûts par modèle
cost_by_model = {}
for entry in entries:
model = entry.model
if model not in cost_by_model:
cost_by_model[model] = {"usd": 0.0, "cny": 0.0, "tokens": 0}
cost_by_model[model]["usd"] += entry.cost_usd
cost_by_model[model]["cny"] += entry.cost_cny
cost_by_model[model]["tokens"] += entry.total_tokens
return {
"report_metadata": {
"generated_at": datetime.now(timezone.utc).isoformat(),
"period_start": start_date.isoformat() if start_date else None,
"period_end": end_date.isoformat() if end_date else None,
"user_filter": user_id
},
"summary": {
"total_requests": total_entries,
"total_tokens": sum(e.total_tokens for e in entries),
"total_cost_usd": sum(e.cost_usd for e in entries),
"total_cost_cny": sum(e.cost_cny for e in entries),
"avg_latency_ms": sum(e.latency_ms for e in entries) / max(1, total_entries),
"violations_detected": violation_count,
"pii_incidents": pii_count,
"compliance_rate": ((total_entries - violation_count) / max(1, total_entries)) * 100
},
"category_breakdown": category_breakdown,
"cost_by_model": cost_by_model,
"violations_detail": [
{"entry_id": e.entry_id, "violations": e.violations}
for e in entries if e.violations
]
}
def get_stats(self) -> Dict[str, Any]:
"""Retourne les statistiques actuelles"""
return self.stats.copy()
class InMemoryAuditStorage:
"""Stockage en mémoire pour les entrées d'audit (production: utilisez une DB)"""
def __init__(self):
self.entries: List[AuditEntry] = []
self._lock = asyncio.Lock()
async def store(self, entry: AuditEntry):
"""Stocke une entrée d'audit"""
async with self._lock:
self.entries.append(entry)
async def query(
self,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None,
user_id: Optional[str] = None
) -> List[AuditEntry]:
"""Interroge les entrées d'audit"""
async with self._lock:
results = self.entries.copy()
if start_date:
results = [e for e in results if e.timestamp >= start_date]
if end_date:
results = [e for e in results if e.timestamp <= end_date]
if user_id:
results = [e for e in results if e.user_id == user_id]
return results
Example d'utilisation
async def main():
"""Exemple d'utilisation du client d'audit"""
client = ComplianceAuditClient(
api_key=HOLYSHEEP_API_KEY,
audit_level=AuditLevel.COMPLIANCE
)
# Exemple de conversation à auditer
messages = [
{"role": "system", "content": "Tu es un assistant médical."},
{"role": "user", "content": "Mon patient Jean Dupont, né le 15/03/1980, "
"contact: [email protected], téléphone: 06 12 34 56 78, "
"présente ces symptômes depuis une semaine..."}
]
async with client.audit_request(
model="deepseek-v3.2",
messages=messages,
user_id="user_12345",
session_id="session_abc",
metadata={"context": "medical_consultation"}
) as result:
if result["error"]:
print(f"Erreur: {result['error']}")
else:
print(f"Réponse reçue: {result['response']}")
print(f"Entrée d'audit: {result['entry'].entry_id}")
# Générer le rapport de conformité
report = await client.generate_compliance_report()
print(f"\n=== RAPPORT DE CONPLIANCE ===")
print(f"Requêtes totales: {report['summary']['total_requests']}")
print(f"Coût total USD: ${report['summary']['total_cost_usd']:.4f}")
print(f"Coût total CNY: ¥{report['summary']['total_cost_cny']:.4f}")
print(f"Taux de conformité: {report['summary']['compliance_rate']:.1f}%")
print(f"Incidents PII: {report['summary']['pii_incidents']}")
if __name__ == "__main__":
asyncio.run(main())Module de Détection de Contenu Sensible
Ce module avancé utilise l'analyse sémantique de HolySheep pour détecter les contenus réglementés dans les deux sens (entrée et sortie), avec une précision accrue grâce aux modèles de langage.
#!/usr/bin/env python3
"""
Content Policy Enforcer - Module de détection de contenu sensible
Version: 1.5.0
Ce module implémente:
- Détection de contenus prohibés
- Vérification de conformité age-gating
- Analyse de sentiments pour contenu sensible
- Blacklist/whitelist configurable
- Alertes en temps réel
"""
import asyncio
import re
from typing import Dict, List, Optional, Tuple, Set
from dataclasses import dataclass
from enum import Enum
import json
class ContentType(Enum):
"""Types de contenus sensibles"""
VIOLENCE = "violence"
SEXUAL = "sexual"
HATE = "hate_speech"
SELF_HARM = "self_harm"
FRAUD = "fraud"
ILLEGAL = "illegal_content"
HARASSMENT = "harassment"
MISINFORMATION = "misinformation"
POLITICAL = "political"
RELIGIOUS = "religious"
class SeverityLevel(Enum):
"""Niveaux de sévérité"""
SAFE = "safe"
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class PolicyViolation:
"""Représente une violation de politique détectée"""
category: ContentType
severity: SeverityLevel
confidence: float # 0.0 à 1.0
matched_patterns: List[str]
suggestion: str
requires_review: bool
auto_action: Optional[str] = None # "block", "redact", "warn"
@dataclass
class PolicyCheckResult:
"""Résultat complet d'une vérification de politique"""
is_compliant: bool
violations: List[PolicyViolation]
overall_risk_score: float
requires_human_review: bool
metadata: Dict[str, any]
class ContentPolicyDatabase:
"""Base de données de politiques de contenu"""
# Patterns regex pour détection rapide
PATTERNS = {
ContentType.VIOLENCE: [
(r'\b(tuer|murder|assassiner|homicide)\b', 0.7),
(r'\b(arme|weapon|gun|pistolet)\b', 0.5),
(r'\b(bombe|bomb|explosif|explosive)\b', 0.9),
],
ContentType.SEXUAL: [
(r'\b(sexe|sex|nudité|nudity)\b', 0.6),
(r'\b(pornographie|pornography)\b', 0.9),
],
ContentType.SELF_HARM: [
(r'\b(suicide|automutilation|self-harm)\b', 0.9),
(r'\b(se blesser|hurt oneself)\b', 0.8),
],
ContentType.FRAUD: [
(r'\b(arnaque|fraud|scam|phishing)\b', 0.7),
(r'\b(faux compte|fake account)\b', 0.6),
],
ContentType.HARASSMENT: [
(r'\b(harcèlement|harassment|menace|threat)\b', 0.7),
(r'\b(doxing|doxxing)\b', 0.9),
],
}
# Blacklist de mots spécifiques
BLACKLIST = {
ContentType.VIOLENCE: {"terrorisme", "terrorist", "attentat"},
ContentType.SELF_HARM: {"suicider", "ends life"},
ContentType.FRAUD: {"pyramid scheme", "ponzi"},
ContentType.ILLEGAL: {"drogue", "drug", "stupéfiant"},
}
class ContentPolicyEnforcer:
"""
Moteur de vérification de politique de contenu
Fonctionnalités:
- Analyse par patterns regex (rapide)
- Analyse sémantique via HolySheep API (précise)
- Système de scoring de risque
- Actions automatiques configurables
"""
def __init__(
self,
api_key: str,
base_url: str = HOLYSHEEP_BASE_URL,
strict_mode: bool = True
):
self.api_key = api_key
self.base_url = base_url
self.strict_mode = strict_mode
self.policy_db = ContentPolicyDatabase()
# Configuration des seuils
self.thresholds = {
SeverityLevel.CRITICAL: 0.95,
SeverityLevel.HIGH: 0.80,
SeverityLevel.MEDIUM: 0.60,
SeverityLevel.LOW: 0.40,
}
# Compteurs de statistiques
self.stats = {
"total_checks": 0,
"violations_blocked": 0,
"violations_redacted": 0,
"warnings_issued": 0,
}
async def check_content(
self,
text: str,
context: Optional[Dict[str, any]] = None
) -> PolicyCheckResult:
"""
Vérifie le contenu contre toutes les politiques
Args:
text: Texte à vérifier
context: Contexte additionnel (langage, région, etc.)
Returns:
PolicyCheckResult avec le verdict complet
"""
self.stats["total_checks"] += 1
violations = []
# Étape 1: Vérification par patterns (rapide)
pattern_violations = await self._check_patterns(text)
violations.extend(pattern_violations)
# Étape 2: Vérification par blacklist
blacklist_violations = self._check_blacklist(text)
violations.extend(blacklist_violations)
# Étape 3: Analyse sémantique via API (si violations suspectées)
if violations or self.strict_mode:
semantic_violations = await self._semantic_analysis(text, context)
violations.extend(semantic_v