# feature_flags.py
from dataclasses import dataclass
from typing import Optional

@dataclass
class FeatureFlag:
    name: str
    enabled: bool
    rollout_percentage: float = 100.0
    target_users: list[str] = None

class FeatureFlagManager:
    def __init__(self):
        self.flags: dict[str, FeatureFlag] = {}
    
    def register(self, flag: FeatureFlag):
        self.flags[flag.name] = flag
    
    def is_enabled(self, flag_name: str, user_id: Optional[str] = None) -> bool:
        flag = self.flags.get(flag_name)
        if not flag or not flag.enabled:
            return False
        
        # 目标用户列表
        if flag.target_users and user_id:
            return user_id in flag.target_users
        
        # 百分比灰度
        if flag.rollout_percentage < 100 and user_id:
            hash_value = hash(f"{flag_name}:{user_id}") % 100
            return hash_value < flag.rollout_percentage
        
        return True

# 使用
manager = FeatureFlagManager()
manager.register(FeatureFlag(
    name="new_checkout_flow",
    enabled=True,
    rollout_percentage=10  # 10% 用户
))

@app.get("/checkout")
async def checkout(user: User = Depends(get_current_user)):
    if manager.is_enabled("new_checkout_flow", user.id):
        return new_checkout_page()
    else:
        return old_checkout_page()

import json
from datetime import datetime

class DatabaseFeatureFlags:
    def __init__(self, db):
        self.db = db
        self.cache = {}
        self.cache_ttl = 60  # 秒
    
    async def get_flag(self, name: str) -> Optional[dict]:
        # 检查缓存
        if name in self.cache:
            cached = self.cache[name]
            if datetime.utcnow().timestamp() - cached["time"] < self.cache_ttl:
                return cached["data"]
        
        # 从数据库获取
        result = await self.db.fetch_one(
            "SELECT * FROM feature_flags WHERE name = $1 AND active = true",
            name
        )
        
        if result:
            data = {
                "enabled": result["enabled"],
                "rollout_percentage": result["rollout_percentage"],
                "conditions": json.loads(result["conditions"] or "{}")
            }
            self.cache[name] = {"data": data, "time": datetime.utcnow().timestamp()}
            return data
        
        return None
    
    async def is_enabled(self, name: str, context: dict = None) -> bool:
        flag = await self.get_flag(name)
        if not flag or not flag["enabled"]:
            return False
        
        context = context or {}
        conditions = flag["conditions"]
        
        # 用户白名单
        if "whitelist" in conditions:
            if context.get("user_id") in conditions["whitelist"]:
                return True
        
        # 用户属性条件
        if "user_segment" in conditions:
            segment = conditions["user_segment"]
            if context.get("segment") != segment:
                return False
        
        # 百分比灰度
        percentage = flag["rollout_percentage"]
        if percentage < 100:
            user_id = context.get("user_id", "")
            hash_val = hash(f"{name}:{user_id}") % 100
            return hash_val < percentage
        
        return True

class ProgressiveRollout:
    def __init__(self, flag_manager):
        self.flag_manager = flag_manager
    
    async def start_rollout(self, flag_name: str, stages: list[dict]):
        """
        stages = [
            {"percentage": 1, "duration_hours": 24},
            {"percentage": 5, "duration_hours": 24},
            {"percentage": 25, "duration_hours": 48},
            {"percentage": 50, "duration_hours": 48},
            {"percentage": 100, "duration_hours": 0}
        ]
        """
        for stage in stages:
            await self.flag_manager.update_rollout(
                flag_name,
                stage["percentage"]
            )
            
            # 监控指标
            metrics = await self.monitor_metrics(flag_name)
            
            if metrics["error_rate"] > 0.01:  # 错误率超过 1%
                await self.rollback(flag_name)
                raise Exception(f"Rollback triggered: error rate {metrics['error_rate']}")
            
            if stage["duration_hours"] > 0:
                await asyncio.sleep(stage["duration_hours"] * 3600)
    
    async def rollback(self, flag_name: str):
        await self.flag_manager.update_rollout(flag_name, 0)

import hashlib

class ABTest:
    def __init__(self, test_name: str, variants: list[str], weights: list[float]):
        self.test_name = test_name
        self.variants = variants
        self.weights = weights  # 权重之和应为 100
    
    def get_variant(self, user_id: str) -> str:
        """根据用户 ID 确定性地分配变体"""
        hash_input = f"{self.test_name}:{user_id}"
        hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16) % 100
        
        cumulative = 0
        for variant, weight in zip(self.variants, self.weights):
            cumulative += weight
            if hash_value < cumulative:
                return variant
        
        return self.variants[-1]

# 使用
checkout_test = ABTest(
    test_name="checkout_redesign",
    variants=["control", "variant_a", "variant_b"],
    weights=[50, 25, 25]  # 50% 对照，25% A，25% B
)

@app.get("/checkout")
async def checkout(user: User = Depends(get_current_user)):
    variant = checkout_test.get_variant(user.id)
    
    # 记录曝光
    await log_exposure(user.id, "checkout_redesign", variant)
    
    if variant == "control":
        return render_checkout_v1()
    elif variant == "variant_a":
        return render_checkout_v2a()
    else:
        return render_checkout_v2b()

async def analyze_ab_test(test_name: str) -> dict:
    """分析 A/B 测试结果"""
    results = await db.fetch_all("""
        SELECT
            variant,
            COUNT(*) as exposures,
            SUM(CASE WHEN converted THEN 1 ELSE 0 END) as conversions
        FROM ab_test_events
        WHERE test_name = $1
        GROUP BY variant
    """, test_name)
    
    analysis = {}
    for row in results:
        conversion_rate = row["conversions"] / row["exposures"] if row["exposures"] > 0 else 0
        analysis[row["variant"]] = {
            "exposures": row["exposures"],
            "conversions": row["conversions"],
            "conversion_rate": conversion_rate
        }
    
    # 计算统计显著性（简化版）
    control = analysis.get("control", {})
    for variant_name, data in analysis.items():
        if variant_name == "control":
            continue
        
        lift = (data["conversion_rate"] - control["conversion_rate"]) / control["conversion_rate"]
        data["lift"] = lift
    
    return analysis