from model.DataBaseUtils import MysqlUtils, CkUtils from model.DateUtils import DateUtils import logging db = MysqlUtils() ck = CkUtils() dt = DateUtils() def dw_daily_channel_cost(ymd): logging.info("run> dw_daily_channel_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, if(stage ='趣程15期' or stage ='趣程26期' or stage ='趣程30期','GDT','MP') type ,0 require_roi,0 require_mult 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 """.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_channel_cost drop partition '{ymd}' ") logging.info(len(data1)) ck.insertMany("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 ) 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 = """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)) 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('2020-04-23','2021-02-05'): # print(i) # dw_daily_channel_cost(i) ods_order('2021-05-06')