影刀RPA新手教程:影刀云调度完全指南——任务中心配置、机器人管理与并发执行

作者:林焱 | 真实案例驱动,每篇覆盖12大核心模块,禁止空话。

案例背景:从手动点"运行"到全自动调度

早期我写的影刀流程,全靠手动点"运行"按钮触发。

有一次我休年假,流程没人点运行,连续5天没执行,运营来问的时候我才发现。

后来用影刀云调度,现在所有流程全自动运行,不需要人工干预,我自己也敢放心休假了。


一、安装与准备工作

影刀云调度是影刀的企业版功能,社区版没有。

确认你的影刀版本支持云调度:打开影刀控制台,能看到"任务中心"菜单就是支持的。

云调度的基本概念:

  • 应用:在影刀设计器里开发的流程
  • 任务:在影刀控制台配置的执行计划(什么时候运行、运行哪个应用、参数是什么)
  • 机器人:实际执行应用的客户端(可以是你的电脑,也可以是云机器人)

使用前要做的准备:

  1. 在影刀控制台创建应用,并"发布"到云端(不是只保存在本地)
  2. 至少一台机器人切换到"调度模式"(在影刀客户端里,点击"切换到调度模式")
  3. 在控制台"机器人管理"里,确认机器人状态是"空闲"
  4. 在这里插入图片描述

我当时犯的第一个错:流程只保存在本地,没发布到云端,在控制台创建任务时找不到这个应用。

发布应用到云端:在影刀设计器里,点击"发布"按钮,选择"发布到控制台"。


二、元素定位:云调度不涉及界面操作

云调度的配置是在影刀控制台网页上完成的,不涉及元素定位。

但有一个常见需求:查看任务执行记录和日志,要在控制台页面上操作。

用影刀的网页自动化功能,可以自动登录控制台,批量下载执行日志。

控制台登录后的关键页面XPath:

# 任务中心菜单
//a[contains(text(),'任务中心')]

# 执行记录列表
//table[@class='task-record-table']//tr

# 执行状态标签
//span[contains(@class,'status')]/text()

我当时踩过这个坑:控制台页面用了动态加载,登录后直接跳转到任务中心,但页面还没渲染完,影刀就去点按钮,点了空。

解决方式:用"等待元素出现"指令,等页面关键元素出现后再操作。


三、变量与数据类型:调度参数的正确传递

云调度任务可以配置"应用参数",运行时传递给流程。

在影刀设计器里,先定义主流程参数:

右键应用 → 应用设置 → 主流程参数 → 添加参数。

在这里插入图片描述

参数类型支持:字符串、数字、布尔、文件。

在控制台创建任务时,"参数配置"里填入参数值。

参数值可以是固定值,也可以是"动态参数"(从上游任务传过来)。

动态参数的配置:

在控制台创建任务时,选择"参数来源"为"上游任务输出",配置上游任务的 jobUuid

我当时踩过这个坑:参数值里有中文,在控制台配置参数时没注意编码,流程里读到的中文是乱码。

解决方式:控制台参数配置里,所有中文值用UTF-8编码填写,流程里用 ensure_ascii=False 读取。


四、流程控制:多任务依赖与顺序执行

有些场景,任务B必须等任务A完成后才能执行。

店群矩阵自动化突破运营极限!

影刀云调度支持"任务依赖"配置:

在控制台创建任务时,“高级设置"里可以配置"依赖任务”。

配置后,任务B会自动等任务A执行成功后才会触发。

如果任务A执行失败,任务B不会触发,并收到"上游任务失败"的通知。

在这里插入图片描述

我当时有一个场景:先要从ERP导出数据(任务A),再处理数据发报告(任务B)。

早期没配置依赖,两个任务同时触发,任务B跑的时候任务A还没导出完,数据处理失败。

现在所有有依赖关系的任务,都在控制台配置好依赖,不用在流程里手写等待逻辑。


五、网页自动化:无人值守流程的注意事项

云调度执行的流程,运行在"调度模式"下,没有界面显示( headless 模式)。

有些在本地运行时正常的流程,在云调度里会失败,原因:

  1. 元素定位用了"基于图像的定位",headless模式下截不到图
  2. 流程里有"模拟鼠标移动到元素上"的操作,headless模式下鼠标事件不触发
  3. 弹窗处理用了"点击确认按钮",headless模式下弹窗可能不出现

解决方案:

  • 元素定位优先用XPath或CSS,不用图像定位
  • 鼠标悬停效果用"执行JavaScript"触发,而不是模拟鼠标移动
  • 弹窗处理用"检查元素是否存在"来判断,而不是等弹窗出现

我当时踩过这个坑:本地运行完全正常,发到云调度就失败,报错"元素不可见"。

查了半天才发现,流程里有一个"点击"操作,在headless模式下,元素在页面上但不在可视区域,点击失败。

解决方式:在点击前用"执行JavaScript"把元素滚动到可视区域:

document.querySelector('选择器').scrollIntoView();

在这里插入图片描述

六、数据处理:云调度环境下的文件路径处理

云调度执行的流程,工作目录和本地不一样。

本地运行时,当前目录是流程文件所在的目录。

云调度运行时,当前目录是影刀客户端的安装目录下的某个临时目录。

流程里如果用了相对路径读写文件,在云调度里会找不到文件。

正确做法:所有文件路径用绝对路径,或者用影刀的"获取应用数据目录"指令获取稳定的路径。

应用数据目录 = 获取应用数据目录
文件路径 = 应用数据目录 + "/data/orders.xlsx"

或者把所有文件操作统一到固定目录:

数据目录 = "C:/yingdao_data/"   # 云机器人上必须存在这个目录
确保目录存在(数据目录)
文件路径 = 数据目录 + "orders.xlsx"

我当时踩过这个坑:本地运行时文件存在 ./data/orders.xlsx,发到云调度后报错"文件不存在"。

后来所有文件路径全部改成绝对路径,问题解决了。


七、鼠标键盘图像:云调度模式下的特殊处理

云调度是headless模式,不涉及鼠标键盘操作,但有可能涉及截图。

截图保存路径要特别注意:

在这里插入图片描述

截图目录 = "C:/yingdao_screenshots/"
确保目录存在(截图目录)
截图文件路径 = 截图目录 + "error_" + 时间戳 + ".png"
截取页面截图 保存到 截图文件路径

截图保存后,如果要发给其他人看,需要上传到公网图床,因为云机器人的本地文件别人访问不到。

用Python把截图上传到OSS:

import oss2

def upload_screenshot(local_path):
    auth = oss2.Auth('key_id', 'key_secret')
    bucket = oss2.Bucket(auth, 'https://oss-cn-hangzhou.aliyuncs.com', 'bucket')
    object_name = 'screenshots/' + local_path.split('/')[-1]
    bucket.put_object_from_file(object_name, local_path)
    return f"https://bucket.oss-cn-hangzhou.aliyuncs.com/{object_name}"

我当时踩过这个坑:截图保存到了云机器人的本地目录,我想看截图得远程登录到那台机器,非常麻烦。

现在所有截图自动上传到OSS,URL直接发到飞书群里,点开就能看。


八、进阶技能:多机器人并发执行与负载均衡

当一个任务要在多台机器人上并行执行时(比如要从100个网站抓数据,一台机器人太慢),可以配置"多机器人执行"。

在控制台创建任务时,“机器人选择"里可以选"指定机器人"或"机器人分组”。

如果选"机器人分组",可以把多台机器人加到同一个分组里,任务触发时,控制台会自动分配机器人来执行。

分配策略有两种:

  • 随机分配:从空闲机器人里随机选一台
  • 负载均衡:选当前任务队列最短的机器人

配置方式:在控制台"任务中心 → 新建任务 → 高级设置 → 执行策略"。

我当时有一个任务要在50个网站上抓数据,一台机器人要跑2小时。

在这里插入图片描述

配置了5台机器人并发执行,每台机器人分10个网站,20分钟就全部跑完了。


九、平台实战:在影刀控制台配置定时任务

定时任务的配置,在影刀控制台"任务中心 → 新建任务"。

关键配置项:

  • 任务名称:要有意义,比如"每日销售数据推送"
  • 执行应用:选择已发布到云端的应用
  • 触发方式:定时触发 / API触发 / 手动触发
  • 定时规则:用Cron表达式,或选"每天/每周/每月"
  • 执行机器人:指定机器人或机器人分组
  • 参数配置:传入应用参数
  • 失败策略:失败时停止 / 失败时继续 / 失败时重试

Cron表达式速查:

0 9 * * *       每天9点
0 9,18 * * *    每天9点和18点
0 9 * * 1-5     周一到周五每天9点
0 0 1 * * *     每天凌晨1点

我当时踩过这个坑:Cron表达式 @ Daily 的意思是"每天触发一次",但具体时间是在创建任务的时间,不是凌晨0点。

如果想让任务每天凌晨执行,必须用 0 0 * * * 而不是 @ Daily


十、系统联动:云调度与开放API的联动

云调度的任务,可以通过开放API来触发和控制。

常用API:

  • POST /job/start:触发任务执行

  • GET /job/query:查询执行状态

  • POST /job/stop:停止执行

  • 在这里插入图片描述

  • GET /task/newest/list:查询任务执行记录

用开放API,可以实现"业务事件触发影刀流程":

比如ERP系统里创建了新订单,ERP调影刀API触发"订单处理"流程。

API调用示例(Python):

import requests

def trigger_yingdao_job(token, app_uuid, params={}):
    url = "https://api.winrobot360.com/oapi/dispatch/v2/job/start"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    data = {
        "appUuid": app_uuid,
        "params": params
    }
    resp = requests.post(url, headers=headers, json=data, timeout=30)
    return resp.json()

我当时踩过这个坑:API触发的任务,执行结果只能通过"查询执行状态"API来获取,没有主动回调。

如果需要执行完成后主动通知,要在流程的最后一步加一个"发送消息"指令。


十一、工程化规范:云调度任务的可维护性

云调度任务多了以后,必须要有规范,否则没人知道哪个任务干什么用的。

规范一:任务命名规范。

格式:[业务线] 任务描述 - 负责人

例子:[销售] 每日数据推送 - 张三[财务] 发票自动录入 - 李四

规范二:参数配置外置化。

所有可能变化的配置(Webhook地址、阈值、收件人),全部通过"应用参数"传入,不硬编码在流程里。

规范三:任务分组管理。

temu店群自动化报活动案例


在这里插入图片描述

在控制台"任务中心"里,用标签或分组功能,把同类任务放一起,方便管理。

我当时有30多个任务,命名全是"测试1"、“测试2”,后来自己都分不清哪个是哪个。

现在所有任务都用统一的命名规范,还在任务描述里写上"创建时间、用途、负责人、告警方式"。


十二、速查表与常见报错

报错1:任务触发后一直"等待调度"

原因:没有空闲机器人,或者机器人没切换到调度模式。

解决:检查机器人管理里,是否有状态为"空闲"的机器人,如果没有,等机器人空闲或增加机器人。

报错2:任务执行失败,错误信息"应用未发布"

原因:应用只保存在本地,没发布到云端。

解决:在影刀设计器里,点击"发布",把应用发布到控制台。

报错3:参数传递失败,流程里读到的参数是空

原因:参数名和流程里定义的主流程参数名不一致(大小写敏感)。

解决:检查参数名完全一致,包括大小写。

报错4:定时任务没有按时触发

原因:Cron表达式写错了,或者任务的"生效时间"设置不对。

在这里插入图片描述

解决:检查Cron表达式,用在线Cron表达式工具验证。

影刀云调度常用Cron表达式速查

需求 Cron表达式
每天9点 0 9 * * *
每天9点和18点 0 9,18 * * *
周一到周五每天9点 0 9 * * 1-5
每月1号凌晨 0 0 1 * *
每30分钟 */30 * * * *

完整案例代码参考

查询任务执行状态(Python,在影刀"执行Python代码"指令里调用):

import requests
import json
import time

YINGDAO_TOKEN = "你的accessToken"
YINGDAO_API_BASE = "https://api.winrobot360.com/oapi/dispatch/v2"

def trigger_job(app_uuid, params={}, robot_uuid=None):
    """触发任务执行"""
    url = f"{YINGDAO_API_BASE}/job/start"
    headers = {
        "Authorization": f"Bearer {YINGDAO_TOKEN}",
        "Content-Type": "application/json"
    }
    data = {"appUuid": app_uuid}
    if params:
        data["params"] = params
    if robot_uuid:
        data["robotClientUuid"] = robot_uuid
    
    resp = requests.post(url, headers=headers, json=data, timeout=30)
    result = resp.json()
    if result.get('code') == 200:
        job_uuid = result['data']['jobUuid']
        return job_uuid
    else:
        raise Exception(f"触发任务失败:{result}")

def wait_job_complete(job_uuid, timeout=600):
    """等待任务执行完成,最多等timeout秒"""
    url = f"{YINGDAO_API_BASE}/job/query"
    headers = {"Authorization": f"Bearer {YINGDAO_TOKEN}"}
    
    start_time = time.time()
    while time.time() - start_time < timeout:
        resp = requests.get(url, headers=headers, params={"jobUuid": job_uuid}, timeout=30)
        result = resp.json()
        if result.get('code') == 200:
            status = result['data']['status']
            if status in ['finish', 'error', 'stopped']:
                return status, result['data']
        time.sleep(10)
    
    raise Exception(f"任务执行超时:{job_uuid}")

# 用法示例:
# job_uuid = trigger_job("应用UUID", {"param1": "value1"})
# status, details = wait_job_complete(job_uuid)
# print(f"任务执行完成,状态:{status}")

XPath元素定位参考(影刀控制台任务中心):

# 任务列表中的状态列
//table[contains(@class,'task-table')]//tr[td[1]/text()='任务名称']/td[contains(@class,'status')]

# 执行记录详情
//div[@class='job-detail']//div[@class='log-content']/text()

我当时踩过这个坑:三个最深刻的教训

教训一:机器人在调度模式下弹出了Windows更新重启对话框。

机器人电脑自动更新重启了,重启后影刀客户端没有自动启动,所有任务排队等了半天没人知道。

现在所有机器人电脑都关闭了自动更新,并配置了"开机自启动影刀客户端"。

教训二:任务A依赖任务B,但任务B失败后任务A也失败了,没有告警。

在这里插入图片描述

依赖任务失败后的默认行为是"不触发",但没有发告警,我以为任务A正常跑完了。

现在所有依赖任务都配置了"上游失败时发告警"。

教训三accessToken 过期导致API触发失败。

开放API的 accessToken 有效期是24小时,我没做自动刷新,过期后所有API触发全部失败。

现在用一个定时任务每天自动刷新 accessToken 并保存到文件,所有API调用前先读文件里的token。


延伸

影刀云调度的完整配置手册,包含多机器人负载均衡、任务依赖配置、API触发封装,我都整理到了 home.linyan.cloud。

还有一套"机器人健康度监控"方案,可以自动检测机器人是否在线、是否有任务排队过长。


#影刀RPA #RPA教程 #云调度 #任务中心 #RPA #林焱

作者:林焱

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐