影刀RPA 短信验证码自动接收：Webhook与API集成实战

拼多多店群自动化报活动上架影刀RPA 短信验证码自动接收Webhook与API集成实战作者林焱什么情况用这个做自动化测试和运维的同学经常碰到这个需求注册账号要收短信验证码、登录要收验证码、操作确认要收验证码。流程跑到一半卡在验证码输入界面必须人工去看手机——自动化的最后一公里断在这里。还有另一个场景监控告警。你的业务系统出现异常后自动发短信通知相关人员或者需要确认对方收到了通知才继续执行后续流程。不管你是用接码平台的虚拟号还是公司自有的短信通道把短信验证码的接收和处理接入影刀RPA流程才能真正实现端到端自动化。怎么做第一步选择短信接收方案根据你的场景选择方案方案A接码平台API适合测试、批量注册用接码平台如短信接码平台的开放API获取手机号→等待短信→获取验证码。流程全自动。方案B自有短信通道Webhook适合生产环境如果你们公司有短信通道阿里云短信、腾讯云短信等可以配置上行短信Webhook用户回复的短信自动推送到你的接口。方案C安卓手机转发零成本方案手机上装一个短信转发App如SMS Forwarder把收到的短信通过HTTP推送到影刀的Webhook接收地址。本文重点讲方案B和方案C因为这两种最通用。第二步搭建短信Webhook接收端影刀本身没有内置Webhook服务器但可以通过Python节点启动一个轻量HTTP服务来接收外部推送fromhttp.serverimportHTTPServer,BaseHTTPRequestHandlerimportjsonimportthreadingimportqueue# 全局消息队列sms_queuequeue.Queue()classSMSWebhookHandler(BaseHTTPRequestHandler):短信Webhook接收处理器defdo_POST(self):content_lengthint(self.headers.get(Content-Length,0))bodyself.rfile.read(content_length)try:datajson.loads(body)# 提取短信信息sms_info{手机号:data.get(phone,data.get(sender,)),内容:data.get(content,data.get(message,)),时间:data.get(time,data.get(timestamp,)),原始数据:data}sms_queue.put(sms_info)# 返回成功响应self.send_response(200)self.send_header(Content-Type,application/json)self.end_headers()self.wfile.write(json.dumps({code:0,msg:ok}).encode())exceptExceptionase:self.send_response(400)self.end_headers()defdo_GET(self):# 健康检查self.send_response(200)self.end_headers()self.wfile.write(bOK)defstart_webhook_server(port8899):启动Webhook服务器后台线程serverHTTPServer((0.0.0.0,port),SMSWebhookHandler)threadthreading.Thread(targetserver.serve_forever,daemonTrue)thread.start()print(f短信Webhook服务已启动端口{port})returnserver,sms_queue第三步在影刀流程中等待并提取验证码影刀流程操作【Python节点】→ 启动Webhook服务启动后得到一个消息队列变量 【循环】→ 等待短信到达 【Python节点】→ 检查队列中是否有新消息 【IF】→ 有消息提取验证码 【Python节点】→ 正则提取6位数字验证码 【填写】→ 将验证码填入目标输入框 【跳出循环】 【ELSE】→ 没消息 【等待】→ 等待2秒 【继续循环】Python节点中提取验证码importredefextract_verification_code(sms_content): 从短信内容中提取验证码 支持多种常见格式 - 【xx平台】验证码123456 - 您的验证码是123456 - code: 123456 - 123456xx平台验证码 # 常见验证码模式patterns[r验证码[:]\s*(\d{4,6}),# 验证码123456r验证码[是为]\s*(\d{4,6}),# 验证码是123456rcode[:]\s*(\d{4,6}),# code: 123456r码[:]\s*(\d{4,6}),# 动态码123456r【.{2,8}】\s*(\d{4,6}),# 【xx平台】123456r(?!\d)(\d{6})(?!\d),# 独立的6位数字兜底]forpatterninpatterns:matchre.search(pattern,sms_content,re.IGNORECASE)ifmatch:returnmatch.group(1)returnNone第四步影刀Webhook接收指令如果你的影刀账号支持企业版或服务器版可以直接用影刀内置的【Webhook接收】指令——不过很多场景下直接写Python更灵活。这里给出一个完整的前端到填写的流程1. 【Python节点】→ 启动接收服务start_webhook_server 2. 【获取变量】→ 拿到sms_queue对象 3. 【打开网页】→ 目标网站的注册/登录页 4. 【填写】→ 手机号 5. 【点击】→ 获取验证码按钮 6. 【For次数循环】→ 最多循环60次每次等2秒共等2分钟 7. 【Python节点】→ 检查队列try_get_sms 8. 【IF】→ code不为空 9. 【填写】→ 填入验证码输入框 10. 【跳出循环】 11. 【等待】→ 2秒 12. 【IF】→ code仍为空 13. 【记录日志】→ 验证码超时未收到 14. 【异常处理】等待取码的Python节点deftry_get_sms(sms_queue,target_phoneNone,timeout_seconds2): 尝试从队列中获取短信非阻塞 返回 (验证码, 短信内容) 或 (None, None) try:smssms_queue.get(timeouttimeout_seconds)# 如果指定了目标手机号过滤iftarget_phoneandtarget_phonenotinsms[手机号]:sms_queue.put(sms)# 放回去不是给我们的returnNone,Nonecodeextract_verification_code(sms[内容])returncode,sms[内容]exceptqueue.Empty:returnNone,None第五步对接阿里云短信上行如果你们用的是阿里云短信服务可以配置上行消息推送importtimeimporthashlibimporthmacimportbase64defget_aliyun_sms_up_messages(access_key_id,access_key_secret,phone_number,max_wait120): 通过阿里云短信API查询上行消息用户回复的短信 适合生产环境无需自建Webhook importrequests# 阿里云短信上行消息查询API# 注意需要短信服务已开通上行功能# 这里用一种更简单的方案使用阿里云短信的回执查询# 实际使用时建议配置MNS消息队列接收上行短信returnNone# 简化示例实际需要处理签名和API调用第六步安卓手机短信转发方案零成本如果你只是个人使用买接码平台不划算可以用安卓手机做短信转发在安卓手机上安装短信转发器App如SMS Forwarder、“Tasker 插件”配置转发规则匹配包含验证码的短信 → POST到你电脑的IP:8899确保网络可达手机和电脑在同一局域网或者用内网穿透转发规则配置示例SMS Forwarder的JSON配置{rules:[{sender_contains:,content_contains:验证码,forward_url:http://你的电脑IP:8899/sms,forward_method:POST,forward_format:{phone:{{sender}},content:{{content}},time:{{time}}}}]}有什么坑坑一防火墙/网络隔离现象手机发了短信电脑上Webhook没收到。原因手机和电脑不在同一网络或者Windows防火墙拦截了入站连接。解决确保手机和电脑在同一个WiFi以管理员身份运行以下命令开放端口# PowerShell管理员模式New-NetFirewallRule-DisplayName影刀短信Webhook-Direction Inbound-LocalPort 8899-Protocol TCP-Action Allow如果不方便配防火墙用内网穿透工具ngrok、frp暴露公网地址坑二多线程安全问题现象流程跑了几天后偶尔出现验证码填错或消息丢失。原因Python的queue.Queue本身是线程安全的但如果你在多个地方同时读队列可能产生竞态。解决确保只有一个消费者在读取队列。如果影刀流程中多个并行分支需要共享短信队列用锁保护TEMU店群矩阵自动化运营核价报活动importthreading sms_lockthreading.Lock()defsafe_get_sms(sms_queue,lock):withlock:try:returnsms_queue.get(timeout1)exceptqueue.Empty:returnNone坑三短信格式千奇百怪现象有的短信验证码提取不出来。原因不同平台发送的验证码短信格式差异大“您的验证码是”、“dynamic code”、“安全码”、“OTP”…花样百出。解决在正则匹配中不断积累模式同时保留独立的6位数字作为兜底策略。建议在日志中记录所有未能提取验证码的短信原文每周更新匹配规则# 持续积累的模式库PATTERNS{验证码:[r验证码[:是]\s*(\d{4,8}),r码[:]\s*(\d{4,8})],英文:[rcode[:]\s*(\d{4,8}),rOTP[:]\s*(\d{4,8})],平台:[r【.{2,10}】\s*(\d{4,8})],通用:[r(?!\d)(\d{4,8})(?!\d)],}坑四Webhook服务不稳定现象Webhook服务器偶尔崩溃流程一直等不到短信。原因Python的http.server比较简单高并发或异常请求可能导致崩溃。解决在流程中增加心跳检测如果Webhook挂了自动重启importrequestsdefhealth_check(port8899):检查Webhook是否存活try:resprequests.get(fhttp://127.0.0.1:{port},timeout2)returnresp.status_code200except:returnFalse# 在循环中每轮检查# 【IF】→ health_check()返回False → 【Python节点】→ 重启start_webhook_server()坑五验证码过期时间现象有时候收到了验证码但填入后被提示验证码已过期。原因Webhook接收→队列等待→流程提取→填写输入框中间可能过了几十秒而有些平台的验证码有效期只有60秒。解决流程中减少不必要的等待优先处理验证码填写到队列中的消息记录时间戳超过50秒的消息直接丢弃如果超时自动点击重新发送不等人工操作importtimedeftry_get_sms_with_ttl(sms_queue,max_age_seconds50):获取短信自动丢弃过期消息whileTrue:try:smssms_queue.get(timeout1)agetime.time()-sms.get(接收时间,0)ifagemax_age_seconds:codeextract_verification_code(sms[内容])returncode,sms[内容]else:# 验证码已过期丢弃continueexceptqueue.Empty:returnNone,None总结短信验证码自动接收是实现注册→登录→操作全链路自动化最关键的一环。三种方案各有适用场景接码平台适合批量测试、自有通道Webhook适合生产环境、手机转发适合个人零成本使用。无论选哪种核心思路都是接收→队列→提取→填写四步走其中队列机制是关键——它让异步的短信接收和同步的流程执行解耦了。