qinglong/dns_optimizer.py

import requests
from time import sleep
import time
import os
import re
import sys
from datetime import datetime, timezone, timedelta
from retry import retry
import socket

# 原始脚本地址： https://github.com/cnwikee/CheckTMDB.git

DOMAINS = [
    'tmdb.org',
    'api.tmdb.org',
    'files.tmdb.org',
    'themoviedb.org',
    'api.themoviedb.org',
    'www.themoviedb.org',
    'auth.themoviedb.org',
    'image.tmdb.org',
    'images.tmdb.org',
    'imdb.com',
    'www.imdb.com',
    'secure.imdb.com',
    's.media-imdb.com',
    'us.dd.imdb.com',
    'www.imdb.to',
    'origin-www.imdb.com',
    'ia.media-imdb.com',
    'thetvdb.com',
    'api.thetvdb.com',
    'ia.media-imdb.com',
    'f.media-amazon.com',
    'imdb-video.media-imdb.com'
]

# 将 Tmdb_Host_TEMPLATE 修改为更通用的名称
HOSTS_TEMPLATE = """# Fast DNS Hosts Start
{content}
# Update time: {update_time}
# Fast DNS Hosts End\n"""


def write_host_file(file_path: str, hosts_content: str, filename: str) -> None:
    # 修改文件名生成逻辑，使其更通用
    output_file_path = os.path.join(file_path, "fast_dns_hosts_" + filename)
    with open(output_file_path, "w", encoding='utf-8') as output_fb:
        output_fb.write(hosts_content)
        print(f"\n~最新{filename}地址已更新~")

def ping_ip(ip, port=80):
    print(f"使用TCP连接测试IP地址的延迟（毫秒）")
    try:
        print(f"\n开始 ping {ip}...")
        start_time = time.time()
        with socket.create_connection((ip, port), timeout=2) as sock:
            latency = (time.time() - start_time) * 1000  # 转换为毫秒
            print(f"IP: {ip} 的平均延迟: {latency}ms")
            return latency
    except Exception as e:
        print(f"Ping {ip} 时发生错误: {str(e)}")
        return float('inf')


def find_fastest_ip(ips):
    """找出延迟最低的IP地址"""
    if not ips:
        return None

    fastest_ip = None
    min_latency = float('inf')
    ip_latencies = []  # 存储所有IP及其延迟

    for ip in ips:
        ip = ip.strip()
        if not ip:
            continue

        print(f"正在测试 IP: {ip}")
        latency = ping_ip(ip)
        ip_latencies.append((ip, latency))
        print(f"IP: {ip} 延迟: {latency}ms")

        if latency < min_latency:
            min_latency = latency
            fastest_ip = ip

        sleep(0.5)

    print("\n所有IP延迟情况:")
    for ip, latency in ip_latencies:
        print(f"IP: {ip} - 延迟: {latency}ms")

    if fastest_ip:
        print(f"\n最快的IP是: {fastest_ip}，延迟: {min_latency}ms")

    return fastest_ip


# async def process_domain(session, domain, csrf_token, udp):
#     print(f"\n正在处理域名: {domain}")
#     ipv4_ips = await async_get_domain_ips(session, domain, csrf_token, udp, "A")
#
#     if not ipv4_ips:
#         print(f"无法获取 {domain} 的IP列表，跳过该域名")
#         return None
#
#     # 处理 IPv4 地址
#     if ipv4_ips:
#         fastest_ipv4 = find_fastest_ip(ipv4_ips)
#         if fastest_ipv4:
#             print(f"域名 {domain} 的最快IPv4是: {fastest_ipv4}")
#             return [fastest_ipv4, domain]
#         else:
#             return [ipv4_ips[0], domain]

def validate_ip(ip):
    """
    验证IP是否为合法的IPv4或IPv6地址
    :param ip: 待验证的IP字符串
    :return: True（合法）/False（非法）
    """
    # IPv4正则（严格验证：每个段0-255，无前置零（除0.0.0.0等合法场景））
    ipv4_pattern = r'^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$'
    # IPv6正则（兼容压缩格式、本地链路地址、IPv4映射地址等所有合法格式）
    ipv6_pattern = r'^(?:(?:[0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}|(?:[0-9a-fA-F]{1,4}:){1,7}:|(?:[0-9a-fA-F]{1,4}:){1,6}:[0-9a-fA-F]{1,4}|(?:[0-9a-fA-F]{1,4}:){1,5}(?::[0-9a-fA-F]{1,4}){1,2}|(?:[0-9a-fA-F]{1,4}:){1,4}(?::[0-9a-fA-F]{1,4}){1,3}|(?:[0-9a-fA-F]{1,4}:){1,3}(?::[0-9a-fA-F]{1,4}){1,4}|(?:[0-9a-fA-F]{1,4}:){1,2}(?::[0-9a-fA-F]{1,4}){1,5}|[0-9a-fA-F]{1,4}:(?:(?::[0-9a-fA-F]{1,4}){1,6})|:(?:(?::[0-9a-fA-F]{1,4}){1,7}|:)|fe80:(?::[0-9a-fA-F]{0,4}){0,4}%[0-9a-zA-Z]{1,}|::(?:ffff(?::0{1,4}){0,1}:){0,1}(?:(?:25[0-5]|(?:2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(?:25[0-5]|(?:2[0-4]|1{0,1}[0-9]){0,1}[0-9])|(?:[0-9a-fA-F]{1,4}:){1,4}:(?:(?:25[0-5]|(?:2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(?:25[0-5]|(?:2[0-4]|1{0,1}[0-9]){0,1}[0-9]))$'

    # 忽略大小写验证IPv6，优先验证IPv4
    if re.match(ipv4_pattern, ip):
        return True
    elif re.match(ipv6_pattern, ip, re.IGNORECASE):
        return True
    else:
        return False

@retry(tries=3)
def get_domain_ips(domain, record_type):
    """
    从Google DNS获取域名的A/AAAA记录IP列表
    :param domain: 目标域名（如 tmdb.org）
    :param record_type: 记录类型（A/AAAA，或数字1/28）
    :return: 去重后的IP列表
    """
    all_ips = []  # 存储所有DNS服务器返回的IP

    print(f"正在从Google DNS获取 {domain} 的{record_type}记录...")
    url = f'https://cloudflare-dns.com/dns-query'
    headers = {
        "accept": "application/dns-json"
        # "accept-encoding": "gzip, deflate, br, zstd",
        # "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
        # "content-type": "application/json; charset=UTF-8",  # 关键：指定JSON格式负载
        # "referer": f"https://dns.google/query?name={domain}&rr_type={record_type}&ecs=",
        # "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/141.0.0.0 Safari/537.36 Edg/141.0.0.0"
    }

    params = {
        'name': domain,
        'type': record_type
    }

    # 初始化IP列表（默认空列表，确保后续使用安全）
    ips_str = []

    try:
        # 改用GET请求（Google DNS resolve接口标准用法）
        response = requests.get(url, headers=headers, params=params, timeout=10)
        response.raise_for_status()  # 主动抛出HTTP错误（如4xx/5xx）

        # 解析JSON响应
        data = response.json()
        if not isinstance(data, dict):
            print("返回数据不是字典格式，无法解析")
            return all_ips

        # 核心：提取Answer数组中的data字段（IP地址）
        answer_list = data.get("Answer", [])
        if not answer_list:
            print(f"未找到 {domain} 的{record_type}记录（Answer字段为空）")
            return all_ips

        # 遍历Answer数组，提取每个条目的data值（IP）
        for answer in answer_list:
            ip = answer.get("data")
            if not ip:  # 过滤空值
                continue

            # 验证IP格式合法性
            if validate_ip(ip):
                all_ips.append(ip)
                print(f"提取到合法IP：{ip}")
            else:
                print(f"跳过非法IP格式：{ip}")

    except requests.exceptions.RequestException as e:
        # 捕获所有网络/请求异常
        print(f"请求Google DNS失败：{e}")
        if hasattr(e, 'response') and e.response:
            print(f"响应内容：{e.response.text[:500]}")  # 打印前500字符避免过长
    except ValueError:
        # JSON解析失败
        print(f"响应内容不是有效的JSON格式：{response.text[:500]}")
    time.sleep(1)
    # 去重并返回（保持列表格式）
    unique_ips = list(set(all_ips))
    print(f"最终提取到 {domain} 的{record_type}记录IP（去重后）：{unique_ips}")
    return unique_ips


def main():
    print("开始检测域名的最快IP...")
    file_path = "/ql/data"

    ipv4_ips, ipv6_ips, ipv4_results, ipv6_results = [], [], [], []

    for domain in DOMAINS:
        print(f"\n正在处理域名: {domain}")
        ipv4_ips = get_domain_ips(domain, "A")
        # ipv6_ips = get_domain_ips(domain, "AAAA")

        if not ipv4_ips and not ipv6_ips:
            print(f"无法获取 {domain} 的IP列表，跳过该域名")
            continue

        # 处理 IPv4 地址
        if ipv4_ips:
            fastest_ipv4 = find_fastest_ip(ipv4_ips)
            if fastest_ipv4:
                ipv4_results.append([fastest_ipv4, domain])
                print(f"域名 {domain} 的最快IPv4是: {fastest_ipv4}")
            else:
                ipv4_results.append([ipv4_ips[0], domain])

        # 处理 IPv6 地址
        # if ipv6_ips:
        #     fastest_ipv6 = find_fastest_ip(ipv6_ips)
        #     if fastest_ipv6:
        #         ipv6_results.append([fastest_ipv6, domain])
        #         print(f"域名 {domain} 的最快IPv6是: {fastest_ipv6}")
        #     else:
        #         # 兜底：可能存在无法正确获取 fastest_ipv6 的情况，则将第一个IP赋值
        #         ipv6_results.append([ipv6_ips[0], domain])

        sleep(1)  # 避免请求过于频繁


    # 保存结果到文件
    if not ipv4_results and not ipv6_results:
        print(f"程序出错：未获取任何domain及对应IP，请检查接口~")
        sys.exit(1)

    # 生成更新时间
    update_time = datetime.now(timezone(timedelta(hours=8))).replace(microsecond=0).isoformat()

    ipv4_hosts_content = HOSTS_TEMPLATE.format(
        content="\n".join(f"{ip:<27} {domain}" for ip, domain in ipv4_results),
        update_time=update_time) if ipv4_results else ""
    # ipv6_hosts_content = HOSTS_TEMPLATE.format(
    #     content="\n".join(f"{ip:<50} {domain}" for ip, domain in ipv6_results),
    #     update_time=update_time) if ipv6_results else ""

    # 写入文件
    if ipv4_hosts_content:
        write_host_file(file_path, ipv4_hosts_content, 'ipv4')

if __name__ == "__main__":
    main()