Python根据域名批量查询服务器所在地的方法

import socket
import requests
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def get_ip_from_domain(domain):
    """解析域名获取IP地址"""
    try:
        # 使用socket获取IP，可考虑添加超时设置
        ip = socket.gethostbyname(domain)
        return ip
    except socket.gaierror:
        return None
    except Exception as e:
        print(f"解析 {domain} 时出错: {e}")
        return None

def get_location_from_ip(ip):
    """通过IP查询地理位置（使用ip-api.com免费服务）"""
    if not ip:
        return None
    
    try:
        url = f"http://ip-api.com/json/{ip}"
        response = requests.get(url, timeout=5)
        data = response.json()
        
        if data['status'] == 'success':
            return {
                'ip': ip,
                'country': data.get('country', ''),
                'region': data.get('regionName', ''),
                'city': data.get('city', ''),
                'isp': data.get('isp', ''),
                'org': data.get('org', '')
            }
        return None
    except requests.exceptions.RequestException as e:
        print(f"查询IP {ip} 位置时出错: {e}")
        return None
    except json.JSONDecodeError:
        print(f"解析IP {ip} 响应时出错")
        return None

def batch_query_domains(domains, max_workers=10):
    """批量查询域名服务器所在地"""
    results = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有域名解析任务
        future_to_domain = {
            executor.submit(get_ip_from_domain, domain): domain 
            for domain in domains
        }
        
        # 处理解析结果并提交位置查询
        location_futures = []
        for future in as_completed(future_to_domain):
            domain = future_to_domain[future]
            try:
                ip = future.result()
                if ip:
                    # 提交位置查询任务
                    location_future = executor.submit(get_location_from_ip, ip)
                    location_futures.append((domain, ip, location_future))
                else:
                    results.append({'domain': domain, 'error': '无法解析IP'})
            except Exception as e:
                results.append({'domain': domain, 'error': str(e)})
        
        # 收集位置查询结果
        for domain, ip, location_future in location_futures:
            try:
                location = location_future.result()
                if location:
                    results.append({
                        'domain': domain,
                        'ip': ip,
                        'country': location['country'],
                        'region': location['region'],
                        'city': location['city'],
                        'isp': location['isp'],
                        'org': location['org']
                    })
                else:
                    results.append({'domain': domain, 'ip': ip, 'error': '无法获取位置信息'})
            except Exception as e:
                results.append({'domain': domain, 'ip': ip, 'error': str(e)})
    
    return results

def main():
    # 示例域名列表
    domains = [
        'google.com',
        'github.com',
        'baidu.com',
        'example.com',
        'invalid-domain-test-123.com'  # 无效域名测试
    ]
    
    print("开始批量查询域名服务器所在地...")
    start_time = time.time()
    
    results = batch_query_domains(domains)
    
    # 输出结果
    print(f"\n查询完成，耗时: {time.time() - start_time:.2f}秒")
    print("=" * 80)
    
    for result in results:
        print(f"\n域名: {result['domain']}")
        if 'error' in result:
            print(f"  错误: {result['error']}")
        else:
            print(f"  IP地址: {result['ip']}")
            print(f"  国家: {result['country']}")
            print(f"  地区: {result['region']}")
            print(f"  城市: {result['city']}")
            print(f"  ISP: {result['isp']}")
            print(f"  组织: {result['org']}")

if __name__ == "__main__":
    main()

这个方案分三步走：先用socket解析域名拿IP，再通过ip-api.com的免费API查地理位置，最后用线程池批量处理。注意ip-api.com有频率限制（每分钟45次），生产环境建议用付费API或自建IP库。