import os
import apscheduler
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger
import logging
import time
class Shelduler:
task_index = 0
wait_index = 0
def __init__(self):
self.scheduler = BackgroundScheduler()
logging.basicConfig(level=logging.ERROR,
format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
filename='log1.txt',
filemode='a')
def run(self):
print("run init")
scheduler = self.scheduler
mytime = "00:00:00"
times = mytime.split(":")
step = 5
# 间隔5秒钟执行一次
trigger = IntervalTrigger(seconds=step)
scheduler.add_job(self.mytask, trigger)
scheduler.add_listener(self.myListener,
apscheduler.events.EVENT_JOB_EXECUTED | apscheduler.events.EVENT_JOB_ERROR)
trigger = CronTrigger(day_of_week="0-6", hour=int(times[0]), minute=int(times[1]), second=int(times[2]))
scheduler.add_job(self.mytask2, trigger)
scheduler.start()
print('按下 Ctrl+{0} 退出'.format("Break" if os.name !='nt' else 'C'))
try:
# 其他任务 独立的线程执行
while True:
time.sleep(2)
self.wait_index += 1
print("\r" + "waitting" + "..." * self.wait_index, end="OK")
self.task_index += 1
if self.task_index % 10 == 0: # 执行10次 清空屏幕
os.system("cls")
self.task_index = 0
self.wait_index = 0
except (KeyboardInterrupt, SystemExit):
scheduler.shutdown()
print('\n退出任务!')
def myListener(self, ev):
if ev.exception:
logging.info('\n%s error.', str(ev.exception))
else:
print("执行成功")
def mytask(self):
print("\nmy task1")
self.wait_index = 0
def mytask2(self):
print("\nmy task2")
if __name__ == "__main__":
shed = Shelduler()
shed.run()
如果本文对您有所帮助,请支持下本站哦!!!^_^