Python脚本自动查ICP备案，告别手动查询的苦日子

pip install requests

import requests
import time
import json

class ICPQuery:
    """ICP备案查询工具"""

    def __init__(self, api_url, api_key=None):
        """
        初始化查询器
        :param api_url: 第三方API地址
        :param api_key: API密钥（如需要）
        """
        self.api_url = api_url
        self.api_key = api_key
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }

    def query_single(self, domain):
        """
        查询单个域名的ICP备案信息
        :param domain: 域名
        :return: 备案信息字典
        """
        try:
            params = {'domain': domain, 'key': self.api_key}
            response = requests.get(
                self.api_url,
                params=params,
                headers=self.headers,
                timeout=10
            )
            response.raise_for_status()
            result = response.json()

            # 根据实际API返回结构解析
            if result.get('code') == 200 or result.get('status') == 'success':
                data = result.get('data', {})
                return {
                    'domain': domain,
                    'unit_name': data.get('unitName', '未知'),        # 主办单位
                    'nature_name': data.get('natureName', '未知'),    # 单位性质
                    'icp': data.get('icp', '未知'),                   # 备案号
                    'update_time': data.get('updateTime', '未知'),    # 审核时间
                    'status': '成功'
                }
            else:
                return {'domain': domain, 'status': '未查到备案信息', 'msg': result.get('msg', '')}

        except requests.exceptions.Timeout:
            return {'domain': domain, 'status': '查询超时'}
        except requests.exceptions.RequestException as e:
            return {'domain': domain, 'status': f'请求异常: {str(e)}'}
        except json.JSONDecodeError:
            return {'domain': domain, 'status': '返回数据解析失败'}

    def query_batch(self, domains, delay=1.5):
        """
        批量查询域名备案信息
        :param domains: 域名列表
        :param delay: 请求间隔(秒)，防止被封
        :return: 查询结果列表
        """
        results = []
        total = len(domains)

        for i, domain in enumerate(domains, 1):
            print(f"[{i}/{total}] 正在查询: {domain}")
            result = self.query_single(domain.strip())
            results.append(result)

            # 打印结果
            if result.get('status') == '成功':
                print(f"  ✅ {result['unit_name']} | {result['icp']}")
            else:
                print(f"  ❌ {result['status']}")

            # 控制请求频率，避免被限制
            if i < total:
                time.sleep(delay)

        return results

    def export_csv(self, results, filename='icp_results.csv'):
        """
        导出查询结果为CSV文件
        :param results: 查询结果列表
        :param filename: 输出文件名
        """
        import csv
        with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
            writer = csv.writer(f)
            writer.writerow(['域名', '主办单位', '单位性质', '备案号', '审核时间', '状态'])
            for r in results:
                writer.writerow([
                    r.get('domain', ''),
                    r.get('unit_name', ''),
                    r.get('nature_name', ''),
                    r.get('icp', ''),
                    r.get('update_time', ''),
                    r.get('status', '')
                ])
        print(f"\n📁 结果已导出: {filename}")


if __name__ == '__main__':
    # ========== 配置区 ==========
    # 替换为你自己的API地址和密钥
    API_URL = 'https://api.example.com/icp/query'
    API_KEY = 'your_api_key_here'

    # 需要查询的域名列表
    DOMAINS = [
        'baidu.com',
        'qq.com',
        'github.com',
        'example.com',  # 这个大概率查不到
    ]
    # ============================

    query = ICPQuery(API_URL, API_KEY)

    # 批量查询
    results = query.query_batch(DOMAINS, delay=1.5)

    # 导出CSV
    query.export_csv(results)

def read_domains_from_file(filepath):
    """从文本文件读取域名列表，每行一个"""
    with open(filepath, 'r', encoding='utf-8') as f:
        domains = [line.strip() for line in f if line.strip()]
    return domains

# 使用
domains = read_domains_from_file('domains.txt')
results = query.query_batch(domains)

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    filename='icp_query.log',
    filemode='a'
)

# 在 query_single 方法里加一行
logging.info(f"查询 {domain}: {result}")

import aiohttp
import asyncio

async def async_query(domain, api_url, api_key, semaphore):
    async with semaphore:
        try:
            params = {'domain': domain, 'key': api_key}
            async with aiohttp.ClientSession() as session:
                async with session.get(api_url, params=params, timeout=10) as resp:
                    result = await resp.json()
                    return {'domain': domain, 'data': result}
        except Exception as e:
            return {'domain': domain, 'error': str(e)}

async def batch_async(domains, api_url, api_key, concurrency=3):
    semaphore = asyncio.Semaphore(concurrency)
    tasks = [async_query(d, api_url, api_key, semaphore) for d in domains]
    return await asyncio.gather(*tasks)

# 运行
# results = asyncio.run(batch_async(domains, API_URL, API_KEY))