126 lines
4.2 KiB
Plaintext
126 lines
4.2 KiB
Plaintext
# wechat_callback.dspy - 微信回调（验签 + 消息接收 + openid映射 + gateway路由 + 被动回复）
|
||
# GET: 服务器配置验证（验签返回 echostr）
|
||
# POST: 接收消息 → openid 映射 user_id → gateway.run_message → 回复 XML
|
||
|
||
import json
|
||
import hashlib
|
||
import time
|
||
import xml.etree.ElementTree as ET
|
||
|
||
signature = (params_kw or {}).get('signature', '')
|
||
timestamp = (params_kw or {}).get('timestamp', '')
|
||
nonce = (params_kw or {}).get('nonce', '')
|
||
echostr = (params_kw or {}).get('echostr', '')
|
||
|
||
dbname = get_module_dbname('pipeline-sdlc')
|
||
|
||
|
||
async def _get_channel_config(sor):
|
||
"""取启用的机构公众号配置（简化：取第一条启用配置；多机构时按 ToUserName 区分）。"""
|
||
recs = await sor.sqlExe(
|
||
"SELECT org_id, appid, token, encoding_aes_key FROM wechat_channel_config "
|
||
"WHERE enabled='1' LIMIT 1", {})
|
||
if recs:
|
||
r = recs[0]
|
||
return {
|
||
'org_id': getattr(r, 'org_id', ''),
|
||
'appid': getattr(r, 'appid', ''),
|
||
'token': getattr(r, 'token', ''),
|
||
'encoding_aes_key': getattr(r, 'encoding_aes_key', ''),
|
||
}
|
||
return None
|
||
|
||
|
||
async def _resolve_user(sor, openid):
|
||
"""openid → user_id（未绑定返回空）。"""
|
||
recs = await sor.sqlExe(
|
||
"SELECT user_id FROM wechat_user_binding WHERE openid=${o}$", {"o": openid})
|
||
return getattr(recs[0], 'user_id', '') if recs else ''
|
||
|
||
|
||
# ── GET 验签 ──
|
||
if echostr:
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
cfg = await _get_channel_config(sor)
|
||
if not cfg:
|
||
return "未配置微信通道"
|
||
tmp = sorted([cfg['token'], timestamp, nonce])
|
||
sha1 = hashlib.sha1(''.join(tmp).encode('utf-8')).hexdigest()
|
||
if sha1 == signature:
|
||
return echostr
|
||
return "验签失败"
|
||
|
||
|
||
# ── POST 消息处理 ──
|
||
# 微信消息 XML 在 request body。Sage DSPY 环境：优先取 params_kw 里的 xml，否则取 request body。
|
||
xml_body = (params_kw or {}).get('xml', '')
|
||
if not xml_body:
|
||
try:
|
||
xml_body = await request.text() if hasattr(request, 'text') else ''
|
||
except Exception:
|
||
xml_body = ''
|
||
if not xml_body:
|
||
try:
|
||
xml_body = (await request.read()).decode('utf-8') if hasattr(request, 'read') else ''
|
||
except Exception:
|
||
xml_body = ''
|
||
|
||
if not xml_body:
|
||
return ""
|
||
|
||
try:
|
||
root = ET.fromstring(xml_body)
|
||
except Exception:
|
||
return ""
|
||
|
||
def _x(node, name):
|
||
e = node.find(name)
|
||
return e.text if e is not None else ''
|
||
|
||
to_user = _x(root, 'ToUserName') # 公众号（机构）
|
||
from_user = _x(root, 'FromUserName') # openid
|
||
msg_type = _x(root, 'MsgType')
|
||
content = _x(root, 'Content')
|
||
|
||
if msg_type != 'text' or not from_user:
|
||
# 非文本消息，回复统一提示
|
||
reply_text = "暂只支持文本消息"
|
||
else:
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
cfg = await _get_channel_config(sor)
|
||
user_id = await _resolve_user(sor, from_user)
|
||
|
||
if not user_id:
|
||
reply_text = "未绑定账号，请先登录系统在「微信绑定」中绑定您的微信"
|
||
else:
|
||
# 走 gateway 统一入口（微信通道）
|
||
from pipeline_service.gateway import get_gateway
|
||
gw = get_gateway()
|
||
parts = []
|
||
try:
|
||
async for chunk in gw.run_message("wechat", user_id, content):
|
||
d = json.loads(chunk)
|
||
t = d.get('type', '')
|
||
if t == 'reply':
|
||
parts.append(d.get('message', ''))
|
||
elif t == 'ask_user':
|
||
parts.append("❓ " + d.get('message', ''))
|
||
elif t == 'error':
|
||
parts.append("⚠️ " + d.get('message', ''))
|
||
# tool_call/tool_result 不推送给微信（避免刷屏）
|
||
except Exception as e:
|
||
parts.append(f"ERROR: {str(e)[:200]}")
|
||
reply_text = "\n".join(parts) if parts else "(无回复)"
|
||
|
||
# ── 被动回复 XML ──
|
||
reply_xml = (
|
||
"<xml>"
|
||
f"<ToUserName><![CDATA[{from_user}]]></ToUserName>"
|
||
f"<FromUserName><![CDATA[{to_user}]]></FromUserName>"
|
||
f"<CreateTime>{int(time.time())}</CreateTime>"
|
||
"<MsgType><![CDATA[text]]></MsgType>"
|
||
f"<Content><![CDATA[{reply_text}]]></Content>"
|
||
"</xml>"
|
||
)
|
||
return reply_xml
|