DataCenterGame/app/etl/sync_to_ck_task.py

from model.DataBaseUtils import MysqlUtils,CkUtils
from model.DateUtils import DateUtils
import logging
db = MysqlUtils()
ck = CkUtils()
dt = DateUtils()



def order_sync_ck(ymd):
    logging.info("sync order")
    col = """date,stage,platform,channel,channel_id,user_id,order_time,reg_time,amount,from_novel,order_id,status,
    platform_user_id,wechat_app_id,book_tags,order_type"""
    sql =f"""select  DATE_FORMAT(date,"%Y-%m-%d") date,
            stage,platform,channel,channel_id,user_id,order_time,reg_time,amount,from_novel,order_id,status,
    ifnull(platform_user_id,''),
    ifnull(wechat_app_id,''),
    ifnull(book_tags,''),
    ifnull(order_type,1) 
    from ods_order where date='{ymd}'"""
    data = db.quchen_text.getData(sql)
    data1 = []
    for x in data:
        li = list(x)
        li[0]=str(li[0])
        li[3]='' if li[3]==None else li[3].replace(' ','')
        li[5]='' if li[5]==None else li[5]
        li[6]=str(li[6])
        li[7]='0000-00-00 00:00:00' if li[7]==None else str(li[7])
        li[9]='' if li[9]==None else li[9]
        data1.append(tuple(li))

    # 删除分区
    ck.execute("alter table game_data.order drop  partition '{}' ".format(ymd))
    ck.insertMany("game_data.order", col, tuple(data1))


# 广告计划
def daily_vx_campaign(ymd):
    table = 'daily_vx_campaign'
    sql = 'select * from ' + table + ' where date="{} 00:00:00" '.format(ymd)
    data = db.quchen_text.getData(sql)
    data1 = []
    for i in data:
        li = list(i)
        li[1] = str(li[1])
        li[5] = round(li[5], 4)
        li[6] = round(li[6], 4)
        li[7] = round(li[7], 4)
        li[8] = round(li[8], 4)
        li[12] = round(li[12], 4)
        li[13] = round(li[13], 4)
        li[15] = round(li[15], 4)
        data1.append(tuple(li))
    col = db.quchen_text.getColumn("daily_vx_campaign")
    ck.execute("alter table order drop  partition '{}' ".format(ymd))
    ck.insertMany(table, col, tuple(data1))


def adcreative():
    """todo:表需要重新设计 无法分区"""
    pass


def campaign_vx():
    """todo:表需要重新设计 无法分区"""
    # table="campaign_vx"
    # sql="select * from {}".format(table)
    # data=db.quchen_text.getData(sql)
    pass


if __name__ == '__main__':
    du = DateUtils()
    # order_sync_ck('2021-06-03')
    # dw_order_channel_cost_sync_ck(dt.get_n_days(-2))
    for i in dt.getDateLists('2021-03-19',du.getNow()):
    # #     order(i)
    #     dw_order_channel_sync_ck(i)
        print(i)
        order_sync_ck(i)