企业官网建设流程全解析

影刀RPA店群自动化：接口与UI混合执行引擎设计

纯UI自动化有很多局限。
慢、不稳定、容易被风控。但有些场景又绕不开UI——比如登录时的滑块验证码、平台没有开放API的操作、需要模拟人工浏览的行为。
我们一直在想：能不能把两者的优势结合起来？
API快、稳定、风控友好；UI能处理复杂交互、模拟真人。如果能根据场景自动选择最优执行方式，失败时自动降级切换，系统的整体稳定性会上一个大台阶。
这套接口与UI混合执行引擎我们跑了半年，效果远超预期。
这篇文章不讲调度，也不讲监控。专门聊聊如何设计双引擎架构，让店群自动化既快又稳。

适用场景：平台有开放API、但部分操作仍需UI的店群项目。

技术栈：平台API SDK + 影刀RPA + 决策引擎 + 状态同步。

一、API和UI的各自优劣势

先列一个对比表格：
| 维度 | API方式 | UI自动化（影刀RPA） |
|------|---------|---------------------|
| 执行速度 | 毫秒级 | 秒级到分钟级 |
| 稳定性 | 高（接口不变就稳） | 低（页面改版即挂） |
| 风控风险 | 低（正规调用） | 中高（行为模拟不完美） |

| 覆盖场景 | 平台开放的功能子集 | 任何人类能操作的功能 | | 维护成本 | 低（SDK版本更新） | 高（页面元素维护） | | 验证码处理 | 不需要 | 需要复杂处理 |

理想策略：能用API的用API，不得不用UI的用UI。
但店群业务中有很多混合场景。例如：

• 商品上架：API创建商品草稿，UI上传图片（因为图片API不稳定）
• • 订单发货：API批量发货，但遇到异常订单（风控拦截）时，切到UI人工处理
• • 登录维护：API登录获取token，但如果token过期且刷新失败，用UI重新登录
    混合执行引擎的核心就是：一个任务，多种执行方式，动态选择，自动降级。

二、双引擎架构全景

整体架构分为三层：
第一层：任务解析层
接收任务（如“上架商品”），解析出可执行的步骤序列。每个步骤标记是否支持API、是否有UI备选。
第二层：执行决策层
对每个步骤，根据当前环境（API可用性、风控状态、历史成功率）决定使用API还是UI。决策规则可配置、可动态调整。
第三层：执行层
API执行器（调用平台官方SDK）和UI执行器（调用影刀RPA脚本）。两者互相独立，通过统一的状态接口上报结果。
第四层：状态同步层
API和UI可能操作相同的数据（比如修改商品价格）。需要保证最终一致性，避免状态冲突。
下面详细展开每一层。

三、执行决策引擎

决策引擎的核心是一个优先级+降级规则表。

# decision_engine.pyclassExecutionDecider:def__init__(self):self.rules={"login":{"primary":"api",# 首选API登录"fallback":"ui",# API失败后降级到UI"conditions":{"api_success_rate":0.95,# API成功率低于95%时跳过API"force_ui":False}},"upload_product":{"primary":"api","fallback":"ui","conditions":{"product_count":10,# 超过10个商品用API批量，否则UI"image_count":5# 图片超过5张用UI（API不稳定）}},"upload_image":{"primary":"ui",# 图片上传API经常超时，默认UI"fallback":None,"conditions":{}},"batch_price_update":{"primary":"api","fallback":None,# 没有UI备选，失败就重试"conditions":{}}}defdecide(self,step_type,context):rule=self.rules.get(step_type)ifnotrule:return"ui"# 默认UI# 检查条件ifnotself._check_conditions(rule["conditions"],context):returnrule["fallback"]or"ui"# 检查历史成功率（从Redis读取）success_rate=self._get_api_success_rate(step_type)ifsuccess_rate<rule["conditions"].get("api_success_rate",0):returnrule["fallback"]or"ui"returnrule["primary"]``` 决策可以细到每个步骤，而不是整个任务。例如“商品上架”任务中：-创建商品记录：API--上传图片：UI--设置价格库存：API--提交审核：API 这样既发挥了API的速度，又避开了API的弱项。 决策结果会缓存在Redis中，同一类型的步骤短时间内不会重复决策（除非条件变化）。---## 四、API执行器的标准化封装不同平台的API差异很大。我们封装了一个统一的API执行器接口，屏蔽平台差异。 ```python# api_executor.pyclassUnifiedAPIExecutor:def__init__(self,platform,shop_credentials):self.platform=platform# "pdd", "temu", "tiktok"self.client=self._create_client(platform,shop_credentials)defexecute(self,action,params):""" 统一入口 action: "login", "upload_product", "update_price", "get_orders" """method_name=f"{self.platform}_{action}"ifhasattr(self,method_name):returngetattr(self,method_name)(params)raiseNotImplementedErrordefpdd_upload_product(self,params):# 调用拼多多开放APIimportpdd_sdk client=pdd_sdk.Client(self.client_id,self.client_secret)response=client.request("pdd.goods.add",params)returnself._normalize_response(response)deftemu_upload_product(self,params):# 调用TEMU API...``` API执行器内置了：-Token自动刷新--请求重试（指数退避）--限流保护（令牌桶）--响应标准化（统一返回格式） 执行失败时，API执行器会抛出特定异常（`APIRateLimitError`,`APITokenExpiredError`,`APINetworkError`），上层决策引擎据此决定是否降级到UI。---## 五、UI执行器的影刀RPA集成UI执行器负责调用影刀RPA脚本，但做了增强封装。 ```python# ui_executor.pyclassUIExecutor:def__init__(self,shop_id,script_name):self.shop_id=shop_id self.script_name=script_namedefexecute(self,params):# 调用影刀脚本，传入参数result=self._call_yingdao(self.script_name,params)# 解析结果ifresult.get("code")==0:returnresult["data"]else:# 抛出特定异常供决策引擎捕获if"element not found"inresult["msg"]:raiseUIElementNotFoundError(result["msg"])elif"captcha"inresult["msg"]:raiseUICaptchaRequiredError(result["msg"])else:raiseUIExecutionError(result["msg"])``` 影刀脚本本身也做了双模式适配：通过传入参数`execution_mode`来决定是否执行某些步骤。例如，如果API已经创建了商品草稿，UI脚本就跳过创建步骤，直接上传图片。 这样避免了API和UI做重复工作。---## 六、状态同步与冲突解决混合执行最大的隐患是状态不一致。比如API修改了商品价格，但UI脚本里缓存了旧价格，再次提交时覆盖了API的修改。 我们采用**主数据源+事件通知**的同步策略。-**主数据源**：以平台实际状态为准（通过API查询获取）--**写操作**：无论API还是UI修改数据，都要发布一个`data_changed`事件到Kafka--**缓存失效**：所有本地缓存（包括影刀脚本的内存变量）在收到事件后必须失效，下次操作前重新查询 ```python# state_sync.pyclassStateManager:def__init__(self):self.cache_ttl=60# 缓存有效期60秒defget_product_price(self,product_id):# 先查本地缓存，过期则查APIcached=redis.get(f"price:{product_id}")ifcachedandnotself._is_expired(cached):returncached# 从平台API获取最新价格price=api.get_product_price(product_id)redis.setex(f"price:{product_id}",self.cache_ttl,price)returnpricedefinvalidate(self,product_id):redis.delete(f"price:{product_id}")# 发布事件，通知其他执行器kafka.send("price_changed",{"product_id":product_id})``` 影刀脚本在执行敏感操作前，也会调用`StateManager.get_product_price`获取最新值，而不是使用之前读取的旧值。**冲突场景**：API和UI同时修改同一个商品的库存。解决方案：给每个写操作带上时间戳和来源，后执行的覆盖先执行的，但记录冲突日志供审计。因为店群场景中库存修改一般是增加或减少固定值，而不是覆盖，所以冲突较少。对于加减操作，我们要求API和UI都先读后写，读的时候加分布式锁。---## 七、降级与恢复的自动化当API执行器连续失败时，系统自动降级到UI。降级是步骤级的，而不是整个任务。 降级策略：-单个步骤API失败 → 尝试UI（如果配置了fallback）--UI也失败 → 整个任务失败，进入重试队列（使用指数退避）--连续3次API失败（不同任务）→ 标记该API端点不可用，后续任务直接跳过API，使用UI--API端点恢复检测：每5分钟尝试一次API调用，成功后清除不可用标记 ```python# health_check.pyclassAPIHealthChecker:def__init__(self):self.unhealthy_apis=set()defreport_failure(self,api_name):self.unhealthy_apis.add(api_name)# 启动恢复检测threading.Timer(300,self._test_recovery,[api_name]).start()def_test_recovery(self,api_name):ifself._test_api(api_name):self.unhealthy_apis.remove(api_name)# 发送恢复通知alert(f"API{api_name}is healthy again")defis_healthy(self,api_name):returnapi_namenotinself.unhealthy_apis ``` 决策引擎在决定使用API前，先检查健康状态。如果API被标记为不健康，直接走UI，避免浪费时间。---## 八、混合执行的典型应用场景**场景一：TEMU商品上架**TEMU的图片上传API经常超时（成功率约85%），但商品创建API稳定。我们设计：-创建商品基础信息：API（秒级成功）--上传图片：UI（影刀脚本上传，成功率99%）--设置价格库存：API--提交审核：API 混合执行后，上架成功率达到98%，平均耗时从UI纯执行（90秒）降到30秒。**场景二：拼多多批量改价**拼多多API支持批量改价，但有频率限制。我们设计：-当修改商品数>50时，使用API批量提交（快）--当修改商品数≤10时，使用UI（因为API有最低调用成本）--当遇到API限流错误时，降级到UI逐个修改 这避免了小批量任务浪费API配额，又保证了大批量任务的效率。**场景三：TikTok Shop登录**TikTok的token有效期较短，且刷新机制复杂。我们设计：-优先使用API登录（带缓存的token）--如果API返回token过期且刷新失败，启动UI登录脚本（模拟人工扫码或短信验证）--UI登录成功后，将新的token存入API执行器的凭证存储 这样既保证了无人值守时的自动化，又能在必要时人工介入（通过UI）。---## 九、数据闭环与持续优化混合执行引擎不是静态的。我们收集每次执行的数据，持续优化决策模型。 收集的指标：-API/UI各自的成功率、耗时、错误类型分布--降级触发频率和原因--同一操作在API和UI下的最终业务结果一致性 每周离线分析一次，自动调整决策规则：-如果某操作在API下成功率持续高于阈值，且UI备选从未被使用，可以移除UI备选简化逻辑--如果某操作的UI备选成功率显著高于API，调整primary为UI 例如：拼多多的“上传图片”操作，我们发现API成功率长期低于70%，而UI成功率98%。决策规则被自动改为`primary=ui`。---## 十、真实踩坑与经验**坑1：API和UI操作同一资源时的竞态条件**同一商品的价格，API刚改完，UI脚本里又改回旧值。因为UI脚本启动时读取了旧价格并缓存。 解决：UI脚本在每次修改前，都重新从平台获取最新值（通过API查询），不使用脚本启动时的参数。**坑2：API限流后降级到UI，UI又触发了更严格的风控**限流通常是因为操作频率过高。降级到UI后，UI脚本的频率可能更高（模拟人类点击其实不慢），导致平台更严厉的限制。 解决：降级到UI后，自动应用更保守的限速策略（操作间隔加长、随机停顿），同时通知运营降低该店铺的任务优先级。**坑3：API调用成功但平台异步处理未完成，UI读取到旧状态**平台API返回成功，但后台异步更新数据库，UI立即查询还是旧值。导致重复操作。 解决：API调用成功后，强制等待一段时间（根据平台经验值，如3秒），或者轮询查询直到状态更新。UI操作前也增加短暂延迟。**坑4：API凭证泄露风险**API密钥存储在调度器配置中，UI脚本通过浏览器模拟登录，也有暴露风险。 解决：API密钥使用云KMS加密存储，运行时解密。UI登录凭证（用户名密码）也加密，且仅由UI执行器解密，不经过影刀脚本变量。---## 总结：取长补短的艺术接口与UI混合执行，不是简单的“二选一”，而是根据场景动态组合。 核心收益：1.**效率提升**：API处理高频、批量操作，UI处理复杂、低频操作，整体吞吐量提升2-3倍2.2.**稳定性增强**：单点失败有降级路径，整体可用性从95%提升到99.5%3.3.**风控友好**：减少UI操作的频率和时长，降低被风控的概率4.4.**维护成本降低**：API变更由平台保证兼容，UI脚本只维护API覆盖不到的部分 如果你正在为UI脚本的不稳定而烦恼，不妨评估一下平台开放了哪些API。哪怕只把登录、查订单这类操作切换到API，也能减少一大半的脚本报错。 双引擎架构的建设周期大约需要2-4周，但长期回报非常显著。我们在50个店铺的TEMU项目中，该架构让每日人工介入次数从15次降到了2次。 自动化没有银弹，但取长补短的组合，往往是最优解。---作者：林焱