域名批量查询dns

import dns.resolver
import concurrent.futures

def query_dns(domain, record_type):
    try:
        answers = dns.resolver.resolve(domain, record_type)
        return {domain: [rdata.to_text() for rdata in answers]}
    except dns.resolver.NXDOMAIN:
        return {domain: 'Domain does not exist'}
    except dns.resolver.NoAnswer:
        return {domain: 'No answer'}
    except dns.resolver.Timeout:
        return {domain: 'Timeout'}

def batch_query(domains, record_type):
    results = {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = {executor.submit(query_dns, domain, record_type): domain for domain in domains}
        for future in concurrent.futures.as_completed(futures):
            domain = futures[future]
            try:
                results.update(future.result())
            except Exception as e:
                results[domain] = str(e)
    return results

# 示例使用
domains = ['example1.com', 'example2.com', 'example3.com']
record_type = 'A'
results = batch_query(domains, record_type)
print(results)