DataCenterGame/app/etl/dw/dw_channel_daily.py

"""
@desc 号维度全量表 ck上跑完dw_daily_channel 并同步到mysql dw_channel_daily
@auth ck
"""
import time
from model.DateUtils import DateUtils
from model.DataBaseUtils import MysqlUtils,CkUtils
du = DateUtils()
db= MysqlUtils()
ck = CkUtils()


def dw_daily_channel():
    print("run> dw_daily_channel")

    sql="""insert into dw_daily_channel
select
       dt,channel,pitcher,stage,platform,book,
       order_count,order_user,order_amount,first_order_count,first_order_user,first_order_amount,
       view_count,click_count,follow_user,cost,reg_order_count,reg_order_user,
       reg_order_amount,reg_order_amount30,web_view_count,platform_view_count,web_order_count,
       0 total_cost,0 total_amount,
       reg_order_user_again,reg_order_user7,reg_order_user30,reg_order_amount7,type,
       0 total_first_amount,
       require_roi,require_mult
from
(select dt,channel, pitcher,stage,platform,book,cost,view_count,click_count,follow_user,web_view_count,platform_view_count,web_order_count,
case type when 'vx' then 'MP' when 'qq' then 'GDT' else type end type,require_roi,require_mult from dw_daily_channel_cost) a
    left outer join
    (select toDate(formatDateTime(reg_time,'%Y-%m-%d')) as dt2,channel as channel2,
       sum(amount) as reg_order_amount,
       count(distinct user_id) as reg_order_user,
       count(1) as reg_order_count,
       sum(if(subtractDays(date, 30)>reg_time,0,amount)) as reg_order_amount30,
       count(distinct if(subtractDays(date, 7)>reg_time,'',user_id))-1 reg_order_user7,
       sum(if(subtractDays(date, 7)>reg_time,0,amount)) as reg_order_amount7,
       count(distinct if(subtractDays(date, 30)>reg_time,'',user_id))-1 reg_order_user30
       from order where reg_time>'2019-03-18 00:00:00' group by toDate(formatDateTime(reg_time,'%Y-%m-%d')),channel) b
        on dt=dt2 and channel=channel2 left outer join
     (select date as dt3,channel as channel3,
     count(1) as first_order_count,
        count(distinct user_id) as first_order_user,
             sum(amount) as first_order_amount
       from order where toDate(reg_time)=date  group by date,channel) c
        on dt=dt3 and channel=channel3
    left outer join
    (select date as dt4,channel as channel4,
     count(1) as order_count,
        count(distinct user_id) as order_user,
             sum(amount) as order_amount
       from order  group by date,channel) d
        on dt=dt4 and channel=channel4
left outer join (
select sum(if(user_order_count>1,1,0)) reg_order_user_again,channel channel6,toDate(reg_date) dt6  from (
select formatDateTime(reg_time,'%Y-%m-%d') reg_date,channel,count(1) user_order_count
    from order group by formatDateTime(reg_time,'%Y-%m-%d') ,user_id,channel
    ) x group by reg_date,channel
    ) f on dt=dt6 and channel=channel6
    having order_amount+cost+reg_order_amount>0
        """
    ck.execute("truncate table dw_daily_channel")
    ck.execute(sql)
    print("ok")

    data=ck.execute('select * from dw_daily_channel')
    insert_sql="insert into dw_channel_daily values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"
    db.dm.execute("truncate table dw_channel_daily")
    db.dm.executeMany(insert_sql,data)

    db.quchen_text.execute("truncate table dw_channel_daily")
    db.quchen_text.executeMany(insert_sql,data)



def dw_channel_daily():
    sql="""
select
       dt,channel,pitcher,stage,platform,book,case type when 'vx' then 'MP' when 'qq' then 'GDT' else type end type,
       order_count,order_user,order_amount,
       first_order_count,first_order_user,first_order_amount,
       view_count,click_count,follow_user,
       cost,reg_order_count,reg_order_user,reg_order_amount,
       web_view_count,platform_view_count,web_order_count,
       reg_order_user_again,
       reg_order_user_again3,
       reg_order_user_again4,
       reg_order_user_again5,
       reg_order_user_again6
from
(select dt,channel, pitcher,stage,platform,book,cost,view_count,click_count,   ---基础属性和消耗数据
        follow_user,web_view_count,platform_view_count,web_order_count,type,
        require_roi,require_mult from dw_daily_channel_cost) a


left outer join
     (select date as dt3,channel as channel3,count(1) as first_order_count,          ---新用户首日充值
     count(distinct user_id) as first_order_user,sum(amount) as first_order_amount
    from order where toDate(reg_time)=date  group by date,channel) c
    on dt=dt3 and channel=channel3

left outer join
        (select date as dt4,channel as channel4,count(1) as order_count,   ---账面充值
        count(distinct user_id) as order_user,sum(amount) as order_amount
        from order  group by date,channel) d
        on dt=dt4 and channel=channel4

left outer join (
    select sum(if(user_order_count>=2,1,0)) reg_order_user_again,channel channel6,toDate(reg_date) dt6,  ---复冲人数
           sum(if(user_order_count>=3,1,0)) reg_order_user_again3,
           sum(if(user_order_count>=4,1,0)) reg_order_user_again4,
           sum(if(user_order_count>=5,1,0)) reg_order_user_again5,
           sum(if(user_order_count>=6,1,0)) reg_order_user_again6
    from (select formatDateTime(reg_time,'%Y-%m-%d') reg_date,channel,count(1) user_order_count
    from order group by formatDateTime(reg_time,'%Y-%m-%d') ,user_id,channel) x group by reg_date,channel
    ) f on dt=dt6 and channel=channel6

left outer join
   (
       select toDate(formatDateTime(reg_time,'%Y-%m-%d')) as dt2,    ---新用户累计充值数据
   channel as channel2,
   sum(amount) as reg_order_amount,
   count(distinct user_id) as reg_order_user,
   count(1) as reg_order_count
   from order where reg_time>'2019-03-18 00:00:00' group by toDate(formatDateTime(reg_time,'%Y-%m-%d')),channel) b
    on dt=dt2 and channel=channel2
    having order_amount+cost+reg_order_amount>0"""

    data=ck.execute(sql)
    isql="insert into dw_channel values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"
    db.dm.execute("truncate table dw_channel")
    db.dm.executeMany(isql,data)

def dw_channel_user_daily():
    sql="""
select toDate(formatDateTime(reg_time,'%Y-%m-%d')) as dt,
            channel,
           count(distinct if(subtractDays(date, 1)>=toDate(reg_time),null,user_id)) dc1,
           count(distinct if(subtractDays(date, 2)>=toDate(reg_time),null,user_id)) dc2,
           count(distinct if(subtractDays(date, 3)>=toDate(reg_time),null,user_id)) dc3,
           count(distinct if(subtractDays(date, 4)>=toDate(reg_time),null,user_id)) dc4,
           count(distinct if(subtractDays(date, 5)>=toDate(reg_time),null,user_id)) dc5,
           count(distinct if(subtractDays(date, 6)>=toDate(reg_time),null,user_id)) dc6,
           count(distinct if(subtractDays(date, 7)>=toDate(reg_time),null,user_id)) dc7,
           count(distinct if(subtractDays(date, 8)>=toDate(reg_time),null,user_id)) dc8,
           count(distinct if(subtractDays(date, 9)>=toDate(reg_time),null,user_id)) dc9,
           count(distinct if(subtractDays(date, 10)>=toDate(reg_time),null,user_id)) dc10,
           count(distinct if(subtractDays(date, 11)>=toDate(reg_time),null,user_id)) dc11,
           count(distinct if(subtractDays(date, 12)>=toDate(reg_time),null,user_id)) dc12,
           count(distinct if(subtractDays(date, 13)>=toDate(reg_time),null,user_id)) dc13,
           count(distinct if(subtractDays(date, 14)>=toDate(reg_time),null,user_id)) dc14,
           count(distinct if(subtractDays(date, 15)>=toDate(reg_time),null,user_id)) dc15,
           count(distinct if(subtractDays(date, 16)>=toDate(reg_time),null,user_id)) dc16,
           count(distinct if(subtractDays(date, 17)>=toDate(reg_time),null,user_id)) dc17,
           count(distinct if(subtractDays(date, 18)>=toDate(reg_time),null,user_id)) dc18,
           count(distinct if(subtractDays(date, 19)>=toDate(reg_time),null,user_id)) dc19,
           count(distinct if(subtractDays(date, 20)>=toDate(reg_time),null,user_id)) dc20,
           count(distinct if(subtractDays(date, 21)>=toDate(reg_time),null,user_id)) dc21,
           count(distinct if(subtractDays(date, 22)>=toDate(reg_time),null,user_id)) dc22,
           count(distinct if(subtractDays(date, 23)>=toDate(reg_time),null,user_id)) dc23,
           count(distinct if(subtractDays(date, 24)>=toDate(reg_time),null,user_id)) dc24,
           count(distinct if(subtractDays(date, 25)>=toDate(reg_time),null,user_id)) dc25,
           count(distinct if(subtractDays(date, 26)>=toDate(reg_time),null,user_id)) dc26,
           count(distinct if(subtractDays(date, 27)>=toDate(reg_time),null,user_id)) dc27,
           count(distinct if(subtractDays(date, 28)>=toDate(reg_time),null,user_id)) dc28,
           count(distinct if(subtractDays(date, 29)>=toDate(reg_time),null,user_id)) dc29,
           count(distinct if(subtractDays(date, 30)>=toDate(reg_time),null,user_id)) dc30
from order where reg_time>'2019-03-18 00:00:00' group by toDate(formatDateTime(reg_time,'%Y-%m-%d')),channel"""

    data =ck.execute(sql)
    isql="insert into dw_channel_user_daily values " \
         "(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s," \
         "%s,%s,%s,%s,%s,%s,%s,%s,%s,%s," \
         "%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"
    db.dm.execute("truncate table dw_channel_user_daily")
    db.dm.executeMany(isql,data)

def dw_channel_amount_daily():
    sql="""
 select toDate(formatDateTime(reg_time,'%Y-%m-%d')) as dt,
   channel as channel,
           sum(if(subtractDays(date, 1)>=toDate(reg_time),0,amount)) as da1,
           sum(if(subtractDays(date, 2)>=toDate(reg_time),0,amount)) as da2,
           sum(if(subtractDays(date, 3)>=toDate(reg_time),0,amount)) as da3,
           sum(if(subtractDays(date, 4)>=toDate(reg_time),0,amount)) as da4,
           sum(if(subtractDays(date, 5)>=toDate(reg_time),0,amount)) as da5,
           sum(if(subtractDays(date, 6)>=toDate(reg_time),0,amount)) as da6,
           sum(if(subtractDays(date, 7)>=toDate(reg_time),0,amount)) as da7,
           sum(if(subtractDays(date, 8)>=toDate(reg_time),0,amount)) as da8,
           sum(if(subtractDays(date, 9)>=toDate(reg_time),0,amount)) as da9,
           sum(if(subtractDays(date, 10)>=toDate(reg_time),0,amount)) as da10,
           sum(if(subtractDays(date, 11)>=toDate(reg_time),0,amount)) as da11,
           sum(if(subtractDays(date, 12)>=toDate(reg_time),0,amount)) as da12,
           sum(if(subtractDays(date, 13)>=toDate(reg_time),0,amount)) as da13,
           sum(if(subtractDays(date, 14)>=toDate(reg_time),0,amount)) as da14,
           sum(if(subtractDays(date, 15)>=toDate(reg_time),0,amount)) as da15,
           sum(if(subtractDays(date, 16)>=toDate(reg_time),0,amount)) as da16,
           sum(if(subtractDays(date, 17)>=toDate(reg_time),0,amount)) as da17,
           sum(if(subtractDays(date, 18)>=toDate(reg_time),0,amount)) as da18,
           sum(if(subtractDays(date, 19)>=toDate(reg_time),0,amount)) as da19,
           sum(if(subtractDays(date, 20)>=toDate(reg_time),0,amount)) as da20,
           sum(if(subtractDays(date, 21)>=toDate(reg_time),0,amount)) as da21,
           sum(if(subtractDays(date, 22)>=toDate(reg_time),0,amount)) as da22,
           sum(if(subtractDays(date, 23)>=toDate(reg_time),0,amount)) as da23,
           sum(if(subtractDays(date, 24)>=toDate(reg_time),0,amount)) as da24,
           sum(if(subtractDays(date, 25)>=toDate(reg_time),0,amount)) as da25,
           sum(if(subtractDays(date, 26)>=toDate(reg_time),0,amount)) as da26,
           sum(if(subtractDays(date, 27)>=toDate(reg_time),0,amount)) as da27,
           sum(if(subtractDays(date, 28)>=toDate(reg_time),0,amount)) as da28,
           sum(if(subtractDays(date, 29)>=toDate(reg_time),0,amount)) as da29,
           sum(if(subtractDays(date, 30)>=toDate(reg_time),0,amount)) as da30,
           sum(if(subtractDays(date, 31)>=toDate(reg_time),0,amount)) as da31,
           sum(if(subtractDays(date, 32)>=toDate(reg_time),0,amount)) as da32,
           sum(if(subtractDays(date, 33)>=toDate(reg_time),0,amount)) as da33,
           sum(if(subtractDays(date, 34)>=toDate(reg_time),0,amount)) as da34,
           sum(if(subtractDays(date, 35)>=toDate(reg_time),0,amount)) as da35,
           sum(if(subtractDays(date, 36)>=toDate(reg_time),0,amount)) as da36,
           sum(if(subtractDays(date, 37)>=toDate(reg_time),0,amount)) as da37,
           sum(if(subtractDays(date, 38)>=toDate(reg_time),0,amount)) as da38,
           sum(if(subtractDays(date, 39)>=toDate(reg_time),0,amount)) as da39,
           sum(if(subtractDays(date, 40)>=toDate(reg_time),0,amount)) as da40,
           sum(if(subtractDays(date, 41)>=toDate(reg_time),0,amount)) as da41,
           sum(if(subtractDays(date, 42)>=toDate(reg_time),0,amount)) as da42,
           sum(if(subtractDays(date, 43)>=toDate(reg_time),0,amount)) as da43,
           sum(if(subtractDays(date, 44)>=toDate(reg_time),0,amount)) as da44,
           sum(if(subtractDays(date, 45)>=toDate(reg_time),0,amount)) as da45,
           sum(if(subtractDays(date, 46)>=toDate(reg_time),0,amount)) as da46,
           sum(if(subtractDays(date, 47)>=toDate(reg_time),0,amount)) as da47,
           sum(if(subtractDays(date, 48)>=toDate(reg_time),0,amount)) as da48,
           sum(if(subtractDays(date, 49)>=toDate(reg_time),0,amount)) as da49,
           sum(if(subtractDays(date, 50)>=toDate(reg_time),0,amount)) as da50,
           sum(if(subtractDays(date, 51)>=toDate(reg_time),0,amount)) as da51,
           sum(if(subtractDays(date, 52)>=toDate(reg_time),0,amount)) as da52,
           sum(if(subtractDays(date, 53)>=toDate(reg_time),0,amount)) as da53,
           sum(if(subtractDays(date, 54)>=toDate(reg_time),0,amount)) as da54,
           sum(if(subtractDays(date, 55)>=toDate(reg_time),0,amount)) as da55,
           sum(if(subtractDays(date, 56)>=toDate(reg_time),0,amount)) as da56,
           sum(if(subtractDays(date, 57)>=toDate(reg_time),0,amount)) as da57,
           sum(if(subtractDays(date, 58)>=toDate(reg_time),0,amount)) as da58,
           sum(if(subtractDays(date, 59)>=toDate(reg_time),0,amount)) as da59,
           sum(if(subtractDays(date, 60)>=toDate(reg_time),0,amount)) as da60,
           sum(if(subtractDays(date, 90)>=toDate(reg_time),0,amount)) as dm3,
           sum(if(subtractDays(date, 120)>=toDate(reg_time),0,amount)) as dm4,
           sum(if(subtractDays(date, 150)>=toDate(reg_time),0,amount)) as dm5
 from order where reg_time>'2019-03-18 00:00:00' group by toDate(formatDateTime(reg_time,'%Y-%m-%d')),channel"""
    data=ck.execute(sql)
    isql="insert into dw_channel_amount_daily values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s," \
         "%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"
    db.dm.execute("truncate table dw_channel_amount_daily")
    db.dm.executeMany(isql,data)


def dw_channel():
    print('run> dw_channel')
    dw_channel_daily()
    dw_channel_user_daily()
    dw_channel_amount_daily()


if __name__ == '__main__':

    # dw_daily_channel()
    dw_channel()
    # dw_channel_amount_daily()
    # dw_channel_user_daily()