5.1 ホスト管理 ​

#!/usr/bin/env python3
import json
import requests

# Zabbix APIエンドポイント
ZABBIX_URL = "https://zabbix.example.com/api_jsonrpc.php"
USERNAME = "admin"
PASSWORD = "password"

def zabbix_auth():
    """Zabbix認証"""
    data = {
        "jsonrpc": "2.0",
        "method": "user.login",
        "params": {
            "user": USERNAME,
            "password": PASSWORD
        },
        "id": 1
    }
    response = requests.post(ZABBIX_URL, json=data)
    return response.json()["result"]

def create_host_bulk(auth_token, hosts_data):
    """ホスト一括作成"""
    hosts = []
    for host_data in hosts_data:
        host = {
            "host": host_data["hostname"],
            "name": host_data["display_name"],
            "groups": [{"groupid": host_data["group_id"]}],
            "interfaces": [{
                "type": 1,  # Zabbix agent
                "main": 1,
                "useip": 1,
                "ip": host_data["ip"],
                "port": "10050"
            }],
            "templates": [{"templateid": host_data["template_id"]}],
            "macros": host_data.get("macros", [])
        }
        hosts.append(host)
    
    data = {
        "jsonrpc": "2.0",
        "method": "host.create",
        "params": hosts,
        "auth": auth_token,
        "id": 1
    }
    
    response = requests.post(ZABBIX_URL, json=data)
    return response.json()

# 使用例
hosts_to_create = [
    {
        "hostname": "web-01",
        "display_name": "Webサーバー01",
        "ip": "192.168.1.100",
        "group_id": "4",
        "template_id": "10001",
        "macros": [
            {"macro": "{$SERVICE_NAME}", "value": "nginx"},
            {"macro": "{$LOG_PATH}", "value": "/var/log/nginx"}
        ]
    },
    {
        "hostname": "web-02",
        "display_name": "Webサーバー02",
        "ip": "192.168.1.101",
        "group_id": "4",
        "template_id": "10001",
        "macros": [
            {"macro": "{$SERVICE_NAME}", "value": "nginx"},
            {"macro": "{$LOG_PATH}", "value": "/var/log/nginx"}
        ]
    }
]

auth_token = zabbix_auth()
result = create_host_bulk(auth_token, hosts_to_create)
print(json.dumps(result, indent=2))

#!/usr/bin/env python3
import csv
import json
import requests
from typing import List, Dict

def import_hosts_from_csv(csv_file: str, auth_token: str):
    """CSVファイルからホストをインポート"""
    hosts_data = []
    
    with open(csv_file, 'r', encoding='utf-8') as file:
        reader = csv.DictReader(file)
        for row in reader:
            host_data = {
                "host": row["hostname"],
                "name": row["display_name"],
                "groups": [{"groupid": get_group_id(row["group_name"], auth_token)}],
                "interfaces": [{
                    "type": 1,
                    "main": 1,
                    "useip": 1,
                    "ip": row["ip_address"],
                    "port": "10050"
                }],
                "templates": [{"templateid": get_template_id(row["template_name"], auth_token)}],
                "macros": [
                    {"macro": "{$SERVICE_TYPE}", "value": row["service_type"]},
                    {"macro": "{$ENVIRONMENT}", "value": row["environment"]}
                ]
            }
            hosts_data.append(host_data)
    
    return create_hosts_batch(hosts_data, auth_token)

def get_group_id(group_name: str, auth_token: str) -> str:
    """グループ名からグループIDを取得"""
    data = {
        "jsonrpc": "2.0",
        "method": "hostgroup.get",
        "params": {"filter": {"name": [group_name]}},
        "auth": auth_token,
        "id": 1
    }
    response = requests.post(ZABBIX_URL, json=data)
    result = response.json()["result"]
    return result[0]["groupid"] if result else None

def get_template_id(template_name: str, auth_token: str) -> str:
    """テンプレート名からテンプレートIDを取得"""
    data = {
        "jsonrpc": "2.0",
        "method": "template.get",
        "params": {"filter": {"host": [template_name]}},
        "auth": auth_token,
        "id": 1
    }
    response = requests.post(ZABBIX_URL, json=data)
    result = response.json()["result"]
    return result[0]["templateid"] if result else None

def create_emergency_maintenance(hostids: List[str], duration_hours: int = 4):
    """緊急メンテナンス作成"""
    from datetime import datetime, timedelta
    
    start_time = int(datetime.now().timestamp())
    end_time = int((datetime.now() + timedelta(hours=duration_hours)).timestamp())
    
    data = {
        "jsonrpc": "2.0",
        "method": "maintenance.create",
        "params": {
            "name": f"Emergency Maintenance {datetime.now().strftime('%Y%m%d-%H%M')}",
            "active_since": start_time,
            "active_till": end_time,
            "hostids": hostids,
            "maintenance_type": 0,  # データ収集あり
            "description": "緊急メンテナンス作業"
        },
        "auth": auth_token,
        "id": 1
    }
    
    response = requests.post(ZABBIX_URL, json=data)
    return response.json()