python_physical_world_v3/start.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
import json
from physical_world_exporter.MyUtils.ConfigUtils import ConfigUtils
from physical_world_exporter.MyUtils.ZillionUtil import ZillionUtil
import os,time,datetime

#插入hbase
def put_hbase(hbasedatabase,hbasetable,datas):
    for i in range(0, len(datas), 1000):
        dataranges = datas[i:i + 1000]
        zillionUtil.put(hbasedatabase, hbasetable, dataranges)
#删除hbase数据
def remove_hbase(hbasedatabase,hbasetable,key):
    zillionUtil.remove(hbasedatabase, hbasetable, key)

#删除hbase数据
def delete_hbase(hbasedatabase,hbasetable):
    Criteria = {"project_id":"3301100002"}
    zillionUtil.delete(hbasedatabase, hbasetable, Criteria)

#获取表主键
def get_hbase_key(hbasedatabase, hbasetable):
    keys = zillionUtil.table_key(hbasedatabase, hbasetable)
    return keys


datetimenow = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
datetimenow_strp = datetime.datetime.strptime(datetimenow,"%Y-%m-%d %H:%M:%S")
print(type(datetimenow_strp),datetimenow_strp)

config = ConfigUtils("physical_world_exporter/config.xml")
url, hbasedatabase = config.readTop("metadata", ["url", "database"])
sourcepath, targetpath = config.readTop("cmd", ["sourcepath", "targetpath"])

#连接hbase
zillionUtil = ZillionUtil(url)

# #切换到输出目录，清空文件
# os.chdir(targetpath)
# os.system("rm -rf *")

#切换到java程序工作目录，执行导出数据程序
os.chdir(sourcepath)
print("进入%s目录，开始执行java程序"%os.getcwd())
cmd = "java -jar -Dfile.encoding=UTF-8 data-migration.jar"
status = os.system(cmd)
#如果程序执行成功
if status == 0:
    print("导出程序执行成功")
    #切换到输出目录，判断文件是否是最新文件
    os.chdir(targetpath)
    print("进入%s目录，检查文件并写入hbase"%os.getcwd())
    list = os.listdir(os.getcwd())
    for file in list:
        #linux 导出程序bug，需要处理下文件名称
        # print(i)
        # file = i.lstrip("\\")
        # cmd_mv = "mv \%s %s"%(i,file)
        # print(cmd_mv)
        # os.system(cmd_mv)
        updatetime = os.path.getmtime(file) #查询文件修改时间
        timeArray = time.localtime(updatetime)
        otherStyleTime = time.strftime("%Y-%m-%d %H:%M:%S", timeArray)
        otherStyleTime_strp = datetime.datetime.strptime(otherStyleTime,"%Y-%m-%d %H:%M:%S")
        detal_time = (otherStyleTime_strp - datetimenow_strp).total_seconds()
        print("%s文件最新修改时间：%s"%(file,otherStyleTime))
        hbasetable = file.split(".")[0]
        if int(detal_time) < 10000:
            with open(file,"r",encoding = 'utf-8') as fp:
                print(fp.name)
                if file == "rel_btw_objs.json":
                    datas_rel = []
                    for line in fp.readlines():
                        data = json.loads(line)
                        datas_rel.append(data)
                    print("删除%s"%fp.name)
                    delete_hbase(hbasedatabase,hbasetable)
                    put_hbase(hbasedatabase, hbasetable, datas_rel)
                else:
                    keys = get_hbase_key(hbasedatabase, hbasetable)
                    datas = []
                    for line in fp.readlines():
                        data = json.loads(line)
                        hbase_key = {}
                        for key in keys:
                            if key in data:
                                hbase_key[key] = data[key]
                        datas.append(data)
                        remove_hbase(hbasedatabase,hbasetable,hbase_key)
                    time.sleep(0.1)
                    put_hbase(hbasedatabase,hbasetable,datas)
            print("%s写入完成"%hbasetable)
        else:
            print("检查%s导出文件数据，可能不是最新数据"%file)
else:
    print("执行java程序失败")