DNS管理API客户端文档

概述

本文档提供了DNS管理API的Python客户端调用代码，涵盖完整的CRUD操作以及暂停/恢复功能。

配置

API_KEY = "js@12345678"
BASE_URL = "http://192.168.0.20:8080"

API接口概览

方法	端点	描述
GET	/api/dns/record/<host>	获取DNS记录信息
POST	/api/dns/record/<host>	添加新的DNS记录
PUT	/api/dns/record/<host>	更新DNS记录IP地址
DELETE	/api/dns/record/<host>	删除DNS记录
PUT	/api/dns/record/<host>/pause	暂停DNS记录解析
PUT	/api/dns/record/<host>/resume	恢复DNS记录解析

完整的DNS管理类

import requests
import json

class DNSManager:
    """DNS管理API客户端"""

    def __init__(self, api_key, base_url):
        """
        初始化DNS管理器

        Args:
            api_key: API密钥
            base_url: API基础URL
        """
        self.api_key = api_key
        self.base_url = base_url
        self.headers = {
            'X-API-Key': api_key,
            'Content-Type': 'application/json'
        }

    # 1. 获取DNS记录
    def get_record(self, host):
        """
        获取DNS记录信息

        Args:
            host: 主机名 (如: www.onlinenet.com.cn)

        Returns:
            dict: 记录信息
        """
        endpoint = f"api/dns/record/{host}"
        url = f"{self.base_url}/{endpoint}"

        try:
            response = requests.get(url, headers=self.headers, timeout=30)
            print(f"GET 请求状态: {response.status_code}")
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            print(f"HTTP错误: {e}")
            if e.response.status_code == 404:
                print(f"记录不存在: {host}")
            return {"status": "error", "message": f"请求失败: {e}"}
        except Exception as e:
            print(f"请求失败: {e}")
            return {"status": "error", "message": f"请求失败: {e}"}

    # 2. 添加DNS记录
    def add_record(self, host, ip, record_type='A'):
        """
        添加新的DNS记录

        Args:
            host: 主机名
            ip: IP地址
            record_type: 记录类型 (A, AAAA, CNAME)

        Returns:
            dict: 操作结果
        """
        endpoint = f"api/dns/record/{host}"
        url = f"{self.base_url}/{endpoint}"

        data = {
            'ip': ip,
            'record_type': record_type.upper()
        }

        try:
            response = requests.post(url, headers=self.headers, json=data, timeout=30)
            print(f"POST 请求状态: {response.status_code}")

            result = response.json()

            if response.status_code == 201:
                print("✅ DNS记录添加成功！")
            elif response.status_code == 409:
                print("⚠️ DNS记录已存在")
            elif response.status_code == 400:
                print("❌ 请求参数错误")
            elif response.status_code == 404:
                print("❌ 域名配置不存在")

            return result
        except requests.exceptions.HTTPError as e:
            print(f"HTTP错误: {e}")
            return {"status": "error", "message": f"请求失败: {e}"}
        except Exception as e:
            print(f"请求失败: {e}")
            return {"status": "error", "message": f"请求失败: {e}"}

    # 3. 更新DNS记录
    def update_record(self, host, new_ip, record_type='A'):
        """
        更新DNS记录

        Args:
            host: 主机名
            new_ip: 新的IP地址
            record_type: 记录类型 (A, AAAA, CNAME)

        Returns:
            dict: 操作结果
        """
        endpoint = f"api/dns/record/{host}"
        url = f"{self.base_url}/{endpoint}"

        data = {
            'ip': new_ip,
            'record_type': record_type.upper()
        }

        try:
            response = requests.put(url, headers=self.headers, json=data, timeout=30)
            print(f"PUT 请求状态: {response.status_code}")

            result = response.json()

            if response.status_code == 200:
                if result.get('status') == 'success':
                    print("✅ DNS记录更新成功！")
                elif result.get('status') == 'warning':
                    print("⚠️ DNS记录已修改，但重新加载DNS服务失败")
            elif response.status_code == 404:
                print("❌ DNS记录不存在")
            elif response.status_code == 400:
                print("❌ 请求参数错误")

            return result
        except requests.exceptions.HTTPError as e:
            print(f"HTTP错误: {e}")
            return {"status": "error", "message": f"请求失败: {e}"}
        except Exception as e:
            print(f"请求失败: {e}")
            return {"status": "error", "message": f"请求失败: {e}"}

    # 4. 删除DNS记录
    def delete_record(self, host):
        """
        删除DNS记录

        Args:
            host: 主机名

        Returns:
            dict: 操作结果
        """
        endpoint = f"api/dns/record/{host}"
        url = f"{self.base_url}/{endpoint}"

        try:
            response = requests.delete(url, headers=self.headers, timeout=30)
            print(f"DELETE 请求状态: {response.status_code}")

            result = response.json()

            if response.status_code == 200:
                print("✅ DNS记录删除成功！")
            elif response.status_code == 404:
                print("❌ DNS记录不存在")

            return result
        except requests.exceptions.HTTPError as e:
            print(f"HTTP错误: {e}")
            return {"status": "error", "message": f"请求失败: {e}"}
        except Exception as e:
            print(f"请求失败: {e}")
            return {"status": "error", "message": f"请求失败: {e}"}

    # 5. 暂停DNS记录
    def pause_record(self, host):
        """
        暂停DNS记录解析

        Args:
            host: 主机名

        Returns:
            dict: 操作结果
        """
        endpoint = f"api/dns/record/{host}/pause"
        url = f"{self.base_url}/{endpoint}"

        try:
            response = requests.put(url, headers=self.headers, timeout=30)
            print(f"PUT 请求状态: {response.status_code}")

            result = response.json()

            if response.status_code == 200:
                if result.get('status') == 'success':
                    print("✅ DNS记录解析已暂停！")
            elif response.status_code == 404:
                print("❌ DNS记录不存在")
            elif response.status_code == 400:
                print("❌ 请求参数错误")

            return result
        except requests.exceptions.HTTPError as e:
            print(f"HTTP错误: {e}")
            return {"status": "error", "message": f"请求失败: {e}"}
        except Exception as e:
            print(f"请求失败: {e}")
            return {"status": "error", "message": f"请求失败: {e}"}

    # 6. 恢复DNS记录
    def resume_record(self, host):
        """
        恢复DNS记录解析

        Args:
            host: 主机名

        Returns:
            dict: 操作结果
        """
        endpoint = f"api/dns/record/{host}/resume"
        url = f"{self.base_url}/{endpoint}"

        try:
            response = requests.put(url, headers=self.headers, timeout=30)
            print(f"PUT 请求状态: {response.status_code}")

            result = response.json()

            if response.status_code == 200:
                if result.get('status') == 'success':
                    print("✅ DNS记录解析已恢复！")
            elif response.status_code == 404:
                print("❌ DNS记录不存在或未被暂停")
            elif response.status_code == 400:
                print("❌ 请求参数错误")

            return result
        except requests.exceptions.HTTPError as e:
            print(f"HTTP错误: {e}")
            return {"status": "error", "message": f"请求失败: {e}"}
        except Exception as e:
            print(f"请求失败: {e}")
            return {"status": "error", "message": f"请求失败: {e}"}

使用示例

1. 初始化管理器

# 配置参数
API_KEY = "js@12345678"
BASE_URL = "http://192.168.0.20:8080"

# 创建DNS管理器实例
dns_manager = DNSManager(API_KEY, BASE_URL)

2. 基本操作示例

# 添加A记录
add_result = dns_manager.add_record(
    host="www.onlinenet.com.cn",
    ip="192.168.1.100",
    record_type="A"
)

# 获取记录信息
get_result = dns_manager.get_record("www.onlinenet.com.cn")

# 更新记录
update_result = dns_manager.update_record(
    host="www.onlinenet.com.cn",
    new_ip="192.168.1.200",
    record_type="A"
)

# 暂停记录
pause_result = dns_manager.pause_record("www.onlinenet.com.cn")

# 恢复记录
resume_result = dns_manager.resume_record("www.onlinenet.com.cn")

# 删除记录
delete_result = dns_manager.delete_record("www.onlinenet.com.cn")

批量操作函数

批量添加记录

def batch_add_records(records_list):
    """
    批量添加DNS记录

    Args:
        records_list: 记录列表，每个元素为 (host, ip, record_type) 元组
    """
    dns = DNSManager(API_KEY, BASE_URL)
    results = []

    for i, (host, ip, record_type) in enumerate(records_list, 1):
        print(f"\n[{i}/{len(records_list)}] 处理: {host}")
        result = dns.add_record(host, ip, record_type)
        results.append({
            'host': host,
            'ip': ip,
            'record_type': record_type,
            'result': result
        })

    # 统计结果
    success_count = sum(1 for r in results if r['result'].get('status') == 'success')
    warning_count = sum(1 for r in results if r['result'].get('status') == 'warning')
    error_count = len(results) - success_count - warning_count

    print(f"\n批量操作完成:")
    print(f"✅ 成功: {success_count}")
    print(f"⚠️ 警告: {warning_count}")
    print(f"❌ 失败: {error_count}")

    return results

# 使用示例
records_to_add = [
    ("web1.onlinenet.com.cn", "192.168.1.101", "A"),
    ("web2.onlinenet.com.cn", "192.168.1.102", "A"),
    ("db.onlinenet.com.cn", "192.168.1.103", "A"),
]

batch_results = batch_add_records(records_to_add)

批量更新记录

def batch_update_records(updates_list):
    """
    批量更新DNS记录

    Args:
        updates_list: 更新列表，每个元素为 (host, new_ip, record_type) 元组
    """
    dns = DNSManager(API_KEY, BASE_URL)
    results = []

    for i, (host, new_ip, record_type) in enumerate(updates_list, 1):
        print(f"\n[{i}/{len(updates_list)}] 更新: {host}")
        result = dns.update_record(host, new_ip, record_type)
        results.append({
            'host': host,
            'new_ip': new_ip,
            'record_type': record_type,
            'result': result
        })

    # 统计结果
    success_count = sum(1 for r in results if r['result'].get('status') == 'success')
    warning_count = sum(1 for r in results if r['result'].get('status') == 'warning')
    error_count = len(results) - success_count - warning_count

    print(f"\n批量更新完成:")
    print(f"✅ 成功: {success_count}")
    print(f"⚠️ 警告: {warning_count}")
    print(f"❌ 失败: {error_count}")

    return results

批量删除记录

def batch_delete_records(hosts_list):
    """
    批量删除DNS记录

    Args:
        hosts_list: 主机名列表
    """
    dns = DNSManager(API_KEY, BASE_URL)
    results = []

    for i, host in enumerate(hosts_list, 1):
        print(f"\n[{i}/{len(hosts_list)}] 删除: {host}")
        result = dns.delete_record(host)
        results.append({
            'host': host,
            'result': result
        })

    # 统计结果
    success_count = sum(1 for r in results if r['result'].get('status') == 'success')
    error_count = len(results) - success_count

    print(f"\n批量删除完成:")
    print(f"✅ 成功: {success_count}")
    print(f"❌ 失败: {error_count}")

    return results

交互式命令行工具

def interactive_dns_manager():
    """交互式DNS管理工具"""
    dns = DNSManager(API_KEY, BASE_URL)

    while True:
        print("\n" + "="*40)
        print("DNS记录管理工具")
        print("="*40)
        print("1. 查询DNS记录")
        print("2. 添加DNS记录")
        print("3. 更新DNS记录")
        print("4. 删除DNS记录")
        print("5. 暂停DNS记录")
        print("6. 恢复DNS记录")
        print("7. 批量操作")
        print("8. 退出")

        choice = input("\n请选择操作 (1-8): ").strip()

        if choice == '1':
            host = input("请输入主机名: ").strip()
            if host:
                result = dns.get_record(host)
                if result.get('status') == 'success':
                    data = result.get('data', {})
                    print(f"\n✅ 记录信息:")
                    for key, value in data.items():
                        print(f"  {key}: {value}")
                else:
                    print(f"\n❌ 查询失败: {result.get('message')}")

        elif choice == '2':
            host = input("请输入主机名: ").strip()
            ip = input("请输入IP地址: ").strip()
            record_type = input("请输入记录类型 (A/AAAA/CNAME, 默认为A): ").strip() or "A"

            if host and ip:
                result = dns.add_record(host, ip, record_type)
                if result.get('status') == 'success':
                    print(f"\n✅ 添加成功!")
                else:
                    print(f"\n❌ 添加失败: {result.get('message')}")
            else:
                print("❌ 主机名和IP地址不能为空")

        elif choice == '3':
            host = input("请输入主机名: ").strip()
            new_ip = input("请输入新的IP地址: ").strip()
            record_type = input("请输入记录类型 (A/AAAA/CNAME, 默认为A): ").strip() or "A"

            if host and new_ip:
                result = dns.update_record(host, new_ip, record_type)
                if result.get('status') in ['success', 'warning']:
                    print(f"\n✅ 更新成功!")
                else:
                    print(f"\n❌ 更新失败: {result.get('message')}")
            else:
                print("❌ 主机名和IP地址不能为空")

        elif choice == '4':
            host = input("请输入主机名: ").strip()
            if host:
                confirm = input(f"确认删除记录 '{host}'? (y/N): ").strip().lower()
                if confirm in ['y', 'yes']:
                    result = dns.delete_record(host)
                    if result.get('status') == 'success':
                        print(f"\n✅ 删除成功!")
                    else:
                        print(f"\n❌ 删除失败: {result.get('message')}")
                else:
                    print("操作已取消")

        elif choice == '5':
            host = input("请输入主机名: ").strip()
            if host:
                result = dns.pause_record(host)
                if result.get('status') == 'success':
                    print(f"\n✅ 暂停成功!")
                else:
                    print(f"\n❌ 暂停失败: {result.get('message')}")

        elif choice == '6':
            host = input("请输入主机名: ").strip()
            if host:
                result = dns.resume_record(host)
                if result.get('status') == 'success':
                    print(f"\n✅ 恢复成功!")
                else:
                    print(f"\n❌ 恢复失败: {result.get('message')}")

        elif choice == '7':
            print("\n批量操作:")
            print("a) 批量添加")
            print("b) 批量更新")
            print("c) 批量删除")
            sub_choice = input("请选择 (a/b/c): ").strip().lower()

            if sub_choice == 'a':
                print("\n请输入记录信息，格式: 主机名,IP,记录类型")
                print("每行一条记录，输入空行结束")
                records = []
                while True:
                    line = input("> ").strip()
                    if not line:
                        break
                    parts = [p.strip() for p in line.split(',')]
                    if len(parts) >= 2:
                        host = parts[0]
                        ip = parts[1]
                        record_type = parts[2] if len(parts) > 2 else 'A'
                        records.append((host, ip, record_type))

                if records:
                    batch_add_records(records)

            elif sub_choice == 'b':
                print("\n请输入更新信息，格式: 主机名,新IP,记录类型")
                print("每行一条记录，输入空行结束")
                updates = []
                while True:
                    line = input("> ").strip()
                    if not line:
                        break
                    parts = [p.strip() for p in line.split(',')]
                    if len(parts) >= 2:
                        host = parts[0]
                        new_ip = parts[1]
                        record_type = parts[2] if len(parts) > 2 else 'A'
                        updates.append((host, new_ip, record_type))

                if updates:
                    batch_update_records(updates)

            elif sub_choice == 'c':
                print("\n请输入要删除的主机名")
                print("每行一个主机名，输入空行结束")
                hosts = []
                while True:
                    host = input("> ").strip()
                    if not host:
                        break
                    hosts.append(host)

                if hosts:
                    confirm = input(f"确认删除 {len(hosts)} 条记录? (y/N): ").strip().lower()
                    if confirm in ['y', 'yes']:
                        batch_delete_records(hosts)
                    else:
                        print("操作已取消")

            else:
                print("❌ 无效选择")

        elif choice == '8':
            print("退出工具")
            break

        else:
            print("❌ 无效选择，请重新输入")

# 运行交互式工具
# interactive_dns_manager()

错误处理

常见错误状态码

状态码	含义	可能原因
200	成功	操作成功完成
201	创建成功	记录添加成功
400	请求错误	参数格式错误、IP地址无效、域名格式错误
404	未找到	记录不存在、域名配置不存在
409	冲突	记录已存在
500	服务器错误	服务器内部错误

错误处理示例

def safe_dns_operation(operation, *args, **kwargs):
    """
    安全的DNS操作，包含详细的错误处理

    Args:
        operation: 操作函数
        *args: 位置参数
        **kwargs: 关键字参数

    Returns:
        操作结果
    """
    try:
        result = operation(*args, **kwargs)

        if result.get('status') == 'error':
            print(f"❌ 操作失败: {result.get('message')}")

            # 根据错误类型提供建议
            if "无效的域名格式" in result.get('message', ''):
                print("  建议: 检查域名格式是否正确")
            elif "DNS记录已存在" in result.get('message', ''):
                print("  建议: 使用更新操作或先删除记录")
            elif "DNS记录不存在" in result.get('message', ''):
                print("  建议: 使用添加操作创建新记录")
            elif "无效的IP地址" in result.get('message', ''):
                print("  建议: 检查IP地址格式是否正确")

        return result

    except Exception as e:
        print(f"❌ 操作异常: {e}")
        return {"status": "error", "message": f"操作异常: {e}"}

# 使用安全操作
result = safe_dns_operation(
    dns_manager.add_record,
    "test.onlinenet.com.cn",
    "192.168.1.100",
    "A"
)

实用工具函数

1. 域名验证

import re

def validate_domain_name(domain):
    """
    验证域名格式

    Args:
        domain: 域名

    Returns:
        bool: 是否有效
    """
    pattern = r'^([a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}$'
    return bool(re.match(pattern, domain))

def extract_domain_parts(hostname):
    """
    从主机名中提取各部分

    Args:
        hostname: 完整主机名

    Returns:
        dict: 域名各部分
    """
    parts = hostname.split('.')
    if len(parts) >= 3:
        return {
            'host': parts[0],
            'domain': '.'.join(parts[1:]),
            'tld': parts[-1],
            'subdomain': '.'.join(parts[:-2]) if len(parts) > 2 else None
        }
    return None

2. 记录状态检查

def check_record_status(host):
    """
    检查DNS记录状态

    Args:
        host: 主机名

    Returns:
        str: 记录状态 (active, paused, not_found)
    """
    dns = DNSManager(API_KEY, BASE_URL)

    # 首先尝试获取记录
    result = dns.get_record(host)

    if result.get('status') == 'success':
        return 'active'
    else:
        # 尝试暂停操作，如果返回404且提示"记录不存在"，则可能是被暂停或不存在
        pause_result = dns.pause_record(host)
        if pause_result.get('status') == 'error' and "记录不存在" in pause_result.get('message', ''):
            return 'not_found'
        else:
            # 如果暂停操作成功或返回其他错误，可能记录是活跃状态
            # 但为了简化，这里假设如果get失败但pause也失败(不是404)，则可能是暂停状态
            return 'paused'

3. 配置导出和导入

import json

def export_dns_config(hosts_list):
    """
    导出DNS配置

    Args:
        hosts_list: 主机名列表

    Returns:
        dict: 配置信息
    """
    dns = DNSManager(API_KEY, BASE_URL)
    config = {'records': []}

    for host in hosts_list:
        result = dns.get_record(host)
        if result.get('status') == 'success':
            config['records'].append(result.get('data', {}))

    # 保存到文件
    with open('dns_config_backup.json', 'w') as f:
        json.dump(config, f, indent=2)

    print(f"✅ 配置已导出到 dns_config_backup.json")
    return config

def import_dns_config(config_file):
    """
    导入DNS配置

    Args:
        config_file: 配置文件路径

    Returns:
        list: 导入结果
    """
    with open(config_file, 'r') as f:
        config = json.load(f)

    dns = DNSManager(API_KEY, BASE_URL)
    results = []

    for record in config.get('records', []):
        host = record.get('host')
        ip = record.get('ip')
        record_type = record.get('record_type', 'A')

        if host and ip:
            print(f"导入记录: {host}")
            result = dns.add_record(host, ip, record_type)
            results.append({'host': host, 'result': result})

    return results

部署和最佳实践

1. 环境配置

# config.py
import os

class DNSConfig:
    """DNS配置管理"""

    @staticmethod
    def load_from_env():
        """从环境变量加载配置"""
        api_key = os.getenv('DNS_API_KEY', 'js@12345678')
        base_url = os.getenv('DNS_API_URL', 'http://192.168.0.20:8080')
        timeout = int(os.getenv('DNS_API_TIMEOUT', '30'))

        return {
            'api_key': api_key,
            'base_url': base_url,
            'timeout': timeout
        }

    @staticmethod
    def load_from_file(filepath='config.json'):
        """从配置文件加载配置"""
        try:
            with open(filepath, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            print(f"配置文件 {filepath} 不存在，使用默认配置")
            return {
                'api_key': 'js@12345678',
                'base_url': 'http://192.168.0.20:8080',
                'timeout': 30
            }

2. 日志配置

import logging

def setup_logging():
    """配置日志系统"""
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler('dns_manager.log'),
            logging.StreamHandler()
        ]
    )

    return logging.getLogger(__name__)

# 使用日志
logger = setup_logging()

3. 性能优化

from functools import lru_cache

class CachedDNSManager(DNSManager):
    """带缓存的DNS管理器"""

    @lru_cache(maxsize=100)
    def get_record(self, host):
        """带缓存的获取记录"""
        return super().get_record(host)

    def clear_cache(self):
        """清空缓存"""
        self.get_record.cache_clear()

故障排除

常见问题及解决方案

1. 401未授权错误
2. 检查API密钥是否正确
3. 确认使用了正确的认证头部（X-API-Key）

4. 验证API密钥是否有访问权限

5. 404未找到错误

6. 确认主机名是否正确
7. 检查域名是否存在

8. 确认记录是否存在（对于更新、删除操作）

9. 500服务器错误

10. 检查API服务器是否运行正常
11. 查看服务器日志获取详细信息

12. 确认DNS服务是否正常运行

13. 连接超时

14. 检查网络连接
15. 确认API服务器地址和端口是否正确
16. 增加超时时间设置

调试模式

def enable_debug_mode():
    """启用调试模式"""
    import http.client as http_client
    import logging

    http_client.HTTPConnection.debuglevel = 1

    logging.basicConfig()
    logging.getLogger().setLevel(logging.DEBUG)
    requests_log = logging.getLogger("requests.packages.urllib3")
    requests_log.setLevel(logging.DEBUG)
    requests_log.propagate = True

# 在需要调试时调用
# enable_debug_mode()

总结

本文档提供了完整的DNS管理API客户端实现，包括：

1. 基础操作：增删改查、暂停恢复
2. 批量处理：批量添加、更新、删除
3. 交互工具：命令行交互界面
4. 错误处理：详细的错误处理和诊断
5. 实用工具：配置管理、状态检查等
6. 最佳实践：部署建议和性能优化

使用这些代码可以方便地管理DNS记录，确保操作的可靠性和安全性。