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()
如果本文對您有所幫助,請支援下本站哦!!!^_^