# community.py
class CommunityManager:
    """社群运营"""
    
    def __init__(self):
        self.groups = []
        self.members = {}
    
    def onboarding_flow(self, user_id: str):
        """新用户引导流程"""
        flow = [
            {
                'day': 0,
                'action': 'send_welcome',
                'content': '欢迎加入！这是Skill使用指南'
            },
            {
                'day': 1,
                'action': 'send_tutorial',
                'content': '今天教你一个实用技巧'
            },
            {
                'day': 3,
                'action': 'check_usage',
                'content': '使用中有遇到问题吗？'
            },
            {
                'day': 7,
                'action': 'offer_upgrade',
                'content': '升级到专业版解锁更多功能'
            }
        ]
        
        return flow
    
    def create_engagement_event(self, event_type: str):
        """创建互动活动"""
        events = {
            'weekly_challenge': {
                'name': '每周挑战',
                'description': '完成指定任务赢取奖励',
                'reward': '1个月专业版'
            },
            'expert_sharing': {
                'name': '专家分享',
                'description': '邀请行业专家分享经验',
                'format': '直播+Q&A'
            },
            'user_showcase': {
                'name': '用户 showcase',
                'description': '展示优秀用户案例',
                'benefit': '曝光+奖励'
            }
        }
        
        return events.get(event_type)

# review.py
class ReviewManager:
    """评价管理"""
    
    def __init__(self, storage):
        self.storage = storage
    
    def request_review(self, user_id: str, trigger: str):
        """请求用户评价"""
        # 选择合适的时机
        good_triggers = [
            '完成重要任务后',
            '使用满7天后',
            '成功解决问题后',
            '升级付费后'
        ]
        
        if trigger in good_triggers:
            self._send_review_request(user_id)
    
    def handle_negative_review(self, review: Dict):
        """处理负面评价"""
        # 1. 及时响应
        self._respond_promptly(review)
        
        # 2. 了解详情
        self._follow_up(review['user_id'])
        
        # 3. 解决问题
        self._resolve_issue(review['issue'])
        
        # 4. 请求更新评价
        self._request_update(review['user_id'])
    
    def showcase_testimonials(self) -> List[Dict]:
        """展示精选评价"""
        testimonials = self.storage.query(
            rating=5,
            has_detail=True,
            is_public=True
        )
        
        # 按行业/场景分类
        categorized = {}
        for t in testimonials:
            category = t.get('user_category', 'other')
            if category not in categorized:
                categorized[category] = []
            categorized[category].append(t)
        
        return categorized

# marketing_analytics.py
class MarketingAnalytics:
    """营销分析"""
    
    def __init__(self):
        self.channels = {}
    
    def track_acquisition(self, channel: str, user_id: str):
        """追踪获客"""
        self._record_event('acquisition', {
            'channel': channel,
            'user_id': user_id,
            'timestamp': datetime.now().isoformat()
        })
    
    def calculate_cac(self, channel: str, period: str) -> float:
        """计算获客成本"""
        spend = self._get_marketing_spend(channel, period)
        new_users = self._get_new_users(channel, period)
        
        return spend / new_users if new_users > 0 else 0
    
    def calculate_ltv(self, user_segment: str) -> float:
        """计算用户生命周期价值"""
        avg_revenue = self._get_avg_monthly_revenue(user_segment)
        avg_lifespan = self._get_avg_lifespan(user_segment)
        
        return avg_revenue * avg_lifespan
    
    def optimize_channels(self) -> List[Dict]:
        """优化渠道投放"""
        channels = ['公众号', '小红书', '抖音', '知乎']
        
        analysis = []
        for channel in channels:
            cac = self.calculate_cac(channel, 'last_30_days')
            ltv = self.calculate_ltv(f'from_{channel}')
            
            analysis.append({
                'channel': channel,
                'cac': cac,
                'ltv': ltv,
                'roi': ltv / cac if cac > 0 else 0,
                'recommendation': self._get_recommendation(ltv, cac)
            })
        
        return sorted(analysis, key=lambda x: x['roi'], reverse=True)

# pricing.py
from enum import Enum
from typing import Dict, List

class PricingTier(Enum):
    FREE = "free"
    BASIC = "basic"
    PRO = "pro"
    ENTERPRISE = "enterprise"

class SubscriptionPlan:
    """订阅计划"""
    
    PLANS = {
        PricingTier.FREE: {
            'name': '免费版',
            'price': 0,
            'features': [
                '每月10次使用额度',
                '基础功能',
                '社区支持'
            ],
            'limits': {
                'monthly_calls': 10,
                'max_file_size': '5MB',
                'support': 'community'
            }
        },
        PricingTier.BASIC: {
            'name': '基础版',
            'price': 29,  # 元/月
            'features': [
                '每月100次使用额度',
                '全部基础功能',
                '邮件支持',
                '数据导出'
            ],
            'limits': {
                'monthly_calls': 100,
                'max_file_size': '50MB',
                'support': 'email'
            }
        },
        PricingTier.PRO: {
            'name': '专业版',
            'price': 99,  # 元/月
            'features': [
                '无限次使用',
                '高级功能',
                '优先支持',
                'API 访问',
                '自定义配置'
            ],
            'limits': {
                'monthly_calls': float('inf'),
                'max_file_size': '500MB',
                'support': 'priority'
            }
        },
        PricingTier.ENTERPRISE: {
            'name': '企业版',
            'price': None,  # 联系销售
            'features': [
                '无限次使用',
                '全部功能',
                '专属客服',
                '私有化部署',
                '定制开发',
                'SLA 保障'
            ],
            'limits': {
                'monthly_calls': float('inf'),
                'max_file_size': '10GB',
                'support': 'dedicated'
            }
        }
    }
    
    @classmethod
    def get_plan(cls, tier: PricingTier) -> Dict:
        """获取计划详情"""
        return cls.PLANS.get(tier)
    
    @classmethod
    def check_feature_access(cls, tier: PricingTier, feature: str) -> bool:
        """检查功能访问权限"""
        plan = cls.get_plan(tier)
        return feature in plan['features']
    
    @classmethod
    def check_limit(cls, tier: PricingTier, limit_type: str, value: any) -> bool:
        """检查是否超出限制"""
        plan = cls.get_plan(tier)
        limit = plan['limits'].get(limit_type)
        
        if limit == float('inf'):
            return True
        
        return value <= limit

# freemium.py
class FreemiumManager:
    """免费增值管理器"""
    
    def __init__(self, storage):
        self.storage = storage
    
    def check_usage_limit(self, user_id: str, action: str) -> Dict:
        """检查使用限制"""
        user_tier = self._get_user_tier(user_id)
        usage = self._get_usage(user_id, action)
        limit = self._get_limit(user_tier, action)
        
        remaining = limit - usage if limit != float('inf') else float('inf')
        
        return {
            'allowed': usage < limit if limit != float('inf') else True,
            'used': usage,
            'limit': limit,
            'remaining': remaining,
            'upgrade_prompt': self._generate_prompt(user_tier, action) if remaining <= 0 else None
        }
    
    def _generate_prompt(self, tier: str, action: str) -> str:
        """生成升级提示"""
        prompts = {
            'monthly_calls': {
                'title': '使用额度已用完',
                'message': '本月使用次数已达上限，升级到专业版享受无限次使用',
                'cta': '立即升级',
                'plan': 'pro'
            },
            'file_size': {
                'title': '文件超出大小限制',
                'message': '免费版支持最大5MB文件，升级到基础版支持50MB',
                'cta': '查看方案',
                'plan': 'basic'
            }
        }
        
        return prompts.get(action, {})
    
    def track_conversion(self, user_id: str, from_tier: str, to_tier: str):
        """追踪转化"""
        conversion = {
            'user_id': user_id,
            'from_tier': from_tier,
            'to_tier': to_tier,
            'converted_at': datetime.now().isoformat(),
            'days_since_signup': self._get_days_since_signup(user_id),
            'usage_before_conversion': self._get_usage_stats(user_id)
        }
        
        self.storage.save(f"conversion:{user_id}", conversion)

# enterprise_sales.py
class EnterpriseSales:
    """企业销售管理"""
    
    def __init__(self):
        self.leads = []
        self.deals = []
    
    def qualify_lead(self, company_info: Dict) -> Dict:
        """线索评分"""
        score = 0
        factors = {
            'company_size': {
                '1-50': 10,
                '51-200': 20,
                '201-1000': 30,
                '1000+': 40
            },
            'industry': {
                '科技': 20,
                '金融': 25,
                '制造': 15,
                '其他': 10
            },
            'budget_range': {
                '<1万': 5,
                '1-5万': 15,
                '5-20万': 25,
                '20万+': 35
            }
        }
        
        for factor, values in factors.items():
            value = company_info.get(factor)
            score += values.get(value, 0)
        
        return {
            'score': score,
            'priority': 'high' if score >= 70 else 'medium' if score >= 40 else 'low',
            'estimated_deal_size': self._estimate_deal_size(company_info)
        }
    
    def generate_proposal(self, requirements: Dict) -> Dict:
        """生成方案建议书"""
        proposal = {
            'overview': self._generate_overview(requirements),
            'solution': self._design_solution(requirements),
            'timeline': self._create_timeline(requirements),
            'pricing': self._calculate_pricing(requirements),
            'case_studies': self._get_relevant_cases(requirements['industry'])
        }
        
        return proposal

# feedback.py
from datetime import datetime
from typing import Dict, List

class FeedbackCollector:
    """反馈收集器"""
    
    def __init__(self, storage):
        self.storage = storage
    
    def collect_feedback(self, user_id: str, feedback_type: str, 
                        content: str, metadata: Dict = None) -> str:
        """收集用户反馈"""
        feedback_id = generate_id()
        
        feedback = {
            'id': feedback_id,
            'user_id': user_id,
            'type': feedback_type,  # bug, feature, improvement, other
            'content': content,
            'metadata': metadata or {},
            'status': 'new',  # new, processing, resolved, closed
            'created_at': datetime.now().isoformat(),
            'updated_at': datetime.now().isoformat()
        }
        
        self.storage.save(f"feedback:{feedback_id}", feedback)
        
        # 发送通知
        self._notify_team(feedback)
        
        return feedback_id
    
    def get_feedback_stats(self, days: int = 30) -> Dict:
        """获取反馈统计"""
        feedbacks = self._get_recent_feedbacks(days)
        
        stats = {
            'total': len(feedbacks),
            'by_type': {},
            'by_status': {},
            'avg_response_time': 0
        }
        
        for fb in feedbacks:
            # 按类型统计
            fb_type = fb['type']
            stats['by_type'][fb_type] = stats['by_type'].get(fb_type, 0) + 1
            
            # 按状态统计
            status = fb['status']
            stats['by_status'][status] = stats['by_status'].get(status, 0) + 1
        
        return stats

# analytics.py
from dataclasses import dataclass
from typing import List

@dataclass
class SkillMetrics:
    """Skill 指标"""
    # 使用指标
    total_sessions: int
    active_users: int
    avg_session_duration: float
    
    # 性能指标
    avg_response_time: float
    error_rate: float
    availability: float
    
    # 满意度指标
    completion_rate: float
    fallback_rate: float
    user_rating: float

class AnalyticsDashboard:
    """分析仪表板"""
    
    def __init__(self, storage):
        self.storage = storage
    
    def get_daily_metrics(self, date: str) -> SkillMetrics:
        """获取每日指标"""
        data = self.storage.load(f"metrics:{date}")
        
        return SkillMetrics(
            total_sessions=data.get('sessions', 0),
            active_users=data.get('active_users', 0),
            avg_session_duration=data.get('avg_duration', 0),
            avg_response_time=data.get('avg_response_time', 0),
            error_rate=data.get('error_rate', 0),
            availability=data.get('availability', 100),
            completion_rate=data.get('completion_rate', 0),
            fallback_rate=data.get('fallback_rate', 0),
            user_rating=data.get('user_rating', 0)
        )
    
    def get_intent_distribution(self, days: int = 7) -> Dict:
        """获取意图分布"""
        intents = {}
        
        for i in range(days):
            date = (datetime.now() - timedelta(days=i)).strftime('%Y-%m-%d')
            daily_intents = self.storage.load(f"intents:{date}") or {}
            
            for intent, count in daily_intents.items():
                intents[intent] = intents.get(intent, 0) + count
        
        return intents
    
    def identify_issues(self) -> List[Dict]:
        """识别问题"""
        issues = []
        
        # 检查错误率
        recent_errors = self._get_recent_errors(hours=24)
        if len(recent_errors) > 10:
            issues.append({
                'type': 'high_error_rate',
                'severity': 'high',
                'description': f'24小时内发生 {len(recent_errors)} 次错误',
                'suggestion': '检查近期更新或外部依赖'
            })
        
        # 检查响应时间
        avg_response_time = self._get_avg_response_time(hours=24)
        if avg_response_time > 3.0:
            issues.append({
                'type': 'slow_response',
                'severity': 'medium',
                'description': f'平均响应时间 {avg_response_time:.2f}s',
                'suggestion': '优化性能或增加资源'
            })
        
        # 检查用户满意度
        recent_ratings = self._get_recent_ratings(days=7)
        if recent_ratings and statistics.mean(recent_ratings) < 3.0:
            issues.append({
                'type': 'low_satisfaction',
                'severity': 'high',
                'description': '用户满意度低于 3.0',
                'suggestion': '收集反馈并改进体验'
            })
        
        return issues

# ab_test.py
import random
from typing import Dict, Any

class ABTestManager:
    """A/B 测试管理器"""
    
    def __init__(self, storage):
        self.storage = storage
        self.tests = {}
    
    def create_test(self, test_id: str, variants: List[str], 
                   traffic_split: List[float]):
        """创建测试"""
        self.tests[test_id] = {
            'variants': variants,
            'traffic_split': traffic_split,
            'results': {v: {'impressions': 0, 'conversions': 0} 
                       for v in variants}
        }
    
    def get_variant(self, test_id: str, user_id: str) -> str:
        """获取用户对应的测试版本"""
        # 根据用户ID哈希分配版本，确保同一用户始终看到同一版本
        hash_value = hash(f"{test_id}:{user_id}") % 100
        
        cumulative = 0
        for variant, split in zip(
            self.tests[test_id]['variants'],
            self.tests[test_id]['traffic_split']
        ):
            cumulative += split * 100
            if hash_value < cumulative:
                return variant
        
        return self.tests[test_id]['variants'][0]
    
    def track_event(self, test_id: str, variant: str, event: str):
        """追踪事件"""
        if event == 'impression':
            self.tests[test_id]['results'][variant]['impressions'] += 1
        elif event == 'conversion':
            self.tests[test_id]['results'][variant]['conversions'] += 1
    
    def get_results(self, test_id: str) -> Dict:
        """获取测试结果"""
        results = self.tests[test_id]['results']
        
        # 计算转化率
        for variant, data in results.items():
            if data['impressions'] > 0:
                data['conversion_rate'] = (
                    data['conversions'] / data['impressions']
                )
        
        return results

# user_segmentation.py
from datetime import datetime, timedelta

class UserSegmentation:
    """用户分层"""
    
    def __init__(self, storage):
        self.storage = storage
    
    def segment_users(self) -> Dict:
        """用户分层"""
        segments = {
            'power_users': [],      # 高频用户
            'regular_users': [],    # 常规用户
            'occasional_users': [], # 偶尔使用
            'at_risk_users': [],    # 流失风险
            'churned_users': []     # 已流失
        }
        
        users = self.storage.get_all_users()
        
        for user in users:
            segment = self._classify_user(user)
            segments[segment].append(user['id'])
        
        return segments
    
    def _classify_user(self, user: Dict) -> str:
        """分类单个用户"""
        last_active = datetime.fromisoformat(user['last_active'])
        days_since_active = (datetime.now() - last_active).days
        
        # 已流失（30天未使用）
        if days_since_active > 30:
            return 'churned_users'
        
        # 流失风险（14天未使用）
        if days_since_active > 14:
            return 'at_risk_users'
        
        # 根据使用频率分类
        monthly_sessions = user.get('monthly_sessions', 0)
        
        if monthly_sessions >= 20:
            return 'power_users'
        elif monthly_sessions >= 5:
            return 'regular_users'
        else:
            return 'occasional_users'
    
    def get_engagement_strategy(self, segment: str) -> Dict:
        """获取运营策略"""
        strategies = {
            'power_users': {
                'action': 'reward',
                'message': '感谢你的支持！邀请好友获得奖励',
                'features': ['beta_access', 'priority_support']
            },
            'regular_users': {
                'action': 'educate',
                'message': '发现新功能，提升你的效率',
                'features': ['new_feature_tips']
            },
            'occasional_users': {
                'action': 'activate',
                'message': '我们想念你！看看有什么新功能',
                'features': ['tutorial', 'use_case_showcase']
            },
            'at_risk_users': {
                'action': 'retain',
                'message': '有什么我们可以帮助的吗？',
                'features': ['feedback_request', 'discount_offer']
            },
            'churned_users': {
                'action': 'win_back',
                'message': '我们做了重大改进，回来看看吧',
                'features': ['changelog', 'special_offer']
            }
        }
        
        return strategies.get(segment, {})

# config.py
import os

class Config:
    """基础配置"""
    DEBUG = False
    LOG_LEVEL = 'INFO'
    
class DevelopmentConfig(Config):
    """开发环境"""
    DEBUG = True
    LOG_LEVEL = 'DEBUG'
    DATABASE_URL = 'sqlite:///dev.db'

class ProductionConfig(Config):
    """生产环境"""
    LOG_LEVEL = 'WARNING'
    DATABASE_URL = os.getenv('DATABASE_URL')
    API_KEY = os.getenv('API_KEY')
    
config = {
    'development': DevelopmentConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}

# utils/logger.py
import logging
import sys

class SkillLogger:
    """Skill 日志记录器"""
    
    def __init__(self, name: str):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(logging.INFO)
        
        # 控制台输出
        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setLevel(logging.INFO)
        
        # 文件输出
        file_handler = logging.FileHandler(f'logs/{name}.log')
        file_handler.setLevel(logging.DEBUG)
        
        # 格式化
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        console_handler.setFormatter(formatter)
        file_handler.setFormatter(formatter)
        
        self.logger.addHandler(console_handler)
        self.logger.addHandler(file_handler)
    
    def info(self, message: str):
        self.logger.info(message)
    
    def error(self, message: str, exc_info=False):
        self.logger.error(message, exc_info=exc_info)
    
    def metric(self, name: str, value: float):
        """记录指标"""
        self.logger.info(f"METRIC: {name}={value}")

# health_check.py
from typing import Dict
import time

class HealthChecker:
    """健康检查"""
    
    def __init__(self):
        self.checks = {}
    
    def register(self, name: str, check_func):
        """注册检查项"""
        self.checks[name] = check_func
    
    def check_all(self) -> Dict:
        """执行所有检查"""
        results = {}
        
        for name, check_func in self.checks.items():
            try:
                start = time.time()
                check_func()
                elapsed = time.time() - start
                results[name] = {
                    'status': 'healthy',
                    'response_time': elapsed
                }
            except Exception as e:
                results[name] = {
                    'status': 'unhealthy',
                    'error': str(e)
                }
        
        return results

# 使用示例
health_checker = HealthChecker()

# 注册数据库检查
health_checker.register('database', lambda: db.ping())

# 注册 API 检查
health_checker.register('api', lambda: requests.get('/health'))

# rollback.py
class RollbackManager:
    """回滚管理器"""
    
    def __init__(self):
        self.versions = []
        self.current_version = None
    
    def deploy(self, version: str):
        """部署新版本"""
        # 保存当前版本
        if self.current_version:
            self.versions.append(self.current_version)
        
        # 部署新版本
        self._do_deploy(version)
        self.current_version = version
    
    def rollback(self):
        """回滚到上一版本"""
        if not self.versions:
            raise Exception("没有可回滚的版本")
        
        previous_version = self.versions.pop()
        self._do_deploy(previous_version)
        self.current_version = previous_version
    
    def _do_deploy(self, version: str):
        """执行部署"""
        # 实际部署逻辑
        pass

# tests/test_invoice.py
import pytest
from unittest.mock import Mock, patch
from services.invoice_service import InvoiceService

class TestInvoiceService:
    """发票服务测试"""
    
    @pytest.fixture
    def service(self):
        """测试夹具"""
        mock_ocr = Mock()
        mock_db = Mock()
        return InvoiceService(mock_ocr, mock_db)
    
    def test_extract_invoice_code(self, service):
        """测试发票代码提取"""
        text = "发票代码：123456789012"
        result = service._extract_pattern(text, r'发票代码[:：]\s*(\d{12})')
        assert result == '123456789012'
    
    def test_extract_invoice_code_not_found(self, service):
        """测试发票代码不存在"""
        text = "这是一段普通文本"
        result = service._extract_pattern(text, r'发票代码[:：]\s*(\d{12})')
        assert result is None
    
    def test_validate_invoice_success(self, service):
        """测试发票验证成功"""
        info = {
            'invoice_code': '123456789012',
            'invoice_number': '12345678',
            'amount': '1000.00'
        }
        service.db.check_duplicate.return_value = False
        
        result = service._validate_invoice(info)
        
        assert result['is_valid'] is True
        assert len(result['errors']) == 0
    
    def test_validate_invoice_duplicate(self, service):
        """测试重复发票"""
        info = {
            'invoice_code': '123456789012',
            'invoice_number': '12345678',
            'amount': '1000.00'
        }
        service.db.check_duplicate.return_value = True
        
        result = service._validate_invoice(info)
        
        assert result['is_valid'] is False
        assert '发票已存在' in result['errors']

# tests/test_with_mock.py
from unittest.mock import Mock, patch, MagicMock

class TestWithMock:
    """Mock 测试示例"""
    
    def test_mock_function(self):
        """Mock 函数"""
        mock_func = Mock(return_value=42)
        result = mock_func()
        
        assert result == 42
        mock_func.assert_called_once()
    
    def test_patch_decorator(self):
        """使用 patch 装饰器"""
        with patch('services.invoice_service.OCRService') as mock_ocr:
            mock_instance = MagicMock()
            mock_instance.recognize.return_value = "识别结果"
            mock_ocr.return_value = mock_instance
            
            # 执行测试
            from services.invoice_service import InvoiceService
            service = InvoiceService(mock_instance, Mock())
            result = service.ocr.recognize("test.jpg")
            
            assert result == "识别结果"
    
    @patch('services.invoice_service.requests.get')
    def test_api_call(self, mock_get):
        """测试 API 调用"""
        mock_get.return_value.json.return_value = {'status': 'ok'}
        
        # 执行 API 调用
        import requests
        response = requests.get('http://api.example.com')
        
        assert response.json()['status'] == 'ok'

# tests/test_integration.py
import pytest
import sqlite3
from services.invoice_service import InvoiceService

class TestIntegration:
    """集成测试"""
    
    @pytest.fixture(scope='function')
    def db(self):
        """创建测试数据库"""
        conn = sqlite3.connect(':memory:')
        conn.execute('''
            CREATE TABLE invoices (
                id INTEGER PRIMARY KEY,
                invoice_code TEXT,
                invoice_number TEXT,
                amount REAL
            )
        ''')
        yield conn
        conn.close()
    
    def test_save_and_query_invoice(self, db):
        """测试保存和查询发票"""
        # 保存发票
        db.execute('''
            INSERT INTO invoices (invoice_code, invoice_number, amount)
            VALUES (?, ?, ?)
        ''', ('123456789012', '12345678', 1000.00))
        db.commit()
        
        # 查询发票
        cursor = db.execute('SELECT * FROM invoices WHERE invoice_code = ?', 
                          ('123456789012',))
        result = cursor.fetchone()
        
        assert result[1] == '123456789012'
        assert result[3] == 1000.00

# tests/test_e2e.py
class TestEndToEnd:
    """端到端测试"""
    
    def test_invoice_recognition_flow(self):
        """测试发票识别完整流程"""
        skill = FinanceAssistantSkill()
        
        # 用户请求识别发票
        response1 = skill.handle_message("帮我识别这张发票", {})
        assert "请上传发票图片" in response1
        
        # 用户上传图片
        context = {'last_intent': 'recognize_invoice'}
        response2 = skill.handle_file("test_invoice.jpg", context)
        
        # 验证响应包含关键信息
        assert "发票识别成功" in response2 or "发票代码" in response2
    
    def test_report_generation_flow(self):
        """测试报表生成流程"""
        skill = FinanceAssistantSkill()
        
        # 用户请求生成报表
        response = skill.handle_message("生成2024年1月的资产负债表", {})
        
        # 验证响应
        assert "资产负债表" in response or "已生成" in response

# tests/test_performance.py
import time
import statistics
from concurrent.futures import ThreadPoolExecutor

class TestPerformance:
    """性能测试"""
    
    def test_response_time(self):
        """测试响应时间"""
        skill = FinanceAssistantSkill()
        
        times = []
        for _ in range(10):
            start = time.time()
            skill.handle_message("查询发票", {})
            elapsed = time.time() - start
            times.append(elapsed)
        
        avg_time = statistics.mean(times)
        print(f"平均响应时间: {avg_time:.3f}s")
        
        # 断言平均响应时间小于 1 秒
        assert avg_time < 1.0
    
    def test_concurrent_requests(self):
        """测试并发请求"""
        skill = FinanceAssistantSkill()
        
        def make_request(i):
            start = time.time()
            skill.handle_message(f"请求 {i}", {})
            return time.time() - start
        
        with ThreadPoolExecutor(max_workers=5) as executor:
            times = list(executor.map(make_request, range(20)))
        
        avg_time = statistics.mean(times)
        print(f"并发平均响应时间: {avg_time:.3f}s")
        
        assert avg_time < 2.0