好看视频作为百度旗下的短视频 / 中长视频平台，汇聚了海量生活、科技、娱乐、知识等领域的优质视频内容。item_search_video 接口（基于平台合规数据服务封装）是通过关键词批量获取视频列表的核心工具，支持按关键词、发布时间、视频时长、播放量、作者类型等多维度筛选，返回视频标题、播放链接、作者信息、互动数据等关键内容，广泛应用于内容聚合、舆情监测、行业分析、短视频二次创作素材调研等场景。
本攻略结合好看视频平台特性（百度生态赋能、内容覆盖全品类、数据合规性要求高），从接口认知、前置准备、实操落地、调试优化到进阶技巧，全方位拆解对接流程，兼顾入门易用性与生产级稳定性，帮助开发者高效完成接口对接。
一、接口核心认知：先明确 “能做什么”“适配什么场景”
- 接口定位与核心价值
核心功能：输入关键词（支持多关键词组合），筛选好看视频平台的视频列表，支持按发布时间、视频时长、播放量、点赞量、作者认证类型等条件过滤，返回分页视频数据（含视频基础信息、互动数据、作者信息），可联动 item_get_video 接口获取视频详情（如完整播放地址、字幕、分镜信息）；
平台特性：
数据覆盖：收录好看视频 90% 以上公开视频，涵盖短视频（≤3 分钟）、中长视频（3-30 分钟）、长视频（＞30 分钟），新视频收录延迟 5-10 分钟；
筛选能力：支持精确 / 模糊匹配、多关键词组合（AND/OR 逻辑）、时间范围（近 1 天 / 7 天 / 30 天 / 自定义）、视频时长区间（如 1-5 分钟）、播放量排序、作者类型（个人 / 企业 / 官方）；
百度生态联动：部分服务商支持同步返回视频在百度搜索的曝光量、相关关键词排名，适配 SEO / 舆情分析场景；
无需官方授权：依赖第三方合规服务商封装能力，无需好看视频创作者 / 开放平台账号，仅需服务商提供的密钥；
典型应用：
内容聚合：垂直领域关键词（如 “Python 教程”“家常菜做法”）采集，搭建领域视频资讯平台；
舆情监测：按品牌名、事件关键词（如 “某品牌 质量问题”）实时追踪相关视频传播；
行业分析：按行业关键词（如 “新能源汽车 评测”）采集视频，统计热门话题、用户偏好；
素材调研：短视频创作者通过关键词（如 “旅行 vlog 素材”）采集参考视频，激发创作灵感；
竞品监测：输入竞品品牌 / 产品关键词，分析竞品在视频平台的宣传策略、用户反馈。 - 核心参数与返回字段（关键词搜索场景适配版）
（1）请求参数（必填 + 可选，按优先级排序）
参数名称 类型 是否必填 说明 应用场景示例
appkey string 是 接口调用密钥，由合规服务商分配（注册并通过审核后获取） haokan_video_abc123
secret string 是 签名密钥，用于请求合法性校验（不可泄露，定期轮换） haokan_video_def456
keyword string 是 搜索关键词（支持多关键词组合，AND 用空格分隔，OR 用
time_range string 否 发布时间范围筛选，默认 “all”（全部时间） all（全部）、1day（1 天内）、7days（7 天内）、30days（30 天内）、90days（90 天内）、custom（自定义，需配合 start_date/end_date）
start_date string 否 自定义时间范围开始日期（time_range=custom 时必填），格式 “YYYY-MM-DD” 2024-01-01
end_date string 否 自定义时间范围结束日期（time_range=custom 时必填），格式 “YYYY-MM-DD” 2024-12-31
duration_min int 否 视频最小时长（单位：秒），默认 0（无下限） 60（筛选 1 分钟以上视频）
duration_max int 否 视频最大时长（单位：秒），默认无上限 300（筛选 5 分钟以下视频）
sort_type string 否 排序方式，默认 “relevance”（相关度优先） relevance（相关度）、play_count（播放量倒序）、like_count（点赞量倒序）、pub_time（发布时间倒序）、comment_count（评论数倒序）
author_type string 否 作者认证类型筛选，默认 “all”（全部） all（全部）、personal（个人认证）、enterprise（企业认证）、official（官方认证，如政府 / 媒体）
is_original int 否 是否原创视频，默认 0（全部） 0（全部）、1（仅原创）、2（仅非原创）
page_no int 否 页码，默认 1（支持分页，最大页码通常≤100，不同服务商限制不同） 1、2、3
page_size int 否 每页视频条数，默认 20（最大支持 50 条 / 页，企业版可申请 100 条） 20、30、50
match_type string 否 关键词匹配类型，默认 “fuzzy”（模糊匹配） fuzzy（模糊匹配，视频标题 / 描述 / 标签含关键词）、exact（精确匹配，视频标题完整包含关键词）
need_baidu_rank int 否 是否返回百度搜索排名数据，默认 0（不返回） 1（返回百度搜索曝光量、排名）、0（仅返回视频基础数据）
timestamp long 是 请求时间戳（毫秒级，有效期 5 分钟，避免重复请求） 1735689600000
sign string 是 签名值（按服务商规则加密生成，核心校验项） 32 位 MD5 大写串（如 A8F7C3D2E1B0967453120FEDCBA9876）
注：
时间范围优先级：time_range=custom 时，start_date 和 end_date 必须同时传入，且 end_date≥start_date，时间跨度最大支持 1 年；
视频时长筛选：duration_min 和 duration_max 可单独传入（仅限制下限 / 上限），也可同时传入（区间筛选）；
部分服务商支持高级筛选参数（如 “视频标签”“是否带字幕”“分辨率”），需以实际文档为准。
（2）返回核心字段（按业务场景分类，关键词搜索重点标注）
视频基础信息：视频 ID（video_id，好看视频唯一标识，用于item_get_video接口）、视频标题（title，含核心卖点）、视频封面图（cover_url，高清封面 URL）、视频时长（duration，单位：秒，转换为 “分：秒” 格式）、发布时间（pub_time，精确到秒）、视频类型（video_type：short/medium/long，对应短 / 中 / 长视频）、原创标识（is_original，true/false）、视频标签（tags，如 “Python”“美食”“旅行”）、视频描述（description，视频简介）、分辨率（resolution，如 “1080P”“720P”，部分服务商支持）；
互动数据：播放量（play_count，累计播放次数）、点赞量（like_count）、评论数（comment_count）、转发量（share_count）、收藏数（collect_count）、弹幕数（danmu_count，部分服务商支持）；
作者信息：作者 ID（author_id）、作者名称（author_name）、作者头像（author_avatar）、作者认证类型（author_type，如 “企业认证”）、作者粉丝数（author_fans，部分服务商支持）、作者简介（author_intro，部分服务商支持）；
百度生态数据（need_baidu_rank=1时返回）：百度搜索曝光量（baidu_exposure）、百度相关关键词排名（baidu_rank，如 “Python 教程” 关键词排名 TOP10）、百度索引状态（baidu_indexed，是否被百度收录）；
扩展信息：视频播放链接（play_url，可直接跳转播放）、是否带字幕（has_subtitle，true/false）、是否支持下载（can_download，true/false）、相关度评分（relevance_score，0-100 分）、视频状态（is_available，true/false，是否正常播放，未下架 / 未违规）。 - 接口限制与注意事项
调用频率：普通用户 10-30 次 / 分钟，企业用户 50-200 次 / 分钟（不同服务商限制不同，需提前确认）；
数据缓存：搜索结果缓存 30 分钟 - 1 小时，热门关键词缓存可能缩短至 10 分钟，企业版支持refresh=1强制刷新（需申请权限）；
权限差异：
权限类型 核心差异 适用场景
普通用户 基础筛选（时间、时长、排序）、基础字段、单页最大 20 条、低调用频率 个人素材调研、小型分析
企业用户 高级筛选（标签、分辨率、是否带字幕）、完整字段（百度生态数据、作者粉丝数）、单页最大 100 条、高调用频率、强制刷新 商业舆情监测、行业分析、内容聚合平台
内容限制：
已下架 / 违规视频、隐私作者视频、未被服务商收录的视频无法返回；
部分原创视频可能因版权限制，仅返回基础信息（标题、封面、播放链接），完整视频需跳转至好看视频平台观看；
合规要求：
必须使用合规服务商接口，禁止自行爬取好看视频数据（违反《反不正当竞争法》及平台规则）；
数据仅可用于自身合规业务，禁止商业化售卖、恶意传播或侵犯视频作者版权；
引用视频数据时需标注 “数据来源：好看视频”，视频内容二次使用需获得作者授权；
不得用于批量抓取用户隐私、敏感信息或从事违法活动（如传播不良内容、恶意诋毁）。
二、对接前准备：3 步搞定前置条件 - 获取接口密钥（核心步骤）
好看视频关键词视频列表接口无官方公开申请渠道，需通过合规第三方服务商（如聚合数据、APISpace、阿里云 API 市场、百度智能云市场等）获取，步骤如下：
选择合规服务商：优先选择有百度生态合作资质、口碑好、支持企业认证的服务商，要求提供《数据合规承诺书》《服务协议》，避免使用非法爬虫接口（存在法律风险）；
注册与认证：
个人用户：注册账号，完成实名认证（提供身份证信息），获取基础接口权限；
企业用户：注册账号，完成企业认证（提供营业执照、法人信息、业务用途说明），获取高级接口权限；
申请接口权限：搜索 “好看视频关键词搜索接口” 或 “好看视频 item_search_video 接口”，选择对应接口，提交申请（需明确业务用途，如 “内容聚合平台开发”“舆情监测系统搭建”）；
获取密钥：审核通过后，在服务商后台 “应用管理” 中获取 appkey 和 secret（部分服务商按调用次数收费，需提前充值或购买套餐）；
查阅文档：下载服务商提供的接口文档，重点确认：
参数格式（如time_range取值、视频时长单位）；
签名生成规则（MD5/SHA256、参数排序要求）；
返回字段含义（不同服务商字段命名可能差异）；
调用频率、分页上限、权限限制等细节。 - 技术环境准备
（1）支持语言与协议
接口采用 HTTPS 协议，支持所有主流开发语言：Python、Java、PHP、Go、Node.js 等，无框架限制，推荐使用 Python（数据处理、批量请求效率高，适配关键词搜索的多条件筛选场景）。
（2）必备工具与依赖
调试工具：
Postman（快速验证接口可用性，批量测试参数组合）；
curl（命令行调试，快速排查网络问题）；
浏览器开发者工具（分析好看视频搜索结果页结构，辅助字段映射）；
在线 MD5/SHA256 工具（验证签名生成正确性）；
开发依赖：
网络请求：requests（Python）、OkHttp（Java）、axios（Node.js）；
加密工具：语言内置 MD5/SHA256 库（签名生成用，按服务商规则选择）；
数据处理：json（解析响应数据）、pandas（批量整理视频列表）、datetime（时间格式转换）、time（时长格式转换）；
辅助工具：
日志库（如 logging，记录请求 / 响应 / 错误，便于追溯）；
Redis（缓存搜索结果，减少重复请求）；
定时任务框架（如 APScheduler，实时关键词监测）；
异常监控工具（如 Sentry，生产级报错追踪）；
多线程 / 异步库（如 aiohttp、concurrent.futures，提升批量分页请求效率）；
视频链接解析工具（如 requests-html，如需提取视频真实播放地址，需合规使用）。 - 业务需求梳理
关键词策略：明确核心关键词及组合规则（如舆情监测需 “品牌名 + 负面关键词”，素材调研需 “领域 + 素材类型”），避免关键词过于宽泛导致结果冗余；
筛选条件定义：
时间范围：实时监测用 “1day/7days”，历史数据采集用 “custom” 自定义时间；
视频时长：短视频聚合选 “duration_max=300”（5 分钟内），知识类内容选 “duration_min=600”（10 分钟以上）；
排序方式：爆款分析选 “play_count”（播放量倒序），新品追踪选 “pub_time”（发布时间倒序）；
字段筛选：仅保留业务必需字段（如内容聚合需 “title、cover_url、play_url、duration、pub_time”，舆情监测需 “author_name、play_count、comment_count、description”），减少数据传输量；
分页与批量需求：确认是否需要获取全部搜索结果（如行业分析），或仅需前 N 页（如素材调研），避免无效分页请求；
异常场景预设：关键词无匹配结果、频率超限、网络波动、视频已下架等场景，需设计降级方案（如返回 “无相关视频” 提示、缓存历史搜索结果）。
三、实操步骤：从调试到落地（Python 示例）
步骤 1：理解请求流程
拼接关键词、筛选条件等核心参数；
按服务商规则生成签名（sign），确保请求合法性；
发送 POST/GET 请求（多数服务商推荐 POST，参数传输更安全）；
接收响应数据，解析 JSON 格式结果，提取视频列表；
数据标准化处理（时间格式统一、时长格式转换、字段映射）；
按需分页请求（遍历所有页码，获取完整结果）；
异常处理（签名错误、权限不足、无结果等）。
步骤 2：签名生成规则（关键！不同服务商可能差异，以实际文档为准）
多数合规服务商采用以下通用签名规则（若有差异，需按服务商文档调整）：
按参数名ASCII 升序排序所有请求参数（不含sign字段）；
将排序后的参数拼接为 “key1=value1&key2=value2&...” 格式（中文 / 特殊字符需 URL 编码）；
在拼接字符串末尾追加 &secret=你的secret；
对拼接后的字符串进行MD5 加密（32 位大写） 或 SHA256 加密（64 位大写），结果即为sign。
签名示例（参数排序与拼接）
假设请求参数：
appkey=haokan_video_abc123
keyword=Python 教程 零基础
time_range=30days
duration_min=60
duration_max=300
sort_type=play_count
page_no=1
page_size=20
timestamp=1735689600000
secret=haokan_video_def456
排序后参数：appkey、duration_max、duration_min、keyword、page_no、page_size、sort_type、time_range、timestamp；
拼接字符串（中文已 URL 编码）
步骤 3：完整代码实现（Python）
（1）依赖安装
bash
运行
pip install requests pandas aiohttp apscheduler # requests：网络请求；pandas：数据整理；aiohttp：异步请求；APScheduler：定时任务
（2）完整代码（含签名生成、接口调用、批量分页、数据保存）
```jsimport requests
import hashlib
import time
import json
import pandas as pd
from urllib.parse import urlencode, unquote
from typing import Dict, List, Optional
import logging
from apscheduler.schedulers.blocking import BlockingScheduler
import aiohttp
import asyncio
配置日志（记录接口调用、错误信息，便于合规追溯）
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
handlers=[logging.FileHandler("haokan_item_search_video.log"), logging.StreamHandler()]
)
接口核心配置（替换为服务商提供的appkey、secret、API_URL）
APP_KEY = "你的appkey"
SECRET = "你的secret"
API_URL = "https://api.haokan-search.com/item_search_video" # 服务商接口地址（以实际为准）
SAVE_PATH = "好看视频列表.xlsx" # 数据保存路径
def generate_sign(params: Dict) -> str:
"""生成接口签名（通用MD5版，若服务商要求SHA256可调整）"""
# 1. 按参数名ASCII升序排序（排除sign字段）
sorted_params = sorted(params.items(), key=lambda x: x[0])
# 2. 拼接参数字符串（urlencode自动处理中文、特殊字符）
param_str = urlencode(sorted_params, encoding="utf-8") + f"&secret={SECRET}"
# 3. MD5加密（32位大写）
md5 = hashlib.md5()
md5.update(param_str.encode("utf-8"))
return md5.hexdigest().upper()
def format_duration(seconds: int) -> str:
"""将秒数转换为“分:秒”格式（如125秒→2:05）"""
if not seconds or seconds < 0:
return "0:00"
minutes = seconds // 60
secs = seconds % 60
return f"{minutes}:{secs:02d}"
def standardize_data(raw_video: Dict) -> Dict:
"""标准化视频数据（统一字段格式，适配业务展示/分析）"""
# 解析作者数据
author_data = raw_video.get("author_info", {}) or raw_video
# 标准化时间格式
pub_time = raw_video.get("pub_time", 0)
pub_time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(pub_time/1000)) if pub_time else ""
# 标准化视频时长
duration = raw_video.get("duration", 0)
duration_str = format_duration(duration)
# 标准化视频类型
video_type = "短视频" if duration <= 180 else "中长视频" if duration <= 1800 else "长视频"
return {
"请求时间": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
"关键词": raw_video.get("keyword", ""),
"视频ID": raw_video.get("video_id", ""),
"视频标题": raw_video.get("title", ""),
"视频封面": raw_video.get("cover_url", ""),
"视频时长": duration_str,
"视频类型": video_type,
"发布时间": pub_time_str,
"是否原创": "是" if raw_video.get("is_original", False) else "否",
"视频标签": ",".join(raw_video.get("tags", [])) if raw_video.get("tags") else "",
"视频描述": raw_video.get("description", ""),
"分辨率": raw_video.get("resolution", "未知"),
"是否带字幕": "是" if raw_video.get("has_subtitle", False) else "否",
"播放量": raw_video.get("play_count", 0),
"点赞量": raw_video.get("like_count", 0),
"评论数": raw_video.get("comment_count", 0),
"转发量": raw_video.get("share_count", 0),
"收藏数": raw_video.get("collect_count", 0),
"作者ID": author_data.get("author_id", ""),
"作者名称": author_data.get("author_name", ""),
"作者认证类型": author_data.get("author_type", ""),
"作者粉丝数": author_data.get("author_fans", 0),
"作者简介": author_data.get("author_intro", ""),
"播放链接": raw_video.get("play_url", ""),
"是否支持下载": "是" if raw_video.get("can_download", False) else "否",
"百度搜索曝光量": raw_video.get("baidu_exposure", 0),
"百度关键词排名": raw_video.get("baidu_rank", "无"),
"相关度评分": raw_video.get("relevance_score", 0),
"视频状态": "正常播放" if raw_video.get("is_available", True) else "已下架/违规"
}
def get_video_list_sync(
keyword: str,
time_range: str = "all",
start_date: Optional[str] = None,
end_date: Optional[str] = None,
duration_min: Optional[int] = None,
duration_max: Optional[int] = None,
sort_type: str = "relevance",
author_type: str = "all",
is_original: int = 0,
page_no: int = 1,
page_size: int = 20,
match_type: str = "fuzzy",
need_baidu_rank: int = 0
) -> Dict:
"""
同步调用item_search_video接口，获取单页好看视频列表
:param keyword: 搜索关键词（支持多关键词组合）
:param time_range: 时间范围
:param start_date: 自定义开始日期（time_range=custom时必填）
:param end_date: 自定义结束日期（time_range=custom时必填）
:param duration_min: 视频最小时长（秒）
:param duration_max: 视频最大时长（秒）
:param sort_type: 排序方式
:param author_type: 作者认证类型
:param is_original: 是否原创（0=全部，1=仅原创，2=仅非原创）
:param page_no: 页码
:param page_size: 每页条数
:param match_type: 关键词匹配类型
:param need_baidu_rank: 是否返回百度排名数据（1=是，0=否）
:return: 标准化后的视频列表结果
"""
# 1. 校验参数合法性
if time_range == "custom" and not (start_date and end_date):
logging.error("time_range=custom时，start_date和end_date为必填参数")
return {"success": False, "error_msg": "缺少自定义时间范围参数", "error_code": -1}
# 2. 构建基础参数（必填项）
params = {
"appkey": APP_KEY,
"keyword": keyword,
"time_range": time_range,
"author_type": author_type,
"is_original": is_original,
"sort_type": sort_type,
"page_no": page_no,
"page_size": page_size,
"match_type": match_type,
"need_baidu_rank": need_baidu_rank,
"timestamp": int(time.time() * 1000)
}
# 3. 补充可选参数
if time_range == "custom":
params["start_date"] = start_date
params["end_date"] = end_date
if duration_min is not None:
params["duration_min"] = duration_min
if duration_max is not None:
params["duration_max"] = duration_max
# 4. 生成签名
params["sign"] = generate_sign(params)
try:
# 5. 发送POST请求（HTTPS协议，超时15秒）
response = requests.post(
url=API_URL,
data=json.dumps(params),
headers={"Content-Type": "application/json"},
timeout=15,
verify=True
)
response.raise_for_status() # 抛出HTTP请求异常（如404、500）
result = response.json()
# 6. 处理响应（不同服务商返回格式可能不同，需按实际文档调整）
# 假设服务商返回格式：{"code":200,"msg":"success","data":{"video_list":[],"total":0,"page_total":1}}
if result.get("code") == 200:
raw_data = result.get("data", {})
video_list = raw_data.get("video_list", [])
total = raw_data.get("total", 0) # 匹配视频总条数
page_total = raw_data.get("page_total", 1) # 总页码
# 标准化视频数据
standard_videos = [standardize_data(video) for video in video_list]
for video in standard_videos:
video["关键词"] = keyword # 补充关键词字段，便于多关键词筛选时区分
return {
"success": True,
"data": standard_videos,
"total": total,
"page_no": page_no,
"page_total": page_total,
"error_msg": ""
}
else:
error_msg = result.get("msg", "接口调用失败")
error_code = result.get("code", -2)
logging.error(f"接口返回错误：code={error_code}, msg={error_msg}（关键词：{keyword}，页码：{page_no}）")
return {
"success": False,
"data": [],
"total": 0,
"page_no": page_no,
"page_total": 0,
"error_code": error_code,
"error_msg": error_msg
}
except requests.exceptions.RequestException as e:
logging.error(f"网络异常（关键词：{keyword}，页码：{page_no}）：{str(e)}")
return {
"success": False,
"data": [],
"total": 0,
"page_no": page_no,
"page_total": 0,
"error_code": -3,
"error_msg": f"网络异常：{str(e)}"
}
except Exception as e:
logging.error(f"数据处理异常（关键词：{keyword}，页码：{page_no}）：{str(e)}")
return {
"success": False,
"data": [],
"total": 0,
"page_no": page_no,
"page_total": 0,
"error_code": -4,
"error_msg": f"处理异常：{str(e)}"
}
async def get_video_list_async(session, keyword: str, **kwargs) -> Dict:
"""异步调用item_search_video接口，提升批量请求效率"""
page_no = kwargs.get("page_no", 1)
# 构建参数（逻辑同同步方法）
params = {
"appkey": APP_KEY,
"keyword": keyword,
"time_range": kwargs.get("time_range", "all"),
"author_type": kwargs.get("author_type", "all"),
"is_original": kwargs.get("is_original", 0),
"sort_type": kwargs.get("sort_type", "relevance"),
"page_no": page_no,
"page_size": kwargs.get("page_size", 20),
"match_type": kwargs.get("match_type", "fuzzy"),
"need_baidu_rank": kwargs.get("need_baidu_rank", 0),
"timestamp": int(time.time() * 1000)
}
if params["time_range"] == "custom":
params["start_date"] = kwargs.get("start_date")
params["end_date"] = kwargs.get("end_date")
if kwargs.get("duration_min") is not None:
params["duration_min"] = kwargs["duration_min"]
if kwargs.get("duration_max") is not None:
params["duration_max"] = kwargs["duration_max"]
# 生成签名
params["sign"] = generate_sign(params)
try:
async with session.post(
API_URL,
json=params,
headers={"Content-Type": "application/json"},
timeout=15
) as response:
response.raise_for_status()
result = await response.json()
if result.get("code") == 200:
raw_data = result.get("data", {})
video_list = raw_data.get("video_list", [])
standard_videos = [standardize_data(video) for video in video_list]
for video in standard_videos:
video["关键词"] = keyword
return {
"success": True,
"data": standard_videos,
"total": raw_data.get("total", 0),
"page_no": page_no,
"page_total": raw_data.get("page_total", 1),
"error_msg": ""
}
else:
error_msg = result.get("msg", "接口调用失败")
logging.error(f"异步请求失败（关键词：{keyword}，页码：{page_no}）：{error_msg}")
return {"success": False, "data": [], "error_msg": error_msg}
except Exception as e:
logging.error(f"异步请求异常（关键词：{keyword}，页码：{page_no}）：{str(e)}")
return {"success": False, "data": [], "error_msg": str(e)}
def batch_get_video_list_sync(
keyword: str,
max_page: int = 5, # 最大获取页码（避免无限制分页）
kwargs
) -> List[Dict]:
"""同步批量获取多页视频列表（遍历所有页码或指定max_page）"""
all_videos = []
page_no = 1
while True:
logging.info(f"正在获取关键词「{keyword}」第{page_no}页视频列表")
result = get_video_list_sync(keyword=keyword, page_no=page_no, kwargs)
if not result["success"]:
logging.error(f"第{page_no}页获取失败：{result['error_msg']}")
break
page_videos = result["data"]
if not page_videos:
logging.info(f"第{page_no}页无匹配视频，批量获取结束")
break
all_videos.extend(page_videos)
logging.info(f"第{page_no}页获取成功，新增{len(page_videos)}条数据（累计{len(all_videos)}条）")
# 终止条件：达到最大页码或总页码
if page_no >= result["page_total"] or page_no >= max_page:
break
page_no += 1
# 控制调用频率（普通用户10次/分钟，间隔6秒；企业用户50次/分钟，间隔1秒）
time.sleep(6)
return all_videos
async def batch_get_video_list_async(
keyword: str,
max_page: int = 5,
**kwargs
) -> List[Dict]:
"""异步批量获取多页视频列表（并行请求，提升效率）"""
all_videos = []
# 先获取总页码
first_page_result = await get_video_list_async(
session=None, keyword=keyword, page_no=1, **kwargs
)
if not first_page_result["success"]:
logging.error(f"关键词「{keyword}」第1页获取失败：{first_page_result['error_msg']}")
return all_videos
all_videos.extend(first_page_result["data"])
page_total = min(first_page_result["page_total"], max_page)
if page_total <= 1:
logging.info(f"关键词「{keyword}」仅1页数据，批量获取结束")
return all_videos
# 并行请求剩余页码
async with aiohttp.ClientSession() as session:
tasks = []
for page_no in range(2, page_total + 1):
tasks.append(get_video_list_async(session, keyword=keyword, page_no=page_no, **kwargs))
# 控制并发数（避免频率超限）
semaphore = asyncio.Semaphore(5) # 最大并发5个请求
async def bounded_task(task):
async with semaphore:
return await task
results = await asyncio.gather(*[bounded_task(task) for task in tasks])
for result in results:
if result["success"]:
all_videos.extend(result["data"])
logging.info(f"异步获取成功，新增{len(result['data'])}条数据")
logging.info(f"关键词「{keyword}」异步批量获取结束，累计{len(all_videos)}条数据")
return all_videos
def save_video_list(videos: List[Dict], save_path: str = SAVE_PATH):
"""将视频列表保存为Excel文件（便于归档/分析）"""
if not videos:
logging.warning("无视频数据可保存")
return
df = pd.DataFrame(videos)
# 筛选常用字段，优化Excel可读性
columns = [
"请求时间", "关键词", "视频ID", "视频标题", "视频时长", "视频类型",
"发布时间", "是否原创", "播放量", "点赞量", "评论数", "转发量",
"作者名称", "作者认证类型", "作者粉丝数", "视频标签", "视频状态",
"播放链接", "视频封面", "百度搜索曝光量", "相关度评分"
]
df = df[columns].drop_duplicates(subset=["视频ID"]) # 按视频ID去重（避免重复保存）
# 增量保存（避免覆盖历史数据）
try:
history_df = pd.read_excel(save_path, engine="openpyxl")
df = pd.concat([history_df, df], ignore_index=True).drop_duplicates(subset=["视频ID"])
except FileNotFoundError:
pass
df.to_excel(save_path, index=False, engine="openpyxl")
logging.info(f"视频列表已归档至：{save_path}（累计{len(df)}条数据）")
def keyword_monitor_task(keywords: List[str], **kwargs):
"""关键词定时监测任务（实时追踪视频动态）"""
logging.info("=== 开始执行关键词定时监测任务 ===")
all_videos = []
for keyword in keywords:
logging.info(f"=== 开始监测关键词：{keyword} ===")
# 同步批量获取（如需高效可改用异步）
videos = batch_get_video_list_sync(keyword=keyword, **kwargs)
all_videos.extend(videos)
logging.info(f"=== 关键词「{keyword}」监测完成，获取{len(videos)}条数据 ===")
# 保存所有关键词的视频列表
save_video_list(all_videos)
logging.info("=== 关键词定时监测任务执行完成 ===")
调用示例（支持单页/批量/异步/定时监测）
if name == "main":
# 模式1：获取单页视频列表（关键词“Python教程 零基础”，30天内，1-5分钟，按播放量排序）
single_page_result = get_video_list_sync(
keyword="Python教程 零基础",
time_range="30days",
duration_min=60,
duration_max=300,
sort_type="play_count",
page_no=1,
page_size=20,
need_baidu_rank=1
)
if single_page_result["success"]:
print("="*80)
print(f"关键词「Python教程 零基础」第1页视频列表（共{len(single_page_result['data'])}条）")
print("="*80)
for idx, video in enumerate(single_page_result["data"][:10], 1): # 打印前10条
print(f"{idx:2d}. 标题：{video['视频标题']}")
print(f" 时长：{video['视频时长']} | 类型：{video['视频类型']} | 原创：{video['是否原创']}")
print(f" 播放量：{video['播放量']} | 点赞量：{video['点赞量']} | 评论数：{video['评论数']}")
print(f" 作者：{video['作者名称']}（{video['作者认证类型']}） | 发布时间：{video['发布时间']}")
print(f" 百度曝光：{video['百度搜索曝光量']} | 相关度：{video['相关度评分']}分 | 状态：{video['视频状态']}")
print(f" 播放链接：{video['播放链接']}")
print("-"*80)
else:
print(f"单页视频列表获取失败：{single_page_result['error_msg']}（错误码：{single_page_result['error_code']}）")
# 模式2：同步批量获取多页视频列表（关键词“旅行vlog”，按发布时间倒序，最多获取5页）
# batch_videos = batch_get_video_list_sync(
# keyword="旅行vlog",
# sort_type="pub_time",
# max_page=5,
# page_size=30,
# is_original=1
# )
# save_video_list(batch_videos)
# 模式3：异步批量获取多页视频列表（关键词“家常菜做法”，自定义时间2024-01-01至2024-12-31）
# asyncio.run(batch_get_video_list_async(
# keyword="家常菜做法",
# time_range="custom",
# start_date="2024-01-01",
# end_date="2024-12-31",
# duration_min=120,
# max_page=10,
# page_size=50
# ))
# 模式4：启动定时监测任务（每1小时监测一次关键词“新能源汽车 评测”“手机 新品”）
# scheduler = BlockingScheduler()
# scheduler.add_job(
# keyword_monitor_task,
# 'interval',
# hours=1,
# args=[["新能源汽车 评测", "手机 新品"]],
# kwargs={"time_range": "1day", "sort_type": "pub_time", "max_page": 3}
# )
# logging.info("关键词定时监测任务已启动，每1小时执行一次...")
# try:
# scheduler.start()
# except (KeyboardInterrupt, SystemExit):
# logging.info("关键词定时监测任务已停止")
```
四、调试与验证：快速定位问题
- 调试步骤（优先用 Postman 验证，排除代码干扰）
手动拼接参数：在 Postman 中创建 POST 请求，填写appkey、keyword、timestamp等必填项，按业务需求补充duration_min、sort_type等可选参数；
生成签名：按服务商规则手动计算sign（用在线 MD5 工具验证，输入拼接后的字符串）；
配置请求头：设置Content-Type: application/json，将参数以 JSON 格式传入请求体；
发送请求：点击发送，查看响应结果；
验证结果：
若返回 200 且视频列表完整：接口正常，可对接代码；
若返回 401（签名无效）：检查参数排序、secret 是否正确、时间戳是否过期、中文关键词是否编码；
若返回 403（权限不足）：确认账号类型（普通 / 企业）是否匹配筛选条件 / 字段需求（如need_baidu_rank=1需企业权限），是否已付费或申请高级权限；
若返回 400（参数错误）：核对time_range/sort_type/author_type等参数取值是否合法、自定义时间格式是否正确、视频时长是否为整数；
若返回 429（频率超限）：降低调用频率，或申请提升配额；
若返回 404（接口不存在）：核对 API_URL 是否正确，确认item_search_video接口已申请开通；
若返回 500（服务器异常）：记录日志，稍后重试（可能是服务商接口问题）；
若返回 503（服务不可用）：联系服务商确认接口是否维护；
若返回 “无匹配视频”：调整关键词（如扩大范围）、时间范围或筛选条件，确认好看视频官网是否有相关视频。 - 常见调试问题排查（关键词搜索场景高频问题）
问题现象 常见原因 排查方案
签名错误（401） 1. 参数排序错误；2. secret 错误；3. 时间戳过期；4. 中文关键词未编码；5. 服务商签名规则差异 1. 按 ASCII 升序排序参数并打印验证；2. 核对 secret 与服务商后台一致；3. 校准本地时间（误差≤5 分钟）；4. 用urlencode处理中文关键词；5. 重新查看服务商文档，确认是否需要 SHA256 加密或额外参数（如nonce随机数）
无匹配视频 1. 关键词过于精准 / 宽泛；2. 时间范围过窄；3. 筛选条件过于严格（如时长区间过小）；4. 视频未被服务商收录；5. 关键词组合逻辑错误 1. 调整关键词（如 “Python 零基础” 改为 “Python 教程 零基础”）；2. 扩大时间范围（如从 1day 改为 7days）；3. 放宽筛选条件（如duration_max从 300 改为 600）；4. 直接在好看视频官网搜索关键词，确认是否有相关视频；5. 确认多关键词组合是否正确（AND 用空格，OR 用
权限不足（403） 1. 普通用户使用高级筛选（如need_baidu_rank=1、is_original=1）；2. 调用次数耗尽；3. 业务用途未通过审核 1. 移除高级筛选条件，或升级为企业用户；2. 查看服务商后台调用次数剩余量，及时充值；3. 补充业务合规说明（如 “仅用于内部内容分析”），重新提交审核
频率超限（429） 单 IP / 账号调用次数超过服务商限制 1. 批量分页时增加间隔（普通用户 6 秒 / 次，企业用户 1 秒 / 次）；2. 异步请求控制并发数（≤5）；3. 分时段调用，避免集中请求；4. 用ratelimit库控制调用频率（Python 示例：@limits(calls=10, period=60)）；5. 联系服务商申请提升配额
参数错误（400） 1. time_range=custom未传start_date/end_date；2. 日期格式错误（非 “YYYY-MM-DD”）；3. 视频时长为非整数；4. is_original取值非法（仅支持 0/1/2） 1. 确保自定义时间范围参数完整；2. 统一日期格式为 “YYYY-MM-DD”；3. 确保duration_min/duration_max为整数；4. 核对is_original取值，仅支持 0（全部）、1（仅原创）、2（仅非原创）
视频列表重复 1. 分页时缓存不一致；2. 关键词多义性导致重复收录；3. 未去重处理 1. 批量获取时一次性请求所有页码，减少缓存影响；2. 优化关键词（如增加限定词 “2024 最新”）；3. 按视频ID去重（代码中已实现）
视频状态为 “已下架 / 违规” 1. 视频已被作者删除；2. 视频涉及违规内容被平台处罚；3. 服务商数据未及时同步 1. 直接访问play_url确认视频是否存在；2. 过滤该类视频，避免影响数据质量；3. 企业版可申请 “实时状态同步” 权限，提升数据时效性
百度排名数据未返回（need_baidu_rank=1） 1. 普通用户无权限；2. 服务商未支持该字段；3. 视频未被百度收录 1. 升级为企业用户，申请百度生态数据权限；2. 联系服务商确认是否支持need_baidu_rank参数；3. 确认视频在百度搜索中是否可检索到
响应速度慢 1. 关键词匹配结果过多；2. 分页page_size过大；3. 启用了need_baidu_rank=1（百度数据查询耗时）；4. 网络波动 1. 优化关键词，缩小筛选范围；2. 减小page_size（如从 50 改为 20）；3. 非必需场景关闭need_baidu_rank；4. 增加超时时间（如从 10 秒改为 15 秒）
五、进阶优化：提升效率与稳定性（生产级必备） - 性能优化（批量 / 实时监测场景重点）
（1）批量请求优化
异步并发请求：多页码 / 多关键词批量获取时，用异步请求并行拉取（控制并发数≤5，避免频率超限），如代码中batch_get_video_list_async方法，效率比同步提升 3-5 倍；
分页智能停止：获取第 1 页后，根据page_total计算总页码，仅请求有效页码，避免无效分页（如page_no超过page_total）；
字段筛选精准化：仅保留业务必需字段（如fields=video_id,title,play_count,author_name,pub_time），减少数据传输量和解析耗时（需服务商支持字段筛选参数）。
（2）缓存策略优化
搜索结果缓存：用 Redis 缓存关键词 + 筛选条件组合对应的搜索结果（缓存 key = 关键词 + 时间范围 + 筛选条件，有效期 30 分钟 - 1 小时），避免重复请求相同关键词；
缓存穿透防护：对无匹配结果的关键词，缓存空结果（有效期 10 分钟），避免频繁无效请求；
热点关键词缓存：对高频监测的关键词（如舆情关键词），缩短缓存有效期（如 10 分钟），平衡实时性与效率；
视频状态缓存：对已获取的视频，缓存其is_available状态（有效期 1 小时），避免重复请求已下架视频。
（3）关键词策略优化
关键词分词与扩展：利用 jieba 分词对核心关键词进行扩展（如 “Python 教程” 扩展为 “Python 零基础教程、Python 进阶教程”），提升匹配覆盖率；
多关键词优先级：按业务重要性划分关键词优先级（如核心舆情关键词、次要素材关键词），核心关键词高频监测，次要关键词低频监测；
关键词去重：多关键词监测时，避免关键词重复（如 “旅行 vlog” 和 “vlog 旅行”），减少冗余请求。 - 稳定性优化（生产级必备）
异常重试机制：
对 429（频率超限）、500（服务器异常）、503（服务不可用）错误，采用指数退避策略重试（5s→10s→20s）；
重试次数≤3 次，避免无效重试导致更多错误；
对 401（签名错误）、403（权限不足）、400（参数错误）错误，直接返回并日志告警（需人工介入）。
密钥与权限安全：
定期轮换secret（每 3 个月更新 1 次），在服务商后台操作；
生产环境将appkey和secret存储在环境变量或配置中心（如 Nacos），避免硬编码；
限制接口调用 IP（部分服务商支持 IP 白名单），防止密钥泄露后被滥用；
企业用户可申请 “子账号” 权限，按业务模块分配不同appkey，便于权限管控。
日志与监控：
详细记录每次请求的参数、签名、响应结果、错误信息、调用耗时，便于问题追溯；
配置日志告警（如通过邮件 / 钉钉推送高频错误、频率超限、权限不足提示），实时监控接口状态；
监控数据质量（如连续 10 次请求无结果、高频关键词响应时间超过 5 秒），及时触发告警；
统计关键词匹配数量、视频状态分布，评估接口使用效果。 - 业务场景专属适配
（1）内容聚合场景
优质视频筛选：按 “播放量≥10000 + 点赞量≥500 + 相关度≥80 分” 筛选，结合tags分类展示（如 “科技”“美食” 栏目）；
时效性内容优先：对新闻、热点类关键词，按 “pub_time” 排序，仅获取近 7 天视频，确保内容新鲜度；
作者分级：按author_fans和author_type分级（如头部作者、腰部作者），优先展示头部作者视频，提升内容质量；
字幕适配：筛选has_subtitle=true的视频，便于用户静音观看，提升用户体验。
（2）舆情监测场景
敏感视频识别：基于title、description和关键词匹配（如 “投诉”“故障”“负面”），筛选敏感舆情视频，触发告警；
传播趋势分析
