Trong bối cảnh các quy định bảo vệ dữ liệu ngày càng nghiêm ngặt như GDPR, LGPD, PDPA Việt Nam, việc xử lý log yêu cầu API trở thành thách thức lớn cho doanh nghiệp. Bài viết này sẽ hướng dẫn bạn xây dựng giải pháp log anonymization và compliant storage hoàn chỉnh với HolySheep AI API, giúp tiết kiệm 85%+ chi phí so với các giải pháp truyền thống.
So Sánh Chi Phí API Providers — 2026
Trước khi đi vào chi tiết kỹ thuật, hãy cùng xem bức tranh chi phí thực tế khi xử lý 10 triệu token/tháng:
| Provider | Giá Output/MTok | 10M Tokens/tháng | Tỷ lệ tiết kiệm |
|---|---|---|---|
| OpenAI GPT-4.1 | $8.00 | $80 | Baseline |
| Anthropic Claude Sonnet 4.5 | $15.00 | $150 | +87.5% |
| Google Gemini 2.5 Flash | $2.50 | $25 | -68.75% |
| HolySheep DeepSeek V3.2 | $0.42 | $4.20 | -94.75% ✓ |
Điều này có nghĩa: với chi phí bạn bỏ ra cho 1 tháng sử dụng Claude Sonnet 4.5, bạn có thể chạy 35 tháng với DeepSeek V3.2 trên HolySheep — đủ budget để đầu tư vào hệ thống log compliance hoàn chỉnh.
Giải Pháp Log Anonymization — Kiến Trúc Tổng Quan
Tại Sao Cần Anonymize Log?
- Tuân thủ pháp luật: GDPR Article 25, Vietnamese PDPD 2012, China's PIPL
- Bảo vệ người dùng: Tránh rò rỉ PII (Personally Identifiable Information)
- An ninh mạng: Giảm thiểu surface attack khi log bị leak
- Tối ưu chi phí lưu trữ: Dữ liệu anonymized nén tốt hơn, query nhanh hơn
Flow Xử Lý
┌─────────────┐ ┌──────────────────┐ ┌─────────────────┐ ┌──────────────┐
│ Client │───▶│ HolySheep API │───▶│ Anonymizer │───▶│ Storage │
│ (Raw Data) │ │ Gateway │ │ (Local/Cloud) │ │ (Compliant) │
└─────────────┘ └──────────────────┘ └─────────────────┘ └──────────────┘
│ │ │ │
│ ▼ ▼ ▼
│ ┌──────────────────┐ ┌─────────────────┐ ┌──────────────┐
│ │ Raw Log Buffer │ │ Hash + Mask │ │ Analytics │
│ │ (Temporary) │ │ + Tokenize │ │ Dashboard │
│ └──────────────────┘ └─────────────────┘ └──────────────┘
│ │ │
└───────────────────┴───────────────────────┘
Retention Policy: 24h max
Triển Khai Giải Pháp — Code Mẫu Hoàn Chỉnh
1. Logging Middleware với HolySheep API Integration
// holy_sheep_logger.py
// Framework: Python 3.11+, FastAPI
// Giải pháp anonymization + compliant storage
import hashlib
import re
import json
import zlib
import base64
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, asdict
from cryptography.fernet import Fernet
from enum import Enum
class SensitiveField(Enum):
"""Danh sách trường nhạy cảm cần anonymize"""
EMAIL = "email"
PHONE = "phone"
CREDIT_CARD = "credit_card"
SSN = "ssn"
IP_ADDRESS = "ip_address"
API_KEY = "api_key"
PASSWORD = "password"
ADDRESS = "address"
NAME = "name"
DATE_OF_BIRTH = "dob"
@dataclass
class AnonymizedLogEntry:
"""Cấu trúc log sau khi anonymize"""
request_id: str
timestamp: str
endpoint: str
method: str
masked_user_id: str
masked_session: str
anonymized_params: Dict[str, Any]
token_usage: Optional[Dict[str, int]]
response_status: int
processing_time_ms: float
anonymization_hash: str
data_classification: str
class HolySheepAPILogger:
"""
HolySheep API Gateway Logger với anonymization và compliance
Compatible: FastAPI, Flask, Django
"""
def __init__(
self,
api_key: str,
encryption_key: Optional[str] = None,
retention_days: int = 30,
anonymize_fields: Optional[List[SensitiveField]] = None
):
"""
Khởi tạo Logger
Args:
api_key: HolySheep API Key
encryption_key: Fernet encryption key cho log storage
retention_days: Số ngày lưu trữ theo compliance
anonymize_fields: Danh sách trường cần anonymize
"""
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.encryption_key = encryption_key or Fernet.generate_key()
self.fernet = Fernet(self.encryption_key)
self.retention_days = retention_days
self.anonymize_fields = anonymize_fields or list(SensitiveField)
# Regex patterns cho PII detection
self._patterns = {
SensitiveField.EMAIL: r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
SensitiveField.PHONE: r'\b\d{10,15}\b',
SensitiveField.CREDIT_CARD: r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b',
SensitiveField.SSN: r'\b\d{3}-\d{2}-\d{4}\b',
SensitiveField.IP_ADDRESS: r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b',
}
def anonymize_value(self, value: str, field_type: SensitiveField) -> str:
"""
Anonymize giá trị theo type
Support: Hash, Mask, Tokenize
"""
if not value or not isinstance(value, str):
return "[REDACTED]"
if field_type == SensitiveField.EMAIL:
# Giữ 2 ký tự đầu và domain
parts = value.split('@')
if len(parts) == 2:
prefix = parts[0][:2] + '***'
return f"{prefix}@{parts[1]}"
return self._hash_value(value)
elif field_type == SensitiveField.PHONE:
# Chỉ giữ 4 số cuối
digits = re.sub(r'\D', '', value)
return f"****{digits[-4:]}"
elif field_type == SensitiveField.CREDIT_CARD:
# Giữ 4 số cuối
digits = re.sub(r'\D', '', value)
return f"****-****-****-{digits[-4:]}"
elif field_type in [SensitiveField.SSN, SensitiveField.API_KEY, SensitiveField.PASSWORD]:
# Hash hoàn toàn
return self._hash_value(value)
elif field_type == SensitiveField.IP_ADDRESS:
# Keep first 2 octets
parts = value.split('.')
if len(parts) >= 2:
return f"{parts[0]}.{parts[1]}.xxx.xxx"
return self._hash_value(value)
elif field_type == SensitiveField.NAME:
# Giữ chữ cái đầu
if len(value) > 0:
return f"{value[0]}***"
return "[REDACTED]"
else:
return self._hash_value(value)
def _hash_value(self, value: str) -> str:
"""Tạo SHA-256 hash với salt"""
salt = "holy_sheep_compliance_2026"
salted = f"{salt}{value}".encode('utf-8')
return hashlib.sha256(salted).hexdigest()[:16]
def detect_pii_in_text(self, text: str) -> Dict[str, List[str]]:
"""
Detect PII trong text tự do
Dùng cho prompt/response logging
"""
detected = {}
for field_type, pattern in self._patterns.items():
matches = re.findall(pattern, text)
if matches:
detected[field_type.value] = matches
return detected
def anonymize_request_params(
self,
params: Dict[str, Any],
user_context: Optional[Dict] = None
) -> Dict[str, Any]:
"""
Anonymize toàn bộ request parameters
Args:
params: Dict chứa request parameters
user_context: Context của user (để generate consistent mask)
"""
anonymized = {}
for key, value in params.items():
key_lower = key.lower()
# Detect field type từ key name
detected_type = self._detect_field_type(key_lower)
if detected_type:
anonymized[key] = self.anonymize_value(str(value), detected_type)
elif isinstance(value, dict):
anonymized[key] = self.anonymize_request_params(value, user_context)
elif isinstance(value, list):
anonymized[key] = [
self.anonymize_request_params(item, user_context)
if isinstance(item, dict) else str(item)
for item in value
]
else:
# Check nếu value chứa PII
pii_found = self.detect_pii_in_text(str(value))
if pii_found:
anonymized[key] = "[PII_DETECTED_AND_REDACTED]"
else:
anonymized[key] = value
return anonymized
def _detect_field_type(self, key: str) -> Optional[SensitiveField]:
"""Detect field type từ key name"""
mapping = {
'email': SensitiveField.EMAIL,
'phone': SensitiveField.PHONE,
'mobile': SensitiveField.PHONE,
'credit_card': SensitiveField.CREDIT_CARD,
'cc_number': SensitiveField.CREDIT_CARD,
'card_number': SensitiveField.CREDIT_CARD,
'ssn': SensitiveField.SSN,
'social_security': SensitiveField.SSN,
'ip': SensitiveField.IP_ADDRESS,
'ip_address': SensitiveField.IP_ADDRESS,
'client_ip': SensitiveField.IP_ADDRESS,
'api_key': SensitiveField.API_KEY,
'apikey': SensitiveField.API_KEY,
'password': SensitiveField.PASSWORD,
'passwd': SensitiveField.PASSWORD,
'secret': SensitiveField.PASSWORD,
'name': SensitiveField.NAME,
'full_name': SensitiveField.NAME,
'first_name': SensitiveField.NAME,
'last_name': SensitiveField.NAME,
'address': SensitiveField.ADDRESS,
'street': SensitiveField.ADDRESS,
'dob': SensitiveField.DATE_OF_BIRTH,
'birthday': SensitiveField.DATE_OF_BIRTH,
'birth_date': SensitiveField.DATE_OF_BIRTH,
}
return mapping.get(key)
def mask_user_id(self, user_id: str) -> str:
"""Tạo consistent mask cho user_id (same input = same output)"""
return f"usr_{self._hash_value(user_id)}"
def create_compliant_log(
self,
request_id: str,
endpoint: str,
method: str,
user_id: str,
params: Dict[str, Any],
response: Dict[str, Any],
processing_time_ms: float,
token_usage: Optional[Dict[str, int]] = None
) -> AnonymizedLogEntry:
"""
Tạo log entry tuân thủ compliance
Returns:
AnonymizedLogEntry đã được anonymize
"""
# Anonymize params
anonymized_params = self.anonymize_request_params(params)
# Mask user info
masked_user_id = self.mask_user_id(user_id)
masked_session = f"sess_{self._hash_value(request_id)}"
# Classify data
classification = self._classify_log(anonymized_params)
# Generate verification hash
log_data = f"{request_id}{endpoint}{masked_user_id}{processing_time_ms}"
anon_hash = hashlib.sha256(log_data.encode()).hexdigest()[:24]
return AnonymizedLogEntry(
request_id=request_id,
timestamp=datetime.utcnow().isoformat() + "Z",
endpoint=endpoint,
method=method,
masked_user_id=masked_user_id,
masked_session=masked_session,
anonymized_params=anonymized_params,
token_usage=token_usage,
response_status=response.get('status_code', 200),
processing_time_ms=processing_time_ms,
anonymization_hash=anon_hash,
data_classification=classification
)
def _classify_log(self, params: Dict[str, Any]) -> str:
"""Phân loại log theo mức độ nhạy cảm"""
sensitive_count = sum(
1 for v in params.values()
if isinstance(v, str) and '***' in v
)
if sensitive_count > 3:
return "HIGH"
elif sensitive_count > 0:
return "MEDIUM"
return "LOW"
def encrypt_for_storage(self, log_entry: AnonymizedLogEntry) -> bytes:
"""Mã hóa log trước khi lưu trữ"""
json_data = json.dumps(asdict(log_entry))
return self.fernet.encrypt(json_data.encode())
def batch_encrypt_logs(self, logs: List[AnonymizedLogEntry]) -> bytes:
"""Batch encrypt nhiều logs"""
batch = {"logs": [asdict(log) for log in logs], "batch_id": self._hash_value(str(datetime.now()))}
json_data = json.dumps(batch)
return self.fernet.encrypt(json_data.encode())
============== USAGE EXAMPLE ==============
Khởi tạo logger
logger = HolySheepAPILogger(
api_key="YOUR_HOLYSHEEP_API_KEY",
retention_days=30,
anonymize_fields=[field for field in SensitiveField]
)
Example request
sample_request = {
"user_id": "user_123456",
"email": "[email protected]",
"phone": "0909123456",
"message": "Tôi muốn tìm hiểu về sản phẩm AI",
"api_key": "sk-holysheep-live-xxxxxxxxxxxx",
"ip_address": "192.168.1.100"
}
Create compliant log
log_entry = logger.create_compliant_log(
request_id="req_abc123xyz",
endpoint="/v1/chat/completions",
method="POST",
user_id="user_123456",
params=sample_request,
response={"status_code": 200, "id": "chatcmpl_123"},
processing_time_ms=145.32,
token_usage={"prompt_tokens": 150, "completion_tokens": 89}
)
Encrypt for storage
encrypted_log = logger.encrypt_for_storage(log_entry)
print(f"Log đã được anonymize và mã hóa: {len(encrypted_log)} bytes")
print(f"Hash xác thực: {log_entry.anonymization_hash}")
print(f"Classification: {log_entry.data_classification}")
2. FastAPI Integration với Auto-Logging
// api_gateway_with_logging.py
// FastAPI integration với automatic anonymization
from fastapi import FastAPI, Request, Response
from fastapi.middleware.base import BaseHTTPMiddleware
from starlette.datastructures import Headers
import time
import uuid
import json
from typing import Callable
from holy_sheep_logger import HolySheepAPILogger, AnonymizedLogEntry
Initialize HolySheep API Client
class HolySheepAIClient:
"""
HolySheep AI API Client với built-in logging
base_url: https://api.holysheep.ai/v1
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.logger = HolySheepAPILogger(
api_key=api_key,
retention_days=90, # Compliance: 90 ngày theo GDPR
anonymize_fields=[field for field in SensitiveField]
)
async def chat_completions(
self,
messages: list,
model: str = "deepseek-v3",
temperature: float = 0.7,
max_tokens: int = 1000,
user_id: str = "anonymous",
stream: bool = False
) -> dict:
"""
Gọi HolySheep Chat Completions API với automatic logging
Args:
messages: List of message objects
model: Model name (deepseek-v3, gpt-4.1, claude-sonnet-4.5, etc.)
user_id: User identifier cho logging
Returns:
API response dict
"""
import httpx
request_id = f"req_{uuid.uuid4().hex[:12]}"
start_time = time.time()
# Build request payload
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": stream,
"user": user_id # Sẽ được anonymize trong log
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": request_id
}
try:
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers
)
processing_time_ms = (time.time() - start_time) * 1000
# Parse response
response_data = response.json()
# Extract token usage
token_usage = None
if 'usage' in response_data:
token_usage = {
'prompt_tokens': response_data['usage'].get('prompt_tokens', 0),
'completion_tokens': response_data['usage'].get('completion_tokens', 0),
'total_tokens': response_data['usage'].get('total_tokens', 0)
}
# Auto-create compliant log
log_entry = self.logger.create_compliant_log(
request_id=request_id,
endpoint="/v1/chat/completions",
method="POST",
user_id=user_id,
params={
"model": model,
"message_count": len(messages),
"temperature": temperature,
"max_tokens": max_tokens
},
response={
"status_code": response.status_code,
"id": response_data.get('id')
},
processing_time_ms=processing_time_ms,
token_usage=token_usage
)
# Store encrypted log
await self._store_compliant_log(log_entry)
return response_data
except httpx.HTTPStatusError as e:
# Log failed request
log_entry = self.logger.create_compliant_log(
request_id=request_id,
endpoint="/v1/chat/completions",
method="POST",
user_id=user_id,
params={"model": model, "message_count": len(messages)},
response={"status_code": e.response.status_code, "error": str(e)},
processing_time_ms=(time.time() - start_time) * 1000
)
await self._store_compliant_log(log_entry)
raise
async def embeddings(
self,
input_text: str,
model: str = "text-embedding-v3",
user_id: str = "anonymous"
) -> dict:
"""Gọi HolySheep Embeddings API với logging"""
import httpx
request_id = f"req_{uuid.uuid4().hex[:12]}"
start_time = time.time()
payload = {
"model": model,
"input": input_text,
"user": user_id
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": request_id
}
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/embeddings",
json=payload,
headers=headers
)
processing_time_ms = (time.time() - start_time) * 1000
response_data = response.json()
# Log request
log_entry = self.logger.create_compliant_log(
request_id=request_id,
endpoint="/v1/embeddings",
method="POST",
user_id=user_id,
params={"model": model, "input_length": len(input_text)},
response={"status_code": response.status_code},
processing_time_ms=processing_time_ms,
token_usage=response_data.get('usage')
)
await self._store_compliant_log(log_entry)
return response_data
async def _store_compliant_log(self, log_entry: AnonymizedLogEntry):
"""
Lưu trữ log đã được anonymize và mã hóa
Support multiple backends: PostgreSQL, MongoDB, S3, etc.
"""
# Encrypt log
encrypted_data = self.logger.encrypt_for_storage(log_entry)
# Compress
import zlib
compressed = zlib.compress(encrypted_data)
# Store to compliance storage
# Implementation depends on your storage backend
storage_key = f"logs/{log_entry.timestamp[:10]}/{log_entry.request_id}.enc"
# Example: Save to file (replace with your storage)
# await self._save_to_storage(storage_key, compressed)
print(f"✅ Log stored compliantly: {storage_key}")
print(f" Classification: {log_entry.data_classification}")
print(f" Masked User: {log_entry.masked_user_id}")
============== FastAPI APP EXAMPLE ==============
app = FastAPI(title="HolySheep AI Gateway with Compliance Logging")
Initialize client
hs_client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
class ComplianceLoggingMiddleware(BaseHTTPMiddleware):
"""Middleware tự động log mọi request"""
async def dispatch(self, request: Request, call_next: Callable):
start_time = time.time()
# Process request
response = await call_next(request)
# Log request details (đã được anonymize tự động)
process_time = (time.time() - start_time) * 1000
# Extract user ID (sẽ được mask trong log)
user_id = request.headers.get("X-User-ID", "anonymous")
# Create anonymized log
log_entry = hs_client.logger.create_compliant_log(
request_id=request.headers.get("X-Request-ID", str(uuid.uuid4())),
endpoint=str(request.url.path),
method=request.method,
user_id=user_id,
params=dict(request.query_params) if request.query_params else {},
response={"status_code": response.status_code},
processing_time_ms=process_time
)
# Store log
await hs_client._store_compliant_log(log_entry)
return response
app.add_middleware(ComplianceLoggingMiddleware)
@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
"""Chat Completions endpoint"""
body = await request.json()
response = await hs_client.chat_completions(
messages=body.get("messages", []),
model=body.get("model", "deepseek-v3"),
temperature=body.get("temperature", 0.7),
max_tokens=body.get("max_tokens", 1000),
user_id=request.headers.get("X-User-ID", "anonymous")
)
return response
@app.post("/v1/embeddings")
async def embeddings(request: Request):
"""Embeddings endpoint"""
body = await request.json()
response = await hs_client.embeddings(
input_text=body.get("input", ""),
model=body.get("model", "text-embedding-v3"),
user_id=request.headers.get("X-User-ID", "anonymous")
)
return response
Run: uvicorn api_gateway_with_logging:app --host 0.0.0.0 --port 8000
3. Compliance Storage với PostgreSQL/MongoDB
// compliance_storage.py
// Backend storage với retention policy và audit trail
import asyncio
from datetime import datetime, timedelta
from typing import List, Optional
import json
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, String, Integer, DateTime, Text, Index
from sqlalchemy.dialects.postgresql import JSONB
Base = declarative_base()
class CompliantLog(Base):
"""Bảng lưu trữ log tuân thủ compliance"""
__tablename__ = 'compliant_api_logs'
id = Column(Integer, primary_key=True, autoincrement=True)
request_id = Column(String(64), unique=True, nullable=False, index=True)
timestamp = Column(DateTime, nullable=False, index=True)
# Anonymized identifiers
masked_user_id = Column(String(64), index=True)
masked_session = Column(String(64), index=True)
# Request details (đã anonymize)
endpoint = Column(String(255), nullable=False)
method = Column(String(10), nullable=False)
anonymized_params = Column(JSONB) # Đã được anonymize hoàn toàn
# Response metadata
response_status = Column(Integer)
processing_time_ms = Column(Integer)
token_usage = Column(JSONB) # {prompt_tokens, completion_tokens, total_tokens}
# Compliance fields
anonymization_hash = Column(String(64), nullable=False)
data_classification = Column(String(20), index=True) # LOW, MEDIUM, HIGH
# Encryption & Storage
encrypted_payload = Column(Text) # Full payload encrypted
compressed_size = Column(Integer)
original_size = Column(Integer)
# Audit
created_at = Column(DateTime, default=datetime.utcnow)
retention_until = Column(DateTime, index=True)
deleted_at = Column(DateTime, nullable=True)
# Indexes cho query efficiency
__table_args__ = (
Index('idx_timestamp_classification', 'timestamp', 'data_classification'),
Index('idx_retention', 'retention_until', 'deleted_at'),
)
class ComplianceStorageManager:
"""
Quản lý lưu trữ log tuân thủ compliance
Features: Auto-retention, Encryption, Audit Trail, Data Recovery
"""
def __init__(
self,
database_url: str,
encryption_key: bytes,
default_retention_days: int = 90
):
"""
Khởi tạo Storage Manager
Args:
database_url: PostgreSQL connection string
encryption_key: Fernet encryption key
default_retention_days: Số ngày lưu trữ mặc định (GDPR compliant: 90 days)
"""
self.engine = create_async_engine(database_url, echo=False)
self.session_factory = AsyncSession(bind=self.engine)
self.encryption_key = encryption_key
self.default_retention_days = default_retention_days
async def initialize(self):
"""Khởi tạo database schema"""
async with self.engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async def store_log(
self,
log_entry: 'AnonymizedLogEntry',
raw_payload: dict
) -> str:
"""
Lưu trữ log với encryption
Args:
log_entry: Log entry đã được anonymize
raw_payload: Payload gốc để encrypt
Returns:
request_id của log đã lưu
"""
from cryptography.fernet import Fernet
import zlib
fernet = Fernet(self.encryption_key)
# Encrypt full payload
json_payload = json.dumps(raw_payload)
encrypted = fernet.encrypt(json_payload.encode())
# Compress
compressed = zlib.compress(encrypted)
# Calculate sizes
original_size = len(json_payload.encode())
compressed_size = len(compressed)
# Retention policy
retention_until = datetime.utcnow() + timedelta(days=self.default_retention_days)
# Create record
async with self.session_factory() as session:
record = CompliantLog(
request_id=log_entry.request_id,
timestamp=datetime.fromisoformat(log_entry.timestamp.replace('Z', '+00:00')),
masked_user_id=log_entry.masked_user_id,
masked_session=log_entry.masked_session,
endpoint=log_entry.endpoint,
method=log_entry.method,
anonymized_params=log_entry.anonymized_params,
response_status=log_entry.response_status,
processing_time_ms=int(log_entry.processing_time_ms),
token_usage=log_entry.token_usage,
anonymization_hash=log_entry.anonymization_hash,
data_classification=log_entry.data_classification,
encrypted_payload=compressed.decode('latin-1'),
compressed_size=compressed_size,
original_size=original_size,
retention_until=retention_until
)
session.add(record)
await session.commit()
return log_entry.request_id
async def query_logs(
self,
start_date: datetime,
end_date: datetime,
classification: Optional[str] = None,
limit: int = 1000
) -> List[CompliantLog]:
"""
Query logs với filters
Args:
start_date: Ngày bắt đầu
end_date: Ngày kết thúc
classification: Filter theo data classification
limit: Số lượng kết quả tối đa
Returns:
List of anonymized log entries
"""
async with self.session_factory() as session:
from sqlalchemy import select, and_
query = select(CompliantLog).where(
and_(
CompliantLog.timestamp >= start_date,
CompliantLog.timestamp <= end_date,
CompliantLog.deleted_at.is_(None)
)
)
if classification:
query = query.where(CompliantLog.data_classification == classification)
query = query.order_by(CompliantLog.timestamp.desc()).limit(limit)
result = await session.execute(query)
return result.scalars().all()
async def get_analytics_summary(
self,
start_date: datetime,
end_date: datetime
) -> dict:
"""
T