saga_transfer_energy/main.py

import datetime
import json,pymysql
import os
import time
from MyUtils.ZillionUtil import ZillionUtil
from MyUtils.MysqlUtils import MysqlUtils
from MyUtils.Dingtalk import send_message
import pytz
import multiprocessing

INSERT_SQL = "replace into %s.%s(project_id,date,energy_cooling,energy_heating,energy_ac_terminal,energy_light,energy_others,create_time,update_time) values "
DELETE_SQL = "DELETE FROM `energy_week_day` WHERE `project_id` = '%s' AND `date` >= '%s' AND `date` < '%s' "
SELETE_COUNTLASTDATA_SQL = "SELECT count(*) FROM `energy_week_day` WHERE `project_id` = '%s' AND `date` >= '%s' AND `date` < '%s'"
SELETE_SUMLASTDATA_SQL = "SELECT SUM(energy_ac_terminal)+SUM(energy_heating)+SUM(energy_cooling)+SUM(energy_light)+sum(energy_others) as last_data FROM `energy_week_day` WHERE `project_id` = '%s' AND `date` >= '%s' AND `date` < '%s'"

def datetime_now():
    # datetime_now = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
    #容器时间
    # tz = pytz.timezone('Asia/Shanghai')  # 东八区
    datetime_now = datetime.datetime.fromtimestamp(int(time.time()),
                                    pytz.timezone('Asia/Shanghai')).strftime('%Y-%m-%d %H:%M:%S')
    return datetime_now



# 获取能耗数据
def get_data_time(zillionUtil,hbase_database, hbase_table, building, from_time, to_time):
    Criteria = {
        "building": building,
        "energyModelSign": building + "EM01",
        "energyModelNodeSign": {
            "in":[
                "EI10102010101001",
                "EI10102010102001",
                "EI1010202001",
                "EI1001"
            ]
        },
        "data_time": {
            "$gte": from_time,
            "$lt": to_time
        }
    }
    datas = zillionUtil.select(hbase_database, hbase_table, Criteria)
    return datas
# # mysql插入数据
# def insert_mysql(sqls,building,my_table):
#     print("%s,开始往mysql插入%s的%s数据..."%(datetime_now(),building,my_table))
#     for i in range(0, len(sqls), 1000):
#         sqlranges = sqls[i:i + 1000]
#         sqlranges = INSERT_SQL % (my_database,my_table) + ",".join(sqlranges)
#         MysqlUtil.update(sqlranges)
#     print("%s,mysql数据%s,项目%s插入成功,合计%s条..." % (datetime_now(),my_table,building,len(sqls)))


# with open("config.json", "r") as f:
#     data = json.load(f)
#     hbase_database = data["metadata"]["database"]
#     url = data["metadata"]["url"]
#     building = data["building"]
#     mysql = data["mysql"]
#     my_database = mysql["database"]
#     dingding = data["dingding"]
#     at_mobiles = data["at_mobiles"]


hbase_database = os.getenv("hbase_database")
url = os.getenv("url")
building = os.getenv("building")
dingding = os.getenv("dingding")
at_mobiles = os.getenv("at_mobiles")
mysql = {
    "host": os.getenv("host"),
    "port": os.getenv("port"),
    "user": os.getenv("user"),
    "passwd": os.getenv("passwd"),
    "database": os.getenv("database")
}
my_database = os.getenv("database")



def check_energy():
    while True:
        project_id = "Pj" + building
        # 容器里获取时间
        today = datetime.datetime.fromtimestamp(int(time.time()),
                                                pytz.timezone('Asia/Shanghai'))
        # 获取星期几0~6 0代表星期一
        weekday = today.weekday()
        if weekday == 0:
            time_now = datetime.datetime.fromtimestamp(int(time.time()),
                                                       pytz.timezone('Asia/Shanghai')).strftime('%H:%M:%S')
            if time_now == "09:00:00":
                ##查询本周能耗和上周能耗
                lastmonday = today - datetime.timedelta(days=weekday + 7)
                conn = pymysql.connect(**mysql)
                mysql_cur = conn.cursor()

                count_sql = SELETE_COUNTLASTDATA_SQL % (project_id, lastmonday.strftime("%Y%m%d"), today.strftime("%Y%m%d"))
                mysql_cur.execute(count_sql)
                lastdatacount = mysql_cur.fetchall()[0][0]
                print("上周能耗天数%s"%lastdatacount)
                if lastdatacount == 7:
                    lastdata_sql = SELETE_SUMLASTDATA_SQL % (
                    project_id, lastmonday.strftime("%Y%m%d"), today.strftime("%Y%m%d"))
                    mysql_cur.execute(lastdata_sql)
                    lastdata = mysql_cur.fetchall()[0][0]
                    last_lastmonday = lastmonday - datetime.timedelta(days=weekday + 7)
                    last_lastdata_sql = SELETE_SUMLASTDATA_SQL % (
                    project_id, last_lastmonday.strftime("%Y%m%d"), lastmonday.strftime("%Y%m%d"))
                    mysql_cur.execute(last_lastdata_sql)
                    last_lastdata = mysql_cur.fetchall()[0][0]
                    message = "【博锐周报数据】 %s,上上周能耗数据：%s,上周能耗数据：%s %s" % (datetime_now(), last_lastdata, lastdata, at_mobiles)
                    print(message)
                    send_message(message, dingding, at_mobiles)

                else:
                    message = "【博锐周报数据】 %s,上周能耗数据有缺失，请及时核对！！！%s" % (datetime_now(), at_mobiles)
                    print(message)
                    send_message(message, dingding, at_mobiles)

                mysql_cur.close()
                conn.close()
                time.sleep(2)

def transfer_energy():
    print("同步%s项目数据"%(building))
    while True:
        project_id = "Pj" + building
        time_now = datetime.datetime.fromtimestamp(int(time.time()),
                                            pytz.timezone('Asia/Shanghai')).strftime('%H:%M:%S')
        # time_now = time.strftime("%H:%M:%S", time.localtime())  # 刷新
        am_set_time = "08:00:00"
        pm_set_time = "22:00:00"
        if time_now == am_set_time or time_now ==pm_set_time:#此处设置每天定时的时间
            today = datetime.date.today().strftime("%Y%m%d")
            yesterday = (datetime.date.today()-datetime.timedelta(days=1)).strftime("%Y%m%d")

            # #连接hbase
            zillionUtil = ZillionUtil(url)
            # #连接hbase
            MysqlUtil = MysqlUtils(**mysql)
            datas = get_data_time(zillionUtil,hbase_database, "data_energydata_1d", building, yesterday, today)
            if datas:

                # 删除上月数据
                print("%s,开始删除%s的数据..."%(datetime_now(),yesterday))
                delete_sql = DELETE_SQL% (project_id,yesterday,today)
                MysqlUtil.update(delete_sql)
                # sqls = []
                energy_cooling = "0"
                energy_heating = "0"
                energy_ac_terminal = "0"
                energy_light = "0"
                sum_data_value = "0"
                for i in datas:

                    if i["energyModelNodeSign"] == "EI1001":
                        sum_data_value = i["data_value"]

                    #冷热源
                    if i["energyModelNodeSign"] == "EI10102010101001":
                        energy_cooling = i["data_value"]
                    # #热
                    # if i["energyModelNodeSign"] == "EI1010201010101001":
                    #     energy_heating = i["data_value"]
                    #空调末端
                    if i["energyModelNodeSign"] == "EI10102010102001":
                        energy_ac_terminal = i["data_value"]
                    #照明
                    if i["energyModelNodeSign"] == "EI1010202001":
                        energy_light = i["data_value"]
                energy_other = float(sum_data_value) - float(energy_cooling) - float(energy_heating) -float(energy_light) - float(energy_ac_terminal)
                sql = "('%s','%s','%s','%s','%s','%s','%s','%s','%s')" % (
                project_id, yesterday,energy_cooling,energy_heating,energy_ac_terminal,energy_light,energy_other, datetime_now(), datetime_now())
                inser_sql = INSERT_SQL % (my_database, "energy_week_day") + sql
                print("%s,开始插入数据..."%datetime_now())
                MysqlUtil.update(inser_sql)


            else:
                print("%s,没有查询到数据...")
            time.sleep(2)
            print("%s,等待下次程序执行" % (datetime_now()))




if __name__ == '__main__':
    check_energy_process = multiprocessing.Process(target=check_energy)
    transfer_energy_process = multiprocessing.Process(target=transfer_energy)
    check_energy_process.start()
    transfer_energy_process.start()