mohe
406446c3f1
SessionRouter approach failed on Linux because chat_bridge.py has
hardcoded Windows paths for provider config and opencode.db.
Instead, use the same pattern as xmpp_agent_core.py: direct Hermes API
call with X-Hermes-Session-Id=wechat-{sender_id} for per-user sessions.
This gives each WeChat user persistent context in Hermes without
depending on cross-platform path issues in chat_bridge.py.
330 lines
13 KiB
Python
330 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""WeChat webhook receiver v3 — per-user session context via Hermes session ID."""
|
|
|
|
import os, sys, json, logging, threading, queue, time
|
|
from http.server import HTTPServer, BaseHTTPRequestHandler
|
|
|
|
HERMES_API = "http://192.168.1.246:8646/v1/chat/completions"
|
|
HERMES_KEY = "hermes123"
|
|
PORT = 5804
|
|
|
|
ARTICLE_PROCESSOR_API = "http://192.168.1.16:5810/process"
|
|
KB_API = "http://192.168.1.246:9090/api/articles"
|
|
|
|
WECHAT_BOT_TOKEN = os.environ.get("WECHAT_BOT_TOKEN", "mowechat_fixed_token_001")
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(message)s",
|
|
handlers=[
|
|
logging.FileHandler("/home/hmo/AgentsMeeting/gateway/linux/logs/webhook.log"),
|
|
logging.StreamHandler()
|
|
]
|
|
)
|
|
log = logging.getLogger("wc-webhook")
|
|
|
|
# ── 消息队列（串行处理，防止并发打爆 Hermes）──────────────
|
|
_msg_queue = queue.Queue()
|
|
|
|
|
|
def _hermes_worker():
|
|
"""单线程 worker，从队列取消息逐条发给 Hermes。"""
|
|
while True:
|
|
sender, sender_id, text = _msg_queue.get()
|
|
try:
|
|
_do_forward(sender, sender_id, text)
|
|
except Exception as e:
|
|
log.error(f"Worker error: {e}")
|
|
finally:
|
|
_msg_queue.task_done()
|
|
# 每条消息之间稍作间隔，避免 LLM API 限流
|
|
time.sleep(2)
|
|
|
|
|
|
def _do_forward(sender, sender_id, text):
|
|
"""调用 Hermes API 并回复微信（带 X-Hermes-Session-Id 实现 per-user 上下文）。"""
|
|
session_id = f"wechat-{sender_id}"
|
|
payload = json.dumps({
|
|
"model": "nova-4",
|
|
"messages": [
|
|
{"role": "user", "content": f"[微信消息] 来自 {sender}({sender_id}): {text}"}
|
|
]
|
|
}).encode()
|
|
|
|
try:
|
|
import urllib.request as ureq
|
|
handler = ureq.ProxyHandler({})
|
|
opener = ureq.build_opener(handler)
|
|
req = ureq.Request(HERMES_API, data=payload,
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {HERMES_KEY}",
|
|
"X-Hermes-Session-Id": session_id,
|
|
})
|
|
resp = opener.open(req, timeout=600)
|
|
resp_data = json.loads(resp.read())
|
|
reply = resp_data.get('choices', [{}])[0].get('message', {}).get('content', '')
|
|
log.info(f"Hermes OK [{sender[:8]}], reply: {reply[:60]}")
|
|
|
|
if reply and sender:
|
|
WebhookHandler._send_wechat_static(sender, reply)
|
|
except Exception as e:
|
|
log.error(f"Hermes error [{sender[:8]}]: {e}")
|
|
|
|
|
|
# 启动 worker 线程
|
|
_worker_thread = threading.Thread(target=_hermes_worker, daemon=True)
|
|
_worker_thread.start()
|
|
|
|
|
|
# ── 文章抓取 + 入库 ──────────────────────────────────────
|
|
def _fetch_article(url):
|
|
"""调用 article_processor 抓取文章全文。"""
|
|
import urllib.request as ureq
|
|
payload = json.dumps({"url": url}).encode()
|
|
req = ureq.Request(ARTICLE_PROCESSOR_API, data=payload,
|
|
headers={"Content-Type": "application/json"})
|
|
with ureq.urlopen(req, timeout=300) as resp:
|
|
return json.loads(resp.read().decode())
|
|
|
|
|
|
def _register_article(data):
|
|
"""注册文章到知识库 DB。"""
|
|
import urllib.request as ureq
|
|
payload = json.dumps(data, ensure_ascii=False).encode("utf-8")
|
|
req = ureq.Request(KB_API, data=payload,
|
|
headers={"Content-Type": "application/json"}, method="POST")
|
|
try:
|
|
with ureq.urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read().decode())
|
|
except Exception as e:
|
|
log.error(f"KB register error: {e}")
|
|
return None
|
|
|
|
|
|
def _is_login_expired(text, result):
|
|
"""检测抓取结果是否为登录失效/验证码页面。
|
|
只有内容极短(<200字)且包含明确验证码关键词时才判定为风控。"""
|
|
if not text:
|
|
return True
|
|
# 内容足够长 = 正常文章，不可能是验证码页
|
|
if len(text) >= 200:
|
|
return False
|
|
# 内容太短但没有验证码关键词 = 可能只是短文章，不误判
|
|
captcha_keywords = ["环境异常", "安全验证", "滑动验证", "扫码验证", "操作频繁", "请完成安全验证"]
|
|
if any(k in text for k in captcha_keywords):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _process_article(sender, sender_id, content):
|
|
"""处理转发文章：抓取 → 入库 → 交 Hermes 回复。"""
|
|
import re as _re
|
|
import urllib.request as ureq
|
|
|
|
# 提取 URL
|
|
url_match = _re.search(r'"url":"(https://mp\.weixin\.qq\.com/[^"]+)"', content)
|
|
if not url_match:
|
|
log.warning(f"No URL found in urlLink content")
|
|
_do_forward(sender, sender_id, f"[分享链接] {content}")
|
|
return
|
|
article_url = url_match.group(1).replace("&", "&")
|
|
log.info(f"Fetching article: {article_url[:80]}...")
|
|
|
|
# 抓取
|
|
try:
|
|
result = _fetch_article(article_url)
|
|
title = result.get("title", "unknown")
|
|
text = result.get("content", "")
|
|
word_count = result.get("word_count", 0)
|
|
log.info(f"Fetched: {title[:40]} ({word_count} chars)")
|
|
|
|
# 检测风控/登录失效
|
|
if _is_login_expired(text, result):
|
|
log.warning(f"Login expired or captcha detected for {article_url[:60]}")
|
|
hermes_msg = (
|
|
f"[微信消息] 来自 {sender}({sender_id}): [分享链接] {content}\n\n"
|
|
f"[系统] 抓取这篇文章时检测到微信读者登录态已失效（页面返回验证码/环境异常）。"
|
|
f"这不是文章本身的问题，而是 article_processor 的 Chrome 微信读者 session 过期了。"
|
|
f"需要老莫在 Windows 上重新运行 wechat_fetcher.py --login 扫码登录后，抓取服务才能恢复。"
|
|
f"请告诉老莫这个情况，让他知道需要重新扫码。"
|
|
)
|
|
_do_forward(sender, sender_id, hermes_msg)
|
|
return
|
|
|
|
except Exception as e:
|
|
log.error(f"Fetch failed: {e}")
|
|
_do_forward(sender, sender_id, f"[分享链接] {content}")
|
|
return
|
|
|
|
# 入库
|
|
kb_data = {
|
|
"url": article_url,
|
|
"title": title,
|
|
"source": "user_sent",
|
|
"platform": "wechat",
|
|
"lifecycle": "draft",
|
|
"full_text": text[:5000] if text else "",
|
|
}
|
|
kb_result = _register_article(kb_data)
|
|
kb_id = kb_result.get("id") if kb_result else None
|
|
log.info(f"KB registered: id={kb_id}")
|
|
|
|
# 交 Hermes 回复（带入库结果）
|
|
kb_status = f"已入库(id={kb_id})" if kb_id else "入库失败"
|
|
hermes_msg = f"[微信消息] 来自 {sender}({sender_id}): [分享链接] {content}\n\n[系统] 文章已自动抓取并{kb_status}，全文 {word_count} 字。"
|
|
_do_forward(sender, sender_id, hermes_msg)
|
|
|
|
|
|
class WebhookHandler(BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
content_type = self.headers.get('Content-Type', '')
|
|
content_length = int(self.headers.get('Content-Length', 0))
|
|
|
|
# Read the body
|
|
body = self.rfile.read(content_length)
|
|
|
|
# Parse multipart/form-data
|
|
msg_type = 'unknown'
|
|
content = ''
|
|
source_raw = '{}'
|
|
is_system = '0'
|
|
|
|
if 'multipart/form-data' in content_type:
|
|
# Manual multipart parsing
|
|
boundary = content_type.split('boundary=')[1].strip()
|
|
if boundary.startswith('"') and boundary.endswith('"'):
|
|
boundary = boundary[1:-1]
|
|
|
|
parts = body.split(b'--' + boundary.encode())
|
|
for part in parts:
|
|
if b'Content-Disposition' not in part:
|
|
continue
|
|
|
|
# Parse headers
|
|
header_end = part.find(b'\r\n\r\n')
|
|
if header_end < 0:
|
|
continue
|
|
part_headers = part[:header_end].decode('utf-8', errors='replace')
|
|
part_body = part[header_end + 4:]
|
|
|
|
# Get field name
|
|
name_start = part_headers.find('name="')
|
|
if name_start < 0:
|
|
continue
|
|
name_start += 6
|
|
name_end = part_headers.find('"', name_start)
|
|
field_name = part_headers[name_start:name_end]
|
|
|
|
# Trim trailing \r\n--
|
|
if part_body.endswith(b'\r\n'):
|
|
part_body = part_body[:-2]
|
|
if part_body.endswith(b'--'):
|
|
part_body = part_body[:-2]
|
|
if part_body.endswith(b'\r\n'):
|
|
part_body = part_body[:-2]
|
|
|
|
if field_name == 'type':
|
|
msg_type = part_body.decode('utf-8', errors='replace')
|
|
elif field_name == 'content':
|
|
content = part_body.decode('utf-8', errors='replace')
|
|
elif field_name == 'source':
|
|
source_raw = part_body.decode('utf-8', errors='replace')
|
|
elif field_name == 'isSystemEvent':
|
|
is_system = part_body.decode('utf-8', errors='replace')
|
|
else:
|
|
# Try as regular form or JSON
|
|
try:
|
|
data = json.loads(body)
|
|
msg_type = data.get('type', 'unknown')
|
|
content = data.get('content', '')
|
|
source_raw = data.get('source', '{}')
|
|
except:
|
|
pass
|
|
|
|
# Parse source
|
|
try:
|
|
source = json.loads(source_raw) if isinstance(source_raw, str) else source_raw
|
|
except:
|
|
source = {}
|
|
|
|
# Extract sender info
|
|
sender_name = "unknown"
|
|
sender_id = "unknown"
|
|
if isinstance(source, dict):
|
|
from_data = source.get('from', {})
|
|
if isinstance(from_data, dict):
|
|
payload = from_data.get('payload', {})
|
|
sender_name = payload.get('name', 'unknown')
|
|
sender_id = payload.get('id', 'unknown')
|
|
|
|
# Skip system events
|
|
if is_system == '1':
|
|
log.info(f"System event: {msg_type}")
|
|
self._respond(200, {"status": "ok"})
|
|
return
|
|
|
|
log.info(f"From: {sender_name} ({sender_id}), Type: {msg_type}")
|
|
|
|
if msg_type == 'text':
|
|
log.info(f"Text: {content}")
|
|
self._forward_to_hermes(sender_name, sender_id, content)
|
|
elif msg_type == 'urlLink':
|
|
log.info(f"URL: {content}")
|
|
# 转发文章：抓取 → 入库 → 交 Hermes 回复
|
|
_process_article(sender_name, sender_id, content)
|
|
elif msg_type == 'file':
|
|
log.info(f"File received, length={len(body)}")
|
|
else:
|
|
log.info(f"Other: {msg_type}")
|
|
|
|
self._respond(200, {"status": "ok"})
|
|
|
|
def _forward_to_hermes(self, sender, sender_id, text):
|
|
"""把消息放入队列，由 worker 串行处理。"""
|
|
_msg_queue.put((sender, sender_id, text))
|
|
log.info(f"Queued for Hermes (queue size: {_msg_queue.qsize()})")
|
|
|
|
@staticmethod
|
|
def _send_wechat_static(to_name, text):
|
|
"""Send message back to WeChat user via bot API (static, for worker)."""
|
|
import urllib.request as ureq
|
|
token = WECHAT_BOT_TOKEN
|
|
api = f"http://localhost:3001/webhook/msg/v2?token={token}"
|
|
data = json.dumps({"to": to_name, "data": {"type": "text", "content": text}}).encode()
|
|
try:
|
|
handler = ureq.ProxyHandler({})
|
|
opener = ureq.build_opener(handler)
|
|
req = ureq.Request(api, data=data,
|
|
headers={"Content-Type": "application/json"})
|
|
resp = opener.open(req, timeout=10)
|
|
log.info(f"WeChat send OK: {resp.status}")
|
|
except Exception as e:
|
|
log.error(f"WeChat send error: {e}")
|
|
|
|
def _send_wechat(self, to_name, text):
|
|
"""Instance method wrapper."""
|
|
self._send_wechat_static(to_name, text)
|
|
|
|
def _respond(self, code, data):
|
|
self.send_response(code)
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(data).encode())
|
|
|
|
def log_message(self, format, *args):
|
|
pass
|
|
|
|
|
|
def main():
|
|
server = HTTPServer(('0.0.0.0', PORT), WebhookHandler)
|
|
log.info(f"Webhook receiver on :{PORT}")
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
server.shutdown()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|