DataCenterGame/app/etl/sync_to_ck_task.py

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


def order(ymd):
    log.debug("sync order")
    col = "date,stage,platform,channel,channel_id,user_id,order_time,reg_time,amount,from_novel,order_id"

    sql ="select * from ods_order where date='{}'".format(ymd)

    data = db.quchen_text.getData(sql)

    # # 掌中云时间格式转换
    data1 = []
    for x in data:
        li = list(x)

        li[0]=str(li[0])
        li[6]=str(li[6])
        li[7]=str(li[7])
        li[9]='' if li[9]==None else li[9]

        data1.append(tuple(li))

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


def daily_qq(mydt):
    log.debug("sync daily_qq")
    table='daily_qq'
    col=db.quchen_text.getColumn("daily_qq")
    sql = "select * from {} where date='{} 00:00:00'".format(table,mydt)
    data = db.quchen_text.getData(sql)
    li1 = []

    for i in data:
        li2 = list(i)
        li2[1] = str(li2[1])
        li2[12]=round(i[12],4)
        li1.append(tuple(li2))
    ck.execute("alter table {} drop  partition '{}' ".format(table, mydt))
    ck.insertMany(table, col, li1)


def daily_vx(mydt):
    log.debug("sync daily_vx")
    table='daily_vx'
    col=db.quchen_text.getColumn(table)
    sql = "select * from {} where date='{} 00:00:00'".format(table,mydt)
    data = db.quchen_text.getData(sql)
    # print(len(data))
    li1 = []


    for i in data:
        li2 = list(i)
        li2[1] = str(li2[1])
        li2[5] = round(i[5],4)
        li2[6] = round(i[6],4)
        li2[7] = round(i[7],4)
        li2[8] = round(i[8],4)
        li2[10] = round(i[10],4)
        li1.append(tuple(li2))

    ck.execute("alter table {} drop  partition '{}' ".format(table, mydt))
    ck.insertMany(table, col, li1)


# 广告计划
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 advertiser_vx_qq():
    sql = "select account_id,name,stage,pitcher,platform,book,'vx' as tag from advertiser_vx " \
          " union " \
          "select account_id,name,stage,pitcher,platform,book,'qq' as tag from advertiser_qq"
    data = db.quchen_text.getData(sql)
    data1 = []

    for i in data:
        li = list(i)
        for idx, val in enumerate(li):
            if val == None:
                li[idx] = ''
        data1.append(tuple(li))

    ck.execute("truncate table advertiser_vx_qq")
    ck.execute("insert into advertiser_vx_qq  (account_id,name,stage,pitcher,platform,book,tag) values " + str(data1)[1:-1])


def pitcher_change():
    sql="select channel,pitcher,ifnull(start_time,'') as start_time," \
        " ifnull(end_time,'') as end_time  from pitcher_change"
    data=db.quchen_text.getData(sql)
    ck.execute("truncate table pitcher")
    ck.execute("insert into pitcher_change values {}".format(str(data)[1:-1]))


def book_change():
    table="book_change"
    col="name,book,start_time,end_time"
    sql="select name,book,ifnull(start_time,'') as start_time,ifnull(end_time,'') as end_time from {}".format(table)
    data=db.quchen_text.getData(sql)
    ck.execute("truncate table {}".format(table))
    ck.insertMany(table,col,data)

def account_change():
    table = "account_change"
    col = "account_id,name,pitcher,start_time,end_time"
    sql = "select account_id,name,pitcher,ifnull(start_time,'') as start_time,ifnull(end_time,'') as end_time from {}".format(table)
    data = db.quchen_text.getData(sql)
    ck.execute("truncate table {}".format(table))
    ck.insertMany(table, col, data)


def platform_change():
    table = "platform_change"
    col = "name,primary_platform,current_platform,change_date"
    sql = "select name,primary_platform,current_platform,date_format(change_date,'%Y-%m-%d') as change_date from {}".format(table)
    data = db.quchen_text.getData(sql)
    ck.execute("truncate table {}".format(table))
    ck.insertMany(table, col, data)


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__':
    for i in dt.getDateLists('2019-05-20','2020-12-27'):
        order(i)