影刀RPA 时间窗口控制:工作时间执行与业务时段判断
作者:林焱
什么情况用这个
有些自动化任务不是随时都能跑的。发客户通知的邮件最好在上午9-12点发,采集电商数据要避开平台维护时间(凌晨2-5点),付款操作只能是工作日的工作时间。如果你的流程半夜自动运行并发了一堆通知短信——等着被客户投诉吧。
时间窗口控制就是给你的流程加一个"智能开关":判断当前是不是合适的执行时间,不是就等着、或者跳过今天的执行。这篇文章覆盖了所有你会在实际业务中遇到的时间窗口判断需求。
怎么做
第一步:最基础的时间窗口——工作时间判断
fromdatetimeimportdatetime,timedefis_working_hours():"""判断当前是否是工作时间(周一到周五 9:00-18:00)"""now=datetime.now()# 周末判断ifnow.weekday()>=5:# 5=周六, 6=周日returnFalse,"周末不执行"# 工作时间判断current_time=now.time()ifcurrent_time<time(9,0):returnFalse,"未到工作时间(9:00开始)"ifcurrent_time>time(18,0):returnFalse,"已过工作时间(18:00结束)"returnTrue,"工作时间"# 在流程中使用can_run,reason=is_working_hours()ifnotcan_run:print(f"当前不可执行:{reason},流程跳过")exit()第二步:灵活的时间窗——多段工作时间
有些业务有午休时间需要避开:
拼多多店群自动化报活动上架!
defis_in_time_windows(windows):""" 判断当前是否在指定的时间窗口内 windows: [(开始时间, 结束时间), ...] """now=datetime.now()current_time=now.time()forstart_str,end_strinwindows:start=time.fromisoformat(start_str)end=time.fromisoformat(end_str)ifstart<=current_time<=end: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_hours=4):""" 如果不在窗口内,计算需要等多久 如果等待时间超过max_wait_hours,建议跳过本次执行 """now=datetime.now()current_time=now.time()forstart_str,_inwindows:start_time=time.fromisoformat(start_str)ifcurrent_time<start_time:# 还没到这个窗口——计算等待时间target=now.replace(hour=start_time.hour,minute=start_time.minute,second=0)wait_seconds=(target-now).total_seconds()wait_hours=wait_seconds/3600ifwait_hours<=max_wait_hours:print(f"等待{wait_hours:.1f}小时到下一个窗口...")importtimeas_time _time.sleep(wait_seconds)returnTrueelse:print(f"等待时间过长({wait_hours:.1f}小时),跳过本次执行")returnFalse# 检查是否在最后一个窗口之后_,last_end=windows[-1]last_end_time=time.fromisoformat(last_end)ifcurrent_time>last_end_time:# 今天的窗口都过去了,等明天print("今天的时间窗口已过,建议明天执行")returnFalsereturnTrue# 已经在窗口内第四步:避开特定时间段(反时间窗口)
有些业务需要"不在某段时间"执行:
defis_blocked_time():""" 判断当前是否在禁止执行的时间段 比如:系统维护时间、平台封禁高发时段 """now=datetime.now()current_time=now.time()# 凌晨2-5点:平台维护,不稳定iftime(2,0)<=current_time<=time(5,0):returnTrue,"平台维护时段"# 整点前后2分钟:很多系统在整点做定时任务,不稳定ifnow.minute>=58ornow.minute<=2:returnTrue,"整点波动期"# 每月1号零点前后:月结,系统压力大ifnow.day==1andcurrent_time<=time(1,0):returnTrue,"月初结算期"returnFalse,""blocked,reason=is_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():"""判断今天是否是工作日"""now=datetime.now()date_key=now.strftime("%m-%d")weekday=now.weekday()# 调休上班日优先判断ifdate_keyinWORKDAYS_2025:returnTrue,f"调休工作日({WORKDAYS_2025[date_key]})"# 法定节假日ifdate_keyinHOLIDAYS_2025:returnFalse,f"法定假日({HOLIDAYS_2025[date_key]})"# 正常周末ifweekday>=5: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_type="normal"):""" 判断是否可以立即执行 task_type: "normal" | "urgent" | "silent" - normal: 只工作时间执行 - urgent: 任何时候都可以 - silent: 工作时间但不发通知 """now=datetime.now()# 紧急任务:不检查时间窗口iftask_type=="urgent":returnTrue,"紧急任务,立即执行"# 检查是否工作日is_wd,wd_reason=is_workday()ifnotis_wd:returnFalse,wd_reason# 检查是否在禁止时段current_time=now.time()forstart_str,end_strinself.blocked_times:start=time.fromisoformat(start_str)end=time.fromisoformat(end_str)ifstart<=current_time<=end:returnFalse,f"禁止时段:{start_str}-{end_str}"# 检查是否在工作窗口in_window=any(time.fromisoformat(s)<=current_time<=time.fromisoformat(e)fors,einself.work_windows)ifnotin_window:returnFalse,"不在工作时间窗口内"returnTrue,"可以执行"# 使用scheduler=ExecutionScheduler()can_run,reason=scheduler.can_execute_now(task_type="normal")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_tz=timezone(timedelta(hours=8))cn_now=datetime.now(cn_tz)坑三:法定假日表没更新
现象:去年写的假日表,今年还在用,清明节没排除。
解决:每年年初更新假日表。更好的做法是从API获取(如调用国务院的放假通知接口)。
坑四:sleep等待跨越了时间边界
现象:在11:58分开始sleep等12:00窗口,结果醒来时刚好错过了中午窗口的截止时间。
解决:等待后重新检查一次窗口状态:
time.sleep(wait_seconds)# 醒来后再检查一次can_run,reason=scheduler.can_execute_now()ifnotcan_run:print(f"等待后已过窗口期:{reason}")坑五:跨天执行没有重新判断
现象:流程在23:50开始跑,跑了20分钟到了00:10,流程还在用"开始时的判断结果"。
解决:长流程中在关键步骤前重新检查时间窗口。
总结:时间窗口判断的核心就两条规则——工作日+工作时间。先把这两条实现清楚,再根据实际业务需要加禁止时段和特殊日期。判断逻辑写到流程入口处,不符合条件直接跳过今天的执行,不要等到跑了半小时才发现不该跑。