QihuAI Design
29a1ce2e54
- 新增 app/core/capability_discovery.py：统一能力发现层，Web/IM 双Claw共用单一实现 （三级降级：用户安装 → 平台published → 相关性裁剪；MCP 自动挂载已启用服务） - im_llm.py：IM Bot Claw 改用统一发现层，恢复 skills(12) + MCP(26工具) 能力， 修复 #866 部署回退导致的 agents=[]/mcp=[] 回归；mcp__ 前缀分流到 call_mcp_tool - mcp_client.py：注入 httpx_client_factory(trust_env=False) 绕过容器 HTTP_PROXY， 修复外部 MCP(xiaobenyang.com / anysearch.com) tools/list 全部静默失败（mcp_tools 0→26） - llm_client.py / im_bot_user.py：2026-07-31 代理劫持内部请求 + 媒体断连修复（随镜像已上线）
724 lines
30 KiB
Python
724 lines
30 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 os
|
||
import random
|
||
import threading
|
||
import time
|
||
from datetime import datetime, timedelta, timezone
|
||
from typing import Optional
|
||
from urllib.parse import quote as _quote
|
||
|
||
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()
|
||
|
||
# 北京时间（UTC+8）。容器系统时区为 UTC，所有对外展示/持久化的时间戳必须按北京时间存储，
|
||
# 否则前台微信 Bot 对话窗会显示成 UTC（比系统时间早 8 小时）。
|
||
TZ_BEIJING = timezone(timedelta(hours=8))
|
||
|
||
|
||
def _beijing_now() -> datetime:
|
||
"""返回北京时间的 naive datetime（去掉 tzinfo 以便直接存入 MySQL DATETIME）。"""
|
||
return datetime.now(TZ_BEIJING).replace(tzinfo=None)
|
||
|
||
# ============================================================
|
||
# 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 _recent_history(user_id: str, limit: int = 6) -> str:
|
||
"""拼接最近若干轮对话，作为多轮上下文。"""
|
||
db = None
|
||
try:
|
||
db = SessionLocal()
|
||
rows = _execute_raw(db, text(
|
||
"SELECT role, content FROM im_bot_chat_log "
|
||
"WHERE user_id=:uid ORDER BY id DESC LIMIT :lim"
|
||
), {"uid": user_id, "lim": limit})
|
||
pairs = []
|
||
for r in reversed(rows):
|
||
dd = _row_to_dict(r)
|
||
role = "用户" if dd["role"] == "user" else "七虎AI"
|
||
pairs.append(f"{role}：{dd['content']}")
|
||
return "\n".join(pairs)
|
||
except Exception:
|
||
return ""
|
||
finally:
|
||
if db:
|
||
db.close()
|
||
|
||
|
||
def _qihua_reply(text: str, user_id: str = None) -> str:
|
||
"""调用七虎AI 生成回复。自行创建 DB 会话，避免上层线程无可用 session。"""
|
||
prompt = text
|
||
if user_id:
|
||
hist = _recent_history(user_id)
|
||
if hist:
|
||
prompt = (
|
||
"以下是你与该用户最近的对话历史，请结合上下文作答：\n"
|
||
f"{hist}\n\n"
|
||
f"用户当前问题：{text}"
|
||
)
|
||
# 磁盘调试日志（不依赖 uvicorn logger）
|
||
try:
|
||
with open("/tmp/im_llm_debug.log", "a", encoding="utf-8") as _f:
|
||
_f.write(_beijing_now().isoformat() + f" _qihua_reply user={user_id} prompt_len={len(prompt)} prompt={repr(prompt)[:500]}\n")
|
||
except Exception:
|
||
pass
|
||
db = None
|
||
try:
|
||
from app.api.im_llm import _call_real_llm
|
||
db = SessionLocal()
|
||
result = _call_real_llm(db, None, prompt, "个人微信", user_id=user_id)
|
||
try:
|
||
with open("/tmp/im_llm_debug.log", "a", encoding="utf-8") as _f:
|
||
_f.write(_beijing_now().isoformat() + f" _qihua_reply RESULT len={len(result) if result else 0} preview={repr(result)[:200]}\n")
|
||
except Exception:
|
||
pass
|
||
return result
|
||
except Exception as e:
|
||
import traceback as _tb2
|
||
import logging
|
||
_log = logging.getLogger('im_bot_user')
|
||
_log.warning(f'[_qihua_reply] 异常: {e}\n{_tb2.format_exc()[:1000]}')
|
||
try:
|
||
with open("/tmp/im_llm_debug.log", "a", encoding="utf-8") as _f:
|
||
_f.write(_beijing_now().isoformat() + f" _qihua_reply EXCEPTION: {e}\n{_tb2.format_exc()[:800]}\n")
|
||
except Exception:
|
||
pass
|
||
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},
|
||
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_post_msg(baseurl: str, bot_token: str, to_user_id: str, context_token: str, item_list: list):
|
||
"""向 iLink 发送一条消息（item_list 可含文本/图片/视频等）。返回 (ok, ret, body)。
|
||
ret=-14=会话过期需重扫；ret=-2=token 失效。"""
|
||
payload = {
|
||
"msg": {
|
||
"to_user_id": to_user_id,
|
||
"message_type": 2,
|
||
"message_state": 2,
|
||
"context_token": context_token,
|
||
"item_list": item_list,
|
||
}
|
||
}
|
||
last_body = {}
|
||
# 07-30 diagnostic
|
||
import json as _json3
|
||
try:
|
||
with open("/tmp/wechat_send_debug.log", "a", encoding="utf-8") as _f:
|
||
_f.write(_beijing_now().isoformat() + " SEND_DEBUG token=" + repr(bot_token[:20] if bot_token else "NULL")
|
||
+ " to=" + repr(to_user_id[:30]) + " ctx=" + repr(context_token[:30]) + "\n")
|
||
_f.write(" payload_keys=" + str(list(payload.get("msg", {}).keys())) + "\n")
|
||
except Exception:
|
||
pass
|
||
|
||
for _ in range(2):
|
||
try:
|
||
resp = requests.post(
|
||
baseurl.rstrip("/") + "/ilink/bot/sendmessage",
|
||
headers=_wechat_headers(bot_token),
|
||
json=payload,
|
||
timeout=20,
|
||
)
|
||
try:
|
||
body = resp.json() if resp.content else {}
|
||
except Exception:
|
||
body = {}
|
||
last_body = body
|
||
ret = body.get("ret") if isinstance(body, dict) else None
|
||
try:
|
||
with open("/tmp/wechat_send.log", "a", encoding="utf-8") as _f:
|
||
_f.write(_beijing_now().isoformat() + " SEND_HTTP=%s ret=%s body=%s\n"
|
||
% (resp.status_code, ret, json.dumps(body, ensure_ascii=False)[:300]))
|
||
except Exception:
|
||
pass
|
||
if resp.ok and (ret is None or ret == 0 or not body):
|
||
return True, ret, body
|
||
if ret in (-2, -14):
|
||
return False, ret, body # 鉴权/会话失效，立即停止重试
|
||
except Exception:
|
||
pass
|
||
return False, (last_body.get("ret") if isinstance(last_body, dict) else None), last_body
|
||
|
||
|
||
def _wechat_send_message(baseurl: str, bot_token: str, to_user_id: str, context_token: str, text: str):
|
||
"""发送纯文本回复。"""
|
||
return _wechat_post_msg(baseurl, bot_token, to_user_id, context_token,
|
||
[{"type": 1, "text_item": {"text": text}}])
|
||
|
||
|
||
def _platform_relay_url(raw_url: str) -> str:
|
||
"""把 AGNES 等境外媒体 URL 改写为平台中转地址，使微信/iLink 可内联渲染。
|
||
|
||
iLink 服务器(中国大陆)直连境外 AGNES 常被墙/超时，微信里只能看到"链接"。
|
||
改写为平台中转地址(平台服务器经香港代理可达 AGNES)，微信即可内联显示图片/视频。
|
||
"""
|
||
access = os.environ.get("APP_ACCESS_URL", "http://210.16.161.218:12888").rstrip("/")
|
||
return f"{access}/app/api/chat/media/relay?url={_quote(raw_url, safe='')}"
|
||
|
||
|
||
def _wechat_send_media(baseurl: str, bot_token: str, to_user_id: str, context_token: str,
|
||
media_type: str, url: str):
|
||
"""发送媒体（图片/视频）。
|
||
|
||
问题：AGNES 媒体 URL 位于境外，iLink(中国大陆服务器)直连常被墙/超时。
|
||
解决：改写为平台中转地址(经香港代理可达 AGNES)。
|
||
|
||
2026-07-31 兼容修复：iLink 对 type=2/5 的媒体 item 返回 ret=-2 invalid arguments，
|
||
且旧代码把 ret=-2 误判为 token 失效，导致连接假性断开。先以纯文本链接形式发送，
|
||
确保用户可点击访问且不会破坏会话；待拿到 iLink 媒体协议文档后再恢复内联 item。
|
||
"""
|
||
relay_url = _platform_relay_url(url)
|
||
if media_type == "image":
|
||
caption = "🖼️ 图片已生成，点击查看：" + relay_url
|
||
else:
|
||
caption = "🎬 视频已生成，点击查看：" + relay_url
|
||
return _wechat_send_message(baseurl, bot_token, to_user_id, context_token, caption)
|
||
|
||
|
||
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"])
|
||
if not parts:
|
||
for k in ("text", "content", "message", "msg"):
|
||
v = msg.get(k)
|
||
if isinstance(v, str) and v.strip():
|
||
parts.append(v)
|
||
return "\n".join(parts)
|
||
|
||
|
||
def _wechat_sender_id(m: dict) -> str:
|
||
"""从入站消息中提取发送者 id（兼容多套字段名）。"""
|
||
for k in ("from_user_id", "from_user_name", "from_username", "user_id",
|
||
"sender_id", "wxid", "from_wxid", "username"):
|
||
v = m.get(k)
|
||
if v:
|
||
return str(v)
|
||
return ""
|
||
|
||
|
||
def _wechat_context_token(m: dict) -> str:
|
||
"""提取 context_token（回复必须回带，否则不关联到对话窗口）。"""
|
||
for k in ("context_token", "ctx_token", "conversation_id", "conv_id",
|
||
"msg_id", "context_id", "session_id"):
|
||
v = m.get(k)
|
||
if v:
|
||
return str(v)
|
||
return ""
|
||
|
||
|
||
def _log_chat(user_id, platform, from_user, context_token, role, content):
|
||
"""持久化 IM Bot 对话历史，便于平台内查阅。"""
|
||
db = None
|
||
try:
|
||
db = SessionLocal()
|
||
_execute_raw(db, text(
|
||
"INSERT INTO im_bot_chat_log "
|
||
"(user_id, platform, from_user, context_token, role, content, created_at) "
|
||
"VALUES (:uid, :platform, :fu, :ctx, :role, :content, :created)"
|
||
), {
|
||
"uid": user_id, "platform": platform, "fu": from_user,
|
||
"ctx": context_token, "role": role, "content": content,
|
||
"created": _beijing_now(),
|
||
})
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
if db:
|
||
db.close()
|
||
|
||
|
||
# ============================================================
|
||
# 镜像到 AI 聊天（标准 conversations/messages）
|
||
# 个人微信对话作为一个独立会话窗（默认名 "WeXin Bot"）出现在 AI 聊天中，
|
||
# 从而复用 AI 聊天模块的全部功能（滚动翻阅、重命名、置顶、删除等）。
|
||
# ============================================================
|
||
WECHAT_CONV_TITLE = "WeXin Bot"
|
||
|
||
|
||
def _get_or_create_wechat_conv(db: Session, user_id: str) -> str:
|
||
"""按 user_id 找/建 source='wechat_bot' 的会话，返回会话 id。"""
|
||
rows = _execute_raw(db, text(
|
||
"SELECT id FROM conversations WHERE user_id=:uid AND source='wechat_bot' "
|
||
"AND status<>'archived' ORDER BY updated_at DESC LIMIT 1"
|
||
), {"uid": user_id})
|
||
if rows:
|
||
return _row_to_dict(rows[0])["id"]
|
||
cid = str(_uuid.uuid4())
|
||
now = _beijing_now()
|
||
_execute_raw(db, text(
|
||
"INSERT INTO conversations "
|
||
"(id, user_id, title, status, pinned, knowledge_graph_count, source, created_at, updated_at) "
|
||
"VALUES (:id, :uid, :title, 'active', 0, 0, 'wechat_bot', :c, :u)"
|
||
), {"id": cid, "uid": user_id, "title": WECHAT_CONV_TITLE, "c": now, "u": now})
|
||
return cid
|
||
|
||
|
||
def _append_ai_message(db: Session, conv_id: str, role: str, content: str, created=None):
|
||
_execute_raw(db, text(
|
||
"INSERT INTO messages (id, conversation_id, role, content, token_used, created_at) "
|
||
"VALUES (:id, :cid, :role, :content, 0, :c)"
|
||
), {
|
||
"id": str(_uuid.uuid4()), "cid": conv_id, "role": role,
|
||
"content": content, "c": created or _beijing_now(),
|
||
})
|
||
|
||
|
||
def _mirror_to_ai_chat(user_id: str, user_content: str, assistant_content: str):
|
||
"""把一轮微信问答写入用户的 WeXin Bot AI 聊天会话。"""
|
||
if not user_id:
|
||
return
|
||
db = None
|
||
try:
|
||
db = SessionLocal()
|
||
cid = _get_or_create_wechat_conv(db, user_id)
|
||
now = _beijing_now()
|
||
_append_ai_message(db, cid, "user", user_content, created=now)
|
||
# assistant 比 user 晚 50ms，保证排序严格递增，避免同秒时 ORDER BY 乱序
|
||
_append_ai_message(db, cid, "assistant", assistant_content,
|
||
created=now + timedelta(milliseconds=50))
|
||
_execute_raw(db, text("UPDATE conversations SET updated_at=:u WHERE id=:id"),
|
||
{"u": _beijing_now(), "id": cid})
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
if db:
|
||
db.close()
|
||
|
||
|
||
# ============================================================
|
||
# 后台接收线程（每用户独立）
|
||
# ============================================================
|
||
def _mark_wechat_expired(account_id: int):
|
||
"""发送会话失效（ret=-14/-2）：标记账号需重扫，并清除失效 token。"""
|
||
db = None
|
||
try:
|
||
db = SessionLocal()
|
||
_execute_raw(db, text(
|
||
"UPDATE im_bot_wechat_account SET status='expired', bot_token=NULL, "
|
||
"updated_at=:ts WHERE id=:aid"
|
||
), {"aid": account_id, "ts": _beijing_now()})
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
if db:
|
||
db.close()
|
||
|
||
|
||
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)
|
||
owner_uid = d.get("user_id")
|
||
for m in msgs:
|
||
if m.get("message_type") == 2:
|
||
continue # 仅跳过 bot 自己发出的回声，其余（含后续追问）均尝试处理
|
||
content = _wechat_extract_text(m)
|
||
if not content:
|
||
continue
|
||
sender = _wechat_sender_id(m)
|
||
ctx = _wechat_context_token(m)
|
||
# 诊断：记录原始入站消息，定位字段名 / 第二轮不回复
|
||
try:
|
||
with open("/tmp/wechat_inbound.log", "a", encoding="utf-8") as _f:
|
||
_f.write(_beijing_now().isoformat() + " sender=" + repr(sender)
|
||
+ " ctx=" + repr(ctx) + " raw=" + json.dumps(m, ensure_ascii=False) + "\n")
|
||
except Exception:
|
||
pass
|
||
# 先生成回复（此时历史尚未包含本条，避免重复）
|
||
import logging as _lrx, inspect as _insp
|
||
_lrx.warning(f'[receiver] calling _qihua_reply({repr(content)}, {repr(owner_uid)})')
|
||
# 验证使用的 _call_real_llm 版本
|
||
from app.api.im_llm import _call_real_llm as _crl
|
||
_src = _insp.getsource(_crl)
|
||
_lrx.warning(f'[receiver] _call_real_llm source len={len(_src)}, has _try_models={"_try_models" in _src}')
|
||
reply = _qihua_reply(content, owner_uid)
|
||
_lrx.warning(f'[receiver] reply ({type(reply).__name__}): {repr(reply)[:200]}')
|
||
# 媒体结构化返回：_call_real_llm 生成图片/视频时返回
|
||
# {"__media__": True, "media_type": "image"/"video", "url": "..."}
|
||
if isinstance(reply, dict) and reply.get("__media__"):
|
||
media_type = reply.get("media_type") or "image"
|
||
media_url = reply.get("url") or ""
|
||
ok, ret, _ = _wechat_send_media(baseurl, bot_token, sender, ctx, media_type, media_url)
|
||
# 对话历史里存可渲染的媒体标签（经平台中转，AI 聊天 WeXin Bot 窗与历史页均可内联显示）
|
||
relay_url = _platform_relay_url(media_url)
|
||
reply_for_log = (
|
||
f'<img src="{relay_url}" alt="生成图片"/>'
|
||
if media_type == "image"
|
||
else f'<video src="{relay_url}" controls></video>'
|
||
)
|
||
else:
|
||
reply_text = reply if isinstance(reply, str) else str(reply)
|
||
ok, ret, _ = _wechat_send_message(baseurl, bot_token, sender, ctx, reply_text)
|
||
reply_for_log = reply_text
|
||
if ok:
|
||
# 仅发送成功才写对话历史（避免“发了但没收到”的误导记录）
|
||
if owner_uid:
|
||
_log_chat(owner_uid, "personal_wechat", sender, ctx, "user", content)
|
||
_log_chat(owner_uid, "personal_wechat", sender, ctx, "assistant", reply_for_log)
|
||
# 同步镜像到 AI 聊天的 WeXin Bot 会话窗
|
||
_mirror_to_ai_chat(owner_uid, content, reply_for_log)
|
||
else:
|
||
try:
|
||
with open("/tmp/wechat_send.log", "a", encoding="utf-8") as _f:
|
||
_f.write(_beijing_now().isoformat() + " SEND_FAIL ret=%s sender=%s\n"
|
||
% (ret, repr(sender)))
|
||
except Exception:
|
||
pass
|
||
if ret == -14:
|
||
# -14=会话过期，需重扫
|
||
_mark_wechat_expired(d["id"])
|
||
elif ret == -2:
|
||
# -2=参数错误（如媒体消息格式不被 iLink 接受），并非 token 失效。
|
||
# 标记 expired 会导致连接假性断开，后续消息全部丢失，故仅记录。
|
||
pass
|
||
time.sleep(1)
|
||
except Exception:
|
||
# 2026-07-31 修复：此前裸 except 静默吞掉所有异常，导致接收器出错时毫无日志、
|
||
# 完全无法排查。改为记录异常堆栈到磁盘，便于定位。
|
||
import traceback as _tb
|
||
try:
|
||
with open("/tmp/wechat_receiver_err.log", "a", encoding="utf-8") as _f:
|
||
_f.write(_beijing_now().isoformat() + " RECEIVER EXCEPTION:\n" + _tb.format_exc() + "\n")
|
||
except Exception:
|
||
pass
|
||
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",
|
||
"status": d.get("status"),
|
||
"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 {}
|
||
qr_url = j.get("qrcode_img_content") or ""
|
||
qr_img = None
|
||
try:
|
||
import io as _io
|
||
import qrcode as _qrcode
|
||
_buf = _io.BytesIO()
|
||
_q = _qrcode.QRCode(box_size=10, border=2)
|
||
_q.add_data(qr_url)
|
||
_q.make(fit=True)
|
||
_img = _q.make_image(fill_color="black", back_color="white")
|
||
_img.save(_buf, format="PNG")
|
||
qr_img = "data:image/png;base64," + base64.b64encode(_buf.getvalue()).decode()
|
||
except Exception:
|
||
qr_img = None
|
||
return {"success": True, "data": {
|
||
"qrcode": j.get("qrcode"),
|
||
"qrcode_img_content": qr_url,
|
||
"qrcode_img": qr_img,
|
||
}}
|
||
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":
|
||
# 07-30 diagnostic: log full iLink confirm response
|
||
import json as _json
|
||
try:
|
||
with open("/tmp/wechat_confirm.log", "a", encoding="utf-8") as _f:
|
||
_f.write(_beijing_now().isoformat() + " CONFIRM_RAW=" + _json.dumps(j, ensure_ascii=False)[:2000] + "\n")
|
||
except Exception:
|
||
pass
|
||
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,
|
||
# 统一按北京时间存储，避免容器 UTC 导致显示偏差
|
||
"connected_at": _beijing_now(),
|
||
"created_at": _beijing_now(), "updated_at": _beijing_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": _beijing_now(),
|
||
"updated_at": _beijing_now(), "id": _row_to_dict(row)["id"],
|
||
})
|
||
row = _wechat_row_for_user(db, user.id)
|
||
saved = _row_to_dict(row) if row else {}
|
||
import json as _json2
|
||
try:
|
||
with open("/tmp/wechat_confirm.log", "a", encoding="utf-8") as _f:
|
||
_f.write(_beijing_now().isoformat() + " SAVED bot_token=" + repr(saved.get("bot_token", "NULL")[:20])
|
||
+ " ilink_bot_id=" + repr(saved.get("ilink_bot_id")) + "\n")
|
||
except Exception:
|
||
pass
|
||
aid = saved.get("id", 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": _beijing_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()
|
||
|
||
|
||
# ============================================================
|
||
# 启动恢复：为所有已连接用户账号拉起接收线程
|
||
# ============================================================
|
||
@router.get("/history")
|
||
async def wechat_history(
|
||
user: User = Depends(get_current_user),
|
||
limit: int = Query(200),
|
||
):
|
||
"""查询当前用户通过个人微信与七虎AI 的对话历史"""
|
||
db = SessionLocal()
|
||
try:
|
||
rows = _execute_raw(
|
||
db,
|
||
text("SELECT id, role, content, platform, from_user, created_at "
|
||
"FROM im_bot_chat_log WHERE user_id=:uid ORDER BY id DESC LIMIT :lim"),
|
||
{"uid": user.id, "lim": min(int(limit), 500)},
|
||
)
|
||
items = [_row_to_dict(r) for r in rows]
|
||
items.reverse()
|
||
return {"success": True, "data": {"messages": items, "total": len(items)}}
|
||
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()
|