行业资讯
影刀RPA 时间窗口控制：工作时间执行与业务时段判断
影刀RPA 时间窗口控制工作时间执行与业务时段判断作者林焱什么情况用这个有些自动化任务不是随时都能跑的。发客户通知的邮件最好在上午9-12点发采集电商数据要避开平台维护时间凌晨2-5点付款操作只能是工作日的工作时间。如果你的流程半夜自动运行并发了一堆通知短信——等着被客户投诉吧。时间窗口控制就是给你的流程加一个智能开关判断当前是不是合适的执行时间不是就等着、或者跳过今天的执行。这篇文章覆盖了所有你会在实际业务中遇到的时间窗口判断需求。怎么做第一步最基础的时间窗口——工作时间判断fromdatetimeimportdatetime,timedefis_working_hours():判断当前是否是工作时间周一到周五 9:00-18:00nowdatetime.now()# 周末判断ifnow.weekday()5:# 5周六, 6周日returnFalse,周末不执行# 工作时间判断current_timenow.time()ifcurrent_timetime(9,0):returnFalse,未到工作时间9:00开始ifcurrent_timetime(18,0):returnFalse,已过工作时间18:00结束returnTrue,工作时间# 在流程中使用can_run,reasonis_working_hours()ifnotcan_run:print(f当前不可执行:{reason}流程跳过)exit()第二步灵活的时间窗——多段工作时间有些业务有午休时间需要避开拼多多店群自动化报活动上架defis_in_time_windows(windows): 判断当前是否在指定的时间窗口内 windows: [(开始时间, 结束时间), ...] nowdatetime.now()current_timenow.time()forstart_str,end_strinwindows:starttime.fromisoformat(start_str)endtime.fromisoformat(end_str)ifstartcurrent_timeend:returnTruereturnFalse# 定义多个时间窗口WORK_WINDOWS[(09:00,12:00),# 上午(14:00,18:00),# 下午避开午休]ifnotis_in_time_windows(WORK_WINDOWS):print(当前不在工作时间窗口内)第三步计算距离下一个时间窗口还要等多久defwait_until_next_window(windows,max_wait_hours4): 如果不在窗口内计算需要等多久 如果等待时间超过max_wait_hours建议跳过本次执行 nowdatetime.now()current_timenow.time()forstart_str,_inwindows:start_timetime.fromisoformat(start_str)ifcurrent_timestart_time:# 还没到这个窗口——计算等待时间targetnow.replace(hourstart_time.hour,minutestart_time.minute,second0)wait_seconds(target-now).total_seconds()wait_hourswait_seconds/3600ifwait_hoursmax_wait_hours:print(f等待{wait_hours:.1f}小时到下一个窗口...)importtimeas_time _time.sleep(wait_seconds)returnTrueelse:print(f等待时间过长({wait_hours:.1f}小时)跳过本次执行)returnFalse# 检查是否在最后一个窗口之后_,last_endwindows[-1]last_end_timetime.fromisoformat(last_end)ifcurrent_timelast_end_time:# 今天的窗口都过去了等明天print(今天的时间窗口已过建议明天执行)returnFalsereturnTrue# 已经在窗口内第四步避开特定时间段反时间窗口有些业务需要不在某段时间执行defis_blocked_time(): 判断当前是否在禁止执行的时间段 比如系统维护时间、平台封禁高发时段 nowdatetime.now()current_timenow.time()# 凌晨2-5点平台维护不稳定iftime(2,0)current_timetime(5,0):returnTrue,平台维护时段# 整点前后2分钟很多系统在整点做定时任务不稳定ifnow.minute58ornow.minute2:returnTrue,整点波动期# 每月1号零点前后月结系统压力大ifnow.day1andcurrent_timetime(1,0):returnTrue,月初结算期returnFalse,blocked,reasonis_blocked_time()ifblocked:# 不用等了——直接跳过这次执行print(f当前处于禁止时段:{reason}跳过)第五步工作日判断含法定节假日判断工作日不只是看周一到周五还要避开法定假日# 中国法定节假日需要每年更新HOLIDAYS_2025{01-01:元旦,01-28:春节,01-29:春节,01-30:春节,01-31:春节,02-01:春节,02-02:春节,02-03:春节,02-04:春节,04-04:清明节,04-05:清明节,04-06:清明节,05-01:劳动节,05-02:劳动节,05-03:劳动节,05-04:劳动节,05-05:劳动节,05-31:端午节,06-01:端午节,06-02:端午节,10-01:国庆节,10-02:国庆节,10-03:国庆节,10-04:国庆节,10-05:国庆节,10-06:国庆节,10-07:国庆节,10-08:国庆节,}# 调休上班日周末但要上班WORKDAYS_2025{01-26:春节调休,02-08:春节调休,04-27:五一调休,09-28:国庆调休,10-11:国庆调休,}defis_workday():判断今天是否是工作日nowdatetime.now()date_keynow.strftime(%m-%d)weekdaynow.weekday()# 调休上班日优先判断ifdate_keyinWORKDAYS_2025:returnTrue,f调休工作日{WORKDAYS_2025[date_key]}# 法定节假日ifdate_keyinHOLIDAYS_2025:returnFalse,f法定假日{HOLIDAYS_2025[date_key]}# 正常周末ifweekday5:returnFalse,周末returnTrue,工作日第六步综合判断——完整的执行许可把以上组合成一个完整的决策函数classExecutionScheduler:执行调度器——判断现在能不能跑def__init__(self):self.work_windows[(09:00,12:00),(14:00,18:00)]self.blocked_times[(02:00,05:00)]defcan_execute_now(self,task_typenormal): 判断是否可以立即执行 task_type: normal | urgent | silent - normal: 只工作时间执行 - urgent: 任何时候都可以 - silent: 工作时间但不发通知 nowdatetime.now()# 紧急任务不检查时间窗口iftask_typeurgent:returnTrue,紧急任务立即执行# 检查是否工作日is_wd,wd_reasonis_workday()ifnotis_wd:returnFalse,wd_reason# 检查是否在禁止时段current_timenow.time()forstart_str,end_strinself.blocked_times:starttime.fromisoformat(start_str)endtime.fromisoformat(end_str)ifstartcurrent_timeend:returnFalse,f禁止时段:{start_str}-{end_str}# 检查是否在工作窗口in_windowany(time.fromisoformat(s)current_timetime.fromisoformat(e)fors,einself.work_windows)ifnotin_window:returnFalse,不在工作时间窗口内returnTrue,可以执行# 使用schedulerExecutionScheduler()can_run,reasonscheduler.can_execute_now(task_typenormal)ifcan_run:print(f✓ 开始执行 ({reason}))else:print(f✗ 不执行 ({reason}))有什么坑坑一只判断了时间没判断日期现象周末没排除周六上午9点照样给客户发催款通知。解决时间窗口必须同时检查星期几或者直接用is_workday()函数。坑二使用服务器时间而不是北京时间现象部署在云服务器上UTC时区datetime.now()是UTC时间9点到18点变成了北京时间17点到凌晨2点。解决显式指定时区TEMU店群矩阵自动化运营核价报活动fromdatetimeimportdatetime,timezone,timedelta# 获取北京时间cn_tztimezone(timedelta(hours8))cn_nowdatetime.now(cn_tz)坑三法定假日表没更新现象去年写的假日表今年还在用清明节没排除。解决每年年初更新假日表。更好的做法是从API获取如调用国务院的放假通知接口。坑四sleep等待跨越了时间边界现象在11:58分开始sleep等12:00窗口结果醒来时刚好错过了中午窗口的截止时间。解决等待后重新检查一次窗口状态time.sleep(wait_seconds)# 醒来后再检查一次can_run,reasonscheduler.can_execute_now()ifnotcan_run:print(f等待后已过窗口期:{reason})坑五跨天执行没有重新判断现象流程在23:50开始跑跑了20分钟到了00:10流程还在用开始时的判断结果。解决长流程中在关键步骤前重新检查时间窗口。总结时间窗口判断的核心就两条规则——工作日工作时间。先把这两条实现清楚再根据实际业务需要加禁止时段和特殊日期。判断逻辑写到流程入口处不符合条件直接跳过今天的执行不要等到跑了半小时才发现不该跑。
郑州网站建设
网页设计
企业官网
