从 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 对象参数如下字段类型必填说明示例值platformstring否平台筛选。值为all全部、weibo、zhihu、bilibili、tieba多个用逗号分隔如weibo,zhihu。默认all。weibo,zhihulimitnumber否每个平台返回的热搜条数最大 50默认值以文档为准。10timeoutnumber否请求超时秒数建议设置 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:0008: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, jsonpayload, timeouttimeout or self.timeout) resp.raise_for_status() data resp.json() if data.get(code) ! 0: raise ValueError(fAPI 返回业务错误: {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(platformall, limit5) 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_keyYOUR_API_KEY) try: result client.fetch(platformweibo,zhihu, limit3, timeout10) 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:0008: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:...}] } } } }code0 表示成功非 0 表示错误具体含义见文档。request_id每次请求的唯一标识用于日志追踪。data.requested_platforms本次实际请求的平台列表。data.failed_platforms返回失败的平台列表及其错误信息为空对象则表示全部成功。data.platforms每个平台一个对象status为success或error此时items为空数组。items中每个元素包含rank排名1 开始、title热搜标题、hot热度值字符串。常见 HTTP 状态码与错误处理状态码含义处理建议200正常返回解析 JSON判断code是否为 0401鉴权失败检查AuthorizationHeader 是否正确设置429请求频率超过 QPS 限制3次/秒加入 sleep 或使用令牌桶限流5xx服务端错误根据重试策略进行指数退避重试最多 3 次业务错误码code非 0常见场景无效平台参数如拼写错误limit超过 50timeout为非数字建议对所有可能的code枚举进行容错避免强依赖业务逻辑。工程化注意事项1. 鉴权安全不要在代码中硬编码 API Key应通过环境变量或配置中心注入。例如os.getenv(HOT_SEARCH_API_KEY)。2. 限流与重试策略QPS 只有 3若需高频轮询建议使用协程并在请求间加asyncio.sleep(0.35)或使用aiolimiter等限流库。网络错误超时、连接重置应配合幂等机制重试注意重试次数不要超过合理范围。3. 日志与监控记录request_id与响应耗时便于排查。对于status为error的平台需要输出告警日志。监控接口成功率设定告警阈值。4. 数据缓存策略热搜数据通常分钟级更新若不需要实时刷洗可设置本地内存缓存如 TTL60s减少对上游请求压力。缓存 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, jsonpayload, headers{Authorization: client.api_key} )) responses await asyncio.gather(*tasks, return_exceptionsTrue) return [r.json() if isinstance(r, httpx.Response) else None for r in responses]注意单次 API 调用已聚合四个平台除非业务需要不同平台不同limit或不同轮询频率否则建议直接使用一次all请求更高效。参考文档全网热搜聚合 API 文档页原始接口文档markdown本文所有接口参数均以官方文档为准如遇不一致请优先参考文档。