Tôi vẫn nhớ rõ cái ngày tháng 6 năm ngoái khi đội ngũ kỹ thuật của một startup e-commerce quy mô 50 nhân viên gặp phải vấn đề nghiêm trọng: không ai có thể trả lời được câu hỏi đơn giản nhất từ phía kiểm toán - "AI đã xử lý bao nhiêu request của khách hàng trong tháng vừa qua, chi phí là bao nhiêu, và có vi phạm GDPR không?". Đó là khoảnh khắc tôi bắt đầu nghiên cứu sâu về audit trail cho AI API usage compliance.

Bài viết này sẽ chia sẻ toàn bộ kiến thức và kinh nghiệm thực chiến của tôi trong việc xây dựng hệ thống audit trail hoàn chỉnh, từ thiết kế database đến implementation với HolySheep AI.

Tại Sao Audit Trail Quan Trọng Với AI API?

Khi doanh nghiệp sử dụng AI API cho customer service, RAG system, hoặc bất kỳ use case nào liên quan đến dữ liệu khách hàng, việc maintain audit trail không chỉ là best practice mà còn là yêu cầu pháp lý. GDPR, SOC2, ISO 27001 đều yêu cầu khả năng truy vết mọi thao tác xử lý dữ liệu.

Với HolySheheep AI, việc implement audit trail càng quan trọng hơn vì tỷ giá chỉ ¥1=$1 giúp chi phí giảm 85%+ so với các provider khác, đồng nghĩa với việc volume request sẽ cao hơn rất nhiều và cần tracking chính xác hơn.

Kiến Trúc Hệ Thống Audit Trail

Để xây dựng một audit trail hoàn chỉnh, tôi đề xuất kiến trúc 4 tầng:

Implementation Chi Tiết

1. Database Schema Cho Audit Trail

-- Tạo bảng chính cho audit trail
CREATE TABLE ai_api_audit_logs (
    id BIGSERIAL PRIMARY KEY,
    request_id UUID NOT NULL DEFAULT gen_random_uuid(),
    user_id VARCHAR(255),
    session_id VARCHAR(255),
    request_timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    
    -- Request details
    model_name VARCHAR(100) NOT NULL,
    endpoint VARCHAR(255) NOT NULL,
    request_tokens INTEGER,
    input_tokens INTEGER,
    output_tokens INTEGER,
    input_characters INTEGER,
    output_characters INTEGER,
    
    -- Cost tracking
    cost_usd DECIMAL(10, 6) NOT NULL,
    cost_cny DECIMAL(10, 6) NOT NULL,
    
    -- Response metadata
    response_time_ms INTEGER NOT NULL,
    http_status_code INTEGER,
    model_response_id VARCHAR(255),
    
    -- Content tracking
    prompt_hash VARCHAR(64), -- SHA256 hash of prompt for deduplication
    prompt_category VARCHAR(50), -- customer_service, rag_query, etc.
    contains_pii BOOLEAN DEFAULT FALSE,
    pii_types JSONB, -- ["email", "phone", "credit_card"]
    
    -- Compliance fields
    data_classification VARCHAR(50), -- public, internal, confidential, restricted
    consent_obtained BOOLEAN DEFAULT FALSE,
    retention_days INTEGER DEFAULT 365,
    legal_basis VARCHAR(100), -- GDPR article basis
    
    -- Error tracking
    error_code VARCHAR(50),
    error_message TEXT,
    retry_count INTEGER DEFAULT 0,
    
    -- Metadata
    client_ip INET,
    user_agent TEXT,
    metadata JSONB,
    
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Indexes cho performance
CREATE INDEX idx_audit_timestamp ON ai_api_audit_logs (request_timestamp DESC);
CREATE INDEX idx_audit_user_id ON ai_api_audit_logs (user_id);
CREATE INDEX idx_audit_session ON ai_api_audit_logs (session_id);
CREATE INDEX idx_audit_cost ON ai_api_audit_logs (cost_usd);
CREATE INDEX idx_audit_model ON ai_api_audit_logs (model_name);

-- Partition theo tháng cho scalability
CREATE TABLE ai_api_audit_logs_2025_06 
PARTITION OF ai_api_audit_logs
FOR VALUES FROM ('2025-06-01') TO ('2025-07-01');

2. Python Client Wrapper Với Audit Logging

import hashlib
import json
import time
import re
from datetime import datetime
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, asdict
import httpx

@dataclass
class AuditLogEntry:
    request_id: str
    user_id: str
    session_id: str
    model_name: str
    endpoint: str
    request_tokens: int
    input_tokens: int
    output_tokens: int
    cost_usd: float
    cost_cny: float
    response_time_ms: int
    http_status_code: int
    prompt_hash: str
    prompt_category: str
    contains_pii: bool
    pii_types: List[str]
    data_classification: str
    error_code: Optional[str]
    error_message: Optional[str]

class HolySheepAIWithAudit:
    """
    HolySheep AI client với built-in audit trail.
    Documentation: https://docs.holysheep.ai
    """
    
    # Pricing structure (updated 2026)
    PRICING = {
        "gpt-4.1": {"input": 2.0, "output": 8.0},      # $2/$8 per MTok
        "claude-sonnet-4.5": {"input": 3.0, "output": 15.0},  # $3/$15 per MTok
        "gemini-2.5-flash": {"input": 0.10, "output": 2.50},  # $0.10/$2.50 per MTok
        "deepseek-v3.2": {"input": 0.07, "output": 0.42},   # $0.07/$0.42 per MTok
    }
    
    CNY_TO_USD_RATE = 1.0  # ¥1 = $1
    
    PII_PATTERNS = {
        "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
        "phone": r'\b\d{10,15}\b',
        "credit_card": r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b',
        "ssn": r'\b\d{3}-\d{2}-\d{4}\b',
    }
    
    def __init__(
        self, 
        api_key: str,
        audit_callback=None,  # Function to save audit logs
        default_user_id: str = "anonymous",
        default_session_id: str = None
    ):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.audit_callback = audit_callback
        self.default_user_id = default_user_id
        self.default_session_id = default_session_id or str(int(time.time()))
    
    def _detect_pii(self, text: str) -> tuple[bool, List[str]]:
        """Detect PII in text"""
        found_types = []
        for pii_type, pattern in self.PII_PATTERNS.items():
            if re.search(pattern, text):
                found_types.append(pii_type)
        return len(found_types) > 0, found_types
    
    def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> tuple[float, float]:
        """Calculate cost in USD and CNY"""
        pricing = self.PRICING.get(model, {"input": 0, "output": 0})
        input_cost = (input_tokens / 1_000_000) * pricing["input"]
        output_cost = (output_tokens / 1_000_000) * pricing["output"]
        cost_usd = input_cost + output_cost
        return cost_usd, cost_usd * self.CNY_TO_USD_RATE
    
    def _hash_prompt(self, prompt: str) -> str:
        """Create SHA256 hash of prompt for deduplication"""
        return hashlib.sha256(prompt.encode()).hexdigest()
    
    async def chat_completion(
        self,
        messages: List[Dict[str, str]],
        model: str = "deepseek-v3.2",
        user_id: str = None,
        session_id: str = None,
        prompt_category: str = "general",
        data_classification: str = "internal",
        **kwargs
    ) -> Dict[str, Any]:
        """
        Gửi request đến HolySheep AI với automatic audit logging.
        """
        import uuid
        
        request_id = str(uuid.uuid4())
        user_id = user_id or self.default_user_id
        session_id = session_id or self.default_session_id
        
        # Combine messages for analysis
        full_prompt = "\n".join([m.get("content", "") for m in messages])
        contains_pii, pii_types = self._detect_pii(full_prompt)
        prompt_hash = self._hash_prompt(full_prompt)
        
        start_time = time.time()
        
        try:
            async with httpx.AsyncClient(timeout=60.0) as client:
                response = await client.post(
                    f"{self.base_url}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {self.api_key}",
                        "Content-Type": "application/json"
                    },
                    json={
                        "model": model,
                        "messages": messages,
                        **kwargs
                    }
                )
                
                response_time_ms = int((time.time() - start_time) * 1000)
                
                if response.status_code == 200:
                    result = response.json()
                    usage = result.get("usage", {})
                    input_tokens = usage.get("prompt_tokens", 0)
                    output_tokens = usage.get("completion_tokens", 0)
                    cost_usd, cost_cny = self._calculate_cost(model, input_tokens, output_tokens)
                    
                    audit_entry = AuditLogEntry(
                        request_id=request_id,
                        user_id=user_id,
                        session_id=session_id,
                        model_name=model,
                        endpoint=f"{self.base_url}/chat/completions",
                        request_tokens=input_tokens + output_tokens,
                        input_tokens=input_tokens,
                        output_tokens=output_tokens,
                        cost_usd=cost_usd,
                        cost_cny=cost_cny,
                        response_time_ms=response_time_ms,
                        http_status_code=200,
                        prompt_hash=prompt_hash,
                        prompt_category=prompt_category,
                        contains_pii=contains_pii,
                        pii_types=pii_types,
                        data_classification=data_classification,
                        error_code=None,
                        error_message=None
                    )
                    
                    # Save audit log
                    if self.audit_callback:
                        await self.audit_callback(audit_entry)
                    
                    return {
                        "success": True,
                        "data": result,
                        "audit": asdict(audit_entry)
                    }
                else:
                    raise Exception(f"API Error: {response.status_code} - {response.text}")
                    
        except Exception as e:
            response_time_ms = int((time.time() - start_time) * 1000)
            error_entry = AuditLogEntry(
                request_id=request_id,
                user_id=user_id,
                session_id=session_id,
                model_name=model,
                endpoint=f"{self.base_url}/chat/completions",
                request_tokens=0,
                input_tokens=0,
                output_tokens=0,
                cost_usd=0,
                cost_cny=0,
                response_time_ms=response_time_ms,
                http_status_code=500,
                prompt_hash=prompt_hash,
                prompt_category=prompt_category,
                contains_pii=contains_pii,
                pii_types=pii_types,
                data_classification=data_classification,
                error_code="API_ERROR",
                error_message=str(e)
            )
            
            if self.audit_callback:
                await self.audit_callback(error_entry)
            
            return {
                "success": False,
                "error": str(e),
                "audit": asdict(error_entry)
            }

3. Audit Database Service

import asyncpg
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional

class AuditDatabaseService:
    """
    Service để lưu trữ và query audit logs.
    Kết nối với PostgreSQL sử dụng asyncpg.
    """
    
    def __init__(self, database_url: str):
        self.database_url = database_url
        self.pool: Optional[asyncpg.Pool] = None
    
    async def connect(self):
        """Khởi tạo connection pool"""
        self.pool = await asyncpg.create_pool(
            self.database_url,
            min_size=5,
            max_size=20
        )
    
    async def save_audit_log(self, audit_entry: Dict[str, Any]):
        """
        Lưu một audit log entry vào database.
        """
        query = """
        INSERT INTO ai_api_audit_logs (
            request_id, user_id, session_id, request_timestamp,
            model_name, endpoint, request_tokens, input_tokens,
            output_tokens, input_characters, output_characters,
            cost_usd, cost_cny, response_time_ms, http_status_code,
            prompt_hash, prompt_category, contains_pii, pii_types,
            data_classification, error_code, error_message, metadata
        ) VALUES (
            $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15,
            $16, $17, $18, $19, $20, $21, $22, $23
        )
        """
        
        async with self.pool.acquire() as conn:
            await conn.execute(
                query,
                audit_entry.get("request_id"),
                audit_entry.get("user_id"),
                audit_entry.get("session_id"),
                datetime.now(),
                audit_entry.get("model_name"),
                audit_entry.get("endpoint"),
                audit_entry.get("input_tokens", 0) + audit_entry.get("output_tokens", 0),
                audit_entry.get("input_tokens", 0),
                audit_entry.get("output_tokens", 0),
                len(str(audit_entry.get("prompt", ""))),
                0,  # output_characters
                audit_entry.get("cost_usd", 0),
                audit_entry.get("cost_cny", 0),
                audit_entry.get("response_time_ms", 0),
                audit_entry.get("http_status_code", 200),
                audit_entry.get("prompt_hash"),
                audit_entry.get("prompt_category", "general"),
                audit_entry.get("contains_pii", False),
                audit_entry.get("pii_types", []),
                audit_entry.get("data_classification", "internal"),
                audit_entry.get("error_code"),
                audit_entry.get("error_message"),
                audit_entry.get("metadata", {})
            )
    
    async def get_cost_summary(
        self, 
        start_date: datetime,
        end_date: datetime,
        model_name: Optional[str] = None
    ) -> Dict[str, Any]:
        """
        Lấy tổng hợp chi phí theo khoảng thời gian.
        """
        conditions = ["request_timestamp BETWEEN $1 AND $2"]
        params = [start_date, end_date]
        
        if model_name:
            conditions.append("model_name = $3")
            params.append(model_name)
        
        where_clause = " AND ".join(conditions)
        
        query = f"""
        SELECT 
            model_name,
            COUNT(*) as total_requests,
            SUM(input_tokens) as total_input_tokens,
            SUM(output_tokens) as total_output_tokens,
            SUM(cost_usd) as total_cost_usd,
            SUM(cost_cny) as total_cost_cny,
            AVG(response_time_ms) as avg_response_time_ms,
            COUNT(CASE WHEN error_code IS NOT NULL THEN 1 END) as error_count
        FROM ai_api_audit_logs
        WHERE {where_clause}
        GROUP BY model_name
        ORDER BY total_cost_usd DESC
        """
        
        async with self.pool.acquire() as conn:
            rows = await conn.fetch(query, *params)
            
            return {
                "period": {
                    "start": start_date.isoformat(),
                    "end": end_date.isoformat()
                },
                "summary": [
                    {
                        "model": dict(row)["model_name"],
                        "total_requests": dict(row)["total_requests"],
                        "total_input_tokens": dict(row)["total_input_tokens"],
                        "total_output_tokens": dict(row)["total_output_tokens"],
                        "total_cost_usd": float(dict(row)["total_cost_usd"] or 0),
                        "total_cost_cny": float(dict(row)["total_cost_cny"] or 0),
                        "avg_response_time_ms": float(dict(row)["avg_response_time_ms"] or 0),
                        "error_rate": dict(row)["error_count"] / dict(row)["total_requests"] * 100
                    }
                    for row in rows
                ],
                "grand_total_usd": sum(float(r["total_cost_usd"] or 0) for r in [dict(row) for row in rows]),
                "grand_total_cny": sum(float(r["total_cost_cny"] or 0) for r in [dict(row) for row in rows])
            }
    
    async def get_user_activity(
        self,
        user_id: str,
        limit: int = 100
    ) -> List[Dict[str, Any]]:
        """Lấy lịch sử hoạt động của một user"""
        query = """
        SELECT * FROM ai_api_audit_logs
        WHERE user_id = $1
        ORDER BY request_timestamp DESC
        LIMIT $2
        """
        
        async with self.pool.acquire() as conn:
            rows = await conn.fetch(query, user_id, limit)
            return [dict(row) for row in rows]
    
    async def generate_compliance_report(
        self,
        start_date: datetime,
        end_date: datetime
    ) -> Dict[str, Any]:
        """
        Generate compliance report cho audit.
        """
        query = """
        WITH pii_stats AS (
            SELECT 
                COUNT(*) FILTER (WHERE contains_pii = TRUE) as requests_with_pii,
                jsonb_array_length(pii_types) as pii_type_count
            FROM ai_api_audit_logs
            WHERE request_timestamp BETWEEN $1 AND $2
        ),
        error_stats AS (
            SELECT 
                error_code,
                COUNT(*) as count,
                AVG(response_time_ms) as avg_time
            FROM ai_api_audit_logs
            WHERE request_timestamp BETWEEN $1 AND $2
            AND error_code IS NOT NULL
            GROUP BY error_code
        )
        SELECT 
            (SELECT * FROM pii_stats) as pii_stats,
            (SELECT array_agg(row_to_json(e)) FROM error_stats e) as error_breakdown
        """
        
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow(query, start_date, end_date)
            
            return {
                "report_period": {
                    "start": start_date.isoformat(),
                    "end": end_date.isoformat()
                },
                "generated_at": datetime.now().isoformat(),
                "pii_processing": {
                    "total_requests_with_pii": row["pii_stats"]["requests_with_pii"],
                    "pii_types_detected": row["pii_stats"]["pii_type_count"]
                },
                "error_analysis": row["error_breakdown"] or [],
                "compliance_status": "COMPLIANT"  # Add actual validation logic
            }

4. Sử Dụng Trong Thực Tế

import asyncio
from holy_sheep_audit import HolySheepAIWithAudit, AuditDatabaseService

async def main():
    """
    Ví dụ sử dụng HolySheep AI với audit trail.
    """
    # Khởi tạo audit database service
    audit_db = AuditDatabaseService("postgresql://user:pass@localhost/audit_db")
    await audit_db.connect()
    
    # Wrapper function để lưu audit logs
    async def save_audit(audit_entry):
        await audit_db.save_audit_log(audit_entry)
    
    # Khởi tạo HolySheep AI client với audit
    client = HolySheepAIWithAudit(
        api_key="YOUR_HOLYSHEEP_API_KEY",  # Thay bằng API key thực tế
        audit_callback=save_audit,
        default_user_id="customer_12345",
        default_session_id="session_ecommerce_checkout"
    )
    
    # Request 1: Customer service query
    response1 = await client.chat_completion(
        messages=[
            {"role": "system", "content": "Bạn là trợ lý khách hàng thân thiện."},
            {"role": "user", "content": "Tôi muốn hỏi về tình trạng đơn hàng #ORD-2025-7890"}
        ],
        model="deepseek-v3.2",  # Model tiết kiệm 85%+ chi phí
        user_id="customer_12345",
        prompt_category="customer_service",
        data_classification="confidential"
    )
    
    print(f"Response 1: {response1.get('success')}")
    print(f"Cost: ${response1['audit']['cost_usd']:.6f} (¥{response1['audit']['cost_cny']:.6f})")
    print(f"Response time: {response1['audit']['response_time_ms']}ms")
    
    # Request 2: RAG query cho product search
    response2 = await client.chat_completion(
        messages=[
            {"role": "user", "content": "Tìm laptop phù hợp cho lập trình viên, budget 20 triệu"}
        ],
        model="deepseek-v3.2",
        user_id="customer_67890",
        prompt_category="rag_product_search",
        data_classification="public"
    )
    
    # Generate cost summary for the day
    from datetime import datetime, timedelta
    
    summary = await audit_db.get_cost_summary(
        start_date=datetime.now() - timedelta(days=1),
        end_date=datetime.now()
    )
    
    print(f"\n=== Cost Summary ===")
    print(f"Total requests: {sum(s['total_requests'] for s in summary['summary'])}")
    print(f"Total cost: ${summary['grand_total_usd']:.2f} (¥{summary['grand_total_cny']:.2f})")
    
    # Generate compliance report
    compliance = await audit_db.generate_compliance_report(
        start_date=datetime.now() - timedelta(days=30),
        end_date=datetime.now()
    )
    
    print(f"\n=== Compliance Report ===")
    print(f"Requests with PII: {compliance['pii_processing']['total_requests_with_pii']}")

if __name__ == "__main__":
    asyncio.run(main())

Dashboard Monitoring Thực Tế

Với hệ thống audit trail đã xây dựng, tôi có thể tạo dashboard monitoring với các metrics quan trọng:

Lỗi Thường Gặp Và Cách Khắc Phục

1. Lỗi "Authentication Error" - API Key Không Hợp Lệ

# ❌ Sai: Sử dụng base_url không đúng
client = HolySheepAIWithAudit(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.openai.com/v1"  # SAI - Không dùng OpenAI endpoint
)


✅ Đúng: Sử dụng HolySheep AI endpoint

client = HolySheepAIWithAudit(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    # base_url mặc định đã là "https://api.holysheep.ai/v1"
)


Hoặc kiểm tra API key trước khi sử dụng

import httpx

async def verify_api_key(api_key: str) -> bool:
    """Verify HolySheep API key validity"""
    async with httpx.AsyncClient() as client:
        try:
            response = await client.post(
                "https://api.holysheep.ai/v1/chat/completions",
                headers={
                    "Authorization": f"Bearer {api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "deepseek-v3.2",
                    "messages": [{"role": "user", "content": "test"}]
                },
                timeout=10.0
            )
            return response.status_code == 200
        except Exception:
            return False

2. Lỗi "Cost Calculation Incorrect" - Tính Sai Chi Phí

# ❌ Sai: Hardcode giá không đúng
cost = (tokens / 1000) * 0.002  # Giá cũ không còn áp dụng


✅ Đúng: Sử dụng bảng giá chính xác từ HolySheep AI

PRICING_2026 = {
    "deepseek-v3.2": {"input": 0.07, "output": 0.42},      # $0.07/$0.42 per MTok
    "gemini-2.5-flash": {"input": 0.10, "output": 2.50},  # $0.10/$2.50 per MTok
    "gpt-4.1": {"input": 2.0, "output": 8.0},            # $2.00/$8.00 per MTok
    "claude-sonnet-4.5": {"input": 3.0, "output": 15.0},  # $3.00/$15.00 per MTok
}

def calculate_cost_accurate(model: str, input_tokens: int, output_tokens: int) -> dict:
    """
    Tính chi phí chính xác theo bảng giá 2026.
    Tỷ giá: ¥1 = $1 (tiết kiệm 85%+)
    """
    pricing = PRICING_2026.get(model, {"input": 0, "output": 0})
    
    input_cost = (input_tokens / 1_000_000) * pricing["input"]
    output_cost = (output_tokens / 1_000_000) * pricing["output"]
    
    total_usd = input_cost + output_cost
    
    return {
        "input_cost_usd": round(input_cost, 6),
        "output_cost_usd": round(output_cost, 6),
        "total_cost_usd": round(total_usd, 6),
        "total_cost_cny": round(total_usd, 6),  # ¥1 = $1
        "currency": "USD/CNY"
    }


Ví dụ: 1 triệu input tokens + 500K output tokens với DeepSeek V3.2

cost = calculate_cost_accurate("deepseek-v3.2", 1_000_000, 500_000)
print(f"Total: ${cost['total_cost_usd']}")  # Output: ~0.28

3. Lỗi "Response Time Timeout" - Request Timeout

# ❌ Sai: Timeout quá ngắn hoặc không set timeout
async with httpx.AsyncClient() as client:
    response = await client.post(url, json=data)  # Không có timeout


✅ Đúng: Set timeout phù hợp với HolySheep AI (<50ms latency)

async with httpx.AsyncClient(timeout=60.0) as client:
    # Retry logic với exponential backoff
    max_retries = 3
    retry_delay = 1.0
    
    for attempt in range(max_retries):
        try:
            response = await client.post(
                "https://api.holysheep.ai/v1/chat/completions",
                headers={
                    "Authorization": f"Bearer {api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "deepseek-v3.2",
                    "messages": messages,
                    "max_tokens": 1000
                },
                timeout=30.0  # 30 seconds timeout
            )
            
            # Check response time từ headers
            response_time = response.headers.get("x-response-time-ms")
            
            if response_time and int(response_time) > 1000:
                print(f"Warning: Slow response {response_time}ms")
            
            return response.json()
            
        except httpx.TimeoutException as e:
            if attempt < max_retries - 1:
                await asyncio.sleep(retry_delay * (2 ** attempt))
                continue
            raise Exception(f"Request timeout after {max_retries} attempts")

4. Lỗi "PII Detection Missed" - Không Phát Hiện PII

# ❌ Sai: Chỉ check email
def detect_pii_simple(text):
    if "@" in text:
        return True
    return False


✅ Đúng: Multi-pattern PII detection

import re

class PIIValidator:
    """Advanced PII detection cho GDPR compliance"""
    
    PATTERNS = {
        "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
        "phone_vn": r'\b(0\d{9,10}|84\d{9,10})\b',
        "phone_global": r'\b\+?1?\d{9,15}\b',
        "credit_card": r'\b(?:\d{4}[-\s]?){3}\d{4}\b',
        "ssn": r'\b\d{3}-\d{2}-\d{4}\b',
        "passport": r'\b[A-Z]{1,2}\d{6,9}\b',
        "vietnam_id": r'\b\d{9,12}\b',  # CCCD Việt Nam
        "ip_address": r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b',
        "dob": r'\b\d{2}/\d{2}/\d{4}\b',
    }
    
    def scan(self, text: str) -> dict:
        """Scan text for all PII types"""
        results = {
            "has_pii": False,
            "pii_types": [],
            "locations": []
        }
        
        for pii_type, pattern in self.PATTERNS.items():
            matches = re.finditer(pattern, text, re.IGNORECASE)
            for match in matches:
                results["has_pii"] = True
                results["pii_types"].append(pii_type)
                results["locations"].append({
                    "type": pii_type,
                    "start": match.start(),
                    "end": match.end(),
                    "value": match.group()[:3] + "***"  # Mask for logging
                })
        
        return results
    
    def anonymize(self, text: str) -> str:
        """Replace PII with [REDACTED]"""
        anonymized = text
        for pii_type, pattern in self.PATTERNS.items():
            anonymized = re.sub(pattern, f"[REDACTED-{pii_type}]", anonymized)
        return anonymized


Sử dụng

validator = PIIValidator()
text = "Email của tôi là [email protected], SDT: 0912345678"
result = validator.scan(text)
print(result["has_pii"])  # True
print(result["pii_types"])  # ['email', 'phone_vn']

Kết Quả Đạt Được

Sau khi implement hệ thống audit trail hoàn chỉnh, đội ngũ e-commerce của tôi đã đạt được: