aaPanel/class/panelDnsapi.py

# coding: utf-8
# +-------------------------------------------------------------------
# | aaPanel
# +-------------------------------------------------------------------
# | Copyright (c) 2015-2099 aaPanel(www.aapanel.com) All rights reserved.
# +-------------------------------------------------------------------
# | Author: aaPanel
# +-------------------------------------------------------------------
import ipaddress
import os
import re
import sys
import time
from datetime import datetime
from typing import Optional
from urllib.parse import urljoin

import requests

os.chdir("/www/server/panel")
sys.path.insert(0, "class/")
sys.path.insert(0, "class_v2/")

import public
from public.exceptions import HintException

__all__ = [
    "aaPanelDns",
    "NameSiloDns",
    "NameCheapDns",
    "CloudFlareDns",
    "PorkBunDns",
    "GodaddyDns",
    "ClouDns",
    "SpaceShipDNS",
    "AmazonRoute53Dns",
]


class ExtractZoneTool(object):
    def __call__(self, domain_name):
        root, zone = public.split_domain_sld(domain_name)
        if not zone:
            acme_txt = "_acme-challenge"
        else:
            acme_txt = "_acme-challenge.%s" % zone
        return root, zone, acme_txt


extract_zone = ExtractZoneTool()


def sync_log(body, mode="a"):
    dns_sync_log = os.path.join(public.get_panel_path(), "logs/dns_sync.log")
    body += "\n"
    with open(dns_sync_log, mode) as f:
        f.write(body)


def white_kwargs(body: dict, kw_prefix: dict, kwargs: dict) -> dict:
    """
    kwargs:  create 从 kw获取, update 从 new_record 获取
    """
    # when create or update
    if kwargs.get("priority") and kwargs.get("priority") != -1:  # 接受其他关键参数
        if kw_prefix.get("priority"):
            body[kw_prefix.get("priority")] = int(kwargs.get("priority"))
    return body


class BaseDns(object):
    def __init__(self):
        self.dns_provider_name = self.__class__.__name__
        self.api_user = ""
        self.api_key = ""

    def log_response(self, response: requests.Response):
        try:
            log_body = response.json()
        except ValueError:
            log_body = response.content
        return log_body

    # =============== acme ======================
    def create_dns_record(self, domain_name: str, domain_dns_value: str) -> None:
        raise NotImplementedError("create_dns_record method must be implemented.")

    def delete_dns_record(self, domain_name: str, domain_dns_value: str) -> None:
        raise NotImplementedError("delete_dns_record method must be implemented.")

    # =============== 域名管理同步信息 =====================
    def get_domains(self, **kwargs) -> list:
        raise NotImplementedError("get_domains method must be implemented.")

    def get_dns_record(self, domain_name: str) -> list:
        raise NotImplementedError("get_dns_record method must be implemented.")

    def create_org_record(
            self, domain_name: str, record: str, record_value: str, record_type: str, ttl: int, **kwargs
    ) -> Optional[dict]:
        raise NotImplementedError("create_org_record method must be implemented.")

    def remove_record(self, domain_name: str, record: str, record_type: str, **kwargs) -> Optional[dict]:
        raise NotImplementedError("remove_record method must be implemented.")

    def update_record(self, domain_name: str, record: dict, new_record: dict, **kwargs) -> Optional[dict]:
        raise NotImplementedError("update_record method must be implemented.")

    def raise_resp_error(self, response: requests.Response):
        raise HintException(
            "Error {dns_name}: status_code={status_code} response={response}".format(
                dns_name=self.dns_provider_name,
                status_code=response.status_code,
                response=self.log_response(response),
            )
        )

    def verify(self) -> Optional[bool]:
        try:
            self.get_domains()
        except Exception:
            raise HintException("Verify fail, please check your Api Account and Password")
        return True


class aaPanelDns(BaseDns):
    """
    遵循 ssl v2入参, 转发dnsmanager
    """
    dns_provider_name = "aapanel"
    kw_prefix = {
        "priority": "priority"
    }

    def __init__(self, api_user: str = None, api_key: str = None, **kwargs):  # noqa
        super().__init__()
        self.api_user = api_user
        self.api_key = api_key
        from ssl_dnsV2.dns_manager import DnsManager
        self.manager = DnsManager()

    # ============== acme ======================
    def create_dns_record(self, domain_name, domain_dns_value):
        domain_name = domain_name.lstrip("*.")
        _, _, acme_txt = extract_zone(domain_name)
        self.create_org_record(
            domain_name=domain_name,
            record=acme_txt,
            record_value=domain_dns_value,
            record_type="TXT",
            ttl=600,
        )

    def delete_dns_record(self, domain_name, domain_dns_value) -> None:
        domain_name = domain_name.lstrip("*.")
        root, _, acme_txt = extract_zone(domain_name)
        self.remove_record(root, acme_txt, "TXT", record_value=domain_dns_value)

    # =============== 域名管理 ====================
    def get_domains(self) -> list:
        return self.manager.get_domains()

    def get_dns_record(self, domain_name: str) -> list:
        records = []
        for x in self.manager.parser.get_zones_records(domain_name):
            try:
                # todo 更多类型特俗处理
                if x.get("type") == "SOA":
                    continue
                if x.get("type") == "MX":
                    priority = re.findall(r"^\s*(\d+)\s+", x.get("value"))
                    if priority:
                        x["priority"] = int(priority[0])
                record = {
                    "record": x.get("name"),
                    "record_type": x.get("type"),
                    "record_value": x.get("value"),
                    "ttl": x.get("ttl"),
                    "proxy": x.get("proxy", -1),
                    "priority": x.get("priority", -1),
                }
                records.append(record)
            except Exception as e:
                public.print_log(f"aaPanelDns get_dns_record error: {e}")
                continue
        return records

    def create_org_record(self, domain_name, record, record_value, record_type, ttl=600, **kwargs):
        root, _, _ = extract_zone(domain_name)
        body = {
            "name": record,
            "type": record_type.upper(),
            "value": record_value,
            "ttl": ttl,
            "proxy": kwargs.get("proxy", -1),
            "priority": kwargs.get("priority", -1),
        }
        body = white_kwargs(body, self.kw_prefix, kwargs)
        try:
            self.manager.add_record(
                domain=root, **body
            )
            return {"status": True, "msg": "Success"}
        except HintException as he:
            return {"status": False, "msg": str(he)}
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def remove_record(self, domain_name, record, record_type="TXT", **kwargs) -> dict:
        root, _, _ = extract_zone(domain_name)
        body = {
            "name": record,
            "type": record_type.upper(),
        }
        if kwargs.get("record_value"):
            body["value"] = kwargs.get("record_value")
        try:
            self.manager.delete_record(
                domain=root, **body
            )
            return {"status": True, "msg": "Success"}
        except HintException as he:
            return {"status": False, "msg": str(he)}
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def update_record(self, domain_name: str, record: dict, new_record: dict, **kwargs):
        domain, _, _ = extract_zone(domain_name)
        body = {
            "name": record.get("record"),
            "type": record.get("record_type").upper(),
            "value": record.get("record_value"),
            "ttl": record.get("ttl", 600),
            "priority": record.get("priority", 10),
            "proxy": record.get("proxy", -1),
            "new_record": {
                "name": new_record.get("record"),
                "type": new_record.get("record_type").upper(),
                "value": new_record.get("record_value"),
                "ttl": new_record.get("ttl", 600),
                "proxy": new_record.get("proxy", -1),
                "priority": new_record.get("priority", -1),
            }
        }
        try:
            self.manager.update_record(
                domain=domain, **body
            )
            return {"status": True, "msg": "Success"}
        except HintException as he:
            return {"status": False, "msg": str(he)}
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def verify(self) -> bool:
        if os.path.exists(public.get_panel_path() + "/class_v2/ssl_dnsV2/aadns.pl"):
            return True
        return False


# noinspection PyUnusedLocal
class NameCheapDns(BaseDns):
    dns_provider_name = "namecheap"
    kw_prefix = {
        "priority": "MXPref"
    }

    def __init__(self, api_user, api_key, **kwargs):
        super().__init__()
        self.api_user = api_user
        self.api_key = api_key
        self.timeout = 30
        self.base_url = "https://api.namecheap.com/xml.response"

    def _get_local_ip(self):
        ip = public.GetLocalIp()
        if isinstance(ipaddress.ip_address(ip), ipaddress.IPv6Address):
            raise HintException("Namecheap Api does not support IPv6")
        return ip

    def _get_hosts(self, domain_name) -> list:
        time.sleep(2)  # 限流
        params = {
            "ApiUser": self.api_user,
            "ApiKey": self.api_key,
            "UserName": self.api_user,
            "Command": "namecheap.domains.dns.getHosts",
            "ClientIp": self._get_local_ip(),
            "SLD": domain_name.split(".")[0],
            "TLD": domain_name.split(".")[1],
        }
        resp = requests.get(url=self.base_url, params=params, timeout=self.timeout)
        if resp.status_code != 200:
            self.raise_resp_error(resp)
        hosts = []
        index = 0
        hosts_info = self._generate_xml_tree(resp.text, ".//host")
        for host in hosts_info:
            index += 1
            try:
                ttl = int(host.get("TTL", 1))
            except Exception:
                ttl = 1
            try:
                mx_pref = int(host.get("MXPref", -1))
            except Exception:
                mx_pref = -1

            hosts.append({
                f"HostName{index}": host.get("Name"),
                f"RecordType{index}": host.get("Type"),
                f"Address{index}": host.get("Address"),
                f"TTL{index}": ttl,
                f"MXPref{index}": mx_pref if host.get("Type") == "MX" else -1,
            })
        return hosts

    # =============== acme ======================
    def create_dns_record(self, domain_name: str, domain_dns_value: str) -> None:
        # acme 调用
        domain_name = domain_name.lstrip("*.")
        _, _, acme_txt = extract_zone(domain_name)
        self.create_org_record(
            domain_name=domain_name,
            record=acme_txt,
            record_value=domain_dns_value,
            record_type="TXT",
        )

    def delete_dns_record(self, domain_name, dns_value=None) -> None:
        # 移除挑战值
        _, _, dns_name = extract_zone(domain_name)
        self.remove_record(domain_name, dns_name, "TXT")

    # =============== 域名管理 ====================
    @staticmethod
    def _generate_xml_tree(resp_body: str, findall: str):
        import xml.etree.ElementTree as EtTree
        from xml.etree.ElementTree import ParseError as ETParseError
        # noinspection HttpUrlsUsage
        tree_root = resp_body.replace('xmlns="http://api.namecheap.com/xml.response"', '')
        try:
            targets = EtTree.fromstring(tree_root).findall(findall)
            return targets
        except ETParseError:
            raise HintException("Error parsing XML response from Namecheap API")
        except Exception as e:
            raise HintException(e)

    def get_domains(self, verify: bool = False) -> list | bool:
        # 获取账号下所有域名, 判断域名nameserver归属, 并且所有返回均为xml
        domains = []
        page = 1
        while 1:
            params = {
                "ApiUser": self.api_user,
                "ApiKey": self.api_key,
                "UserName": self.api_user,
                "Command": "namecheap.domains.getList",  # returns a list of domains for the particular user
                "ClientIp": self._get_local_ip(),
                "Page": page,
                "PageSize": 100,
                "SortBy": "NAME",
            }
            resp = requests.get(url=self.base_url, params=params, timeout=self.timeout)
            if "API Key is invalid or API access has not been enabled" in resp.text:
                raise HintException("API Key is invalid or API access has not been enabled")

            if "ERROR" in resp.text:
                try:
                    err_msg = re.search(r"<Errors>(.*?)</Errors>", resp.text, re.DOTALL).group(1)
                except:
                    err_msg = resp.text
                raise HintException(err_msg)

            if verify:
                return True

            try:
                temp_domains = self._generate_xml_tree(resp.text, ".//Domain")
            except HintException as hit:
                public.print_log(hit)
                continue

            if not temp_domains:
                break

            time.sleep(3)  # 限流
            domains.extend(temp_domains)
            page += 1

        for expired in [d.get("Name", "") for d in domains if d.get("IsExpired") == "true"]:
            sync_log(f"|-- warning: [{expired}] is Expired, Skip It...")

        domains = [
            domain.get("Name") for domain in domains if domain.get("IsExpired") == "false"
        ]
        res = []
        for d in domains:
            try:
                params = {
                    "ApiUser": self.api_user,
                    "ApiKey": self.api_key,
                    "UserName": self.api_user,
                    # gets a list of DNS servers associated with the requested domain.
                    "Command": "namecheap.domains.dns.getList",
                    "ClientIp": self._get_local_ip(),
                    "SLD": d.split(".")[0],
                    "TLD": d.split(".")[1],
                }
                resp = requests.get(url=self.base_url, params=params, timeout=self.timeout)
                if resp.status_code != 200:
                    continue

                time.sleep(3)  # 限流
                for t in self._generate_xml_tree(resp.text, ".//DomainDNSGetListResult"):
                    if t.get("Domain") == d:
                        if t.get("IsUsingOurDNS") == "true":
                            res.append(d)
                            break
                        else:
                            sync_log(
                                f"|-- warning: [{t.get('Domain')}] is Not Using NameCheap DNS, Skip It..."
                            )
                            break
            except Exception as e:
                public.print_log(f"get_domains error {e}")
                continue
        return res

    # nc 所有record 不带域名
    def get_dns_record(self, domain_name: str):
        domain_name, _, _ = extract_zone(domain_name)
        try:
            records = self._get_hosts(domain_name)
        except Exception as e:
            public.print_log(f"get_dns_record error {e}")
            records = []
        res = []
        for r in records:
            temp = {}
            for k, v in r.items():
                if k.startswith("HostName"):
                    temp["record"] = v
                elif k.startswith("RecordType"):
                    temp["record_type"] = v
                elif k.startswith("Address"):
                    temp["record_value"] = v
                elif k.startswith("TTL"):
                    temp["ttl"] = r.get("ttl", 1)
                elif k.startswith("MXPref"):
                    temp["priority"] = v
                else:
                    temp[k] = v
            res.append(temp)
        return res

    def __set_hosts_with_params(self, domain_name: str, new_params: dict):
        try:
            setHosts_resp = requests.get(url=self.base_url, params=new_params, timeout=self.timeout)
        except Exception as e:
            return {"status": False, "msg": str(e)}
        if any([
            setHosts_resp.status_code != 200,
            f'Domain="{domain_name}" IsSuccess="true"' not in setHosts_resp.text
        ]):
            return {"status": False, "msg": setHosts_resp.text}
        return {"status": True, "msg": setHosts_resp.text}

    def __generate_new_params(self, new_params: dict, new_hosts: list) -> dict:
        for i, host in enumerate(new_hosts):
            index = i + 1
            for key in list(host.keys()):
                if key.startswith(("HostName", "Address", "RecordType", "TTL", "MXPref")):
                    base_key = ''.join([c for c in key if not c.isdigit()])
                    if base_key == "MXPref" and host[key] == -1:
                        continue
                    new_params[f"{base_key}{index}"] = host[key]
        return new_params

    # 创建record
    def create_org_record(self, domain_name, record, record_value, record_type, ttl=1, **kwargs):
        domain_name, _, _ = extract_zone(domain_name)
        hosts = self._get_hosts(domain_name)
        add_index = len(hosts) + 1
        params = {
            "ApiUser": self.api_user,
            "ApiKey": self.api_key,
            "UserName": self.api_user,
            "ClientIp": self._get_local_ip(),
            "Command": "namecheap.domains.dns.setHosts",
            "SLD": domain_name.split(".")[0],
            "TLD": domain_name.split(".")[1],
            "DomainName": domain_name,
        }
        # nc 单独处理
        for index, host in enumerate(hosts):
            idx = index + 1
            for field in ["HostName", "Address", "RecordType", "TTL", "MXPref"]:
                if field == "MXPref" and host[f"{field}{idx}"] == -1:
                    continue
                params[f"{field}{idx}"] = host[f"{field}{idx}"]

        params.update({
            f"HostName{add_index}": record,
            f"Address{add_index}": record_value,
            f"RecordType{add_index}": record_type,
            f"TTL{add_index}": ttl
        })
        return self.__set_hosts_with_params(domain_name, params)

    # 删除record
    def remove_record(self, domain_name, record, record_type="TXT", **kwargs) -> dict:
        domain_name, _, _ = extract_zone(domain_name)
        hosts_info = self._get_hosts(domain_name)
        new_hosts = [
            host for host in hosts_info if not (record in host.values() and record_type in host.values())
        ]
        if not new_hosts:
            # is empty
            return {"status": True, "msg": "Dns Record is empty."}
        new_params = {
            "ApiUser": self.api_user,
            "ApiKey": self.api_key,
            "UserName": self.api_user,
            "ClientIp": self._get_local_ip(),
            "Command": "namecheap.domains.dns.setHosts",
            "SLD": domain_name.split(".")[0],
            "TLD": domain_name.split(".")[1],
            "DomainName": domain_name,
        }
        new_params = self.__generate_new_params(new_params, new_hosts)
        return self.__set_hosts_with_params(domain_name, new_params)

    # 更新record
    def update_record(self, domain_name: str, record: dict, new_record: dict, **kwargs):
        domain_name, _, _ = extract_zone(domain_name)
        hosts_info = self._get_hosts(domain_name)
        new_hosts = []
        for index, host in enumerate(hosts_info):
            if all([
                record.get("record") in host.values(),
                record.get("record_type") in host.values(),
                record.get("record_value") in host.values(),
            ]):
                host[f"HostName{index + 1}"] = new_record.get("record")
                host[f"RecordType{index + 1}"] = new_record.get("record_type")
                host[f"Address{index + 1}"] = new_record.get("record_value")
                host[f"TTL{index + 1}"] = kwargs.get("ttl", 1)
                if new_record.get("priority") != -1:
                    host[f"{self.kw_prefix.get("priority")}{index + 1}"] = new_record.get("priority")
                new_hosts.append(host)
            else:
                new_hosts.append(host)
        if not new_hosts:  # is empty
            return {"status": True, "msg": "Dns Record is empty."}
        new_params = {
            "ApiUser": self.api_user,
            "ApiKey": self.api_key,
            "UserName": self.api_user,
            "ClientIp": self._get_local_ip(),
            "Command": "namecheap.domains.dns.setHosts",
            "SLD": domain_name.split(".")[0],
            "TLD": domain_name.split(".")[1],
            "DomainName": domain_name,
        }
        new_params = self.__generate_new_params(new_params, new_hosts)
        return self.__set_hosts_with_params(domain_name, new_params)

    # 验证
    def verify(self):
        # namecheap token 请求时会切掉多余的str长度
        try:
            self.get_domains(verify=True)
        except HintException as e:
            raise e
        except Exception:
            raise HintException(
                "Verify fail, please check your Api Account and Password, "
                "Add the server address to the NameCheapDns API whitelist."
            )
        return True


# noinspection PyUnusedLocal
class CloudFlareDns(BaseDns):
    dns_provider_name = "cloudflare"
    kw_prefix = {
        "priority": "priority"
    }

    def __init__(self, api_user, api_key, limit: bool = True, **kwargs):
        super().__init__()
        self.cf_zone_id = None
        self.api_user = api_user
        self.api_key = api_key
        self.limit = limit
        self.cf_base_url = "https://api.cloudflare.com/client/v4/"
        self.time_out = 65  # seconds

    def _get_auth_headers(self) -> dict:
        if self.limit:
            return {"Authorization": "Bearer " + self.api_key}
        else:  # api limit False, is global permissions
            return {"X-Auth-Email": self.api_user, "X-Auth-Key": self.api_key}

    def find_dns_zone(self, domain_name):
        domain = domain_name.lstrip("*.")
        url = self.cf_base_url + "zones?status=active&per_page=20"
        headers = self._get_auth_headers()
        page = 1

        while True:
            paginated_url = url + f"&page={page}"
            find_dns_zone_response = requests.get(paginated_url, headers=headers, timeout=self.time_out)
            if find_dns_zone_response.status_code != 200:
                raise HintException(
                    "Error find cloudflare dns zone, please check your Api Account and Password:"
                    " status_code={status_code} response={response}".format(
                        status_code=find_dns_zone_response.status_code,
                        response=self.log_response(find_dns_zone_response),
                    )
                )
            result = find_dns_zone_response.json()["result"]

            matched_zones = [
                x for x in result if domain == x.get("name") or domain.endswith("." + x.get("name"))
            ]
            if matched_zones:
                # 优先最长匹配, max时len相等, 即eq, not eq时max为最精确
                best_match = max(matched_zones, key=lambda x: len(x.get("name", "")))
                self.cf_zone_id = best_match.get("id")
                return

            if len(result) < 20:
                break
            page += 1

        raise HintException(
            "Error unable to get DNS zone for domain={domain}".format(
                domain=domain
            )
        )

    # =============== acme ======================

    @staticmethod
    def _build_cf_value(record_type: str, record_value: str, priority: int = -1) -> dict:
        """
        A/AAAA/CNAME/NS/TXT -> {"content": value}
        MX -> {"data": {"priority": N, "target": value}}
        CAA -> {"data": {"flags": N, "tag": tag, "value": value}}
        """
        record_type = str(record_type or "").upper()

        if record_type == "MX":
            pref = priority if str(priority) not in {"", "-1", "None"} else 10
            return {"data": {"priority": pref, "target": record_value}}

        if record_type == "CAA":
            parts = str(record_value or "").strip().split(None, 2)
            if len(parts) == 3:
                flag_str, tag, caa_val = parts
                try:
                    flags = int(flag_str)
                except ValueError:
                    flags = 0
                return {"data": {"flags": flags, "tag": tag, "value": caa_val}}
            return {"content": record_value}

        return {"content": record_value}

    @staticmethod
    def _parse_cf_value(record_type: str, item: dict) -> str:
        """从返回的记录中解析值"""
        record_type = str(record_type or "").upper()
        data = item.get("data")
        if record_type == "MX" and data:
            return str(data.get("target", ""))
        if record_type == "CAA" and data:
            flags = data.get("flags", 0)
            tag = data.get("tag", "")
            val = data.get("value", "")
            return f"{flags} {tag} {val}"
        return str(item.get("content", ""))

    @staticmethod
    def _parse_cf_priority(record_type: str, item: dict) -> int:
        """提取优先级: MX -> data.priority 或顶层 priority"""
        if str(record_type or "").upper() == "MX":
            data = item.get("data", {})
            pref = data.get("priority", item.get("priority", -1))
        else:
            pref = item.get("priority", -1)
        try:
            return int(pref)
        except (TypeError, ValueError):
            return -1

    def create_dns_record(self, domain_name, domain_dns_value):
        # acme 调用
        domain_name = domain_name.lstrip("*.")
        self.find_dns_zone(domain_name)
        # CloudFlare 的 acme 记录名称是 _acme-challenge.{完整域名}
        acme_txt = "_acme-challenge." + domain_name
        url = urljoin(
            self.cf_base_url,
            "zones/{0}/dns_records".format(self.cf_zone_id),
        )
        headers = self._get_auth_headers()
        body = {
            "type": "TXT",
            "name": acme_txt,
            "content": str(domain_dns_value),
        }

        create_cloudflare_dns_record_response = requests.post(
            url, headers=headers, json=body, timeout=self.time_out
        )
        if create_cloudflare_dns_record_response.status_code != 200:
            raise HintException(
                "Error creating cloudflare dns record: status_code={status_code} response={response}".format(
                    status_code=create_cloudflare_dns_record_response.status_code,
                    response=self.log_response(create_cloudflare_dns_record_response),
                )
            )

    def delete_dns_record(self, domain_name, domain_dns_value):
        # 移除挑战值
        domain_name = domain_name.lstrip("*.")
        self.find_dns_zone(domain_name)
        # CloudFlare 的 acme 记录名称是 _acme-challenge.{完整域名}
        acme_txt = "_acme-challenge." + domain_name
        self.remove_record(domain_name, acme_txt, "TXT", record_value=domain_dns_value)

    # =============== 域名管理 ====================

    def get_domains(self) -> list:
        url = self.cf_base_url + "zones?status=active&order=name"
        headers = self._get_auth_headers()
        domains = []
        page = 1
        per_page = 50
        fail_count = 0
        while True:
            params = {"page": page, "per_page": per_page}
            try:
                res = requests.get(url, headers=headers, params=params, timeout=self.time_out)
                time.sleep(1)  # 限流
                resp = res.json()
                if not resp.get("success"):
                    fail_count += 1
                    if fail_count >= 3:
                        raise HintException("get domains fail")
                    else:
                        time.sleep(1)
                        continue
                result = resp.get("result", [])
                domains.extend([i.get("name", "") for i in result])
                if len(result) < per_page:
                    break
                page += 1
            except Exception as e:
                public.print_log(f"cloudflare get_domains error {e}")
                raise HintException(e)
        return domains

    def get_dns_record(self, domain_name: str) -> list:
        domain_name, _, _ = extract_zone(domain_name)
        self.find_dns_zone(domain_name)
        url = self.cf_base_url + f"zones/{self.cf_zone_id}/dns_records"
        result = []
        page = 1
        per_page = 500
        fail_count = 0
        while True:
            params = {"page": page, "per_page": per_page}
            try:
                response = requests.get(
                    url, headers=self._get_auth_headers(), params=params, timeout=self.time_out
                )
                data = response.json()
                if data.get("success"):
                    records = data.get("result", [])
                    for i in records:
                        r_type = i.get("type", "").upper()
                        # 跳过 SOA 和根 NS 记录
                        if r_type == "SOA":
                            continue
                        r_value = self._parse_cf_value(r_type, i)
                        priority = self._parse_cf_priority(r_type, i)
                        result.append({
                            "record": i.get("name", ""),
                            "record_value": r_value,
                            "record_type": r_type,
                            "proxy": i.get("proxied", False),
                            "priority": priority,
                            "ttl": i.get("ttl", 1),
                        })
                    if len(records) < per_page:
                        break
                    page += 1
                else:
                    fail_count += 1
                    if fail_count >= 3:
                        break
            except requests.RequestException as e:
                public.print_log(f"get_dns_record error {e}")
                break
        return result

    # 创建record
    def create_org_record(self, domain_name, record, record_value, record_type, ttl=1, proxied=0, **kwargs):
        domain_name, _, _ = extract_zone(domain_name)
        proxied = True if proxied == 1 else False
        self.find_dns_zone(domain_name)
        url = self.cf_base_url + f"zones/{self.cf_zone_id}/dns_records"
        headers = self._get_auth_headers()
        ttl_val = 600 if int(ttl) == 1 else int(ttl)
        priority = kwargs.get("priority", -1)

        body = {"name": record, "type": record_type, "ttl": ttl_val, "proxied": proxied}
        body.update(self._build_cf_value(record_type, record_value, priority))
        body = white_kwargs(body, self.kw_prefix, kwargs)

        try:
            create_res = requests.post(url, headers=headers, json=body, timeout=self.time_out)
            create_res = create_res.json()
            if create_res.get("success"):
                return {"status": True, "msg": create_res}
            return {"status": False, "msg": str(create_res.get("errors"))}
        except requests.exceptions.HTTPError as http_err:
            return {"status": False, "msg": http_err}
        except Exception as e:
            return {"status": False, "msg": str(e)}

    # 删除record
    def remove_record(self, domain_name, record, record_type="TXT", **kwargs) -> dict:
        domain_name, _, _ = extract_zone(domain_name)
        self.find_dns_zone(domain_name)
        headers = self._get_auth_headers()
        list_dns_payload = {"type": record_type, "name": record}
        list_dns_url = self.cf_base_url + f"zones/{self.cf_zone_id}/dns_records"
        list_dns_response = requests.get(
            list_dns_url, params=list_dns_payload, headers=headers, timeout=self.time_out
        )
        if list_dns_response.status_code != 200:
            return {"status": False, "msg": f"HTTP {list_dns_response.status_code}: {list_dns_response.text}"}
        try:
            results = list_dns_response.json().get("result", [])
            if not results:
                return {"status": True, "msg": "Dns Record is empty."}

            # 如果提供了record_value，需要匹配正确的记录
            record_value = kwargs.get("record_value")
            if record_value:
                # 解析记录值以进行匹配
                for item in results:
                    parsed_value = self._parse_cf_value(record_type, item)
                    if parsed_value == record_value:
                        dns_record_id = item["id"]
                        url = self.cf_base_url + f"zones/{self.cf_zone_id}/dns_records/{dns_record_id}"
                        remove_res = requests.delete(url, headers=headers, timeout=self.time_out)
                        remove_res = remove_res.json()
                        if remove_res.get("success"):
                            return {"status": True, "msg": remove_res}
                        return {"status": False, "msg": str(remove_res.get("errors"))}
                # 没有找到匹配的记录
                return {"status": False, "msg": "Dns Record not found with specified value"}
            else:
                # 没有提供record_value，删除第一条匹配的记录
                dns_record_id = results[0]["id"]
                url = self.cf_base_url + f"zones/{self.cf_zone_id}/dns_records/{dns_record_id}"
                remove_res = requests.delete(url, headers=headers, timeout=self.time_out)
                remove_res = remove_res.json()
                if remove_res.get("success"):
                    return {"status": True, "msg": remove_res}
                return {"status": False, "msg": str(remove_res.get("errors"))}
        except requests.exceptions.HTTPError as http_err:
            return {"status": False, "msg": http_err}
        except Exception as e:
            return {"status": False, "msg": str(e)}

    # 更新record
    def update_record(self, domain_name, record: dict, new_record: dict, **kwargs):
        domain_name, _, _ = extract_zone(domain_name)
        self.find_dns_zone(domain_name)
        record_type = record.get("record_type")
        record_name = record.get("record")
        get_url = self.cf_base_url + f"zones/{self.cf_zone_id}/dns_records?type={record_type}&name={record_name}"
        try:
            # get record id
            get_response = requests.get(get_url, headers=self._get_auth_headers(), timeout=self.time_out)
            get_result = get_response.json()
            if get_result.get("success") and get_result.get("result"):
                record_id = get_result["result"][0]["id"]
                update_url = self.cf_base_url + f"zones/{self.cf_zone_id}/dns_records/{record_id}"

                new_type = new_record.get("record_type", record_type)
                new_value = new_record.get("record_value", "")
                new_ttl = int(new_record.get("ttl", 600))
                if new_ttl == 1:
                    new_ttl = 600
                proxied = True if new_record.get("proxy") == 1 else False
                priority = new_record.get("priority", -1)

                body = {
                    "type": new_type,
                    "name": new_record.get("record"),
                    "ttl": new_ttl,
                    "proxied": proxied,
                }
                body.update(self._build_cf_value(new_type, new_value, priority))
                body = white_kwargs(body, self.kw_prefix, new_record)

                update_response = requests.put(update_url, headers=self._get_auth_headers(), json=body,
                                               timeout=self.time_out)
                update_result = update_response.json()
                if update_result.get("success"):
                    return {"status": True, "msg": update_result}
                return {"status": False, "msg": str(update_result.get("errors"))}
            else:
                return {"status": False, "msg": "Dns Record Not Found!"}
        except requests.exceptions.HTTPError as http_err:
            return {"status": False, "msg": http_err}
        except Exception as err:
            return {"status": False, "msg": err}

    # 验证
    def verify(self):
        try:
            self.get_domains()
        except Exception:
            raise HintException("Verify fail, please check your Api Account and Password")
        return True


# noinspection PyUnusedLocal
class PorkBunDns(BaseDns):
    dns_provider_name = "porkbun"
    kw_prefix = {
        "priority": "prio"
    }

    def __init__(self, api_user, api_key, **kwargs):
        super().__init__()
        self.api_user = api_user  # secretapikey
        self.api_key = api_key
        self.timeout = 30
        self.base_url = "https://api.porkbun.com/api/json/v3"

    def _json_data(self) -> dict:
        return {
            "secretapikey": self.api_user,
            "apikey": self.api_key,
        }

    def _retrieve_record_by_domain(self, domain_name: str) -> list:
        domain_name, _, _ = extract_zone(domain_name)
        url = self.base_url + f"/dns/retrieve/{domain_name}"
        data = self._json_data()
        try:
            response = requests.post(url, json=data, timeout=self.timeout)
            time.sleep(1)  # 限流
            res = response.json()
            if res.get("status") != "SUCCESS":
                # 如果没有任何记录，porkbun可能会返回错误
                if "domain not found" in res.get("message", "").lower():
                    return []
                raise HintException("get record fail")
            return res.get("records", [])
        except Exception as err:
            raise HintException(err)

    def _get_sub_domain(self, root: str, record: str) -> str:
        """
        标准化子域名处理：将record 转换为子域名部分
        例如: root="example.com", record="www.example.com" -> 返回 "www"
        例如: root="example.com", record="example.com" -> 返回 ""
        """
        if not record or record == root:
            return ""
        if record.endswith("." + root):
            return record[:-(len(root) + 1)]
        return record

    def _get_fqdn(self, root: str, record: str) -> str:
        """
        标准化为完整域名，用于匹配 retrieve 返回的结果
        """
        sub = self._get_sub_domain(root, record)
        return f"{sub}.{root}" if sub else root

    # =============== acme ======================
    def create_dns_record(self, domain_name, domain_dns_value) -> None:
        if domain_name.startswith("*."):
            domain_name = domain_name[2:]
        root, _, acme_txt = extract_zone(domain_name)
        self.create_org_record(
            domain_name=root,
            record=acme_txt,
            record_value=domain_dns_value,
            record_type="TXT",
        )

    def delete_dns_record(self, domain_name, domain_dns_value) -> None:
        if domain_name.startswith("*."):
            domain_name = domain_name[2:]
        root, _, acme_txt = extract_zone(domain_name)
        fqdn = f"{acme_txt}.{root}" if acme_txt else root
        self.remove_record(root, fqdn, "TXT")

    # =============== 域名管理 ====================
    def get_domains(self) -> list:
        url = self.base_url + "/domain/listAll"
        data = self._json_data()

        response = requests.post(url, json=data, timeout=self.timeout)
        res = response.json()
        if not res.get("status") == "SUCCESS":
            raise HintException("get domains fail")

        res_domains = []
        for d in res.get("domains"):
            if d.get("status") != "ACTIVE":
                sync_log(f"|-- warning: [{d.get('domain')}] is Not ACTIVE, Skip It...")
                continue
            if d.get("expireDate") and d.get("expireDate") < datetime.now().strftime("%Y-%m-%d %H:%M:%S") and \
                    d.get("expireDate") != "0000-00-00 00:00:00":
                sync_log(f"|-- warning: [{d.get('domain')}] is Expired, Skip It...")
                continue

            try:
                detail_url = self.base_url + f"/domain/getNs/{d.get('domain')}"
                detail_resp = requests.post(detail_url, json=data, timeout=self.timeout)
                detail_res = detail_resp.json()
                if not detail_res.get("status") == "SUCCESS":
                    continue

                if all("porkbun.com" in d.lower() for d in detail_res.get("ns")):
                    res_domains.append(d.get("domain"))
                else:
                    sync_log(f"|-- warning: [{d.get('domain')}] is Not Using Porkbun DNS, Skip It...")

            except Exception as err:
                public.print_log(f"{self.dns_provider_name} get domains detail error {err}")
                continue
        res_domains.sort()
        return res_domains

    def get_dns_record(self, domain_name: str) -> list:
        try:
            return [
                {
                    "record": x.get("name"),  # Porkbun 返回的是 FQDN
                    "record_value": x.get("content"),
                    "record_type": x.get("type"),
                    "proxy": False,
                    "ttl": int(x.get("ttl", 1)),
                    "priority": int(x.get("prio", -1)) if x.get("prio") not in [None, "0"] else -1,
                } for x in self._retrieve_record_by_domain(domain_name)
            ]
        except Exception as err:
            public.print_log(f"{self.dns_provider_name} get_domains error {err}")
            raise HintException(err)

    def create_org_record(self, domain_name, record, record_value, record_type, ttl=1, **kwargs):
        root, _, _ = extract_zone(domain_name)
        sub_name = self._get_sub_domain(root, record)

        url = self.base_url + f"/dns/create/{root}"
        data = self._json_data()
        data.update({
            "name": sub_name,
            "type": record_type,
            "content": record_value,
            "ttl": 600 if ttl == 1 else int(ttl),
        })
        data = white_kwargs(data, self.kw_prefix, kwargs)  # 接受其他关键参数

        try:
            response = requests.post(url, json=data, timeout=self.timeout)
            res = response.json()
            if res.get("status") != "SUCCESS":
                return {"status": False, "msg": res.get("message")}
            return {"status": True, "msg": res.get("status")}
        except Exception as err:
            return {"status": False, "msg": err}

    def remove_record(self, domain_name, record, record_type="TXT", **kwargs) -> dict:
        try:
            root, _, _ = extract_zone(domain_name)
            # 无论输入,都标准化为 FQDN 进行匹配
            target_fqdn = self._get_fqdn(root, record).lower()
            res = {"status": "False", "msg": "Dns Record is Not Found."}
            for r in self._retrieve_record_by_domain(domain_name):
                if r.get("name", "").lower() == target_fqdn and r.get("type", "").lower() == record_type.lower():
                    url = self.base_url + f"/dns/delete/{root}/{int(r.get('id'))}"
                    response = requests.post(url, json=self._json_data(), timeout=self.timeout)
                    res = response.json()
                    break

            if res.get("status") != "SUCCESS":
                return {"status": False, "msg": res.get("message", "Fail, please try again later.")}
            return {"status": True, "msg": res.get("status", "SUCCESS")}

        except Exception as err:
            public.print_log(f"{self.dns_provider_name} remove record error {err}")
            return {"status": False, "msg": err}

    def update_record(self, domain_name, record: dict, new_record: dict, **kwargs):
        try:
            root, _, _ = extract_zone(domain_name)
            # 无论输入,都标准化为 FQDN 进行匹配
            old_fqdn = self._get_fqdn(root, record.get("record")).lower()

            res = {"status": "False", "msg": "Dns Record is Not Found."}
            data = self._json_data()
            data.update({
                "name": new_record.get("record"),
                "type": new_record.get("record_type"),
                "content": new_record.get("record_value"),
                "ttl": 600 if new_record.get("ttl") == 1 else int(new_record.get("ttl")),
            })
            data = white_kwargs(data, self.kw_prefix, new_record)  # 接受其他关键参数

            for r in self._retrieve_record_by_domain(domain_name):
                if all([
                    r.get("name", "").lower() == old_fqdn,
                    r.get("type", "").lower() == record.get("record_type", "").lower(),
                    r.get("content") == record.get("record_value"),
                ]):
                    url = self.base_url + f"/dns/edit/{root}/{int(r.get('id'))}"
                    response = requests.post(url, json=data, timeout=self.timeout)
                    res = response.json()
                    break

            if res.get("status") != "SUCCESS":
                return {"status": False, "msg": res.get("message", "Fail, please try again later.")}
            return {"status": True, "msg": res.get("status", "SUCCESS")}

        except Exception as err:
            public.print_log(f"{self.dns_provider_name} remove record error {err}")
            return {"status": False, "msg": err}

    def verify(self):
        try:
            url = self.base_url + "/ping"
            data = self._json_data()
            response = requests.post(url, json=data, timeout=self.timeout)
            res = response.json()
            if res.get("status") == "SUCCESS":
                return True
        except:
            pass
        raise HintException("Verify fail, please check your Api Account and Password")


# noinspection PyUnusedLocal
class NameSiloDns(BaseDns):
    dns_provider_name = "namesilo"
    kw_prefix = {
        "priority": "rrdistance"
    }

    def __init__(self, api_user, api_key, **kwargs):
        super().__init__()
        self.api_user = api_user
        self.api_key = api_key
        self.timeout = 30
        self.base_url = "https://www.namesilo.com/api"
        self.ver_type_key = f"version=1&type=json&key={api_key}"

    def _remote_record(self, domain_name: str) -> list:
        try:
            url = f"{self.base_url}/dnsListRecords?{self.ver_type_key}&domain={domain_name}"
            response = requests.get(url, params={"page": 1, "pageSize": 100})
            time.sleep(1)  # 限流
            res = response.json()
            if res.get("reply").get("code") != 300:
                raise HintException(res.get("reply").get("detail"))
            return res.get("reply", {}).get("resource_record", [])
        except Exception as e:
            public.print_log(f"get_dns_record error {e}")
            raise HintException(e)

    def _find_rrid(self, domain_name: str, record: dict) -> str:
        rrid = ""
        try:
            for r in self._remote_record(domain_name):
                if all([
                    r.get("host") == record.get("record"),
                    r.get("type") == record.get("record_type"),
                ]):
                    rrid = r.get("record_id")
                    break
        except Exception as e:
            public.print_log(f"NameSilo find rrid error {e}")
        return rrid

    # =============== acme ======================
    def create_dns_record(self, domain_name, domain_dns_value) -> None:
        _, _, acme_txt = extract_zone(domain_name)
        self.create_org_record(
            domain_name=domain_name,
            record=acme_txt,
            record_value=domain_dns_value,
            record_type="TXT",
            ttl=3600,
        )

    def delete_dns_record(self, domain_name, domain_dns_value) -> None:
        domain_name, _, acme_txt = extract_zone(domain_name)
        acme_txt = acme_txt + "." + domain_name
        self.remove_record(domain_name, acme_txt, "TXT")

    # =============== 域名管理 ====================
    def get_domains(self, verify: bool = False) -> list | bool:
        url = f"{self.base_url}/listDomains?{self.ver_type_key}&withBid=1&skipExpired=1"
        page = 1
        domains = []
        fail_count = 0
        while 1:
            params = {"page": page, "pageSize": 100}
            try:
                response = requests.get(url, params=params)
                if "Invalid API Key" in response.text or "Permission denied" in response.text:
                    raise HintException("Invalid API Key, Verify fail, please check your Api Key")
                res = response.json()
                time.sleep(1)  # 限流
                if res.get("reply").get("code") != 300:
                    fail_count += 1
                    if fail_count >= 2:
                        raise HintException(res.get("reply").get("detail"))
                    else:
                        continue
                if verify:
                    return True
                temp_domains = res["reply"].get("domains")
                if not temp_domains:
                    break
                temp_domains = [temp_domains] if isinstance(temp_domains, dict) else temp_domains
                domains.extend(temp_domains)
                page += 1
            except Exception as e:
                raise HintException(f"get domains errro {e}")
        try:
            domains = sorted(domains, key=lambda x: x["domain"]["domain"])
        except:
            pass
        res_domains = []
        detail_url_pre = f"{self.base_url}/getDomainInfo?{self.ver_type_key}"
        for d in domains:
            try:
                d = d.get("domain", {})
                if d.get("expires") and d.get("expires") < datetime.now().strftime("%Y-%m-%d"):
                    sync_log(f"|-- warning: [{d.get('domain')}] is Expired, Skip It...")
                    continue
                # domain detail info
                detail_url = f"{detail_url_pre}&domain={d.get("domain")}"
                detail_resp = requests.get(detail_url)
                time.sleep(1)  # 限流
                rp = detail_resp.json()
                if rp.get("reply", {}).get("code") != 300:
                    continue
                if all(
                        "DNSOWL.COM" in n.get("nameserver", "").upper() for n in rp["reply"].get("nameservers", [])
                ):
                    res_domains.append(d.get("domain"))
                else:
                    sync_log(
                        f"|-- warning: [{d.get('domain')}] is Not Using NameSilo DNS, Skip It..."
                    )
            except Exception as err:
                public.print_log(f"{self.dns_provider_name} get domains detail error {err}")
                continue

        return res_domains

    def get_dns_record(self, domain_name: str) -> list:
        domain_name, _, _ = extract_zone(domain_name)
        try:
            records = self._remote_record(domain_name)
            return [
                {
                    "record": x.get("host"),
                    "record_value": x.get("value"),
                    "record_type": x.get("type"),
                    "proxy": False,
                    "ttl": int(x.get("ttl", 1)),
                    "priority": int(x.get("distance", -1)) if x.get("distance") not in [None, 0] else -1,
                } for x in records
            ]
        except Exception as e:
            public.print_log(f"get_dns_record error {e}")
            raise HintException(e)

    def create_org_record(self, domain_name, record, record_value, record_type, ttl=1, **kwargs):
        domain_name, _, _ = extract_zone(domain_name)
        try:
            url = f"{self.base_url}/dnsAddRecord?{self.ver_type_key}&domain={domain_name}"
            params = {
                "rrhost": record,
                "rrvalue": record_value,
                "rrtype": record_type,
                "rrttl": 7207 if int(ttl) == 1 else int(ttl),  # default is 7207
            }
            params = white_kwargs(params, self.kw_prefix, kwargs)  # 接受其他关键参数
            response = requests.get(url, params=params, timeout=self.timeout)
            resp = response.json()
            if resp.get("reply").get("code") != 300:
                return {"status": False, "msg": resp.get("reply").get("detail")}

            return {"status": True, "msg": response.json()}
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def remove_record(self, domain_name: str, record: str, record_type: str, **kwargs) -> dict:
        domain_name, _, _ = extract_zone(domain_name)
        try:
            rrid = self._find_rrid(domain_name, {"record": record, "record_type": record_type})
            url = f"{self.base_url}/dnsDeleteRecord?{self.ver_type_key}&domain={domain_name}&rrid={rrid}"
            response = requests.get(url, timeout=self.timeout)
            resp = response.json()
            if resp.get("reply").get("code") != 300:
                return {"status": False, "msg": resp.get("reply").get("detail")}
            return {"status": True, "msg": resp.get("reply").get("detail")}
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def update_record(self, domain_name: str, record: dict, new_record: dict, **kwargs):
        domain_name, _, _ = extract_zone(domain_name)
        try:
            rrid = self._find_rrid(domain_name, record)
            if not rrid:
                return {"status": False, "msg": "Dns Record Not Found!"}

            url = f"{self.base_url}/dnsUpdateRecord?{self.ver_type_key}&domain={domain_name}"
            params = {
                "rrid": rrid,
                "rrhost": new_record.get("record"),
                "rrvalue": new_record.get("record_value"),
                "rrtype": new_record.get("record_type"),
                "rrttl": 7207 if new_record.get("ttl") == 1 else int(new_record.get("ttl")),  # default is 7207
            }
            params = white_kwargs(params, self.kw_prefix, new_record)  # 接受其他关键参数
            response = requests.get(url, params=params, timeout=self.timeout)
            resp = response.json()
            if resp.get("reply").get("code") != 300:
                return {"status": False, "msg": resp.get("reply").get("detail")}
            return {"status": True, "msg": resp.get("reply").get("detail")}
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def verify(self) -> bool:
        try:
            self.get_domains(verify=True)
        except Exception as e:
            raise HintException(e)
        return True


class GodaddyDns(BaseDns):
    dns_provider_name = "godaddy"
    kw_prefix = {
        "priority": "priority"
    }

    def __init__(self, api_user, api_key, **kwargs):  # noqa
        super().__init__()
        self.api_user = api_user  # secret key
        self.api_key = api_key
        self.timeout = 30
        # self.base_url = "https://api.ote-godaddy.com"
        self.base_url = "https://api.godaddy.com"

        self.headers = {
            "Authorization": f"sso-key {api_key}:{api_user}",
            "Accept": "application/json",
            "Content-Type": "application/json",
        }

    def _make_request(self, method, endpoint, payload=None):
        url = f"{self.base_url}{endpoint}"
        response = requests.request(method, url, headers=self.headers, json=payload)
        if response.status_code in [200, 204]:
            return response
        if response.json().get("code") == "UNABLE_TO_AUTHENTICATE":
            # godaddy api 需要拥有50个域名方可调用, 官方链接说明
            # https://www.godaddy.com/zh/help/how-do-i-access-domain-related-apis-42424
            raise Exception('GoDaddy API have been rejected, '
                            'Official Documentation link:\n'
                            '"https://www.godaddy.com/zh/help/how-do-i-access-domain-related-apis-42424"')
        raise Exception(f"Godaddy API Error {response.status_code}: {response.text}")

    # =============== acme ======================
    def create_dns_record(self, domain_name, domain_dns_value):
        # acme 调用
        domain_name = domain_name.lstrip("*.")
        self.create_org_record(
            domain_name=domain_name,
            record="_acme-challenge." + domain_name,
            record_value=domain_dns_value,
            record_type="TXT",
            ttl=600,
        )

    def delete_dns_record(self, domain_name, domain_dns_value):
        # 移除挑战值
        domain_name, _, acme_txt = extract_zone(domain_name)
        acme_txt = acme_txt + "." + domain_name
        self.remove_record(domain_name, acme_txt, "TXT")

    # =============== ote test ==================
    def ote_buy_domain(self, domain_name):
        if "ote" not in self.base_url:
            raise HintException("Only support ote env")
        url = self.base_url + f"/v1/domains/purchase"
        data = {
            "domain": domain_name,
            "consent": {
                "agreementKeys": [
                    "DNRA"
                ],
                "agreedBy": "bt-dev3",
                "agreedAt": datetime.utcnow().isoformat() + "Z"  # noqa
            },
            "period": 1,
            "renewAuto": False,
            "privacy": False,
            "contactRegistrant": {
                "nameFirst": "abc",
                "nameLast": "abc",
                "email": "abc@example.com",
                "phone": "+1.1234567890",
                "addressMailing": {
                    "address1": "123 Main St",
                    "city": "town",
                    "state": "AZ",
                    "postalCode": "85001",
                    "country": "US"
                }
            }
        }
        try:
            response = requests.post(url, headers=self.headers, json=data, timeout=self.timeout)
            res = response.json()
            if response.status_code != 200:
                raise HintException(res.get("message", "buy domain fail"))
            return res
        except Exception as e:
            raise HintException(e)

    # =============== 域名管理 ====================
    def get_domains(self) -> list:
        try:
            all_domains = []
            limit = 300
            endpoint = f"/v1/domains?limit={limit}"
            while 1:
                res = self._make_request("GET", endpoint).json()
                if not res:
                    break
                for domain in res:
                    if domain.get("status") != "ACTIVE":
                        sync_log(f"|-- warning: [{domain.get('domain')}] is Not ACTIVE, Skip It...")
                        continue
                    if domain.get("expires") and domain.get("expires") < datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"):
                        sync_log(f"|-- warning: [{domain.get('domain')}] is Expired, Skip It...")
                        continue
                    all_domains.append(domain.get("domain"))

                if len(res) < limit:
                    break

                last_domain = res[-1]["domain"]
                endpoint = f"/v1/domains?limit={limit}&marker={last_domain}"
                time.sleep(1)  # 限流

            all_domains.sort()
            return all_domains
        except Exception as e:
            raise HintException(f"get domains error {e}")

    def get_dns_record(self, domain_name: str) -> list:
        domain_name, _, _ = extract_zone(domain_name)
        try:
            res = self._make_request(
                "GET", f"/v1/domains/{domain_name}/records"
            ).json()
            return [
                {
                    "record": x.get("name"),
                    "record_value": x.get("data"),
                    "record_type": x.get("type"),
                    "ttl": int(x.get("ttl", 1)),
                    "priority": int(x.get("priority", -1)) if x.get("priority") not in [None, 0] else -1,
                    "proxy": -1,
                } for x in res
            ]
        except Exception as e:
            raise HintException(e)

    def create_org_record(self, domain_name, record, record_value, record_type, ttl=1, **kwargs):
        domain_name, _, _ = extract_zone(domain_name)
        body = {
            "data": record_value,
            "name": record,
            "ttl": 600 if ttl == 1 else int(ttl),
            "type": record_type
        }
        body = white_kwargs(body, self.kw_prefix, kwargs)
        try:
            self._make_request(
                "PATCH",
                f"/v1/domains/{domain_name}/records",
                payload=[body]
            )
            return {"status": True, "msg": "success"}
        except Exception as e:
            raise HintException(e)

    def remove_record(self, domain_name, record, record_type="TXT", **kwargs) -> dict:
        domain_name, _, _ = extract_zone(domain_name)
        try:
            self._make_request(
                "DELETE", f"/v1/domains/{domain_name}/records/{record_type}/{record}"
            )
            return {"status": True, "msg": "success"}
        except Exception as e:
            raise HintException(e)

    def update_record(self, domain_name, record: dict, new_record: dict, **kwargs):
        domain_name, _, _ = extract_zone(domain_name)
        try:
            body = {
                "data": new_record.get("record_value"),
                "name": new_record.get("record"),
                "ttl": 600 if new_record.get("ttl") == 1 else int(new_record.get("ttl")),
                "type": new_record.get("record_type")
            }
            body = white_kwargs(body, self.kw_prefix, new_record)
            self._make_request(
                "PUT",
                f"/v1/domains/{domain_name}/records/{record['record_type']}/{record['record']}",
                payload=[body]
            )
            return {"status": True, "msg": "success"}
        except Exception as e:
            raise HintException(e)

    def verify(self) -> Optional[bool]:
        try:
            self.get_domains()
        except Exception as e:
            raise HintException(f"Verify fail, please check your Api Key and Secret Key: {e}")
        return True


class ClouDns(BaseDns):
    dns_provider_name = "cloudns"
    kw_prefix = {
        "priority": "priority"
    }

    def __init__(self, api_user, api_key, **kwargs):  # noqa
        super().__init__()
        self.api_user = str(api_user or "").strip()
        self.api_key = str(api_key or "").strip()
        self.timeout = 30
        self.base_url = "https://api.cloudns.net/dns/"

    def _auth_params(self) -> dict:
        if self.api_user.isdigit():  # 接受 api auth id
            return {
                "auth-id": self.api_user,
                "auth-password": self.api_key,
            }
        # 回退账号策略
        return {
            "auth-user": self.api_user,
            "auth-password": self.api_key,
        }

    def _make_request(self, endpoint: str, payload: dict = None, method: str = "GET"):
        params = self._auth_params()
        if payload:
            params.update(payload)
        url = urljoin(self.base_url, endpoint)
        response = requests.request(method, url, params=params, timeout=self.timeout)
        if response.status_code != 200:
            self.raise_resp_error(response)

        try:
            data = response.json()
        except ValueError:
            raise HintException(f"Cloudns API returned invalid JSON: {response.text}")

        if isinstance(data, dict):
            status = str(data.get("status", "")).lower()
            if status in {"failed", "failure", "error"}:
                raise HintException(data.get("statusDescription") or data.get("status") or str(data))
            if str(data.get("result", "")).lower() in {"failed", "failure", "error"}:
                raise HintException(data.get("statusDescription") or data.get("message") or str(data))
        return data

    @staticmethod
    def _normalize_host(domain_name: str, record: str) -> str:
        record = (record or "@").strip().rstrip(".")
        if not record or record == "@":
            return "@"

        if record == domain_name:
            return "@"

        suffix = f".{domain_name}"
        if record.endswith(suffix):
            host = record[:-len(suffix)]
            return host or "@"
        return record

    def _normalize_zone_domain(self, domain_name: str) -> str:
        domain = str(domain_name or "").strip().rstrip(".").lstrip("*.")
        if not domain:
            raise HintException("Missing domain-name")

        root, _, _ = extract_zone(domain)
        domain = (root or domain).strip().rstrip(".")
        if not domain:
            raise HintException("Missing domain-name")
        return domain

    def _find_record_id(self, domain_name: str, record: str, record_type: str, record_value: str = None):
        host = self._normalize_host(domain_name, record)
        try:
            for item in self.get_dns_record(domain_name):
                if item.get("record_type", "").upper() != record_type.upper():
                    continue
                if self._normalize_host(domain_name, item.get("record")) != host:
                    continue
                if record_value is not None and item.get("record_value") != record_value:
                    continue
                return item.get("record_id")
        except Exception as e:
            public.print_log(f"{self.dns_provider_name} find record id error {e}")
        return None

    def _build_acme_record(self, domain_name: str) -> tuple[str, str]:
        domain = str(domain_name or "").strip().rstrip(".").lstrip("*.")
        zone_domain = self._normalize_zone_domain(domain)
        suffix = f".{zone_domain}"
        if domain == zone_domain:
            acme_host = "_acme-challenge"
        elif domain.endswith(suffix):
            sub = domain[:-len(suffix)]
            acme_host = f"_acme-challenge.{sub}"
        else:
            acme_host = "_acme-challenge"
        return zone_domain, acme_host

    @staticmethod
    def _parse_caa_value(record_value: str) -> tuple[int, str, str]:
        parts = str(record_value or "").strip().split(None, 2)
        if len(parts) != 3:
            raise HintException("CAA record_value format error, expected: '<flag> <type> <value>'")

        flag_raw, caa_type, caa_value = parts
        if not caa_type or not caa_value:
            raise HintException("CAA record_value format error, expected: '<flag> <type> <value>'")

        # Accept UI-style quoted values such as: 0 issue "letsencrypt.org"
        if (caa_value.startswith('"') and caa_value.endswith('"')) or (
                caa_value.startswith("'") and caa_value.endswith("'")
        ):
            caa_value = caa_value[1:-1].strip()
        if caa_value == "":
            raise HintException("CAA record_value format error, expected: '<flag> <type> <value>'")

        try:
            caa_flag = int(flag_raw)
        except Exception:
            raise HintException("CAA flag must be integer")
        return caa_flag, caa_type, caa_value

    def _build_record_payload(self, domain_name: str, record: str, record_value: str, record_type: str,
                              ttl: int, extra: dict) -> dict:
        record_type = str(record_type or "").upper()
        payload = {
            "host": self._normalize_host(domain_name, record),
            "record-type": record_type,
            "ttl": 600 if int(ttl) == 1 else int(ttl),
        }

        if record_type == "CAA":
            # Match get_dns_record output format: "flag type value".
            caa_flag, caa_type, caa_value = self._parse_caa_value(record_value)
            payload.update({
                "caa_flag": caa_flag,
                "caa_type": caa_type,
                "caa_value": caa_value,
            })
            return payload

        payload["record"] = record_value
        if record_type == "MX":
            priority = extra.get("priority", -1)
            if str(priority) not in {"", "-1", "None"}:
                payload["priority"] = int(priority)
        return payload

    # =============== acme ======================
    def create_dns_record(self, domain_name: str, domain_dns_value: str) -> None:
        zone_domain, acme_txt = self._build_acme_record(domain_name)
        res = self.create_org_record(
            domain_name=zone_domain,
            record=acme_txt,
            record_value=domain_dns_value,
            record_type="TXT",
            ttl=600,
        )
        if not res.get("status"):
            raise HintException(res.get("msg") or "create dns record failed")

    def delete_dns_record(self, domain_name: str, domain_dns_value: str) -> None:
        zone_domain, acme_txt = self._build_acme_record(domain_name)
        res = self.remove_record(zone_domain, acme_txt, "TXT", record_value=domain_dns_value)
        if not res.get("status"):
            raise HintException(res.get("msg") or "delete dns record failed")

    # =============== 域名管理 ====================
    def get_domains(self, verify: bool = False, page: int = 1, per_page: int = 100) -> list | bool:
        try:
            domains = []
            current_page = max(int(page), 1)
            page_size = max(int(per_page), 1)

            def _is_active_domain(item: dict) -> str:
                name = str(item.get("name", "")).strip().rstrip(".")
                zone = str(item.get("zone", "")).strip().lower()
                status = str(item.get("status", "")).strip().lower()
                type = str(item.get("type", "")).strip().lower()
                if zone != "domain":
                    return ""
                if status not in {"1", "active", "true"}:
                    return ""
                if not name or name.endswith(".arpa"):
                    return ""
                if type != "master":
                    return ""
                return name

            while 1:
                payload = {
                    "page": current_page,
                    "rows-per-page": page_size,
                }
                data = self._make_request("list-zones.json", payload)
                if verify:
                    return True

                page_domains = []
                if not isinstance(data, list):
                    raise HintException(f"list-zones response type error: {type(data).__name__}")

                raw_count = len(data)
                for item in data:
                    if isinstance(item, dict):
                        domain = _is_active_domain(item)
                        if domain:
                            page_domains.append(domain)

                if raw_count == 0:
                    break

                domains.extend(page_domains)
                if raw_count < page_size:
                    break

                current_page += 1
                time.sleep(1)  # 限流

            return sorted(set(domains))
        except Exception as e:
            raise HintException(f"get domains error {e}")

    def get_dns_record(self, domain_name: str) -> list:
        domain_name = self._normalize_zone_domain(domain_name)
        try:
            data = self._make_request("records.json", {"domain-name": domain_name})
            items = []
            if isinstance(data, dict):
                items = [v for v in data.values() if isinstance(v, dict)]
            elif isinstance(data, list):
                items = [v for v in data if isinstance(v, dict)]

            result = []
            for x in items:
                try:
                    status = str(x.get("status", "")).strip().lower()
                    if status not in {"1", "active", "true"}:
                        continue

                    record_type = str(x.get("type", "")).strip().upper()
                    host = str(x.get("host", "")).strip()
                    host = host if host else "@"

                    # CAA style "flag type value", default flag=0 when missing.
                    if record_type == "CAA":
                        caa_type = str(x.get("caa_type", "")).strip()
                        caa_value = str(x.get("caa_value", "")).strip()
                        if not caa_type or caa_value == "":
                            continue
                        caa_flag = str(x.get("caa_flag", "0")).strip() or "0"
                        record_value = f"{caa_flag} {caa_type} {caa_value}"
                    else:
                        record_value = x.get("record", "")

                    ttl = int(x.get("ttl", 1))
                    priority = x.get("priority", -1)
                    result.append({
                        "record": host,
                        "record_value": record_value,
                        "record_type": record_type,
                        "ttl": ttl,
                        "priority": int(priority) if str(priority) not in {"", "0", "-1", "None"} else -1,
                        "proxy": -1,
                        "record_id": x.get("id") or x.get("record_id"),
                    })
                except Exception:
                    continue
            return result
        except Exception as e:
            raise HintException(e)

    def create_org_record(self, domain_name, record, record_value, record_type, ttl=1, **kwargs):
        domain_name = self._normalize_zone_domain(domain_name)
        try:
            payload = {
                "domain-name": domain_name,
                **self._build_record_payload(
                    domain_name=domain_name,
                    record=record,
                    record_value=record_value,
                    record_type=record_type,
                    ttl=ttl,
                    extra=kwargs,
                ),
            }
            res = self._make_request("add-record.json", payload, method="POST")
            return {"status": True, "msg": res}
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def remove_record(self, domain_name, record, record_type="TXT", **kwargs) -> dict:
        domain_name = self._normalize_zone_domain(domain_name)
        try:
            record_id = kwargs.get("record_id") or self._find_record_id(
                domain_name,
                record,
                record_type,
                kwargs.get("record_value"),
            )
            if not record_id:
                return {"status": True, "msg": "Dns Record is empty."}

            payload = {
                "domain-name": domain_name,
                "record-id": int(record_id),
            }
            res = self._make_request("delete-record.json", payload, method="POST")
            return {"status": True, "msg": res}
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def update_record(self, domain_name, record: dict, new_record: dict, **kwargs):
        domain_name = self._normalize_zone_domain(domain_name)
        try:
            record_id = record.get("record_id") or self._find_record_id(
                domain_name,
                record.get("record"),
                record.get("record_type", ""),
                record.get("record_value"),
            )
            if not record_id:
                return {"status": False, "msg": "Dns Record Not Found!"}

            payload = {
                "domain-name": domain_name,
                "record-id": int(record_id),
                **self._build_record_payload(
                    domain_name=domain_name,
                    record=new_record.get("record"),
                    record_value=new_record.get("record_value"),
                    record_type=new_record.get("record_type", ""),
                    ttl=new_record.get("ttl", 1),
                    extra=new_record,
                ),
            }
            res = self._make_request("mod-record.json", payload, method="POST")
            return {"status": True, "msg": res}
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def verify(self) -> Optional[bool]:
        try:
            self.get_domains(verify=True)
        except Exception as e:
            public.print_log(f"e = {e}")
            if "You don't have access to the HTTP API. Check your plan." in str(e):
                raise HintException(public.lang(
                    "Cloudns API have been rejected, "
                    "Free accounts (Free plan) don't have access to API, Please upgrade your plan."
                ))
            raise HintException(public.lang(f"Verify fail, please check your Api Account and Password: {e}"))
        return True


class SpaceShipDNS(BaseDns):
    dns_provider_name = "spaceshipdns"
    kw_prefix = {
        "priority": "priority"
    }

    def __init__(self, api_user, api_key, **kwargs):  # noqa
        super().__init__()
        self.api_user = str(api_user or "").strip()  # API Key
        self.api_key = str(api_key or "").strip()  # API Secret
        self.timeout = 30
        self.base_url = "https://spaceship.dev/api/v1"

    def _get_auth_headers(self) -> dict:
        return {
            "X-Api-Key": self.api_user,
            "X-Api-Secret": self.api_key,
            "Accept": "application/json",
            "Content-Type": "application/json",
        }

    def _make_request(self, method: str, endpoint: str, payload=None, params=None):
        url = f"{self.base_url}{endpoint}"
        headers = self._get_auth_headers()
        response = requests.request(
            method, url, headers=headers, json=payload, params=params, timeout=self.timeout
        )
        if response.status_code not in (200, 201, 204):
            self.raise_resp_error(response)
        if response.status_code == 204 or not response.text.strip():
            return {}
        try:
            return response.json()
        except ValueError:
            raise HintException(f"Spaceship API returned invalid JSON: {response.text}")

    def _normalize_zone_domain(self, domain_name: str) -> str:
        domain = str(domain_name or "").strip().rstrip(".").lstrip("*.")
        if not domain:
            raise HintException("Missing domain-name")
        root, _, _ = extract_zone(domain)
        domain = (root or domain).strip().rstrip(".")
        if not domain:
            raise HintException("Missing domain-name")
        return domain

    @staticmethod
    def _get_sub_domain(root: str, record: str) -> str:
        """
        标准化子域名
        root="example.com", record="www.example.com" -> "www"
        root="example.com", record="example.com" -> ""
        root="example.com", record="www" -> "www"
        """
        record = str(record or "").strip().rstrip(".")
        if not record or record == root or record == "@":
            return ""
        suffix = f".{root}"
        if record.endswith(suffix):
            return record[:-len(suffix)]
        return record

    @staticmethod
    def _parse_record_value(item: dict) -> str:
        """从返回记录中解析值, 各类型字段不同
        A/AAAA  -> address
        CNAME   -> target
        MX      -> exchange
        NS      -> host
        TXT     -> value
        CAA     -> "flag tag value"
        """
        record_type = str(item.get("type", "")).upper()
        if record_type in ("A", "AAAA"):
            return str(item.get("address", ""))
        if record_type == "CNAME":
            return str(item.get("cname", ""))
        if record_type == "MX":
            return str(item.get("exchange", ""))
        if record_type == "NS":
            return str(item.get("nameserver", ""))
        if record_type == "CAA":
            flag = item.get("flag", 0)
            tag = item.get("tag", "")
            value = item.get("value", "")
            return f"{flag} {tag} {value}"
        return str(item.get("value", ""))

    @staticmethod
    def _parse_record_priority(item: dict) -> int:
        """提取优先级: MX -> preference"""
        record_type = str(item.get("type", "")).upper()
        if record_type == "MX":
            pref = item.get("preference", -1)
        else:
            pref = item.get("priority", -1)
        try:
            pref = int(pref)
        except (TypeError, ValueError):
            pref = -1
        return pref if str(pref) not in {"", "-1", "None"} else -1

    def _build_record_body(self, zone_domain: str, record: str, record_value: str,
                           record_type: str, ttl: int, extra: dict) -> dict:
        """
        构造PUT创建记录的单条 body
        各类型使用不同值字段:
          A/AAAA -> address,
          CNAME -> target,
          MX -> exchange+preference
          NS -> host,
          TXT -> value,
          CAA -> flag+tag+value
        """
        record_type = str(record_type or "").upper()
        name = self._get_sub_domain(zone_domain, record) or record
        ttl_val = 600 if int(ttl) == 1 else int(ttl)

        if record_type in ("A", "AAAA"):
            body = {"type": record_type, "name": name, "address": record_value, "ttl": ttl_val}

        elif record_type == "CNAME":
            body = {"type": record_type, "name": name, "cname": record_value, "ttl": ttl_val}

        elif record_type == "NS":
            body = {"type": record_type, "name": name, "nameserver": record_value, "ttl": ttl_val}

        elif record_type == "MX":
            body = {
                "type": record_type,
                "name": name,
                "exchange": record_value,
                "preference": int(extra.get("priority", 10)),
                "ttl": ttl_val,
            }

        elif record_type == "CAA":
            parts = str(record_value).strip().split(None, 2)
            if len(parts) == 3:
                body = {
                    "type": record_type, "name": name,
                    "flag": int(parts[0]), "tag": parts[1],
                    "value": parts[2].strip('"').strip("'"),
                    "ttl": ttl_val,
                }
            else:
                body = {"type": record_type, "name": name, "value": record_value, "ttl": ttl_val}
        else:
            # TXT 及其他
            body = {"type": record_type, "name": name, "value": record_value, "ttl": ttl_val}
        return body

    def _build_delete_body(self, zone_domain: str, record: str, record_type: str, **kwargs) -> dict:
        """构造DELETE单条记录匹配 body, 各类型值字段不同"""
        record_type = str(record_type or "").upper()
        name = self._get_sub_domain(zone_domain, record) or record
        val = kwargs.get("record_value") or ""

        if record_type in ("A", "AAAA"):
            body = {"type": record_type, "name": name, "address": val}

        elif record_type == "CNAME":
            body = {"type": record_type, "name": name, "cname": val}

        elif record_type == "NS":
            body = {"type": record_type, "name": name, "nameserver": val}

        elif record_type == "MX":
            body = {
                "type": record_type, "name": name,
                "exchange": val,
                "preference": int(kwargs.get("priority", 0)),
            }

        elif record_type == "CAA":
            parts = str(val).strip().split(None, 2)
            if len(parts) == 3:
                body = {
                    "type": record_type, "name": name,
                    "flag": int(parts[0]), "tag": parts[1],
                    "value": parts[2].strip('"').strip("'"),
                }
            else:
                body = {"type": record_type, "name": name, "value": val}

        else:
            # TXT / 其他
            body = {"type": record_type, "name": name, "value": val}
        return body

    # =============== acme ======================
    def create_dns_record(self, domain_name: str, domain_dns_value: str) -> None:
        _, _, acme_txt = extract_zone(domain_name)
        res = self.create_org_record(
            domain_name=domain_name,
            record=acme_txt,
            record_value=domain_dns_value,
            record_type="TXT",
            ttl=600,
        )
        if not res.get("status"):
            raise HintException(res.get("msg") or "create dns record failed")

    def delete_dns_record(self, domain_name: str, domain_dns_value: str) -> None:
        _, _, acme_txt = extract_zone(domain_name)
        res = self.remove_record(domain_name, acme_txt, "TXT", record_value=domain_dns_value)
        if not res.get("status"):
            raise HintException(res.get("msg") or "delete dns record failed")

    # =============== 域名管理 ====================
    def get_domains(self, verify: bool = False) -> list:
        try:
            domains = []
            skip = 0
            take = 100
            while True:
                params = {"take": take, "skip": skip}
                data = self._make_request("GET", "/domains", params=params)
                if verify:
                    return []

                items = data.get("items", [])
                if not items:
                    break

                for item in items:
                    if not isinstance(item, dict):
                        continue
                    name = str(item.get("name", "")).strip().rstrip(".")
                    if not name:
                        continue
                    status = str(item.get("status", "")).strip().lower()
                    if status not in {"active", "1", "true", ""}:
                        sync_log(f"|-- warning: [{name}] is Not ACTIVE, Skip It...")
                        continue
                    domains.append(name)

                if len(items) < take:
                    break
                skip += take
                time.sleep(1)  # 限流

            return sorted(set(domains))
        except HintException:
            raise
        except Exception as e:
            raise HintException(f"get domains error {e}")

    def get_dns_record(self, domain_name: str) -> list:
        zone_domain = self._normalize_zone_domain(domain_name)
        try:
            result = []
            skip = 0
            take = 100
            while True:
                params = {"take": take, "skip": skip}
                data = self._make_request("GET", f"/dns/records/{zone_domain}", params=params)
                items = data.get("items", [])
                if not items:
                    break

                for x in items:
                    if not isinstance(x, dict):
                        continue
                    try:
                        record_type = str(x.get("type", "")).strip().upper()
                        host = str(x.get("name", "")).strip() or "@"
                        record_value = self._parse_record_value(x)
                        ttl = int(x.get("ttl", 1))
                        priority = self._parse_record_priority(x)
                        result.append({
                            "record": host,
                            "record_value": record_value,
                            "record_type": record_type,
                            "ttl": ttl,
                            "priority": priority,
                            "proxy": -1,
                        })
                    except Exception:
                        continue

                if len(items) < take:
                    break
                skip += take
                time.sleep(1)  # 限流

            return result
        except HintException:
            raise
        except Exception as e:
            raise HintException(f"get dns record error {e}")

    def create_org_record(self, domain_name, record, record_value, record_type, ttl=1, **kwargs):
        """PUT /dns/records/{domain} + {"items": [...]} 创建记录"""
        zone_domain = self._normalize_zone_domain(domain_name)
        try:
            body = self._build_record_body(zone_domain, record, record_value, record_type, ttl, kwargs)
            res = self._make_request("PUT", f"/dns/records/{zone_domain}", payload={"items": [body]})
            return {"status": True, "msg": res}
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def remove_record(self, domain_name, record, record_type="TXT", **kwargs) -> dict:
        """DELETE /dns/records/{domain} + 裸数组 [...] 删除记录"""
        zone_domain = self._normalize_zone_domain(domain_name)
        try:
            body = self._build_delete_body(zone_domain, record, record_type, **kwargs)
            res = self._make_request("DELETE", f"/dns/records/{zone_domain}", payload=[body])
            return {"status": True, "msg": res}
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def update_record(self, domain_name, record: dict, new_record: dict, **kwargs):
        """无原生更新支持, 先 delete 再 create 复用"""
        zone_domain = self._normalize_zone_domain(domain_name)
        try:
            # 1. 删除旧记录
            del_body = self._build_delete_body(
                zone_domain,
                record.get("record", ""),
                record.get("record_type", ""),
                **record,
            )
            self._make_request("DELETE", f"/dns/records/{zone_domain}", payload=[del_body])

            # 2. 创建新记录
            new_body = self._build_record_body(
                zone_domain,
                new_record.get("record", ""),
                new_record.get("record_value", ""),
                new_record.get("record_type", ""),
                new_record.get("ttl", 600),
                new_record,
            )
            res = self._make_request("PUT", f"/dns/records/{zone_domain}", payload={"items": [new_body]})
            return {"status": True, "msg": res}
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def verify(self) -> Optional[bool]:
        try:
            self.get_domains(verify=True)
        except Exception as e:
            raise HintException(f"Verify fail, please check your Api Key and Secret: {e}")
        return True


# noinspection PyUnusedLocal
class AmazonRoute53Dns(BaseDns):
    dns_provider_name = "route53"
    kw_prefix = {
        "priority": "priority"
    }
    # DNS 记录类型中支持多值的(同一 name + type 可存多个value)
    _MULTI_VALUE_TYPES = {"TXT", "CAA", "MX"}

    def __init__(self, api_user, api_key, **kwargs):  # noqa
        super().__init__()
        self.api_user = str(api_user or "").strip()  # AWS AccessKeyId
        self.api_key = str(api_key or "").strip()  # AWS SecretAccessKey
        try:
            import boto3
            from botocore.config import Config
            from botocore.exceptions import ClientError
        except ImportError:
            os.system("btpip install boto3")
            try:
                import boto3
                from botocore.config import Config
                from botocore.exceptions import ClientError
            except ImportError:
                raise HintException("Aws SDK boto3 is not installed, please install it first: btpip install boto3")

        self.boto3 = boto3
        self.ClientError = ClientError
        self.Config = Config
        session = self.boto3.Session(
            aws_access_key_id=self.api_user,
            aws_secret_access_key=self.api_key,
            region_name="us-east-1",
        )
        config = self.Config(retries={"max_attempts": 3, "mode": "standard"})
        self.r53 = session.client("route53", config=config)
        self._hosted_zones_cache = None

    def _find_hosted_zone(self, domain_name: str) -> Optional[str]:
        """查找域名对应的Zone (最长匹配, 参考cf处理)"""
        domain = domain_name.strip().rstrip(".").lstrip("*.")
        zones = self.get_domains()
        matched = [
            z for z in zones if domain == z or domain.endswith("." + z)
        ]
        if not matched:
            return None
        return max(matched, key=lambda x: len(x))

    def _get_zone_id(self, domain_name: str) -> str:
        """获取托管区 ID"""
        domain = str(domain_name or "").strip().rstrip(".")
        if not domain.endswith("."):
            domain = domain + "."

        paginator = self.r53.get_paginator("list_hosted_zones")
        for page in paginator.paginate():
            for zone in page.get("HostedZones", []):
                zone_name = zone.get("Name", "").rstrip(".")
                if zone_name == domain.rstrip("."):
                    zone_id = zone.get("Id", "")
                    if zone_id.startswith("/hostedzone/"):
                        zone_id = zone_id[len("/hostedzone/"):]
                    return zone_id

        raise HintException(f"Hosted zone not found for domain: {domain_name}")

    @staticmethod
    def _normalize_host(domain_name: str, record: str) -> str:
        """标准化子域名

        处理输入的record可能带域名也可能不带域名的情况:
        - "test" -> "test"
        - "test.example.com" -> "test"
        - "test.example.com." -> "test"
        - "@" -> "@"
        - "example.com" -> "@"
        """
        # 标准化domain_name，去掉末尾点
        domain = str(domain_name or "").strip().rstrip(".")
        # 标准化record，去掉末尾点和空格
        record = str(record or "").strip().rstrip(".")

        if not record or record in ("@", domain):
            return "@"

        # 检查record是否包含完整域名，如果是则提取子域名部分
        suffix = f".{domain}"
        if record.endswith(suffix):
            return record[:-len(suffix)] or "@"

        # record本身就是一个子域名（不带domain后缀）
        return record

    @staticmethod
    def _parse_caa_value(value: str) -> tuple:
        """解析 CAA 值: 'flag tag value' -> (flag, tag, value)"""
        parts = str(value or "").strip().split(None, 2)
        if len(parts) != 3:
            raise HintException("CAA record_value format error, expected: '<flag> <tag> <value>'")
        flag_str, tag, caa_value = parts
        if not tag or caa_value == "":
            raise HintException("CAA record_value format error, expected: '<flag> <tag> <value>'")
        if (caa_value.startswith('"') and caa_value.endswith('"')) or (
                caa_value.startswith("'") and caa_value.endswith("'")
        ):
            caa_value = caa_value[1:-1].strip()
        if caa_value == "":
            raise HintException("CAA record_value format error, expected: '<flag> <tag> <value>'")
        try:
            flag = int(flag_str)
        except Exception:
            raise HintException("CAA flag must be integer")
        return flag, tag, caa_value

    @staticmethod
    def _format_caa_value(flag: int, tag: str, caa_value: str) -> str:
        """格式化 CAA 值为 Route53 Value: 'flag tag "value"'"""
        return f'{flag} {tag} "{caa_value}"'

    @staticmethod
    def _ensure_upper(s: str) -> str:
        """确保字符串大写"""
        return str(s or "").upper()

    @staticmethod
    def _ensure_trailing_dot(name: str) -> str:
        """确保域名以点结尾"""
        return name + "." if name and not name.endswith(".") else name

    def _build_r53_value(self, record_type: str, record_value: str, priority: int = -1) -> str:
        """构造各类型使用不同字段格式"""
        record_type = self._ensure_upper(record_type)

        if record_type == "MX":
            pref = priority if str(priority) not in {"", "-1", "None"} else 10
            return f"{pref} {record_value}"

        if record_type == "TXT":
            val = str(record_value or "")
            if (val.startswith('"') and val.endswith('"')) or (val.startswith("'") and val.endswith("'")):
                val = val[1:-1]
            return f'"{val}"'

        if record_type == "CAA":
            flag, tag, caa_val = self._parse_caa_value(record_value)
            return self._format_caa_value(flag, tag, caa_val)

        return record_value

    def _parse_r53_value(self, record_type: str, value: str) -> str:
        """解析返回的 Value 为外部格式"""
        record_type = self._ensure_upper(record_type)
        value = str(value or "")

        if record_type == "TXT":
            if (value.startswith('"') and value.endswith('"')) or (value.startswith("'") and value.endswith("'")):
                value = value[1:-1]
            return value

        if record_type == "MX":
            parts = value.strip().split(None, 1)
            return parts[1] if len(parts) > 1 else value

        if record_type == "CAA":
            parts = value.strip().split(None, 2)
            if len(parts) == 3:
                flag_str, tag, caa_val = parts
                if (caa_val.startswith('"') and caa_val.endswith('"')) or (
                        caa_val.startswith("'") and caa_val.endswith("'")):
                    caa_val = caa_val[1:-1]
                return f"{flag_str} {tag} {caa_val}"
            return value

        return value

    @staticmethod
    def _parse_priority(record_type: str, value: str) -> int:
        """提取优先级: MX 从 value 解析"""
        record_type = str(record_type or "").upper()
        if record_type == "MX":
            parts = str(value or "").strip().split(None, 1)
            try:
                return int(parts[0])
            except (ValueError, IndexError):
                return -1
        return -1

    def _get_full_name(self, domain_name: str, record: str) -> str:
        """获取完整域名 (不带末尾点)"""
        host = self._normalize_host(domain_name, record)
        return domain_name if host == "@" else f"{host}.{domain_name}"

    def _make_change(self, action: str, name: str, record_type: str,
                     ttl: int, value: str, priority: int = -1) -> dict:
        """构造单条 Change 字典, 支持 A/AAAA/MX/TXT/NS/CAA/CNAME"""
        record_type = self._ensure_upper(record_type)

        # CAA 需要特殊处理(值格式不同)
        if record_type == "CAA":
            flag, tag, caa_val = self._parse_caa_value(value)
            r53_val = self._format_caa_value(flag, tag, caa_val)
            return {
                "Action": action,
                "ResourceRecordSet": {
                    "Name": self._ensure_trailing_dot(name),
                    "Type": "CAA",
                    "TTL": ttl,
                    "ResourceRecords": [{"Value": r53_val}],
                },
            }

        # DELETE 操作使用原始值，CREATE 操作需要格式化
        r_value = value if action == "DELETE" else self._build_r53_value(record_type, value, priority)

        return {
            "Action": action,
            "ResourceRecordSet": {
                "Name": self._ensure_trailing_dot(name),
                "Type": record_type,
                "TTL": ttl,
                "ResourceRecords": [{"Value": r_value}],
            },
        }

    # =============== acme ======================
    def create_dns_record(self, domain_name: str, domain_dns_value: str) -> None:
        """ACME 挑战专用, 自动合并相同记录名的多个挑战值"""
        domain_name = domain_name.lstrip("*.")
        _, _, acme_txt = extract_zone(domain_name)
        # 优化获取现有 TXT 记录
        existing_values = []
        try:
            existing_records = self._get_acme_txt_record(domain_name, acme_txt)
            for r in existing_records:
                val = r.get("record_value", "")
                if val and val not in existing_values:
                    existing_values.append(val)
        except Exception:
            pass

        # 合并去重
        all_values = existing_values + [domain_dns_value]
        all_values = list(dict.fromkeys(all_values).keys())

        # 创建或更新记录
        if len(all_values) == 1:
            # 单值使用普通创建
            res = self.create_org_record(
                domain_name=domain_name,
                record=acme_txt,
                record_value=domain_dns_value,
                record_type="TXT",
                ttl=600,
            )
        else:
            # 多值使用UPSERT合并所有值
            res = self._upsert_dns_record_multi(
                domain_name=domain_name,
                record=acme_txt,
                record_values=all_values,
                record_type="TXT",
                ttl=600,
            )

        if not res.get("status"):
            raise HintException(res.get("msg") or "create dns record failed")

    def _make_change_multi(self, action: str, name: str, record_type: str,
                           ttl: int, values: list[str]) -> dict:
        """
        构造多值的 Change 字典
        用于一个记录名存储多个 TXT 值（ACME 多域名挑战）
        """
        record_type = self._ensure_upper(record_type)

        # 为 TXT 记录构建多个值
        resource_records = []
        for value in values:
            if record_type == "TXT":
                # TXT 记录需要带引号
                formatted_value = self._build_r53_value(record_type, value, -1)
                resource_records.append({"Value": formatted_value})
            else:
                resource_records.append({"Value": value})

        return {
            "Action": action,
            "ResourceRecordSet": {
                "Name": self._ensure_trailing_dot(name),
                "Type": record_type,
                "TTL": ttl,
                "ResourceRecords": resource_records,
            },
        }

    def _upsert_dns_record_multi(self, domain_name: str, record: str, record_values: list[str],
                                  record_type: str, ttl: int = 600) -> dict:
        """
        创建或更新 DNS 记录，支持多值（ACME 批处理）
        """
        domain_name = self._find_hosted_zone(domain_name) or domain_name
        try:
            zone_id = self._get_zone_id(domain_name)
            full_name = self._get_full_name(domain_name, record)

            ttl_val = 600 if int(ttl) == 1 else int(ttl)

            # 使用多值 Change
            change = self._make_change_multi("UPSERT", full_name, record_type, ttl_val, record_values)
            if self._change_record(zone_id, "UPSERT", [change]):
                return {"status": True, "msg": "success"}
            return {"status": False, "msg": "upsert dns record failed"}
        except HintException:
            raise
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def delete_dns_record(self, domain_name: str, domain_dns_value: str) -> None:
        domain_name = domain_name.lstrip("*.")
        _, _, acme_txt = extract_zone(domain_name)
        res = self.remove_record(
            domain_name, acme_txt, "TXT", record_value=domain_dns_value
        )
        if not res.get("status"):
            raise HintException(res.get("msg") or "delete dns record failed")

    # =============== 域名管理 ====================
    def get_domains(self, verify: bool = False) -> list:
        if self._hosted_zones_cache is not None and not verify:
            return self._hosted_zones_cache

        try:
            domains = []
            paginator = self.r53.get_paginator("list_hosted_zones")
            for page in paginator.paginate():
                for zone in page.get("HostedZones", []):
                    # 私有托管区
                    if zone.get("Config", {}).get("PrivateZone", False):
                        continue
                    if verify:
                        return []
                    name = zone.get("Name", "").strip().rstrip(".")
                    if name:
                        domains.append(name)
            result = sorted(set(domains))
            if not verify:
                self._hosted_zones_cache = result
            return result
        except self.ClientError as e:
            raise HintException(f"get domains error: {e}")
        except Exception as e:
            raise HintException(f"get domains error {e}")

    def get_dns_record(self, domain_name: str) -> list:
        zone_domain = self._find_hosted_zone(domain_name)
        if not zone_domain:
            raise HintException(f"Unable to find hosted zone for domain: {domain_name}")

        try:
            zone_id = self._get_zone_id(zone_domain)
            result = []

            paginator = self.r53.get_paginator("list_resource_record_sets")
            for page in paginator.paginate(HostedZoneId=zone_id):
                for rs in page.get("ResourceRecordSets", []):
                    r_type = rs.get("Type", "").upper()
                    r_name = rs.get("Name", "").strip().rstrip(".")
                    r_ttl = rs.get("TTL", 600)
                    host = self._normalize_host(zone_domain, r_name)
                    # 跳过 SOA 记录
                    if r_type == "SOA":
                        continue
                    # 跳过zone根 NS 记录
                    if r_type == "NS" and host == "@" and zone_domain in r_name:
                        continue
                    for rr in rs.get("ResourceRecords", []):
                        r_value = rr.get("Value", "")
                        record_value = self._parse_r53_value(r_type, r_value)
                        priority = self._parse_priority(r_type, r_value)
                        result.append({
                            "record": host,
                            "record_value": record_value,
                            "record_type": r_type,
                            "ttl": r_ttl,
                            "priority": priority,
                            "proxy": -1,
                        })

            return result
        except self.ClientError as e:
            raise HintException(f"get dns record error: {e}")
        except Exception as e:
            raise HintException(f"get dns record error {e}")

    def _get_acme_txt_record(self, domain_name: str, record: str) -> list:
        """
        获取 ACME record TXT 挑战记录（ACME 专用优化快速查找）
        """
        zone_domain = self._find_hosted_zone(domain_name)
        if not zone_domain:
            return []

        try:
            zone_id = self._get_zone_id(zone_domain)
            target_host = self._normalize_host(domain_name, record)
            full_name_with_dot = self._get_full_name(domain_name, record) + "."

            paginator = self.r53.get_paginator("list_resource_record_sets")
            for page in paginator.paginate(HostedZoneId=zone_id):
                for rs in page.get("ResourceRecordSets", []):
                    r_type = rs.get("Type", "").upper()
                    r_name = rs.get("Name", "")
                    if r_type != "TXT":
                        continue
                    # 标准化记录名进行比较
                    r_name_stripped = r_name.strip().rstrip(".")
                    if r_name_stripped != full_name_with_dot.rstrip("."):
                        continue
                    # 解析值并返回
                    r_ttl = rs.get("TTL", 600)
                    result = []
                    for rr in rs.get("ResourceRecords", []):
                        r_value = rr.get("Value", "")
                        record_value = self._parse_r53_value(r_type, r_value)
                        result.append({
                            "record": target_host,
                            "record_value": record_value,
                            "record_type": r_type,
                            "ttl": r_ttl,
                            "priority": -1,
                            "proxy": -1,
                        })
                    return result
            return []
        except Exception as e:
            public.print_log(f"get acme txt record error: {e}")
            return []

    def _change_record(self, zone_id: str, action: str, changes: list) -> bool:
        """执行 Record 变更, action: CREATE/DELETE/UPSERT"""
        try:
            self.r53.change_resource_record_sets(
                HostedZoneId=zone_id,
                ChangeBatch={
                    "Changes": changes,
                },
            )
            return True
        except self.ClientError as e:
            error_code = e.response.get("Error", {}).get("Code", "")
            error_msg = e.response.get("Error", {}).get("Message", "")
            # DELETE 时记录不存在视为成功
            if action == "DELETE" and error_code == "InvalidChangeBatch" and "but it does not exist" in error_msg:
                return True
            public.print_log(f"Route53 {action} error: {error_msg}")
            return False
        except Exception as e:
            public.print_log(f"Route53 {action} error: {e}")
            return False

    def create_org_record(self, domain_name, record, record_value, record_type, ttl=1, **kwargs):
        """
        创建 DNS 记录
        - 可合并类型存在时合并新值
        - 不可合并类型存在时直接返回失败, 避免创建表数据
        - 不存在时正常创建
        """
        domain_name = self._find_hosted_zone(domain_name) or domain_name
        try:
            zone_id = self._get_zone_id(domain_name)
            full_name = self._get_full_name(domain_name, record)
            record_type_upper = self._ensure_upper(str(record_type))

            ttl_val = 600 if int(ttl) == 1 else int(ttl)
            priority = kwargs.get("priority", -1)

            if record_type_upper in self._MULTI_VALUE_TYPES:
                # 可合并类型: 查询现有值并追加
                existing_values = []
                records = self.get_dns_record(domain_name)
                target_host = self._normalize_host(domain_name, record)
                for r in records:
                    if (self._normalize_host(domain_name, r.get("record")) == target_host and
                            self._ensure_upper(r.get("record_type", "")) == record_type_upper):
                        val = r.get("record_value", "")
                        if val and val not in existing_values:
                            existing_values.append(val)

                # 追加新值(去重)
                normalized_new_value = self._parse_r53_value(
                    record_type_upper, record_value
                ) if record_type_upper in ("TXT", "CAA") else record_value

                if normalized_new_value not in existing_values:
                    existing_values.append(normalized_new_value)

                if len(existing_values) == 1:
                    change = self._make_change("UPSERT", full_name, record_type_upper,
                                               ttl_val, existing_values[0], priority)
                else:
                    change = self._make_change_multi("UPSERT", full_name, record_type_upper,
                                                     ttl_val, existing_values)

                if self._change_record(zone_id, "UPSERT", [change]):
                    return {"status": True, "msg": "success"}
                return {"status": False, "msg": "upsert record failed"}
            else:
                # 不可合并类型: 检查是否存在
                records = self.get_dns_record(domain_name)
                target_host = self._normalize_host(domain_name, record)
                for r in records:
                    if (self._normalize_host(domain_name, r.get("record")) == target_host and
                            self._ensure_upper(r.get("record_type", "")) == record_type_upper):
                        # 已存在，返回失败, 避免添加到缓存表中
                        return {"status": False, "msg": "Record already exists."}

                # 不存在，正常创建
                change = self._make_change("UPSERT", full_name, record_type_upper,
                                           ttl_val, record_value, priority)
                if self._change_record(zone_id, "UPSERT", [change]):
                    return {"status": True, "msg": "success"}
                return {"status": False, "msg": "create record failed"}
        except HintException:
            raise
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def remove_record(self, domain_name, record, record_type="TXT", **kwargs) -> dict:
        """
        删除 DNS 记录
        多值类型+指定值: 仅移除该值, 保留其余
        其余情况: 删除全部匹配记录
        """
        domain_name = self._find_hosted_zone(domain_name) or domain_name
        try:
            zone_id = self._get_zone_id(domain_name)
            full_name = self._get_full_name(domain_name, record)
            record_type_upper = str(record_type).upper()

            # 查询所有记录
            records = self.get_dns_record(domain_name)
            target_host = self._normalize_host(domain_name, record)

            # 收集所有匹配的记录值
            matched_records = []
            for r in records:
                if (self._normalize_host(domain_name, r.get("record")) != target_host or
                        r.get("record_type", "").upper() != record_type_upper):
                    continue
                matched_records.append(r)

            if not matched_records:
                return {"status": True, "msg": "Dns Record not found."}

            provided_value = kwargs.get("record_value", "")
            ttl = matched_records[0].get("ttl", 600)
            priority = matched_records[0].get("priority", -1)

            # === 多值类型 + 指定了具体值 → 只删这一个值, 保留其余 ===
            if record_type_upper in self._MULTI_VALUE_TYPES and provided_value:
                # 标准化要删除的值
                compare_value = self._parse_r53_value(record_type_upper, provided_value)

                # 检查值是否存在
                if compare_value not in [r.get("record_value", "") for r in matched_records]:
                    return {"status": False, "msg": f"No matching record found with value: {provided_value}"}

                # 过滤出要保留的值
                remaining_values = [
                    r.get("record_value", "") for r in matched_records
                    if r.get("record_value", "") != compare_value
                ]

                if remaining_values:
                    # 还有剩余值 → UPSERT 保留
                    if len(remaining_values) == 1:
                        change = self._make_change("UPSERT", full_name, record_type_upper,
                                                   ttl, remaining_values[0], priority)
                    else:
                        change = self._make_change_multi("UPSERT", full_name, record_type_upper,
                                                         ttl, remaining_values)
                else:
                    # 没有剩余值 → DELETE 整个记录集
                    change = self._make_change_multi("DELETE", full_name, record_type_upper,
                                                     ttl, [r.get("record_value", "") for r in matched_records])

                if self._change_record(zone_id, change.get("Action"), [change]):
                    return {"status": True, "msg": "success"}
                return {"status": False, "msg": "delete record failed"}

            # === 其余情况: 删除全部匹配记录 ===
            change = self._make_change_multi("DELETE", full_name, record_type_upper, ttl,
                                              [r.get("record_value", "") for r in matched_records])
            if self._change_record(zone_id, "DELETE", [change]):
                return {"status": True, "msg": "success"}
            return {"status": False, "msg": "delete record failed"}
        except HintException:
            raise
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def update_record(self, domain_name, record: dict, new_record: dict, **kwargs):
        """
        更新 DNS 记录
        多值类型同 name+type 时: 在值列表中替换, 保留其他值
        其余情况: 先删后建或直接 UPSERT
        """
        domain_name = self._find_hosted_zone(domain_name) or domain_name
        try:
            old_full_name = self._get_full_name(domain_name, record.get("record"))
            new_full_name = self._get_full_name(domain_name, new_record.get("record"))
            old_record_type = record.get("record_type", "TXT")
            new_record_type = new_record.get("record_type", "TXT")
            old_record_type_upper = self._ensure_upper(old_record_type)
            new_record_type_upper = self._ensure_upper(new_record_type)

            ttl_val = int(new_record.get("ttl", 600))
            if ttl_val == 1:
                ttl_val = 600

            priority = new_record.get("priority", -1)

            # === 记录名或类型变化 → 先删后建 ===
            if old_full_name != new_full_name or old_record_type_upper != new_record_type_upper:
                del_res = self.remove_record(domain_name, record.get("record"), old_record_type)
                if not del_res.get("status"):
                    return {"status": False, "msg": f"delete old record failed: {del_res.get('msg')}"}

                create_change = self._make_change(
                    "UPSERT", new_full_name, new_record_type_upper,
                    ttl_val, new_record.get("record_value", ""), priority,
                )
                if self._change_record(self._get_zone_id(domain_name), "UPSERT", [create_change]):
                    return {"status": True, "msg": "success"}
                return {"status": False, "msg": "create new record failed"}

            # === name+type 相同 ===
            if new_record_type_upper in self._MULTI_VALUE_TYPES:
                # 多值类型: 在值列表中替换指定值, 保留其他值
                records = self.get_dns_record(domain_name)
                target_host = self._normalize_host(domain_name, record.get("record"))
                existing_values = [
                    r.get("record_value", "") for r in records
                    if (self._normalize_host(domain_name, r.get("record")) == target_host and
                        self._ensure_upper(r.get("record_type", "")) == new_record_type_upper)
                ]

                old_value = record.get("record_value", "")
                new_value = new_record.get("record_value", "")

                # 找到旧值并替换
                if old_value in existing_values:
                    idx = existing_values.index(old_value)
                    existing_values[idx] = new_value
                else:
                    # 旧值不在列表中, 检查是否只是格式差异(TXT引号)
                    normalized_old = self._parse_r53_value(new_record_type_upper, old_value)
                    found = False
                    for i, v in enumerate(existing_values):
                        if v == normalized_old:
                            existing_values[i] = new_value
                            found = True
                            break
                    if not found:
                        # 找不到旧值, 视为新增
                        if new_value not in existing_values:
                            existing_values.append(new_value)

                # 去重
                existing_values = list(dict.fromkeys(existing_values))

                if len(existing_values) == 1:
                    change = self._make_change("UPSERT", new_full_name, new_record_type_upper,
                                               ttl_val, existing_values[0], priority)
                else:
                    change = self._make_change_multi("UPSERT", new_full_name, new_record_type_upper,
                                                     ttl_val, existing_values)
            else:
                # 单值类型: 直接 UPSERT 覆盖
                change = self._make_change(
                    "UPSERT", new_full_name, new_record_type_upper,
                    ttl_val, new_record.get("record_value", ""), priority,
                )

            if self._change_record(self._get_zone_id(domain_name), "UPSERT", [change]):
                return {"status": True, "msg": "success"}
            return {"status": False, "msg": "update record failed"}

        except HintException:
            raise
        except Exception as e:
            return {"status": False, "msg": str(e)}

    def verify(self) -> Optional[bool]:
        try:
            self.get_domains(verify=True)
        except Exception as e:
            raise HintException(f"Verify fail, please check your AccessKeyId and SecretAccessKey: {e}")
        return True