Dnspod域名设置（python脚本实现ddns）

家用宽带申请到公网ip后，ip地址是动态的，为了使用方便，需申请一个域名并设置ddns，这样就可以使用固定的域名来访问自己的服务器，不需要使用难记且经常变化的ip地址。

一、域名申请

有一些免费三级域名申请的地方，但是方便起见，我在腾讯云购买了一个 数字.xyz的域名，一年8块，也便宜。

使用免费的域名解析功能可以设置

• 子域名级数5 级

• 负载均衡2 条

• url转发2条

• 子域名数量不限

个人对域名没太大需求，目前只设置3级A类域名，免费解析足够用，ddns功能就需要自己编写脚本来实现了。

二、脚本编写

使用python编写，通过dnspod api获取个人域名内的dns解析记录，与外网地址比较，如果不同就更新dns记录，每隔5-10分钟运行一次脚本。

2.1 获取外网地址

访问https://myip.ipip.net就能得到自己的ip地址。

2.2 获得域名解析信息

使用dnspod api， 说明见腾讯云文档：https://docs.dnspod.cn/api/call-requency/

2.2.1 获取token

访问https://console.dnspod.cn/account/token/token，创建一个秘钥，

完成后程序中可以使用ID,TOKEN来访问api。

2.2.2 函数功能

实现了获取域名，添加域名，删除域名，获取记录，添加记录，删除记录，修改记录功能，都通过requests 的post方法访问对应网址即可。

1. 获取记录函数需要参数为 域名

2. 删除记录需要参数为 记录id和域名， 记录id通过获取记录函数获得。

3. 修改记录需要参数为 域名，子域名，记录id，要修改的值， 记录类型， 线路类型。

4. 添加记录如果仅能添加2个相同的子域名，原因是负载均衡的限制。

具体代码如下：

"""
访问和修改dnspod域名的类
get_domain无需参数，输出login_token对应的用户拥有的域名。
域名下包含记录，使用getrecord（domain_list）获取,每条记录对应一个子域名，也是最常用的。
对仅拥有一个域名一个公网ip的用户来说，日常使用仅需要在ip地址发生变化时，更新每条记录就可以了。
"""
import requests
import logging
import re
from dnspod_setting import Settings
from time import sleep


class Dnspod:
    """
    操作dnspod域名的类
    """

    def __init__(self, login_token):
        self._login_token = login_token
        self._format = 'json'

        # 设置 logger
        logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        self._logger = logging.getLogger(__name__)
        handler = logging.FileHandler(Settings.log_file, encoding='utf-8')
        handler.setLevel(logging.INFO)
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        self._logger.addHandler(handler)

        self._add_domain_URL = 'https://dnsapi.cn/Domain.Create'  # 添加域名
        self._del_domain_URL = 'https://dnsapi.cn/Domain.Remove'  # 删除域名
        self._add_record_URL = 'https://dnsapi.cn/Record.Create'  # 添加记录
        self._alter_record_URL = 'https://dnsapi.cn/Record.Modify'  # 修改记录
        self._get_record_URL = 'https://dnsapi.cn/Record.List'  # 获取记录列表
        self._del_record_URL = 'https://dnsapi.cn/Record.Remove'  # 删除记录
        self._get_domain_list_URL = 'https://dnsapi.cn/Domain.List'  # 获取域名列表

    @staticmethod
    def get_outer_ip():
        """
        获取外网ip
        :return:
        """
        ip = ''
        for count in range(10):     # 重试10次
            try:
                res = requests.get('https://myip.ipip.net', timeout=5).text
                ip = re.findall(r'(\d+\.\d+\.\d+\.\d+)', res)
                ip = ip[0] if ip else ''
            except:
                pass
            if ip:          # 一旦ip不为空，直接返回
                return ip
            sleep(1)
        return ip

    def get_domain_list(self):
        """
        获取域名列表
        :return: 域名列表： ['xxx.xx', 'xxx.xx'] or []
        """
        try:
            r = requests.post(self._get_domain_list_URL, data={'login_token': self._login_token,
                                                               'format': self._format})
            response_record_json = r.json()
            if response_record_json['status']['code'] == '1':
                domains_list = [doms['name'] for doms in response_record_json['domains'] if doms['status'] == 'enable']
                for domain in domains_list:
                    self._logger.info(f"域名:{domain}")
                return domains_list
            else:
                return []
        except Exception as e:
            self._logger.error(e)

    def add_domain(self, domains_list):
        """
        添加域名
        :param domains_list:['xxx.xxx.xxx', 'xx2.xxx.xxx']
        :return: True or False
        """
        try:
            for domain in domains_list:
                r = requests.post(self._add_domain_URL, data={'login_token': self._login_token,
                                                              'format': self._format,
                                                              'domain': domain,
                                                              })
                response_record_json = r.json()
                if response_record_json["status"]["code"] == "1":
                    self._logger.info(
                        f'域名：{response_record_json["domain"]["domain"]}，添加成功;域名ID：{response_record_json["domain"]["id"]}')
                else:
                    self._logger.error(f'域名：{domain},添加失败，错误信息：{response_record_json["status"]["message"]}')
        except Exception as e:
            self._logger.error(e)

    def del_domain(self, domains_list):
        """
        删除域名
        :param domains_list:['xxx.xxx.xxx', 'xx2.xxx.xxx']
        :return: True or False
        """
        try:
            for domain in domains_list:
                r = requests.post(self._del_domain_URL, data={'login_token': self._login_token,
                                                              'format': self._format,
                                                              'domain': domain,
                                                              })
                response_record_json = r.json()
                if response_record_json["status"]["code"] == "1":
                    self._logger.info(f'删除域名：{domain}成功.')
                else:
                    self._logger.error(f'删除域名：{domain},添加失败，错误信息：{response_record_json["status"]["message"]}')
        except Exception as e:
            self._logger.error(e)

    def add_record(self, domain, sub_domain_list, value, record_type='A', record_line_value='默认'):
        """
        添加域名和解析记录
        :param domain: 要添加记录的域名
        :param sub_domain_list: 要添加的子域名列表
        :param record_type: 添加记录类型：默认 A
        :param record_line_value: 线路类型， 使用 ‘默认’即可
        :param value: 要添加的值， A类型是ip地址。
        :return:
        """
        try:
            for sub_domain in sub_domain_list:
                r = requests.post(self._add_record_URL, data={'login_token': self._login_token,
                                                              'format': self._format,
                                                              'domain': domain,
                                                              'sub_domain': sub_domain,
                                                              'record_type': record_type,
                                                              'record_line': record_line_value,
                                                              'value': value,
                                                              })
                response_record_json = r.json()
                if response_record_json["status"]["code"] == "1":
                    self._logger.info(f'域名：{domain}，成功添加记录：{response_record_json["record"]["name"]} ')
                else:
                    self._logger.error(
                        f'域名：{domain}，添加：{sub_domain}，记录失败，错误信息：{response_record_json["status"]["message"]}')
        except Exception as e:
            self._logger.error(e)

    def get_record(self, domain):
        """
        获取域名下的记录
        :param domain: "xxx.xxx.xx"
        :return:
        """
        records_list = []
        try:
            r = requests.post(self._get_record_URL, data={'login_token': self._login_token,
                                                          'format': self._format,
                                                          'domain': domain,
                                                          })

            response_record_json = r.json()
            if response_record_json['status']['code'] == '1':
                self._logger.info(
                    f"域名：{response_record_json['domain']['name']}，"
                    f"共有：{response_record_json['info']['sub_domains']}个子域名"
                    f"和：{response_record_json['info']['record_total']}条解析记录.")
                for record in response_record_json['records']:
                    if record['type'] == 'A':  # 只修改A类型的记录，避免修改dnspod默认记录。
                        records_list.append({
                            'domain': domain,
                            'id': record['id'],
                            'name': record['name'],
                            'line': record['line'],
                            'type': record['type'],
                            'value': record['value']
                        })
        except Exception as e:
            self._logger.error(e)
            return []

        return records_list

    def alter_record(self, domain, sub_domain, record_id, value, record_type='A', record_line='默认'):
        """
        修改解析记录
        :param domain: 域名，不可缺
        :param sub_domain: 子域名，可选，默认值为@
        :param record_id: 记录id
        :param value: 值
        :param record_type: 记录类型，默认 A
        :param record_line: 线路类型，使用‘默认’即可
        :return:
        """
        try:
            r = requests.post(self._alter_record_URL, data={'login_token': self._login_token,
                                                            'format': self._format,
                                                            'domain': domain,
                                                            'sub_domain': sub_domain,
                                                            'record_id': record_id,
                                                            'record_type': record_type,
                                                            'record_line': record_line,
                                                            'value': value,
                                                            })
            response_record_json = r.json()
            if response_record_json['status']['code'] == '1':
                self._logger.info(f"修改记录{record_id}成功。")
            else:
                self._logger.error(f"记录ID：{record_id}修改失败，错误信息: {response_record_json['status']['message']}。")
        except Exception as e:
            self._logger.error(e)

    def del_record(self, domain, record_id):
        """
        删除解析记录
        :param domain: 域名，
        :param record_id: 记录id，通过get_record获取
        :return:
        """
        try:
            r = requests.post(self._del_record_URL, data={'login_token': self._login_token,
                                                          'format': self._format,
                                                          'domain': domain,
                                                          'record_id': record_id,
                                                          })
            response_record_json = r.json()
            if response_record_json['status']['code'] == '1':
                self._logger.info(
                    f"您正在解析记录，ID为：{record_id}，"
                    f"删除状态值为：{response_record_json['status']['code']}, "
                    f"信息为：{response_record_json['status']['message']}")
            else:
                self._logger.error(
                    f"您正在解析记录，ID为：{record_id}，"
                    f"删除状态值为：{response_record_json['status']['code']}, "
                    f"信息为：{response_record_json['status']['message']}")
        except Exception as e:
            self._logger.error(e)


def update_ip():
    """
    更新公网ip地址，定期运行保证域名正确，
    因为只有一个域名，只有一个ip，所以每次运行，获取域名的A类型解析记录，将所有记录ip修改为外网ip
    :return:
    """
    pod = Dnspod(Settings.login_token)
    ip_public = pod.get_outer_ip()
    if ip_public:
        records = pod.get_record(Settings.domain)
        for record in records:
            if ip_public == record['value']:
                print('地址未改变，无需更改')
            else:
                pod.alter_record(record['domain'], record['name'], record['id'], ip_public)


if __name__ == '__main__':
    update_ip()