PlantAI-dev/wechat_task_bridge.py

#!/usr/bin/env python3
"""
WeChat Task Bridge - 微信与 bbav20311 任务系统的桥接

功能:
1. 用户从微信发送分析请求
2. 自动创建并执行任务
3. 任务完成后微信通知用户

使用:
    python wechat_task_bridge.py

作者: xialab的幸运星
日期: 2026-04-10
"""

import asyncio
import json
import re
import sys
from pathlib import Path
from typing import Dict, Optional
from datetime import datetime

# 添加项目路径
PROJECT_ROOT = Path(__file__).parent
sys.path.insert(0, str(PROJECT_ROOT))

from wechat_integration import WeChatIntegration, WeChatTaskNotifier

# 导入 bbav20311 的任务模块
try:
    from apps.web.pipeline_backend import new_job, run_stage
    from gfa_core.agents.router import route, is_analysis_task
    BACKEND_AVAILABLE = True
except ImportError as e:
    print(f"⚠️  无法导入后端模块: {e}")
    print("   将以模拟模式运行")
    BACKEND_AVAILABLE = False


class WeChatTaskBridge:
    """
    微信任务桥接器
    
    将微信消息转换为 bbav20311 分析任务
    """
    
    def __init__(self):
        self.wechat = WeChatIntegration()
        self.active_tasks: Dict[str, dict] = {}  # 存储活跃任务
        self.user_sessions: Dict[str, dict] = {}   # 用户会话状态
        
    async def start(self):
        """启动桥接服务"""
        print("🚀 启动 WeChat Task Bridge")
        print("="*60)
        
        # 1. 登录微信
        print("\n🔐 正在登录微信...")
        success = await self.wechat.login(timeout=300)
        if not success:
            print("❌ 微信登录失败，服务无法启动")
            return
        
        print("✅ 微信登录成功")
        
        # 2. 启动消息监听
        print("\n🎧 启动消息监听...")
        print("   用户现在可以通过微信发送分析请求")
        print("   按 Ctrl+C 停止服务\n")
        
        await self.wechat.start_message_listener(
            callback=self._handle_message,
            poll_interval=5
        )
    
    async def _handle_message(self, msg: dict):
        """处理收到的微信消息"""
        user_id = msg.get("from")
        content = msg.get("content", "").strip()
        
        print(f"\n📨 收到消息 [{user_id}]: {content}")
        
        # 显示正在输入状态
        await self.wechat.send_typing(user_id)
        
        # 解析命令
        response = await self._parse_command(user_id, content)
        
        # 发送回复
        if response:
            await self.wechat.send_text_message(user_id, response)
    
    async def _parse_command(self, user_id: str, content: str) -> str:
        """解析用户命令并返回回复"""
        content_lower = content.lower()
        
        # 1. 帮助命令
        if content in ["帮助", "help", "?", "菜单"]:
            return self._get_help_message()
        
        # 2. 状态查询
        if content in ["状态", "status", "系统状态"]:
            return await self._get_system_status()
        
        # 3. 任务列表
        if content in ["任务", "任务列表", "tasks"]:
            return self._get_task_list(user_id)
        
        # 4. 任务详情
        detail_match = re.match(r"详情\s+(\w+)", content)
        if detail_match:
            task_id = detail_match.group(1)
            return self._get_task_detail(task_id)
        
        # 5. 取消任务
        cancel_match = re.match(r"取消\s+(\w+)", content)
        if cancel_match:
            task_id = cancel_match.group(1)
            return await self._cancel_task(user_id, task_id)
        
        # 6. 分析请求 - 尝试解析为基因家族分析任务
        if BACKEND_AVAILABLE:
            analysis_response = await self._handle_analysis_request(user_id, content)
            if analysis_response:
                return analysis_response
        
        # 7. 未知命令
        return f"""❓ 无法理解的命令: "{content}"

请使用以下格式：
• 帮助 - 显示帮助信息
• 状态 - 查询系统状态
• 任务列表 - 查看任务
• 详情 <任务ID> - 查看详情
• 或直接发送分析需求，如："Ft.SPL 基因家族鉴定"

发送"帮助"查看完整命令列表。"""
    
    def _get_help_message(self) -> str:
        """获取帮助信息"""
        return """🤖 bbav20311 基因家族分析助手

📋 可用命令:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━
• 帮助 / help - 显示此帮助
• 状态 / status - 查询系统状态
• 任务 / 任务列表 - 查看进行中的任务
• 详情 <任务ID> - 查看任务详情
• 取消 <任务ID> - 取消任务

🧬 分析任务:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━
直接发送分析需求，例如：
• "Ft.SPL 基因家族鉴定"
• "分析 Ft.WRKY 的进化树"
• "Ft.MYB motif分析"

📊 支持的 stage:
• identify - 基因家族鉴定
• phylogeny - 进化树构建
• motif - Motif分析
• domain - 结构域分析
• promoter - 启动子分析
• synteny - 共线性分析
• expression - 表达分析
• combo - 综合分析

💡 提示:
• 任务完成后会自动通知您
• 大任务可能需要几分钟到几小时
• 可随时查询任务状态"""
    
    async def _get_system_status(self) -> str:
        """获取系统状态"""
        if not BACKEND_AVAILABLE:
            return "⚠️ 后端服务未连接（模拟模式）"
        
        # 检查后端服务状态
        active_count = len(self.active_tasks)
        
        return f"""📊 系统状态

✅ 微信服务: 在线
✅ 后端服务: 运行中
📋 活跃任务: {active_count} 个

系统运行正常，可以接收分析请求。"""
    
    def _get_task_list(self, user_id: str) -> str:
        """获取用户的任务列表"""
        user_tasks = [
            tid for tid, task in self.active_tasks.items()
            if task.get("user_id") == user_id
        ]
        
        if not user_tasks:
            return "📭 您当前没有进行中的任务"
        
        lines = [f"📋 您的任务列表 ({len(user_tasks)}个):"]
        for tid in user_tasks[-5:]:  # 只显示最近5个
            task = self.active_tasks[tid]
            status_emoji = {
                "pending": "⏳",
                "running": "🔄",
                "completed": "✅",
                "failed": "❌"
            }.get(task.get("status"), "❓")
            
            lines.append(f"{status_emoji} {tid} - {task.get('task_type', '未知')}")
        
        lines.append("\n发送'详情 <任务ID>'查看详细信息")
        return "\n".join(lines)
    
    def _get_task_detail(self, task_id: str) -> str:
        """获取任务详情"""
        task = self.active_tasks.get(task_id)
        if not task:
            return f"❌ 未找到任务: {task_id}"
        
        status_emoji = {
            "pending": "⏳",
            "running": "🔄",
            "completed": "✅",
            "failed": "❌"
        }.get(task.get("status"), "❓")
        
        lines = [
            f"📋 任务详情: {task_id}",
            f"状态: {status_emoji} {task.get('status', '未知')}",
            f"类型: {task.get('task_type', '未知')}",
            f"创建时间: {task.get('created_at', '未知')}",
        ]
        
        if task.get("started_at"):
            lines.append(f"开始时间: {task['started_at']}")
        
        if task.get("completed_at"):
            lines.append(f"完成时间: {task['completed_at']}")
        
        if task.get("error"):
            lines.append(f"错误信息: {task['error']}")
        
        if task.get("output_files"):
            lines.append(f"\n输出文件:")
            for f in task["output_files"][:5]:
                lines.append(f"  • {f}")
        
        return "\n".join(lines)
    
    async def _cancel_task(self, user_id: str, task_id: str) -> str:
        """取消任务"""
        task = self.active_tasks.get(task_id)
        if not task:
            return f"❌ 未找到任务: {task_id}"
        
        if task.get("user_id") != user_id:
            return "❌ 您无权取消此任务"
        
        if task.get("status") in ["completed", "failed"]:
            return "❌ 任务已完成或已失败，无法取消"
        
        # 取消任务逻辑
        task["status"] = "cancelled"
        return f"✅ 任务 {task_id} 已取消"
    
    async def _handle_analysis_request(self, user_id: str, content: str) -> Optional[str]:
        """
        处理分析请求
        
        使用 bbav20311 的意图识别和任务系统
        """
        try:
            # 使用 router 进行意图识别
            intent = route(content)
            
            if not is_analysis_task(intent):
                return None  # 不是分析任务
            
            # 提取参数
            prefix = intent.prefix
            stage = intent.task_type if intent.task_type != "pipeline" else "full"
            
            if not prefix:
                return "❌ 无法识别基因家族前缀，请使用格式如：Ft.SPL 基因家族"
            
            # 创建任务
            task_id = f"wx_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{user_id[:8]}"
            
            task_info = {
                "task_id": task_id,
                "user_id": user_id,
                "task_type": stage,
                "prefix": prefix,
                "status": "pending",
                "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                "parameters": intent.parameters
            }
            
            self.active_tasks[task_id] = task_info
            
            # 启动异步任务执行
            asyncio.create_task(self._execute_task(task_info))
            
            return f"""✅ 分析任务已创建

📋 任务ID: {task_id}
🧬 基因家族: {prefix}
🔬 分析类型: {stage}
⏳ 状态: 排队中

任务完成后将通过微信通知您。
发送"详情 {task_id}"查看进度。"""
            
        except Exception as e:
            print(f"处理分析请求失败: {e}")
            return None
    
    async def _execute_task(self, task_info: dict):
        """执行任务"""
        task_id = task_info["task_id"]
        user_id = task_info["user_id"]
        
        # 创建通知器
        notifier = WeChatTaskNotifier(user_wechat_id=user_id, wechat=self.wechat)
        
        try:
            # 更新状态
            task_info["status"] = "running"
            task_info["started_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            
            # 通知开始
            await notifier.notify_task_started(
                task_id=task_id,
                task_type=task_info["task_type"]
            )
            
            if BACKEND_AVAILABLE:
                # 创建 bbav20311 job
                job = new_job()
                
                # 运行任务
                # 这里需要根据 task_type 调用相应的 stage
                stage = task_info["task_type"]
                prefix = task_info["prefix"]
                
                print(f"执行任务: {task_id}, stage={stage}, prefix={prefix}")
                
                # 模拟任务执行（实际实现需要调用 pipeline_backend）
                await asyncio.sleep(5)  # 模拟执行时间
                
                # 更新任务信息
                task_info["status"] = "completed"
                task_info["completed_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                task_info["output_files"] = [
                    f"{prefix}.{stage}.result.txt",
                    f"{prefix}.{stage}.report.pdf"
                ]
                
                # 通知完成
                await notifier.notify_task_completed(
                    task_id=task_id,
                    task_type=task_info["task_type"],
                    output_files=task_info["output_files"]
                )
            else:
                # 模拟模式
                await asyncio.sleep(3)
                task_info["status"] = "completed"
                task_info["completed_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                task_info["output_files"] = ["simulated_result.txt"]
                
                await notifier.notify_task_completed(
                    task_id=task_id,
                    task_type=task_info["task_type"],
                    output_files=task_info["output_files"]
                )
                
        except Exception as e:
            task_info["status"] = "failed"
            task_info["error"] = str(e)
            task_info["completed_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            
            await notifier.notify_task_failed(
                task_id=task_id,
                task_type=task_info["task_type"],
                error=str(e)
            )


# ==================== 测试 ====================

async def test_bridge():
    """测试桥接器"""
    print("="*60)
    print("🧪 WeChat Task Bridge 测试")
    print("="*60)
    
    bridge = WeChatTaskBridge()
    
    # 模拟消息处理
    test_cases = [
        ("user_001", "帮助"),
        ("user_001", "状态"),
        ("user_001", "Ft.SPL 基因家族鉴定"),
        ("user_001", "任务列表"),
    ]
    
    for user_id, content in test_cases:
        print(f"\n[Test] 用户: {user_id}, 消息: {content}")
        response = await bridge._parse_command(user_id, content)
        print(f"[Response] {response[:200]}...")
        await asyncio.sleep(1)
    
    print("\n" + "="*60)
    print("✅ 测试完成")
    print("="*60)


if __name__ == "__main__":
    import sys
    
    if len(sys.argv) > 1 and sys.argv[1] == "test":
        # 运行测试模式
        asyncio.run(test_bridge())
    else:
        # 运行服务
        bridge = WeChatTaskBridge()
        try:
            asyncio.run(bridge.start())
        except KeyboardInterrupt:
            print("\n\n🛑 服务已停止")