影刀RPA 时间日期处理：定时任务与日期计算完全攻略

作者：林焱

什么情况用

RPA流程几乎离不开时间处理：每天凌晨2点自动执行、只处理最近7天的数据、计算两个日期之间的工作日天数、将「3天前」「上周五」这种自然语言转成具体日期。

影刀内置了「定时运行」和「等待」节点，但遇到复杂的日期计算——比如「本月第三个工作日」、「排除节假日的截止日期」——就需要Python的datetime模块出手。

核心场景：流程中涉及日期计算、定时调度、工作日统计等。

怎么做

第一步：datetime模块速查

fromdatetimeimportdatetime,date,timedeltaimportcalendar# ===== 获取当前时间 =====now=datetime.now()# 2024-06-26 14:30:00.123456today=date.today()# 2024-06-26# ===== 创建指定日期 =====d=datetime(2024,6,26,14,30,0)d=datetime.strptime("2024-06-26","%Y-%m-%d")# ===== 格式化输出 =====print(now.strftime("%Y-%m-%d %H:%M:%S"))# 2024-06-26 14:30:00print(now.strftime("%Y年%m月%d日"))# 2024年06月26日print(now.strftime("%A"))# Wednesday（星期几）# ===== 日期加减 =====yesterday=today-timedelta(days=1)next_week=today+timedelta(days=7)one_hour_ago=now-timedelta(hours=1)three_months_later=today+timedelta(days=90)# 粗略用90天代替3个月# ===== 日期差 =====diff=datetime(2024,12,31)-datetime(2024,1,1)print(f"相差{diff.days}天")# 相差 365 天

第二步：常用日期计算函数库

classDateUtils:"""RPA常用日期工具"""@staticmethoddefget_this_week_range():"""获取本周一和周日的日期"""today=date.today()monday=today-timedelta(days=today.weekday())sunday=monday+timedelta(days=6)returnmonday,sunday@staticmethoddefget_this_month_range():"""获取本月第一天和最后一天"""today=date.today()first_day=today.replace(day=1)last_day=today.replace(day=calendar.monthrange(today.year,today.month)[1])returnfirst_day,last_day@staticmethoddefget_last_month_range():"""获取上月第一天和最后一天"""today=date.today()# 上月最后一天 = 本月第一天 - 1天first_of_this_month=today.replace(day=1)last_of_last_month=first_of_this_month-timedelta(days=1)first_of_last_month=last_of_last_month.replace(day=1)returnfirst_of_last_month,last_of_last_month@staticmethoddefget_quarter_range():"""获取本季度第一天和最后一天"""today=date.today()quarter_month=((today.month-1)//3)*3+1first_day=today.replace(month=quarter_month,day=1)last_month=quarter_month+2last_day=today.replace(month=last_month,day=calendar.monthrange(today.year,last_month)[1])returnfirst_day,last_day@staticmethoddefis_weekend(d):"""判断是否是周末"""returnd.weekday()>=5@staticmethoddefadd_workdays(start_date,days):"""增加N个工作日（跳过周末）"""current=start_date added=0whileadded<days:current+=timedelta(days=1)ifcurrent.weekday()<5:# 周一到周五added+=1returncurrent@staticmethoddefcount_workdays(start_date,end_date):"""计算两个日期之间的工作日天数"""days=0current=start_datewhilecurrent<=end_date:ifcurrent.weekday()<5:days+=1current+=timedelta(days=1)returndays@staticmethoddefget_recent_days(num_days):"""获取最近N天的日期列表"""today=date.today()return[(today-timedelta(days=i)).strftime("%Y-%m-%d")foriinrange(num_days)]@staticmethoddefparse_relative_date(text):""" 解析相对日期表达 "今天" → 今天日期 "昨天" → 昨天日期 "3天前" → 3天前的日期 "上周一" → 上周一日期 """today=date.today()text=text.strip()iftextin("今天","今日"):returntodayiftextin("昨天","昨日"):returntoday-timedelta(days=1)iftextin("前天",):returntoday-timedelta(days=2)iftextin("明天","明日"):returntoday+timedelta(days=1)importre# N天前 / N天后match=re.match(r'(\d+)\s*天[前后]',text)ifmatch:n=int(match.group(1))if"后"intext:returntoday+timedelta(days=n)else:returntoday-timedelta(days=n)# 上周Xweekdays={"一":0,"二":1,"三":2,"四":3,"五":4,"六":5,"日":6}match=re.match(r'上周([一二三四五六日])',text)ifmatch:target=weekdays[match.group(1)]days_since_monday=today.weekday()last_monday=today-timedelta(days=days_since_monday+7)returnlast_monday+timedelta(days=target)returnNone# 测试du=DateUtils()print(f"本周：{du.get_this_week_range()}")print(f"本月：{du.get_this_month_range()}")print(f"10个工作日后：{du.add_workdays(date.today(),10)}")print(f"本月工作日：{du.count_workdays(*du.get_this_month_range())}")print(f"「3天前」：{du.parse_relative_date('3天前')}")print(f"「上周五」：{du.parse_relative_date('上周五')}")

第三步：影刀定时任务的最佳实践

影刀支持cron定时，但有几个常见需求需要自定义：

importschedule# pip install scheduleimporttimeclassTaskScheduler:"""自定义定时任务调度器"""def__init__(self):self.tasks=[]defrun_at(self,time_str,task_func,*args):"""每天在指定时间运行"""schedule.every().day.at(time_str).do(task_func,*args)defrun_every_n_minutes(self,minutes,task_func,*args):"""每N分钟运行"""schedule.every(minutes).minutes.do(task_func,*args)defrun_at_weekday(self,weekday,time_str,task_func,*args):"""每周某天某时运行"""day_map={"monday":schedule.every().monday,"tuesday":schedule.every().tuesday,"wednesday":schedule.every().wednesday,"thursday":schedule.every().thursday,"friday":schedule.every().friday,"saturday":schedule.every().saturday,"sunday":schedule.every().sunday}day_map[weekday.lower()].at(time_str).do(task_func,*args)defrun_on_month_day(self,day,time_str,task_func,*args):"""每月某天某时运行（如每月1号凌晨生成报表）"""# schedule库不直接支持「每月第N天」# 变通方案：每天检查是否是目标日期defwrapper():ifdatetime.now().day==day:task_func(*args)schedule.every().day.at(time_str).do(wrapper)defstart(self):"""开始调度循环"""print("定时调度器已启动...")whileTrue:schedule.run_pending()time.sleep(30)# 每30秒检查一次# 示例——生成报表的调度# scheduler = TaskScheduler()# scheduler.run_at("08:00", generate_daily_report) # 每天8点生成日报# scheduler.run_at_weekday("monday", "09:00", generate_weekly_report) # 每周一9点生成周报# scheduler.run_on_month_day(1, "08:00", generate_monthly_report) # 每月1号8点生成月报# scheduler.start()

第四步：超时控制——别让流程无限等待

importthreadingdefrun_with_timeout(func,timeout_seconds,*args,**kwargs):""" 给函数调用加超时限制 超时后抛出 TimeoutError """result=[None]exception=[None]deftarget():try:result[0]=func(*args,**kwargs)exceptExceptionase:exception[0]=e thread=threading.Thread(target=target)thread.daemon=Truethread.start()![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/32f7d9b08bac4690b19f3b0dafe7cbeb.png#pic_center)thread.join(timeout=timeout_seconds)ifthread.is_alive():raiseTimeoutError(f"操作超时（{timeout_seconds}秒）")ifexception[0]:raiseexception[0]returnresult[0]# 使用try:data=run_with_timeout(fetch_large_dataset,120,url="https://api.example.com")print(f"获取到{len(data)}条数据")exceptTimeoutError:print("数据获取超时，跳过本次处理")

第五步：生成时间戳文件名

defgenerate_timestamp_filename(prefix="report",ext="xlsx"):"""生成带时间戳的文件名，防止覆盖"""timestamp=datetime.now().strftime("%Y%m%d_%H%M%S")returnf"{prefix}_{timestamp}.{ext}"# report_20240626_143000.xlsx# 唯一的文件名 → 多次运行不会互相覆盖

另外要注意：strftime里的%f（微秒）非常有用，高频率运行时避免同名冲突。

有什么坑

坑1：时区——datetime.now()不一定是北京时间

如果你的服务器时区设置不对，datetime.now()返回的可能不是北京时间。凌晨2点触发的任务可能下午才跑。

解决方法：

fromdatetimeimporttimezone,timedelta# 明确使用北京时间beijing_tz=timezone(timedelta(hours=8))beijing_now=datetime.now(beijing_tz)

坑2：月份加减的天数陷阱

timedelta(days=30)不等于「一个月后」。1月31日 + 30天 = 3月2日，但你可能期望是2月28日。

fromdateutil.relativedeltaimportrelativedelta# pip install python-dateutil# ✅ 正确的月份加减next_month=datetime(2024,1,31)+relativedelta(months=1)print(next_month)# 2024-02-29 —— 自动处理月末three_months_ago=datetime.now()-relativedelta(months=3)

坑3：strftime不支持中文

# ❌ 不能直接在strftime里写中文# now.strftime("%Y年%m月%d日") # 这个可以，因为中文在格式字符串里# now.strftime("星期%A") # 这个不行，%A输出的是英文星期几# ✅ 用字典映射WEEKDAY_CN={0:"周一",1:"周二",2:"周三",3:"周四",4:"周五",5:"周六",6:"周日"}weekday_cn=WEEKDAY_CN[datetime.now().weekday()]

坑4：定时任务的累积效应

如果定时任务执行时间超过了间隔时间（比如每5分钟运行一次，但单次执行要8分钟），任务会堆积。

解决方法：在执行前检查是否有上一次任务还在跑。

总结：RPA的时间处理远不止「定时运行」四个字。核心技能：timedelta日期加减、relativedelta月份处理、工作日计算、自然语言日期解析、超时控制。把这些封装成工具函数随用随取，别每次从零写。