# date: 2021-10-13
"""
# 每天0点：add_job(trigger='cron', hour='0')
# 每1小时：add_job(trigger='interval', hour='1')
# 每1分钟：add_job(trigger='cron', minutes='*/1')
tips: 如果希望加强测试或重度使用，则推荐循环操作；如果轻度使用或保守推进，建议手动指定下次操作时间。
"""
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.executors.pool import ThreadPoolExecutor
from pymongo import MongoClient
import uuid
class APS(object):
def __init__(self, db_type='mongo', db_host='aibox-middleware-mongo', db_port=27017,
username='admin', password='admin', database='aibox', collection='LoopTask', is_clean=True):
"""
内部访问:
host: sri-thirdparty-mongo
port: 27017
远程访问:
host: 118.190.217.96、192.168.20.162
port: 7030
"""
# todo 默认使用sqlite数据库，而不是mongo
mongo = MongoClient(db_host, db_port, username=username, password=password)
if is_clean:
mongo[database][collection].delete_many({})
jobstores = {
'default': MongoDBJobStore(client=mongo, database=database, collection=collection),
}
executors = {
'default': ThreadPoolExecutor(max_workers=3)
}
self.scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors, timezone='Asia/Shanghai')
self.scheduler.start() # todo 不支持client方式查看job列表
def get_all(self):
"""
获取任务列表
"""
job_list = []
jobs = self.scheduler.get_jobs()
for job in jobs:
data = {
'id': job.id,
'name': job.name,
'func': job.func.__name__,
'args': list(job.args),
'trigger': str(job.trigger),
'next_run_time': job.next_run_time.strftime('%Y-%m-%d %H:%M:%S'),
}
job_list.append(data)
return job_list
def get_one(self, job_uuid):
"""获取任务详情"""
job = self.scheduler.get_job(job_uuid)
data = {
'id': job.id,
'name': job.name,
'func': job.func.__name__,
'args': list(job.args),
'trigger': str(job.trigger),
'next_run_time': job.next_run_time.strftime('%Y-%m-%d %H:%M:%S'),
}
return data
def create_job(self, **params):
"""
创建任务
trigger: 触发器
date 固定时间调度（只会执行一次）
interval 固定时间间隔
cron 定时调度
"""
params['id'] = str(uuid.uuid4())
# return self.scheduler.add_job(**params)
return self.scheduler.add_job(**params, max_instances=100) # fix maximum number of running instances reached
def remove_all(self):
"""删除全部任务"""
jobs = self.scheduler.get_jobs()
job_ids = [i.id for i in jobs]
for job_id in job_ids:
self.remove_one(job_id)
return job_ids
def remove_one(self, job_uuid):
"""删除指定任务"""
return self.scheduler.remove_job(job_uuid)
def pause_one(self, job_uuid):
"""暂停指定任务"""
return self.scheduler.pause_job(job_uuid)
def resume_one(self, job_uuid):
"""恢复指定任务"""
return self.scheduler.resume_job(job_uuid)
def pause_all(self):
"""暂停全部任务"""
return self.scheduler.pause()
def resume_all(self):
"""恢复全部任务"""
return self.scheduler.resume()
@classmethod
def create_job_by_hour(cls, run_date, ins_id):
"""
定时执行
run_date: 执行时间（%Y-%m-%d %H:%M:%S）
ins_id: 数据id
"""
# str_task_id = str(uuid.uuid4())
# str_item = mdb.get_one_by_id('UserEventStrategy', ins_id)
# str_task_id_list = str_item.get('str_task_id_list', [])
# methods.debug_log('ApsClient.create_job_by_hour', f"str_task_id_list: {str_task_id_list}")
# str_task_id_list.append(str_task_id)
# mdb.update_one_by_id('UserEventStrategy', ins_id, {'str_task_id_list': str_task_id_list})
# run_date = datetime.datetime.strptime(run_date, "%Y-%m-%d %H:%M:%S")
# cls.scheduler.add_job(func=action_event_strategy, trigger='date', run_date=run_date, id=str_task_id,
# args=[ins_id, ])
@classmethod
def create_job_by_interval(cls, ins_id, start_date, end_date):
"""
每小时执行
:param ins_id: 数据id
:param start_date: 开始时间（"%Y-%m-%d %H:%M:%S"）
:param end_date: 结束时间（"%Y-%m-%d %H:%M:%S"）
:return:
"""
# str_task_id = str(uuid.uuid4())
# str_item = mdb.get_one_by_id('UserEventStrategy', ins_id)
# str_task_id_list = str_item.get('str_task_id_list', [])
# methods.debug_log('ApsClient.create_job_by_interval', f"str_task_id_list: {str_task_id_list}")
# str_task_id_list.append(str_task_id)
# mdb.update_one_by_id('UserEventStrategy', ins_id, {'str_task_id_list': str_task_id_list})
# cls.scheduler.add_job(func=action_event_strategy, trigger='interval', id=str_task_id, hours=1,
# start_date=start_date, end_date=end_date, args=[ins_id, ])
@classmethod
def create_job_by_mon(cls, hour, minute, end_date, ins_id):
"""
指定每周一固定时间执行
:param hour: 指定时
:param minute: 指定分
:param end_date: 结束时间（"%Y-%m-%d %H:%M:%S"）
:param ins_id: 数据id
:return:
"""
# end_date = datetime.datetime.strptime(end_date, "%Y-%m-%d %H:%M:%S")
# cls.scheduler.add_job(func=action_event_strategy, trigger='cron', day_of_week='mon', id=str(uuid.uuid4()),
# hour=hour, minute=minute, end_date=end_date, args=[ins_id, ])
@classmethod
def create_job_by_week(cls, _args, day_of_week, hour, minute, start_date, instance_id):
"""每周"""
# cron_id = str(uuid.uuid4())
# cls.scheduler.add_job(send_charts, 'cron', day_of_week=day_of_week, id=cron_id, hour=hour,
# minute=minute,
# start_date=start_date, args=_args)
# cls.save_job_id(instance_id, cron_id)
# cls.scheduler.shutdown()
@classmethod
def create_job_by_month(cls, _args, month, day, hour, minute, start_date, instance_id):
"""每月"""
# cron_id = str(uuid.uuid4())
#
# cls.scheduler.add_job(send_charts, 'cron', month=month, day=day, id=cron_id, hour=hour,
# minute=minute, start_date=start_date, args=_args, )
# cls.save_job_id(instance_id, cron_id)
# cls.scheduler.shutdown()
@classmethod
def create_job_by_day(cls, _args, hour, minute, start_date, instance_id):
"""每天"""
# cron_id = str(uuid.uuid4())
#
# cls.scheduler.add_job(send_charts, 'cron', id=cron_id, hour=hour, start_date=start_date,
# minute=minute, args=_args)
# cls.save_job_id(instance_id, cron_id)
# cls.scheduler.shutdown()
@classmethod
def create_job_by_hours(cls, _args, minute, start_date, instance_id):
"""每小时"""
# cron_id = str(uuid.uuid4())
# cls.scheduler.add_job(send_charts, 'cron', id=cron_id,
# minute=minute, start_date=start_date, args=_args)
# cls.save_job_id(instance_id, cron_id)
# cls.scheduler.shutdown()
@classmethod
def create_job_by_n_day(cls, _args, day, hour, minute, start_date, instance_id):
"""n天"""
# cron_id = str(uuid.uuid4())
# cls.scheduler.add_job(send_charts, 'interval', id=cron_id,
# days=day, hours=hour, minutes=minute, start_date=start_date,
# args=_args)
# cls.save_job_id(instance_id, cron_id)
# cls.scheduler.shutdown()
@classmethod
def create_job_by_n_hour(cls, _args, hour, minute, start_date, instance_id):
"""n小时"""
# cron_id = str(uuid.uuid4())
# cls.scheduler.add_job(send_charts, 'interval', id=cron_id, hours=hour, minutes=minute,
# start_date=start_date, args=_args)
# cls.save_job_id(instance_id, cron_id)
# cls.scheduler.shutdown()
@classmethod
def create_job_by_n_minute(cls, _args, minute, start_date, instance_id):
"""n分钟"""
# cron_id = str(uuid.uuid4())
# cls.scheduler.add_job(send_charts, 'interval', id=cron_id, minutes=minute, start_date=start_date,
# args=_args)
# cls.save_job_id(instance_id, cron_id)
# cls.scheduler.shutdown()
@classmethod
def create_job_by_n_minute_alarm(cls, minute, start_date, end_date):
"""告警收敛"""
# cls.scheduler.add_job(early_count, 'interval', id=str(uuid.uuid4()), minutes=minute, start_date=start_date,
# end_date=end_date)
# cls.scheduler.shutdown()
if __name__ == '__main__':
# --- init ---
aps = APS(db_type='mongo', db_host='192.168.30.13', db_port=7030,
username='admin', password='admin', database='vms', collection='LoopTask')
# --- 并不支持client方式测试 ---
pass