当前位置：首页 > news >正文

用Python调用百度热搜榜API：从零实现实时热搜数据抓取与可视化

news 2026/6/29 22:22:28

引言

在信息爆炸的时代，实时掌握热点话题对内容创作、舆情分析、营销决策等场景至关重要。百度热搜榜作为国内最具影响力的热词榜单之一，其数据具有极高的时效性和参考价值。然而，直接爬取百度页面存在反爬难、结构不稳定等问题。通过专业的聚合API平台（如极数本源 ApiZero），我们可以用稳定、合法的方式获取结构化热搜数据。

本文将以百度热搜榜API为例，演示从申请API密钥、编写请求代码、解析响应数据到可视化展示的完整流程。所有代码均基于Python 3.10+编写，使用requests和matplotlib库，确保可直接运行。

准备工作

1. 注册并获取API密钥

前往极数本源 API平台注册账号，在“API商城”中搜索“百度热搜榜”接口。申请后可在控制台获取专属的API Key（示例：sk_your_api_key_here）。注意：不同接口可能有不同的鉴权方式，本文基于平台常见的key参数传值。

2. 安装必要的Python库

pip install requests matplotlib pandas

requests: 发送HTTP请求
matplotlib: 数据可视化
pandas: 辅助数据分析（可选）

请求百度热搜榜API

接口说明

根据平台文档，百度热搜榜API的请求方式如下：

请求方法：GET
请求URL：https://api.apizero.cn/v1/hot/baidu（仅为示例路径，实际以平台文档为准）
请求参数：
- key：你的API密钥（必填）
- num：返回条数（可选，默认20，最大50）
响应格式：application/json

编写请求函数

import requests # 请替换为你的真实key API_KEY = "sk_your_api_key_here" URL = "https://api.apizero.cn/v1/hot/baidu" def fetch_baidu_hot(key: str, count: int = 30) -> dict: """ 获取百度热搜榜数据 :param key: API密钥 :param count: 返回条数（1-50） :return: 解析后的JSON字典 """ params = { "key": key, "num": count } try: response = requests.get(URL, params=params, timeout=10) response.raise_for_status() # 检查HTTP状态码 return response.json() except requests.exceptions.RequestException as e: print(f"请求出错：{e}") return None # 测试 if __name__ == "__main__": data = fetch_baidu_hot(API_KEY) if data: print(data)

解析响应数据

正常响应会返回类似如下结构（字段名基于常见API设计，实际可能略有不同，需参照文档调整）：

{ "code": 200, "message": "success", "data": [ { "rank": 1, "keyword": "高考成绩公布", "heat": 9854321, "url": "https://www.baidu.com/s?wd=高考成绩公布" }, { "rank": 2, "keyword": "中美会谈最新消息", "heat": 8732100, "url": "..." } ] }

其中heat表示热度值（单位通常为“次”），rank为排名。

编写解析函数提取关键词与热度：

def parse_hot_data(raw: dict) -> list: """ 从API响应中提取热搜列表 :param raw: API返回的字典 :return: 列表，每个元素为 (rank, keyword, heat) """ if raw.get("code") != 200: print("API返回异常：", raw.get("message")) return [] items = raw.get("data", []) result = [] for item in items: result.append(( item.get("rank", 0), item.get("keyword", ""), int(item.get("heat", 0)) )) return result

数据可视化：动态条形图展示Top热搜

使用matplotlib绘制横向条形图，直观展示当前热点热度对比。

import matplotlib.pyplot as plt import numpy as np def plot_hot_bar(data: list, top_n: int = 10): """ 绘制热搜Top-N条形图 :param data: [(rank, keyword, heat), ...] :param top_n: 展示前N条 """ # 取前top_n条，并反转以便排名最前的在顶部 sorted_data = sorted(data, key=lambda x: x[2], reverse=True)[:top_n] # 反转顺序使排名1在顶部 sorted_data.reverse() ranks, keywords, heats = zip(*sorted_data) # 创建颜色映射：热度越高颜色越深 norm = plt.Normalize(min(heats), max(heats)) colors = plt.cm.Blues(norm(heats)) fig, ax = plt.subplots(figsize=(10, 6)) bars = ax.barh(keywords, heats, color=colors) # 添加数值标签 for bar, heat in zip(bars, heats): ax.text(bar.get_width() + 0.02 * max(heats), bar.get_y() + bar.get_height()/2, f'{heat:,}', va='center', fontsize=9) ax.set_xlabel('热度值') ax.set_title(f'百度热搜榜 Top {top_n}（实时）', fontsize=14) ax.tick_params(axis='y', labelsize=10) plt.tight_layout() plt.show() # 调用 data = fetch_baidu_hot(API_KEY, count=30) hot_list = parse_hot_data(data) if hot_list: plot_hot_bar(hot_list, top_n=15)

运行后应生成类似下图的条形图（由于实时数据变化，具体关键词不同）：

（实际运行时请替换为本地生成的图片或留空，此处仅为说明）

进阶技巧

1. 定时任务

使用schedule库或系统cron每5分钟抓取一次，保存到数据库或CSV，用于趋势分析。

import schedule import time def job(): data = fetch_baidu_hot(API_KEY, count=50) hot_list = parse_hot_data(data) # 追加到CSV文件 import csv from datetime import datetime with open('baidu_hot_history.csv', 'a', newline='', encoding='utf-8') as f: writer = csv.writer(f) for _, keyword, heat in hot_list: writer.writerow([datetime.now().isoformat(), keyword, heat]) print(f"{datetime.now()}: 数据已保存") schedule.every(5).minutes.do(job) while True: schedule.run_pending() time.sleep(1)

2. 数据清洗与去重

由于同一关键词可能在不同时刻出现，可依据关键词+时间戳构建唯一索引。使用pandas处理：

import pandas as pd df = pd.read_csv('baidu_hot_history.csv', names=['timestamp','keyword','heat']) df['time'] = pd.to_datetime(df['timestamp']) # 去重：保留每个关键词的最新记录 df_latest = df.sort_values('time').drop_duplicates(subset='keyword', keep='last') print(df_latest.head())

3. 错误处理与重试

API可能因网络波动返回非200状态码，可加入重试机制：

from tenacity import retry, stop_after_attempt, wait_fixed @retry(stop=stop_after_attempt(3), wait=wait_fixed(2)) def fetch_with_retry(key, count=30): return fetch_baidu_hot(key, count)