影刀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():判断今天是否是工作日简单版todaydatetime.date.today()# 0周一, 6周日weekdaytoday.weekday()# 周六(5)和周日(6)不运行ifweekday5:returnFalsereturnTrueifnotis_workday():print(今天是周末跳过执行)# 使用影刀的【退出流程】指令退出else:print(今天是工作日正常执行)更准确的工作日判断考虑法定节假日importdatetimedefis_workday_accurate():使用chinesecalendar判断工作日try:fromchinese_calendarimportis_workday todaydatetime.date.today()returnis_workday(today)exceptImportError:print(未安装chinesecalendar使用基础判断)# 降级为基础判断weekdaydatetime.date.today().weekday()returnweekday5ifnotis_workday_accurate():print(今天不是工作日节假日或周末跳过执行)# 退出流程else:print(今天是工作日正常执行)2.3 文件触发模式有些场景不是按时间触发而是文件到了就处理importosimporttimeimportdatetimedefwait_for_file(file_path,check_interval60,max_wait3600): 等待文件出现 check_interval: 检查间隔秒 max_wait: 最大等待时间秒 start_timetime.time()whiletime.time()-start_timemax_wait:ifos.path.exists(file_path):# 检查文件是否还在写入文件大小稳定size1os.path.getsize(file_path)time.sleep(2)size2os.path.getsize(file_path)ifsize1size2andsize10:print(f文件已就绪:{file_path}({size1}bytes))returnTruetime.sleep(check_interval)print(f等待超时:{file_path}({max_wait}秒))returnFalse# 等待数据文件出现最多等1小时# if wait_for_file(rD:\data\input\today_data.csv, check_interval60, max_wait3600):# print(开始处理数据...)# else:# print(文件未出现流程结束)2.4 流程链式调用一个流程完成后自动启动下一个流程importsubprocessimporttimedefrun_flow_chain(flows):串行运行多个流程results[]fori,flowinenumerate(flows):print(f\n[{i1}/{len(flows)}] 正在运行:{flow[name]})# 运行流程用影刀的命令行接口# 这里用模拟的命令实际根据影刀的CLI调整try:resultsubprocess.run(flow[command],shellTrue,capture_outputTrue,textTrue,timeoutflow.get(timeout,3600))status成功ifresult.returncode0else失败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_pathschedule_config.json):self.config_pathconfig_path self.schedulesself._load_schedules()def_load_schedules(self):加载定时任务配置ifos.path.exists(self.config_path):withopen(self.config_path,r,encodingutf-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,encodingutf-8)asf:json.dump(default,f,ensure_asciiFalse,indent2)returndefaultdefcheck_and_run(self):检查并运行到期的任务nowdatetime.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):判断任务是否应该运行# 简化版判断逻辑scheduletask[schedule]current_timenow.strftime(%H:%M)ifschedule.startswith(daily):timesschedule.replace(daily ,).split(, )returncurrent_timein[t.strip()fortintimes]elifschedule.startswith(weekly):partsschedule.split()day_nameparts[1].lower()time_strparts[2]days[monday,tuesday,wednesday,thursday,friday,saturday,sunday]ifday_nameindaysandnow.weekday()days.index(day_name):returncurrent_timetime_strreturnFalsedef_save_schedules(self):保存配置withopen(self.config_path,w,encodingutf-8)asf:json.dump(self.schedules,f,ensure_asciiFalse,indent2)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,timeout300):获取文件锁starttime.time()whiletime.time()-starttimeout: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 rD:\locks\采集任务.lock# if acquire_lock(lock):# try:# # 执行采集流程# pass# finally:# release_lock(lock)坑3定时任务运行失败但没人知道定时任务在后台运行失败了如果不看日志根本不知道。解决在流程末尾加通知参考上一篇通知文章# 流程末尾的通知逻辑flow_result{status:success,# 或 failedflow_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(fUTC时间:{datetime.datetime.utcnow()})print(f时区:{datetime.datetime.now().astimezone().tzinfo})确认服务器时区设置正确。影刀定时任务用的是本地时间服务器所在时区。总结定时任务管理的核心要点禁止休眠——确保电脑在任务触发时是醒着的加文件锁——防止多个任务同时操作同一资源失败要通知——后台运行的任务失败了一定要推送通知错开时间——多个任务至少间隔5分钟工作日判断放流程里——定时任务本身不支持工作日过滤