# coding: utf-8
# -------------------------------------------------------------------
# aaPanel
# -------------------------------------------------------------------
# Copyright (c) 2014-2099 aaPanel(www.aapanel.com) All rights reserved.
# -------------------------------------------------------------------
# Author: aapanel
# -------------------------------------------------------------------
# ------------------------------
# aaDNS app
# ------------------------------
import json
import os
import random
import sys
import time
from dataclasses import dataclass, fields
from typing import Optional, Type, TypeVar, List

sys.path.insert(0, '/www/server/panel/class')
sys.path.insert(0, '/www/server/panel/class_v2')
import public
from ssl_dnsV2.model import DnsResolve
from ssl_dnsV2.helper import DnsParser
from ssl_dnsV2.conf import *
from public.exceptions import HintException
from ssl_domainModelV2.service import DomainValid, SyncService
from ssl_domainModelV2.model import DnsDomainProvider, dns_logger

PANEL_PATH = public.get_panel_path()

S = TypeVar("S", bound="Soa")

TIMEOUT = 1  # 用于验证ns超时



def backup_file(path: str, suffix: str = None) -> None:
    """创建 pdns的备份文件"""
    if not os.path.exists(path):
        return
    if suffix is None:
        if path == ZONES:
            suffix = "aabak"  # 主配置
        elif path.startswith(ZONES_DIR):
            suffix = "aadef"  # 区域文件
        else:
            suffix = "bak"
    backup_path = f"{path}_{suffix}"
    public.ExecShell(f"cp -a {path} {backup_path}")


def pdns_rollback(path: str) -> None:
    """根据 pdns文件路径回滚自己得备份文件"""
    if path.startswith(ZONES_DIR):
        suffix = "aadef"
    else:
        suffix = "aabak"
    backup_path = f"{path}_{suffix}"
    if os.path.exists(backup_path):
        public.ExecShell(f"cp -a {backup_path} {path}")


@dataclass(slots=True)
class Soa:
    domain: str
    nameserver: str
    admin_mail: str
    refresh: int
    retry: int
    expire: int
    minimum: int

    @classmethod
    def from_dict(cls: Type[S], data: dict) -> S:
        class_fields = {f.name for f in fields(cls)}
        filtered_data = {k: v for k, v in data.items() if k in class_fields}
        return cls(**filtered_data)


class Templater:
    @staticmethod
    def generate_zone(domain: str) -> str:
        template = """
zone "%s" IN {
        type master;
        file "/var/named/chroot/var/named/%s.zone";
        allow-update { none; };
};
""" % (domain, domain)
        public.writeFile(ZONES, template, "a+")
        return template

    @staticmethod
    def generate_record(domain: str, ns1: str, ns2: str, soa: str, ip: str, **kwargs) -> str:
        # 确保FQDN格式
        for k in [ns1, ns2, soa]:
            if not k.endswith("."):
                k += "."
        ttl = str(kwargs.get("ttl", "14400"))
        priority = str(kwargs.get("priority", "10"))
        r_type = "A" if ":" not in ip else "AAAA"
        serial = time.strftime("%Y%m%d", time.localtime()) + "01"

        template = f"""$TTL 86400
{domain}.      IN SOA  {soa}     admin.{domain}. (
                                        {serial}       ; serial
                                        3600      ; refresh
                                        1800      ; retry
                                        1209600      ; expire
                                        1800 )    ; minimum
{domain}.            86400     IN      NS        {ns1}
{domain}.            86400     IN      NS        {ns2}
{domain}.            {ttl}     IN      {r_type}        {ip}
{domain}.            {ttl}     IN      MX {priority}      mail.{domain}.
www             {ttl}     IN      {r_type}        {ip}
mail            {ttl}     IN      {r_type}        {ip}
ns1             {ttl}     IN      {r_type}        {ip}
ns2             {ttl}     IN      {r_type}        {ip}
"""
        public.writeFile(os.path.join(ZONES_DIR, f"{domain}.zone"), template)
        return template


class MailManager:
    def __init__(self, domain: str = None):
        self.domain = domain
        self.dns_manager = DnsManager()

    def _get_provider(self, provider: DnsDomainProvider = None) -> DnsDomainProvider:
        if not provider:
            provider = self.dns_manager.aapanel_dns_obj
            if provider:
                return provider
            raise HintException("DNS provider 'aaPanelDns' not found.")
        return provider

    def _get_records(self) -> list:
        return self.dns_manager.parser.get_zones_records(self.domain)

    def __append_spf_record(self, spf_record: dict, new_spf_ip: str) -> None:
        parts = spf_record["value"].strip('"').split()
        all_index = -1
        for i, part in enumerate(parts):
            if part.endswith("all"):
                all_index = i
                break
        if all_index != -1:
            parts.insert(all_index, f"+{new_spf_ip}")
        else:
            parts.append(f"+{new_spf_ip}")

        new_spf_value = " ".join(parts)
        self.dns_manager.update_record(
            domain=self.domain,
            name=spf_record["name"],
            type="TXT",
            value=spf_record["value"],
            new_record={
                "name": spf_record["name"],
                "type": "TXT",
                "ttl": spf_record["ttl"],
                "value": f'"{new_spf_value}"',
            }
        )

    def add_dmarc(self, policy: str = "none", provider: DnsDomainProvider = None, **kwargs) -> bool:
        """
        添加 DMARC 记录
        DMARC 策略 "none" (接受) "quarantine" (隔离) 或 "reject" (拒绝)
        """
        if policy not in ["none", "quarantine", "reject"]:
            raise HintException("Invalid DMARC policy specified.")

        for record in self._get_records():
            if record.get("name") == "_dmarc" and record.get("type") == "TXT":
                dns_logger(f"domain [{self.domain}] has DMARC record, skipping addition.")
                return True

        dmarc_value = f'"v=DMARC1; p={policy}; rua=mailto:admin@{self.domain}"'
        body = {
            "domain": self.domain,
            "record": "_dmarc",
            "record_type": "TXT",
            "record_value": dmarc_value,
            "ttl": kwargs.get("ttl", 600),
            "ps": "Auto DMARC Record",
        }
        self._get_provider(provider).model_create_dns_record(body)
        return True

    def add_spf(self, provider: DnsDomainProvider = None, **kwargs) -> bool:
        records = self._get_records()
        server_ip = public.GetLocalIp()
        new_spf_ip = f"ip4:{server_ip}" if DomainValid.is_ip4(server_ip) else f"ip6:{server_ip}"
        spf_record = None
        for record in records:
            if all([
                record.get("type") == "TXT",
                record.get("name") == "@" or record.get("name") == self.domain,
                record.get("value", "").startswith("v=spf1") or record.get("value", "").startswith('"v=spf1'),
            ]):
                if spf_record is not None:
                    dns_logger(
                        f"domain [{self.domain}] has multiple SPF records, it will auto fix it,"
                        f" keep the first one."
                    )
                    # 多个spf记录异常
                    self.dns_manager.delete_record(
                        domain=self.domain, name=record["name"], type="TXT", value=record["value"]
                    )
                else:
                    spf_record = record
        # 存在spf
        if spf_record:
            if new_spf_ip not in spf_record.get("value", ""):
                self.__append_spf_record(spf_record, new_spf_ip)
                dns_logger(
                    f"domain [{self.domain}] SPF record updated to include server IP: [{server_ip}]."
                    f"Current SPF: {spf_record.get('value', '')}"
                )
                return True
            else:
                dns_logger(f"domain [{self.domain}] has SPF record with server IP, skipping addition.")
                return True
        # 不存在SPF记录
        body = {
            "domain": self.domain,
            "record": "@",
            "record_type": "TXT",
            "record_value": f"v=spf1 +mx +a +{new_spf_ip} -all",
            "ttl": kwargs.get("ttl", 600),
            "ps": "Auto SPF Record",
        }
        self._get_provider(provider).model_create_dns_record(body)

        return True

    def add_dkim(self, provider: DnsDomainProvider = None, **kwargs) -> bool:
        dkim_name = "default._domainkey"
        records = self._get_records()
        for record in records:
            if record.get("name") == dkim_name and record.get("type") == "TXT":
                dns_logger(f"domain [{self.domain}] has DKIM record, skipping addition.")
                return True
        try:
            plugin_path = "/www/server/panel/plugin/mail_sys"
            if os.path.exists(plugin_path) and plugin_path not in sys.path:
                sys.path.insert(0, plugin_path)
            from plugin.mail_sys.mail_sys_main import mail_sys_main
            dkim_value = mail_sys_main()._get_dkim_value(self.domain)
            if dkim_value == "":
                mail_sys_main().set_rspamd_dkim_key(self.domain)
                dkim_value = mail_sys_main()._get_dkim_value(self.domain)
        except ImportError:
            dns_logger(f"domain [{self.domain}] Mail System Plugin ImportError, not installed...")
            raise HintException("Mail System Plugin ImportError, not installed.")
        except Exception:
            dns_logger(f"domain [{self.domain}] Failed to retrieve DKIM value from mail system.")
            raise HintException("Failed to retrieve DKIM value from mail system.")

        if not dkim_value:
            dns_logger(f"domain [{self.domain}] DKIM value is empty, cannot add record.")
            raise HintException("DKIM value is empty, cannot add record.")

        body = {
            "domain": self.domain,
            "record": dkim_name,
            "record_type": "TXT",
            "record_value": f'"{dkim_value}"',
            "ttl": kwargs.get("ttl", 600),
            "ps": "Auto SPF Record",
        }
        self._get_provider(provider).model_create_dns_record(body)
        return True


class DnsManager:
    def __init__(self):
        self.parser = DnsParser()
        self.config = self.parser.config

    @property
    def aapanel_dns_obj(self) -> Optional[DnsDomainProvider]:
        return DnsDomainProvider.objects.filter(name="aaPanelDns").first()

    def _quotes(self, s: str) -> bool:
        if not s or not isinstance(s, str):
            return False
        if s.startswith('"') and s.endswith('"'):
            return True
        return False

    def _get_glb_ns(self) -> List[tuple]:
        """获取公共NS服务器, 打乱"""
        shuffled_servers = [x for x in PUBLIC_SERVER]
        random.shuffle(shuffled_servers)
        return shuffled_servers

    def makesuer_port(self) -> None:
        if not public.S("firewall_new").where(
                "ports = ? AND protocol = ?", ("53", "tcp/udp")
        ).count():
            try:
                from firewallModelV2.comModel import main as firewall_main
                get = public.dict_obj()
                get.protocol = "all"
                get.port = "53"
                get.choose = "all"
                get.types = "accept"
                get.strategy = "accept"
                get.chain = "INPUT"
                get.brief = "DNS Service Port"
                get.operation = "add"
                firewall_main().set_port_rule(get)
            except Exception as e:
                dns_logger("Add Firewall Port 53 Error: {}".format(e))
                public.print_log("add firewall port 53 error: {}".format(e))

    def query_dns(self, q_name: str, record_type: str, ns_server: Optional[list] = None, time_out: int = None) -> list:
        """
        DNS查询
        :param q_name: 要查询的名称
        :param record_type: 记录类型
        :param ns_server: 要使用的DNS服务器列表。如果为None，则使用系统默认
        :param time_out: 超时 s
        :return: 记录值字符串的列表
        :raises: HintException 如果查询失败
        """
        try:
            import dns.resolver
        except ImportError:
            dns_logger("dnspython ImportError, not installed...")
            public.ExecShell("btpip install dnspython")
            public.print_log("dnspython not installed, skipping DNS query.")
            return []

        resolver = dns.resolver.Resolver()
        if time_out:
            resolver.timeout = time_out
            resolver.lifetime = time_out
        if ns_server:
            resolver.nameservers = ns_server
        time.sleep(0.5)

        try:
            answers = resolver.resolve(q_name, record_type)
            found_records = []
            for rdata in answers:
                record_str = str(rdata)
                # 格式化
                if record_type == "TXT" and self._quotes(record_str):
                    record_str = record_str[1:-1]
                elif record_type == "MX":
                    record_str = record_str.split(" ", 1)[1]
                found_records.append(record_str.rstrip('.'))
            return found_records
        except (dns.resolver.Timeout, dns.resolver.LifetimeTimeout):
            raise HintException(f"DNS query for {q_name} ({record_type}) timed out.")
        except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN, dns.resolver.NoNameservers):
            # 没有找到记录
            return []
        except Exception as e:
            raise HintException(f"DNS query for {q_name} ({record_type}) failed: {e}")

    def validate_with_resolver(self, action: str, domain: str, **kwargs) -> bool:
        # 本地校验dns记录
        if not action:
            return False
        record_data = kwargs
        if action == "update" and "new_record" in kwargs:
            record_data = kwargs["new_record"]

        record_type = record_data.get("type", "A").upper()
        name = record_data.get("name", "@")
        value_to_check = record_data.get("value")

        if name == "@":
            q_name = domain
        elif name.endswith("."):
            q_name = name
        else:
            q_name = f"{name}.{domain}"

        found_records = self.query_dns(
            q_name=q_name, record_type=record_type, ns_server=["127.0.0.1"]
        )
        time.sleep(1)

        if not found_records:
            if action in ["create", "update"]:  # 对于创建||更新，是失败的
                raise HintException(
                    f"Validation failed: Record {q_name} ({record_type}) not found after {action}."
                )
            if action == "delete":  # 对于删除，是成功的
                return True
        # 去引号, 去尾点
        value_to_check_cleaned = value_to_check.strip('"').rstrip(".")

        if action in ["create", "update"]:
            if value_to_check_cleaned not in found_records:
                raise HintException(
                    f"Validation failed: Record {q_name} ({record_type})"
                    f" with value '{value_to_check}' not found after {action}."
                )
            return True

        if action == "delete":
            if value_to_check_cleaned in found_records:
                raise HintException(
                    f"Validation failed: Record {q_name} ({record_type})"
                    f" with value '{value_to_check}' still exists after delete."
                )
            return True

        return False

    def _validate_NS_A(self, ns_list: List[str], time_out: int = None) -> bool:
        """备用尝试直接验证A记录方法: 直接查询ns1和ns2的A记录"""
        time_out = int(time_out) if time_out else TIMEOUT
        server_ip = public.GetLocalIp()
        try:
            for n in ns_list:
                ns_ip = self.query_dns(
                    q_name=n, record_type="A", time_out=time_out
                )
                if server_ip not in ns_ip:
                    return False
            return True
        except Exception as e:
            public.print_log(f"Warning: DNS query to NS A record failed: {e}.")
            return False

    def _validate_NS(self, domain: str, ns_list: List[str], addr_list: list, time_out: int = None) -> bool:
        try:
            time_out = int(time_out) if time_out else TIMEOUT
            found_ns = self.query_dns(
                q_name=domain, record_type="NS", ns_server=addr_list, time_out=time_out
            )
            if found_ns:
                ns_set = {ns.rstrip(".").lower() for ns in found_ns}
                # 是否子集
                if set(ns_list).issubset(ns_set):
                    return True
            return False
        except Exception as e:
            public.print_log(f"Warning: DNS query to NS record failed: {e}.")
            return False

    def _validate_any_ns_msg(self, domain: str, ns_list: List[str], serv_name: str, addr_list: list) -> Optional[str]:
        """单次验证域名的权威"""
        # NS 阶段
        if self._validate_NS(domain, ns_list, addr_list):
            return (f"Validate NameServer Success: NS records are correctly set. "
                    f"Found in Global Public DNS '{serv_name}' {addr_list}")

        public.print_log(
            f"Warning: NS for {domain} are Not Found! "
            f"which do not match provided [{', '.join(ns_list)}]. Proceeding to fallback validation."
        )

        # fallback 尝试直接解析NS主机的A记录
        if self._validate_NS_A(ns_list):
            return (
                f"Validate NameServer Success: NS A records point to server IP."
                f"Found in Global Public DNS '{serv_name}' {addr_list}"
            )

        return None

    def _any_ns_hit(self, domain: str, ns_list: List[str]) -> str:
        """
        any one策略
        验证域名的权威NS记录。
        1. 优先查询域名的权威NS记录是否已在公网指向ns1/ns2。
        2. 如果查询失败或不匹配（例如新域名、未设置胶水记录），则回退到备用验证。
        3. 备用验证：直接查询ns1和ns2的A记录，看它们是否指向本机IP。
        成功返回str, 失败抛异常。
        """
        ns_list = [x.rstrip(".").lower() for x in ns_list if x]
        for server_name, addr_list in self._get_glb_ns():
            try:
                any_tips = self._validate_any_ns_msg(
                    domain, ns_list, server_name, addr_list
                )
                # 返回成功信息tips
                if any_tips and isinstance(any_tips, str):
                    body = {
                        "ns_resolve": 0,
                        "a_resolve": 0,
                        "tips": any_tips,
                    }
                    if "NS records" in any_tips:
                        body["ns_resolve"] = 1
                        body["a_resolve"] = 0

                    elif "NS A records" in any_tips:
                        body["ns_resolve"] = 0
                        body["a_resolve"] = 1
                    DnsResolve.update_or_create(domain, **body)
                    return any_tips
                else:
                    continue
            except Exception as e:
                public.print_log(e)

        # for loop end, all failed
        raise HintException("Validate NameServer Failed: NS records are not correctly set.")

    def __read_zone_lines(self, zone_file: str) -> list:
        """读取zone文件返回行列表"""
        content = public.readFile(zone_file) or ""
        if not content:
            raise HintException(f"Zone file [{zone_file}] is empty.")
        # 移除末尾可能存在的空行
        lines = content.rstrip("\n").split("\n") if content else []
        if not lines:
            raise HintException(f"Zone file [{zone_file}] is empty.")
        return lines

    def _update_soa(self, lines: list, soa_obj: Optional[Soa] = None) -> bool:
        """更新SOA 必然更新序列号, 如果带了 soa_obj 参数必须全提供"""
        serial_update = False
        other_update = True if soa_obj is not None else False

        ns_admin_mail_changed = False
        refresh_changed = False
        retry_changed = False
        expire_changed = False
        minimum_changed = False

        for i, line in enumerate(lines):
            if "IN SOA" in line:  # SOA 行开始
                # ns 和 admin mail
                if soa_obj and soa_obj.nameserver and soa_obj.admin_mail:
                    new_soa_line = (
                        f"{soa_obj.domain}.      IN SOA  "
                        f"{soa_obj.nameserver}.     "
                        f"{soa_obj.admin_mail}. ("
                    )
                    lines[i] = new_soa_line
                    ns_admin_mail_changed = True
                # 剩余参数
                go_on = lines[i:]
                for j, soa_line in enumerate(go_on):
                    # '; serial' 注释结尾, (看模板)
                    if ";" in soa_line and "serial" in soa_line:
                        serial_str = soa_line.split(';')[0].strip()
                        if serial_str.isdigit():
                            today_prefix = time.strftime("%Y%m%d", time.localtime())
                            if serial_str.startswith(today_prefix):
                                new_serial = str(int(serial_str) + 1)
                            else:
                                new_serial = today_prefix + "01"
                            # SOA 序列号变更
                            lines[i + j] = soa_line.replace(serial_str, new_serial, 1)
                            serial_update = True

                    elif "; refresh" in soa_line and soa_obj and soa_obj.refresh:
                        lines[i + j] = (f"                                        "
                                        f"{soa_obj.refresh}      ; refresh")
                        refresh_changed = True

                    elif "; retry" in soa_line and soa_obj and soa_obj.retry:
                        lines[i + j] = (f"                                        "
                                        f"{soa_obj.retry}      ; retry")
                        retry_changed = True

                    elif "; expire" in soa_line and soa_obj and soa_obj.expire:
                        lines[i + j] = (f"                                        "
                                        f"{soa_obj.expire}      ; expire")
                        expire_changed = True

                    elif "; minimum" in soa_line and soa_obj and soa_obj.minimum:
                        lines[i + j] = (f"                                        "
                                        f"{soa_obj.minimum} )    ; minimum")
                        minimum_changed = True

                break

        if other_update:
            if not (ns_admin_mail_changed and refresh_changed and retry_changed
                    and expire_changed and minimum_changed):
                return False

        if serial_update:
            return True

        return False

    def _build_record_line(self, domain: str, **kwargs) -> str:
        """构建DNS记录行"""

        def _get_params(kwargs):
            name = kwargs.get("name", "@")
            ttl = kwargs.get("ttl", "600")
            ttl = "600" if int(ttl) == 1 else str(ttl)
            record_type = kwargs.get("type", "A").upper()
            value = kwargs.get("value")
            priority = kwargs.get("priority", -1),
            if not value:
                raise HintException("value is required!")
            return name, record_type, ttl, value, priority

        name, record_type, ttl, value, priority = _get_params(kwargs)
        if "new_record" in kwargs:
            name, record_type, ttl, value, priority = _get_params(kwargs["new_record"])

        # === 仅校验 ===
        if record_type == "A" and not DomainValid.is_ip4(value):
            raise HintException(f"Invalid A record value: {value}")

        elif record_type == "AAAA" and not DomainValid.is_ip6(value):
            raise HintException(f"Invalid AAAA record value: {value}")

        elif record_type in ["CNAME", "NS"] and not DomainValid.is_valid_domain(value):
            raise HintException(f"Invalid {record_type} record value: {value}")

        elif record_type == "CAA":
            parts = value.split(None, 2)
            if len(parts) != 3:
                raise HintException(f"Invalid CAA record format: {value}")
            flags, tag, ca_value = parts
            if not flags.isdigit() or not (0 <= int(flags) <= 255):
                raise HintException(f"Invalid CAA flags: {flags}. Must be 0-255.")
            if tag not in ["issue", "issuewild", "iodef"]:
                raise HintException(f"Invalid CAA tag: {tag}. Must be 'issue', 'issuewild', or 'iodef'.")
            ca_value = ca_value.strip('"')
            if tag == "iodef":
                if not (ca_value.startswith("mailto:") or DomainValid.is_valid_domain(ca_value)):
                    raise HintException(f"Invalid CAA iodef value: {ca_value}")
            elif not DomainValid.is_valid_domain(ca_value):
                raise HintException(f"Invalid CAA domain value: {ca_value}")

        elif record_type == "TXT":
            # TXT 记录值如果包含空格，使用引号包裹
            if " " in value and not self._quotes(value):
                value = f'"{value}"'

        # === 参数追加尾部作为 整体value 写入 conf ===
        elif record_type == "SRV":
            priority = kwargs.get("priority", "10")
            weight = kwargs.get("weight", "5")
            port = kwargs.get("port")
            if not port:
                raise HintException("SRV record requires a 'port'.")
            if not all(str(p).isdigit() for p in [priority, weight, port]):
                raise HintException("SRV priority, weight, and port must be integers.")
            if not DomainValid.is_valid_domain(value):
                raise HintException(f"Invalid SRV target domain: {value}")
            value = f"{priority} {weight} {port} {value}"

        record_name = name
        if record_type == "MX":
            if not DomainValid.is_valid_domain(value):
                raise HintException(f"Invalid MX record value: {value}")
            # 如果value不是完全限定域名FQDN(不以.结尾)，则补充主域名
            if not value.endswith("."):
                value = f"{value}.{domain}."

            if record_name != "@" and not record_name.endswith("."):
                record_name = f"{record_name}.{domain}"
            # record_type = f"{record_type} {kwargs.get("priority", "10")}"
            value = f"{kwargs.get('priority', '10')} {value}"

        # 构造兼容旧格式
        return f"{record_name}\t{ttl}\tIN\t{record_type}\t{value}"

    def _find_record_line_index(self, lines: list, **kwargs) -> Optional[int]:
        """查找DNS记录行索引"""
        name = kwargs.get("name", "@")
        record_type = kwargs.get("type", "A").upper()
        value = kwargs.get("value")
        # 剥离引号
        if record_type in ["TXT", "CAA"] and value and self._quotes(value):
            value = value[1:-1]

        for i, line in enumerate(lines):
            match = record_pattern.match(line)
            if match:
                r_name, _, _, r_type, r_value_raw = match.groups()
                r_type = r_type.upper()
                r_value = r_value_raw.strip()
                # todo 更多类型的特殊处理
                if r_type in ["TXT", "CAA"]:
                    if self._quotes(r_value):
                        r_value = r_value[1:-1]  # 剥离引号
                    else:  # 如果没有引号，可能存在的注释
                        r_value = r_value.split(';', 1)[0].strip()

                elif r_type == "MX":
                    # MX解析后格式 "优先级 目标主机"
                    r_value = r_value.split(";", 1)[0].strip()
                    try:
                        _, mx_host = r_value.split(None, 1)
                        r_value = mx_host.rstrip(".")
                    except (ValueError, IndexError):
                        pass

                # todo 统一
                if r_name == name and r_type.upper() == record_type and r_value == value:
                    return i
        return None

    def _modify_record(self, domain: str, action: str, **kwargs) -> bool:
        # C, U, D
        zone_file = os.path.join(ZONES_DIR, f"{domain}.zone")
        if not os.path.exists(zone_file):
            raise HintException("Zone file not found!")
        # 开事务
        backup_file(zone_file)
        try:
            lines = self.__read_zone_lines(zone_file)
            # 更新SOA序列号
            if not self._update_soa(lines):
                raise HintException("SOA record not found, cannot update serial number.")
            modify = False
            line_index = self._find_record_line_index(lines, **kwargs)
            if action.lower() == "create":
                if line_index is not None:
                    public.print_log("Record already exists, skipping creation.")
                    return True

                new_record_line = self._build_record_line(domain, **kwargs)
                lines.append(new_record_line)
                modify = True

            elif action.lower() in ["update", "delete"]:
                if line_index is None:
                    raise HintException("Record not found!")

                if action.lower() == "delete":
                    lines.pop(line_index)
                    modify = True
                elif action.lower() == "update":
                    updated_line = self._build_record_line(domain, **kwargs)
                    lines[line_index] = updated_line
                    modify = True

            if modify:
                public.writeFile(zone_file, "\n".join(lines) + "\n")
                return True

            return False
        except Exception as e:
            pdns_rollback(zone_file)
            import traceback
            public.print_log(traceback.format_exc())
            raise HintException(e)

    def __apply_and_validate_change(self, action: str, domain: str, **kwargs) -> None:
        """配置变动和验证管理"""
        zone_file = os.path.join(ZONES_DIR, f"{domain}.zone")
        try:
            if domain not in self.parser.get_zones():
                raise HintException("Domian Not Found!")
            if not self._modify_record(domain, action, **kwargs):
                return
            self.reload_service()
            self.validate_with_resolver(action, domain, **kwargs)
            # 校验后对区域文件进行备份稳定版本
            backup_file(zone_file)
        except Exception as e:
            import traceback
            public.print_log(traceback.format_exc())
            pdns_rollback(zone_file)
            self.reload_service()
            raise HintException(f"Record {action} failed and was rolled back: {e}")

    # ================= script ====================

    def _generate_auth_body(self, domain: str, ns_res: list, a_res: list, shuffled_servers: list) -> dict:
        body = {
            "domain": domain,
            "ns_resolve": 0,
            "a_resolve": 0,
            "tips": "",
        }
        if not shuffled_servers:
            return body

        ns_rate = len(ns_res) / len(shuffled_servers)
        a_rate = len(a_res) / len(shuffled_servers)
        threshold = 0.3
        ns_ok = ns_rate >= threshold
        a_ok = a_rate >= threshold
        body["ns_resolve"] = 1 if ns_ok else 0
        body["a_resolve"] = 1 if a_ok else 0

        tips_parts = []
        # 前缀
        if ns_ok and a_ok:
            tips_parts.append("Validate NameServer Success: ")
        else:
            tips_parts.append("Validate NameServer Failed: ")
        # NS
        if ns_ok:
            tips_parts.append("NS records are correctly set.")
        else:
            tips_parts.append("NS records are not correctly set.")
        # A
        if a_ok:
            tips_parts.append("NS A records point to server IP.")
        else:
            tips_parts.append("NS A records do not point to server IP.")

        body["tips"] = " ".join(tips_parts)
        return body

    def builtin_dns_checker(self, dns_obj: Optional[DnsDomainProvider] = None) -> None:
        """验证域名的权威NS记录, 用于任务"""
        obj = self.aapanel_dns_obj if not dns_obj else dns_obj
        if os.path.exists(DNS_AUTH_LOCK) or not obj or obj.status == 0 or len(obj.domains) == 0:
            return
        with open(DNS_AUTH_LOCK, "w") as f:
            f.write("0")
        try:
            SyncService(obj.id).process(nohup=True)
            for domain in obj.domains:
                try:
                    ns_list = [
                        x.get("value", "").rstrip(".").lower() for x in self.parser.get_zones_records(domain)
                        if x.get("type", "").upper() == "NS"
                    ]
                    shuffled_servers = self._get_glb_ns()
                    ns_res = []
                    a_res = []
                    for server_name, addr_list in shuffled_servers:
                        try:
                            if self._validate_NS(domain, ns_list, addr_list, time_out=5):
                                ns_res.append(server_name)
                            if self._validate_NS_A(ns_list, time_out=5):
                                a_res.append(server_name)
                        except Exception:
                            continue

                    body = self._generate_auth_body(
                        domain, ns_res, a_res, shuffled_servers
                    )
                    DnsResolve.update_or_create(**body)
                    dns_logger(f"domain [{domain}] dns auth : {body.get('tips', 'Msg Not Found')}")
                except Exception as e:
                    public.print_log(f"builtin_dns_auth error : {e}")
                    continue
        except Exception as e:
            public.print_log(f"builtin_dns_auth outer error : {e}")
        finally:
            public.ExecShell(f"rm -f {DNS_AUTH_LOCK}")

    # ================= public ====================

    def change_service_status(self, service_name: str = "pdns", status: str = "restart") -> bool:
        if service_name == "pdns":
            service_packname = self.config.pdns_paths['service_name']
        elif service_name == "bind":
            service_packname = self.config.bind_paths['service_name']
        else:
            raise HintException("Unsupported service!")

        if status not in ["stop", "restart", "reload", "start"]:
            raise HintException("Invalid action specified!")

        if status != "stop":
            status = "restart"
            self.makesuer_port()
            if service_name == "pdns":  # bind 互斥
                stop_name = self.config.bind_paths['service_name']
            else:  # pdns 互斥
                stop_name = self.config.pdns_paths['service_name']
            a0, e0 = public.ExecShell(f"ps -ef|grep {stop_name}|grep -v grep")
            if a0:  # 关闭互斥服务
                public.ExecShell(f"systemctl stop {stop_name}")

        _, e = public.ExecShell(f"systemctl {status} {service_packname}")
        if e:
            raise HintException(f"{status} {service_name} service failed error: {e}")
        # 变更账号状态
        provider = self.aapanel_dns_obj
        if provider:
            status_map = {"restart": 1, "stop": 0}
            provider.status = status_map[status]
            provider.save()
        return True

    def reload_service(self, service_name: str = "pdns") -> bool:
        return self.change_service_status(service_name, "reload")

    def add_zone(self, domain: str, ns1: str, ns2: str, soa: str, ip: str = "127.0.0.1") -> Optional[str]:
        domain = domain.strip().rstrip(".")
        if domain in self.parser.get_zones():
            dns_logger(f"Add Zone Failed: zone [{domain}] Already Exists")
            raise HintException("Zone Already Exists!")

        # 权威解析
        try:
            # add zone 会更新创建首次解析记录
            ns_res_msg = str(self._any_ns_hit(domain, [ns1, ns2]))
        except HintException as he:
            ns_res_msg = f"\nHowever, {str(he)}"
            DnsResolve.update_or_create(domain, **{"ns_resolve": 0, "a_resolve": 0, "tips": ns_res_msg})
        except Exception as e:
            ns_res_msg = (f"\nHowever, An error occurred during NS validation: {e}."
                          f" Your DNS Resolution May Not be Active in the internet.")
            DnsResolve.update_or_create(domain, **{"ns_resolve": 0, "a_resolve": 0, "tips": ns_res_msg})

        zone_file = os.path.join(ZONES_DIR, f"{domain}.zone")
        backup_file(ZONES)  # 备份主配置文件
        try:
            Templater.generate_zone(domain)  # 追加zone 配置
            Templater.generate_record(domain, ns1, ns2, soa, ip)  # 生成zone区域文件
            self.reload_service()
            result_msg = f"Zone [{domain}] Added Successfully. "
            if ns_res_msg and isinstance(ns_res_msg, str):
                result_msg += ns_res_msg
            dns_logger(result_msg)
            return result_msg
        except Exception as e:
            pdns_rollback(ZONES)
            if os.path.exists(zone_file):
                try:
                    os.remove(zone_file)  # 移除异常的区域文件
                    os.remove(f"{zone_file}_aadef")  # 移除其备份
                except:
                    pass
            self.reload_service()
            dns_logger(f"Add Zone Failed: {e}")
            if isinstance(e, HintException):
                raise e
            raise HintException("Add Zone Failed Error: {}".format(e))
        finally:
            # 同步DNS服务
            provider = self.aapanel_dns_obj
            if provider:
                sync = SyncService(provider.id)
                sync.force = True
                sync.sync_dns_domains()

    def delete_zone(self, domain: str) -> Optional[str]:
        zone_file = os.path.join(ZONES_DIR, f"{domain}.zone")
        # 备份主配置文件和区域文件
        backup_file(ZONES)
        backup_file(zone_file)
        try:
            zones_content = public.readFile(ZONES) or ""
            lines = zones_content.splitlines()
            new_lines: list = []
            in_block_to_delete = False
            brace_count = 0
            for line in lines:
                stripped_line = line.strip()
                if stripped_line.startswith(f'zone "{domain}"'):
                    in_block_to_delete = True
                if in_block_to_delete:  # 配对大括号
                    brace_count += line.count("{")
                    brace_count -= line.count("}")
                    if brace_count <= 0:
                        in_block_to_delete = False
                    continue
                # 只保留一个换行
                if not stripped_line and new_lines and not new_lines[-1].strip():
                    continue
                new_lines.append(line)

            # 只保留一个换行
            while new_lines and not new_lines[-1].strip():
                new_lines.pop()
            final_content = "\n".join(new_lines)
            if final_content:
                final_content += "\n"

            public.writeFile(ZONES, final_content)
            try:
                os.remove(zone_file)
                os.remove(zone_file + "_aadef")
            except Exception as ex:
                public.print_log(f"Error removing zone file {zone_file}: {ex}")
            self.reload_service()
            msg = f"Zone [{domain}] Deleted Successfully."
            dns_logger(msg)
            DnsResolve.objects.filter(domain=domain).delete()
            return msg
        except Exception as e:
            import traceback
            public.print_log(traceback.format_exc())
            pdns_rollback(ZONES)
            pdns_rollback(zone_file)
            self.reload_service()
            dns_logger(f"Delete Zone Failed: {e}")
            raise HintException("Delete zone failed error: {}".format(e))
        finally:
            # 同步DNS服务
            provider = self.aapanel_dns_obj
            if provider:
                sync = SyncService(provider.id)
                sync.force = True
                sync.sync_dns_domains()

    def add_record(self, domain: str, **kwargs) -> bool:
        self.__apply_and_validate_change("create", domain, **kwargs)
        return True

    def delete_record(self, domain: str, **kwargs) -> bool:
        self.__apply_and_validate_change("delete", domain, **kwargs)
        return True

    def update_record(self, domain: str, **kwargs) -> bool:
        if not kwargs.get("new_record"):
            raise HintException("update record is required for update operation.")
        self.__apply_and_validate_change("update", domain, **kwargs)
        return True

    def get_domains(self) -> list:
        return self.parser.get_zones() or []

    def get_default_nameserver(self) -> dict:
        try:
            if os.path.exists(aaDNS_CONF):
                ns = public.readFile(aaDNS_CONF)
                return json.loads(ns) if ns else {}
            return {}
        except Exception as e:
            public.print_log("Error reading nameserver config: {}".format(e))
            return {}

    def set_default_nameserver(self, n1: str, n2: str) -> bool:
        for k in [n1, n2]:
            if not k.endswith("."):
                k += "."
        default = {"NS1": n1, "NS2": n2}
        self.config.ns_server = default
        public.writeFile(aaDNS_CONF, json.dumps(default))
        dns_logger("Default Nameserver Set To: {}".format(default))
        return True

    def get_soa(self, domain: str) -> dict:
        if not domain:
            return {}
        for record in self.parser.get_zones_records(domain=domain, witSOA=True):
            if record.get("type") == "SOA":
                return {
                    "nameserver": record.get("nameserver"),
                    "admin_mail": record.get("admin_mail"),
                    "serial": record.get("serial"),
                    "refresh": record.get("refresh"),
                    "retry": record.get("retry"),
                    "expire": record.get("expire"),
                    "minimum": record.get("minimum"),
                }
        return {}

    def set_soa(self, **kwargs) -> bool:
        soa_obj = Soa.from_dict(kwargs)
        zone_file = os.path.join(ZONES_DIR, f"{soa_obj.domain}.zone")
        if not os.path.exists(zone_file):
            raise HintException(f"Zone file for domain '{soa_obj.domain}' not found!")
        backup_file(zone_file)
        try:
            lines = self.__read_zone_lines(zone_file)
            # 更新整个SOA
            if not self._update_soa(lines, soa_obj):
                raise HintException("SOA record not found, cannot update serial number.")
            public.writeFile(zone_file, "\n".join(lines) + "\n")
        except Exception as e:
            pdns_rollback(zone_file)
            import traceback
            public.print_log(traceback.format_exc())
            raise HintException(f"Set Soa Failed: {e}")
        return True

    def get_logger(self, p: int = 1, limit: int = 20, search: str = None) -> dict:
        if search is None:
            sql = ("type = ?", ("DnsSSLManager",))
            count = public.S("logs").where(*sql).count()
        else:
            sql = ("type = ? AND log LIKE ?", ("DnsSSLManager", f"%{search}%"))
            count = public.S("logs").where(*sql).count()
        logs = public.S("logs").where(*sql).order("id", "DESC").limit(limit, (p - 1) * limit).field(
            "log, addtime"
        ).select()
        length = 150

        if logs and isinstance(logs, list):
            for log_entry in logs:
                if "log" in log_entry and len(log_entry["log"]) > length:
                    original_log = log_entry["log"]
                    chunks = [original_log[i:i + length] for i in range(0, len(original_log), length)]
                    log_entry["log"] = "\n".join(chunks)

        return {"data": logs or [], "count": count or 0}

    def clear_logger(self) -> bool:
        public.S("logs").where("type = ?", ("DnsSSLManager",)).delete()
        return True


if __name__ == '__main__':
    DnsManager().builtin_dns_checker()
