main.py 4.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113
  1. from MyUtils.MysqlUtils import MysqlUtils
  2. from MyUtils.ZillionUtil import ZillionUtil
  3. from MyUtils.Dingtalk import send_message_markdown
  4. import datetime
  5. import pytz
  6. import json
  7. import time
  8. SELETE_SQL = "SELECT infos FROM `object_data_equipment_infos` WHERE `category` LIKE '%ACAT%'"
  9. INSERT_SQL = "replace into sagacloud_setup.zj_check_EquipSwitch(EquipLocalID,EquipName,EquipSwitchSet,RunStatus,remark) values "
  10. DELETE_SQL = "truncate table sagacloud_setup.zj_check_equipswitch"
  11. def datetime_now():
  12. # datetime_now = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
  13. #容器时间
  14. # tz = pytz.timezone('Asia/Shanghai') # 东八区
  15. datetime_now = datetime.datetime.fromtimestamp(int(time.time()),
  16. pytz.timezone('Asia/Shanghai')).strftime('%Y-%m-%d %H:%M:%S')
  17. return datetime_now
  18. # 获取能耗数据
  19. def get_data_time(zillionUtil,hbase_database, hbase_table, building, meter,funid1,funid2,timedelta_5min):
  20. Criteria = {
  21. "building": building,
  22. "meter": meter,
  23. "funcid": {
  24. "in":[
  25. funid1,
  26. funid2
  27. ]
  28. },
  29. "receivetime": {
  30. "$gte": timedelta_5min
  31. }
  32. }
  33. datas = zillionUtil.select(hbase_database, hbase_table, Criteria)
  34. return datas
  35. with open("config.json", "r") as f:
  36. data = json.load(f)
  37. hbase_database = data["metadata"]["database"]
  38. url = data["metadata"]["url"]
  39. mysql = data["mysql"]
  40. my_database = mysql["database"]
  41. dingding = data["dingding"]
  42. timedelta_5min = (datetime.datetime.now() - datetime.timedelta(minutes=5)).strftime(f"%Y%m%d%H%M")
  43. current_month = datetime.datetime.now().strftime("%Y%m")
  44. # #连接hbase
  45. zillionUtil = ZillionUtil(url)
  46. # #连接mysql
  47. MysqlUtil = MysqlUtils(**mysql)
  48. object_data_infos = MysqlUtil.query(SELETE_SQL)
  49. sqls = []
  50. for info in object_data_infos:
  51. if "EquipSwitchSet" in json.loads(info[0]) and "RunStatus" in json.loads(info[0]):
  52. info_data = json.loads(info[0])
  53. EquipSwitchSet = info_data["EquipSwitchSet"]
  54. RunStatus = info_data["RunStatus"]
  55. EquipLocalID = info_data["EquipLocalID"]
  56. EquipName = info_data["EquipName"]
  57. if EquipSwitchSet != RunStatus:
  58. EquipSwitchSet_meter,EquipSwitchSet_funcid = str.split(EquipSwitchSet,"-")[0],str.split(EquipSwitchSet,"-")[1]
  59. RunStatus_meter,RunStatus_funcid = str.split(RunStatus,"-")[0],str.split(RunStatus,"-")[1]
  60. datas = get_data_time(zillionUtil, hbase_database, "original_month_"+current_month, "3301100002",EquipSwitchSet_meter,eval(EquipSwitchSet_funcid),eval(RunStatus_funcid),timedelta_5min)
  61. EquipSwitchSet_data_list = []
  62. RunStatus_data_list = []
  63. for i in datas:
  64. if str(i["funcid"]) == EquipSwitchSet_funcid:
  65. EquipSwitchSet_data_list.append(i["data"])
  66. else:
  67. RunStatus_data_list.append(i["data"])
  68. if EquipSwitchSet_funcid and RunStatus_data_list:
  69. diff_data = set(EquipSwitchSet_data_list) | set(RunStatus_data_list)
  70. if len(set(EquipSwitchSet_data_list)) == 1 and len(diff_data) == 2:
  71. remark = "%s %s设定值与运行状态不一致,检查时间%s"%(datetime_now(),EquipSwitchSet_meter,timedelta_5min)
  72. print(remark)
  73. sql = "('%s','%s','%s','%s','%s')" % (EquipLocalID,EquipName,EquipSwitchSet,RunStatus,remark)
  74. sqls.append(sql)
  75. else:
  76. remark = "%s %s数据有丢失,检查时间%s"%(datetime_now(),EquipSwitchSet_meter,timedelta_5min)
  77. print(remark)
  78. sql = "('%s','%s','%s','%s','%s')" % (EquipLocalID, EquipName, EquipSwitchSet, RunStatus, remark)
  79. sqls.append(sql)
  80. if sqls:
  81. print("%s 清空数据库表"%datetime_now())
  82. MysqlUtil.update(DELETE_SQL)
  83. print("%s 数据库表清空完成"%datetime_now())
  84. print("%s 开始往数据库插入数据,共计%s条"%(datetime_now(),len(sqls)))
  85. for i in range(0,len(sqls),1000):
  86. sqlranges = sqls[i:i + 1000]
  87. sqlranges = INSERT_SQL + ",".join(sqlranges)
  88. MysqlUtil.update(sqlranges)
  89. print("%s 数据库插入成功"%datetime_now())
  90. messages = []
  91. message = (" - **类型:** 空调疑似掉线 \n" + " - **数量:** <font color=#FF0000>%s</font> \n")% (len(sqls))
  92. messages.append(message)
  93. messages.append("___")
  94. messages.append("<font color=#0080FF>详情查看MySQL数据库sagacloud_setup里的zj_check_equipswitch表</font>")
  95. messages_info = "\n".join(messages)
  96. title = "## **【之江实验室】空调控制状态报警** "
  97. messages_info = title + "\n" + "___" + "\n" + messages_info
  98. send_message_markdown(title, messages_info, dingding)