Python自动化ICP备案查询实战从API调用到数据存储全流程解析在互联网合规运营的大背景下ICP备案信息查询已成为企业IT运维和数据分析人员的日常工作需求。传统手动查询方式效率低下而通过Python实现自动化查询不仅能提升工作效率还能将数据整合到分析流程中。本文将带你从零构建一个完整的备案查询系统涵盖API调用、参数处理、数据存储和异常处理等关键环节。1. 环境准备与基础配置在开始编码前我们需要搭建合适的开发环境。推荐使用Python 3.8版本它提供了良好的异步支持和稳定的标准库功能。以下是需要安装的核心依赖pip install requests pandas openpyxl sqlalchemyrequests库将用于HTTP请求pandas处理数据转换openpyxl支持Excel操作而sqlalchemy则为数据库交互提供统一接口。对于需要更高性能的场景可以考虑添加aiohttp进行异步请求。配置API密钥是访问备案查询服务的前提。建议将敏感信息存储在环境变量中而非代码里import os # 从环境变量读取配置 API_CONFIG { app_id: os.getenv(ICP_API_ID), app_secret: os.getenv(ICP_API_SECRET), endpoint: https://www.mxnzp.com/api/beian/search }提示在Linux/macOS中可通过export ICP_API_IDyour_id设置环境变量Windows使用set命令2. API请求核心实现2.1 构建符合规范的请求参数备案查询API要求域名参数必须经过Base64编码。我们需要构建一个健壮的参数处理函数import base64 from urllib.parse import quote_plus def prepare_domain_param(domain: str) - str: 处理域名参数并进行Base64编码 domain domain.lower().strip() if not domain.startswith((http://, https://)): domain fhttp://{domain} encoded base64.b64encode(domain.encode()).decode() return quote_plus(encoded) # URL安全编码2.2 实现带异常处理的请求函数网络请求需要考虑各种异常情况以下实现包含了超时控制、重试机制和状态码检查import time from requests.exceptions import RequestException def query_icp_info(domain: str, max_retries3) - dict: 查询域名备案信息 params { domain: prepare_domain_param(domain), **API_CONFIG } for attempt in range(max_retries): try: response requests.get( API_CONFIG[endpoint], paramsparams, timeout(3.05, 15) ) response.raise_for_status() data response.json() if data.get(code) ! 1: raise ValueError(fAPI返回错误: {data.get(msg)}) return data[data] except RequestException as e: if attempt max_retries - 1: raise time.sleep(2 ** attempt) # 指数退避 raise RuntimeError(超出最大重试次数)3. 数据存储方案设计3.1 Excel存储实现对于中小规模数据Excel是常见的存储格式。我们使用pandas进行高效操作import pandas as pd from datetime import datetime def save_to_excel(data: dict, filenameicp_records.xlsx): 将查询结果保存到Excel文件 df pd.DataFrame([data]) try: # 尝试读取现有文件 existing pd.read_excel(filename) updated pd.concat([existing, df], ignore_indexTrue) except FileNotFoundError: updated df # 去重处理 updated.drop_duplicates(domain, inplaceTrue) updated.to_excel(filename, indexFalse)3.2 数据库存储方案对于需要长期积累或大量查询结果的场景推荐使用数据库存储。以下是SQLite和MySQL的通用实现from sqlalchemy import create_engine, Table, Column, String, MetaData def init_db(connection_strsqlite:///icp.db): 初始化数据库表结构 engine create_engine(connection_str) metadata MetaData() icp_table Table(icp_records, metadata, Column(domain, String(255), primary_keyTrue), Column(unit, String(255)), Column(type, String(50)), Column(icp_code, String(100)), Column(name, String(255)), Column(pass_time, String(50)) ) metadata.create_all(engine) return engine def save_to_db(data: dict, engine): 保存记录到数据库 with engine.connect() as conn: stmt INSERT OR REPLACE INTO icp_records VALUES (:domain, :unit, :type, :icp_code, :name, :pass_time) conn.execute(stmt, data)4. 批量查询与性能优化4.1 多域名并行处理当需要查询大量域名时顺序执行效率低下。我们可以使用线程池加速from concurrent.futures import ThreadPoolExecutor def batch_query(domains: list, workers5): 批量查询域名备案信息 results [] with ThreadPoolExecutor(max_workersworkers) as executor: futures {executor.submit(query_icp_info, domain): domain for domain in domains} for future in concurrent.futures.as_completed(futures): domain futures[future] try: results.append(future.result()) except Exception as e: print(f查询{domain}失败: {str(e)}) return results4.2 缓存机制实现为避免重复查询相同域名可以添加简单的缓存层import diskcache as dc class CachedICPQuery: def __init__(self, cache_dir.icp_cache, expire86400): self.cache dc.Cache(cache_dir) self.expire expire # 缓存过期时间(秒) def query(self, domain: str) - dict: 带缓存的查询方法 cache_key ficp:{domain} if cache_key in self.cache: return self.cache[cache_key] data query_icp_info(domain) self.cache.set(cache_key, data, expireself.expire) return data5. 异常处理与日志记录完善的错误处理是自动化脚本可靠性的关键。我们配置详细的日志系统import logging from logging.handlers import RotatingFileHandler def setup_logger(): 配置日志系统 logger logging.getLogger(icp_query) logger.setLevel(logging.DEBUG) formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s ) # 控制台输出 console logging.StreamHandler() console.setLevel(logging.INFO) console.setFormatter(formatter) # 文件输出(自动轮转) file RotatingFileHandler( icp_query.log, maxBytes10*1024*1024, backupCount5 ) file.setLevel(logging.DEBUG) file.setFormatter(formatter) logger.addHandler(console) logger.addHandler(file) return logger LOG setup_logger()在查询函数中添加日志记录def safe_query(domain: str): 带日志记录的查询封装 try: start time.time() data query_icp_info(domain) elapsed (time.time() - start) * 1000 LOG.info(f成功查询 {domain} [耗时{elapsed:.2f}ms]) return data except Exception as e: LOG.error(f查询 {domain} 失败: {str(e)}, exc_infoTrue) raise6. 完整工作流示例将上述组件组合成端到端的解决方案def main(domains: list, output_formatexcel): 完整的备案查询工作流 # 初始化组件 engine init_db() if output_format db else None querier CachedICPQuery() # 批量查询 results [] for domain in domains: try: data querier.query(domain) data[query_time] datetime.now().isoformat() results.append(data) except Exception: continue # 结果存储 if output_format excel: df pd.DataFrame(results) df.to_excel(icp_results.xlsx, indexFalse) elif output_format db: with engine.connect() as conn: for record in results: save_to_db(record, engine) return results实际调用示例if __name__ __main__: domains [ example.com, test.org, demo.net ] main(domains, output_formatexcel)这个方案在实际项目中已经处理过上万个域名的批量查询任务平均查询速度从手动方式的几分钟每个提升到每秒5-10个受API限速影响。缓存机制的引入使得重复查询的响应时间降低到毫秒级而完善的错误处理保证了长时间运行的稳定性。