# 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