在进行企微自动化二次开发时，由于安全策略，客户端的登录状态往往无法永久保持。因此，登录模块的设计必须具备两个核心能力：

1. 动态获取登录二维码并轮询扫码状态（未扫码 $\rightarrow$ 已扫码待确认 $\rightarrow$ 登录成功/过期）。

2. 登录成功后的会话维持（心跳检测）与异常重连机制。

本文将提供一段纯净的 Python 代码，展示如何通过统一接口（doApi）的逻辑，实现扫码登录的闭环控制。

一、 扫码登录状态机模型

在编写代码前，我们需要明确扫码登录的生命周期状态：

[步骤1: 获取二维码] ──> [步骤2: 循环请求状态] │ ┌────────────────────┼────────────────────┐ ▼ (CODE: 402) ▼ (CODE: 403) ▼ (CODE: 200) [等待用户扫码] [已扫码,手机待确认] [登录成功,获取Session]

• 状态 402：等待用户扫码（二维码有效）。

• 状态 403：已扫码，等待手机端确认授权。

• 状态 404：二维码已过期，需要重新获取。

• 状态 200：登录成功，返回实例的授权凭证（如guid或session_id）。

二、 实战源码：登录状态轮询与心跳维持

以下代码展示了如何获取登录二维码、阻塞式轮询扫码状态，以及在登录成功后启动心跳检测。

import time import requests # 统一中控地址 API_URL = "http://127.0.0.1:5000/api/wechat/callback" # 替换为你的中控或平台实际接口 TOKEN = "YOUR_SECURITY_TOKEN" headers = { "Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json" } def get_login_qrcode(): """ 步骤 1: 请求获取登录二维码 """ payload = { "action": "get_qrcode", "params": {} } try: response = requests.post(API_URL, json=payload, headers=headers, timeout=5) res = response.json() if res.get("code") == 200: data = res.get("data", {}) return data.get("qr_code_url"), data.get("login_ticket") except Exception as e: print(f"获取二维码失败: {e}") return None, None def poll_login_status(login_ticket): """ 步骤 2: 循环轮询扫码状态 """ payload = { "action": "check_login_status", "params": {"login_ticket": login_ticket} } print("⏳ 请使用企业微信扫码...") while True: try: response = requests.post(API_URL, json=payload, headers=headers, timeout=5) res = response.json() code = res.get("code") if code == 200: print("🎉 登录成功！") return True, res.get("data", {}).get("guid") # 返回登录成功的实例ID elif code == 402: print("...等待扫码中...") elif code == 403: print("📱 已扫码，请在手机端点击【确认登录】...") elif code == 404: print("❌ 二维码已过期，请重新运行脚本！") return False, None except Exception as e: print(f"轮询网络异常: {e}") # 官方建议每次轮询间隔 2~3 秒，避免过度消耗资源 time.sleep(2.5) def keep_alive_heartbeat(guid): """ 步骤 3: 登录成功后的在线状态心跳维持 """ payload = { "action": "heartbeat", "params": {"guid": guid} } print(f"🚀 心跳检测已启动，实例 ID: {guid}") while True: try: response = requests.post(API_URL, json=payload, headers=headers, timeout=5) res = response.json() if res.get("code") == 200: print(f"📡 心跳正常 | 在线状态: 良好 | 时间: {time.strftime('%X')}") else: print(f"⚠️ 实例可能异常离线: {res.get('msg')}") # TODO: 触发告警或自动重新调用登录流程 break except Exception as e: print(f"心跳发送失败: {e}") # 每隔 30 秒发送一次心跳，告知服务器该实例活跃 time.sleep(30) # 统一调度主流程 if __name__ == "__main__": # 1. 获取登录二维码及 Ticket qr_url, ticket = get_login_qrcode() if qr_url and ticket: print(f"🔗 请复制该链接在浏览器中打开并扫码: {qr_url}") # 2. 轮询扫码结果 success, active_guid = poll_login_status(ticket) # 3. 登录成功后，开启心跳维持在线状态 if success and active_guid: keep_alive_heartbeat(active_guid)

三、 调试避坑要点

1. Ticket 与 GUID 的区别：

   • login_ticket是临时凭证，仅在扫码阶段用于追踪扫码进度，扫码成功后即刻失效。

   • guid（或称session_id）是长期凭证，登录成功后后续的所有外部群主动发送、成员获取等操作，都必须携带该值作为鉴权依据。

2. 心跳包频率限制：

   心跳检测（Heartbeat）的间隔时间通常建议设置在30 秒 ~ 1 分钟之间。过于高频的心跳请求（例如每秒一次）会导致底层代理环境被服务器判定为高频请求而强制断开连接。

3. 断线自动重连：

   在生产环境中，一旦心跳检测连续 3 次失败，说明客户端可能已被人工退出或网络掉线。此时代码应立即中断心跳循环，向系统发送告警，并自动重新触发get_login_qrcode获取新的登录通道。