机房作为企业 IT 基础设施的核心，其稳定运行至关重要。动环系统（动力环境监控系统）虽能实时监测机房的温度、湿度、电力供应、空调运行等关键指标，但传统监控模式存在独立系统需单独登录、告警分散、历史数据难以分析等痛点。本文将为您带来一套高效的解决方案 —— 通过 API 将动环系统接入 Zabbix 监控平台，即使是 Zabbix 新手，也能在 1 小时内完成基础监控配置，实现机房环境 7×24 小时智能监控，消除监控孤岛，在一个控制台即可查看所有基础设施状态，让机房监控更智能、更高效。
动环系统（动力环境监控系统）负责监测机房的：温度/湿度 、电力供应、空调运行、安防状态、漏水检测等
传统监控痛点：
脚本功能概述
此Python脚本实现了从动环系统获取设备数据并自动发送到Zabbix监控平台的完整流程。主要功能包括：
环境准备
1.安装Python 3.x
2.安装必要的Python库：
pip install requests
3.确保zabbix_sender工具已安装并可用
（1）下面是一个示例动环监控的整个流程图
请求登录获取会话的接口时，会返回一个PHPSESSID（分别对应着动环系统里不通的设备名），在根据获取到的ID去请求数据，每个动环系统的请求方式都不一定是一样的，可以具体根据对应产商提供的API接口说明文档去进行适配。下面是我获取的ID并做了映射，会根据清晰明了。
（2）通过动环提供的API接口获取到的ID读取的数据，下面是一个示例
（3）把获取的到数据进行调整最终输出为一个临时文本文件，然后在通过zabbix_sender把数据根据对应给予的键值推送到平台对应的zabbix采集器监控项中，推送的键值要和监控项的创建的键值一致，具体可以参考网上的zabbix采集器监控项运用教程。
1. 执行脚本
python3 donghuan_to_zabbix.py <username> <password> <dev_ip> <dev_name> getnodes
参数说明：
<username>: 动环系统登录用户名
<password>: 动环系统登录密码
<dev_ip>: 动环系统服务器IP地址
<dev_name>: Zabbix中配置的主机名称
2. 示例执行命令
python3 donghuan_to_zabbix.py admin password123 192.168.1.100 Donghuan_Main getnodes
import requests
import subprocess
import json
import os
import sys
# 配置
IDS = [1594, 1598, 1595, 1586, 1598, 1593, 1596, 1591, 1599, 1589]
KEY_PREFIX = "device.data."
TEMP_DIR = "/tmp/donghuan"
AUTO_DISCOVERY_FILE = os.path.join(TEMP_DIR, "zabbix_auto_discovery.json")
# 设备ID与名称的映射
DEVICE_NAMES = {
1586: "1号精密空调",
1589: "2号配电柜",
1591: "2号精密空调",
1593: "3号精密空调",
1594: "3号配电柜",
1595: "市电输入",
1596: "环境监控",
1598: "UPS主机",
1599: "1号配电柜"
}
def login_and_get_session(dev_ip, username, password):
"""登录服务器并返回会话对象"""
login_url = f"http://{dev_ip}/action/ajax/login_ajax.php"
session = requests.Session()
response = session.get(login_url, params={"uname": username, "pwd": password, "checked": "1"})
if response.status_code == 200:
cookies = session.cookies.get_dict()
if "PHPSESSID" in cookies:
print("登录成功，已获取PHPSESSID。")
return session
print("登录失败。")
exit(1)
def fetch_device_data(session, dev_ip, device_id):
"""获取特定设备ID的数据"""
data_url = f"http://{dev_ip}/action/devicePage/ajaxGetData.php?id={device_id}"
response = session.get(data_url)
if response.status_code == 200:
data = response.json().get('datas', {})
if data:
return {
"updatetime": data.get('rd_updatetime', ""),
"innertemp": data.get('rd_air_innertemp', ""),
"innerhum": data.get('rd_air_innerhum', ""),
"state": data.get('rd_air_state', ""),
"temp_alarm": data.get('rd_air_temphalarm', "0"), # 温度报警
"hum_alarm": data.get('rd_air_humhalarm', "0"), # 湿度报警
"temp_setpoint": data.get('rd_air_tempsetpoint', ""),
"hum_setpoint": data.get('rd_air_humsetpoint', "")
}
print(f"获取ID为{device_id}的设备数据失败")
return None
def write_device_data_to_file(device_name, device_data):
"""将设备数据写入文件，以设备名称命名文件"""
if not os.path.exists(TEMP_DIR):
os.makedirs(TEMP_DIR)
file_path = os.path.join(TEMP_DIR, f"{device_name}.txt") # 使用设备名称命名文件
with open(file_path, 'w') as f:
json.dump(device_data, f, indent=4, ensure_ascii=False)
return file_path
def process_and_send_data(file_path, dev_name, device_id):
"""处理数据并发送到Zabbix"""
with open(file_path, 'r') as f:
device_data = json.load(f)
zabbix_commands = [
f"zabbix_sender -z {dev_name} -s {dev_name} -k {KEY_PREFIX}updatetime[{device_id}] -o {device_data['updatetime']}",
f"zabbix_sender -z {dev_name} -s {dev_name} -k {KEY_PREFIX}innertemp[{device_id}] -o {device_data['innertemp']}",
f"zabbix_sender -z {dev_name} -s {dev_name} -k {KEY_PREFIX}innerhum[{device_id}] -o {device_data['innerhum']}",
f"zabbix_sender -z {dev_name} -s {dev_name} -k {KEY_PREFIX}state[{device_id}] -o {device_data['state']}",
f"zabbix_sender -z {dev_name} -s {dev_name} -k {KEY_PREFIX}temp_alarm[{device_id}] -o {device_data['temp_alarm']}", # 温度报警
f"zabbix_sender -z {dev_name} -s {dev_name} -k {KEY_PREFIX}hum_alarm[{device_id}] -o {device_data['hum_alarm']}", # 湿度报警
f"zabbix_sender -z {dev_name} -s {dev_name} -k {KEY_PREFIX}temp_setpoint[{device_id}] -o {device_data['temp_setpoint']}", # 温度设定值
f"zabbix_sender -z {dev_name} -s {dev_name} -k {KEY_PREFIX}hum_setpoint[{device_id}] -o {device_data['hum_setpoint']}" # 湿度设定值
]
for command in zabbix_commands:
subprocess.run(command, shell=True, check=True)
def main():
if len(sys.argv) < 5:
print("用法: script.py <username> <password> <dev_ip> <dev_name> <action>")
sys.exit(1)
username = sys.argv[1]
password = sys.argv[2]
dev_ip = sys.argv[3]
dev_name = sys.argv[4]
action = sys.argv[5]
session = login_and_get_session(dev_ip, username, password)
if action == 'getnodes':
for device_id in IDS:
# 获取设备名称
device_name = DEVICE_NAMES.get(device_id)
if not device_name:
print(f"设备ID {device_id} 无对应名称，跳过此设备。")
continue
device_data = fetch_device_data(session, dev_ip, device_id)
if not device_data:
continue
# 将设备数据写入文件，以设备名称命名
file_path = write_device_data_to_file(device_name, device_data)
process_and_send_data(file_path, dev_name, device_id)
if __name__ == "__main__":
main()
