Create dns_resolver.py

dns_resolver.py

@@ -0,0 +1,95 @@
+import socket
+import urllib.request
+import json
+import time
+import logging
+
+class DnsManager:
+    _instance = None
+
+    def __new__(cls, *args, **kwargs):
+        if not cls._instance:
+            cls._instance = super(DnsManager, cls).__new__(cls)
+        return cls._instance
+
+    def __init__(self, cache_ttl=600):
+        # 避免单例模式下重复初始化
+        if hasattr(self, '_initialized'):
+            return
+        self.cache_ttl = cache_ttl
+        self.cache = {}  # 格式: {"hostname": {"ip": "xxx", "expiry": 0}}
+        # 覆盖 Discord 机器人运行所需的所有关键域名
+        self.target_hosts = [
+            'discord.com', 
+            'gateway.discord.gg', 
+            'cdn.discordapp.com', 
+            'discordapp.com'
+        ]
+        self.logger = logging.getLogger("DnsManager")
+        self._initialized = True
+
+    def _fetch_ip_from_http(self, hostname):
+        """通过 HTTPS (TCP 443) 绕过系统 UDP DNS 限制获取 IP"""
+        try:
+            # 使用 Google DNS API
+            url = f"https://dns.google/resolve?name={hostname}&type=A"
+            req = urllib.request.Request(url, headers={"Accept": "application/dns-json"})
+            with urllib.request.urlopen(req, timeout=5) as response:
+                data = json.loads(response.read().decode())
+                if "Answer" in data:
+                    # 找到第一个 A 记录 (IPv4)
+                    for answer in data['Answer']:
+                        if answer['type'] == 1: # A record
+                            return answer['data']
+        except Exception as e:
+            self.logger.warning(f"无法通过 HTTP 解析 {hostname}: {e}")
+        return None
+
+    def get_ip(self, hostname):
+        """获取 IP，优先使用缓存，过期则自动更新"""
+        current_time = time.time()
+        
+        if hostname in self.cache:
+            if current_time < self.cache[hostname]['expiry']:
+                return self.cache[hostname]['ip']
+
+        # 尝试更新 IP
+        new_ip = self._fetch_ip_from_http(hostname)
+        
+        if new_ip:
+            self.cache[hostname] = {
+                "ip": new_ip,
+                "expiry": current_time + self.cache_ttl
+            }
+            print(f"🔄 [DNS库] 动态解析成功: {hostname} -> {new_ip}")
+            return new_ip
+        
+        # 如果获取失败，且缓存里有旧的，则延长旧缓存寿命并继续使用
+        if hostname in self.cache:
+            self.logger.warning(f"解析失败，正在使用 {hostname} 的过期缓存")
+            return self.cache[hostname]['ip']
+        
+        # 最后的保底 hardcode IP (Discord 常用 IP)
+        return "162.159.137.232"
+
+    def patch_socket(self):
+        """核心方法：将补丁注入到 Python 全局 socket 层"""
+        original_getaddrinfo = socket.getaddrinfo
+
+        def patched_getaddrinfo(*args, **kwargs):
+            host = args[0]
+            # 只有在目标域名列表里且不是正在请求 DNS 接口本身时才拦截
+            if host in self.target_hosts:
+                latest_ip = self.get_ip(host)
+                port = args[1]
+                # 强制返回 IPv4 地址族 (AF_INET)，避免 IPv6 解析错误
+                return [(socket.AF_INET, socket.SOCK_STREAM, 6, '', (latest_ip, port))]
+            
+            # 其余请求（如访问 MongoDB 或 Google DNS 接口）走原始解析
+            return original_getaddrinfo(*args, **kwargs)
+
+        socket.getaddrinfo = patched_getaddrinfo
+        print("💉 [DNS库] 全局 Socket 补丁已加载，已绕过系统 DNS")
+
+# 导出实例
+dns_manager = DnsManager()