from model.DataBaseUtils import MysqlUtils,CkUtils
from model.DateUtils import DateUtils
from model.log import logger
log=logger()
db = MysqlUtils()
ck = CkUtils()
dt = DateUtils()
from datetime import datetime
from sync_to_ck_task import dw_order_channel_cost_sync_ck
def dw_daily_channel_cost(ymd):
sql="""replace into dw_daily_channel_cost
select dt,x.channel,pitcher,stage,platform,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,
ifnull(type,'')
from
(select dt,channel,stage,pitcher,platform,book from channel_info_daily where dt='{0}' and channel!='') x
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='{0} 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='{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
left join (SELECT channel,type from channel_by_account_daily GROUP By channel,type) k on x.channel=k.channel
""".format(ymd)
print(sql)
db.quchen_text.execute(sql)
def channel_by_account_daily(ymd):
"""返回当天消耗账户对应的公众号表"""
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,'qq' as type from advertiser_qq
union
select account_id,name,'vx' as type from advertiser_vx
) 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)
print(sql)
db.quchen_text.execute(sql)
def channel_info_daily(ymd):
"""获取公众号某天的期数，投手，平台，书籍
@ return [[]]
"""
# 获取现在的全量公众号信息
sql="""select '{}' 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 advertiser_qq 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_qq where name is not null group by name,stage,pitcher,platform,book
union
select name,ifnull(stage,'') stage,ifnull(pitcher,'') pitcher,ifnull(platform,'') platform,ifnull(book,'') book from advertiser_vx where name is not null and name !=''
) b on a.name=b.name
""".format(ymd)
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))
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]
print(data)
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 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))
def dw_daily_channel():
"""快照表 每日一更新 t-1"""
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,total_cost,total_amount,reg_order_user_again,reg_order_user7,reg_order_user30,reg_order_amount7,type,
total_first_amount
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 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 dt dt5,channel channel5,total_cost,total_amount,total_first_amount from dw_channel_daily_total ) e
on dt=dt5 and channel=channel5
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
"""
ck.execute("truncate table dw_daily_channel")
ck.execute(sql)
print("ok")
def dm_pitcher_daily_page_total():
sql=f"""
insert into dm_pitcher_daily_page_total
select '{dt.get_n_days(0)}' dt,
q.pitcher,
round(cost,2),
round(amount,2),
round(roi,4),
channel_count,
on_channel_count,
channel_count-on_channel_count off_channel_count,
round(this_month_cost,2),
round(this_month_amount,2),
if(this_month_cost = 0, 0, round(this_month_amount / this_month_cost, 4)) this_month_roi,
last_month_cost,
last_month_amount,
if(last_month_cost = 0, 0, last_month_amount / last_month_cost) last_month_roi,
round(last_month_far_amount,2),
follow_user,
if(last_month_cost = 0, 0, last_month_far_amount / last_month_cost) last_month_far_roi
from (select pitcher,
sum(cost) cost,
sum(reg_order_amount) amount,
sum(follow_user) follow_user,
if(cost = 0, 0, round(amount / cost, 2)) roi
from dw_daily_channel
group by pitcher) q
left outer join
(select count(distinct channel) channel_count, pitcher
from dw_daily_channel_cost
where dt = '{dt.get_n_days(0)}'
group by pitcher) w
on q.pitcher = w.pitcher
left outer join(
select count(distinct channel) on_channel_count, pitcher
from dw_daily_channel_cost
where dt >= '{dt.get_n_days(-15)}'
and cost > 0
group by pitcher) e
on q.pitcher = e.pitcher
left outer join (
select pitcher,
sum(cost) this_month_cost
from dw_daily_channel
where dt >= '{dt.get_n_pre_month_first_day(0)}'
group by pitcher) r
on q.pitcher = r.pitcher
left outer join(
select pitcher,
sum(cost) last_month_cost
from dw_daily_channel
where dt >= '{dt.get_n_pre_month_first_day(1)}'
and dt < '{dt.get_n_pre_month_first_day(0)}'
group by pitcher) t on q.pitcher = t.pitcher
left outer join (
select b.pitcher, sum(amount) last_month_far_amount
from order a
left outer join dw_daily_channel_cost b on a.channel = b.channel and a.date = b.dt
where reg_time >= '{dt.get_n_pre_month_first_day(1)} 00:00:00' and reg_time<'{dt.get_n_pre_month_first_day(0)} 00:00:00'
group by pitcher
) y on q.pitcher = y.pitcher
left outer join (
select b.pitcher, sum(amount) last_month_amount
from order a
left outer join dw_daily_channel_cost b on a.channel = b.channel and a.date = b.dt
where reg_time >= '{dt.get_n_pre_month_first_day(1)} 00:00:00' and reg_time<'{dt.get_n_pre_month_first_day(0)} 00:00:00'
and dt<'{dt.get_n_pre_month_first_day(0)}'
group by pitcher
) p on q.pitcher=p.pitcher
left outer join (
select b.pitcher, sum(amount) this_month_amount
from order a
left outer join dw_daily_channel_cost b on a.channel = b.channel and a.date = b.dt
where reg_time >= '{dt.get_n_pre_month_first_day(0)} 00:00:00'
group by pitcher
) g on q.pitcher=g.pitcher
"""
print(sql)
ck.execute(f"alter table dm_pitcher_daily_page_total drop partition '{dt.get_n_days(0)}'")
ck.execute(sql)
def dw_channel_daily_total(ymd):
sql=f"""insert into dw_channel_daily_total
select '{ymd}' dt,channel,total_cost,total_amount,total_first_amount from
(select channel,sum(cost) total_cost from dw_daily_channel_cost where dt<='{ymd}' group by channel)a
left outer join
(select channel,sum(amount) total_amount,
sum(if(toDate(formatDateTime(reg_time,'%Y-%m-%d'))=date,amount,0)) total_first_amount
from order where date<='{ymd}' group by channel) b
on a.channel=b.channel
"""
ck.execute(f"alter table dw_channel_daily_total drop partition '{ymd}'")
print(sql)
ck.execute(sql)
def dw_daily_pitcher(ymd):
sql=f"""
insert into dw_daily_pitcher
select '{ymd}' dt, a.pitcher,
start_cost_date,end_cost_date,total_amount,total_cost
from
(select sum(cost) total_cost,pitcher from dw_daily_channel_cost where dt<='{ymd}' group by pitcher) a
left outer join
(select min(dt) start_cost_date,max(dt) end_cost_date,pitcher
from dw_daily_channel_cost where cost>0 group by pitcher) b on a.pitcher=b.pitcher
left outer join
(
select b.pitcher pitcher,sum(if(toDate(formatDateTime(reg_time,'%Y-%m-%d'))>=start_cost_date,amount,0)) total_amount
from order a
left outer join dw_daily_channel_cost b on a.channel = b.channel and date=dt
left outer join (select min(dt) start_cost_date,pitcher,channel from dw_daily_channel_cost where cost>0 group by pitcher,channel)
c on b.pitcher=c.pitcher and b.channel=c.channel where date<='{ymd}' group by b.pitcher
) d on a.pitcher=d.pitcher
"""
print(sql)
ck.execute(f"alter table dw_daily_pitcher drop partition '{ymd}'")
ck.execute(sql)
if __name__ == '__main__':
# dw_daily_pitcher('2021-01-14')
dm_pitcher_daily_page_total()
exit(0)
# dw_channel_daily_total('2020-07-20')
# channel_by_account_daily('2020-12-17')
# dw_daily_channel_cost('2020-12-17')
# dw_order_channel_cost_sync_ck('2020-12-17')
# exit(0)
# ods_order('2020-12-20')
# dw_daily_channel_plus()
# exit()
# dw_daily_channel()
# exit(0)
# dm_pitcher_daily_page_total()
# dm_pitcher_daily_page_total()
# exit(0)
# dw_channel_daily_total('2021-01-11')
# dw_daily_channel_cost('2021-01-12')
# dw_channel_daily_total('2021-01-13')
# exit(0)
# dw_daily_channel()
# # exit(0)
for i in dt.getDateLists('2019-03-18','2021-01-14'):
print(i)
# dw_channel_daily_total(i)
dw_daily_pitcher(i)
# channel_by_account_daily(i)
# dw_daily_channel_cost(i)
# dw_order_channel_cost_sync_ck(i)
# dw_daily_channel()
# order_sync_ck(today)
# # ods_order(i)
# channel_info_daily(i)
# dw_daily_channel_cost(i)
# dw_order_channel_cost_sync_ck(i)
# dw_channel_daily_total(i)