#!/usr/bin/env python3 """ wechat_qr_notifier.py — 检测 wechat_bridge 未登录时，通过 XMPP 推送登录二维码 ============================================================================ 由 health check 或 crontab 调用。检查 Docker 日志中是否有二维码提示， 如果有则提取登录链接，通过 ejabberdctl 发送 XMPP 消息通知用户。 """ import json, os, re, subprocess, sys, urllib.request, urllib.parse from datetime import datetime GATEWAY_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) TEMP_DIR = os.path.join(GATEWAY_DIR, "temp") STATE_FILE = os.path.join(TEMP_DIR, "wechat_qr_state.json") LOGIN_URL = "http://localhost:3001/login?token=mowechat_fixed_token_001" DOCKER_CONTAINER = "wxBotWebhook" XMPP_FROM = "mohe@yoin.fun" XMPP_TO = "hmo@yoin.fun" NOTIFY_INTERVAL_MIN = 240 # 同一轮掉线最多每 4 小时提醒一次 QR_IMG_API = "https://api.qrserver.com/v1/create-qr-code/?size=500x500&margin=10&data=" def log(msg): ts = datetime.now().strftime("%H:%M:%S") print(f"[{ts}] {msg}") def is_logged_out(): """检查 Docker 日志最近 10 分钟是否有二维码提示。""" try: result = subprocess.run( ["docker", "logs", DOCKER_CONTAINER, "--since", "10m"], capture_output=True, text=True, timeout=10 ) logs = result.stdout + result.stderr # 只匹配二维码提示，不匹配 "already login" 等正常状态 return "扫码以下二维码以登录" in logs or "Or Access the URL to login" in logs except Exception as e: log(f"ERROR checking docker logs: {e}") return False def fetch_qr_url(): """从登录页面提取微信登录二维码链接。""" try: with urllib.request.urlopen(LOGIN_URL, timeout=10) as r: html = r.read().decode("utf-8", errors="replace") # 提取 qrcode.makeCode("...") 中的 URL m = re.search(r'qrcode\.makeCode\("([^"]+)"\)', html) if m: return m.group(1) # 备选：直接找 login.weixin.qq.com 链接 m = re.search(r'(https://login\.weixin\.qq\.com/l/[^\s"\'<>]+)', html) if m: return m.group(1) except Exception as e: log(f"ERROR fetching QR page: {e}") return None def should_notify(): """检查是否应该发送通知（避免频繁打扰）。""" os.makedirs(TEMP_DIR, exist_ok=True) try: with open(STATE_FILE, "r") as f: state = json.load(f) except (FileNotFoundError, json.JSONDecodeError): state = {} last_notify = state.get("last_notify", "") last_qr = state.get("last_qr", "") now = datetime.now() # 如果 QR 链接变了，立即通知（新的一轮掉线） # 如果 QR 链接没变但超过 30 分钟，再次提醒 if last_notify: try: last_dt = datetime.fromisoformat(last_notify) elapsed = (now - last_dt).total_seconds() / 60 if elapsed < NOTIFY_INTERVAL_MIN: return False, last_qr except ValueError: pass return True, last_qr def save_state(qr_url): """保存通知状态。""" state = { "last_notify": datetime.now().isoformat(), "last_qr": qr_url, } with open(STATE_FILE, "w") as f: json.dump(state, f) def send_xmpp(qr_url): """通过 ejabberdctl 发送 XMPP 消息，附带 QR 码图片链接。""" # 生成 QR 码图片链接（公网可访问，500x500 + margin""" qr_img_url = QR_IMG_API + urllib.parse.quote(qr_url, safe="") body = ( f"🔔 莫荷微信掉线了，需要重新扫码登录\n\n" f"👉 打开这个链接，用微信扫页面上的二维码：\n" f"{qr_img_url}\n\n" f"如果相册扫码不生效，试试复制这个链接到微信里打开：\n" f"{qr_url}\n\n" f"⏰ 二维码约 5 分钟过期，请及时扫码" ) # XML-escape body body = body.replace("&", "&").replace("<", "<").replace(">", ">") stanza = f"{body}" try: result = subprocess.run( ["docker", "exec", "ejabberd", "ejabberdctl", "send_stanza", XMPP_FROM, XMPP_TO, stanza], capture_output=True, text=True, timeout=15 ) if result.returncode == 0: log(f"XMPP notification sent to {XMPP_TO}") return True else: log(f"XMPP send failed: {result.stderr}") return False except Exception as e: log(f"XMPP send error: {e}") return False def main(): # 1. 检查是否掉线 if not is_logged_out(): log("wechat_bridge is logged in, nothing to do") return log("wechat_bridge appears to be logged out") # 2. 提取二维码链接 qr_url = fetch_qr_url() if not qr_url: log("ERROR: could not extract QR code URL") return log(f"QR URL: {qr_url}") # 3. 检查是否需要通知 should, last_qr = should_notify() if not should: log(f"Already notified recently (last QR: {last_qr[:50]}...), skipping") return # 4. 发送 XMPP 通知 if send_xmpp(qr_url): save_state(qr_url) log("Done") else: log("Failed to send notification") sys.exit(1) if __name__ == "__main__": main()