DataCenterGame/app/crontab_task/ad_hourly.py

from model.DateUtils import DateUtils
from app.api_data.tx_ad_cost import get_cost
from app.etl.dw import dw_image_cost_day
import time
from model.DingTalkUtils import DingTalkUtils
import logging
from logging import handlers
import threading

du = DateUtils()

def hourly_run():
    # 广告数据
    get_cost.hourly()# 广告相关消耗数据
    dw_image_cost_day.hourly()

if __name__ == '__main__':
    logging.basicConfig(
        handlers=[
            logging.handlers.RotatingFileHandler('./ad_hourly.log',
                                                 maxBytes=10 * 1024 * 1024,
                                                 backupCount=5,
                                                 encoding='utf-8')
            , logging.StreamHandler()  # 供输出使用
        ],
        level=logging.INFO,
        format="%(asctime)s - %(levelname)s %(filename)s %(funcName)s %(lineno)s - %(message)s"
    )
    logging.info('广告素材任务,开始')
    st = time.time()

    hourly_thread=threading.Thread(target=hourly_run)
    hourly_thread.daemon=True
    hourly_thread.start()

    while 1:
        if time.time()-st > 3000:
            logging.info('线程运行超过50分钟,强制停止')
            break
        if not hourly_thread.is_alive():
            break

    if int(time.time()-st)>1500:
        DingTalkUtils.send(f"小时任务耗时{int(time.time()-st)}秒",phone="15168342316")

    logging.info('广告素材任务,结束')