xiaoxia-saas/packages/infrastructure/feature_flags.py

"""Feature Flags - 功能开关系统

支持三级控制：
1. 全局开关：全局启用/禁用某个功能
2. 套餐级别：按订阅套餐（free/basic/premium）控制
3. 用户白名单：针对特定用户单独启用

优先级：用户白名单 > 套餐级别 > 全局开关

Usage:
    from packages.infrastructure.feature_flags import feature_flags

    # 检查功能是否启用
    if feature_flags.is_enabled("ai_voice_generation", user_plan="basic", user_id="u123"):
        ...

    # 管理员动态更新
    feature_flags.set_global("batch_export", enabled=False)
    feature_flags.set_plan_override("multi_platform_output", "premium", True)
    feature_flags.set_user_override("deduplication_report", "user_42", True)
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from typing import Optional

logger = logging.getLogger(__name__)


@dataclass
class FeatureFlag:
    """单个 Feature Flag 的定义"""

    name: str
    description: str = ""
    # 全局开关，默认 True（启用）
    global_enabled: bool = True
    # 套餐级别覆盖：{plan_name: bool}
    plan_overrides: dict[str, bool] = field(default_factory=dict)
    # 用户白名单：user_id -> bool（True=强制启用，False=强制禁用）
    user_overrides: dict[str, bool] = field(default_factory=dict)

    def is_enabled(
        self,
        user_plan: Optional[str] = None,
        user_id: Optional[str] = None,
    ) -> bool:
        """判断功能是否对指定用户启用

        优先级：用户白名单 > 套餐级别 > 全局开关
        """
        # 1. 用户白名单优先
        if user_id and user_id in self.user_overrides:
            return self.user_overrides[user_id]

        # 2. 套餐级别
        if user_plan and user_plan in self.plan_overrides:
            return self.plan_overrides[user_plan]

        # 3. 全局开关
        return self.global_enabled


class FeatureScope:
    """Feature Flag 名称常量，避免硬编码字符串"""

    AI_VOICE_GENERATION = "ai_voice_generation"
    DEDUPLICATION_REPORT = "deduplication_report"
    BATCH_EXPORT = "batch_export"
    MULTI_PLATFORM_OUTPUT = "multi_platform_output"
    RECIPE_REUSE = "recipe_reuse"


class FeatureFlags:
    """Feature Flags 管理器

    单例模式，全局唯一实例（feature_flags）。
    """

    def __init__(self) -> None:
        self._flags: dict[str, FeatureFlag] = {}
        self._init_defaults()

    def _init_defaults(self) -> None:
        """初始化默认 Feature Flags"""
        defaults = [
            FeatureFlag(
                name=FeatureScope.AI_VOICE_GENERATION,
                description="AI 配音生成功能",
                global_enabled=True,
                plan_overrides={"free": False},  # free 套餐不可用
            ),
            FeatureFlag(
                name=FeatureScope.DEDUPLICATION_REPORT,
                description="素材去重检测报告",
                global_enabled=True,
                plan_overrides={"free": False, "basic": False},  # 仅 premium
            ),
            FeatureFlag(
                name=FeatureScope.BATCH_EXPORT,
                description="批量导出功能",
                global_enabled=True,
                plan_overrides={"free": False},
            ),
            FeatureFlag(
                name=FeatureScope.MULTI_PLATFORM_OUTPUT,
                description="多平台发布输出",
                global_enabled=True,
                plan_overrides={"free": False, "basic": False},  # 仅 premium
            ),
            FeatureFlag(
                name=FeatureScope.RECIPE_REUSE,
                description="配方复用功能",
                global_enabled=True,
                plan_overrides={"free": False},  # 仅基础版和高级版可用
            ),
        ]
        for flag in defaults:
            self._flags[flag.name] = flag

    def register(self, flag: FeatureFlag) -> None:
        """注册一个新的 Feature Flag"""
        self._flags[flag.name] = flag
        logger.info(f"Feature flag '{flag.name}' registered")

    def get(self, name: str) -> Optional[FeatureFlag]:
        """获取 Feature Flag 定义"""
        return self._flags.get(name)

    def is_enabled(
        self,
        name: str,
        user_plan: Optional[str] = None,
        user_id: Optional[str] = None,
    ) -> bool:
        """检查功能是否启用

        Args:
            name: Feature flag 名称
            user_plan: 用户订阅套餐 (free/basic/premium)
            user_id: 用户 ID

        Returns:
            True 如果功能启用，False 否则
        """
        flag = self._flags.get(name)
        if flag is None:
            logger.warning(f"Feature flag '{name}' not found, defaulting to disabled")
            return False
        return flag.is_enabled(user_plan=user_plan, user_id=user_id)

    def set_global(self, name: str, enabled: bool) -> None:
        """设置全局开关"""
        flag = self._flags.get(name)
        if flag is None:
            raise KeyError(f"Feature flag '{name}' not found")
        flag.global_enabled = enabled
        logger.info(f"Feature flag '{name}' global set to {enabled}")

    def set_plan_override(self, name: str, plan: str, enabled: bool) -> None:
        """设置套餐级别覆盖"""
        flag = self._flags.get(name)
        if flag is None:
            raise KeyError(f"Feature flag '{name}' not found")
        flag.plan_overrides[plan] = enabled
        logger.info(f"Feature flag '{name}' plan '{plan}' set to {enabled}")

    def set_user_override(self, name: str, user_id: str, enabled: bool) -> None:
        """设置用户白名单覆盖"""
        flag = self._flags.get(name)
        if flag is None:
            raise KeyError(f"Feature flag '{name}' not found")
        flag.user_overrides[user_id] = enabled
        logger.info(f"Feature flag '{name}' user '{user_id}' set to {enabled}")

    def list_flags(self) -> dict[str, FeatureFlag]:
        """列出所有 Feature Flags"""
        return dict(self._flags)

    def get_enabled_for_plan(self, plan: str) -> list[str]:
        """获取指定套餐下所有启用的功能名称"""
        return [name for name, flag in self._flags.items() if flag.is_enabled(user_plan=plan)]


# 全局单例
feature_flags = FeatureFlags()