dc-docs/tb/ai_analyzer.py

import json
import os
import sys
from datetime import datetime
from typing import Dict, Any, Optional

import anthropic

from .tb_config import TBConfig


class AIAnalyzer:
    def __init__(self, config: TBConfig):
        self.config = config
        api_key = config.get('ai.api_key') or os.environ.get('ANTHROPIC_API_KEY')
        if not api_key:
            raise ValueError("未配置 Anthropic API Key。请在 config.json 的 ai.api_key 中设置，或设置环境变量 ANTHROPIC_API_KEY")
        self.client = anthropic.Anthropic(api_key=api_key)
        self.model = config.get('ai.model', 'claude-sonnet-4-20250514')
        self.max_tokens = config.get('ai.max_tokens', 4096)

        if getattr(sys, 'frozen', False):
            base_path = os.path.dirname(sys.executable)
        else:
            base_path = os.path.dirname(os.path.abspath(__file__))
        self.base_path = base_path
        self.sop_path = os.path.join(base_path, 'pm_sop.md')

    def _read_json(self, filename: str) -> Dict:
        path = os.path.join(self.base_path, 'data', filename)
        if not os.path.exists(path):
            return {}
        with open(path, 'r', encoding='utf-8') as f:
            return json.load(f)

    def build_context(self) -> str:
        summary = self._read_json('summary.json')
        task_state = self._read_json('task_state.json')
        teams = self.config.get('teams', {})
        user_roles = self.config.get('user_roles', {})
        focus_iteration = self.config.get('ai.focus_iteration', '')

        truncated_summary = self._truncate_summary(summary, focus_iteration)

        # Filter task_state by iteration if configured
        done_statuses = {'已完成', '已关闭', 'Done', 'Closed', '测试完成', '制作完成'}
        active_tasks = task_state.get('tasks', {})
        if focus_iteration:
            active_tasks = {
                tid: t for tid, t in active_tasks.items()
                if t.get('iteration', '') == focus_iteration and t.get('status') not in done_statuses
            }
        else:
            active_tasks = {
                tid: t for tid, t in active_tasks.items()
                if t.get('status') not in done_statuses
            }

        # Filter recent_wins by iteration
        recent_wins = summary.get('recent_wins', [])
        if focus_iteration:
            # recent_wins don't have iteration, filter from by_user tasks instead
            recent_wins = []

        overview = {
            'focus_iteration': focus_iteration or '全部任务',
            'total_tasks': summary.get('total_tasks'),
            'project_progress': summary.get('project_progress'),
            'by_status': summary.get('by_status'),
            'by_priority': summary.get('by_priority'),
            'by_iteration': summary.get('by_iteration', {}),
            'updated_at': summary.get('updated_at')
        }
        team_cfg = {'teams': teams, 'user_roles': user_roles}

        context_parts = [
            "## 项目概览\n```json\n" + json.dumps(overview, ensure_ascii=False, indent=2) + "\n```",
            "## 团队成员数据（已截断）\n```json\n" + json.dumps(truncated_summary.get('by_user', {}), ensure_ascii=False, indent=2) + "\n```",
            "## 活跃任务快照\n```json\n" + json.dumps(active_tasks, ensure_ascii=False, indent=2) + "\n```",
            "## 团队配置\n```json\n" + json.dumps(team_cfg, ensure_ascii=False, indent=2) + "\n```",
        ]
        return '\n\n'.join(context_parts)

    def _truncate_summary(self, summary: Dict, focus_iteration: str = '') -> Dict:
        result = dict(summary)
        by_user = summary.get('by_user', {})
        truncated = {}
        done_statuses = {'已完成', '已关闭', 'Done', 'Closed', '测试完成', '制作完成'}

        for user, data in by_user.items():
            user_data = dict(data)
            tasks = data.get('tasks', [])
            # Filter by iteration if configured
            if focus_iteration:
                tasks = [t for t in tasks if t.get('iteration', '') == focus_iteration]
            active = [t for t in tasks if t.get('status') not in done_statuses]
            user_data['tasks'] = active[:5]
            user_data['active_task_count'] = len(active)
            user_data['iteration_task_count'] = len(tasks)
            truncated[user] = user_data

        result['by_user'] = truncated
        return result

    def load_sop(self) -> str:
        with open(self.sop_path, 'r', encoding='utf-8') as f:
            content = f.read()
        return content.split('## Chat 模式')[0].strip()

    def run_full_analysis(self) -> Dict[str, Any]:
        context = self.build_context()
        sop = self.load_sop()

        prompt = f"{sop}\n\n---\n\n以下是当前项目数据：\n\n{context}"

        response = self.client.messages.create(
            model=self.model,
            max_tokens=self.max_tokens,
            messages=[{"role": "user", "content": prompt}],
        )

        raw_text = response.content[0].text
        analysis = self._parse_json_response(raw_text)
        analysis['generated_at'] = datetime.now().isoformat()
        analysis['model'] = self.model

        output_path = os.path.join(self.base_path, 'data', 'ai_analysis.json')
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(analysis, f, ensure_ascii=False, indent=2)

        return analysis

    def get_latest(self) -> Optional[Dict]:
        path = os.path.join(self.base_path, 'data', 'ai_analysis.json')
        if not os.path.exists(path):
            return None
        with open(path, 'r', encoding='utf-8') as f:
            return json.load(f)

    def _parse_json_response(self, text: str) -> Dict:
        text = text.strip()
        if text.startswith('```json'):
            text = text[7:]
        if text.startswith('```'):
            text = text[3:]
        if text.endswith('```'):
            text = text[:-3]
        text = text.strip()

        try:
            return json.loads(text)
        except json.JSONDecodeError:
            start = text.find('{')
            end = text.rfind('}')
            if start != -1 and end != -1:
                try:
                    return json.loads(text[start:end + 1])
                except json.JSONDecodeError:
                    pass
            return {'raw_response': text, 'parse_error': True}


if __name__ == '__main__':
    print("正在执行 AI 项目分析...")
    config = TBConfig()
    analyzer = AIAnalyzer(config)
    result = analyzer.run_full_analysis()
    print(f"\n分析完成！结果已保存到 data/ai_analysis.json")
    print(f"一句话总结: {result.get('summary', 'N/A')}")
    risks = result.get('risks', [])
    print(f"识别到 {len(risks)} 个风险")
    recs = result.get('recommendations', [])
    print(f"生成 {len(recs)} 条建议")