场景:
即日起截止到2019年国庆节前,每周工作日的8点-20点,去指定目录查看一个文件当日是否存在,如果存在就拿来使用然后删除文件,如果不存在则每隔15分钟循环执行这个判断,直到文件出现在目录下,如果超过20点文件依然没有出现则不再执行循环。简易流程图如下所示。这里就模拟去桌面查找test.xlsx是否存在。
处理过程:
代码如下所示,因为是个定时任务,这里就不展示结果了。不知道怎么在APScheduler设置任务的结束时间,所以把结束时间加到if语句里面进行判断。
from apscheduler.schedulers.blocking import BlockingScheduler
import schedule
import os
from datetime import datetime
import time
def find_job():
# 实际工作中这里需要替换成需要执行的.py文件
print('execute ****.py %s' % datetime.now())
def remove_file():
# 移除目录下的文件
os.remove(dirpath + '\\' + file)
# 验证文件是否已经从目录下移除
if file in os.listdir(dirpath):
print('remove file fail...')
else:
print('remove file success!')
def not_find_job():
print('continue find ... %s' % datetime.now())
# 把not_find_job函数加入到调度中,设置时间为每隔15分钟执行一次
schedule.every(15).minutes.do(not_find_job)
def myfun():
# 判断当前时间是否超过20点,不超过则执行后续语句块,超过则什么也不做
if datetime.now().strftime('%H:%M') < '20:00':
# 使用while是因为:如果条件满足要求有出口可以退出结束当天的调度任务。
while file not in os.listdir(dirpath):
schedule.run_pending() # 如果文件不在目录下则保持schedule一直运行去查询上面设置的任务
time.sleep(15*60) # 休眠15分钟
else:
find_job()
remove_file()
if __name__ == '__main__':
dirpath = 'C:\Users\***\Desktop'
file = 'test.xlsx'
schedule = BlockingScheduler() # 实例化
# 把上面写的每天需要定时执行的任务添加进来,设置各种参数
schedule.add_job(myfun, 'cron', day_of_week='mon-fri', hour=8, minutes=0, end_date='2019-01-01')
# 开始执行每天的定时任务
schedule.start()
如果是在本地执行会有一个问题:电脑不能关机,关机的话定时任务就关掉了,开机后Python不自动重启这个任务。解决这个问题的方法:部署到服务器上。
这么解决这个场景,我自己感觉不够简洁和简单,但是又不知道其他更好的方法,如果你有更方便的方法和技巧,请让我也了解和学习吧。
更多推荐
Python APScheduler定时执行任务:调度(schedule)一个能退出有出口的循环类型
发布评论