手撕代码：如何用Python通过SNMP协议批量抓取机房温湿度记录仪数据

pip install pysnmp

pip install pysnmp==4.4.12

snmpwalk -v2c -c public 192.168.1.100 .1.3.6.1.4.1.xxx

from pysnmp.hlapi import *
import time

def snmp_get(ip: str, community: str, oid: str, timeout=1, retries=1):
    """
    SNMP v2c GET 单OID读取
    :param ip: 设备IP
    :param community: 读共同体，一般public
    :param oid: 对象标识符字符串，如 ".1.3.6.1.4.1.xx.1.1.1"
    :param timeout: 超时时间 单位s
    :param retries: 重试次数
    :return: 成功返回数值，失败返回None
    """
    errorIndication, errorStatus, errorIndex, varBinds = next(
        getCmd(SnmpEngine(),
               CommunityData(community, mpModel=1),  # mpModel=1 → SNMPv2c
               UdpTransportTarget((ip, 161), timeout=timeout, retries=retries),
               ContextData(),
               ObjectType(ObjectIdentity(oid)))
    )

    if errorIndication:
        # 超时、不可达、共同体错误都会走到这里
        print(f"[SNMP ERROR] {ip} | {errorIndication}")
        return None
    elif errorStatus:
        print(f"[SNMP ERROR] {ip} | {errorStatus.prettyPrint()} @ {errorIndex}")
        return None
    else:
        for varBind in varBinds:
            raw_val = varBind[1].prettyPrint()
            return raw_val

if __name__ == "__main__":
    temp_oid = ".1.3.6.1.4.1.2021.10.1.1"
    hum_oid  = ".1.3.6.1.4.1.2021.10.1.2"
    ip_addr  = "192.168.1.100"
    com = "public"
    t_val = snmp_get(ip_addr, com, temp_oid)
    h_val = snmp_get(ip_addr, com, hum_oid)
    print(f"IP:{ip_addr} 温度:{t_val} 湿度:{h_val}")

from pysnmp.hlapi import *
from concurrent.futures import ThreadPoolExecutor, as_completed
import csv
import time
from datetime import datetime

# ====================== 【配置区，根据现场修改】 ======================
COMMUNITY = "public"
TEMP_OID  = ".1.3.6.1.4.1.37244.1.1.1"   # 替换为你设备温度OID
HUM_OID   = ".1.3.6.1.4.1.37244.1.1.2"   # 替换为你设备湿度OID
SCALE = 10.0   # 很多传感器数值放大10倍，除以10得到真实值；如果是直接浮点返回设置为1
TIMEOUT = 1.2
MAX_WORKERS = 15  # 并发线程数量
# 批量设备IP清单，在这里录入所有机房温湿度记录仪IP列表
DEVICE_LIST = [
    {"ip":"192.168.1.100","name":"机房A-01"},
    {"ip":"192.168.1.101","name":"机房A-02"},
    {"ip":"192.168.1.102","name":"配电室-01"},
    {"ip":"192.168.1.103","name":"库房-01"},
]
LOG_FILE = "snmp_temp_hum_log.csv"
# ====================================================================

def snmp_get_one_oid(ip, community, oid):
    errorIndication, errorStatus, errorIndex, varBinds = next(
        getCmd(SnmpEngine(),
               CommunityData(community, mpModel=1),
               UdpTransportTarget((ip, 161), timeout=TIMEOUT, retries=1),
               ContextData(),
               ObjectType(ObjectIdentity(oid)))
    )
    if errorIndication or errorStatus:
        return None
    for varBind in varBinds:
        return varBind[1].prettyPrint()

def task_read_device(dev):
    """单个设备采集任务：读取温度、湿度，换算、返回字典"""
    ip = dev["ip"]
    dev_name = dev["name"]
    t_raw = snmp_get_one_oid(ip, COMMUNITY, TEMP_OID)
    h_raw = snmp_get_one_oid(ip, COMMUNITY, HUM_OID)
    ret = {
        "name": dev_name,
        "ip": ip,
        "temp": None,
        "hum": None,
        "ts": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    }
    try:
        if t_raw is not None:
            ret["temp"] = round(float(t_raw)/SCALE, 2)
        if h_raw is not None:
            ret["hum"] = round(float(h_raw)/SCALE, 2)
    except Exception as e:
        print(f"数据解析异常 {ip}: {str(e)}")
    return ret

def batch_collect():
    """批量并发采集入口"""
    result_list = []
    print(f"==== 开始批量SNMP采集，设备总数：{len(DEVICE_LIST)} ====")
    start = time.time()
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        future_map = {executor.submit(task_read_device, dev): dev for dev in DEVICE_LIST}
        for future in as_completed(future_map):
            res = future.result()
            result_list.append(res)
    cost = time.time() - start
    print(f"==== 采集完成，耗时 {cost:.2f}s ====")
    return result_list

def save_to_csv(data_list):
    """追加写入CSV日志，用于历史归档，对接后续数据分析"""
    header = ["ts","name","ip","temp","hum"]
    with open(LOG_FILE, "a", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=header)
        # 判断文件是否为空，为空则写表头
        f.seek(0,2)
        if f.tell() == 0:
            writer.writeheader()
        for item in data_list:
            writer.writerow(item)

if __name__ == "__main__":
    while True:
        collect_result = batch_collect()
        # 控制台打印结果
        for item in collect_result:
            print(f"[{item['ts']}] {item['name']:>10} | {item['ip']:>15} | 温度:{item['temp']} ℃ | 湿度:{item['hum']} %RH")
        save_to_csv(collect_result)
        time.sleep(30)  # 采集间隔30s，按需修改