在进行企微自动化二次开发时由于安全策略客户端的登录状态往往无法永久保持。因此登录模块的设计必须具备两个核心能力动态获取登录二维码并轮询扫码状态未扫码 $\rightarrow$ 已扫码待确认 $\rightarrow$ 登录成功/过期。登录成功后的会话维持心跳检测与异常重连机制。本文将提供一段纯净的 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: fBearer {TOKEN}, Content-Type: application/json } def get_login_qrcode(): 步骤 1: 请求获取登录二维码 payload { action: get_qrcode, params: {} } try: response requests.post(API_URL, jsonpayload, headersheaders, timeout5) 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, jsonpayload, headersheaders, timeout5) 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, jsonpayload, headersheaders, timeout5) 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)三、 调试避坑要点Ticket 与 GUID 的区别login_ticket是临时凭证仅在扫码阶段用于追踪扫码进度扫码成功后即刻失效。guid或称session_id是长期凭证登录成功后后续的所有外部群主动发送、成员获取等操作都必须携带该值作为鉴权依据。心跳包频率限制心跳检测Heartbeat的间隔时间通常建议设置在30 秒 ~ 1 分钟之间。过于高频的心跳请求例如每秒一次会导致底层代理环境被服务器判定为高频请求而强制断开连接。断线自动重连在生产环境中一旦心跳检测连续 3 次失败说明客户端可能已被人工退出或网络掉线。此时代码应立即中断心跳循环向系统发送告警并自动重新触发get_login_qrcode获取新的登录通道。
)
