#encoding:utf-8

from apscheduler.schedulers.blocking import BlockingScheduler
import datetime

def sch_test():
    now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print('時(shí)間：{}, 測(cè)試apscheduler'.format(now))

task = BlockingScheduler()
task.add_job(func=sch_test, trigger='cron', second='*/10')
task.start()

# 使用run_date指定運(yùn)行時(shí)間
task.add_job(func='sch_test', trigger='date', run_date=datetime.datetime(2022 ,10 , 8, 16, 1, 30))
# 或者用next_run_time
task.add_job(func=sch_test,trigger='date', next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=3))

# encoding:utf-8
from apscheduler.schedulers.blocking import BlockingScheduler
import datetime

def sch_test(job_type):
    now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print('時(shí)間：{}, {}測(cè)試apscheduler'.format(now, job_type))

task = BlockingScheduler()
task.add_job(func=sch_test, args=('一次性任務(wù)',),trigger='date', next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=3))
task.add_job(func=sch_test, args=('定時(shí)任務(wù)',), trigger='cron', second='*/5')
task.add_job(func=sch_test, args=('循環(huán)任務(wù)',), trigger='interval', seconds=3)
task.start()

from apscheduler.schedulers.blocking import BlockingScheduler

def sch_test(job_type):
    now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print('時(shí)間：{}, {}測(cè)試apscheduler'.format(now, job_type))

sched = BlockingScheduler()
# 使用mysql存儲(chǔ)任務(wù)
sql_url = 'mysql+pymysql://root:root@localhost:3306/db_name?charset=utf8'
sched.add_jobstore('sqlalchemy',url=sql_url)
# 添加任務(wù)
sched.add_job(func=sch_test, args=('定時(shí)任務(wù)',), trigger='cron', second='*/5')
sched.start()

from apscheduler.schedulers.blocking import BlockingScheduler
import datetime
from apscheduler.events import EVENT_JOB_EXECUTED , EVENT_JOB_ERROR
import logging

# logging日志配置打印格式及保存位置
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
                    datefmt='%Y-%m-%d %H:%M:%S',
                    filename='sche.log',
                    filemode='a')

def log_listen(event):
	if event.exception :
		print ( '任務(wù)出錯(cuò)，報(bào)錯(cuò)信息：{}'.format(event.exception))
	else:
		print ( '任務(wù)正常運(yùn)行...' )

def sch_test(job_type):
    now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print('時(shí)間：{}, {}測(cè)試apscheduler'.format(now, job_type))
    print(1/0)

sched = BlockingScheduler()
# 使用mysql存儲(chǔ)任務(wù)
sql_url = 'mysql+pymysql://root:root@localhost:3306/db?charset=utf8'
sched.add_jobstore('sqlalchemy',url=sql_url)
# 添加任務(wù)
sched.add_job(func=sch_test, args=('定時(shí)任務(wù)',), trigger='cron', second='*/5')

# 配置任務(wù)執(zhí)行完成及錯(cuò)誤時(shí)的監(jiān)聽(tīng)
sched.add_listener(log_listen, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
# 配置日志監(jiān)聽(tīng)
sched._logger = logging

sched.start()

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
from apscheduler.events import EVENT_JOB_EXECUTED , EVENT_JOB_ERROR
import logging
import logging.handlers
import os
import datetime


class LoggerUtils():
    def init_logger(self, logger_name):
        # 日志格式
        formatter = logging.Formatter('%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s')
        log_obj = logging.getLogger(logger_name)
        log_obj.setLevel(logging.INFO)
        # 設(shè)置log存儲(chǔ)位置
        path = '/data/logs/'
        filename = '{}{}.log'.format(path, logger_name)
        if not os.path.exists(path):
            os.makedirs(path)
        # 設(shè)置日志按照時(shí)間分割
        timeHandler = logging.handlers.TimedRotatingFileHandler(
           filename,
           when='D',  # 按照什么維度切割， S:秒，M：分，H:小時(shí)，D:天，W:周
           interval=1, # 多少天切割一次
           backupCount=10  # 保留幾天
        )
        timeHandler.setLevel(logging.INFO)
        timeHandler.setFormatter(formatter)
        log_obj.addHandler(timeHandler)
        return log_obj


class Scheduler(LoggerUtils):
    def __init__(self):
        # 執(zhí)行器設(shè)置
        executors = {
            'default': ThreadPoolExecutor(10),  # 設(shè)置一個(gè)名為“default”的ThreadPoolExecutor，其worker值為10
            'processpool': ProcessPoolExecutor(5)  # 設(shè)置一個(gè)名為“processpool”的ProcessPoolExecutor，其worker值為5
        }
        self.scheduler = BlockingScheduler(timezone="Asia/Shanghai", executors=executors)
        # 存儲(chǔ)器設(shè)置
        # 這里使用sqlalchemy存儲(chǔ)器,將任務(wù)存儲(chǔ)在mysql
        sql_url = 'mysql+pymysql://root:root@localhost:3306/db?charset=utf8'
        self.scheduler.add_jobstore('sqlalchemy',url=sql_url)

        def log_listen(event):
            if event.exception:
                # 日志記錄
                self.scheduler._logger.error(event.traceback)
    
        # 配置任務(wù)執(zhí)行完成及錯(cuò)誤時(shí)的監(jiān)聽(tīng)
        self.scheduler.add_listener(log_listen, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
        # 配置日志監(jiān)聽(tīng)
        self.scheduler._logger = self.init_logger('sche_test')

    def add_job(self, *args, **kwargs):
        """添加任務(wù)"""
        self.scheduler.add_job(*args, **kwargs)

    def start(self):
        """開(kāi)啟任務(wù)"""
        self.scheduler.start()

# 測(cè)試任務(wù)
def sch_test(job_type):
    now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print('時(shí)間：{}, {}測(cè)試apscheduler'.format(now, job_type))
    print(1/0)


# 添加任務(wù),開(kāi)啟任務(wù)
sched = Scheduler()
# 添加任務(wù)
sched.add_job(func=sch_test, args=('定時(shí)任務(wù)',), trigger='cron', second='*/5')
# 開(kāi)啟任務(wù)
sched.start()