433 lines
14 KiB
Python
Executable File
433 lines
14 KiB
Python
Executable File
#!/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🛑 服务已停止") |