从curl到工程封装:ICP备案查询
应用场景何时需要自动化查询ICP备案在日常开发或运维中以下场景会频繁用到ICP备案查询接口域名合规审核在用户提交域名绑定、域名接入时自动检测域名是否已备案未备案则拒绝或提醒。备案状态监控定期巡检自有或合作方的域名列表及时发现备案被注销、被吊销的情况避免业务中断。内容平台风控对用户上传的外部链接进行备案检测过滤未备案的违规网站。域名交易/转让在查看文档或出售域名前快速确认当前主体的备案情况。这些场景的共同特点是高频、自动、要求低延迟手动的浏览器查询或curl单次调用已不够用需要将接口封装成可复用的模块融入CI/CD、定时任务或业务后端。接口能力边界你必须知道的几件事本接口基于工信部公开备案数据通过域名查询备案信息。以下是其关键特性请求方式GET单次请求即返回结果非常适合高并发轻量调用。QPS 限制5次/秒。如果业务需要更高吞吐需要设计本地缓存或分布式限流。缓存策略已备案域名24小时缓存备案信息变更频率低未备案域名1小时缓存避免新备案域名被长期误判。这意味着重复查询同一个域名会命中缓存时效性对于备案变更不敏感的业务足够。域名清洗自动剥离https://、http://、www.、端口号、路径输入https://www.baidu.com/abc与baidu.com等效。这一层减少入参的前处理负担。返回结构is_filed布尔字段指示备案状态已备案时填充完整6字段未备案/境外/已注销时返回空字段null或但状态码仍为200便于前端直接判断。注意接口不保证数据实时存在缓存窗也不承诺覆盖所有境外域名或历史注销数据请以工信部官网为准。请求参数与鉴权方式Query 参数参数名必填类型说明示例domain是string要查询的域名支持完整URL自动清洗baidu.com或https://www.baidu.com/abcHeader 参数参数名必填类型说明示例Authorization否stringAPI Key 鉴权格式Bearer sk_live_xxx匿名调用时可省略每日限制30次Bearer sk_live_xxxxxxxxxxxxxx匿名调用每日30次超过会返回限流错误。建议线上使用准备的API Key额度更高具体以文档为准。curl 接入从单次请求开始带鉴权的完整请求# 替换 YOUR_API_KEY 为你的真实密钥 export API_KEYsk_live_xxxxxxxxxxxxxx curl -sS \ -X GET \ -H Authorization: Bearer $API_KEY \ https://v1.apizero.cn/api/icp?domainbaidu.com预期返回已格式化{ code: 0, data: { domain: baidu.com, icp_code: 京ICP证030173号-1, site_name: 百度一下你就知道, company_name: 北京百度网讯科技有限公司, company_type: 企业, audit_time: 2019-05-16 16:06:21, is_filed: true }, msg: 成功, request_id: req_abc123 }未备案域名的返回curl -sS https://v1.apizero.cn/api/icp?domainnotexist12345.com返回示例{ code: 0, data: { domain: notexist12345.com, icp_code: null, site_name: null, company_name: null, company_type: null, audit_time: null, is_filed: false }, msg: 成功, request_id: req_def456 }注意未备案或境外域名返回的is_filedfalse旁边字段为空并非错误。匿名调用每日限30次去掉 Authorization 头即可curl -sS https://v1.apizero.cn/api/icp?domainbaidu.com## 从curl到工程封装Python 类设计 单次curl调试完成后我们需要把它变成可复用的模块。以下是一个 Python 工程化封装涵盖**超时控制、重试、限流、日志、异常分类**。 python import requests import time import logging from functools import lru_cache from typing import Optional, Dict, Any logger logging.getLogger(__name__) class ICPQueryError(Exception): ICP查询基异常 pass class RateLimitError(ICPQueryError): 限流异常 pass class AuthError(ICPQueryError): 鉴权失败 pass class ICPQueryClient: BASE_URL https://v1.apizero.cn/api/icp def __init__(self, api_key: Optional[str] None, qps: float 4.0, retry_times: int 2): :param api_key: API Key为None时使用匿名每日30次 :param qps: 每秒最大请求数建议低于接口QPS5/s :param retry_times: 失败重试次数 self.api_key api_key self.qps qps self._last_request_time 0.0 self.retry_times retry_times self.session requests.Session() # 适配Authorization头 if self.api_key: self.session.headers.update({Authorization: fBearer {self.api_key}}) def _enforce_qps(self): 简单令牌桶确保请求间隔不低于 1/qps 秒 now time.time() elapsed now - self._last_request_time min_interval 1.0 / self.qps if elapsed min_interval: time.sleep(min_interval - elapsed) self._last_request_time time.time() def query(self, domain: str) - Dict[str, Any]: 查询单个域名的ICP备案信息 :param domain: 域名支持完整URL自动清洗 :return: API返回的JSON字典 :raises ICPQueryError: 网络、限流、鉴权等异常 self._enforce_qps() for attempt in range(self.retry_times 1): try: resp self.session.get( self.BASE_URL, params{domain: domain}, timeout(3, 10) # (connect, read) timeout ) resp.raise_for_status() # 非200抛出HTTPError data resp.json() # 检查业务错误码 if data.get(code) ! 0: raise ICPQueryError(f业务错误: {data.get(msg, 未知错误)}) logger.info(fICP query success: {domain}, is_filed{data[data][is_filed]}) return data except requests.exceptions.HTTPError as e: status e.response.status_code if status 401: raise AuthError(API Key 无效或未授权) from e elif status 429: # 限流等待后重试 if attempt self.retry_times: wait min(2 ** attempt, 10) logger.warning(fRate limited, retrying in {wait}s...) time.sleep(wait) continue else: raise RateLimitError(超出速率限制重试后仍失败) from e else: if attempt self.retry_times: time.sleep(1) continue else: raise ICPQueryError(fHTTP错误: {status}) from e except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e: if attempt self.retry_times: wait 2 ** attempt logger.warning(fNetwork error, retrying in {wait}s...) time.sleep(wait) continue else: raise ICPQueryError(f网络请求失败: {e}) from e except ValueError as e: # JSON解析失败 raise ICPQueryError(f响应JSON解析失败: {e}) from e # 不应当到达这里 raise ICPQueryError(未知错误) # 可选添加LRU缓存减少重复请求 lru_cache(maxsize1024) def query_cached(self, domain: str) - Dict[str, Any]: return self.query(domain)使用示例# 实例化客户端匿名模式 client ICPQueryClient() # 查询已备案域名 try: result client.query(baidu.com) data result[data] if data[is_filed]: print(f备案号: {data[icp_code]}) print(f主办单位: {data[company_name]}) else: print(该域名未备案) except ICPQueryError as e: print(f查询失败: {e})注意实际生产使用时建议将api_key通过环境变量注入避免硬编码。返回值详细解读字段类型说明codeint0表示成功非0表示业务异常msgstring状态描述成功为“成功”request_idstring每次请求的唯一ID用于排查日志data.domainstring查询的域名清洗后data.is_filedbooltrue-已备案false-未备案/境外/已注销data.icp_codestring/null备案许可证号如“京ICP证030173号-1”data.site_namestring/null网站名称data.company_namestring/null主办单位名称data.company_typestring/null单位性质企业/个人/事业单位/政府机关等data.audit_timestring/null审核通过时间格式YYYY-MM-DD HH:mm:ss关注is_filed字段未备案时其他字段为null但code仍是0这方便业务层直接做if data[is_filed]:判断无需额外错误处理。常见错误与处理方式400 请求参数缺失若未传domain参数返回类似{ code: 1001, msg: 参数 domain 不能为空, data: {} }解决检查请求URL是否包含?domainxxx。401 鉴权失败场景使用了无效的API Key或者匿名调用但超出了每日30次限制。返回HTTP 401body含错误信息。处理检查API Key是否正确或增加额度。429 请求过于频繁返回HTTP 429body提示超出QPS。处理客户端限速建议QPS控制在4/s并实现指数退避重试。未备案不是错误许多开发者误以为未备案会返回错误码实际上is_filedfalse且code0。请务必用业务逻辑区分不要抛异常。工程化注意事项域名清洗虽然服务端会自动清洗但建议客户端也做一次归一化如去掉末尾点、转小写便于日志统计和缓存key一致。缓存策略服务端已有缓存24h/1h但对于高频巡检场景建议客户端再叠加一层本地缓存如TLRU减少API调用次数。注意缓存失效时间应与服务端保持一致或略少避免数据陈旧。限流QPS 5/s客户端必须严格限制并发。使用信号量或队列控制。ICPQueryClient类中简单的时间间隔控制适合单线程多线程场景改用rate-limiter库。错误分类网络错误、限流、鉴权、业务错误各自独立异常方便上层统一异常处理如限流时降级、鉴权失败时告警。日志与监控记录每个域名的查询结果、request_id、耗时便于排障。对is_filedfalse的域名可单独打标关注是否应有备案却未备案。幂等性GET请求天然幂等重试时不必担心重复写入。参考文档ICP备案查询API文档原始接口定义Markdown