影刀RPA店群自动化：验证码识别与人工介入混合处理机制实战

店群自动化跑得越久，验证码出现的频率就越高。

这不是脚本写得好不好的问题，而是平台风控的正常行为。登录需要滑块、上架需要点选、批量操作需要输入验证码……每一个验证码都是一个阻塞点。

我们的第一版方案很简单：遇到验证码就等待30秒，然后重试。结果经常卡死，一卡就是半天。

后来改成了自动识别。但纯自动方案也有问题：有些验证码识别率低（比如复杂滑块、语序点选），强行重试反而容易被风控标记。

最终我们落地了一套自动识别 + 人工介入混合处理机制。

这篇文章专门聊聊这套机制的工程设计。

适用场景：拼多多、TEMU、TikTok Shop等平台的各类验证码处理。

技术栈：影刀RPA + Python + 打码平台API + WebSocket人工通知 + 任务挂起恢复。

---

一、验证码的分类与处理策略

先按我们遇到的实际情况，把验证码分成三类：

第一类：简单图形验证码

4-6位数字/字母，背景有轻微干扰线。OCR识别率90%以上。

策略：全自动 + 重试上限3次。

第二类：滑块验证码

拖动拼图到缺口位置。有简单滑块和带轨迹检测的滑块。

策略：轨迹模拟算法 + 打码平台（平台返回缺口坐标），成功率70%左右。失败后转人工。

第三类：点选/语序验证码

比如“点击下图中所有的红绿灯”、“按顺序点击汉字”。纯自动识别率低，且平台会记录点击轨迹。

策略：直接走人工介入，不浪费自动识别时间。

我们的原则是：自动能搞定的自动搞，搞不定的秒转人工，绝对不让任务卡死。

---

二、自动识别层的工程实现

对于图形验证码和滑块，我们接入了打码平台（如超级鹰、图鉴、云打码）。

影刀RPA遇到验证码时，截图保存，调用Python接口上传识别，取回结果后模拟输入或拖动。

# captcha_solver.py
import base64
import requests
from typing import Tuple

class CaptchaSolver:
    def __init__(self, platform_config):
        self.platform = platform_config["name"]  # "chaojiying", "tujian"
        self.username = platform_config["username"]
        self.password = platform_config["password"]
        self.soft_id = platform_config.get("soft_id")
        
    def solve_image_captcha(self, image_bytes) -> str:
        """识别图形验证码，返回验证码字符串"""
        if self.platform == "chaojiying":
            return self._solve_chaojiying(image_bytes)
        elif self.platform == "tujian":
            return self._solve_tujian(image_bytes)
            
    def solve_slider_captcha(self, image_bytes_with_gap, image_bytes_bg=None) -> int:
        """识别滑块验证码，返回需要滑动的距离（像素）"""
        # 调用打码平台的滑块识别接口
        # 返回距离值
        pass
        
    def _solve_chaojiying(self, image_bytes):
        # 超鹰接口调用
        files = {'userfile': ('captcha.jpg', image_bytes)}
        data = {
            'user': self.username,
            'pass2': self.password,
            'softid': self.soft_id,
        }
        resp = requests.post('http://upload.chaojiying.net/Upload/Processing.php', 
                             files=files, data=data)
        return resp.json().get('pic_str')

在影刀RPA脚本中，集成方式：

# 影刀RPA可以调用的Python脚本
import sys
import cv2
import numpy as np

def main():
    # 影刀传递截图路径
    screenshot_path = sys.argv[1]
    with open(screenshot_path, 'rb') as f:
        img_bytes = f.read()
    solver = CaptchaSolver(config)
    code = solver.solve_image_captcha(img_bytes)
    # 输出结果，供影刀读取
    print(code)
    
if __name__ == '__main__':
    main()

影刀侧流程：

1. 截图元素（验证码图片区域）
2. 保存到临时文件
3. 调用上述Python脚本，获取识别结果
4. 将结果输入验证码输入框
5. 点击提交

注意：必须设置超时和重试。打码平台可能返回错误（“未识别”），这时不要无限重试，超过2次就转人工。

---

三、滑块轨迹模拟：突破行为检测

简单的滑块验证码，只要求拖动到正确位置。但平台会检测拖动轨迹：是否匀速、是否有停顿、鼠标是否按下后立即移动等。

我们实现了一个轨迹生成器，模拟人类拖动的加速度和抖动。

# trajectory_generator.py
import random
import math

def generate_trajectory(distance: int, duration: float = 0.8) -> list:
    """
    生成模拟人类拖动的轨迹点列表
    distance: 需要拖动的总距离（像素）
    duration: 拖动总时长（秒）
    返回每个点的x坐标列表（相对起点）
    """
    points = []
    t = 0
    # 贝塞尔曲线或者简单的时间-位移函数
    # 人类拖动通常是慢-快-慢
    while t <= duration:
        # 缓动函数 easeOutCubic 变种
        progress = t / duration
        # 加一点随机抖动
        noise = random.uniform(-2, 2) * progress
        x = distance * (1 - (1 - progress) ** 3) + noise
        points.append(max(0, min(distance, int(x))))
        t += 0.01  # 每10ms一个点
    
    # 去重，去掉超出距离的点
    points = sorted(set(points))
    # 确保最后一点等于distance
    if points[-1] != distance:
        points.append(distance)
    return points

在影刀RPA中，我们可以通过模拟鼠标事件来实现这套轨迹。但影刀自带的“鼠标拖动”指令是直线匀速，容易被检测。

解决方案：使用Python的pynput或pyautogui逐点移动。影刀通过“运行Python代码”调用底层鼠标控制。

import time
from pynput.mouse import Button, Controller

def drag_with_trajectory(start_x, start_y, distance, duration):
    mouse = Controller()
    mouse.position = (start_x, start_y)
    mouse.press(Button.left)
    
    points = generate_trajectory(distance, duration)
    prev_x = start_x
    for x in points:
        delta = x - (prev_x - start_x)
        mouse.move(delta, 0)
        prev_x = start_x + x
        time.sleep(0.01)  # 控制速度
    
    mouse.release(Button.left)

这个轨迹模拟在简单滑块上效果不错，但对于带机器学习检测的高级滑块（比如某电商平台的极验升级版），仍然会被识别。那时我们就直接转人工了。

---

四、人工介入池的设计

当自动识别失败或遇到复杂验证码时，任务需要暂停，等待人工处理。

我们设计了一套人工介入池的机制：

• 每个店铺的验证码任务被挂起，推送到一个“待人工队列”
• 前端管理后台实时展示待处理验证码（截图 + 任务上下文）
• 运营人员在前端完成验证码输入或远程协助
• 系统恢复任务继续执行

核心数据结构：

# human_intervention.py
import redis
import json
import uuid

class HumanInterventionPool:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.pending_key = "human:captcha:pending"
        self.processing_key = "human:captcha:processing"
        
    def push_task(self, shop_id, task_id, captcha_type, screenshot_url, context):
        """推送一个待人工处理的验证码任务"""
        intervention_id = str(uuid.uuid4())
        task_data = {
            "id": intervention_id,
            "shop_id": shop_id,
            "task_id": task_id,
            "captcha_type": captcha_type,  # "slider", "click_word", "normal"
            "screenshot_url": screenshot_url,
            "context": context,
            "created_at": time.time(),
            "status": "pending"
        }
        self.redis.lpush(self.pending_key, json.dumps(task_data))
        # 同时通知前端（通过WebSocket或轮询）
        self._notify_frontend(intervention_id)
        return intervention_id
        
    def claim_task(self, operator_id):
        """运营人员认领一个待处理任务"""
        task_json = self.redis.rpop(self.pending_key)
        if not task_json:
            return None
        task = json.loads(task_json)
        task["operator_id"] = operator_id
        task["claimed_at"] = time.time()
        task["status"] = "processing"
        self.redis.hset(self.processing_key, task["id"], json.dumps(task))
        return task
        
    def submit_result(self, intervention_id, result):
        """人工提交验证码结果"""
        task_json = self.redis.hget(self.processing_key, intervention_id)
        if not task_json:
            return False
        task = json.loads(task_json)
        task["result"] = result
        task["status"] = "completed"
        task["completed_at"] = time.time()
        # 将结果写入结果队列，供调度器恢复任务
        self.redis.lpush(f"human:result:{task['task_id']}", json.dumps(result))
        self.redis.hdel(self.processing_key, intervention_id)
        return True

前端管理后台（基于Vue + WebSocket）实时显示待处理任务，运营人员看到截图后，手动完成验证码，将结果提交。

影刀RPA侧，任务在遇到验证码后会：

1. 调用HumanInterventionPool.push_task挂起当前任务
2. 进入等待循环，轮询human:result:{task_id}队列
3. 超时时间设置为10分钟（如果10分钟没人处理，重试自动识别或标记失败）
4. 拿到结果后，继续执行验证码提交操作

这种混合机制保证99%的验证码都能被处理，且人工成本可控（因为大部分简单验证码已被自动识别）。

---

五、任务挂起与恢复的工程细节

任务挂起不是简单的sleep。影刀RPA脚本需要能够“暂停等待外部信号，然后继续”。

我们设计了一个基于文件的信号机制（也可以用Redis发布订阅）：

# 影刀RPA脚本中的挂起逻辑（伪代码）
import time
import os

def wait_for_human_result(task_id, timeout=600):
    result_file = f"/tmp/captcha_result_{task_id}.json"
    start = time.time()
    while time.time() - start < timeout:
        if os.path.exists(result_file):
            with open(result_file, 'r') as f:
                result = json.load(f)
            os.remove(result_file)
            return result.get("value")
        time.sleep(1)
    return None

当人工提交结果时，Python后端将结果写入该文件。影刀脚本读取后继续执行。

这种方式简单可靠，但需要确保临时文件的清理（任务超时或完成后删除）。

更优雅的方式：使用Redis的brpop阻塞等待，影刀通过调用Python脚本来实现阻塞读取。

# 影刀调用此脚本获取人工结果，会阻塞直到有结果
def get_human_result(task_id, timeout=600):
    r = redis.Redis()
    result = r.brpop(f"human:result:{task_id}", timeout=timeout)
    if result:
        return json.loads(result[1])
    return None

这样影刀脚本中只需要执行result = python("get_human_result.py", task_id)，脚本会阻塞，直到人工处理完毕或超时。

坑点：影刀调用外部Python脚本时，如果脚本长时间阻塞，影刀可能会判定为“无响应”。需要调整影刀的超时设置，或者将等待逻辑拆分成多次轮询。

---

六、验证码出现频率的监控与自适应

不同店铺、不同时间段，验证码出现频率差异很大。

我们做了一个简单的监控模块，统计每个店铺的验证码频率，并动态调整策略。

# rate_limiter.py
class CaptchaRateMonitor:
    def __init__(self, redis_client):
        self.redis = redis_client
        
    def record(self, shop_id, captcha_type, solved_by):
        """记录一次验证码事件"""
        key = f"captcha:stats:{shop_id}"
        now = time.time()
        self.redis.zadd(key, {f"{now}:{captcha_type}:{solved_by}": now})
        # 保留最近24小时数据
        self.redis.zremrangebyscore(key, 0, now - 86400)
        
    def get_frequency(self, shop_id, window_hours=1):
        """过去window_hours小时内验证码出现次数"""
        now = time.time()
        count = self.redis.zcount(f"captcha:stats:{shop_id}", now - window_hours*3600, now)
        return count
        
    def recommend_action(self, shop_id):
        freq = self.get_frequency(shop_id)
        if freq > 10:
            return "suspend"  # 频率过高，暂停任务
        elif freq > 5:
            return "slow_down"  # 降低并发
        else:
            return "normal"

当某个店铺在1小时内出现超过10次验证码，调度器会暂时挂起该店铺的所有任务，并通知运营检查账号状态（可能是密码错误、IP被污染等）。

另外，我们根据验证码解决方式来调整优先级：如果某个店铺的验证码大部分依靠人工解决，系统会提示运营检查自动识别配置或更换打码平台。

---

七、真实踩坑与经验

坑1：打码平台返回的结果有时是错的，直接提交导致账号锁定

有些打码平台为了速度，返回随机结果。我们遇到过一次，连续5次识别错误，账号被临时锁定30分钟。

解决：增加二次校验机制。自动识别后，不是直接提交，而是先模拟输入但不点击提交，通过页面是否有“验证码错误”提示来判断。如果识别结果明显错误，立即重试，而不是盲目提交。

坑2：人工介入时，多个运营同时处理同一任务

如果没有锁机制，两个运营可能会同时认领同一个验证码任务，提交两次结果。

解决：使用Redis的rpop原子操作，每个任务只能被一个运营认领。同时前端展示“已被某某处理”的状态。

坑3：滑块轨迹生成太完美，反而被识别

我们第一版轨迹生成太平滑，没有人类的那种微小抖动和超调。平台可以通过机器学习区分人类和机器轨迹。

解决：在轨迹中加入随机噪音，以及最后的微小回弹（人类拖动过头会往回挪一点）。同时随机化拖动时长（0.6~1.2秒之间）。

坑4：影刀截图验证码图片的时机太早或太晚

验证码图片可能异步加载，影刀截图时图片还没完全渲染。导致提交的是空白图或残缺图。

解决：在截图前增加“等待图片加载完成”的逻辑，比如等待某个image元素加载完毕，或者循环截图比对像素变化，稳定后再保存。

---

八、运营后台的远程协助能力

人工介入不仅要能提交验证码，还要能“远程接管”浏览器，处理一些自动脚本无法应对的复杂情况（比如弹窗公告、新手指南）。

我们在管理后台集成了远程浏览器实时画面（通过WebRTC或VNC），运营人员可以直接看到店铺当前浏览器窗口，并操作鼠标键盘。这对处理突发情况非常有用。

实现方式：在每台执行节点上运行一个WebRTC代理服务，将浏览器桌面（或单个标签页）推流到后台。运营连接后，输入事件回传到节点。

这个能力不是必须的，但对于高级运维场景，能大幅降低人工介入的成本。

---

九、总结：自动与人工的边界

店群自动化中，验证码处理是一个绕不开的话题。

我见过有的团队试图100%自动识别，结果投入大量精力调算法，识别率从70%提到85%，但最后15%仍然卡死。

也见过有的团队完全人工处理，运营每天盯着屏幕做验证码，效率极低。

混合机制是务实的选择：

• 简单验证码全自动
• 中等难度的自动+打码平台
• 高难度秒转人工

同时通过监控和自适应策略，避免同一店铺频繁触发验证码（那说明风控级别高，需要降低操作频率或更换环境）。

这套机制让我们的店群系统能够7x24小时运行，人工介入的日均次数控制在个位数。

希望对正在折腾验证码的你有所帮助。

---

作者：林焱