从 curl 到工程封装：构建全网热搜数据聚合层-平芜编程栈

适用场景

全媒体舆情监测、热点事件追踪、内容运营选题挖掘等场景，都需要跨平台获取实时热搜数据。传统做法是逐一调用各平台自有接口，面临鉴权不同、限流分散、数据结构不统一等问题。全网热搜聚合 API 通过一次 POST 请求，同时返回微博、知乎、B站、贴吧四个平台的热搜列表，大幅降低集成复杂度。

接口能力边界

请求方法：POST
接口地址：https://v1.apizero.cn/api/hot-search
分类：内容娱乐
QPS 上限：3 次/秒（超过会被限流，建议客户端做退避重试）
支持平台：微博（weibo）、知乎（zhihu）、B站（bilibili）、百度贴吧（tieba）
单次可查询平台数：支持逗号分隔指定多个平台，或使用all查询全部四个
单平台最大返回条数：limit参数最大值为 50，超过会被截断为 50

请求参数与鉴权

接口要求POST请求，请求体为 JSON 对象，参数如下：

字段	类型	必填	说明	示例值
platform	string	否	平台筛选。值为`all`（全部）、`weibo`、`zhihu`、`bilibili`、`tieba`，多个用逗号分隔，如`"weibo,zhihu"`。默认`all`。	`"weibo,zhihu"`
limit	number	否	每个平台返回的热搜条数，最大 50，默认值以文档为准。	`10`
timeout	number	否	请求超时秒数，建议设置 10–30 秒，防止跨平台聚合时个别平台响应慢导致整体超时。	`15`

鉴权方式：在 HTTP Header 中传入Authorization，值为 API Key。示例：

Authorization: your-api-key-here

注意：API Key 需要向服务提供方申请，本文不涉及申请流程。

快速验证：curl 示例

使用 curl 进行接口联通性测试，是最直接的验证方式。请将YOUR_API_KEY替换为实际密钥。

curl -sS -X POST \ -H "Authorization: YOUR_API_KEY" \ -H "Content-Type: application/json" \ -d '{"platform": "all", "limit": 5, "timeout": 10}' \ "https://v1.apizero.cn/api/hot-search"

返回 JSON 示例（简要）：

{ "code": 0, "msg": "成功", "request_id": "req_abc123", "data": { "generated_at": "2026-05-08T13:00:00+08:00", "requested_platforms": ["weibo","zhihu","bilibili","tieba"], "limit_per_platform": 5, "total_items": 20, "failed_platforms": {}, "platforms": { "weibo": { "name": "微博热搜", "status": "success", "count": 5, "items": [ {"rank": 1, "title": "热搜标题1", "hot": "5234567"}, {"rank": 2, "title": "热搜标题2", "hot": "4234567"} ] } } } }

注意：实际hot字段值代表热度数值（字符串），rank为排名，从 1 开始递增。若平台返回失败，status为"error"且items为空数组。

推荐语言封装：Python 示例

curl 适合测试，但在工程中我们需要稳健的封装。以 Python 为例，推荐使用requests库，并结合重试、超时、日志等机制。

import requests import time from typing import Optional, Dict, Any class HotSearchClient: """全网热搜聚合 API 封装""" BASE_URL = "https://v1.apizero.cn/api/hot-search" def __init__(self, api_key: str, timeout: int = 15, max_retries: int = 3): self.api_key = api_key self.timeout = timeout self.max_retries = max_retries self.session = requests.Session() self.session.headers.update({ "Authorization": api_key, "Content-Type": "application/json" }) def fetch(self, platform: str = "all", limit: int = 10, timeout: Optional[int] = None) -> Dict[str, Any]: """ 获取热搜数据 :param platform: 平台筛选，如 "all", "weibo", "weibo,zhihu" :param limit: 每个平台返回条数 :param timeout: 请求超时（秒），覆盖默认值 :return: 解析后的 JSON 响应字典 :raises: requests.exceptions.RequestException 或 ValueError（非 JSON） """ payload = { "platform": platform, "limit": limit, "timeout": timeout or self.timeout } last_exception = None for attempt in range(1, self.max_retries + 1): try: resp = self.session.post(self.BASE_URL, json=payload, timeout=timeout or self.timeout) resp.raise_for_status() data = resp.json() if data.get("code") != 0: raise ValueError(f"API 返回业务错误: {data.get('msg')}") return data except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: last_exception = e if attempt < self.max_retries: wait = 2 ** attempt # 指数退避 print(f"请求失败 (尝试 {attempt}/{self.max_retries}), {wait}s 后重试: {e}") time.sleep(wait) else: raise last_exception except Exception as e: # 非网络错误直接抛出，不重试 raise e # 不会走到这里 raise RuntimeError("Unexpected exit") def get_all_platforms_summary(self) -> Dict[str, Any]: """获取全部平台的热搜摘要（只返回标题和排名）""" raw = self.fetch(platform="all", limit=5) platforms = raw["data"]["platforms"] summary = {} for key, info in platforms.items(): if info["status"] == "success": summary[key] = [(item["rank"], item["title"]) for item in info["items"]] return summary if __name__ == "__main__": # 注意：需要替换为真实的 API Key client = HotSearchClient(api_key="YOUR_API_KEY") try: result = client.fetch(platform="weibo,zhihu", limit=3, timeout=10) print(f"请求ID: {result['request_id']}") for plat, info in result["data"]["platforms"].items(): if info["status"] == "success": print(f"{info['name']} 共 {info['count']} 条:") for item in info["items"]: print(f" #{item['rank']} {item['title']} (热度 {item['hot']})") except Exception as e: print(f"请求异常: {e}")

封装要点：

使用requests.Session重用连接，减少握手开销。
支持指数退避重试，仅对网络类异常重试，业务错误（如鉴权失败）直接抛出。
提供timeout覆盖，防止接口挂死。
顶层异常全部捕获，便于调用方统一处理。

响应数据结构解读

成功响应格式：

{ "code": 0, "msg": "成功", "request_id": "req_abc123", "data": { "generated_at": "2026-05-08T13:00:00+08:00", "requested_platforms": ["weibo","zhihu"], "limit_per_platform": 10, "total_items": 20, "failed_platforms": {}, "platforms": { "weibo": { "name": "微博热搜", "status": "success", "count": 10, "items": [{"rank":1, "title":"...", "hot":"..."}] } } } }

code：0 表示成功，非 0 表示错误，具体含义见文档。
request_id：每次请求的唯一标识，用于日志追踪。
data.requested_platforms：本次实际请求的平台列表。
data.failed_platforms：返回失败的平台列表及其错误信息，为空对象则表示全部成功。
data.platforms：每个平台一个对象，status为"success"或"error"（此时items为空数组）。
items中每个元素包含rank（排名，1 开始）、title（热搜标题）、hot（热度值，字符串）。

常见 HTTP 状态码与错误处理

状态码	含义	处理建议
200	正常返回	解析 JSON，判断`code`是否为 0
401	鉴权失败	检查`Authorization`Header 是否正确设置
429	请求频率超过 QPS 限制（3次/秒）	加入 sleep 或使用令牌桶限流
5xx	服务端错误	根据重试策略进行指数退避重试，最多 3 次

业务错误码（code非 0）常见场景：

无效平台参数（如拼写错误）
limit超过 50
timeout为非数字

建议对所有可能的code枚举进行容错，避免强依赖业务逻辑。

工程化注意事项

1. 鉴权安全

不要在代码中硬编码 API Key，应通过环境变量或配置中心注入。
例如：os.getenv("HOT_SEARCH_API_KEY")。

2. 限流与重试策略

QPS 只有 3，若需高频轮询，建议使用协程并在请求间加asyncio.sleep(0.35)或使用aiolimiter等限流库。
网络错误（超时、连接重置）应配合幂等机制重试，注意重试次数不要超过合理范围。

3. 日志与监控

记录request_id与响应耗时，便于排查。
对于status为"error"的平台，需要输出告警日志。
监控接口成功率，设定告警阈值。

4. 数据缓存策略

热搜数据通常分钟级更新，若不需要实时刷洗，可设置本地内存缓存（如 TTL=60s），减少对上游请求压力。
缓存 key 建议包含platform和limit组合。

5. 异步请求优化（选读）

若使用 Python asyncio，可借助aiohttp或httpx.AsyncClient并发发起请求（但注意接口本身已聚合多个平台，一般只需单次调用）。
如果业务需要同时查询多个platform组合，可并发调用：

import asyncio import httpx async def fetch_platforms(client: HotSearchClient, platforms: list): async with httpx.AsyncClient() as http_client: tasks = [] for plat in platforms: payload = {"platform": plat, "limit": 5, "timeout": 10} tasks.append(http_client.post( client.BASE_URL, json=payload, headers={"Authorization": client.api_key} )) responses = await asyncio.gather(*tasks, return_exceptions=True) return [r.json() if isinstance(r, httpx.Response) else None for r in responses]

注意：单次 API 调用已聚合四个平台，除非业务需要不同平台不同limit或不同轮询频率，否则建议直接使用一次all请求更高效。