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


def del_channel():
    sql = 'delete from dw_channel where channel in ("腾云文渊", "青渊文学", "玉竹文渊", "赤瞳伴读","扶摇阅文")'
    db.dm.execute(sql)



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

if __name__ == '__main__':

    # dw_daily_channel()
    # dw_channel()
    # dw_channel_amount_daily()
    # dw_channel_user_daily()
    # dw_channel_amount_daily_reverse()
    del_channel()