乐维社区：在线答疑https://forum.lwops.cn/
概述
基于 Zabbix API 和 Python 实现批量日志监控自动化，通过 Excel 导入监控配置，一键创建监控项和触发器，适用于服务器日志的大量监控场景。

一、环境准备与基础方法

1. 环境准备
依赖项 版本要求
Python >= 3.6
Zabbix Server （本文基于 6.4 验证）
pip install requests pandas openpyxl
前置条件：Zabbix Server 已启用日志监控所需的 logrt / log 相关 Agent 主动/被动检查，目标主机上 zabbix_agent2.conf 已配置日志文件访问权限。
2. 登录认证
Zabbix API 采用 JSON-RPC 2.0 协议，操作需携带认证 Token。通过 user.login 方法获取：

import requests
import json
requests.packages.urllib3.disable_warnings()

class PerseusZAPIClient:
    def __init__(self, url: str, user: str, pwd: str):
        self.url = url + "/api_jsonrpc.php"
        self.user = user
        self.pwd = pwd
        self.token = None
        self.headers = {"Content-Type": "application/json"}
        self._login()

    def _login(self):
        payload = {"jsonrpc":"2.0","method":"user.login",
                   "params":{"username":self.user,"password":self.pwd},
                   "id":1,"auth":None}
        res = requests.post(self.url, data=json.dumps(payload), headers=self.headers, verify=False)
        self.token = res.json()["result"]
        print(f"✅ 认证成功，Token: {self.token[:10]}...")

verify=False：内网环境常使用自签名证书，生产环境建议配置 CA 证书路径。
3. 主机查询与监控项检查
批量创建前需要获取主机 ID，同时使用 check_item 检查监控项是否已存在以避免重复创建：

def get_host_id(self, host):
    """根据主机名获取主机 ID"""
    payload = {"jsonrpc":"2.0","method":"host.get",
               "params":{"filter":{"host":[host]},"output":["hostid"]},
               "id":2,"auth":self.token}
    res = requests.post(self.url, data=json.dumps(payload), headers=self.headers, verify=False)
    return res.json()["result"][0]["hostid"]

def check_item(self, host_id, key):
    """检查监控项是否已存在，返回 (是否存在, item_id)"""
    payload = {"jsonrpc":"2.0","method":"item.get",
               "params":{"hostids":[host_id],"filter":{"key_":key},"output":["itemid"]},
               "id":3,"auth":self.token}
    res = requests.post(self.url, data=json.dumps(payload), headers=self.headers, verify=False)
    res_json = res.json()
    return (True, res_json["result"][0]["itemid"]) if res_json["result"] else (False, "")

二、创建日志监控项

1. 日志监控项类型
   Zabbix 原生支持以下日志类 Item 类型：
   
   Key 自动构建规则（调用脚本中根据"发现关键字次数"自动选择）：
   
2. 创建监控项实现
   参数说明：
   
   item_type / value_type 说明：
   创建时固定 type=7（Zabbix agent 主动式）、value_type=2（日志），同时自动添加 Application: 日志 标签用于分类筛选。
   

def create_item(self, host_id, items, delay, name, hostname):
    """创建单个日志监控项，已存在则抛出 ITEM_EXISTS 异常"""
    exists, item_id = self.check_item(host_id, items)
    if exists:
        raise Exception(f"ITEM_EXISTS:{item_id}")
    payload = {"jsonrpc":"2.0","method":"item.create",
               "params":{"hostid":host_id,"name":name,"key_":items,
                         "type":7,"value_type":2,"interfaceid":0,"delay":delay,
                         "tags":[{"tag":"Application","value":"日志"}]},
               "id":4,"auth":self.token}
    res = requests.post(self.url, data=json.dumps(payload), headers=self.headers, verify=False)
    res_json = res.json()
    if "error" in res_json:
        raise Exception(f"项创建失败: {res_json['error']['message']} - {res_json['error'].get('data','')}")
    item_id = res_json["result"]["itemids"][0]
    print(f"✅ 成功创建监控项 {hostname} (ID: {item_id})")
    return item_id
已存在的监控项会抛出 ITEM_EXISTS:{item_id} 异常，上层 batch_create 会捕获并复用已有 ID，保证幂等性。

3. Key 语法参考

log[file,<regexp>,<encoding>,<maxlines>,<mode>,<output>,<maxdelay>,<options>]
logrt[file_regexp,<regexp>,<encoding>,<maxlines>,<mode>,<output>,<maxdelay>,<options>]

# 匹配模式 - 关键字告警（skip 跳过旧数据）
logrt["/var/log/nginx/error.log.*", "5\d{2}",, skip]

# 计数模式 - 次数超阈值告警
logrt.count["/var/log/nginx/error.log.*","5\d{2}"]

三、创建日志触发器

1. 触发器优先级
   脚本使用中文等级映射：
   
   severity_reverse = {“信息”: 1, “警告”: 2, “次要”: 3, “严重”: 4, “紧急”: 5}
2. 创建触发器实现
   参数说明：
   

def create_trigger(self, host, items, trigger_name, trigger_level, trigger_count, keywords):
    """创建单个日志触发器，自动构建表达式"""
    severity_reverse = {"信息": 1, "警告": 2, "次要": 3, "严重": 4, "紧急": 5}
    priority = severity_reverse.get(trigger_level, 4)
    expr = f'find(/{host}/{items},,"regexp","{keywords}")=1'
    if trigger_count != "":
        expr = f'last(/{host}/{items})>{trigger_count}'
    payload = {"jsonrpc":"2.0","method":"trigger.create",
               "params":{"description":trigger_name,"expression":expr,
                         "manual_close":1,"priority":priority,
                         "tags":[{"tag":"Application","value":"日志"}]},
               "id":5,"auth":self.token}
    res = requests.post(self.url, data=json.dumps(payload), headers=self.headers, verify=False)
    res_json = res.json()
    if "error" in res_json:
        raise Exception(f"API错误: {res_json['error']['message']} - {res_json['error'].get('data','')}")
    trigger_id = res_json["result"]["triggerids"][0]
    print(f"✅ 成功创建触发器 {trigger_name} (ID: {trigger_id})")
    return trigger_id

manual_close: 1 允许手动关闭告警，Application: 日志 标签与监控项保持一致。
3. 触发器表达式参考
#匹配到关键字即告警（find 模式 - 脚本默认）
find(/Web-Server-01/logrt[“/var/log/nginx/error.log.*”,“500”],“regexp”,“500”)=1

#值超过阈值触发（last 模式 - trigger_count 非空时）
last(/App-Server-01/logrt.count[“/var/log/app/error.log”,“CRITICAL”])>0

四、完整代码

1. 核心类 items_creants.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import requests
import json
requests.packages.urllib3.disable_warnings()

class PerseusZAPIClient:
    def __init__(self, url: str, user: str, pwd: str):
        self.url = url + "/api_jsonrpc.php"
        self.user = user
        self.pwd = pwd
        self.token = None
        self.headers = {"Content-Type": "application/json"}
        self._login()

    def _login(self):
        payload = {"jsonrpc":"2.0","method":"user.login",
                   "params":{"username":self.user,"password":self.pwd},
                   "id":1,"auth":None}
        res = requests.post(self.url, data=json.dumps(payload), headers=self.headers, verify=False)
        self.token = res.json()["result"]
        print(f"✅ 认证成功，Token: {self.token[:10]}...")

    def get_host_id(self, host):
        payload = {"jsonrpc":"2.0","method":"host.get",
                   "params":{"filter":{"host":[host]},"output":["hostid"]},
                   "id":2,"auth":self.token}
        res = requests.post(self.url, data=json.dumps(payload), headers=self.headers, verify=False)
        return res.json()["result"][0]["hostid"]

    def check_item(self, host_id, key):
        payload = {"jsonrpc":"2.0","method":"item.get",
                   "params":{"hostids":[host_id],"filter":{"key_":key},"output":["itemid"]},
                   "id":3,"auth":self.token}
        res = requests.post(self.url, data=json.dumps(payload), headers=self.headers, verify=False)
        res_json = res.json()
        return (True, res_json["result"][0]["itemid"]) if res_json["result"] else (False, "")

    def create_item(self, host_id, items, delay, name, hostname):
        exists, item_id = self.check_item(host_id, items)
        if exists: raise Exception(f"ITEM_EXISTS:{item_id}")
        payload = {"jsonrpc":"2.0","method":"item.create",
                   "params":{"hostid":host_id,"name":name,"key_":items,
                             "type":7,"value_type":2,"interfaceid":0,"delay":delay,
                             "tags":[{"tag":"Application","value":"日志"}]},
                   "id":4,"auth":self.token}
        res = requests.post(self.url, data=json.dumps(payload), headers=self.headers, verify=False)
        res_json = res.json()
        if "error" in res_json: raise Exception(f"项创建失败: {res_json['error']['message']} - {res_json['error'].get('data','')}")
        item_id = res_json["result"]["itemids"][0]
        print(f"✅ 成功创建监控项 {hostname} (ID: {item_id})")
        return item_id

    def create_trigger(self, host, items, trigger_name, trigger_level, trigger_count, keywords):
        severity_reverse = {"信息": 1, "警告": 2, "次要": 3, "严重": 4, "紧急": 5}
        priority = severity_reverse.get(trigger_level, 4)
        expr = f'find(/{host}/{items},,"regexp","{keywords}")=1'
        if trigger_count != "":
            expr = f'last(/{host}/{items})>{trigger_count}'
        payload = {"jsonrpc":"2.0","method":"trigger.create",
                   "params":{"description":trigger_name,"expression":expr,
                             "manual_close":1,"priority":priority,
                             "tags":[{"tag":"Application","value":"日志"}]},
                   "id":5,"auth":self.token}
        res = requests.post(self.url, data=json.dumps(payload), headers=self.headers, verify=False)
        res_json = res.json()
        if "error" in res_json: raise Exception(f"API错误: {res_json['error']['message']} - {res_json['error'].get('data','')}")
        trigger_id = res_json["result"]["triggerids"][0]
        print(f"✅ 成功创建触发器 {trigger_name} (ID: {trigger_id})")
        return trigger_id

    def batch_create(self, configs):
        """批量创建监控项和触发器，返回分类统计结果"""
        res = {"新建项":[],"已创建":[],"成功触发器":[],"失败触发器":[],"整体失败":[]}
        for config in configs:
            hostname, name, regex_path, keywords, trigger_name, trigger_count, trigger_level, trigger_time, trigger_decs = config['row']
            host = hostname
            line_num = config["line_num"]
            items = config["log_keys"]
            if not host or not hostname:
                res["整体失败"].append(f"第{line_num}行主机配置空: {host}")
                continue
            try:
                host_id = self.get_host_id(host)
                print(f"\n🔍 主机 {host} (ID: {host_id})")
                try:
                    item_id = self.create_item(host_id, items, trigger_time, name, hostname)
                    res["新建项"].append(f"第{line_num}行{host} - {hostname} (ID: {item_id})")
                except Exception as e:
                    if str(e).startswith("ITEM_EXISTS:"):
                        item_id = str(e).split(":")[1]
                        res["已创建"].append(f"第{line_num}行{host} - {hostname} (ID: {item_id})")
                        print(f"⏭️ 监控项 {hostname} 复用ID: {item_id}")
                    else:
                        err = f"第{line_num}行{host} - {hostname}: {str(e)}"
                        res["整体失败"].append(err)
                        print(f"❌ {err}")
                        continue  # 跳过创建触发器
                try:
                    trigger_id = self.create_trigger(host, items, trigger_name, trigger_level, trigger_count, keywords)
                    res["成功触发器"].append(f"第{line_num}行{host} - {hostname} 触发器(ID: {trigger_id})")
                except Exception as e:
                    err = f"第{line_num}行{host} - {hostname} 触发器: {str(e)}"
                    if "already exists" in err:
                        err = f"第{line_num}行{host} - {hostname} 触发器: 重复创建触发器"
                    res["失败触发器"].append(err)
                    print(f"❌ {err}")
            except Exception as e:
                err = f"第{line_num}行{host} - {hostname}: {str(e)}"
                res["整体失败"].append(err)
                print(f"❌ {err}")
        return res

五、使用 Excel 批量导入

1. Excel 格式如图
2. 
   格式说明行（第2行）为字段描述，脚本从第3行开始读取数据；发现关键字次数为空时使用 find() 匹配模式，非空时使用 logrt.count 计数模式。
3. 调用脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import math
import pandas as pd
from items_creants import PerseusZAPIClient

# ── 配置 ──────────────────────────────────────────────
excel_file_path = "日志监控批量导入格式.xlsx"
server_url = "http://xx.xx.xx.xx"
username = "zabbix"
password = "zabbix"
trigger_times = "1m"  # 默认采集间隔

# ── 读取 Excel ────────────────────────────────────────
df = pd.read_excel(excel_file_path, engine="openpyxl")
row_list = df.values.tolist()

# ── 构建配置列表 ──────────────────────────────────────
NEW_BATCH_CONFIG = []
for line_num, row in enumerate(row_list[1:], start=3):  # 跳过格式说明行
    row[7] = trigger_times  # 统一覆盖采集间隔
    # 处理 NaN 值
    cleaned_row = ["" if (isinstance(v, float) and math.isnan(v)) else v for v in row]
    hostname, name, regex_path, keywords, trigger_name, trigger_count, \
        trigger_level, trigger_time, trigger_decs = cleaned_row

    # 根据是否配置次数自动选择 Key 类型
    if trigger_count != "":
        key_ = f'logrt.count["{regex_path}","{keywords}"]'
    else:
        key_ = f'logrt["{regex_path}", "{keywords}",, skip]'

    NEW_BATCH_CONFIG.append({
        "host_name": hostname,
        "log_keys": key_,
        "row": cleaned_row,
        "line_num": line_num,
    })

# ── 执行批量创建 ─────────────────────────────────────
CLIENT = PerseusZAPIClient(server_url, username, password)
print("\n🚀 开始批量创建...")
RESULT = CLIENT.batch_create(NEW_BATCH_CONFIG)

# ── 结果汇总 ─────────────────────────────────────────
print("\n" + "=" * 80)
print("📊 结果汇总:")
print(f"✅ 新建监控项: {len(RESULT['新建项'])} 个")
for i in RESULT["新建项"]: print(f"   - {i}")
print(f"⏭️ 复用监控项: {len(RESULT['已创建'])} 个")
for i in RESULT["已创建"]: print(f"   - {i}")
print(f"✅ 成功触发器: {len(RESULT['成功触发器'])} 个")
for i in RESULT["成功触发器"]: print(f"   - {i}")
print(f"❌ 失败触发器: {len(RESULT['失败触发器'])} 个")
for i in RESULT["失败触发器"]: print(f"   - {i}")
print(f"💥 整体失败: {len(RESULT['整体失败'])} 个")
for i in RESULT["整体失败"]: print(f"   - {i}")

3.生成结果