AgentsMeeting/gateway/linux/wechat_webhook.py

#!/usr/bin/env python3
"""WeChat webhook receiver v2 - receives messages from docker-wechatbot-webhook."""

import os, sys, json, logging, threading, queue, time
from http.server import HTTPServer, BaseHTTPRequestHandler

# ── MoFin session routing (shared with XMPP) ──
_SCRIPTS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts")
if _SCRIPTS_DIR not in sys.path:
    sys.path.insert(0, _SCRIPTS_DIR)
from chat_bridge import SessionBridge
from session_router import SessionRouter

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")

# ── Session router (per-user session context via opencode.db) ──
_router = SessionRouter(
    bridge=SessionBridge(session_id="wechat-mohe"),
    default_session="wechat-mohe",
)

# ── 消息队列（串行处理，防止并发打爆 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):
    """通过 SessionRouter 发送消息（自动带上下文 + 会话管理）。"""
    try:
        reply = _router.route("wechat", sender_id, text)
        log.info(f"Router OK, reply: {reply[:60] if reply else '(empty)'}")

        if reply and sender:
            WebhookHandler._send_wechat_static(sender, reply)
    except Exception as e:
        log.error(f"Router error: {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("&amp;", "&")
    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()