Refactor DNS management logic in main.py to track current IP records and streamline updates. Remove unused update interval variables from .env.example and config.py. Update Dockerfile to set timezone to Asia/Shanghai. Adjust scheduling in main.py to check for updates every 5 minutes.

2025-07-18 05:42:08 +08:00 · 2025-01-12 04:40:18 +08:00 · 2025-01-12 04:40:18 +08:00 · b1d3576763
commit b1d3576763
parent 831b4cd4f7
4 changed files with 122 additions and 72 deletions
--- a/.env.example
+++ b/.env.example
@ -10,7 +10,6 @@ DOMAIN_1=example1.com
 SUB_DOMAIN_1=@           # 子域名，@ 表示根域名
 REMARK_1=优选IP          # 记录备注
 TTL_1=600               # TTL值（秒）
-UPDATE_INTERVAL_1=15     # 更新间隔（分钟）
 IPV4_ENABLED_1=true     # 是否启用IPv4记录
 IPV6_ENABLED_1=true     # 是否启用IPv6记录
 ENABLED_1=true          # 是否启用此域名配置
@ -20,7 +19,6 @@ DOMAIN_2=example2.com
 SUB_DOMAIN_2=www
 REMARK_2=优选IP
 TTL_2=600
-UPDATE_INTERVAL_2=15
 IPV4_ENABLED_2=true
 IPV6_ENABLED_2=true
 ENABLED_2=true
@ -30,7 +28,6 @@ DOMAIN_3=example3.com
 SUB_DOMAIN_3=*
 REMARK_3=CloudFlare优选  # 记录备注
 TTL_3=1800              # 更长的TTL
-UPDATE_INTERVAL_3=30     # 更长的更新间隔
 IPV4_ENABLED_3=true     # 启用IPv4
 IPV6_ENABLED_3=true     # 启用IPv6
 ENABLED_3=true
@ -40,6 +37,5 @@ ENABLED_3=true
 # - 是否启用IPv4 (IPV4_ENABLED_n)
 # - 是否启用IPv6 (IPV6_ENABLED_n)
 # - TTL值 (TTL_n)
-# - 更新间隔 (UPDATE_INTERVAL_n)
 # - 是否启用 (ENABLED_n)
 # - 记录备注 (REMARK_n) 
--- a/4
+++ b/4
@ -1,5 +1,9 @@
 FROM python:3.11-slim

+# 设置时区为中国时区
+ENV TZ=Asia/Shanghai
+RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
+
 WORKDIR /app

 # 复制项目文件
--- a/config.py
+++ b/config.py
@ -26,10 +26,8 @@ while True:
        {
            "domain": domain,
            "sub_domain": os.getenv(f"SUB_DOMAIN_{index}", "@"),
-            "record_type": os.getenv(f"RECORD_TYPE_{index}", "A"),
            "remark": os.getenv(f"REMARK_{index}", "优选IP"),
            "ttl": int(os.getenv(f"TTL_{index}", "600")),
-            "update_interval": int(os.getenv(f"UPDATE_INTERVAL_{index}", "15")),
            "ipv4_enabled": os.getenv(f"IPV4_ENABLED_{index}", "true").lower()
            == "true",
            "ipv6_enabled": os.getenv(f"IPV6_ENABLED_{index}", "true").lower()
--- a/main.py
+++ b/main.py
@ -26,8 +26,33 @@ class DNSPodManager:
        clientProfile.httpProfile = httpProfile
        # 实例化要请求产品的client对象
        self.client = dnspod_client.DnspodClient(cred, "", clientProfile)
-        # 记录每个域名每种记录类型的最后更新时间
-        self.last_update = {}  # 格式: {domain: {'A': timestamp, 'AAAA': timestamp}}
+        # 记录当前使用的IP
+        self.current_ips = (
+            {}
+        )  # 格式: {domain: {'默认': {'A': ip, 'AAAA': ip}, '移动': {...}}}
+        # 初始化时获取所有域名当前的记录
+        self.init_current_records()
+
+    def init_current_records(self):
+        """初始化时获取所有域名当前的解析记录"""
+        logger.info("正在获取所有域名当前的解析记录...")
+        for domain_config in config.DOMAINS:
+            if not domain_config["enabled"]:
+                continue
+
+            domain = domain_config["domain"]
+            sub_domain = domain_config["sub_domain"]
+
+            # 获取当前记录
+            current_records = self.get_current_records(domain, sub_domain)
+            if current_records:
+                self.current_ips[domain] = current_records
+                logger.info(f"域名 {domain} - {sub_domain} 当前记录：")
+                for line, records in current_records.items():
+                    for record_type, ip in records.items():
+                        logger.info(f"  - {line} - {record_type}: {ip}")
+            else:
+                logger.warning(f"域名 {domain} - {sub_domain} 暂无解析记录")

    def get_optimal_ips(self) -> Dict:
        """获取优选IP"""
@ -160,11 +185,24 @@ class DNSPodManager:
            logger.error(f"更新DNS记录失败: {str(e)}")
            return False

+    def get_current_records(self, domain: str, sub_domain: str) -> Dict:
+        """获取当前域名的所有记录"""
+        try:
+            records = self.get_record_list(domain, sub_domain)
+            current_records = {}
+            for record in records:
+                if record.Line not in current_records:
+                    current_records[record.Line] = {}
+                current_records[record.Line][record.Type] = record.Value
+            return current_records
+        except Exception as e:
+            logger.error(f"获取当前记录失败: {str(e)}")
+            return {}
+
    def update_domain_records(self, domain_config: Dict) -> None:
        """更新单个域名的记录"""
        domain = domain_config["domain"]
        sub_domain = domain_config["sub_domain"]
-        record_type = domain_config["record_type"]
        ttl = domain_config["ttl"]
        remark = domain_config["remark"]

@ -173,82 +211,95 @@ class DNSPodManager:
        if not ip_data:
            return

-        # 获取对应版本的IP数据
-        ip_version = "v6" if record_type == "AAAA" else "v4"
-        if ip_version not in ip_data:
-            logger.warning(
-                f"未找到{ip_version}版本的IP数据，跳过更新 {domain} 的 {record_type} 记录"
-            )
-            return
+        # 获取当前记录
+        if domain not in self.current_ips:
+            self.current_ips[domain] = self.get_current_records(domain, sub_domain)
+        current_records = self.current_ips[domain]

-        # 检查是否有可用的IP数据
-        has_ip_data = False
-        for line_key in ["CM", "CU", "CT"]:
-            if line_key in ip_data[ip_version] and ip_data[ip_version][line_key]:
-                has_ip_data = True
-                break
+        # 处理默认线路
+        if domain_config["ipv4_enabled"] and "v4" in ip_data:
+            best_ip = self.find_best_ip(ip_data, "v4")
+            if best_ip:
+                ip, latency = best_ip
+                current_ip = current_records.get("默认", {}).get("A")
+                if current_ip != ip:
+                    logger.info(
+                        f"更新A记录: {domain} - {sub_domain} - 默认 - {ip} (延迟: {latency}ms)"
+                    )
+                    if self.update_record(
+                        domain, sub_domain, "A", "默认", ip, ttl, remark
+                    ):
+                        # 更新成功后更新缓存
+                        if "默认" not in current_records:
+                            current_records["默认"] = {}
+                        current_records["默认"]["A"] = ip
+                    time.sleep(1)

-        if not has_ip_data:
-            logger.warning(
-                f"没有可用的{ip_version}版本IP数据，跳过更新 {domain} 的 {record_type} 记录"
-            )
-            return
-
-        # 先处理默认线路
-        best_ip = self.find_best_ip(ip_data, ip_version)
-        if best_ip:
-            ip, latency = best_ip
-            logger.info(
-                f"更新{record_type}记录: {domain} - {sub_domain} - 默认 - {ip} (延迟: {latency}ms)"
-            )
-            success = self.update_record(
-                domain, sub_domain, record_type, "默认", ip, ttl, remark
-            )
-            if not success:
-                logger.error(f"更新默认线路记录失败: {domain} - {sub_domain}")
-            time.sleep(1)  # 添加延时
+        if domain_config["ipv6_enabled"] and "v6" in ip_data:
+            best_ip = self.find_best_ip(ip_data, "v6")
+            if best_ip:
+                ip, latency = best_ip
+                current_ip = current_records.get("默认", {}).get("AAAA")
+                if current_ip != ip:
+                    logger.info(
+                        f"更新AAAA记录: {domain} - {sub_domain} - 默认 - {ip} (延迟: {latency}ms)"
+                    )
+                    if self.update_record(
+                        domain, sub_domain, "AAAA", "默认", ip, ttl, remark
+                    ):
+                        # 更新成功后更新缓存
+                        if "默认" not in current_records:
+                            current_records["默认"] = {}
+                        current_records["默认"]["AAAA"] = ip
+                    time.sleep(1)

        # 更新其他线路的记录
        line_mapping = {"移动": "CM", "联通": "CU", "电信": "CT"}

        for line, line_key in line_mapping.items():
-            if line_key in ip_data[ip_version] and ip_data[ip_version][line_key]:
-                best_ip = self.find_line_best_ip(ip_data, ip_version, line_key)
+            if domain_config["ipv4_enabled"] and "v4" in ip_data:
+                best_ip = self.find_line_best_ip(ip_data, "v4", line_key)
                if best_ip:
                    ip, latency = best_ip
-                    logger.info(
-                        f"更新{record_type}记录: {domain} - {sub_domain} - {line} - {ip} (延迟: {latency}ms)"
-                    )
-                    success = self.update_record(
-                        domain, sub_domain, record_type, line, ip, ttl, remark
-                    )
-                    if not success:
-                        logger.error(f"更新{line}线路记录失败: {domain} - {sub_domain}")
-                    time.sleep(1)  # 添加延时
+                    current_ip = current_records.get(line, {}).get("A")
+                    if current_ip != ip:
+                        logger.info(
+                            f"更新A记录: {domain} - {sub_domain} - {line} - {ip} (延迟: {latency}ms)"
+                        )
+                        if self.update_record(
+                            domain, sub_domain, "A", line, ip, ttl, remark
+                        ):
+                            # 更新成功后更新缓存
+                            if line not in current_records:
+                                current_records[line] = {}
+                            current_records[line]["A"] = ip
+                        time.sleep(1)
+
+            if domain_config["ipv6_enabled"] and "v6" in ip_data:
+                best_ip = self.find_line_best_ip(ip_data, "v6", line_key)
+                if best_ip:
+                    ip, latency = best_ip
+                    current_ip = current_records.get(line, {}).get("AAAA")
+                    if current_ip != ip:
+                        logger.info(
+                            f"更新AAAA记录: {domain} - {sub_domain} - {line} - {ip} (延迟: {latency}ms)"
+                        )
+                        if self.update_record(
+                            domain, sub_domain, "AAAA", line, ip, ttl, remark
+                        ):
+                            # 更新成功后更新缓存
+                            if line not in current_records:
+                                current_records[line] = {}
+                            current_records[line]["AAAA"] = ip
+                        time.sleep(1)

    def check_and_update(self):
        """检查并更新所有域名"""
-        current_time = time.time()
-
        for domain_config in config.DOMAINS:
            if not domain_config["enabled"]:
                continue
-
-            domain = domain_config["domain"]
-
-            # 处理IPv4记录
-            if domain_config["ipv4_enabled"]:
-                ipv4_config = domain_config.copy()
-                ipv4_config["record_type"] = "A"
-                self.update_domain_records(ipv4_config)
-                time.sleep(1)  # 添加延时
-
-            # 处理IPv6记录
-            if domain_config["ipv6_enabled"]:
-                ipv6_config = domain_config.copy()
-                ipv6_config["record_type"] = "AAAA"
-                self.update_domain_records(ipv6_config)
-                time.sleep(1)  # 添加延时
+            self.update_domain_records(domain_config)
+            time.sleep(1)  # 添加延时


 def main():
@ -257,9 +308,10 @@ def main():
    # 首次运行，更新所有域名
    manager.check_and_update()

-    # 每分钟检查一次是否需要更新
-    schedule.every(1).minutes.do(manager.check_and_update)
+    # 每5分钟检查一次是否需要更新
+    schedule.every(5).minutes.do(manager.check_and_update)

+    logger.info("程序启动成功，开始监控更新（每5分钟检查一次）...")
    while True:
        schedule.run_pending()
        time.sleep(1)