carry_efficiency_tool/workers/select_work_studio.py

import sys
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from loguru import logger
sys.path.append('/app')
from utils.odps_data import OdpsDataCount
from utils.redis import RedisHelper


class StartGetRecommend(object):
    @classmethod
    def run_nrfx(cls):
        logger.info(f"[ODPS] 开始获取内容分析数据")
        data = OdpsDataCount.main()
        if not data:
            logger.info(f"[ODPS] 片尾内容分析没有获取到数据")
            return
        RedisHelper().get_client().rpush("task:carry_redis_by_nrfx", *data)
        logger.info(f"[ODPS] 片尾内容分析共获取{len(data)}条,写入成功")


def run():
    scheduler = BlockingScheduler()
    try:
        scheduler.add_job(StartGetRecommend.run_nrfx, trigger=CronTrigger(minute=55, second=0))  # 每小时获取一次
        scheduler.start()
    except KeyboardInterrupt:
        pass
    except Exception as e:
        pass
    finally:
        scheduler.shutdown()


if __name__ == '__main__':
    run()