天天看點

python定時任務_python 定時任務apscheduler

python定時任務_python 定時任務apscheduler

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()

如果本文對您有所幫助,請支援下本站哦!!!^_^