DataCenterGame/app/api_data/tx_ad_cost/get_cost.py

from app.api_data.tx_ad_cost.cost_util import *
from model.DateUtils import DateUtils
from model.DataBaseUtils import MysqlUtils
from concurrent.futures import ThreadPoolExecutor
from model.DingTalkUtils import DingTalkUtils
import logging

db = MysqlUtils()
du = DateUtils()

max_workers = 10


def get_accounts(filter=None):
    if filter:
        if filter == 'MP':
            return db.quchen_text.getData(
                "select account_id,access_token,name channel from advertiser_vx where (name !='' or name is not null)")
        else:
            return db.quchen_text.getData(
                "select account_id,access_token,name channel from advertiser_qq where (name !='' or name is not null)")

    return db.quchen_text.getData(
        "select account_id,access_token,name channel,'MP' flag from advertiser_vx where (name !='' or name is not null) union "
        "select account_id,access_token,name channel,'GDT' flag from advertiser_qq where (name !='' or name is not null)")


def ad(dt):
    sql = f"""SELECT b.account_id,b.access_token,b.type,GROUP_CONCAT(ad_id)  from ad_cost_day  a
left join (
select account_id,access_token,'MP' type  from advertiser_vx where (name !='' or name is not null) union 
select account_id,access_token,'GDT' type  from advertiser_qq where (name !='' or name is not null)
) b on a.account_id=b.account_id
where a.dt='{dt}'
GROUP BY b.account_id,b.access_token,b.type"""

    accounts = db.quchen_text.getData(sql)
    executor = ThreadPoolExecutor(max_workers=max_workers)

    for account in accounts:
        executor.submit(get_ad_info, account[0], account[1], account[2], account[3], dt)
    executor.shutdown()


"""广告日消耗"""


def ad_cost_day(st, et):
    executor = ThreadPoolExecutor(max_workers=max_workers)
    for account in get_accounts():
        executor.submit(get_ad_cost_day, account[0], account[1], account[3], st, et)
    executor.shutdown()


def adcreative(dt):
    sql = f"""SELECT b.account_id,b.access_token,b.type,GROUP_CONCAT(adcreative_id)  from ad_info  a
left join (
select account_id,access_token,'MP' type  from advertiser_vx where  (name !='' or name is not null) union 
select account_id,access_token,'GDT' type from advertiser_qq where (name !='' or name is not null)
) b on a.account_id=b.account_id
where a.dt='{dt}'
GROUP BY b.account_id,b.access_token,b.type having b.account_id is not null """

    accounts = db.quchen_text.getData(sql)
    # print(accounts)
    executor = ThreadPoolExecutor(max_workers=max_workers)
    for account in accounts:
        executor.submit(get_adcreatives, account[0], account[1], account[2], account[3], dt)
    executor.shutdown()


def image(dt):
    sql = f"""SELECT b.account_id,b.access_token,b.type,GROUP_CONCAT(image_id)  from adcreative_info  a
left join (
select account_id,access_token,'MP' type  from advertiser_vx where (name !='' or name is not null) union 
select account_id,access_token,'GDT' type from advertiser_qq where (name !='' or name is not null)
) b on a.account_id=b.account_id
where a.dt='{dt}' and a.is_video=0
GROUP BY b.account_id,b.access_token,b.type"""
    accounts = db.quchen_text.getData(sql)
    executor = ThreadPoolExecutor(max_workers=max_workers)
    for account in accounts:
        executor.submit(images_info_get, account[0], account[1], account[3])
    executor.shutdown()


def video(dt):
    sql = f"""SELECT b.account_id,b.access_token,b.type,GROUP_CONCAT(image_id)  from adcreative_info  a
	left join (
	select account_id,access_token,'MP' type  from advertiser_vx where (name !='' or name is not null) union 
	select account_id,access_token,'GDT' type from advertiser_qq where (name !='' or name is not null)
	) b on a.account_id=b.account_id
	where a.dt='{dt}' and a.is_video=1
	GROUP BY b.account_id,b.access_token,b.type"""
    accounts = db.quchen_text.getData(sql)
    executor = ThreadPoolExecutor(max_workers=max_workers)
    for account in accounts:
        executor.submit(video_info_get, account[0], account[1], account[3])
    executor.shutdown()


def campaign(dt):
    sql = f"""SELECT b.account_id,b.access_token,b.type,GROUP_CONCAT(campaign_id)  from ad_info  a
	left join (
	select account_id,access_token,'MP' type  from advertiser_vx where  (name !='' or name is not null) union 
	select account_id,access_token,'GDT' type from advertiser_qq where  (name !='' or name is not null)
	) b on a.account_id=b.account_id
	where a.dt='{dt}'
	GROUP BY b.account_id,b.access_token,b.type having b.account_id is not null """

    accounts = db.quchen_text.getData(sql)

    executor = ThreadPoolExecutor(max_workers=max_workers)
    for account in accounts:
        executor.submit(get_campaign, account[0], account[1], account[2], account[3], dt)
    executor.shutdown()


def run(dt):
    """
        1.拉取有消耗的广告
        2.用有消耗的广告id 去拉取广告基础信息
        3.用第2步获取的创意id 去拉取广告创意基础信息
        4.用创意信息中的图片id 去获取图片的基础信息
        """
    try:
        logging.info('获取广告id,开始')
        ad_cost_day(dt, dt)
        logging.info('获取广告id,结束')
        logging.info('获取广告基础信息,开始')
        ad(dt)
        logging.info('获取广告基础信息,结束')
        logging.info('获取广告创意素材,开始')
        adcreative(dt)
        logging.info('获取广告创意素材,结束')
        logging.info('获取广告计划信息,开始')
        campaign(dt)
        logging.info('获取广告计划信息,结束')
        logging.info('获取图片信息,开始')
        image(dt)
        logging.info('获取图片信息,结束')
        logging.info('获取视频信息,开始')
        video(dt)
        logging.info('获取视频信息,结束')

    except:
        DingTalkUtils().send("拉取广告数据出错")


def day():
    ad_cost_day(du.get_n_days(-10), du.get_n_days(0))


def hourly():
    dt = du.getNow()
    run(dt)


if __name__ == '__main__':
    from logging import handlers

    logging.basicConfig(
        handlers=[
            logging.handlers.RotatingFileHandler('./dw_image_cost.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"
    )
    # MP
    # account_id = 18516323
    # access_token = '262deda76aec00c2e144e83bd3c0b2a2'
    #
    # account_id2= 14709511
    # access_token2 = 'e87f7b6f860eaeef086ddcc9c3614678'
    # run()
    # ad_cost_day('2020-04-08','2021-04-07')

    # day()
    #

    # day()
    # run('2021-05-10')
    # adcreative('2021-05-11')
    # video('2021-05-14')
    # campaign('2021-05-14')
    # ad_cost_day(du.get_n_days(-365), du.get_n_days(0))

    for dt in list(du.getDateLists(du.get_n_days(-10), du.get_n_days(0))):
        print(dt)
        campaign(dt)
    # run(dt)
    # ad(dt)
    # adcreative(dt)
    # image(dt)
    # video(dt)
    # campaign(du.get_n_days(0))
    run(du.get_n_days(0))