🤖 自动化反馈集成系统

目标: 将在线表单数据自动导入MBE，由TITANS记忆并驱动全自动化改进

实现状态: ✅ 已实现 (2026-01-28)

🚀 快速开始

1. 运行数据库迁移

# PostgreSQL
psql -U postgres -d mbe -f migrations/003_add_feedback_tables.sql

2. 配置环境变量

# .env 文件添加
FEEDBACK_SYNC_INTERVAL=300
FEEDBACK_AUTO_ANALYZE=true
GIT_PLATFORM=gitea
GIT_API_URL=http://localhost:3000/api/v1
GIT_REPO=zenglx01/mises-behavior-engine
GIT_TOKEN=your-git-token

3. 启动同步服务

# 方式1: 独立运行
python scripts/start_feedback_sync.py

# 方式2: 集成到主应用 (自动启动)
# 在 main.py 中已配置

4. API端点

• POST /api/feedback/auto/webhook/tencent - 腾讯文档Webhook
• GET /api/feedback/auto/submissions - 查询反馈列表
• POST /api/feedback/auto/sync/trigger - 手动触发同步
• POST /api/feedback/auto/analyze/trigger - 手动触发分析
• GET /api/feedback/auto/analyze/summary - 获取分析汇总

🎯 系统架构

┌──────────────────────────────────────────────────────────┐
│                  反馈收集层                               │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐              │
│  │腾讯文档表│  │Google表单│  │Slack消息 │              │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘              │
└───────┼─────────────┼─────────────┼─────────────────────┘
        │             │             │
        └─────────────┼─────────────┘
                      ↓
┌──────────────────────────────────────────────────────────┐
│              数据同步服务（新增）                         │
│  ┌──────────────────────────────────────────────┐       │
│  │  feedback_sync_service.py                    │       │
│  │  • 定时拉取表单数据                           │       │
│  │  • 数据清洗和标准化                           │       │
│  │  • 去重和增量同步                             │       │
│  └──────────────────┬───────────────────────────┘       │
└─────────────────────┼───────────────────────────────────┘
                      ↓
┌──────────────────────────────────────────────────────────┐
│              MBE 反馈数据库                               │
│  ┌──────────────────────────────────────────────┐       │
│  │  feedback_submissions (PostgreSQL)           │       │
│  │  • 原始提交数据                               │       │
│  │  • 提交时间、用户、评分                       │       │
│  │  • 问题描述、优先级                           │       │
│  └──────────────────┬───────────────────────────┘       │
└─────────────────────┼───────────────────────────────────┘
                      ↓
┌──────────────────────────────────────────────────────────┐
│            TITANS 记忆系统（集成）                        │
│  ┌──────────────────────────────────────────────┐       │
│  │  • 用户反馈向量化存储                         │       │
│  │  • 问题模式识别                               │       │
│  │  • 历史反馈关联                               │       │
│  │  • 专家改进建议记忆                           │       │
│  └──────────────────┬───────────────────────────┘       │
└─────────────────────┼───────────────────────────────────┘
                      ↓
┌──────────────────────────────────────────────────────────┐
│            AI 自动分析引擎（新增）                        │
│  ┌──────────────────────────────────────────────┐       │
│  │  feedback_analyzer.py                        │       │
│  │  • 自动问题分类                               │       │
│  │  • 优先级智能排序                             │       │
│  │  • 相似问题聚合                               │       │
│  │  • 根因分析                                   │       │
│  │  • 改进方案生成                               │       │
│  └──────────────────┬───────────────────────────┘       │
└─────────────────────┼───────────────────────────────────┘
                      ↓
┌──────────────────────────────────────────────────────────┐
│          自动化工单系统（新增）                           │
│  ┌──────────────────────────────────────────────┐       │
│  │  • 自动创建GitHub Issues / Gitea Issues      │       │
│  │  • 自动分配优先级和标签                       │       │
│  │  • 自动关联相关代码文件                       │       │
│  │  • 自动生成修复建议                           │       │
│  └──────────────────┬───────────────────────────┘       │
└─────────────────────┼───────────────────────────────────┘
                      ↓
┌──────────────────────────────────────────────────────────┐
│          Cursor AI 自动修复（未来）                       │
│  • AI 自动生成修复代码                                   │
│  • 自动运行测试验证                                       │
│  • 自动提交PR                                            │
│  • 人工审核后合并                                         │
└──────────────────────────────────────────────────────────┘

🔧 实现方案

Phase 1: 数据同步服务（立即实现）✅

文件结构：

src/feedback/
├── __init__.py
├── sync_service.py          # 数据同步服务
├── models.py                # 数据模型
├── analyzers.py             # AI分析引擎
├── integrations/
│   ├── tencent_docs.py      # 腾讯文档API
│   ├── google_forms.py      # Google Forms API
│   └── slack_api.py         # Slack API
└── automation/
    ├── issue_creator.py     # 自动创建Issue
    └── report_generator.py  # 自动生成报告

1. 数据同步服务

src/feedback/sync_service.py

"""
反馈数据自动同步服务
定时从在线表单拉取数据并导入MBE
"""

import asyncio
import hashlib
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from loguru import logger
from sqlalchemy import select, and_

from src.core.database import AsyncSessionLocal
from src.feedback.models import FeedbackSubmission
from src.feedback.integrations.tencent_docs import TencentDocsClient
from src.feedback.integrations.google_forms import GoogleFormsClient
from src.core.titans_conversation_memory import TITANSMemoryIntegration


class FeedbackSyncService:
    """反馈数据同步服务"""
    
    def __init__(self):
        self.tencent_client = TencentDocsClient()
        self.google_client = GoogleFormsClient()
        self.titans_memory = TITANSMemoryIntegration()
        self.sync_interval = 300  # 5分钟同步一次
        
    async def start(self):
        """启动同步服务"""
        logger.info("🚀 反馈同步服务启动")
        while True:
            try:
                await self.sync_all_sources()
                await asyncio.sleep(self.sync_interval)
            except Exception as e:
                logger.error(f"同步失败: {e}")
                await asyncio.sleep(60)  # 失败后等待1分钟重试
    
    async def sync_all_sources(self):
        """同步所有数据源"""
        logger.info("📥 开始同步反馈数据...")
        
        # 同步腾讯文档
        tencent_count = await self.sync_tencent_docs()
        
        # 同步Google Forms
        google_count = await self.sync_google_forms()
        
        # 同步Slack消息
        slack_count = await self.sync_slack_messages()
        
        total = tencent_count + google_count + slack_count
        logger.info(f"✅ 同步完成: 新增 {total} 条反馈")
        
        # 触发AI分析
        if total > 0:
            await self.trigger_analysis()
    
    async def sync_tencent_docs(self) -> int:
        """同步腾讯文档表单数据"""
        try:
            # 获取表单配置
            form_id = await self._get_config("tencent_form_id")
            if not form_id:
                return 0
            
            # 拉取数据
            responses = await self.tencent_client.get_responses(form_id)
            logger.info(f"📋 腾讯文档: 获取到 {len(responses)} 条响应")
            
            # 保存到数据库
            new_count = 0
            async with AsyncSessionLocal() as session:
                for response in responses:
                    # 检查是否已存在（基于哈希去重）
                    response_hash = self._calculate_hash(response)
                    exists = await session.execute(
                        select(FeedbackSubmission).where(
                            FeedbackSubmission.response_hash == response_hash
                        )
                    )
                    if exists.scalar_one_or_none():
                        continue
                    
                    # 创建新记录
                    feedback = FeedbackSubmission(
                        source="tencent_docs",
                        response_hash=response_hash,
                        submitted_at=response.get("submitted_at"),
                        tester_name=response.get("name"),
                        test_duration=response.get("duration"),
                        
                        # AI对话评分
                        expert_matching_score=response.get("expert_matching"),
                        answer_quality_score=response.get("answer_quality"),
                        context_understanding=response.get("context_understanding"),
                        response_speed=response.get("response_speed"),
                        
                        # 文档评分
                        doc_readability_score=response.get("doc_readability"),
                        
                        # 问题描述
                        high_priority_issues=response.get("high_priority"),
                        medium_priority_issues=response.get("medium_priority"),
                        low_priority_issues=response.get("low_priority"),
                        
                        # 总体评价
                        overall_score=response.get("overall_score"),
                        recommendation=response.get("recommendation"),
                        summary=response.get("summary"),
                        
                        # 原始数据
                        raw_data=response
                    )
                    
                    session.add(feedback)
                    new_count += 1
                    
                    # 存入TITANS记忆
                    await self._store_in_titans(feedback)
                
                await session.commit()
            
            logger.info(f"💾 腾讯文档: 新增 {new_count} 条反馈")
            return new_count
            
        except Exception as e:
            logger.error(f"腾讯文档同步失败: {e}")
            return 0
    
    async def sync_google_forms(self) -> int:
        """同步Google Forms数据"""
        try:
            form_id = await self._get_config("google_form_id")
            if not form_id:
                return 0
            
            responses = await self.google_client.get_responses(form_id)
            logger.info(f"📋 Google Forms: 获取到 {len(responses)} 条响应")
            
            # 类似的处理逻辑...
            return 0
            
        except Exception as e:
            logger.error(f"Google Forms同步失败: {e}")
            return 0
    
    async def sync_slack_messages(self) -> int:
        """同步Slack频道中的反馈消息"""
        # 实现Slack消息同步
        return 0
    
    async def _store_in_titans(self, feedback: FeedbackSubmission):
        """将反馈存入TITANS记忆系统"""
        try:
            # 构建记忆文本
            memory_text = f"""
            测试反馈 - {feedback.submitted_at.strftime('%Y-%m-%d %H:%M')}
            测试人: {feedback.tester_name}
            总体评分: {feedback.overall_score}/10
            推荐度: {feedback.recommendation}
            
            高优先级问题:
            {feedback.high_priority_issues}
            
            中优先级问题:
            {feedback.medium_priority_issues}
            
            总结: {feedback.summary}
            """
            
            # 存入TITANS（使用专门的"反馈记忆"专家）
            await self.titans_memory.add_conversation(
                user_id="feedback_system",
                expert_name="测试反馈分析专家",
                user_message=memory_text,
                assistant_message="已记录并分析此反馈",
                metadata={
                    "feedback_id": feedback.id,
                    "priority": self._calculate_priority(feedback),
                    "issues_count": self._count_issues(feedback),
                }
            )
            
            logger.debug(f"💾 TITANS记忆: 已存储反馈 #{feedback.id}")
            
        except Exception as e:
            logger.error(f"TITANS记忆存储失败: {e}")
    
    def _calculate_hash(self, response: Dict) -> str:
        """计算响应哈希（用于去重）"""
        key = f"{response.get('submitted_at')}_{response.get('name')}_{response.get('summary')}"
        return hashlib.md5(key.encode()).hexdigest()
    
    def _calculate_priority(self, feedback: FeedbackSubmission) -> str:
        """计算反馈优先级"""
        if feedback.high_priority_issues and len(feedback.high_priority_issues) > 100:
            return "critical"
        elif feedback.overall_score < 6:
            return "high"
        elif feedback.overall_score < 8:
            return "medium"
        else:
            return "low"
    
    def _count_issues(self, feedback: FeedbackSubmission) -> int:
        """统计问题数量"""
        count = 0
        if feedback.high_priority_issues:
            count += feedback.high_priority_issues.count('\n') + 1
        if feedback.medium_priority_issues:
            count += feedback.medium_priority_issues.count('\n') + 1
        if feedback.low_priority_issues:
            count += feedback.low_priority_issues.count('\n') + 1
        return count
    
    async def trigger_analysis(self):
        """触发AI自动分析"""
        from src.feedback.analyzers import FeedbackAnalyzer
        analyzer = FeedbackAnalyzer()
        await analyzer.analyze_recent_feedback()
    
    async def _get_config(self, key: str) -> Optional[str]:
        """获取配置"""
        # 从数据库或环境变量读取配置
        import os
        return os.getenv(key.upper())

2. 数据模型

src/feedback/models.py

"""
反馈数据模型
"""

from datetime import datetime
from sqlalchemy import Column, Integer, String, Text, Float, DateTime, JSON
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class FeedbackSubmission(Base):
    """测试反馈提交记录"""
    __tablename__ = "feedback_submissions"
    
    id = Column(Integer, primary_key=True)
    
    # 元数据
    source = Column(String(50))  # tencent_docs, google_forms, slack
    response_hash = Column(String(64), unique=True, index=True)
    submitted_at = Column(DateTime, default=datetime.utcnow)
    synced_at = Column(DateTime, default=datetime.utcnow)
    
    # 测试人信息
    tester_name = Column(String(100))
    test_duration = Column(String(50))
    completed_tests = Column(JSON)  # 完成的测试项列表
    
    # AI对话评分
    expert_matching_score = Column(Integer)  # 1-5
    answer_quality_score = Column(Integer)  # 1-5
    context_understanding = Column(String(50))  # 能/不能/有时能
    response_speed = Column(String(50))  # <2秒 / 2-5秒 / 5-10秒 / >10秒
    test_questions_results = Column(Text)
    
    # 文档评分
    doc_readability_score = Column(Integer)  # 1-5
    docs_viewed = Column(JSON)
    doc_issues = Column(Text)
    
    # UI/UX评分
    ui_score = Column(Integer, nullable=True)
    ux_issues = Column(Text, nullable=True)
    
    # 性能评分
    performance_score = Column(Integer, nullable=True)
    performance_issues = Column(Text, nullable=True)
    
    # 问题汇总
    high_priority_issues = Column(Text)
    medium_priority_issues = Column(Text)
    low_priority_issues = Column(Text)
    
    # 总体评价
    overall_score = Column(Integer)  # 1-10
    recommendation = Column(String(50))  # 推荐/修复后/暂不
    summary = Column(Text)
    
    # 分析结果（AI生成）
    ai_analyzed = Column(DateTime, nullable=True)
    issue_category = Column(String(100), nullable=True)
    priority = Column(String(20), nullable=True)  # critical, high, medium, low
    root_cause = Column(Text, nullable=True)
    fix_suggestion = Column(Text, nullable=True)
    related_files = Column(JSON, nullable=True)
    
    # 处理状态
    status = Column(String(50), default="pending")  # pending, analyzing, issue_created, fixed, verified
    github_issue_url = Column(String(500), nullable=True)
    fixed_at = Column(DateTime, nullable=True)
    
    # 原始数据
    raw_data = Column(JSON)
    
    def __repr__(self):
        return f"<FeedbackSubmission #{self.id} from {self.tester_name} score={self.overall_score}>"


class FeedbackPattern(Base):
    """反馈模式识别（聚合相似问题）"""
    __tablename__ = "feedback_patterns"
    
    id = Column(Integer, primary_key=True)
    pattern_name = Column(String(200))
    category = Column(String(100))
    occurrence_count = Column(Integer, default=1)
    first_seen = Column(DateTime, default=datetime.utcnow)
    last_seen = Column(DateTime, default=datetime.utcnow)
    
    # 关联的反馈ID列表
    feedback_ids = Column(JSON)
    
    # 模式描述
    description = Column(Text)
    root_cause = Column(Text, nullable=True)
    fix_priority = Column(String(20))
    
    # 处理状态
    status = Column(String(50), default="identified")
    github_issue_url = Column(String(500), nullable=True)

3. 腾讯文档集成

src/feedback/integrations/tencent_docs.py

"""
腾讯文档API集成
"""

import aiohttp
from typing import List, Dict
from loguru import logger


class TencentDocsClient:
    """腾讯文档客户端"""
    
    def __init__(self):
        self.api_base = "https://docs.qq.com/api"
        self.access_token = None  # 需要配置
    
    async def get_responses(self, form_id: str) -> List[Dict]:
        """
        获取表单响应数据
        
        注意: 腾讯文档需要手动导出Excel然后上传
        或使用腾讯云API（需要企业认证）
        
        这里提供两种方案:
        方案1: Webhook（推荐）- 表单提交时推送到MBE
        方案2: 定期导出Excel并自动解析
        """
        
        # 方案2示例: 解析导出的Excel文件
        return await self._parse_excel_export()
    
    async def _parse_excel_export(self) -> List[Dict]:
        """
        解析导出的Excel文件
        
        假设文件被自动上传到 /data/feedback/latest_export.xlsx
        """
        try:
            import pandas as pd
            
            file_path = "/data/feedback/latest_export.xlsx"
            df = pd.read_excel(file_path)
            
            responses = []
            for _, row in df.iterrows():
                response = {
                    "submitted_at": row.get("提交时间"),
                    "name": row.get("姓名"),
                    "duration": row.get("测试时长"),
                    "expert_matching": row.get("专家匹配准确性"),
                    "answer_quality": row.get("回答质量"),
                    "context_understanding": row.get("上下文理解"),
                    "response_speed": row.get("响应速度"),
                    "doc_readability": row.get("文档可读性"),
                    "high_priority": row.get("高优先级问题"),
                    "medium_priority": row.get("中优先级问题"),
                    "low_priority": row.get("低优先级问题"),
                    "overall_score": row.get("总体评分"),
                    "recommendation": row.get("是否推荐发布"),
                    "summary": row.get("一句话总结"),
                }
                responses.append(response)
            
            return responses
            
        except Exception as e:
            logger.error(f"Excel解析失败: {e}")
            return []
    
    async def setup_webhook(self, form_id: str, webhook_url: str):
        """
        设置Webhook（如果腾讯文档支持）
        
        当有新提交时，自动推送到MBE的webhook端点
        """
        # 实现Webhook配置
        pass


class TencentDocsWebhook:
    """处理腾讯文档Webhook推送"""
    
    @staticmethod
    async def handle_submission(data: Dict):
        """处理表单提交推送"""
        from src.feedback.sync_service import FeedbackSyncService
        service = FeedbackSyncService()
        
        # 直接处理单条提交
        feedback = await service._create_feedback_from_response(
            source="tencent_docs_webhook",
            response=data
        )
        
        logger.info(f"✅ Webhook: 收到新反馈 #{feedback.id}")

4. AI自动分析引擎

src/feedback/analyzers.py

"""
AI反馈自动分析引擎
"""

from datetime import datetime, timedelta
from typing import List, Dict
from loguru import logger
from sqlalchemy import select, and_

from src.core.database import AsyncSessionLocal
from src.feedback.models import FeedbackSubmission, FeedbackPattern
from src.llm.base import LLMClient
from src.core.titans_conversation_memory import TITANSMemoryIntegration


class FeedbackAnalyzer:
    """反馈智能分析器"""
    
    def __init__(self):
        self.llm = LLMClient()
        self.titans = TITANSMemoryIntegration()
    
    async def analyze_recent_feedback(self, hours: int = 24):
        """分析最近的反馈"""
        logger.info(f"🔍 开始分析最近 {hours} 小时的反馈...")
        
        async with AsyncSessionLocal() as session:
            # 获取未分析的反馈
            since = datetime.utcnow() - timedelta(hours=hours)
            result = await session.execute(
                select(FeedbackSubmission).where(
                    and_(
                        FeedbackSubmission.submitted_at >= since,
                        FeedbackSubmission.ai_analyzed.is_(None)
                    )
                )
            )
            feedbacks = result.scalars().all()
            
            if not feedbacks:
                logger.info("✅ 没有待分析的反馈")
                return
            
            logger.info(f"📊 待分析反馈: {len(feedbacks)} 条")
            
            # 逐条分析
            for feedback in feedbacks:
                await self._analyze_single(feedback, session)
            
            # 模式识别
            await self._identify_patterns(feedbacks, session)
            
            # 生成汇总报告
            await self._generate_summary_report(feedbacks)
            
            await session.commit()
            
            logger.info("✅ 反馈分析完成")
    
    async def _analyze_single(self, feedback: FeedbackSubmission, session):
        """分析单条反馈"""
        try:
            # 构建分析提示词
            prompt = f"""
            你是一个专业的软件测试反馈分析专家。请分析以下用户反馈:
            
            测试人: {feedback.tester_name}
            总体评分: {feedback.overall_score}/10
            推荐度: {feedback.recommendation}
            
            AI对话评分:
            - 专家匹配: {feedback.expert_matching_score}/5
            - 回答质量: {feedback.answer_quality_score}/5
            - 上下文理解: {feedback.context_understanding}
            - 响应速度: {feedback.response_speed}
            
            文档评分:
            - 可读性: {feedback.doc_readability_score}/5
            
            高优先级问题:
            {feedback.high_priority_issues or "无"}
            
            中优先级问题:
            {feedback.medium_priority_issues or "无"}
            
            低优先级问题:
            {feedback.low_priority_issues or "无"}
            
            一句话总结:
            {feedback.summary}
            
            请分析并返回JSON格式:
            {{
                "issue_category": "功能/性能/文档/UI/其他",
                "priority": "critical/high/medium/low",
                "root_cause": "根本原因分析",
                "fix_suggestion": "修复建议",
                "related_files": ["可能相关的文件路径"],
                "estimated_effort": "预计工作量（小时）"
            }}
            """
            
            # 调用LLM分析
            response = await self.llm.chat(prompt)
            
            # 解析结果
            import json
            analysis = json.loads(response)
            
            # 更新反馈记录
            feedback.ai_analyzed = datetime.utcnow()
            feedback.issue_category = analysis.get("issue_category")
            feedback.priority = analysis.get("priority")
            feedback.root_cause = analysis.get("root_cause")
            feedback.fix_suggestion = analysis.get("fix_suggestion")
            feedback.related_files = analysis.get("related_files", [])
            
            logger.info(f"✅ 分析完成: #{feedback.id} - {analysis['priority']} - {analysis['issue_category']}")
            
            # 如果是高优先级，自动创建Issue
            if analysis["priority"] in ["critical", "high"]:
                await self._auto_create_issue(feedback, analysis)
            
        except Exception as e:
            logger.error(f"分析失败: {e}")
    
    async def _identify_patterns(self, feedbacks: List[FeedbackSubmission], session):
        """识别反馈模式（相似问题聚合）"""
        logger.info("🔍 识别反馈模式...")
        
        # 使用TITANS的语义搜索能力找相似问题
        # 实现聚类算法
        # ...
        
        pass
    
    async def _auto_create_issue(self, feedback: FeedbackSubmission, analysis: Dict):
        """自动创建GitHub/Gitea Issue"""
        from src.feedback.automation.issue_creator import IssueCreator
        
        creator = IssueCreator()
        issue_url = await creator.create_issue(
            title=f"[测试反馈] {feedback.summary[:50]}",
            body=self._format_issue_body(feedback, analysis),
            labels=[analysis["priority"], analysis["issue_category"]],
            assignees=[]  # 可根据category自动分配
        )
        
        feedback.github_issue_url = issue_url
        feedback.status = "issue_created"
        
        logger.info(f"📝 已创建Issue: {issue_url}")
    
    def _format_issue_body(self, feedback: FeedbackSubmission, analysis: Dict) -> str:
        """格式化Issue正文"""
        return f"""
## 测试反馈详情

**反馈来源**: {feedback.source}  
**提交时间**: {feedback.submitted_at.strftime('%Y-%m-%d %H:%M')}  
**测试人**: {feedback.tester_name}  
**总体评分**: {feedback.overall_score}/10  
**推荐度**: {feedback.recommendation}

---

## AI分析

**问题分类**: {analysis['issue_category']}  
**优先级**: {analysis['priority']}  
**预计工作量**: {analysis.get('estimated_effort', '待评估')}

**根本原因**:
{analysis['root_cause']}

**修复建议**:
{analysis['fix_suggestion']}

**可能相关文件**:
{chr(10).join(f'- `{f}`' for f in analysis.get('related_files', []))}

---

## 详细问题

### 高优先级问题
{feedback.high_priority_issues or '无'}

### 中优先级问题
{feedback.medium_priority_issues or '无'}

### 低优先级问题
{feedback.low_priority_issues or '无'}

---

## 测试数据

**AI对话评分**:
- 专家匹配: {feedback.expert_matching_score}/5
- 回答质量: {feedback.answer_quality_score}/5
- 上下文理解: {feedback.context_understanding}
- 响应速度: {feedback.response_speed}

**文档评分**:
- 可读性: {feedback.doc_readability_score}/5

---

**原始反馈ID**: #{feedback.id}
        """
    
    async def _generate_summary_report(self, feedbacks: List[FeedbackSubmission]):
        """生成汇总报告"""
        # 统计数据
        # 生成趋势分析
        # 发送给团队
        pass

5. 自动Issue创建

src/feedback/automation/issue_creator.py

"""
自动创建GitHub/Gitea Issue
"""

import aiohttp
from typing import List
from loguru import logger


class IssueCreator:
    """Issue自动创建器"""
    
    def __init__(self):
        # 从环境变量读取配置
        import os
        self.git_platform = os.getenv("GIT_PLATFORM", "gitea")  # gitea or github
        self.api_url = os.getenv("GIT_API_URL", "http://localhost:3000/api/v1")
        self.repo = os.getenv("GIT_REPO", "zenglx01/mises-behavior-engine")
        self.token = os.getenv("GIT_TOKEN")
    
    async def create_issue(
        self, 
        title: str, 
        body: str, 
        labels: List[str] = None,
        assignees: List[str] = None
    ) -> str:
        """创建Issue并返回URL"""
        try:
            url = f"{self.api_url}/repos/{self.repo}/issues"
            
            headers = {
                "Authorization": f"token {self.token}",
                "Content-Type": "application/json"
            }
            
            data = {
                "title": title,
                "body": body,
                "labels": labels or [],
                "assignees": assignees or []
            }
            
            async with aiohttp.ClientSession() as session:
                async with session.post(url, headers=headers, json=data) as response:
                    if response.status == 201:
                        result = await response.json()
                        issue_url = result.get("html_url")
                        logger.info(f"✅ Issue创建成功: {issue_url}")
                        return issue_url
                    else:
                        logger.error(f"Issue创建失败: {response.status}")
                        return None
                        
        except Exception as e:
            logger.error(f"Issue创建失败: {e}")
            return None

📝 配置文件

config/feedback_sync.yaml

# 反馈同步配置

sync:
  enabled: true
  interval: 300  # 秒
  
sources:
  tencent_docs:
    enabled: true
    form_id: "${TENCENT_FORM_ID}"
    method: "excel_export"  # excel_export 或 webhook
    export_path: "/data/feedback/latest_export.xlsx"
    
  google_forms:
    enabled: false
    form_id: "${GOOGLE_FORM_ID}"
    credentials_file: "/secrets/google-credentials.json"
    
  slack:
    enabled: true
    channel: "#dev-testing"
    bot_token: "${SLACK_BOT_TOKEN}"

analysis:
  enabled: true
  auto_analyze: true
  llm_model: "deepseek-chat"
  
automation:
  auto_create_issue: true
  issue_priority_threshold: "high"  # critical, high, medium, low
  git_platform: "gitea"  # gitea 或 github
  git_api_url: "http://localhost:3000/api/v1"
  git_repo: "zenglx01/mises-behavior-engine"
  git_token: "${GIT_TOKEN}"

titans:
  store_in_memory: true
  memory_expert: "测试反馈分析专家"

🚀 部署和使用

1. 数据库迁移

# 创建反馈表
alembic revision --autogenerate -m "add feedback tables"
alembic upgrade head

2. 配置环境变量

# .env 文件
TENCENT_FORM_ID=your_form_id
GOOGLE_FORM_ID=your_form_id
SLACK_BOT_TOKEN=your_slack_token
GIT_PLATFORM=gitea
GIT_API_URL=http://localhost:3000/api/v1
GIT_REPO=zenglx01/mises-behavior-engine
GIT_TOKEN=your_git_token

3. 启动同步服务

方法A: 作为独立服务

# scripts/start_feedback_sync.py

import asyncio
from src.feedback.sync_service import FeedbackSyncService

async def main():
    service = FeedbackSyncService()
    await service.start()

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

运行:

python scripts/start_feedback_sync.py

方法B: 集成到主应用

# src/main.py

from src.feedback.sync_service import FeedbackSyncService

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时
    logger.info("🚀 启动反馈同步服务...")
    sync_service = FeedbackSyncService()
    sync_task = asyncio.create_task(sync_service.start())
    
    yield
    
    # 关闭时
    sync_task.cancel()
    logger.info("✅ 反馈同步服务已停止")

app = FastAPI(lifespan=lifespan)

4. 添加API端点（Webhook接收）

# src/api/feedback.py

from fastapi import APIRouter, Request
from src.feedback.integrations.tencent_docs import TencentDocsWebhook

router = APIRouter(prefix="/api/feedback", tags=["feedback"])

@router.post("/webhook/tencent")
async def tencent_webhook(request: Request):
    """接收腾讯文档Webhook推送"""
    data = await request.json()
    await TencentDocsWebhook.handle_submission(data)
    return {"status": "ok"}

@router.get("/submissions")
async def get_submissions(limit: int = 50):
    """获取反馈列表"""
    async with AsyncSessionLocal() as session:
        result = await session.execute(
            select(FeedbackSubmission)
            .order_by(FeedbackSubmission.submitted_at.desc())
            .limit(limit)
        )
        submissions = result.scalars().all()
        return submissions

@router.get("/analysis/summary")
async def get_analysis_summary():
    """获取分析汇总"""
    # 返回统计数据、趋势图等
    pass

📊 使用流程

完整自动化流程

1. 用户填写在线表单
   ↓
2. 数据同步服务（每5分钟）
   - 从腾讯文档/Google Forms拉取数据
   - 或通过Webhook实时接收
   ↓
3. 存入PostgreSQL数据库
   - 去重检查
   - 存储原始数据
   ↓
4. TITANS记忆系统
   - 向量化存储
   - 建立语义索引
   ↓
5. AI自动分析
   - 问题分类
   - 优先级评估
   - 根因分析
   - 修复建议
   ↓
6. 自动创建Issue（高优先级）
   - 生成Issue标题和正文
   - 自动打标签
   - 关联代码文件
   ↓
7. 开发人员（或AI）修复
   - 查看Issue
   - 修复代码
   - 提交PR
   ↓
8. 自动验证
   - 运行测试
   - 更新Issue状态
   ↓
9. 反馈给测试人员
   - 发送通知
   - 请求复测

🎯 Phase 2: 高级功能（未来）

1. AI自动修复

# src/feedback/automation/auto_fixer.py

class AutoFixer:
    """AI自动修复引擎（与Cursor集成）"""
    
    async def attempt_fix(self, issue: FeedbackSubmission):
        """尝试自动修复问题"""
        
        # 1. 分析问题
        analysis = await self._analyze_issue(issue)
        
        # 2. 定位代码
        files = await self._locate_code(analysis)
        
        # 3. 生成修复代码
        fix_code = await self._generate_fix(files, analysis)
        
        # 4. 创建PR分支
        branch = await self._create_branch(f"auto-fix-{issue.id}")
        
        # 5. 应用修复
        await self._apply_fix(branch, fix_code)
        
        # 6. 运行测试
        test_result = await self._run_tests()
        
        # 7. 创建PR（待人工审核）
        if test_result.passed:
            pr_url = await self._create_pr(
                title=f"[Auto-Fix] {issue.summary}",
                description=self._format_pr_description(issue, analysis),
                labels=["auto-generated", "needs-review"]
            )
            return pr_url
        else:
            return None

2. 反馈趋势分析

class TrendAnalyzer:
    """反馈趋势分析"""
    
    async def generate_trend_report(self):
        """生成趋势报告"""
        
        # 分析指标:
        # - 平均评分变化
        # - 问题类型分布
        # - 响应时间趋势
        # - 用户满意度
        # - 修复效率
        
        pass

3. 智能测试用例生成

class TestCaseGenerator:
    """基于反馈自动生成测试用例"""
    
    async def generate_from_feedback(self, issue: FeedbackSubmission):
        """根据用户反馈生成测试用例"""
        
        # 分析用户描述的问题
        # 生成重现步骤
        # 创建自动化测试脚本
        # 添加到测试套件
        
        pass

📋 总结

立即可实现（Phase 1）

✅ 数据同步服务

• 定时从在线表单拉取数据
• 存入PostgreSQL数据库
• 去重和增量同步

✅ TITANS记忆集成

• 所有反馈向量化存储
• 语义搜索相似问题
• 历史模式识别

✅ AI自动分析

• 问题自动分类
• 优先级智能评估
• 根因分析
• 修复建议生成

✅ 自动Issue创建

• 高优先级问题自动建Issue
• 自动打标签和分配
• 关联代码文件

预计工作量

Phase 1 实现:
- 数据同步服务: 4小时
- 数据模型设计: 2小时
- AI分析引擎: 6小时
- Issue自动创建: 2小时
- 测试和调试: 4小时

总计: 约18小时（2-3天）

价值

🎯 完全自动化的反馈处理循环 🚀 零人工干预的问题追踪 💡 AI驱动的持续改进 📊 实时反馈数据分析 🔮 预测性问题识别

这就是AI驱动开发的未来！ 🚀

文档版本: v1.0
创建时间: 2026-01-28
预计实现时间: 2-3天