99 lines
2.6 KiB
Python
99 lines
2.6 KiB
Python
import asyncio
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pyautogui
|
|
import requests
|
|
import yaml
|
|
|
|
from hik_uptime.custom_exception import NoWindowException
|
|
|
|
|
|
def read_config():
|
|
home_dir = Path.home()
|
|
with open(
|
|
home_dir / ".config" / "hik-uptime" / "config.yaml", "r", encoding="utf-8"
|
|
) as f:
|
|
config = yaml.safe_load(f)
|
|
return config
|
|
|
|
|
|
def get_status():
|
|
# 找到窗口并最大化
|
|
window = pyautogui.getWindowsWithTitle("智慧园区客户端")
|
|
if len(window) == 0:
|
|
raise NoWindowException("No window found")
|
|
window = window[0]
|
|
|
|
window.maximize()
|
|
window.activate()
|
|
|
|
time.sleep(2)
|
|
pyautogui.click(x=790, y=133)
|
|
time.sleep(2)
|
|
|
|
offline_devices = []
|
|
# TODO: OCR 获取已离线的设备名称数量
|
|
# 报警主机54 online: #0cb317
|
|
color_54 = pyautogui.pixel(369, 269)
|
|
color_54_hex = "#{:02x}{:02x}{:02x}".format(*color_54)
|
|
if color_54_hex != "#0cb317":
|
|
offline_devices.append("报警主机54")
|
|
|
|
# 报警主机53 online: #0db318
|
|
color_53 = pyautogui.pixel(742, 269)
|
|
color_53_hex = "#{:02x}{:02x}{:02x}".format(*color_53)
|
|
if color_53_hex != "#0db318":
|
|
offline_devices.append("报警主机53")
|
|
|
|
# 报警主机52 online: #0db318
|
|
color_52 = pyautogui.pixel(1116, 269)
|
|
color_52_hex = "#{:02x}{:02x}{:02x}".format(*color_52)
|
|
if color_52_hex != "#0db318":
|
|
offline_devices.append("报警主机52")
|
|
|
|
# 报警主机52 online: #0cb317
|
|
color_51 = pyautogui.pixel(369, 419)
|
|
color_51_hex = "#{:02x}{:02x}{:02x}".format(*color_51)
|
|
if color_51_hex != "#0cb317":
|
|
offline_devices.append("报警主机51")
|
|
|
|
# 报警主机52 online: #0db318
|
|
color_50 = pyautogui.pixel(742, 269)
|
|
color_50_hex = "#{:02x}{:02x}{:02x}".format(*color_50)
|
|
if color_50_hex != "#0db318":
|
|
offline_devices.append("报警主机50")
|
|
|
|
print(offline_devices)
|
|
|
|
for offline_device in offline_devices:
|
|
push_json = {
|
|
"eventType": "报警主机离线",
|
|
"deviceName": offline_device,
|
|
"eventLvl": 2,
|
|
"data": {"typeName": f"{offline_device}离线"},
|
|
"userIds": ["2110"],
|
|
}
|
|
resp = requests.post(
|
|
"http://192.168.20.115:8000/ipaasuat/engine_company/anycross/trigger/callback/MmRhZTE4OTRiYjVkZDQ5YWNmOGRmZDI0NjQ1MTBlODUw",
|
|
json=push_json,
|
|
)
|
|
print(resp.content.decode("utf-8"))
|
|
|
|
|
|
def capture_loop():
|
|
while True:
|
|
try:
|
|
get_status()
|
|
except Exception as e:
|
|
print(e)
|
|
time.sleep(60 * 5)
|
|
|
|
|
|
def run_app():
|
|
asyncio.run(capture_loop())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_app()
|