360 lines
14 KiB
Python
360 lines
14 KiB
Python
"""
|
||
Phase 4: 个人微信 (iLink 协议 / 微信 ClawBot 插件) —— 前台用户模块
|
||
im_bot_user.py
|
||
|
||
为【平台注册用户】提供"各自绑定自己的个人微信"能力：
|
||
- 每个用户独立一条 im_bot_wechat_account 记录（按 user_id 隔离）
|
||
- 扫码登录免 AppID；消息收发走 Bearer bot_token + X-WECHAT-UIN
|
||
- 绑定后由七虎AI 大脑自动应答（复用 channels 的真实 LLM 调用）
|
||
|
||
与后台 im_bot_admin 的区别：
|
||
- 后台：平台级 IM Bot 接入中枢（飞书/企微/Telegram 等多平台、路由、统计），其个人微信
|
||
端点仅限 user_id IS NULL 的"全局 Bot"，与用户绑定互不干扰。
|
||
- 前台：每个登录用户管理自己的个人微信，数据严格按 user_id 隔离。
|
||
"""
|
||
import base64
|
||
import json
|
||
import random
|
||
import threading
|
||
import time
|
||
from datetime import datetime
|
||
from typing import Optional
|
||
|
||
import requests
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..auth import get_current_user
|
||
from ..database import SessionLocal
|
||
from ..models import User
|
||
|
||
router = APIRouter()
|
||
|
||
# ============================================================
|
||
# iLink 协议常量（腾讯官方 iLink Bot HTTP API，直连免 AppID）
|
||
# ============================================================
|
||
ILINK_BASE = "https://ilinkai.weixin.qq.com"
|
||
|
||
# 每个已连接账号一个后台接收线程，key = 账号表主键 id
|
||
_wechat_receiver_threads: dict[int, threading.Thread] = {}
|
||
|
||
|
||
# ============================================================
|
||
# 通用 DB 工具
|
||
# ============================================================
|
||
def _execute_raw(db: Session, sql_text, params=None):
|
||
result = db.execute(sql_text, params or {})
|
||
if result.returns_rows:
|
||
return result.fetchall()
|
||
db.commit()
|
||
return result
|
||
|
||
|
||
def _row_to_dict(row) -> dict:
|
||
if row is None:
|
||
return None
|
||
d = {}
|
||
for col in row._mapping.keys():
|
||
val = row._mapping[col]
|
||
if isinstance(val, datetime):
|
||
val = val.isoformat()
|
||
d[col] = val
|
||
return d
|
||
|
||
|
||
# ============================================================
|
||
# iLink 协议请求工具
|
||
# ============================================================
|
||
def _wechat_headers(bot_token: str) -> dict:
|
||
"""iLink 协议请求头：Bearer bot_token + 随机 X-WECHAT-UIN 防重放"""
|
||
uin = base64.b64encode(str(random.randint(0, 2 ** 32 - 1)).encode("utf-8")).decode("utf-8")
|
||
return {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {bot_token}",
|
||
"AuthorizationType": "ilink_bot_token",
|
||
"X-WECHAT-UIN": uin,
|
||
}
|
||
|
||
|
||
def _wechat_row_for_user(db: Session, user_id: str):
|
||
rows = _execute_raw(
|
||
db,
|
||
text("SELECT * FROM im_bot_wechat_account WHERE user_id = :uid ORDER BY id DESC LIMIT 1"),
|
||
{"uid": user_id},
|
||
)
|
||
return rows[0] if rows else None
|
||
|
||
|
||
def _wechat_mask(row) -> dict:
|
||
d = _row_to_dict(row)
|
||
if d.get("bot_token"):
|
||
bt = d["bot_token"]
|
||
d["bot_token"] = ("***" + bt[-6:]) if len(bt) > 6 else "***"
|
||
d.pop("appid", None)
|
||
d.pop("secret_key", None)
|
||
return d
|
||
|
||
|
||
# ============================================================
|
||
# 七虎AI 大脑（复用 channels 的真实 LLM 调用）
|
||
# ============================================================
|
||
def _qihua_reply(text: str) -> str:
|
||
"""调用七虎AI 生成回复。自行创建 DB 会话，避免上层线程无可用 session。"""
|
||
db = None
|
||
try:
|
||
from app.api.channels import _call_real_llm
|
||
db = SessionLocal()
|
||
return _call_real_llm(db, None, text, "个人微信")
|
||
except Exception as e:
|
||
return f"（AI 回复生成失败：{e}）"
|
||
finally:
|
||
if db:
|
||
db.close()
|
||
|
||
|
||
# ============================================================
|
||
# iLink 消息收发
|
||
# ============================================================
|
||
def _wechat_get_updates(baseurl: str, bot_token: str, cursor: str):
|
||
"""长轮询收消息，返回 (msgs, new_cursor)。最多 hold 35s。"""
|
||
try:
|
||
resp = requests.post(
|
||
baseurl.rstrip("/") + "/ilink/bot/getupdates",
|
||
headers=_wechat_headers(bot_token),
|
||
json={"get_updates_buf": cursor, "base_info": {"channel_version": "1.0.2"}},
|
||
timeout=40,
|
||
)
|
||
if not resp.ok:
|
||
return [], cursor
|
||
j = resp.json() if resp.content else {}
|
||
return j.get("msgs", []) or [], j.get("get_updates_buf", cursor)
|
||
except Exception:
|
||
return [], cursor
|
||
|
||
|
||
def _wechat_send_message(baseurl: str, bot_token: str, to_user_id: str, context_token: str, text: str) -> bool:
|
||
"""发送文本回复（必须带 context_token 关联到正确的对话窗口）"""
|
||
try:
|
||
resp = requests.post(
|
||
baseurl.rstrip("/") + "/ilink/bot/sendmessage",
|
||
headers=_wechat_headers(bot_token),
|
||
json={"msg": {
|
||
"to_user_id": to_user_id,
|
||
"message_type": 2,
|
||
"message_state": 2,
|
||
"context_token": context_token,
|
||
"item_list": [{"type": 1, "text_item": {"text": text}}],
|
||
}},
|
||
timeout=20,
|
||
)
|
||
return resp.ok
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def _wechat_extract_text(msg: dict) -> str:
|
||
parts = []
|
||
for it in (msg.get("item_list") or []):
|
||
if it.get("type") == 1 and it.get("text_item", {}).get("text"):
|
||
parts.append(it["text_item"]["text"])
|
||
return "\n".join(parts)
|
||
|
||
|
||
# ============================================================
|
||
# 后台接收线程（每用户独立）
|
||
# ============================================================
|
||
def _wechat_receiver_loop(account_id: int):
|
||
"""长轮询个人微信消息 → 七虎AI 回复 → 发送。"""
|
||
cursor = ""
|
||
while True:
|
||
db = SessionLocal()
|
||
try:
|
||
row = _execute_raw(
|
||
db,
|
||
text("SELECT * FROM im_bot_wechat_account WHERE id = :aid"),
|
||
{"aid": account_id},
|
||
)
|
||
if not row:
|
||
return
|
||
d = _row_to_dict(row[0])
|
||
if d.get("status") != "connected" or not d.get("bot_token"):
|
||
return
|
||
bot_token = d["bot_token"]
|
||
baseurl = d.get("baseurl") or ILINK_BASE
|
||
msgs, cursor = _wechat_get_updates(baseurl, bot_token, cursor)
|
||
for m in msgs:
|
||
if m.get("message_type") not in (1, None):
|
||
continue # 只处理用户消息（2=bot 自己发出的）
|
||
content = _wechat_extract_text(m)
|
||
if not content:
|
||
continue
|
||
reply = _qihua_reply(content)
|
||
_wechat_send_message(
|
||
baseurl, bot_token,
|
||
m.get("from_user_id", ""),
|
||
m.get("context_token", ""),
|
||
reply,
|
||
)
|
||
time.sleep(1)
|
||
except Exception:
|
||
time.sleep(5)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def _start_wechat_receiver(account_id: int):
|
||
if account_id in _wechat_receiver_threads and _wechat_receiver_threads[account_id].is_alive():
|
||
return
|
||
t = threading.Thread(target=_wechat_receiver_loop, args=(account_id,), daemon=True)
|
||
t.start()
|
||
_wechat_receiver_threads[account_id] = t
|
||
|
||
|
||
# ============================================================
|
||
# 前台用户接口（按 user_id 隔离）
|
||
# ============================================================
|
||
@router.get("/status")
|
||
async def wechat_status(user: User = Depends(get_current_user)):
|
||
"""当前登录用户的个人微信连接状态"""
|
||
db = SessionLocal()
|
||
try:
|
||
row = _wechat_row_for_user(db, user.id)
|
||
if not row:
|
||
return {"success": True, "data": {"connected": False, "configured": False}}
|
||
d = _row_to_dict(row)
|
||
return {"success": True, "data": {
|
||
"connected": d.get("status") == "connected",
|
||
"configured": True,
|
||
"baseurl": d.get("baseurl"),
|
||
"ilink_bot_id": d.get("ilink_bot_id"),
|
||
"username": d.get("username"),
|
||
"connected_at": str(d.get("connected_at")) if d.get("connected_at") else None,
|
||
}}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@router.post("/qrcode")
|
||
async def wechat_qrcode(user: User = Depends(get_current_user)):
|
||
"""获取个人微信扫码登录二维码（iLink 协议，免 AppID）"""
|
||
try:
|
||
resp = requests.get(
|
||
ILINK_BASE + "/ilink/bot/get_bot_qrcode",
|
||
params={"bot_type": 3},
|
||
timeout=20,
|
||
)
|
||
if not resp.ok:
|
||
return {"success": False, "error": f"获取二维码失败 HTTP {resp.status_code}"}
|
||
j = resp.json() if resp.content else {}
|
||
return {"success": True, "data": {
|
||
"qrcode": j.get("qrcode"),
|
||
"qrcode_img_content": j.get("qrcode_img_content"),
|
||
}}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
@router.post("/qrcode/status")
|
||
async def wechat_qrcode_status(
|
||
qrcode: str = Query(...),
|
||
user: User = Depends(get_current_user),
|
||
):
|
||
"""轮询扫码状态；confirmed 时保存该用户的 bot_token 并启动接收线程"""
|
||
try:
|
||
resp = requests.get(
|
||
ILINK_BASE + "/ilink/bot/get_qrcode_status",
|
||
params={"qrcode": qrcode},
|
||
timeout=40,
|
||
)
|
||
if not resp.ok:
|
||
return {"success": False, "error": f"查询状态失败 HTTP {resp.status_code}"}
|
||
j = resp.json() if resp.content else {}
|
||
data = j.get("data", j) if isinstance(j, dict) else j
|
||
status = data.get("status")
|
||
if status == "confirmed":
|
||
creds = data.get("credentials", {}) or {}
|
||
bot_token = data.get("bot_token") or creds.get("bot_token")
|
||
baseurl = data.get("baseurl") or creds.get("baseurl") or ILINK_BASE
|
||
ilink_bot_id = data.get("ilink_bot_id") or creds.get("ilink_bot_id")
|
||
db = SessionLocal()
|
||
try:
|
||
row = _wechat_row_for_user(db, user.id)
|
||
if not row:
|
||
_execute_raw(db, text("""INSERT INTO im_bot_wechat_account
|
||
(user_id, username, bot_token, ilink_bot_id, baseurl, status,
|
||
connected_at, created_at, updated_at)
|
||
VALUES (:user_id, :username, :bot_token, :ilink_bot_id, :baseurl,
|
||
'connected', :connected_at, :created_at, :updated_at)"""),
|
||
{
|
||
"user_id": user.id, "username": user.username,
|
||
"bot_token": bot_token, "ilink_bot_id": ilink_bot_id,
|
||
"baseurl": baseurl,
|
||
"connected_at": datetime.now(),
|
||
"created_at": datetime.now(), "updated_at": datetime.now(),
|
||
})
|
||
else:
|
||
_execute_raw(db, text("""UPDATE im_bot_wechat_account SET
|
||
user_id=:user_id, username=:username, bot_token=:bot_token,
|
||
ilink_bot_id=:ilink_bot_id, baseurl=:baseurl,
|
||
status='connected', connected_at=:connected_at, updated_at=:updated_at
|
||
WHERE id=:id"""),
|
||
{
|
||
"user_id": user.id, "username": user.username,
|
||
"bot_token": bot_token, "ilink_bot_id": ilink_bot_id,
|
||
"baseurl": baseurl, "connected_at": datetime.now(),
|
||
"updated_at": datetime.now(), "id": _row_to_dict(row)["id"],
|
||
})
|
||
row = _wechat_row_for_user(db, user.id)
|
||
aid = _row_to_dict(row)["id"] if row else 1
|
||
_start_wechat_receiver(aid)
|
||
finally:
|
||
db.close()
|
||
return {"success": True, "data": {"status": "confirmed", "connected": True}}
|
||
return {"success": True, "data": {"status": status, "connected": False}}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
@router.post("/disconnect")
|
||
async def wechat_disconnect(user: User = Depends(get_current_user)):
|
||
"""断开当前用户的个人微信连接（清除 bot_token）"""
|
||
db = SessionLocal()
|
||
try:
|
||
row = _wechat_row_for_user(db, user.id)
|
||
if row:
|
||
_execute_raw(db, text(
|
||
"UPDATE im_bot_wechat_account SET bot_token=NULL, status='disconnected', "
|
||
"updated_at=:updated_at WHERE id=:id"
|
||
), {"updated_at": datetime.now(), "id": _row_to_dict(row)["id"]})
|
||
return {"success": True, "message": "已断开个人微信连接"}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
# ============================================================
|
||
# 启动恢复：为所有已连接用户账号拉起接收线程
|
||
# ============================================================
|
||
def start_all_wechat_receivers():
|
||
"""在应用启动时调用，恢复此前已连接用户的个人微信接收线程。"""
|
||
db = SessionLocal()
|
||
try:
|
||
rows = _execute_raw(
|
||
db,
|
||
text("SELECT id FROM im_bot_wechat_account WHERE status='connected' AND user_id IS NOT NULL"),
|
||
)
|
||
for r in rows:
|
||
_start_wechat_receiver(r._mapping["id"])
|
||
if rows:
|
||
import logging
|
||
logging.getLogger("aiflow").info(f"[IM个人微信] 已恢复 {len(rows)} 个用户微信接收线程")
|
||
except Exception as e:
|
||
import logging
|
||
logging.getLogger("aiflow").warning(f"[IM个人微信] 启动恢复失败(不影响服务): {e}")
|
||
finally:
|
||
db.close()
|