main.py 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687
  1. import datetime
  2. import json
  3. import os
  4. from MyUtils.ZillionUtil import ZillionUtil
  5. from MyUtils.MysqlUtils import MysqlUtils
  6. from dateutil.relativedelta import relativedelta
  7. 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 "
  8. DELETE_SQL = "DELETE FROM `energy_week_day` WHERE `project_id` = '%s' AND `date` >= '%s' AND `date` < '%s' "
  9. def datetime_now():
  10. datetime_now = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
  11. return datetime_now
  12. # 获取能耗数据
  13. def get_data_time(hbase_database, hbase_table, building, from_time, to_time):
  14. Criteria = {
  15. "building": building,
  16. "energyModelSign": building + "EM01",
  17. "energyModelNodeSign": "EI1001",
  18. "data_time": {
  19. "$gte": from_time,
  20. "$lt": to_time
  21. }
  22. }
  23. datas = zillionUtil.select(hbase_database, hbase_table, Criteria)
  24. return datas
  25. # mysql插入数据
  26. def insert_mysql(sqls,building,my_table):
  27. print("%s,开始往mysql插入%s的%s数据..."%(datetime_now(),building,my_table))
  28. for i in range(0, len(sqls), 1000):
  29. sqlranges = sqls[i:i + 1000]
  30. sqlranges = INSERT_SQL % (my_database,my_table) + ",".join(sqlranges)
  31. MysqlUtil.update(sqlranges)
  32. print("%s,mysql数据%s,项目%s插入成功,合计%s条..." % (datetime_now(),my_table,building,len(sqls)))
  33. # with open("config.json", "r") as f:
  34. # data = json.load(f)
  35. # hbase_database = data["metadata"]["database"]
  36. # url = data["metadata"]["url"]
  37. # building = data["building"]
  38. # mysql = data["mysql"]
  39. # my_database = mysql["database"]
  40. hbase_database = os.getenv("hbase_database")
  41. url = os.getenv("url")
  42. building = os.getenv("building")
  43. mysql = {
  44. "host": os.getenv("host"),
  45. "port": os.getenv("port"),
  46. "user": os.getenv("user"),
  47. "passwd": os.getenv("passwd"),
  48. "database": os.getenv("database")
  49. }
  50. now_month = datetime.date.today().strftime("%Y%m")
  51. last_month = (datetime.date.today() - relativedelta(months=1)).strftime("%Y%m")
  52. # #连接hbase
  53. zillionUtil = ZillionUtil(url)
  54. # #连接hbase
  55. MysqlUtil = MysqlUtils(**mysql)
  56. datas = get_data_time(hbase_database, "data_energydata_1d", building, last_month, now_month)
  57. if datas:
  58. project_id = "Pj" + building
  59. # 删除上月数据
  60. print("%s,开始删除%s的数据..."%(datetime_now(),last_month))
  61. delete_sql = DELETE_SQL% (project_id,last_month,now_month)
  62. MysqlUtil.update(delete_sql)
  63. sqls = []
  64. for i in datas:
  65. dt = i["data_time"][0:8]
  66. data_value = i["data_value"]
  67. sql = "('%s','%s','0','0','0','0','%s','%s','%s')" % (project_id, dt, data_value, datetime_now(),datetime_now())
  68. sqls.append(sql)
  69. insert_mysql(sqls,building,"energy_week_day")
  70. else:
  71. print("%s,没有查询到数据...")