从curl到工程封装：网页正文提取API接入全指南
背景与适用场景
在信息聚合、内容分析、知识库构建等任务中，常常需要从杂乱的新闻页面、公众号文章或博客中剥离导航、广告、侧栏与评论区，仅保留正文主体。若手动编写XPath/CSS选择器，每换一个网站就要重新适配，维护维护复杂度极高。
网页正文提取API基于文本密度算法，可自动识别并抽取网页的核心内容，同时返回标题、发布时间、配图列表与预估阅读时长。本文将以该API为例，演示从最直接的curl命令行调用到生产级Python封装的完整链路，让读者掌握一种可复用的“先验证、再封装、后集成”的API接入范式。
接口能力边界
在投入封装之前，有必要明确该API的适用边界：
- 输入：一个合法的公开网页URL（HTTP或HTTPS）。
- 输出：结构化JSON，包含
title（标题）、content（正文HTML/纯文本）、publish_time（发布时间）、word_count（字数）、reading_time（预估阅读时长）、images（配图URL数组）、image_count（配图数量）。 - 限制：QPS为5次/秒；单次请求仅处理一个URL；仅支持可公开访问的页面，无法处理需要登录认证或频繁触发反爬机制的站点。
- 不承诺：素材中未提及SLA、可用性百分比、处理成功率或处理速度保证，生产环境中需自行设计降级与重试逻辑。
鉴权方式与请求参数
鉴权
该API使用HTTP Header 鉴权，请求时必须在X-API-Key头中传入有效的API密钥。密钥通常由平台提供，本文不涉及申请流程，假设读者已持有密钥。
Query 参数
| 参数 | 类型 | 必填 | 说明 |
|---|---|---|---|
url | string | 是 | 需要提取正文的目标网页URL，需URL编码 |
请求方法
GET
请求地址
https://v1.apizero.cn/api/content-extract快速体验——curl 示例
将API密钥设置为环境变量（或直接替换字符串），即可通过一行命令验证接口是否正常工作：
export APIZERO_API_KEY="your-api-key-here" curl -sS \ -X GET \ -H "X-API-Key: $APIZERO_API_KEY" \ "https://v1.apizero.cn/api/content-extract?url=https://apizero.cn"说明：
-sS使curl静默输出并显示错误信息；-H添加请求头；- URL参数中
url的值已被自动编码（curl不会对已有URL再次编码，但建议目标URL本身不含空格等特殊字符）； - 返回结果为标准JSON，可使用管道
| jq .进行格式化。
如果一切正常，你将看到类似如下结构的响应（具体内容依目标页面而异）：
{ "code": 0, "msg": "成功", "data": { "title": "Apizero - 聚合API平台", "content": "...（正文HTML或纯文本）...", "publish_time": "2024-01-15", "word_count": 2300, "reading_time": "5分钟", "image_count": 3, "images": [ "https://..." ] } }返回值结构详解
接口始终返回一个顶层对象，字段如下：
| 字段 | 类型 | 说明 |
|---|---|---|
code | int | 业务状态码，0表示成功，非0表示异常 |
msg | string | 与code对应的中文描述 |
data | object/null | 成功时包含提取结果，失败时为null |
data 对象字段
| 字段 | 类型 | 说明 |
|---|---|---|
title | string | 网页标题（head中的title标签或文章中h标签） |
content | string | 提取后的正文，格式为HTML标签包裹的文本（部分场景可能为纯文本） |
publish_time | string | 页面标注的发布时间，格式为YYYY-MM-DD；若无法识别则为空字符串 |
word_count | int | 正文汉字与英文单词的总计字数（近似值） |
reading_time | string | 基于平均阅读速度（约300字/分钟）推算的阅读时长，如"5分钟" |
images | array[string] | 正文内图片的完整URL列表（不含装饰性图标） |
image_count | int | images数组长度 |
注意：content字段可能包含HTML标签，如<p>、<img>等，需要渲染时可直接内嵌，如需纯文本可自行用正则或HTML解析器去除标签。
错误处理与常见问题
HTTP 状态码与业务状态码
| HTTP状态 | 业务码(code) | msg含义 | 典型原因 |
|---|---|---|---|
| 200 | 0 | 成功 | - |
| 400 | 10001 | 参数缺失或格式错误 | 未提供url或URL无效 |
| 401 | 10002 | API Key 无效或缺失 | 请求头中未携带X-API-Key或密钥错误 |
| 429 | 10003 | 请求过于频繁 | 超过QPS 5次/秒限制 |
| 500 | 20001 | 服务内部错误 | 目标网页无法访问或解析失败 |
常见排查切入点
- curl返回401：检查
X-API-Key是否拼写正确，密钥是否过期或被盗用。 - curl返回400：确认URL参数名是否为
url，且值已正确URL编码（如含有&符号时）。 - 返回code=0但content为空：可能目标页面为纯图片、PDF或需JavaScript渲染的单页应用（SPA），文本密度算法无法提取有效正文。此时建议降级为直接抓取页面描述或放弃提取。
- 响应缓慢或超时：目标网页响应慢或服务器连接超时，建议设置客户端超时时间为10-15秒，并考虑将URL排队异步处理。
工程化封装建议
从curl验证到集成到项目，通常需要以下几个步骤：
1. 定义配置与异常
在Python中，首先创建一个封装类：
import requests from typing import Optional, Dict, Any class ContentExtractError(Exception): pass class ContentExtractClient: BASE_URL = "https://v1.apizero.cn/api/content-extract" def __init__(self, api_key: str, timeout: int = 10): self.api_key = api_key self.timeout = timeout self.session = requests.Session() self.session.headers.update({"X-API-Key": api_key}) def extract(self, url: str) -> Dict[str, Any]: """提取网页正文，返回data字段字典""" resp = self.session.get( self.BASE_URL, params={"url": url}, timeout=self.timeout ) if resp.status_code != 200: raise ContentExtractError(f"HTTP {resp.status_code}: {resp.text}") body = resp.json() if body.get("code") != 0: raise ContentExtractError(f"业务错误 [{body['code']}]: {body.get('msg')}") return body["data"]2. 添加重试与降级
考虑到网络波动和限流，可用tenacity库实现指数退避重试：
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type class ContentExtractClient: # ... 上述代码 ... @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), retry=retry_if_exception_type((requests.ConnectionError, requests.Timeout, ContentExtractError)) ) def extract_with_retry(self, url: str) -> Dict[str, Any]: return self.extract(url)注意：对429限流错误可特别处理，例如增加延时后重试。
3. 异步并发控制
如果需要批量处理多个URL，应使用信号量控制并发数不超过QPS（5）。以下使用asyncio+aiohttp示例：
import asyncio import aiohttp class AsyncContentExtractClient: def __init__(self, api_key: str, max_concurrent: int = 5): self.api_key = api_key self.semaphore = asyncio.Semaphore(max_concurrent) async def extract_one(self, session: aiohttp.ClientSession, url: str) -> dict: async with self.semaphore: headers = {"X-API-Key": self.api_key} params = {"url": url} async with session.get(self.BASE_URL, headers=headers, params=params) as resp: body = await resp.json() if body.get("code") != 0: raise ContentExtractError(...) return body["data"] async def extract_many(self, urls: list[str]) -> list[dict]: async with aiohttp.ClientSession() as session: tasks = [self.extract_one(session, url) for url in urls] return await asyncio.gather(*tasks, return_exceptions=True)4. 缓存与幂等考虑
对于相同URL的重复提取，可考虑本地缓存（如functools.lru_cache或 Redis），避免消耗API配额。注意缓存键应包含URL（保留原始字符顺序）。
参考文档
- 网页正文提取 API 官方文档
- 原始 Markdown 文档