利用Python高效批量查询域名可用性

import whois

def check_domain_slow(domain):
    try:
        w = whois.whois(domain)
        # 根据whois查询结果判断域名状态
        # 不同的whois服务器返回的状态字段可能不同，这里以'status'为例
        if w.status is None or "no match" in str(w.status).lower(): # 简单判断为可注册
            return True
        else:
            return False
    except Exception as e:
        print(f"Error checking {domain}: {e}")
        return False

# 假设matches是一个包含50k域名的列表
# available_domains = [match for match in matches if check_domain_slow(match)]
# 这种方式会非常慢

import os
import sys
from multiprocessing import Pool
import pandas as pd
from tqdm import tqdm
from whois import whois

# 函数：抑制标准输出，避免whois库在并行环境中打印信息
def blockPrint():
    sys.stdout = open(os.devnull, "w")

# 函数：恢复标准输出
def enablePrint():
    sys.stdout = sys.__stdout__

# 核心函数：检查单个域名可用性
def check_domain_availability(domain):
    try:
        blockPrint()  # 抑制当前进程的输出
        result = whois(domain)
    except Exception:
        # 捕获所有异常，确保进程不会崩溃，并返回None表示查询失败或未知
        return domain, None
    finally:
        enablePrint()  # 恢复当前进程的输出

    # 根据whois查询结果判断域名状态
    # whois库通常在域名已注册时返回一个对象，其status属性非空或包含注册信息
    # 如果域名未注册，whois查询可能会失败或返回一个结果对象，其中status为None或表示“No Match”
    # 这里我们简化判断：如果status为None，则认为可能是可用的（需要进一步验证）
    # 实际应用中，更严谨的判断可能需要检查result.registrar, result.creation_date等字段
    is_free = not bool(result.status) if result and hasattr(result, 'status') else True
    return domain, is_free

if __name__ == "__main__":
    # 示例域名列表（实际应用中替换为您的50k域名列表）
    domains_to_check = [
        "google.com",
        "yahoo.com",
        "facebook.com",
        "xxxnonexistentzzz.domain", # 假设这是一个不存在的域名
        "python.org",
        "example.com",
        "availabledomain12345.com", # 假设这是一个可用的域名
        "anotherunavailable.net"
    ] * 100 # 放大列表以模拟大规模查询

    results = []
    # 使用进程池，processes参数控制并发进程数
    # 根据您的CPU核心数和网络带宽合理设置，通常为CPU核心数或稍多
    num_processes = 16
    print(f"开始使用 {num_processes} 个进程批量查询域名可用性...")

    with Pool(processes=num_processes) as pool:
        # imap_unordered用于异步获取结果，且不保证顺序，适用于我们不关心结果顺序的场景
        for domain, is_free in tqdm(
            pool.imap_unordered(check_domain_availability, domains_to_check),
            total=len(domains_to_check),
            desc="查询进度"
        ):
            results.append((domain, is_free))

    # 将结果转换为DataFrame并去重，方便查看
    df = pd.DataFrame(results, columns=["domain", "is_free"])
    print("\n查询结果:")
    print(df.drop_duplicates())

    # 打印可用域名
    available_domains_df = df[df['is_free']].drop_duplicates()
    if not available_domains_df.empty:
        print("\n以下域名可能可用:")
        for domain in available_domains_df['domain']:
            print(domain)
    else:
        print("\n未发现可能可用的域名。")