简易 dnspod.cn API 客户端

dnspod.py

import urllib.parse

import requests


class DnspodException(Exception):
    pass


class DnspodErrorResponse(DnspodException):
    def __init__(self, code, message):
        self.code = code
        self.message = message
        super().__init__(code, message)


class DnspodClient:
    def __init__(self, dnspod_id, dnspod_token):
        self.session = requests.Session()
        self.login_token = "{},{}".format(dnspod_id, dnspod_token)
        self.base_url = "https://dnsapi.cn/"

    def request(self, path, **params):
        params.update({"login_token": self.login_token, "format": "json"})
        url = urllib.parse.urljoin(self.base_url, path)

        try:
            resp = self.session.post(url, data=params)
            resp.raise_for_status()
        except Exception as e:
            raise DnspodException(e)

        try:
            body = resp.json()
            code = body["status"]["code"]
            message = body["status"]["message"]
        except Exception as e:
            raise DnspodException(e, resp.text)

        if code != "1":
            raise DnspodErrorResponse(code, message)
        return body

    def get_domains(self, offset=None, length=None):
        body = self.request("Domain.List", offset=offset, length=length)
        return body["info"]["domain_total"], body["domains"]

    def get_domain(self, domain_id):
        body = self.request("Domain.Info", domain_id=domain_id)
        return body["domain"]

    def get_lines(self, domain_grade):
        body = self.request("Record.Line", domain_grade=domain_grade)
        return body["line_ids"]

    def get_records(self, domain_id, record_type=None, offset=None, length=None):
        body = self.request(
            "Record.List",
            domain_id=domain_id,
            record_type=record_type,
            offset=offset,
            length=length,
        )
        return int(body["info"]["record_total"]), body["records"]

    def get_record(self, domain_id, record_id):
        body = self.request("Record.Info", domain_id=domain_id, record_id=record_id)
        return body["record"]

    def create_cname_record(self, domain_id, sub_domain, value):
        body = self.request(
            "Record.Create",
            domain_id=domain_id,
            sub_domain=sub_domain,
            record_type="CNAME",
            record_line_id=0,  # 默认线路
            value=value,
        )
        return body["record"]

    def modify_cname_record(self, domain_id, record_id, sub_domain, value):
        body = self.request(
            "Record.Modify",
            domain_id=domain_id,
            record_id=record_id,
            sub_domain=sub_domain,
            record_type="CNAME",
            record_line_id=0,  # 默认线路
            value=value,
        )
        return body["record"]

    def remove_record(self, domain_id, record_id):
        self.request("Record.Remove", domain_id=domain_id, record_id=record_id)