AICLW/wechatAiclaw
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: bug

Browse Source
This commit is contained in:
丹尼尔
2026-03-11 18:19:30 +08:00
parent 152877cef2
commit b7ef2569c4
13 changed files with 6907 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4276
backend/data/logs/app.log
View File

File diff suppressed because it is too large Load Diff

1609
backend/data/logs/node-access.log
View File

File diff suppressed because it is too large Load Diff

BIN
backend/data/wechat.db
View File

Binary file not shown.

658
backend/main.py
View File

@@ -5,13 +5,14 @@ import os
from contextlib import asynccontextmanager
from logging.handlers import RotatingFileHandler
from datetime import datetime
from typing import Any, List, Optional
from typing import Any, Dict, List, Optional
from urllib.parse import urlencode
import httpx
from fastapi import FastAPI, HTTPException, Query, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse, Response
from fastapi.responses import HTMLResponse, JSONResponse, Response
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
try:
@@ -25,21 +26,23 @@ except ImportError:
WECHAT_UPSTREAM_BASE_URL = os.getenv("WECHAT_UPSTREAM_BASE_URL", "http://localhost:8080").rstrip("/")
CHECK_STATUS_BASE_URL = os.getenv("CHECK_STATUS_BASE_URL", "http://113.44.162.180:7006").rstrip("/")
# 消息实时回调：设置后 7006 将新消息 POST 到该地址，作为主接收入口（与 SetCallback 一致）
CALLBACK_BASE_URL = (os.getenv("CALLBACK_BASE_URL") or "").strip().rstrip("/")
SLIDER_VERIFY_BASE_URL = os.getenv("SLIDER_VERIFY_BASE_URL", "http://113.44.162.180:7765").rstrip("/")
SLIDER_VERIFY_KEY = os.getenv("SLIDER_VERIFY_KEY", os.getenv("KEY", "408449830"))
# 发送文本消息：swagger 中为 POST /message/SendTextMessage，body 为 SendMessageModel（MsgItem 数组）
SEND_MSG_PATH = (os.getenv("SEND_MSG_PATH") or "/message/SendTextMessage").strip()
# 发送图片消息：部分上游为独立接口，或与文本同 path 仅 MsgType 不同（如 3=图片）
SEND_IMAGE_PATH = (os.getenv("SEND_IMAGE_PATH") or "").strip() or SEND_MSG_PATH
# 联系人列表：7006 为 POST /friend/GetContactList，body 传 CurrentChatRoomContactSeq/CurrentWxcontactSeq=0
CONTACT_LIST_PATH = (os.getenv("CONTACT_LIST_PATH") or os.getenv("FRIEND_LIST_PATH") or "/friend/GetContactList").strip()
FRIEND_LIST_PATH = (os.getenv("FRIEND_LIST_PATH") or CONTACT_LIST_PATH).strip()
# 图片消息 MsgType：部分上游为 0，常见为 3
IMAGE_MSG_TYPE = int(os.getenv("IMAGE_MSG_TYPE", "3"))
# 按 key 缓存取码结果与 Data62，供后续步骤使用
qrcode_store: dict = {}
# 按 key 缓存联系人索引：name(微信号/昵称/备注) -> 联系人详情
_contact_index: Dict[str, Dict[str, dict]] = {}
_LOG_FMT = "%(asctime)s [%(levelname)s] %(name)s - %(message)s"
logging.basicConfig(level=logging.INFO, format=_LOG_FMT)
# 日志落盘：写入 data/logs/app.log，便于排查（可按 LOG_DIR 覆盖目录）
@@ -69,14 +72,18 @@ def _allowed_ai_reply(key: str, from_user: str) -> bool:
return False
cfg = store.get_ai_reply_config(key)
if not cfg:
logger.debug("AI reply skipped: no config for key=%s (请在管理页「AI 回复设置」保存超级管理员/白名单)", key[:8])
return False
super_admins = set(cfg.get("super_admin_wxids") or [])
whitelist = set(cfg.get("whitelist_wxids") or [])
return from_user.strip() in super_admins or from_user.strip() in whitelist
allowed = from_user.strip() in super_admins or from_user.strip() in whitelist
if not allowed:
logger.debug("AI reply skipped: from_user=%s not in whitelist/super_admin for key=%s", from_user[:20], key[:8])
return allowed
async def _ai_takeover_reply(key: str, from_user: str, content: str) -> None:
"""收到他人消息时由 AI 接管：生成回复并发送。"""
"""收到他人消息时由 AI 接管：根据指令生成回复或调用内置动作（如代发消息）。"""
if not from_user or not content or not content.strip():
return
try:
@@ -95,33 +102,92 @@ async def _ai_takeover_reply(key: str, from_user: str, content: str) -> None:
break
if not context or context[-1].get("role") != "user":
context.append({"role": "user", "content": content})
text = await llm_chat(context)
if text and text.strip():
await _send_message_upstream(key, from_user, text.strip())
logger.info("AI takeover replied to %s: %s", from_user[:20], text.strip()[:50])
# Function Call 风格：让模型选择是直接回复，还是发起“代发消息”等动作
system_prompt = (
"你是微信客服助手，负责根据用户消息决定是直接回复，还是调用内置动作。"
"只用 JSON 回复，格式如下（不要多余文本）：\n"
"{\n"
' \"type\": \"reply\" | \"send_message\",\n'
' \"reply\"?: \"直接回复给当前联系人的内容（当 type=reply 时必填）\",\n'
' \"target_wxid\"?: \"要代发消息的对象 wxid（当 type=send_message 时必填，可以是当前联系人或其他 wxid）\",\n'
' \"content\"?: \"要代发的消息内容（当 type=send_message 时必填）\"\n'
"}\n"
"注意：\n"
"1）如果用户只是正常聊天，优先使用 type=reply；\n"
"2）只有当用户明确指令你“帮我给某人发消息/转告/通知 xxx”时，才使用 type=send_message；\n"
"3）严禁凭空编造 target_wxid，若无法确定对方 wxid，请继续使用 type=reply 询问用户；\n"
"4）不要用自然语言解释 JSON，只输出 JSON 本身。"
)
messages = [{"role": "system", "content": system_prompt}, *context]
raw = await llm_chat(messages)
if not raw or not raw.strip():
return
action = None
try:
import json as _json
action = _json.loads(raw)
except Exception:
# 回退为普通文本回复
reply_text = raw.strip()
await _send_message_upstream(key, from_user, reply_text)
logger.info("AI takeover replied (fallback) to %s: %s", from_user[:20], reply_text[:50])
return
if not isinstance(action, dict) or "type" not in action:
return
action_type = str(action.get("type") or "").strip()
if action_type == "send_message":
target = str(action.get("target_wxid") or "").strip()
msg = str(action.get("content") or "").strip()
if target and msg:
# 先按“昵称/备注/微信号”解析成真正的 wxid
real_wxid = await _resolve_contact_username(key, target)
if not real_wxid:
# 回一条提示给当前联系人，说明未找到该联系人
warn = f"未找到名为「{target}」的联系人，请确认称呼或直接提供对方微信号。"
await _send_message_upstream(key, from_user, warn)
logger.info("AI send_message resolve failed for %s (key=***%s)", target, key[-4:] if len(key) >= 4 else "****")
return
await _send_message_upstream(key, real_wxid, msg)
logger.info("AI function-call send_message to %s (raw=%s): %s", real_wxid[:20], target[:20], msg[:50])
elif action_type == "reply":
reply_text = str(action.get("reply") or "").strip()
if reply_text:
await _send_message_upstream(key, from_user, reply_text)
logger.info("AI takeover replied to %s: %s", from_user[:20], reply_text[:50])
except Exception as e:
logger.exception("AI takeover reply error (from=%s): %s", from_user, e)
def _on_ws_message(key: str, data: dict) -> None:
"""GetSyncMsg 收到数据时：写入 store；若为他人消息则 AI 接管对话。"""
msg_list = data.get("MsgList") or data.get("List") or data.get("msgList")
if isinstance(msg_list, list) and msg_list:
store.append_sync_messages(key, msg_list)
for m in msg_list:
if _is_self_sent(m):
continue
from_user = (m.get("FromUserName") or m.get("from") or "").strip()
content = (m.get("Content") or m.get("content") or "").strip()
msg_type = m.get("MsgType") or m.get("msgType")
if from_user and content and (msg_type in (1, None) or str(msg_type) == "1"): # 仅文本触发 AI
if not _allowed_ai_reply(key, from_user):
def _on_ws_message(key: str, data: Any) -> None:
"""GetSyncMsg / 回调 收到数据时：写入 store；若为他人消息则 AI 接管对话。"""
# 1）上游典型结构：{"MsgList": [...]} / {"List": [...]} / {"msgList": [...]}
if isinstance(data, dict):
msg_list = data.get("MsgList") or data.get("List") or data.get("msgList")
if isinstance(msg_list, list) and msg_list:
store.append_sync_messages(key, msg_list)
for m in msg_list:
if _is_self_sent(m):
continue
try:
asyncio.get_running_loop().create_task(_ai_takeover_reply(key, from_user, content))
except RuntimeError:
pass
elif isinstance(data, list):
from_user = (m.get("FromUserName") or m.get("from") or "").strip()
content = (m.get("Content") or m.get("content") or "").strip()
msg_type = m.get("MsgType") or m.get("msgType")
if from_user and content and (msg_type in (1, None) or str(msg_type) == "1"): # 仅文本触发 AI
if not _allowed_ai_reply(key, from_user):
continue
try:
asyncio.get_running_loop().create_task(_ai_takeover_reply(key, from_user, content))
except RuntimeError:
pass
return
# 2）如果 data 本身就是列表（例如回调已归一化为 [normalized_msg]）
if isinstance(data, list):
store.append_sync_messages(key, data)
for m in data:
if not isinstance(m, dict) or _is_self_sent(m):
@@ -136,21 +202,42 @@ def _on_ws_message(key: str, data: dict) -> None:
asyncio.get_running_loop().create_task(_ai_takeover_reply(key, from_user, content))
except RuntimeError:
pass
else:
store.append_sync_messages(key, [data])
m = data if isinstance(data, dict) else {}
if not _is_self_sent(m):
from_user = (m.get("FromUserName") or m.get("from") or "").strip()
content = (m.get("Content") or m.get("content") or "").strip()
msg_type = m.get("MsgType") or m.get("msgType")
if from_user and content and (msg_type in (1, None) or str(msg_type) == "1"):
if not _allowed_ai_reply(key, from_user):
pass
else:
try:
asyncio.get_running_loop().create_task(_ai_takeover_reply(key, from_user, content))
except RuntimeError:
pass
return
# 3）兜底：单条 dict / 其它类型
store.append_sync_messages(key, [data])
m = data if isinstance(data, dict) else {}
if not _is_self_sent(m):
from_user = (m.get("FromUserName") or m.get("from") or "").strip()
content = (m.get("Content") or m.get("content") or "").strip()
msg_type = m.get("MsgType") or m.get("msgType")
if from_user and content and (msg_type in (1, None) or str(msg_type) == "1"):
if not _allowed_ai_reply(key, from_user):
return
try:
asyncio.get_running_loop().create_task(_ai_takeover_reply(key, from_user, content))
except RuntimeError:
pass
async def _register_message_callback(key: str) -> bool:
"""向 7006 注册消息回调：POST /message/SetCallback?key=xxx，使 7006 将新消息推送到本服务。"""
if not CALLBACK_BASE_URL or not key:
return False
url = f"{CHECK_STATUS_BASE_URL.rstrip('/')}/message/SetCallback"
callback_url = f"{CALLBACK_BASE_URL.rstrip('/')}/api/callback/wechat-message"
body = {"CallbackURL": callback_url, "Enabled": True}
try:
async with httpx.AsyncClient(timeout=10.0) as client:
resp = await client.post(url, params={"key": key}, json=body)
if resp.status_code >= 400:
logger.warning("SetCallback %s key=%s: %s %s", url, key[-4:] if len(key) >= 4 else "****", resp.status_code, resp.text[:200])
return False
logger.info("SetCallback registered for key=***%s, CallbackURL=%s", key[-4:] if len(key) >= 4 else "****", callback_url)
return True
except Exception as e:
logger.warning("SetCallback error key=%s: %s", key[-4:] if len(key) >= 4 else "****", e)
return False
async def _run_greeting_scheduler() -> None:
@@ -219,7 +306,15 @@ async def _run_greeting_scheduler() -> None:
@asynccontextmanager
async def lifespan(app: FastAPI):
set_message_callback(_on_ws_message)
asyncio.create_task(start_ws_sync())
_callback_key = (os.getenv("WECHAT_WS_KEY") or os.getenv("KEY") or os.getenv("WS_KEY") or "").strip()
if CALLBACK_BASE_URL and _callback_key:
ok = await _register_message_callback(_callback_key)
if ok:
logger.info("消息接收已切换为实时回调入口，不再启动 WS GetSyncMsg")
else:
asyncio.create_task(start_ws_sync())
else:
asyncio.create_task(start_ws_sync())
scheduler = asyncio.create_task(_run_greeting_scheduler())
yield
scheduler.cancel()
@@ -847,6 +942,57 @@ async def api_list_messages(key: str = Query(..., description="账号 key"), lim
return {"items": store.list_sync_messages(key, limit=limit)}
@app.get("/api/callback-status")
async def api_callback_status(key: Optional[str] = Query(None, description="账号 key，传入时会向 7006 重新注册 SetCallback 并返回是否成功")):
"""检查消息回调配置：是否配置了 CALLBACK_BASE_URL、回调地址，以及（传入 key 时）向 7006 注册是否成功。"""
callback_url = ""
if CALLBACK_BASE_URL:
callback_url = f"{CALLBACK_BASE_URL.rstrip('/')}/api/callback/wechat-message"
registered: Optional[bool] = None
if key and key.strip():
k = key.strip()
registered = await _register_message_callback(k)
return {
"configured": bool(CALLBACK_BASE_URL),
"callback_url": callback_url or None,
"registered": registered,
}
@app.post("/api/callback/wechat-message")
async def api_callback_wechat_message(request: Request, key: Optional[str] = Query(None, description="账号 key（7006 回调时可能带在 query）")):
"""7006 消息实时回调入口：与 SetCallback 配合，收到新消息时 7006 POST 到此地址，与 WS GetSyncMsg 同结构，统一走 _on_ws_message 处理。"""
try:
body = await request.json()
except Exception:
body = {}
# 打印回调原始内容，便于排查结构与字段（截断避免日志过大）
try:
logger.info("callback/wechat-message raw body: %s", str(body)[:1000])
except Exception:
pass
k = (key or (body.get("key") or body.get("Key") or "") or "").strip()
if not k:
logger.warning("callback/wechat-message: missing key in query and body")
return JSONResponse(content={"ok": False, "error": "missing key"}, status_code=200)
try:
payload: Any = body
# 7006 回调当前格式示例：{"key": "...", "message": {...}, "type": "message"}
# 优先按回调 message 结构归一化，再回退到 Data/data 解包。
if isinstance(body, dict) and body.get("message"):
normalized = _normalize_callback_message(body)
if normalized:
payload = [normalized]
elif isinstance(body, dict):
inner = body.get("Data") or body.get("data")
if isinstance(inner, (dict, list)):
payload = inner
_on_ws_message(k, payload)
except Exception as e:
logger.exception("callback/wechat-message key=%s: %s", k[-4:] if len(k) >= 4 else "****", e)
return {"ok": True}
async def _send_message_upstream(key: str, to_user_name: str, content: str) -> dict:
"""调用上游发送文本消息；成功时写入发出记录并返回响应，失败抛 HTTPException。"""
url = f"{WECHAT_UPSTREAM_BASE_URL.rstrip('/')}{SEND_MSG_PATH}"
@@ -968,9 +1114,93 @@ async def api_send_image(body: SendImageBody):
raise HTTPException(status_code=502, detail=f"upstream_error: {exc}") from exc
def _log_contact_list_response_structure(raw: dict) -> None:
"""首轮无数据时打印上游响应结构，便于排查为何数据未解析到。"""
keys_top = list(raw.keys()) if isinstance(raw, dict) else []
data = raw.get("Data") or raw.get("data")
keys_data = list(data.keys()) if isinstance(data, dict) else (type(data).__name__ if data is not None else "None")
logger.info(
"GetContactList response structure (no items extracted): top_level_keys=%s, Data_keys=%s",
keys_top,
keys_data,
)
if isinstance(data, dict):
for k, v in list(data.items())[:5]:
preview = str(v)[:80] if v is not None else "null"
logger.info(" Data.%s: %s", k, preview)
def _unwrap_wechat_field(v: Any) -> Any:
"""上游字段有时为 {'str': 'xxx'} 或 {'len': 0} 这种包装，这里尝试取出内部值。"""
if isinstance(v, dict):
if "str" in v:
return v.get("str")
if "len" in v and len(v) == 1:
return v.get("len")
return v
def _normalize_callback_message(raw: dict) -> dict:
"""
将 7006 回调的 message 结构统一为与 WS GetSyncMsg 类似的消息字典，
便于复用 _on_ws_message / 实时消息面板的展示与 AI 接管逻辑。
示例 raw:
{
"key": "HBpEnbtj9BJZ",
"message": {
"msg_id": 126545176,
"from_user_name": {"str": "zhang499142409"},
"to_user_name": {"str": "wxid_xxx"},
"msg_type": 1,
"content": {"str": "测试"},
...
},
"type": "message"
}
"""
msg = raw.get("message") or raw
if not isinstance(msg, dict):
return {}
from_user = _unwrap_wechat_field(msg.get("from_user_name") or msg.get("FromUserName"))
to_user = _unwrap_wechat_field(msg.get("to_user_name") or msg.get("ToUserName"))
content = _unwrap_wechat_field(msg.get("content") or msg.get("Content"))
msg_type = msg.get("msg_type") or msg.get("MsgType")
create_time = msg.get("create_time") or msg.get("CreateTime")
# 回放到统一结构，字段名尽量与 WS GetSyncMsg 一致
normalized = {
"MsgId": msg.get("msg_id") or msg.get("MsgId") or msg.get("new_msg_id"),
"FromUserName": from_user or "",
"ToUserName": to_user or "",
"Content": content or "",
"MsgType": msg_type,
"CreateTime": create_time,
}
# 附带原始字段，便于调试 / 扩展展示
for k, v in msg.items():
if k in (
"msg_id",
"MsgId",
"new_msg_id",
"from_user_name",
"FromUserName",
"to_user_name",
"ToUserName",
"content",
"Content",
"msg_type",
"MsgType",
"create_time",
"CreateTime",
):
continue
normalized[k] = v
return normalized
def _normalize_contact_list(raw: Any) -> List[dict]:
"""将上游 GetContactList 多种返回格式统一为 [ { wxid, remark_name, ... } ]。"""
items = []
items: Any = []
if isinstance(raw, list):
items = raw
elif isinstance(raw, dict):
@@ -978,20 +1208,37 @@ def _normalize_contact_list(raw: Any) -> List[dict]:
if isinstance(data, list):
items = data
elif isinstance(data, dict):
items = (
contact_list = (
data.get("ContactList")
or data.get("contactList")
or data.get("WxcontactList")
or data.get("wxcontactList")
or data.get("CachedContactList")
or data.get("List")
or data.get("list")
or data.get("items")
or []
)
items = items or raw.get("items") or raw.get("list") or raw.get("List") or []
# 7006 格式：ContactList 为对象，联系人 id 在 contactUsernameList 字符串数组里
if isinstance(contact_list, dict):
username_list = (
contact_list.get("contactUsernameList")
or contact_list.get("ContactUsernameList")
or contact_list.get("UserNameList")
or []
)
if isinstance(username_list, list) and username_list:
items = [{"wxid": (x if isinstance(x, str) else str(x)), "remark_name": (x if isinstance(x, str) else str(x))} for x in username_list]
else:
items = []
else:
items = contact_list if isinstance(contact_list, list) else (
data.get("List") or data.get("list") or data.get("items")
or data.get("Friends") or data.get("friends")
or data.get("MemberList") or data.get("memberList") or []
)
items = items or raw.get("items") or raw.get("list") or raw.get("List") or raw.get("ContactList") or raw.get("WxcontactList") or []
result = []
for x in items:
if isinstance(x, str):
result.append({"wxid": x, "remark_name": x})
continue
if not isinstance(x, dict):
continue
wxid = (
@@ -1010,44 +1257,260 @@ def _normalize_contact_list(raw: Any) -> List[dict]:
or x.get("DisplayName")
or wxid
)
result.append({"wxid": wxid, "remark_name": remark, **{k: v for k, v in x.items() if k not in ("wxid", "Wxid", "remark_name", "RemarkName")}})
result.append({"wxid": wxid, "remark_name": remark})
return result
# 上游 GetContactList 请求体：CurrentChatRoomContactSeq、CurrentWxcontactSeq 传 0 表示拉取全量
GET_CONTACT_LIST_BODY = {"CurrentChatRoomContactSeq": 0, "CurrentWxcontactSeq": 0}
async def _fetch_all_contact_usernames(key: str) -> List[str]:
"""
调用 /friend/GetContactList 拉取全部联系人，返回去重后的 UserName 列表。
仅用于后续调用 GetContactDetailsList 获取详情。
"""
url = f"{CHECK_STATUS_BASE_URL.rstrip('/')}/friend/GetContactList"
usernames: List[str] = []
seen: set = set()
body: dict = {"CurrentChatRoomContactSeq": 0, "CurrentWxcontactSeq": 0}
max_rounds = 50
try:
async with httpx.AsyncClient(timeout=30.0) as client:
for round_num in range(max_rounds):
resp = await client.post(url, params={"key": key}, json=body)
if resp.status_code >= 400:
logger.warning("GetContactList(round=%s) %s: %s", round_num + 1, resp.status_code, resp.text[:200])
break
raw = resp.json()
chunk = _normalize_contact_list(raw)
if not chunk and isinstance(raw, dict):
chunk = _normalize_contact_list(raw.get("Data") or raw.get("data") or raw)
if round_num == 0 and not chunk and isinstance(raw, dict):
_log_contact_list_response_structure(raw)
for item in chunk:
wxid = (item.get("wxid") or "").strip()
if wxid and wxid not in seen:
seen.add(wxid)
usernames.append(wxid)
next_chat, next_wx = _next_contact_seq(raw)
if next_chat == 0 and next_wx == 0:
break
if next_chat == body.get("CurrentChatRoomContactSeq") and next_wx == body.get("CurrentWxcontactSeq"):
break
body = {
"CurrentChatRoomContactSeq": next_chat or body.get("CurrentChatRoomContactSeq", 0),
"CurrentWxcontactSeq": next_wx or body.get("CurrentWxcontactSeq", 0),
}
if not body["CurrentChatRoomContactSeq"] and not body["CurrentWxcontactSeq"]:
break
except Exception as e:
logger.warning("GetContactList usernames error: %s", e)
logger.info("GetContactList usernames total=%s", len(usernames))
return usernames
async def _build_contact_index(key: str) -> Dict[str, dict]:
"""
通用联系人索引：
- 先通过 GetContactList 拿到全部 UserName 列表；
- 再通过 /friend/GetContactDetailsList 批量拉取详情；
- 构建 name(微信号/昵称/备注) -> 联系人详情 的索引。
"""
if key in _contact_index and _contact_index[key]:
return _contact_index[key]
usernames = await _fetch_all_contact_usernames(key)
if not usernames:
_contact_index[key] = {}
return _contact_index[key]
url = f"{CHECK_STATUS_BASE_URL.rstrip('/')}/friend/GetContactDetailsList"
index: Dict[str, dict] = {}
async with httpx.AsyncClient(timeout=30.0) as client:
chunk_size = 50
for i in range(0, len(usernames), chunk_size):
batch = usernames[i : i + chunk_size]
body = {
"RoomWxIDList": [],
"UserNames": batch,
}
try:
resp = await client.post(url, params={"key": key}, json=body)
except Exception as e:
logger.warning("GetContactDetailsList batch error: %s", e)
continue
if resp.status_code >= 400:
logger.warning("GetContactDetailsList %s: %s", resp.status_code, resp.text[:200])
continue
raw = resp.json()
data = raw.get("Data") or raw.get("data") or raw
# 打印一次结构，便于排查为何没有解析出联系人详情
try:
if i == 0:
top_keys = list(raw.keys()) if isinstance(raw, dict) else type(raw).__name__
data_keys = list(data.keys()) if isinstance(data, dict) else type(data).__name__
logger.info(
"GetContactDetailsList structure: top_keys=%s, Data_keys=%s, batch_size=%s",
top_keys,
data_keys,
len(batch),
)
except Exception:
pass
items = []
if isinstance(data, dict):
# 7006 GetContactDetailsList 当前结构：Data.contactList 为联系人详情数组
items = (
data.get("List")
or data.get("list")
or data.get("ContactDetailsList")
or data.get("contacts")
or data.get("contactList")
or []
)
elif isinstance(data, list):
items = data
if not isinstance(items, list):
# 结构不符时记录一条日志，帮助判断需要从哪里取联系人列表
logger.info(
"GetContactDetailsList no list items parsed, data_type=%s, sample=%s",
type(data).__name__,
str(data)[:200],
)
continue
# 追加一次示例项日志，便于确认字段名（UserName/NickName/RemarkName 等）
try:
if i == 0 and items:
sample = items[0]
if isinstance(sample, dict):
logger.info(
"GetContactDetailsList first item keys=%s, sample=%s",
list(sample.keys()),
str(sample)[:200],
)
except Exception:
pass
for d in items:
if not isinstance(d, dict):
continue
# 只保留 bitVal == 3 的联系人（如上游定义的「有效联系人」），其它忽略
try:
bit_val = int(d.get("bitVal") or 0)
except (TypeError, ValueError):
bit_val = 0
if bit_val != 3:
continue
# 7006 联系人详情字段为 userName/nickName/pyinitial/quanPin 等，内部多为 {'str': 'xxx'} 包装
wxid = _unwrap_wechat_field(
d.get("userName") or d.get("UserName") or d.get("user_name") or d.get("wxid")
)
wxid = (wxid or "").strip()
if not wxid:
continue
nick = _unwrap_wechat_field(d.get("nickName") or d.get("NickName") or d.get("nick_name")) or ""
nick = str(nick).strip()
remark = _unwrap_wechat_field(
d.get("remark") or d.get("RemarkName") or d.get("remark_name")
) or ""
remark = str(remark).strip()
pyinitial = _unwrap_wechat_field(d.get("pyinitial") or d.get("pyInitial") or d.get("PYInitial")) or ""
pyinitial = str(pyinitial).strip()
quan_pin = _unwrap_wechat_field(d.get("quanPin") or d.get("QuanPin") or d.get("fullPinyin")) or ""
quan_pin = str(quan_pin).strip()
info = {
"wxid": wxid,
"remark_name": remark or nick or wxid,
"nick_name": nick,
"pyinitial": pyinitial,
"quan_pin": quan_pin,
"raw": d,
}
# 主键：wxid
index[wxid] = info
# 昵称 / 备注作为别名，便于按自然语言查找
if nick and nick not in index:
index[nick] = info
if remark and remark not in index:
index[remark] = info
_contact_index[key] = index
logger.info("Contact index built for key=***%s, size=%s", key[-4:] if len(key) >= 4 else "****", len(index))
return index
async def _resolve_contact_username(key: str, name: str) -> Optional[str]:
"""
将用户提到的“昵称/备注/微信号”解析成真正的 wxid(UserName)。
返回 wxid，找不到则返回 None。
"""
if not name:
return None
idx = await _build_contact_index(key)
info = idx.get(name.strip())
if info and isinstance(info, dict):
wxid = (info.get("wxid") or info.get("UserName") or "").strip()
return wxid or None
return None
# 上游 GetContactList：传 0 拉首页，响应可能带 NextWxcontactSeq/NextChatRoomContactSeq 表示还有后续页，需循环拉取全量
def _next_contact_seq(raw: dict) -> tuple:
"""从上游响应中解析下一页的 seq，返回 (next_chatroom_seq, next_wxcontact_seq)。无下一页则返回 (0, 0)。"""
def _int(v, default: int = 0) -> int:
if v is None:
return default
try:
return int(v)
except (TypeError, ValueError):
return default
data = raw.get("Data") or raw.get("data") or raw
chatroom_seq = 0
wxcontact_seq = 0
if isinstance(data, dict):
chatroom_seq = _int(data.get("NextChatRoomContactSeq") or data.get("CurrentChatRoomContactSeq"), 0)
wxcontact_seq = _int(data.get("NextWxcontactSeq") or data.get("CurrentWxcontactSeq"), 0)
for k in ("NextChatRoomContactSeq", "NextWxcontactSeq"):
v = raw.get(k)
if v is not None:
if "ChatRoom" in k:
chatroom_seq = _int(v, 0)
else:
wxcontact_seq = _int(v, 0)
return (chatroom_seq, wxcontact_seq)
# 联系人列表等接口禁止缓存，避免 304 导致前端拿到旧数据
_NO_CACHE_HEADERS = {"Cache-Control": "no-store, no-cache, must-revalidate", "Pragma": "no-cache"}
@app.get("/api/contact-list")
async def api_contact_list(key: str = Query(..., description="账号 key")):
"""获取全部联系人：POST 上游，body 为 CurrentChatRoomContactSeq/CurrentWxcontactSeq=0，key 走 query。"""
base = WECHAT_UPSTREAM_BASE_URL.rstrip("/")
path = CONTACT_LIST_PATH if CONTACT_LIST_PATH.startswith("/") else f"/{CONTACT_LIST_PATH}"
url = f"{base}{path}"
"""获取全部联系人详情：基于 GetContactList + GetContactDetailsList 构建的通用索引。禁止缓存。"""
try:
async with httpx.AsyncClient(timeout=15.0) as client:
resp = await client.post(
url,
params={"key": key},
json=GET_CONTACT_LIST_BODY,
)
if resp.status_code >= 400:
logger.warning("GetContactList %s: %s", resp.status_code, resp.text[:200])
return {"items": [], "error": resp.text[:200]}
raw = resp.json()
# 日志便于确认 7006 返回结构（不打印完整列表）
if isinstance(raw, dict):
data = raw.get("Data") or raw.get("data")
data_keys = list(data.keys()) if isinstance(data, dict) else getattr(data, "__name__", type(data).__name__)
logger.info("GetContactList response keys: raw=%s, Data=%s", list(raw.keys()), data_keys)
items = _normalize_contact_list(raw)
if not items and isinstance(raw, dict):
items = _normalize_contact_list(raw.get("Data") or raw.get("data") or raw)
logger.info("GetContactList normalized items count: %s", len(items))
return {"items": items}
index = await _build_contact_index(key)
# 只返回去重后的联系人详情（以 wxid 主键）
uniques: Dict[str, dict] = {}
for name, info in index.items():
if not isinstance(info, dict):
continue
wxid = (info.get("wxid") or "").strip()
if not wxid or wxid in uniques:
continue
uniques[wxid] = {
"wxid": wxid,
# 显示时优先用昵称，其次备注，最后用 wxid
"nick_name": info.get("nick_name") or "",
"remark_name": info.get("remark_name") or info.get("nick_name") or wxid,
"pyinitial": info.get("pyinitial") or "",
"quanPin": info.get("quan_pin") or "",
}
items = list(uniques.values())
logger.info("api_contact_list key=***%s -> %s contacts", key[-4:] if len(key) >= 4 else "****", len(items))
return JSONResponse(content={"items": items}, headers=_NO_CACHE_HEADERS)
except Exception as e:
logger.warning("GetContactList error: %s", e)
return {"items": [], "error": str(e)}
# 打印完整异常与 key，便于排查加载联系人报错
logger.exception("contact-list error for key=***%s: %s", key[-4:] if len(key) >= 4 else "****", e)
return JSONResponse(content={"items": [], "error": str(e)}, headers=_NO_CACHE_HEADERS)
@app.get("/api/friends")
@@ -1092,6 +1555,33 @@ async def api_update_ai_reply_config(body: AIReplyConfigUpdate):
)
@app.get("/api/ai-reply-status")
async def api_ai_reply_status(key: str = Query(..., description="账号 key")):
"""检查 AI 模型接管是否正常：WS 连接、是否配置白名单/超级管理员、是否有当前模型。"""
ws_ok = is_ws_connected()
cfg = store.get_ai_reply_config(key)
super_list = cfg.get("super_admin_wxids") or [] if cfg else []
white_list = cfg.get("whitelist_wxids") or [] if cfg else []
has_allow_list = bool(super_list or white_list)
model = store.get_current_model()
has_model = bool(model and model.get("api_key"))
ok = ws_ok and has_allow_list and has_model
return {
"ok": ok,
"ws_connected": ws_ok,
"has_ai_reply_config": bool(cfg),
"has_whitelist_or_super_admin": has_allow_list,
"super_admin_count": len(super_list),
"whitelist_count": len(white_list),
"has_current_model": has_model,
"message": "正常" if ok else (
"未连接消息同步(WS)" if not ws_ok else
"请在「AI 回复设置」添加并保存超级管理员或白名单" if not has_allow_list else
"请在「模型管理」添加并选中当前模型" if not has_model else "未知"
),
}
# ---------- 模型管理（多模型切换，API Key 按模型配置） ----------
class ModelCreate(BaseModel):
name: str
@@ -1235,3 +1725,9 @@ async def logout(body: LogoutBody):
)
return resp.json()
# 静态页面目录：与 Node 一致，直接访问后端时也可访问所有静态页
_PUBLIC_DIR = os.path.join(os.path.dirname(__file__), "..", "public")
if os.path.isdir(_PUBLIC_DIR):
app.mount("/", StaticFiles(directory=_PUBLIC_DIR, html=True), name="static")

2
backend/ws_sync.py
View File

@@ -11,7 +11,7 @@ logger = logging.getLogger("wechat-backend.ws_sync")
WS_BASE_URL = os.getenv("WECHAT_WS_BASE_URL", "").rstrip("/") or os.getenv("CHECK_STATUS_BASE_URL", "http://113.44.162.180:7006").rstrip("/").replace("http://", "ws://").replace("https://", "wss://")
# 与 7006 GetSyncMsg 建立连接时使用的 key，必须与登录页使用的账号 key 一致，否则收不到该账号的消息
# 优先读取 WECHAT_WS_KEY，未设置时使用 KEY（与登录参数一致）
DEFAULT_KEY = (os.getenv("WECHAT_WS_KEY") or os.getenv("KEY") or "").strip() or "HBpEnbtj9BJZ"
DEFAULT_KEY = (os.getenv("WECHAT_WS_KEY") or os.getenv("KEY") or os.getenv("WS_KEY") or "").strip() or "HBpEnbtj9BJZ"
try:
import websockets