API を活用した Zabbix のホスト管理

host,groups,templates,interfaces
test host1,"[
{""groupid"": ""4""}
]","[{""templateid"": ""10001""}]","[{""type"": 1,""main"": 1, ""useip"": 1,""ip"": ""127.0.0.1"",""dns"": """",""port"":""10050"",""interfaceid"":52}]"
test host2,"[
{""groupid"": ""4""}
]",,"[{""type"": 1,""main"": 1, ""useip"": 1,""ip"": ""127.0.0.1"",""dns"": """",""port"":""10050"",""interfaceid"":53}]"
test host3,"[
{""groupid"": ""4""}
]",,

#!/usr/bin/env python3

import csv
import json
import argparse
import ssl
from urllib.request import urlopen, Request


class ZabbixAPI:
    def __init__(self, url, username, password):
        # リクエストを送る URL の作成
        self.api_url = f"{url}/api_jsonrpc.php"
        self.api_token = None
        # 認証のための API トークンを取得
        if username and password:
            self.login(username, password)

    def login(self, username, password):
        params = {
                "username": username,
                "password": password
        }
        # ログインし、認証トークンを取得
        self.api_token = self._request("user.login", params)

    def logout(self):
        params = "[]"
        # ログアウト
        self._request("user.logout", params)

    def _request(self, method, params):
        # リクエストの作成
        request = Request(
                url=self.api_url,
                data=json.dumps({
                    "jsonrpc":  "2.0",
                    "method":   method,
                    "params":   params,
                    "id":       1
                }).encode(),
                headers={"Content-Type": "application/json-rpc"})
        # 認証トークンを取得済の場合はヘッダーに追加
        if self.api_token is not None:
            request.add_header("Authorization", "Bearer " + self.api_token)
        # HTTPS 向けに設定
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        # リクエストを送信
        response = json.loads(urlopen(request, context=ctx).read().decode("utf-8"))
        if "error" in response:
            self.logout
            raise Exception(f"Zabbix API error: {response['error']}")
        return response["result"]

    def create_host(self, file):
        # host.create のオブジェクト型のパラメータ
        object_parameter = ["groups",
                            "interfaces",
                            "tags",
                            "templates",
                            "macros",
                            "inventory"]
        # csv ファイルを取得し、辞書型で格納
        with open(file) as f:
            reader = csv.DictReader(f)
            hosts = [row for row in reader]
        for host in hosts:
            for param in list(host.keys()):
                # 該当ホストに存在しないパラメータは削除
                if not host[param]:
                    del host[param]
                # オブジェクト型のパラメータは辞書型で格納
                elif param in object_parameter:
                    host[param] = json.loads(host[param])
            print(self._request("host.create", host))


def get_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("--url",      required=True,  help="Zabbix URL (e.g. https://127.0.0.1/zabbix)")
    parser.add_argument("--username", required=True,  help="Zabbix username")
    parser.add_argument("--password", required=True,  help="Zabbix password")
    parser.add_argument("--file",     required=True,  help="CSV file path")
    return parser.parse_args()


args = get_args()
api = ZabbixAPI(args.url, args.username, args.password)
api.create_host(args.file)
api.logout

def _request(self, method, params):
        # リクエストの作成
        request = Request(
                url=self.api_url,
                data=json.dumps({
                    "jsonrpc":  "2.0",
                    "method":   method,
                    "params":   params,
                    "id":       1
                }).encode(),
                headers={"Content-Type": "application/json-rpc"})
        # 認証トークンを取得済の場合はヘッダーに追加
        if self.api_token is not None:
            request.add_header("Authorization", "Bearer " + self.api_token)
        # HTTPS 向けに設定
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        # リクエストを送信
        response = json.loads(urlopen(request, context=ctx).read().decode("utf-8"))
        if "error" in response:
            self.logout
            raise Exception(f"Zabbix API error: {response['error']}")
        return response["result"]

def create_host(self, file):
        # host.create のオブジェクト型のパラメーター
        object_parameter = ["groups",
                            "interfaces",
                            "tags",
                            "templates",
                            "macros",
                            "inventory"]
        # csv ファイルを取得し、辞書型で格納
        with open(file) as f:
            reader = csv.DictReader(f)
            hosts = [row for row in reader]
        for host in hosts:
            for param in list(host.keys()):
                # 該当ホストに存在しないパラメーターは削除
                if not host[param]:
                    del host[param]
                # オブジェクト型のパラメーターは辞書型で格納
                elif param in object_parameter:
                    host[param] = json.loads(host[param])
            print(self._request("host.create", host))