DataCenterGame/app/etl/data_stat_task.py

from model.DataBaseUtils import MysqlUtils, CkUtils
from datetime import datetime, timedelta, timezone
from model.DateUtils import DateUtils
import logging
import time

db = MysqlUtils()
ck = CkUtils()
dt = DateUtils()


def dw_daily_bytedance_cost(ymd):
    logging.info(f'dw_daily_bytedance_cost 数据填充开始')

    sql = '''
    select x.dt,x.channel,pitcher,stage,x.platform,x.book,
               ifnull(view_count,0),ifnull(click_count,0),
               ifnull(follow_user,0),ifnull(cost,0)/100 as cost,
               ifnull(web_view_count,0) web_view_count,
               ifnull(platform_view_count,0) platform_view_count,
               ifnull(web_order_count,0) web_order_count,
               'BYTEDANCE' type
                ,0 require_roi,0 require_mult
                 from
        ( select dt, channel,stage,pitcher,platform,book  from channel_info_daily cid 
            where dt='{0}' and channel !=''
            and channel in 
            (select distinct(channel) from channel_by_account_daily cbad 
            where dt='{0}'
            and type ='BYTEDANCE' 
            )
        ) x  -- 只允许渠道MP、GDT
        left join
        (select channel,sum(cost) as cost,sum(view_count) as view_count,sum(valid_click_count) as click_count,sum(from_follow_uv)  as follow_user,
         sum(web_view_count) as web_view_count,sum(platform_view_count) as platform_view_count,sum(web_order_count) as web_order_count
         from
            (select account_id,cost,view_count,valid_click_count,round(valid_click_count*official_account_follow_rate,0) as from_follow_uv,
            0 as  web_view_count,
            0 as platform_view_count,
            0 as web_order_count
             from daily_tt where date='{0} 00:00:00' ) a
        left join
        (select account_id,channel from channel_by_account_daily where dt='{0}') b 
         on a.account_id=b.account_id  group by channel) 
        z on x.channel=z.channel  
        
        '''.format(ymd)

    data = db.quchen_text.get_data_list(sql)
    data1 = []
    col = "dt,channel,pitcher,stage,platform,book,view_count,click_count,follow_user,cost,web_view_count,platform_view_count,web_order_count,type,require_roi,require_mult"
    for i in data:
        i[0] = str(i[0])
        i[9] = str(i[9])
        i[6] = float(i[6])
        i[7] = float(i[7])
        i[8] = float(i[8])
        i[9] = float(i[9])
        i[10] = float(i[10])
        i[11] = float(i[11])
        i[12] = float(i[12])
        data1.append(tuple(i))
    ck.execute(f"alter table dw_daily_bytedance_cost drop  partition '{ymd}' ")
    logging.info(len(data1))
    ck.insertMany("dw_daily_bytedance_cost", col, tuple(data1))

    logging.info(f'dw_daily_bytedance_cost 数据填充结束')


def platform_data_sum(ymd):
    logging.info('dw_daily_platform_cost开始数据更新')
    ck.execute("alter table game_data.dw_daily_platform_cost drop  partition '{}' ".format(ymd))
    sql = f'''
        insert into game_data.dw_daily_platform_cost
        select * from game_data.dw_daily_channel_cost b where dt='{ymd}'
    '''
    ck.execute(sql)
    logging.info('dw_daily_platform_cost数据更新，结束')


def dw_daily_channel_cost(ymd):
    logging.info("run> dw_daily_channel_cost")
    datatime_ymd = datetime.strptime(ymd, '%Y-%m-%d').astimezone(timezone(timedelta(hours=8))).timetuple()
    timestamp_ymd = time.mktime(datatime_ymd)
    tomorrow_ymd = (datetime.strptime(ymd, '%Y-%m-%d').astimezone(timezone(timedelta(hours=8))) + timedelta(
        days=1)).timetuple()
    timestamp_tom = time.mktime(tomorrow_ymd)

    sql = f"""
        select x.dt,x.channel,pitcher,stage,x.platform,x.book,
               ifnull(view_count,0),ifnull(click_count,0),
               ifnull(follow_user,0),ifnull(cost,0)/100 as cost,
               ifnull(web_view_count,0) web_view_count,
               ifnull(platform_view_count,0) platform_view_count,
               ifnull(web_order_count,0) web_order_count,
               if(stage ='趣程15期' or stage ='趣程26期' or stage ='趣程30期','GDT','MP') type
                ,0 require_roi,0 require_mult,ifnull(y.reg_num,0),ifnull(w.create_user_num,0)
                 from
        ( select dt, channel,stage,pitcher,platform,book  from channel_info_daily cid 
            where dt='{ymd}' and channel !=''
            and channel in 
            (select distinct(channel) from channel_by_account_daily cbad 
            where dt='{ymd}'
            and (type ='GDT' or type='MP')
            ) 
        ) x  -- 只允许渠道MP、GDT
        left join
        (select channel,sum(cost) as cost,sum(view_count) as view_count,sum(valid_click_count) as click_count,
            sum(from_follow_uv)  as follow_user,
         sum(web_view_count) as web_view_count,
         sum(platform_view_count) as platform_view_count,
         sum(web_order_count) as web_order_count
         from
            (select account_id,cost,view_count,valid_click_count,
            round(valid_click_count*official_account_follow_rate,0) as from_follow_uv,
            0 as  web_view_count,
            0 as platform_view_count,
            0 as web_order_count
             from daily_vx where date='{ymd} 00:00:00'
            union
            select account_id,cost,view_count,valid_click_count,from_follow_uv,
            ifnull(web_commodity_page_view_count,0) as  web_view_count, 
            ifnull(platform_page_view_count,0) as platform_view_count,
            ifnull(web_order_count,0) as web_order_count
             from daily_qq where date='{ymd} 00:00:00' ) a
        left join
        (select account_id,channel from channel_by_account_daily where dt='{ymd}') b 
         on a.account_id=b.account_id  group by channel) 
        z on x.channel=z.channel  
      
       left join     
        (
        select c.name as channel ,DATE(FROM_UNIXTIME(origin.create_time)) as wx_date,
        count(*) as reg_num
        from 
        db_mp.h_member origin left join
        db_mp.mp_conf_agent a  on origin.app_id =a.app_id and origin.agent_id = a.agent_id 
        left join
        db_mp.mp_mp_conf b on a.advertiser_conf_id =b.id
        left join 
        quchen_text.advertiser_vx c on b.mp_id =c.wechat_account_id 
        where c.wechat_account_id is not null
        and origin.create_time > {timestamp_ymd} and origin.create_time < {timestamp_tom}
        group by name,wx_date
        ) y on x.channel= y.channel
        
        left join     
        (
        select f.name as channel,DATE(FROM_UNIXTIME(a.create_time)) as wx_date,
        count(*) as create_user_num
        from db_mp.h_mg_role a
        left join db_mp.h_mem_game b on a.mg_mem_id = b.id
        left join db_mp.h_member c on b.mem_id = c.id
        left join db_mp.mp_conf_agent d on c.app_id = d.app_id  and c.agent_id = d.agent_id 
        left join db_mp.mp_mp_conf e on d.advertiser_conf_id =e.id
        left join quchen_text.advertiser_vx f on e.mp_id = f.wechat_account_id 
        where f.name is not null 
        and a.create_time > {timestamp_ymd} and a.create_time < {timestamp_tom}
        group by f.name,wx_date
        order by wx_date desc
        ) w on x.channel= w.channel

        
        """
    data = db.quchen_text.get_data_list(sql)
    data1 = []
    col = "dt,channel,pitcher,stage,platform,book,view_count,click_count,follow_user,cost,web_view_count,platform_view_count,web_order_count,type,require_roi,require_mult,reg_num,create_user_num"
    for i in data:
        i[0] = str(i[0])
        i[9] = str(i[9])
        i[6] = float(i[6])
        i[7] = float(i[7])
        i[8] = float(i[8])
        i[9] = float(i[9])
        i[10] = float(i[10])
        i[11] = float(i[11])
        i[12] = float(i[12])
        data1.append(tuple(i))
    ck.execute(f"alter table game_data.dw_daily_channel_cost drop  partition '{ymd}' ")
    logging.info(len(data1))
    ck.insertMany("game_data.dw_daily_channel_cost", col, tuple(data1))


def channel_by_account_daily(ymd):
    """返回当天消耗账户对应的公众号表"""
    logging.info("run> channel_by_account_daily")
    sql = """replace into channel_by_account_daily 
            select  '{0}' as dt,a.account_id as account_id, ifnull(ifnull(b.name,a.name),'') as channel,type  from
            (select  account_id,name,'GDT' type  from advertiser_qq 
             union 
             select account_id,name,'MP' type from advertiser_vx
             union
             select advertiser_id,channel,'BYTEDANCE' type from advertiser_bytedance
             ) a
                left join
            (select b.account_id,b.name from
            (select min(end_time) as end_time,account_id from account_change where end_time>'{0}' GROUP BY account_id) a
            left join account_change  b on  a.end_time=b.end_time and a.account_id=b.account_id) b on a.account_id=b.account_id""".format(
        ymd)
    db.quchen_text.execute(sql)


def channel_info_daily(ymd):
    """获取公众号某天的期数，投手，平台，书籍
    @ return [[]]
    """
    # 获取现在的全量公众号信息
    logging.info("run> channel_info_daily")
    sql = f"""select '{ymd}' as dt,a.name ,ifnull(stage,''),ifnull(pitcher,''),ifnull(platform,''),ifnull(book,'') from (
            select  name from advertiser_vx  where name is not null group by name--  公众号全量表
            union
            select  name from account_change group by name
            union
            select channel as name   from pitcher_change group by channel
            union
            select  name from platform_change group by name
            union
            select  name from book_change group by name) a
             
             left join
              ( select name,ifnull(stage,'') stage,ifnull(pitcher,'') pitcher,
                ifnull(platform,'') platform,ifnull(book,'') book 
                from advertiser_vx 
                where name is not null  
                and start_date <= '{ymd}'
                group by name,stage,pitcher,platform,book
                ) b on a.name=b.name
                """
    data = db.quchen_text.get_data_list(sql)
    pitcher_change = db.quchen_text.getData(
        "select b.channel as channel,pitcher from "
        "(select max(start_time) as start_time,channel from pitcher_change "
        " where start_time<='{}' GROUP BY channel) a"
        " left join pitcher_change  b on  a.start_time=b.start_time and a.channel=b.channel".format(ymd))

    platform_change = db.quchen_text.getData(
        "select b.name as channel,current_platform as platform from (select max(change_date) as change_date,name from platform_change "
        "where change_date<='{}' GROUP BY name) a "
        "left join platform_change  b on  a.change_date=b.change_date and a.name=b.name".format(ymd))

    book_change = db.quchen_text.getData(
        "select b.name as channel,book from (select max(start_time) as start_time,name from book_change "
        "where start_time<='{}' GROUP BY name) a "
        "left join book_change  b on  a.start_time=b.start_time and a.name=b.name".format(ymd))

    stage_change = db.quchen_text.getData(
        "select  channel,stage from (select max(start_date) as start_date,channel from stage_change "
        "where start_date<='{}' GROUP BY channel) a "
        "left join stage_change using(start_date,channel)".format(ymd))

    for i in data:
        for j in pitcher_change:
            if i[1] == j[0]:
                i[3] = j[1]
        for k in platform_change:
            if i[1] == k[0]:
                i[4] = k[1]
        for h in book_change:
            if i[1] == h[0]:
                i[5] = h[1]

        for m in stage_change:
            if i[1] == m[0]:
                i[2] = m[1]

    insert_sql = "replace into channel_info_daily values (%s,%s,%s,%s,%s,%s) "

    db.quchen_text.executeMany(insert_sql, data)


def ods_order(dt):
    sql = """	replace into ods_order
			select 
			case platform when '掌中云' then DATE_FORMAT(STR_TO_DATE(order_time,'%Y-%m-%dT%H:%i:%s'),'%Y-%m-%d') 
            when '掌读' then from_unixtime(order_time, '%Y-%m-%d') 
            ELSE order_time end  date,
            stage,platform,channel,channel_id,user_id,
            case when platform='掌中云' then DATE_FORMAT(STR_TO_DATE(order_time,'%Y-%m-%dT%H:%i:%s'),'%Y-%m-%d %H:%i:%s') 
            when platform='掌读' then from_unixtime(order_time, '%Y-%m-%d %H:%i:%s') 
            ELSE order_time end order_time,
            case when platform='掌中云' then DATE_FORMAT(STR_TO_DATE(reg_time,'%Y-%m-%dT%H:%i:%s'),'%Y-%m-%d %H:%i:%s') 
            when platform='掌读' then from_unixtime(reg_time, '%Y-%m-%d %H:%i:%s') 
            ELSE reg_time end reg_time,
			amount,from_novel,order_id,2 from `order` where date=UNIX_TIMESTAMP('{}')
			""".format(dt)
    db.quchen_text.execute(sql)


def order_account_text():
    db.quchen_text.execute("truncate order_account_text")
    with open('./wending_account_config.csv', encoding='utf-8') as f:
        for i in f.readlines():
            db.quchen_text.execute("insert into order_account_text(platform,text) values ('文鼎','{}')".format(i))


if __name__ == '__main__':
    # channel_info_daily('2021-02-06')

    # channel_by_account_daily('2021-02-05')
    for i in dt.getDateLists('2021-04-18', '2021-10-28'):
        print(i)
        channel_info_daily(i)

        dw_daily_channel_cost(i)
        # ods_order('2021-05-06')
        platform_data_sum(i)