影刀RPA 定时任务的配置与管理：让流程自动运行
作者：林焱
RPA最大的价值之一就是"无人值守"——设置好定时任务，流程每天自动运行，你不用手动点启动。但很多人配置定时任务时踩坑：任务没触发、触发了但流程报错、多个任务冲突……
这篇文章讲清楚影刀RPA定时任务的配置方法、常见场景和踩坑经验。
一、什么情况用什么
| 场景 | 触发方式 | 说明 |
|---|---|---|
| 每天固定时间运行 | 定时触发（每天09:00） | 最常见 |
| 工作日运行，周末不运行 | 定时+工作日判断 | 结合chinesecalendar |
| 每隔N小时运行一次 | 间隔触发（每2小时） | 适合监控类流程 |
| 每周一运行 | 定时触发（每周一09:00） | 周报类流程 |
| 每月1号运行 | 定时触发（每月1号） | 月报类流程 |
| 文件出现时触发 | 轮询检测+条件触发 | 文件到达后自动处理 |
| 上一个流程完成后触发 | 流程链式调用 | 串行执行多个流程 |
| 多个流程同时要运行 | 错开时间+队列管理 | 避免资源冲突 |
二、怎么做
2.1 配置定时任务
在影刀客户端配置：
- 打开影刀客户端
- 进入「定时任务」页面
- 点击「新建定时任务」
- 选择要运行的流程
- 设置触发时间和频率
- 保存
触发频率设置：
拼多多店群自动化报活动上架！
每天：选择具体时间，如09:00 每周：选择星期几和时间，如周一09:00 每月：选择日期和时间，如每月1号09:00 间隔：设置间隔时间，如每2小时 自定义：用Cron表达式（高级）2.2 工作日判断（周末不运行）
影刀的定时任务本身不支持"仅工作日"的设置。需要在流程内部加判断：
importdatetimedefis_workday():"""判断今天是否是工作日（简单版）"""today=datetime.date.today()# 0=周一, 6=周日weekday=today.weekday()# 周六(5)和周日(6)不运行ifweekday>=5:returnFalsereturnTrueifnotis_workday():print("今天是周末，跳过执行")# 使用影刀的【退出流程】指令退出else:print("今天是工作日，正常执行")更准确的工作日判断（考虑法定节假日）：
importdatetimedefis_workday_accurate():"""使用chinesecalendar判断工作日"""try:fromchinese_calendarimportis_workday today=datetime.date.today()returnis_workday(today)exceptImportError:print("未安装chinesecalendar，使用基础判断")# 降级为基础判断weekday=datetime.date.today().weekday()returnweekday<5ifnotis_workday_accurate():print("今天不是工作日（节假日或周末），跳过执行")# 退出流程else:print("今天是工作日，正常执行")2.3 文件触发模式
有些场景不是按时间触发，而是"文件到了就处理"：
importosimporttimeimportdatetimedefwait_for_file(file_path,check_interval=60,max_wait=3600):""" 等待文件出现 check_interval: 检查间隔（秒） max_wait: 最大等待时间（秒） """start_time=time.time()whiletime.time()-start_time<max_wait:ifos.path.exists(file_path):# 检查文件是否还在写入（文件大小稳定）size1=os.path.getsize(file_path)time.sleep(2)size2=os.path.getsize(file_path)ifsize1==size2andsize1>0:print(f"文件已就绪:{file_path}({size1}bytes)")returnTruetime.sleep(check_interval)print(f"等待超时:{file_path}({max_wait}秒)")returnFalse# 等待数据文件出现（最多等1小时）# if wait_for_file(r"D:\data\input\today_data.csv", check_interval=60, max_wait=3600):# print("开始处理数据...")# else:# print("文件未出现，流程结束")2.4 流程链式调用
一个流程完成后自动启动下一个流程：
importsubprocessimporttimedefrun_flow_chain(flows):"""串行运行多个流程"""results=[]fori,flowinenumerate(flows):print(f"\n[{i+1}/{len(flows)}] 正在运行:{flow['name']}")# 运行流程（用影刀的命令行接口）# 这里用模拟的命令，实际根据影刀的CLI调整try:result=subprocess.run(flow['command'],shell=True,capture_output=True,text=True,timeout=flow.get('timeout',3600))status="成功"ifresult.returncode==0else"失败"print(f"结果:{status}")results.append({'name':flow['name'],'status':status,'returncode':result.returncode})# 如果失败，决定是否继续ifresult.returncode!=0:ifflow.get('stop_on_fail',False):print(f"流程失败，停止后续执行")breakelse:print(f"流程失败，继续执行下一个")# 流程间间隔time.sleep(flow.get('interval',5))exceptsubprocess.TimeoutExpired:print(f"流程超时:{flow['name']}")results.append({'name':flow['name'],'status':'超时'})ifflow.get('stop_on_fail',False):break# 输出汇总print(f"\n{'='*50}")print("执行汇总:")forrinresults:print(f"{r['name']}:{r['status']}")returnresults# 使用示例# flows = [# {"name": "数据采集", "command": "yingdao run flow1", "timeout": 1800, "stop_on_fail": True},# {"name": "数据处理", "command": "yingdao run flow2", "timeout": 600},# {"name": "生成报表", "command": "yingdao run flow3", "timeout": 300},# {"name": "发送邮件", "command": "yingdao run flow4", "timeout": 120},# ]# run_flow_chain(flows)2.5 定时任务管理器
多个定时任务需要统一管理时，写一个管理器：
importjsonimportosfromdatetimeimportdatetime,timedeltaclassScheduleManager:"""定时任务管理器"""def__init__(self,config_path="schedule_config.json"):self.config_path=config_path self.schedules=self._load_schedules()def_load_schedules(self):"""加载定时任务配置"""ifos.path.exists(self.config_path):withopen(self.config_path,'r',encoding='utf-8')asf:returnjson.load(f)# 默认配置default=[{"name":"日报生成","flow":"日报生成_v2","schedule":"daily 09:00","workday_only":True,"enabled":True,"last_run":None,"last_status":None},{"name":"数据采集","flow":"商品采集_v3","schedule":"daily 08:00, 14:00, 20:00","workday_only":False,"enabled":True,"last_run":None,"last_status":None},{"name":"周报生成","flow":"周报汇总_v1","schedule":"weekly monday 18:00","workday_only":True,"enabled":True,"last_run":None,"last_status":None}]withopen(self.config_path,'w',encoding='utf-8')asf:json.dump(default,f,ensure_ascii=False,indent=2)returndefaultdefcheck_and_run(self):"""检查并运行到期的任务"""now=datetime.now()print(f"检查时间:{now.strftime('%Y-%m-%d %H:%M:%S')}")fortaskinself.schedules:ifnottask['enabled']:continueiftask['workday_only']andnotself._is_workday(now):continueifself._should_run(task,now):print(f" → 运行任务:{task['name']}({task['flow']})")# 这里调用影刀的流程运行接口task['last_run']=now.strftime('%Y-%m-%d %H:%M:%S')task['last_status']="success"# 实际运行后更新# 保存状态self._save_schedules()def_is_workday(self,dt):"""判断是否工作日"""returndt.weekday()<5def_should_run(self,task,now):"""判断任务是否应该运行"""# 简化版判断逻辑schedule=task['schedule']current_time=now.strftime('%H:%M')ifschedule.startswith('daily'):times=schedule.replace('daily ','').split(', ')returncurrent_timein[t.strip()fortintimes]elifschedule.startswith('weekly'):parts=schedule.split()day_name=parts[1].lower()time_str=parts[2]days=['monday','tuesday','wednesday','thursday','friday','saturday','sunday']ifday_nameindaysandnow.weekday()==days.index(day_name):returncurrent_time==time_strreturnFalsedef_save_schedules(self):"""保存配置"""withopen(self.config_path,'w',encoding='utf-8')asf:json.dump(self.schedules,f,ensure_ascii=False,indent=2)deflist_tasks(self):"""列出所有任务"""print(f"\n{'任务名':<15}{'流程':<20}{'计划':<30}{'状态':<8}{'上次运行'}")print("-"*90)fortaskinself.schedules:status="启用"iftask['enabled']else"禁用"print(f"{task['name']:<15}{task['flow']:<20}{task['schedule']:<30}{status:<8}{task.get('last_run','未运行')}")# 使用示例# sm = ScheduleManager()# sm.list_tasks()# sm.check_and_run()三、有什么坑
坑1：电脑休眠导致定时任务不触发
电脑进入休眠状态后，定时任务不会触发。等电脑唤醒后，错过的任务也不会补执行。
解决：
- 在Windows电源设置里禁止休眠（至少在运行时间窗口内不休眠）
- 或者在BIOS里设置定时唤醒
# 用Python禁止系统休眠（运行期间）importctypes# ES_CONTINUOUS | ES_SYSTEM_AWAKEctypes.windll.kernel32.SetThreadExecutionState(0x80000002)print("已禁止系统休眠")# 流程结束后恢复# ctypes.windll.kernel32.SetThreadExecutionState(0x80000000)坑2：多个定时任务同时运行导致冲突
两个采集流程同时运行，可能都去操作同一个浏览器，互相干扰。或者同时写同一个Excel文件，导致文件锁冲突。
解决：错开任务时间，至少间隔5分钟。或者在流程开头加锁：
TEMU店群矩阵自动化运营核价报活动
importosimporttimedefacquire_lock(lock_file,timeout=300):"""获取文件锁"""start=time.time()whiletime.time()-start<timeout:ifnotos.path.exists(lock_file):# 创建锁文件withopen(lock_file,'w')asf:f.write(str(os.getpid()))print(f"已获取锁:{lock_file}")returnTrueelse:# 检查锁是否过期（超过10分钟）iftime.time()-os.path.getmtime(lock_file)>600:os.remove(lock_file)continueprint("等待其他流程释放锁...")time.sleep(10)print("获取锁超时")returnFalsedefrelease_lock(lock_file):"""释放锁"""ifos.path.exists(lock_file):os.remove(lock_file)print(f"已释放锁:{lock_file}")# 使用# lock = r"D:\locks\采集任务.lock"# if acquire_lock(lock):# try:# # 执行采集流程# pass# finally:# release_lock(lock)坑3：定时任务运行失败但没人知道
定时任务在后台运行，失败了如果不看日志根本不知道。
解决：在流程末尾加通知（参考上一篇通知文章）：
# 流程末尾的通知逻辑flow_result={"status":"success",# 或 "failed""flow_name":"日报生成","error":"","duration":180}ifflow_result["status"]!="success":# 发送失败通知notify("error",f"流程失败:{flow_result['flow_name']}",flow_result["error"])else:# 成功只记录日志print(f"流程成功完成:{flow_result['flow_name']}")坑4：流程执行时间超过预期，和下一个任务冲突
流程本来预计10分钟跑完，结果遇到网络慢跑了30分钟，下一个定时任务又开始了。
解决：在流程开头检查是否已有实例在运行（用上面的文件锁机制），如果有就跳过本次执行。
坑5：时区问题
服务器时区和本地时区不一致，导致定时任务触发时间不对。
解决：在流程里打印当前时间确认：
importdatetimeprint(f"当前时间:{datetime.datetime.now()}")print(f"UTC时间:{datetime.datetime.utcnow()}")print(f"时区:{datetime.datetime.now().astimezone().tzinfo}")确认服务器时区设置正确。影刀定时任务用的是本地时间（服务器所在时区）。
总结
定时任务管理的核心要点：
禁止休眠——确保电脑在任务触发时是醒着的
加文件锁——防止多个任务同时操作同一资源
失败要通知——后台运行的任务失败了一定要推送通知
错开时间——多个任务至少间隔5分钟
工作日判断放流程里——定时任务本身不支持工作日过滤
