RPA+API混合架构实现智能群发调度

摘要单一技术方案总有短板。本文提出RPAAPI混合架构用RPA模拟人工点击解决“成员确认”痛点用API处理大规模标签筛选两者结合实现真正的自动化企微无限群发。包含完整的RPA脚本UiPath/影刀和API调用代码以及状态机设计模式。一、问题背景1.1 纯API方案的短板必须成员确认无法全自动标签筛选能力有限复杂逻辑需二次开发1.2 纯RPA方案的短板依赖UI界面稳定性差大规模发送效率低无法获取精确的客户列表1.3 混合架构的优势RPA负责“最后一公里”的确认点击API负责数据筛选和任务创建两者结合既合规又高效二、技术方案2.1 混合架构流程text┌─────────────────────────────────────────────────┐ │ 控制中心 │ │ (状态机待创建-待确认-已发送-数据回传) │ └──────────┬──────────────────┬───────────────────┘ │ │ ▼ ▼ ┌──────────────────┐ ┌──────────────────┐ │ API模块 │ │ RPA模块 │ │ • 获取客户标签 │ │ • 打开企微客户端 │ │ • 创建群发任务 │ │ • 定位群发入口 │ │ • 查询任务状态 │ │ • 点击确认发送 │ │ • 统计发送结果 │ │ • 异常处理 │ └──────────────────┘ └──────────────────┘ │ │ └────────┬─────────┘ ▼ ┌──────────────────┐ │ 数据库(MySQL) │ │ • 任务表 │ │ • 日志表 │ │ • 账号状态表 │ └──────────────────┘2.2 技术选型组件技术选型说明RPA工具影刀/UiPath国内推荐影刀UI元素识别更准API开发Python FastAPI高性能异步框架状态机transitionsPython状态机库数据库MySQL RedisMySQL持久化Redis缓存三、实现步骤步骤1API模块——创建待确认任务python# api_server.py from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel import mysql.connector import redis import requests import json app FastAPI() r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) class MassTask(BaseModel): name: str tag_rules: list content: str scheduled_time: str None app.post(/api/task/create) async def create_task(task: MassTask, background_tasks: BackgroundTasks): 创建群发任务状态待确认 # 1. 获取企业微信API access_token token get_qw_token() # 2. 调用官方API创建群发任务[citation:9] url fhttps://qyapi.weixin.qq.com/cgi-bin/externalcontact/add_msg_template?access_token{token} # 获取符合条件的客户列表 target_users get_users_by_tags(task.tag_rules) data { chat_type: single, external_userid: target_users[:10000], # 分批处理 text: {content: task.content}, sender: zhangsan, # 指定成员 allow_select: True # 允许成员选择 } resp requests.post(url, jsondata).json() if resp.get(errcode) 0: msgid resp.get(msgid) # 3. 存入数据库状态为pending_confirm db get_db() cursor db.cursor() sql INSERT INTO mass_tasks (msgid, name, content, target_count, status, created_at) VALUES (%s, %s, %s, %s, %s, NOW()) cursor.execute(sql, (msgid, task.name, json.dumps(task.content), len(target_users), pending_confirm)) db.commit() # 4. 触发RPA任务异步 background_tasks.add_task(trigger_rpa_confirm, msgid) return {code: 0, msgid: msgid, target_count: len(target_users)} else: return {code: -1, error: resp.get(errmsg)} def trigger_rpa_confirm(msgid): 触发RPA执行确认操作 # 可以通过Redis发布订阅或直接调用RPA的HTTP接口 r.publish(rpa_tasks, json.dumps({ action: confirm_mass, msgid: msgid }))步骤2RPA模块——自动点击确认影刀代码text# 影刀Python脚本 - confirm_mass.py import time import pyautogui import cv2 import numpy as np import redis import json import requests class QwRPA: 企微RPA操作类 def __init__(self): self.r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) def find_and_click(self, template_path, confidence0.8): 图像识别找元素并点击 # 截图 screenshot pyautogui.screenshot() screenshot cv2.cvtColor(np.array(screenshot), cv2.COLOR_RGB2BGR) # 读取模板 template cv2.imread(template_path) # 模板匹配 result cv2.matchTemplate(screenshot, template, cv2.TM_CCOEFF_NORMED) min_val, max_val, min_loc, max_loc cv2.minMaxLoc(result) if max_val confidence: # 计算中心点并点击 h, w template.shape[:2] center_x max_loc[0] w//2 center_y max_loc[1] h//2 pyautogui.click(center_x, center_y) time.sleep(1) return True return False def confirm_mass(self, msgid): 确认群发任务 print(f开始确认任务: {msgid}) # 1. 打开企微客户端如果未打开 if not self.find_and_click(templates/wecom_icon.png): # 从开始菜单打开 pyautogui.hotkey(win) time.sleep(1) pyautogui.write(企业微信) pyautogui.press(enter) time.sleep(5) # 2. 点击“群发助手” self.find_and_click(templates/mass_helper.png) time.sleep(2) # 3. 查找待发送任务 max_retry 5 for i in range(max_retry): if self.find_and_click(templates/to_send_tab.png): break time.sleep(2) # 4. 根据msgid查找对应任务简化默认点第一个 self.find_and_click(templates/task_item.png) time.sleep(1) # 5. 点击“发送” self.find_and_click(templates/send_button.png) time.sleep(2) # 6. 最终确认有些版本需要二次确认 self.find_and_click(templates/confirm_send.png) # 7. 回传状态 self.report_status(msgid, confirmed) def report_status(self, msgid, status): 回传确认状态 requests.post(http://localhost:8000/api/task/status, json{ msgid: msgid, status: status, confirmed_time: time.strftime(%Y-%m-%d %H:%M:%S) }) def listen_and_run(self): 监听Redis队列执行RPA任务 pubsub self.r.pubsub() pubsub.subscribe(rpa_tasks) for message in pubsub.listen(): if message[type] message: task json.loads(message[data]) if task[action] confirm_mass: self.confirm_mass(task[msgid]) if __name__ __main__: rpa QwRPA() rpa.listen_and_run()步骤3状态机管理python# state_machine.py from transitions import Machine import mysql.connector from datetime import datetime class MassTaskState: 群发任务状态机 states [pending_confirm, confirmed, sending, completed, failed] def __init__(self, task_id, msgid): self.task_id task_id self.msgid msgid self.db self.get_db() # 初始化状态机 self.machine Machine(modelself, statesMassTaskState.states, initialpending_confirm) # 定义转移 self.machine.add_transition(confirm, pending_confirm, confirmed, afterupdate_db) self.machine.add_transition(start_send, confirmed, sending, afterupdate_db) self.machine.add_transition(complete, sending, completed, afterupdate_db) self.machine.add_transition(fail, *, failed, afterupdate_db) def update_db(self): 更新数据库状态 cursor self.db.cursor() sql UPDATE mass_tasks SET status%s, updated_at%s WHERE msgid%s cursor.execute(sql, (self.state, datetime.now(), self.msgid)) self.db.commit() staticmethod def get_db(): return mysql.connector.connect( hostlocalhost, userroot, passwordpassword, databaseqw_mass )四、最佳实践4.1 异常处理机制RPA执行失败加入重试队列最多重试3次失败后发送企业微信告警API调用失败使用指数退避重试1s, 2s, 4s, 8s...人工介入超过3次失败的任务转入“人工处理”池由运营手动处理4.2 避坑指南UI元素变化企微客户端更新频繁模板截图需定期更新建议每周重新截图多开冲突RPA运行时不要操作鼠标键盘最好使用专用虚拟机权限问题确保RPA运行账号有企微客户端的完整操作权限五、工具推荐混合架构虽然灵活但RPA脚本维护成本高每次企微升级都可能失效。推荐使用企销宝技术优势内置RPA引擎API双引擎自动切换稳定性99.9%多账号并发支持100账号同时在线每个账号独立RPA实例与自研对比自研需维护RPA脚本API代码状态机企销宝开箱即用适合场景需要稳定、长期运行的群发任务无需投入研发维护