前言
微信自动化流程需要正确处理账号离线和会话失效。恢复登录状态的前提,是当前会话仍满足接口要求;接口提示凭证过期或需要验证时,应转入重新授权流程,不能把恢复功能当作跳过用户确认的方式。
一、区分账号状态与任务状态
账号在线不代表每个业务任务都已成功;请求失败也不一定代表账号已退出。分别保存账号状态、最近检查结果、任务执行进度和失败原因,避免一次网络超时触发重复登录或重复操作。
二、按状态选择处理步骤
- 先用文档规定的状态接口确认账号当前情况。
- 仅在接口支持、会话有效且状态允许时发起恢复。
- 恢复请求返回失败时按错误类型处理,设置重试次数和停止条件。
- 需要重新验证或扫码确认时,提示账号所有者操作。
- 确认状态恢复后,再检查待处理任务,避免重复执行。
三、登录恢复接口与示例
下面保留原有状态管理和调用结构示例,仅作集成参考。接口路径、字段、响应和可恢复条件必须以当前文档为准,示意值不能直接当作生产配置。
客户端(你的服务) WechatApi 网关 微信服务端
| | |
|---[二次登录请求]------->| |
| appId + 设备票据 | |
| |---[票据验证请求]------->|
| | |
| |<--[票据合法,颁发新会话]--|
| | |
|<--[ret:200, 新syncKey]--| |
| | |
|---[后续业务请求]------->| |
httpPOST /api/v1/login/second HTTP/1.1
Host: api.wechatapi.net
Content-Type: application/json
VideosApi-token: your_api_token_here
json{
"appId": "wx_device_abc123",
"loginType": "second",
"proxyRegion": "cn-shanghai"
}
json{
"ret": 200,
"msg": "second login success",
"data": {
"appId": "wx_device_abc123",
"wxId": "wxid_xxxxxxxxxxxx",
"nickName": "张三",
"syncKey": "1_xxxxxxxxxx|2_xxxxxxxxxx|3_xxxxxxxxxx",
"loginTime": 1718245800,
"expireIn": 7200
}
}
json{
"ret": 1102,
"msg": "device token expired, need re-scan",
"data": {}
}
pythonimport time
import requests
import logging
logger = logging.getLogger(__name__)
WECHAT_API_BASE = "https://api.wechatapi.net"
API_TOKEN = "your_api_token_here" # 替换为控制台获取的真实 token
APP_ID = "wx_device_abc123" # 替换为实际设备 ID
HEADERS = {
"Content-Type": "application/json",
"VideosApi-token": API_TOKEN,
}
# 需要重新扫码的错误码
NEED_RESCAN_CODES = {1102, 1103}
# 可重试的错误码
RETRYABLE_CODES = {5001, 5002}
def second_login(app_id: str, max_retries: int = 3) -> dict | None:
"""
尝试恢复已有登录状态。
成功返回 data 字段,失败返回 None(需重新扫码时抛出 RuntimeError)。
"""
payload = {
"appId": app_id,
"loginType": "second",
"timeout": 15,
}
for attempt in range(1, max_retries + 1):
try:
resp = requests.post(
f"{WECHAT_API_BASE}/api/v1/login/second",
json=payload,
headers=HEADERS,
timeout=20,
)
result = resp.json()
ret_code = result.get("ret", -1)
if ret_code == 200:
logger.info(f"[{app_id}] 二次登录成功,wxId={result['data']['wxId']}")
return result["data"]
if ret_code in NEED_RESCAN_CODES:
raise RuntimeError(
f"[{app_id}] 设备票据失效 (ret={ret_code}),需重新扫码登录"
)
if ret_code in RETRYABLE_CODES and attempt < max_retries:
wait = 2 ** attempt # 指数退避:2, 4, 8 秒
logger.warning(f"[{app_id}] 第 {attempt} 次重试失败,{wait}s 后重试 (ret={ret_code})")
time.sleep(wait)
continue
logger.error(f"[{app_id}] 登录失败,不可重试 (ret={ret_code}, msg={result.get('msg')})")
return None
except RuntimeError:
raise
except Exception as e:
logger.error(f"[{app_id}] 请求异常: {e}")
if attempt < max_retries:
time.sleep(2 ** attempt)
return None
def login_guard(app_id: str, check_interval: int = 300):
"""
登录守护进程:每隔 check_interval 秒检查并维持登录态。
"""
logger.info(f"[{app_id}] 登录守护启动,检查间隔 {check_interval}s")
while True:
try:
data = second_login(app_id)
if data:
expire_in = data.get("expireIn", check_interval)
next_check = min(check_interval, expire_in - 60) # 提前 60s 续期
logger.info(f"[{app_id}] 登录态正常,{next_check}s 后下次检查")
time.sleep(next_check)
else:
logger.warning(f"[{app_id}] 登录失败,60s 后重试")
time.sleep(60)
except RuntimeError as e:
# 需要人工扫码
logger.critical(str(e))
# 此处可接入告警通知(钉钉/企微/短信)
break
bash# 设备登录态的 Redis 键设计示例
# 键格式: wechat:device:{appId}:status
# 值: online / offline / need_rescan / banned
# 登录成功后写入状态
redis-cli SET "wechat:device:wx_device_abc123:status" "online" EX 7200
# 设置 syncKey(TTL 与 expireIn 对齐)
redis-cli SET "wechat:device:wx_device_abc123:syncKey" "1_xxx|2_xxx|3_xxx" EX 7200
# 查询所有需要续期的设备(TTL < 300s)
# 在业务代码中实现,扫描 TTL 并提前触发 second_login
四、多设备与并发任务
同一账号的登录与恢复动作应由统一的状态管理入口协调。避免多个工作进程同时发起恢复,也不要在状态尚不明确时持续创建新设备或重复登录。
五、异常处理与人工接管
连续失败、权限限制或验证要求应触发告警与暂停。日志保留必要的任务与设备标识,但不要记录完整访问凭证;由有权处理的人员确认下一步操作。
总结
登录状态恢复是微信自动化流程的异常处理环节。以接口返回和有效授权为依据,配合有限重试、去重和人工确认,才能避免离线期间产生重复操作。
相关能力与接入说明:WechatApi;微信RPA流程集成;个人微信 API;微信二次开发;微信客服机器人;控制台;微信RPA流程集成
