175 lines
4.8 KiB
Python
Executable File
175 lines
4.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
微信自动回复机器人
|
|
|
|
功能：
|
|
- 接收用户消息
|
|
- 自动回复
|
|
- 处理 bbav20311 相关命令
|
|
"""
|
|
|
|
import json
|
|
import subprocess
|
|
import select
|
|
import time
|
|
import sys
|
|
|
|
YOUR_USER_ID = "o9cq801A7b5SQiVmMTKpOFN5DX4g@im.wechat"
|
|
|
|
|
|
def call_mcp_tool(tool_name: str, params: dict):
|
|
"""调用 MCP 工具"""
|
|
mcp_request = {
|
|
"jsonrpc": "2.0",
|
|
"method": "tools/call",
|
|
"params": {
|
|
"name": tool_name,
|
|
"arguments": params
|
|
},
|
|
"id": 1
|
|
}
|
|
|
|
proc = subprocess.Popen(
|
|
["bunx", "mcp-wechat-server"],
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True
|
|
)
|
|
|
|
request_json = json.dumps(mcp_request) + "\n"
|
|
proc.stdin.write(request_json)
|
|
proc.stdin.flush()
|
|
proc.stdin.close()
|
|
|
|
ready, _, _ = select.select([proc.stdout], [], [], 30)
|
|
|
|
if ready:
|
|
response_line = proc.stdout.readline()
|
|
proc.terminate()
|
|
try:
|
|
response = json.loads(response_line)
|
|
if "result" in response:
|
|
result = response["result"]
|
|
if "content" in result and len(result["content"]) > 0:
|
|
return result["content"][0].get("text", "")
|
|
except:
|
|
pass
|
|
else:
|
|
proc.terminate()
|
|
return None
|
|
|
|
|
|
def send_message(to_user: str, text: str):
|
|
"""发送消息"""
|
|
return call_mcp_tool("send_text_message", {
|
|
"to": to_user,
|
|
"text": text
|
|
})
|
|
|
|
|
|
def get_messages():
|
|
"""获取消息"""
|
|
return call_mcp_tool("get_messages", {})
|
|
|
|
|
|
def handle_message(msg: dict):
|
|
"""处理收到的消息"""
|
|
from_user = msg.get("from_user_id", "")
|
|
text = msg.get("text", "").strip()
|
|
|
|
print(f"📨 收到消息 [{from_user}]: {text}")
|
|
|
|
# 只处理你的消息
|
|
if from_user != YOUR_USER_ID:
|
|
print(f" 忽略非目标用户消息")
|
|
return
|
|
|
|
# 处理命令
|
|
if text == "你是谁":
|
|
reply = """🤖 我是 bbav20311 基因家族分析助手
|
|
|
|
我可以帮你：
|
|
• 运行基因家族分析任务
|
|
• 查询任务状态
|
|
• 发送任务完成通知
|
|
• 回答分析相关问题
|
|
|
|
发送"帮助"查看可用命令。"""
|
|
send_message(from_user, reply)
|
|
|
|
elif text in ["帮助", "help", "?"]:
|
|
reply = """📋 可用命令：
|
|
|
|
🧬 分析任务：
|
|
• "分析 Ft.SPL" - 运行基因家族鉴定
|
|
• "状态" - 查询系统状态
|
|
• "任务列表" - 查看进行中的任务
|
|
|
|
💬 其他：
|
|
• "你是谁" - 查看我的介绍
|
|
• "帮助" - 显示此帮助
|
|
|
|
直接发送基因家族名称即可开始分析！"""
|
|
send_message(from_user, reply)
|
|
|
|
elif text == "状态":
|
|
reply = "📊 系统状态：\n\n✅ 微信服务：在线\n✅ 后端服务：运行中\n📋 活跃任务：0 个\n\n系统运行正常，可以接收分析请求。"
|
|
send_message(from_user, reply)
|
|
|
|
elif "分析" in text or " Ft." in text or text.startswith("Ft."):
|
|
# 提取基因家族名称
|
|
import re
|
|
match = re.search(r'Ft\.[A-Z]+', text)
|
|
if match:
|
|
gene_family = match.group(0)
|
|
reply = f"🚀 收到分析请求：{gene_family}\n\n任务已创建，正在处理中...\n完成后将通知您。"
|
|
send_message(from_user, reply)
|
|
# TODO: 实际调用 bbav20311 分析任务
|
|
else:
|
|
reply = "❓ 无法识别基因家族名称，请使用格式如：Ft.SPL、Ft.WRKY"
|
|
send_message(from_user, reply)
|
|
else:
|
|
reply = f"🤖 收到：{text}\n\n发送'帮助'查看可用命令，或直接发送基因家族名称（如 Ft.SPL）开始分析。"
|
|
send_message(from_user, reply)
|
|
|
|
|
|
def main():
|
|
"""主循环"""
|
|
print("="*60)
|
|
print("🤖 bbav20311 微信机器人启动")
|
|
print("="*60)
|
|
print(f"监听用户: {YOUR_USER_ID}")
|
|
print("按 Ctrl+C 停止\n")
|
|
|
|
last_message_id = 0
|
|
|
|
try:
|
|
while True:
|
|
result = get_messages()
|
|
|
|
if result:
|
|
try:
|
|
data = json.loads(result)
|
|
messages = data.get("messages", [])
|
|
|
|
for msg in messages:
|
|
msg_id = msg.get("message_id", 0)
|
|
|
|
# 只处理新消息
|
|
if msg_id > last_message_id:
|
|
last_message_id = msg_id
|
|
handle_message(msg)
|
|
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
time.sleep(2)
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n\n🛑 机器人已停止")
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |