Kaynağa Gözat

transfer_data

李莎 2 yıl önce
işleme
8b22a17d72

+ 49 - 0
MyUtils/ConfigUtils.py

@@ -0,0 +1,49 @@
+import xml.etree.ElementTree as ET
+
+
+#读取xml文件
+class ConfigUtils():
+    def __init__(self,file):
+        self.url = ""
+        self.tree = ET.parse(file)
+        self.root = self.tree.getroot()
+
+    def readTop(self, key, child):
+        datas = []
+        data = self.root.find(key)
+        for key in child:
+            datas.append(data.get(key))
+        return datas
+
+    def readTopDict(self, key, child):
+        datas = {}
+        data = self.root.find(key)
+        for key in child:
+            datas[key] = data.get(key)
+        return datas
+
+    def readConfig(self,parent,child):
+        datas=[]
+        for childLine in self.root.find(parent):
+            data=[]
+            for key in child:
+                data.append(childLine.get(key))
+            datas.append(data)
+        return datas
+
+    def readConfigDict(self,parent,child):
+        datas=[]
+        for childLine in self.root.find(parent):
+            data={}
+            for key in child:
+                data[key]=childLine.get(key)
+            datas.append(data)
+        return datas
+
+    def readConfigSingle(self,key):
+        data = self.root.find(key).text
+        return data
+
+
+if __name__ == '__main__':
+    config = ConfigUtils("logger.conf")

+ 137 - 0
MyUtils/DateUtils.py

@@ -0,0 +1,137 @@
+import datetime
+from dateutil.relativedelta import relativedelta
+from dateutil import rrule
+
+YYmmddHHMMSS = "%Y%m%d%H%M%S"
+YYmmdd = "%Y%m%d"
+YYmm = "%Y%m"
+# 根据开始月份结束月份获取所有月份
+def get_each_month(start_month, end_month):
+    if str(start_month).count('-') != 1 or str(end_month).count('-') != 1:
+        print("Parameter Error: Pls input a string such as '2019-01'")
+        return []
+    if int(str(start_month).split('-')[1]) > 12 or int(str(end_month).split('-')[1]) > 12:
+        print('Parameter Error: Pls input correct month range such as between 1 to 12')
+        return []
+    if int(str(start_month).split('-')[1]) == 0 or int(str(end_month).split('-')[1]) == 13:
+        print('Parameter Error: Pls input correct month range such as between 1 to 12')
+        return []
+    start = datetime.datetime.strptime(start_month, "%Y-%m")
+    end = datetime.datetime.strptime(end_month, "%Y-%m")
+    month_count = rrule.rrule(rrule.MONTHLY, dtstart=start, until=end).count()  # 计算总月份数
+    if end < start:
+        print("Parameter Error: Pls input right date range,start_month can't latter than end_month")
+        return []
+    else:
+        list_month = []
+        year = int(str(start)[:7].split('-')[0])  # 截取起始年份
+        for m in range(month_count):  # 利用range函数填充结果列表
+            month = int(str(start)[:7].split('-')[1])  # 截取起始月份,写在for循环里,作为每次迭代的累加基数
+            month = month + m
+            if month > 12:
+                if month % 12 > 0:
+                    month = month % 12  # 计算结果大于12,取余数
+                    if month == 1:
+                        year += 1  # 只需在1月份的时候对年份加1,注意year的初始化在for循环外
+                else:
+                    month = 12
+            if len(str(month)) == 1:
+                list_month.append(str(year) + '-0' + str(month))
+            else:
+                list_month.append(str(year) + '-' + str(month))
+        return list_month
+
+
+# 转换格式,去掉"-",%Y%m
+def get_eachmonth(start_month, end_month):
+    startmonth = start_month[0:4] + "-" + start_month[4:6]
+    endmonth = end_month[0:4] + "-" + end_month[4:6]
+    months = get_each_month(startmonth, endmonth)
+    list_month = [i[0:4] + i[5:7] for i in months]
+    return list_month
+
+
+def get_month(starttime, endtime):
+    months = []
+    starttime = datetime.datetime.strptime(starttime, YYmmddHHMMSS)
+    endtime = datetime.datetime.strptime(endtime, YYmmddHHMMSS)
+    while starttime <= endtime:
+        start = starttime.strftime("%Y%m")
+        if start not in months:
+            months.append(start)
+        starttime = starttime + datetime.timedelta(days=1)
+    return months
+
+def get_month_range(starttime, endtime):
+    months = []
+    starttime = datetime.datetime.strptime(starttime[0:8], YYmmdd)
+    endtime = datetime.datetime.strptime(endtime[0:8], YYmmdd)
+    while starttime <= endtime:
+        start = starttime.strftime(YYmmddHHMMSS)
+        endMonth = datetime.datetime.strptime(start[0:6], YYmm) + relativedelta(months=+1)
+
+        end = endMonth.strftime(YYmmddHHMMSS)
+        if endtime <= endMonth:
+            end = endtime.strftime(YYmmddHHMMSS)
+        starttime = endMonth
+        months.append([start, end])
+    return months
+
+
+
+def get_month1(starttime, endtime):
+    months = []
+    startyear = int(starttime[0:4])
+    startmonth = int(starttime[4:6])
+    endyear = int(endtime[0:4])
+    endmonth = int(endtime[4:6])
+    while startyear != endyear or startmonth != endmonth:
+        startyearstr = str(startyear)
+        startmonthstr = str(startmonth)
+        if startmonth < 10:
+            startmonthstr = "0" + str(startmonth)
+        months.append(startyearstr + startmonthstr)
+        startmonth += 1
+        if startmonth == 13:
+            startyear += 1
+            startmonth = 1
+    months.append(endtime[0:6])
+    return months
+
+#根据开始结束时间获取最大区间为1天的时间区间
+# def get_day(starttime,endtime):
+#     times = []
+#     starttime = datetime.datetime.strptime(starttime, YYmmddHHMMSS)
+#     endtime = datetime.datetime.strptime(endtime, YYmmddHHMMSS)
+#     while starttime < endtime:
+#         start = starttime.strftime(YYmmddHHMMSS)
+#         starttime_delta = starttime + datetime.timedelta(days=1)
+#         end = starttime_delta.strftime(YYmmddHHMMSS)
+#         if str(starttime)[0:7]<str(starttime_delta)[0:7] :
+#             end = starttime_delta.strftime("%Y%m") +"01000000"
+#             starttime = datetime.datetime.strptime(end, YYmmddHHMMSS)
+#             times.append([start, end])
+#             continue
+#         if starttime >= starttime_delta:
+#             end = starttime_delta.strftime(YYmmddHHMMSS)
+#         starttime = starttime + datetime.timedelta(days=1)
+#         times.append([start,end])
+#     return times
+#根据开始结束时间获取最大区间为1天的时间区间
+def get_day(starttime,endtime):
+    times = []
+    starttime = datetime.datetime.strptime(starttime, YYmmddHHMMSS)
+    endtime = datetime.datetime.strptime(endtime, YYmmddHHMMSS)
+    while starttime < endtime:
+        start = starttime.strftime(YYmmddHHMMSS)
+        stratDay = datetime.datetime.strptime(start[0:8],YYmmdd)
+        starttime_delta = stratDay + datetime.timedelta(days=1)
+        end = starttime_delta.strftime(YYmmddHHMMSS)
+        if endtime <= starttime_delta:
+            end = endtime.strftime(YYmmddHHMMSS)
+        starttime = stratDay + datetime.timedelta(days=1)
+        times.append([start,end])
+    return times
+
+if __name__ == '__main__':
+    print(get_month_range("20200106000000", "20200115000000"))

+ 42 - 0
MyUtils/HttpUtil.py

@@ -0,0 +1,42 @@
+# coding=utf-8
+import sys
+
+if (sys.version_info.major == 3):
+    from urllib import parse
+    import urllib.request as urllib2
+else:
+    import urllib2
+
+
+class HttpUtil(object):
+    #post方法获取数据  application/json
+    @staticmethod
+    def post(url, postData):  #
+        req = urllib2.Request(url, data=postData.encode('utf-8'),
+                              headers={'Content-Type': 'application/json;charset=UTF-8'})
+        res = urllib2.urlopen(req, timeout=60).read().decode("utf-8")
+        return res
+
+    #post方法获取数据  application/x-www-form-urlencoded
+    @staticmethod
+    def postText(url, postData):
+        req = urllib2.Request(url, data=postData.encode('utf-8'),
+                              headers={'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8'})
+        res = urllib2.urlopen(req, timeout=60).read().decode("utf-8")
+        return res
+
+    #get方法获取数据  text/xml
+    @staticmethod
+    def get(url, getData, isquote):
+        if isquote:
+            getData = parse.quote(getData)
+        req = urllib2.Request((url + getData), headers={'Content-Type': 'text/xml;charset=UTF-8'})
+        res = urllib2.urlopen(req, timeout=60)
+        req = res.read()
+        return req.decode()
+
+
+# if __name__ == '__main__':
+    # httputil = HttpUtil()
+    # httputil.postText()
+    # HttpUtil.postText()

+ 53 - 0
MyUtils/MetadataWebUtil.py

@@ -0,0 +1,53 @@
+import json
+from MyUtils.HttpUtil import *
+
+
+class MetadataWebUtil(object):
+    def __init__(self, url):
+        self.url = url
+
+    def get_hbase(self, params):
+        params = json.dumps(params)
+        word = '<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:ser="http://service.persagy.com/" xmlns:ent="http://entity.persagy.com">' \
+               '<soapenv:Header/>' \
+               '<soapenv:Body><ser:Query>' \
+               '<arg0>%s</arg0>' \
+               '</ser:Query></soapenv:Body></soapenv:Envelope>' % (params)
+        res = HttpUtil.post(self.url, word)
+        res = res.split("<return>")[1].split("</return>")[0]
+        res = json.loads(res.replace("&#xd;", ""))
+        return res
+
+    def database_list(self):
+        params = {
+            "QueryType": "database_list"
+        }
+        res = self.get_hbase(params)
+        return res["Content"]
+
+    def table_list(self, database):
+        params = {
+            "QueryType": "table_list_strict",
+            "Database": database
+        }
+        table_list_strict = self.get_hbase(params)["Content"]
+        params = {
+            "QueryType": "table_list",
+            "Database": database
+        }
+        table_list = self.get_hbase(params)["Content"]
+        table_list = [i for i in table_list if i not in table_list_strict]
+        tables = {}
+        for table in table_list_strict:
+            childs = []
+            for i in table_list:
+                if i.startswith(table + "_"):
+                    try:
+                        word = i[len(table + "_"):]
+                        if len(word) == 4 or len(word) == 6:
+                            word = int(word)
+                            childs.append(i)
+                    except:
+                        pass
+            tables[table] = childs
+        return tables

+ 70 - 0
MyUtils/MysqlUtils.py

@@ -0,0 +1,70 @@
+import pymysql
+from dbutils.pooled_db import PooledDB
+
+
+class MysqlUtils():
+    # 连接池对象
+    __pool = None
+    def __init__(self,**config):
+        self.host = config["host"]
+        self.port = int(config["port"])
+        self.user = config["user"]
+        self.passwd = config["passwd"]
+        self.database = None
+        if "database" in config:
+            self.database = config["database"]
+        # self.pool = self.get_conn()
+        self.pool = None
+
+    def get_conn(self):
+        if MysqlUtils.__pool is None:
+            __pool = PooledDB(pymysql, mincached=1, maxcached=5, maxconnections=5,
+                              host=self.host, port=self.port, user=self.user, passwd=self.passwd,
+                              database=self.database,
+                               use_unicode=False, blocking=False, charset="utf8")
+            self.pool = __pool
+
+
+    def query(self,sql):
+        con = None
+        cur = None
+        try:
+            con = self.pool.connection()
+            cur = con.cursor()
+            count = cur.execute(sql)
+            if count >= 0:
+                result = cur.fetchall()
+                result = [[j.decode() if isinstance(j,bytes) else j for j in i] for i in result]
+                # print(result)
+            else:
+                result = False
+            return result
+        except Exception as e:
+            print(e)
+        finally:
+            if cur:
+                cur.close()
+            if con:
+                con.close()
+
+
+    def update(self,sql):
+        con = None
+        cur = None
+        try:
+            con = self.pool.connection()
+            cur = con.cursor()
+            cur.execute(sql)
+            con.commit()
+        except Exception as e:
+            con.rollback()  # 事务回滚
+            print(e)
+        finally:
+            if cur:
+                cur.close()
+            if con:
+                con.close()
+
+    def close(self):
+        if MysqlUtils.__pool:
+            MysqlUtils.__pool.close()

+ 104 - 0
MyUtils/ZillionUtil.py

@@ -0,0 +1,104 @@
+from MyUtils.MetadataWebUtil import MetadataWebUtil
+
+
+class ZillionUtil():
+    def __init__(self, url):
+        self.metadata = MetadataWebUtil(url)
+
+    ##列出所有数据库
+    def database_list(self):
+        param = {
+            "QueryType": "database_list"
+        }
+        databases = self.metadata.get_hbase(param)
+        return databases["Content"]
+
+        # TODO  其他zillion语句
+
+    ##列出所有数据库下的表(不包含分表)
+    def table_list_strict(self,table):
+        param = {
+            	"QueryType":"table_list_strict",
+	            "Database":table
+        }
+        databases = self.metadata.get_hbase(param)
+        return databases["Content"]
+
+
+    ##列出所有数据库
+    # def query(self, database, table, Criteria):
+    #     param = {
+    #         "QueryType": "select",
+    #         "Database": database,
+    #         "Datatable": table,
+    #         "Criteria": Criteria
+    #     }
+    #     print(param)
+    #     databases = self.metadata.get_hbase(param)
+    #     return databases["Content"]
+
+    def select(self, database, table, Criteria):
+        param = {
+            "QueryType": "select",
+            "Database": database,
+            "Datatable": table,
+            "Criteria": Criteria
+        }
+        databases = self.metadata.get_hbase(param)
+        return databases["Content"]
+
+    def table_key(self, database, table):
+        param = {
+            "QueryType": "table_detail",
+            "Database": database,
+            "Datatable": table
+        }
+        databases = self.metadata.get_hbase(param)
+        key = databases["Content"]["Key"]
+        return key
+
+
+
+    def select_count(self, database, table, Criteria):
+        param = {
+            "QueryType": "select_count",
+            "Database": database,
+            "Datatable": table,
+            "Criteria": Criteria
+        }
+        databases = self.metadata.get_hbase(param)
+        return databases["Count"]
+
+    def query_data(self, sql):
+        databases = self.metadata.get_hbase(sql)
+        return databases["Content"]
+
+    def insert(self,database,table,InsertObject):
+        param = {
+            "QueryType": "batch_insert",
+            "Database": database,
+            "Datatable": table,
+            "InsertObjects": InsertObject
+        }
+        databases = self.metadata.get_hbase(param)
+        return databases["Count"]
+
+    def remove(self,database,table,key):
+        param = {
+            "QueryType": "remove",
+            "Database": database,
+            "Datatable": table,
+            "Key": key
+        }
+        databases = self.metadata.get_hbase(param)
+        return databases["Count"]
+
+    def put(self,database,table,InsertObject):
+        param = {
+            "QueryType": "batch_put",
+            "Database": database,
+            "Datatable": table,
+            "InsertObjects": InsertObject
+        }
+        databases = self.metadata.get_hbase(param)
+        return databases["Result"]

+ 1 - 0
config-time-real.json

@@ -0,0 +1 @@
+{"from_time": "20220421000000", "to_time": "20220422000000", "schedule_task": "16:52:00"}

+ 1 - 0
config-time.json

@@ -0,0 +1 @@
+{"from_time": "20210614000000", "to_time": "20210617000000"}

+ 28 - 0
config.json

@@ -0,0 +1,28 @@
+{
+  "metadata": {
+    "database": "db_time_data",
+    "url": "http://47.93.33.207:8890/metadata-web/services/Service1_WS?wsdl"
+  },
+  "kafka": {
+    "topic": "test",
+    "host": "192.168.100.114",
+    "port": 9092
+  },
+  "building": {
+    "id": "1101080259"
+  },
+  "mysql": {
+    "host": "192.168.100.197",
+    "port": 3306,
+    "user": "root",
+    "password": "j5ry0#jZ7vaUt5f4",
+    "database": "saga_dev",
+    "fjd_table": "energy_15_min_fjd_py",
+    "energy_table": "energy_15_min_py",
+    "co2_table": "co2_15_min",
+    "pm25_table": "pm25_15_min",
+    "temperature_table": "temperature_15_min",
+    "hcho_table": "hcho_15_min",
+    "humidity_table": "humidity_15_min"
+  }
+}

+ 107 - 0
start_kafka.py

@@ -0,0 +1,107 @@
+import json
+from MyUtils.ZillionUtil import ZillionUtil
+from MyUtils.DateUtils import get_month_range
+from kafka import KafkaProducer
+import datetime
+import argparse
+
+
+# 获取项目点位
+def get_pointlist(hbase_database, hbase_table, buildingid):
+    Criteria = {
+        "building": buildingid
+    }
+    datas = zillionUtil.select(hbase_database, hbase_table, Criteria)
+    pointlist = []
+    for i in datas:
+        point = []
+        meter = i["meter"]
+        funcid = i["funcid"]
+        point.append(meter)
+        point.append(funcid)
+        pointlist.append(point)
+    return pointlist
+
+# 获取能耗数据
+def get_data_time(hbase_database, hbase_table, buildingid, meter, funcid,from_time,to_time):
+    Criteria = {
+        "building": buildingid,
+        "meter": meter,
+        "funcid": funcid,
+        "data_time":{
+		    "$gte": from_time,
+		    "$lt": to_time
+	    }
+    }
+    datas = zillionUtil.select(hbase_database, hbase_table, Criteria)
+    return datas
+
+#kafka发送数据
+def send_kafka(topic,msg):
+    msg = json.dumps(msg)
+    msg = msg.encode('utf-8')
+    future = producer.send(topic, msg)
+    try:
+        record_metadata = future.get(timeout=10)
+        partition = record_metadata.partition
+        offset = record_metadata.offset
+        # print('save success, partition: {}, offset: {}'.format(partition, offset))
+    except Exception as e:
+        print("Error:{}".format(e))
+
+
+now_time = datetime.datetime.now().strftime("%Y%m%d") + "000000"
+end_time = (datetime.datetime.now() + datetime.timedelta(days=1)).strftime("%Y%m%d")
+# 读取配置文件信息
+with open("config.json", "r") as f:
+    data = json.load(f)
+    hbase_database = data["metadata"]["database"]
+    url = data["metadata"]["url"]
+    building = data["building"]["id"]
+    topic = data["kafka"]["topic"]
+    kafka_host = data["kafka"]["host"]
+    kafka_port = data["kafka"]["port"]
+
+with open("config-time.json", "r") as f:
+    data_time = json.load(f)
+    from_time = data_time["from_time"]
+    to_time = data_time["to_time"]
+
+#可启动传参,python  xxx.py --start_time "20210701000000" --end_time "20210702000000"
+parser = argparse.ArgumentParser()
+parser.add_argument("--start_time",default=from_time,help="--start_time 20220101000000")
+parser.add_argument("--end_time",default=to_time,help="--end_time 20220201000000")
+args = parser.parse_args()
+
+# print(args.start_time)
+# print(args.end_time)
+
+
+#连接kafka
+producer = KafkaProducer(bootstrap_servers='%s:%s'%(kafka_host,kafka_port))  # 连接kafka
+
+
+tables = ["fjd_0_near_15min","data_servicedata_15min"]
+# #连接hbase
+zillionUtil = ZillionUtil(url)
+pointlist = get_pointlist(hbase_database,"dy_pointlist",building)
+
+for table in tables:
+    for i in pointlist:
+        meter,funcid = i[0],i[1]
+        monthrange = get_month_range(args.start_time,args.end_time)
+        for m in monthrange:
+            startdate,enddate = m[0],m[1]
+            print("%s开始查询%s至%s的数据 %s-%s"%(table,startdate,enddate,meter,funcid))
+            if table == "fjd_0_near_15min":
+                data_fjd = get_data_time(hbase_database,table,building,meter,funcid,startdate,enddate)
+                if data_fjd:
+                    for i in data_fjd:
+                        i.pop("data_flag")
+                        send_kafka(topic,i)
+            else:
+                data_fjd = get_data_time(hbase_database, table, building, meter, funcid, startdate, enddate)
+                if data_fjd:
+                    for i in data_fjd:
+                        send_kafka(topic,i)
+producer.close()

+ 111 - 0
start_kafka_real.py

@@ -0,0 +1,111 @@
+import json,time
+from MyUtils.ZillionUtil import ZillionUtil
+from MyUtils.DateUtils import get_month_range
+from kafka import KafkaProducer
+import datetime
+
+
+# 获取项目点位
+def get_pointlist(hbase_database, hbase_table, buildingid):
+    Criteria = {
+        "building": buildingid
+    }
+    datas = zillionUtil.select(hbase_database, hbase_table, Criteria)
+    pointlist = []
+    for i in datas:
+        point = []
+        meter = i["meter"]
+        funcid = i["funcid"]
+        point.append(meter)
+        point.append(funcid)
+        pointlist.append(point)
+    return pointlist
+
+# 获取能耗数据
+def get_data_time(hbase_database, hbase_table, buildingid, meter, funcid,from_time,to_time):
+    Criteria = {
+        "building": buildingid,
+        "meter": meter,
+        "funcid": funcid,
+        "data_time":{
+		    "$gte": from_time,
+		    "$lt": to_time
+	    }
+    }
+    datas = zillionUtil.select(hbase_database, hbase_table, Criteria)
+    return datas
+
+#kafka发送数据
+def send_kafka(topic,msg):
+    msg = json.dumps(msg)
+    msg = msg.encode('utf-8')
+    future = producer.send(topic, msg)
+    try:
+        record_metadata = future.get(timeout=10)
+        # partition = record_metadata.partition
+        # offset = record_metadata.offset
+        # print('save success, partition: {}, offset: {}'.format(partition, offset))
+    except Exception as e:
+        print("Error:{}".format(e))
+
+yesterday_time = (datetime.datetime.now()+datetime.timedelta(days=-1)).strftime("%Y%m%d") + "000000"
+now_time = datetime.datetime.now().strftime("%Y%m%d") + "000000"
+
+
+
+
+
+# 读取配置文件信息
+with open("config.json", "r") as f:
+    data = json.load(f)
+    hbase_database = data["metadata"]["database"]
+    url = data["metadata"]["url"]
+    building = data["building"]["id"]
+    topic = data["kafka"]["topic"]
+    kafka_host = data["kafka"]["host"]
+    kafka_port = data["kafka"]["port"]
+
+#先修改文件时间
+with open("config-time-real.json", "r") as f:
+    data_time = json.load(f)
+data_time["from_time"] = yesterday_time
+data_time["to_time"] = now_time
+with open("config-time-real.json", "w") as f_config:
+    json.dump(data_time,f_config)
+with open("config-time-real.json", "r") as f:
+    data_time = json.load(f)
+    from_time = data_time["from_time"]
+    to_time = data_time["to_time"]
+    schedule_task = data_time["schedule_task"]
+
+tables = ["fjd_0_near_15min","data_servicedata_15min"]
+print("------------------等待下次运行时间%s------------------" % schedule_task)
+while True:
+    time_now = time.strftime("%H:%M:%S", time.localtime())  # 刷新
+    if time_now == schedule_task:  # 此处设置每天定时的时间
+        #连接kafka
+        producer = KafkaProducer(bootstrap_servers='%s:%s'%(kafka_host,kafka_port))  # 连接kafka
+        # #连接hbase
+        zillionUtil = ZillionUtil(url)
+        pointlist = get_pointlist(hbase_database,"dy_pointlist",building)
+
+        for table in tables:
+            for i in pointlist:
+                meter,funcid = i[0],i[1]
+                monthrange = get_month_range(from_time,to_time)
+                for m in monthrange:
+                    startdate,enddate = m[0],m[1]
+                    print("%s开始查询%s至%s的数据 %s-%s"%(table,startdate,enddate,meter,funcid))
+                    if table == "fjd_0_near_15min":
+                        data_fjd = get_data_time(hbase_database,table,building,meter,funcid,startdate,enddate)
+                        if data_fjd:
+                            for i in data_fjd:
+                                i.pop("data_flag")
+                                send_kafka(topic,i)
+                    else:
+                        data_fjd = get_data_time(hbase_database, table, building, meter, funcid, startdate, enddate)
+                        if data_fjd:
+                            for i in data_fjd:
+                                send_kafka(topic,i)
+        producer.close()
+        print("------------------等待下次运行时间%s------------------" % schedule_task)

+ 147 - 0
start_mysql.py

@@ -0,0 +1,147 @@
+import json
+from MyUtils.ZillionUtil import ZillionUtil
+import pymysql
+import datetime
+import argparse
+
+INSERT_SQL = "replace into %s.%s(building,meter,func_id,data_time,data_value,dt) values "
+
+# 获取项目点位
+def get_pointlist(hbase_database, hbase_table, buildingid):
+    Criteria = {
+        "building": buildingid
+    }
+    datas = zillionUtil.select(hbase_database, hbase_table, Criteria)
+    pointlist = []
+    for i in datas:
+        point = []
+        meter = i["meter"]
+        funcid = i["funcid"]
+        point.append(meter)
+        point.append(funcid)
+        pointlist.append(point)
+    return pointlist
+
+# 获取能耗数据
+def get_data_time(hbase_database, hbase_table, buildingid, meter, funcid,from_time,to_time):
+    Criteria = {
+        "building": buildingid,
+        "meter": meter,
+        "funcid": funcid,
+        "data_time":{
+		    "$gte": from_time,
+		    "$lt": to_time
+	    }
+    }
+    datas = zillionUtil.select(hbase_database, hbase_table, Criteria)
+    return datas
+
+#取hbase数据,处理成sql语句
+def hbase_energy_data(points,table):
+    sqls = []
+    for i in points:
+        meter,funcid = i[0],i[1]
+        print("%s开始查询%s至%s的数据 %s-%s"%(table,args.start_time,args.end_time,meter,funcid))
+        datas = get_data_time(hbase_database,table,building,meter,funcid,args.start_time,args.end_time)
+        for i in datas:
+            data_time = i["data_time"]
+            data_value = i["data_value"]
+            dt = data_time[0:4]+ "-" + data_time[4:6] + "-" + data_time[6:8]
+            sqlline = "(%s,%s,%s,%s,%s,'%s')" % (building, meter, funcid, data_time, data_value, dt)
+            sqls.append(sqlline)
+    return sqls
+
+
+# mysql插入数据
+def insert_mysql(sqls,my_table):
+    print("开始往mysql插入%s数据..."%my_table)
+    for i in range(0, len(sqls), 1000):
+        sqlranges = sqls[i:i + 1000]
+        sqlranges = INSERT_SQL % (my_database,my_table) + ",".join(sqlranges)
+        mysql_cur.execute(sqlranges)
+        conn.commit()
+    print("mysql数据%s插入成功,合计%s条..." % (my_table,len(sqls)))
+
+
+end_time = datetime.datetime.now().strftime("%Y%m%d") + "000000"
+start_time = (datetime.datetime.now() + datetime.timedelta(days=-1)).strftime("%Y%m%d") + "000000"
+
+parser = argparse.ArgumentParser()
+parser.add_argument("--start_time",default=start_time,help="--start_time 20220101000000")
+parser.add_argument("--end_time",default=end_time,help="--end_time 20220102000000")
+parser.add_argument("funcid",type=int,help="--funcid 10101")
+args = parser.parse_args()
+
+# print(args.start_time)
+# print(args.end_time)
+# print(args.funcid)
+
+# 读取配置文件信息
+with open("config.json", "r") as f:
+    data = json.load(f)
+    hbase_database = data["metadata"]["database"]
+    url = data["metadata"]["url"]
+    building = data["building"]["id"]
+    my_database = data["mysql"]["database"]
+    my_fjd_table = data["mysql"]["fjd_table"]
+    my_energy_table = data["mysql"]["energy_table"]
+    my_co2_table = data["mysql"]["co2_table"]
+    my_pm25_table = data["mysql"]["pm25_table"]
+    my_temperature_table = data["mysql"]["temperature_table"]
+    my_hcho_table = data["mysql"]["hcho_table"]
+    my_humidity_table = data["mysql"]["humidity_table"]
+
+    mysql = data["mysql"]
+    del mysql["fjd_table"]
+    del mysql["energy_table"]
+    del mysql["co2_table"]
+    del mysql["pm25_table"]
+    del mysql["temperature_table"]
+    del mysql["hcho_table"]
+    del mysql["humidity_table"]
+
+tables = ["fjd_0_near_15min","data_servicedata_15min"]
+# #连接hbase
+zillionUtil = ZillionUtil(url)
+#处理点位表
+pointlist = get_pointlist(hbase_database,"dy_pointlist",building)
+points = []
+for point in pointlist:
+    i = point[1]
+    if i == args.funcid:
+        points.append(point)
+#连接mysql
+conn = pymysql.connect(**mysql)
+mysql_cur = conn.cursor()
+
+#电
+if args.funcid == 10101:
+    sqls = hbase_energy_data(points, "fjd_0_near_15min")
+    insert_mysql(sqls, my_fjd_table)
+    sqls = hbase_energy_data(points, "data_servicedata_15min")
+    insert_mysql(sqls, my_energy_table)
+#CO2
+if args.funcid == 11301:
+    sqls = hbase_energy_data(points, "fjd_0_near_15min")
+    insert_mysql(sqls, my_co2_table)
+#pm2.5
+if args.funcid == 11401:
+    sqls = hbase_energy_data(points, "fjd_0_near_15min")
+    insert_mysql(sqls, my_pm25_table)
+#甲醛
+if args.funcid == 11305:
+    sqls = hbase_energy_data(points, "fjd_0_near_15min")
+    insert_mysql(sqls, my_temperature_table)
+#温度
+if args.funcid == 11101:
+    sqls = hbase_energy_data(points, "fjd_0_near_15min")
+    insert_mysql(sqls, my_hcho_table)
+#湿度
+if args.funcid == 11201:
+    sqls = hbase_energy_data(points, "fjd_0_near_15min")
+    insert_mysql(sqls, my_humidity_table)
+
+mysql_cur.close()
+conn.close()
+
+