DataCenterGame/app/etl/rds_order_ck.py

"""把rds的订单数据同步到clickhouse
@auth ck
@date 20201127
"""

from model.DataBaseUtils import MysqlUtils,CkUtils
from model.DateUtils import DateUtils
from datetime import datetime
from model.log import logger
log=logger()
db = MysqlUtils()
ck = CkUtils()
dt = DateUtils()
import sys
from model.DingTalkUtils import DingTalkUtils


def run(mydt):
    log.debug("run [{}] data".format(mydt))
    col = "date,stage,platform,channel,channel_id,user_id,order_time,reg_time,amount,from_novel,order_id"

    sql ="select * from `order` where date=UNIX_TIMESTAMP('{}') ".format(mydt)

    data = db.quchen_text.getData(sql)

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

        li[6] = li[6].replace('T',' ').replace('+08:00','').replace('/','-')

        if li[6] == None or li[6] == '':
            li[6] = '0000-00-00 00:00:00'

        if li[7]==None or li[7]=='':
            li[7]='0000-00-00 00:00:00'
        else:
            li[7] = li[7].replace('T',' ').replace('+08:00','').replace('/','-')

        try:
            li[6]=str(datetime.strptime(li[6],'%Y-%m-%d %H:%M'))
            li[7]=str(datetime.strptime(li[7],'%Y-%m-%d %H:%M'))
            # li[7]=str(datetime.strptime(li[7],'%Y/%m/%d %H:%M:%S'))
        except:
            pass
            # print("======================")
        #     try:
        #         li[6] = str(datetime.strptime(li[6][:17], '%Y-%m-%d %H:%M'))
        #         li[7] = str(datetime.strptime(li[7][:17], '%Y/%m/%d %H:%M'))
        #     except:
        #         pass
                # li[6]='2020-11-01 00:00:00'
                # li[7]='2020-11-01 00:00:00'

        li[6] =li[6][:19]
        li[7] =li[7][:19]

        li[9] = '' if li[9] is None else li[9]
        li[5] = '' if li[5] is None else li[5]
        li[1] = '' if li[1] is None else li[1]

        data1.append(tuple(li))

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


if __name__ == '__main__':
    # dt_li =dt.getDateLists('2020-11-25','2020-11-27')
    # for i in dt_li:
    #     run(i)

    if sys.argv.__len__()>1:
        run(dt.get_n_days(-1))

    today=dt.get_n_days()
    try:
        run(today)
    except Exception as e:
        DingTalkUtils.send("订单同步到ck出错:"+e,'18860455786,')