简介面向希望掌握社交媒体数据采集的Python爬虫学习者这份资源以快手APP短视频为对象演示从API请求、JSON解析到视频文件下载、表格化存储的完整链路可作为同类短视频App数据提取的参考。压缩包共2个文件包含1个Python脚本和1个Markdown说明文档整体仅2KB脚本覆盖网络请求、JSON数据解析、模拟登录、反爬策略应对、视频分块下载、pandas写入CSV等关键环节说明文档则梳理操作流程与实现要点。已有1606人学习下载适合具备基础Python语法、希望进阶爬虫实战的读者。通过学习可掌握requests会话与模拟登录、代理与User-Agent设置、嵌套JSON字段提取、多线程批量处理等方法也能理解请求头伪装与数据持久化的具体做法最终形成一套可复用的快手短视频数据采集与保存方案。1. 从数据接口到文件落盘快手短视频采集的完整链路多数人拿到快手短视频采集需求时第一反应是写个循环直接下载视频。真正跑一遍就会发现下载视频反而是整条链路里最简单的一步接口签名、JSON嵌套解析、登录态保持这三件事会消耗掉八成调试时间。kuaishou.py 把这条链路拆成了请求构造、元数据解析、二进制落盘、频控退避四段README 里标注了每个模块的输入输出和依赖版本。它能直接解决的问题是给定一批视频ID或用户主页地址自动抓取标题、作者、播放量等元数据并生成CSV同时把视频文件按ID保存在本地目录。适合做短视频内容归档、竞品监控和运营数据分析的工程师阅读尤其是那些已经用 requests 写过简单爬虫、但没处理过大规模下载和数据清洗的人。下面按代码实际执行顺序逐段拆解。2. 接口请求的构造与登录态保持requests 会话的关键参数2.1 先定位数据接口从页面到 XHR 接口快手App的数据来自移动端API网关路径形如/rest/o/photo/list、/rest/w/photo/info。实际操作时最直接的办法是用抓包工具观察App下拉刷新时发出的 XHR 请求或者直接分析 Web 端详情页的网络面板。只要找到一个返回feeds数组的接口后续的批量采集就能围绕它扩展。这里有一个关键点App端接口大多带签名参数例如__NS_sig3、sig、timestamp。这些参数是客户端根据当前时间、请求路径和本地密钥计算的直接用裸requests.get()很难通过校验。常见做法是抓一次真实请求把签名参数和 Cookie 缓存下来在有效期内复用等签名过期后再抓一次。比起逆向客户端加密逻辑这个方案在跨端采集场景下性价比高得多也更容易在接口升级时快速恢复。2.2 用 requests.Session 维持登录态如果目标接口要求登录每次都新建连接显然不合适。requests.Session会自动管理 Cookie 和连接池只要登录一次后续请求都携带会话凭证不再需要手动拼 Cookie 到每个请求头里。import requests session requests.Session() session.headers.update({ User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.6 Mobile/15E148 Safari/604.1, Referer: https://www.kuaishou.com/, Accept: application/json, text/plain, */*, }) def login_with_cookie(cookie_str: str): 把抓包得到的 Cookie 写入会话 for item in cookie_str.split(;): if in item: key, value item.strip().split(, 1) session.cookies.set(key, value)逻辑说明session.headers.update()把移动端 UA 和 Referer 设为会话级全局头后续每个请求自动携带代码里不用重复传 headerslogin_with_cookie把抓包得到的 Cookie 字符串按分号拆成键值对逐项写入会话之后调用session.get(url)时服务端就能识别出登录身份。参数说明cookie_str里常见的键有did、userId、kuaishou.server.web_st等split(, 1)只拆第一个等号防止某个 Cookie 值本身含有等号导致解析错位。strip()去掉键和值两侧的空格避免出现 userId这类带前缀空格的脏键。2.3 请求头里容易被忽略的校验字段除了 UA 和 Cookie服务端还会校验Accept-Language、X-Requested-With等字段。遇到接口返回403或业务码result不为1时先别急着加频率限制而是逐项核对请求头是否和真实抓包一致。可以用 curl 直接对比差异curl -i -X GET https://api.kuaishou.com/rest/o/photo/list?userIdxxxpage1 \ -H User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) \ -H Accept-Language: zh-CN,zh;q0.9 \ -H Referer: https://www.kuaishou.com/-i参数把响应头也打印出来能直接看到Set-Cookie和状态码如果响应头里出现X-Login-Required这类标记说明 Cookie 失效需要重新抓包登录。真机返回状态码的含义大致如下表状态码含义处理方式200请求成功正常解析 JSON403签名或频率校验失败检查签名参数与请求头406Cookie 缺失或过期重新抓包获取 Cookie503服务端限流指数退避后重试我会把 Session 初始化和登录逻辑单独拆成client.py模块采集脚本里只负责拼 URL 和解析返回值。这样抓包换 Cookie 时只改一个文件不会动到业务代码。3. JSON 字段解析与元数据提取从嵌套结构到 CSV 保存3.1 快手接口的返回结构列表接口的返回体是典型的多层嵌套外层是feeds数组每个元素里有photo对象photo下面又挂着user、videoResource、coverUrl等子对象。第一次上手容易一路用data[feeds][0][photo][caption]硬点到底一旦某条视频缺字段就抛KeyError整批数据中断。更稳妥的做法是先打印第一条数据确认字段路径和类型后再写解析逻辑import json def pretty_print_first_item(raw: str): data json.loads(raw) first_photo data[feeds][0][photo] print(json.dumps(first_photo, ensure_asciiFalse, indent2))逻辑说明json.loads把响应文本转成字典json.dumps(first_photo, ensure_asciiFalse, indent2)重新序列化并缩进输出中文标题不会被转成\uXXXX转义序列。这一步输出的字段路径就是后续元数据映射的依据。实测中photo.id、photo.caption、photo.likeCount、photo.viewCount这些字段在不同接口版本下基本固定而videoResource.url会随请求时间变化不适合作为数据行主键。主键只能选photo.id它在整个响应体里唯一且不回变。3.2 递归提取字段应对字段缺失和层级变化接口升级时字段层级常会调整比如旧版photo.user.name到新版可能变成photo.author.name。靠硬编码下标取值的代码接口一变更就要全局修改。我习惯写一个按路径递归取值的函数把字段路径作为配置项传进去接口变动时只改路径字符串。def deep_get(data: dict, path: str, defaultNone): 按点号路径递归取值 示例deep_get(photo, user.name, -) keys path.split(.) node data for key in keys: if isinstance(node, dict) and key in node: node node[key] else: return default return node def extract_video_meta(feeds: list) - list: rows [] for feed in feeds: photo feed.get(photo, {}) user photo.get(user, {}) rows.append({ video_id: deep_get(photo, id, ), title: deep_get(photo, caption, ), author: deep_get(user, name, ), like_count: deep_get(photo, likeCount, 0), play_count: deep_get(photo, viewCount, 0), }) return rows逻辑说明deep_get沿点号路径逐层下探每一层都检查当前节点是否为字典遇到列表或缺失键直接返回默认值。这样某条视频缺作者字段时只影响这一行不会中断整个采集流程。extract_video_meta把每条 feed 压平成一行元数据字段名统一用蛇形命名方便后面写表和分析。参数说明default参数建议根据字段类型设置ID 类字段用空字符串计数类字段用0避免后面参与计算时报TypeError。user字段在部分视频里可能是null所以先用photo.get(user, {})兜底再传给deep_get否则user.name路径会直接返回默认值。3.3 用 pandas 写出 CSV 并做字段类型纠偏元数据量达到几千行时手动用csv.writer要处理转义和类型转换效率太低。pandas 的DataFrame可以一次性完成类型转换、去重和落盘代码量少且不容易出错。import pandas as pd def save_to_csv(rows: list, output_path: str): df pd.DataFrame(rows) # 播放量在接口里可能是字符串或缺失先转成数值 for col in [like_count, play_count]: df[col] pd.to_numeric(df[col], errorscoerce).fillna(0).astype(int) df.drop_duplicates(subset[video_id], keepfirst, inplaceTrue) df.to_csv(output_path, indexFalse, encodingutf-8-sig)参数说明errorscoerce把无法解析的字符串变成NaNfillna(0)填充空值astype(int)统一为整数subset[video_id]以视频 ID 为去重键keepfirst保留先出现的记录encodingutf-8-sig写入带 BOM 的 UTF-8让 Excel 直接打开 CSV 时不出现中文乱码。写完 CSV 后通常会过滤出有效的video_id列表作为下载队列queue [row[video_id] for row in rows if row[video_id]]这里只保留非空 ID避免用None拼接视频 URL 导致后续请求全部失败。实际项目中我会把 CSV 文件名带上采集日期比如meta_20250214.csv这样每天的元数据互不覆盖回溯问题时也能快速定位是哪一批数据。4. 视频二进制流的分块下载与并发加速4.1 streamTrue 与 iter_content 分块写入短视频文件通常几 MB 到几十 MB如果直接requests.get(url).content整个文件会一次性加载进内存。批量下载时内存占用随视频数量线性增长脚本很容易被系统杀死。正确做法是开启流式响应边读边写把内存占用控制在一个固定值附近。import requests def download_video(video_url: str, filepath: str): resp requests.get(video_url, streamTrue, timeout30) resp.raise_for_status() with open(filepath, wb) as f: for chunk in resp.iter_content(chunk_size1024 * 256): if chunk: f.write(chunk)逻辑说明streamTrue让 requests 只下载响应头就返回文件内容通过iter_content(1024 * 256)按 256KB 切块迭代if chunk过滤掉 TCP 保活产生的空字节块避免无意义的磁盘写入。timeout30限制单次请求的等待时间防止某个卡死的视频 URL 拖住整个任务。这里的关键区别是不设置streamTrue时resp.content会聚合完整响应体视频大了以后内存和磁盘 IO 都会成为瓶颈。设置chunk_size也不是越小越好如果改成 1KB每个视频会产生几万次磁盘写入调用反而更慢。注意不要让chunk_size超过服务端单次返回的缓冲上限否则 requests 内部仍会按自己的缓冲大小切块设置值只作为上限参考。256KB 是我在不同平台视频下载场景下试过比较稳的取值。4.2 ThreadPoolExecutor 控制并发下载单线程逐个下载时网络往返的等待时间完全串行几百个视频可能要跑几十分钟。用ThreadPoolExecutor可以把下载任务分发到多个线程但并发数需要权衡太快会触发服务端限流。from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path def batch_download(items: list[dict], out_dir: str, max_workers: int 4): Path(out_dir).mkdir(parentsTrue, exist_okTrue) with ThreadPoolExecutor(max_workersmax_workers) as executor: futures {} for item in items: video_id item[video_id] url item[video_url] ext url.split(?)[0].rsplit(., 1)[-1] if . in url.split(?)[0] else mp4 filepath str(Path(out_dir) / f{video_id}.{ext}) futures[executor.submit(download_video, url, filepath)] video_id for future in as_completed(futures): video_id futures[future] try: future.result() print(fok: {video_id}) except Exception as exc: print(ffail: {video_id} - {exc})逻辑说明executor.submit把download_video(url, filepath)丢进线程池返回的future以video_id为键存进字典as_completed在任务完成时逐个取结果失败时打印异常不影响其他视频继续下载。参数说明max_workers4是面对接口频控的保守值本地网络好且接口无风控时可以提升到 8 或 12ext先切掉 URL 的查询参数再取最后一段后缀否则?后面的参数串会被误判成扩展名。文件名直接使用video_id既避免操作系统文件名非法字符问题也方便与 CSV 里的数据行对应。4.3 下载失败后的重试与断点续传快手视频地址有时是带时效的签名 URL过期后返回 403。面对这种情况工程上常用两个手段一是统一重试二是利用 HTTP Range 实现断点续传。断点续传的意义在于当下载到一半连接断开时不需要从头拉整个文件而是从已写入的字节数继续。import os def download_with_resume(video_url: str, filepath: str, retries: int 3): for attempt in range(retries): existing os.path.getsize(filepath) if os.path.exists(filepath) else 0 headers {Range: fbytes{existing}-} if existing else {} resp requests.get(video_url, streamTrue, headersheaders, timeout30) if resp.status_code 416: print(already complete) return resp.raise_for_status() mode ab if existing else wb with open(filepath, mode) as f: for chunk in resp.iter_content(chunk_size1024 * 256): if chunk: f.write(chunk) if os.path.getsize(filepath) int(resp.headers.get(Content-Length, 0)): return print(give up:, filepath)逻辑说明existing取本地文件已有字节数Range: bytes{existing}-要求服务端从该偏移继续传若服务端返回206追加写入若返回200说明服务端忽略 Range此时应该从头覆盖所以代码用mode区分ab和wb。416表示请求范围超出文件总长度等价于文件已经下完直接退出。参数说明retries3控制最大尝试次数每次重试前应配合退避等待避免连续打到同一个限流接口。下载完成后比较本地文件大小与服务端Content-Length不相等就说明传输异常需要重试。这里有个细节部分服务端不返回Content-Lengthint(resp.headers.get(Content-Length, 0))会取到 0此时条件判断失效稳妥做法是再校验文件尾部是否为完整 FLV/MP4 box但日常采集场景下以上逻辑已经能覆盖九成异常。5. 频控识别的退避策略与合规保存边界5.1 通过日志特征识别限流脚本跑久了最容易出现的不是 IP 被封而是服务端开始对高频接口返回异常。常见的三个信号同一接口连续返回 403、JSON 里的result字段从 1 变为 100、单次响应耗时从 200 毫秒涨到 3 秒以上。把这些信号记录到日志比单纯看状态码更早暴露问题。import time import logging def safe_get(url: str, session: requests.Session, max_retry: int 5): for attempt in range(max_retry): resp session.get(url, timeout15) if resp.status_code 200 and resp.json().get(result) 1: return resp.json() wait min(2 ** attempt * 3, 60) logging.warning(retry %s after %ss, status%s, url, wait, resp.status_code) time.sleep(wait) raise RuntimeError(ffailed after retries: {url})逻辑说明每轮先请求只有状态码 200 且业务码为 1 才认为成功否则按指数退避等待后重试。min(2 ** attempt * 3, 60)计算出的等待序列是 3、6、12、24、48 秒触顶 60 秒后不再增长。日志里同时记录 URL、等待秒数和状态码排查时能直接看出是否集中在某个接口上。5.2 在请求间隔中加入随机抖动纯固定间隔的定时请求在服务端看来节奏过于规整容易被识别。常见做法是在基础间隔上叠加均匀分布的随机值让请求时间点更接近真实用户的滚动节奏。import random def jitter_sleep(base: float 1.5, spread: float 0.8): time.sleep(random.uniform(base - spread, base spread))参数说明base是平均等待秒数spread控制抖动幅度实际等待落在 0.7 到 2.3 秒之间。需要说明的是抖动不是用来对抗反爬而是降低请求在时间轴上碰撞的概率。和上一小节的指数退避搭配使用长时间采集时接口返回成功率会明显提升。5.3 数据保存的合规边界与目录归档采集前查看目标站点的 robots.txt 和平台协议是基本功。快手公开主页的视频元数据、开放平台接口数据与需要登录后可见的完播明细两者的合规性完全不同。kuaishou.py 默认只保存公开视频的标题、作者、播放量和视频文件不涉及非公开数据字段这个范围相对安全。保存时我在每行元数据里追加采集时间戳便于后续审计和数据回溯。rows.append({ video_id: video_id, crawled_at: time.strftime(%Y-%m-%d %H:%M:%S), })另外视频文件建议按日期分目录归档而不是全部平铺在一个目录下。文件数上万之后单目录的索引耗时和删除耗时都会明显增加。归档方式很简单在batch_download里把out_dir拼上当天日期例如videos/20250214/任务的写入范围就限定在了当天目录排查问题也能快速按时间定位。本文还有配套的精品资源点击获取
快手短视频采集全链路解析：从接口签名到文件落盘
简介面向希望掌握社交媒体数据采集的Python爬虫学习者这份资源以快手APP短视频为对象演示从API请求、JSON解析到视频文件下载、表格化存储的完整链路可作为同类短视频App数据提取的参考。压缩包共2个文件包含1个Python脚本和1个Markdown说明文档整体仅2KB脚本覆盖网络请求、JSON数据解析、模拟登录、反爬策略应对、视频分块下载、pandas写入CSV等关键环节说明文档则梳理操作流程与实现要点。已有1606人学习下载适合具备基础Python语法、希望进阶爬虫实战的读者。通过学习可掌握requests会话与模拟登录、代理与User-Agent设置、嵌套JSON字段提取、多线程批量处理等方法也能理解请求头伪装与数据持久化的具体做法最终形成一套可复用的快手短视频数据采集与保存方案。1. 从数据接口到文件落盘快手短视频采集的完整链路多数人拿到快手短视频采集需求时第一反应是写个循环直接下载视频。真正跑一遍就会发现下载视频反而是整条链路里最简单的一步接口签名、JSON嵌套解析、登录态保持这三件事会消耗掉八成调试时间。kuaishou.py 把这条链路拆成了请求构造、元数据解析、二进制落盘、频控退避四段README 里标注了每个模块的输入输出和依赖版本。它能直接解决的问题是给定一批视频ID或用户主页地址自动抓取标题、作者、播放量等元数据并生成CSV同时把视频文件按ID保存在本地目录。适合做短视频内容归档、竞品监控和运营数据分析的工程师阅读尤其是那些已经用 requests 写过简单爬虫、但没处理过大规模下载和数据清洗的人。下面按代码实际执行顺序逐段拆解。2. 接口请求的构造与登录态保持requests 会话的关键参数2.1 先定位数据接口从页面到 XHR 接口快手App的数据来自移动端API网关路径形如/rest/o/photo/list、/rest/w/photo/info。实际操作时最直接的办法是用抓包工具观察App下拉刷新时发出的 XHR 请求或者直接分析 Web 端详情页的网络面板。只要找到一个返回feeds数组的接口后续的批量采集就能围绕它扩展。这里有一个关键点App端接口大多带签名参数例如__NS_sig3、sig、timestamp。这些参数是客户端根据当前时间、请求路径和本地密钥计算的直接用裸requests.get()很难通过校验。常见做法是抓一次真实请求把签名参数和 Cookie 缓存下来在有效期内复用等签名过期后再抓一次。比起逆向客户端加密逻辑这个方案在跨端采集场景下性价比高得多也更容易在接口升级时快速恢复。2.2 用 requests.Session 维持登录态如果目标接口要求登录每次都新建连接显然不合适。requests.Session会自动管理 Cookie 和连接池只要登录一次后续请求都携带会话凭证不再需要手动拼 Cookie 到每个请求头里。import requests session requests.Session() session.headers.update({ User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.6 Mobile/15E148 Safari/604.1, Referer: https://www.kuaishou.com/, Accept: application/json, text/plain, */*, }) def login_with_cookie(cookie_str: str): 把抓包得到的 Cookie 写入会话 for item in cookie_str.split(;): if in item: key, value item.strip().split(, 1) session.cookies.set(key, value)逻辑说明session.headers.update()把移动端 UA 和 Referer 设为会话级全局头后续每个请求自动携带代码里不用重复传 headerslogin_with_cookie把抓包得到的 Cookie 字符串按分号拆成键值对逐项写入会话之后调用session.get(url)时服务端就能识别出登录身份。参数说明cookie_str里常见的键有did、userId、kuaishou.server.web_st等split(, 1)只拆第一个等号防止某个 Cookie 值本身含有等号导致解析错位。strip()去掉键和值两侧的空格避免出现 userId这类带前缀空格的脏键。2.3 请求头里容易被忽略的校验字段除了 UA 和 Cookie服务端还会校验Accept-Language、X-Requested-With等字段。遇到接口返回403或业务码result不为1时先别急着加频率限制而是逐项核对请求头是否和真实抓包一致。可以用 curl 直接对比差异curl -i -X GET https://api.kuaishou.com/rest/o/photo/list?userIdxxxpage1 \ -H User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) \ -H Accept-Language: zh-CN,zh;q0.9 \ -H Referer: https://www.kuaishou.com/-i参数把响应头也打印出来能直接看到Set-Cookie和状态码如果响应头里出现X-Login-Required这类标记说明 Cookie 失效需要重新抓包登录。真机返回状态码的含义大致如下表状态码含义处理方式200请求成功正常解析 JSON403签名或频率校验失败检查签名参数与请求头406Cookie 缺失或过期重新抓包获取 Cookie503服务端限流指数退避后重试我会把 Session 初始化和登录逻辑单独拆成client.py模块采集脚本里只负责拼 URL 和解析返回值。这样抓包换 Cookie 时只改一个文件不会动到业务代码。3. JSON 字段解析与元数据提取从嵌套结构到 CSV 保存3.1 快手接口的返回结构列表接口的返回体是典型的多层嵌套外层是feeds数组每个元素里有photo对象photo下面又挂着user、videoResource、coverUrl等子对象。第一次上手容易一路用data[feeds][0][photo][caption]硬点到底一旦某条视频缺字段就抛KeyError整批数据中断。更稳妥的做法是先打印第一条数据确认字段路径和类型后再写解析逻辑import json def pretty_print_first_item(raw: str): data json.loads(raw) first_photo data[feeds][0][photo] print(json.dumps(first_photo, ensure_asciiFalse, indent2))逻辑说明json.loads把响应文本转成字典json.dumps(first_photo, ensure_asciiFalse, indent2)重新序列化并缩进输出中文标题不会被转成\uXXXX转义序列。这一步输出的字段路径就是后续元数据映射的依据。实测中photo.id、photo.caption、photo.likeCount、photo.viewCount这些字段在不同接口版本下基本固定而videoResource.url会随请求时间变化不适合作为数据行主键。主键只能选photo.id它在整个响应体里唯一且不回变。3.2 递归提取字段应对字段缺失和层级变化接口升级时字段层级常会调整比如旧版photo.user.name到新版可能变成photo.author.name。靠硬编码下标取值的代码接口一变更就要全局修改。我习惯写一个按路径递归取值的函数把字段路径作为配置项传进去接口变动时只改路径字符串。def deep_get(data: dict, path: str, defaultNone): 按点号路径递归取值 示例deep_get(photo, user.name, -) keys path.split(.) node data for key in keys: if isinstance(node, dict) and key in node: node node[key] else: return default return node def extract_video_meta(feeds: list) - list: rows [] for feed in feeds: photo feed.get(photo, {}) user photo.get(user, {}) rows.append({ video_id: deep_get(photo, id, ), title: deep_get(photo, caption, ), author: deep_get(user, name, ), like_count: deep_get(photo, likeCount, 0), play_count: deep_get(photo, viewCount, 0), }) return rows逻辑说明deep_get沿点号路径逐层下探每一层都检查当前节点是否为字典遇到列表或缺失键直接返回默认值。这样某条视频缺作者字段时只影响这一行不会中断整个采集流程。extract_video_meta把每条 feed 压平成一行元数据字段名统一用蛇形命名方便后面写表和分析。参数说明default参数建议根据字段类型设置ID 类字段用空字符串计数类字段用0避免后面参与计算时报TypeError。user字段在部分视频里可能是null所以先用photo.get(user, {})兜底再传给deep_get否则user.name路径会直接返回默认值。3.3 用 pandas 写出 CSV 并做字段类型纠偏元数据量达到几千行时手动用csv.writer要处理转义和类型转换效率太低。pandas 的DataFrame可以一次性完成类型转换、去重和落盘代码量少且不容易出错。import pandas as pd def save_to_csv(rows: list, output_path: str): df pd.DataFrame(rows) # 播放量在接口里可能是字符串或缺失先转成数值 for col in [like_count, play_count]: df[col] pd.to_numeric(df[col], errorscoerce).fillna(0).astype(int) df.drop_duplicates(subset[video_id], keepfirst, inplaceTrue) df.to_csv(output_path, indexFalse, encodingutf-8-sig)参数说明errorscoerce把无法解析的字符串变成NaNfillna(0)填充空值astype(int)统一为整数subset[video_id]以视频 ID 为去重键keepfirst保留先出现的记录encodingutf-8-sig写入带 BOM 的 UTF-8让 Excel 直接打开 CSV 时不出现中文乱码。写完 CSV 后通常会过滤出有效的video_id列表作为下载队列queue [row[video_id] for row in rows if row[video_id]]这里只保留非空 ID避免用None拼接视频 URL 导致后续请求全部失败。实际项目中我会把 CSV 文件名带上采集日期比如meta_20250214.csv这样每天的元数据互不覆盖回溯问题时也能快速定位是哪一批数据。4. 视频二进制流的分块下载与并发加速4.1 streamTrue 与 iter_content 分块写入短视频文件通常几 MB 到几十 MB如果直接requests.get(url).content整个文件会一次性加载进内存。批量下载时内存占用随视频数量线性增长脚本很容易被系统杀死。正确做法是开启流式响应边读边写把内存占用控制在一个固定值附近。import requests def download_video(video_url: str, filepath: str): resp requests.get(video_url, streamTrue, timeout30) resp.raise_for_status() with open(filepath, wb) as f: for chunk in resp.iter_content(chunk_size1024 * 256): if chunk: f.write(chunk)逻辑说明streamTrue让 requests 只下载响应头就返回文件内容通过iter_content(1024 * 256)按 256KB 切块迭代if chunk过滤掉 TCP 保活产生的空字节块避免无意义的磁盘写入。timeout30限制单次请求的等待时间防止某个卡死的视频 URL 拖住整个任务。这里的关键区别是不设置streamTrue时resp.content会聚合完整响应体视频大了以后内存和磁盘 IO 都会成为瓶颈。设置chunk_size也不是越小越好如果改成 1KB每个视频会产生几万次磁盘写入调用反而更慢。注意不要让chunk_size超过服务端单次返回的缓冲上限否则 requests 内部仍会按自己的缓冲大小切块设置值只作为上限参考。256KB 是我在不同平台视频下载场景下试过比较稳的取值。4.2 ThreadPoolExecutor 控制并发下载单线程逐个下载时网络往返的等待时间完全串行几百个视频可能要跑几十分钟。用ThreadPoolExecutor可以把下载任务分发到多个线程但并发数需要权衡太快会触发服务端限流。from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path def batch_download(items: list[dict], out_dir: str, max_workers: int 4): Path(out_dir).mkdir(parentsTrue, exist_okTrue) with ThreadPoolExecutor(max_workersmax_workers) as executor: futures {} for item in items: video_id item[video_id] url item[video_url] ext url.split(?)[0].rsplit(., 1)[-1] if . in url.split(?)[0] else mp4 filepath str(Path(out_dir) / f{video_id}.{ext}) futures[executor.submit(download_video, url, filepath)] video_id for future in as_completed(futures): video_id futures[future] try: future.result() print(fok: {video_id}) except Exception as exc: print(ffail: {video_id} - {exc})逻辑说明executor.submit把download_video(url, filepath)丢进线程池返回的future以video_id为键存进字典as_completed在任务完成时逐个取结果失败时打印异常不影响其他视频继续下载。参数说明max_workers4是面对接口频控的保守值本地网络好且接口无风控时可以提升到 8 或 12ext先切掉 URL 的查询参数再取最后一段后缀否则?后面的参数串会被误判成扩展名。文件名直接使用video_id既避免操作系统文件名非法字符问题也方便与 CSV 里的数据行对应。4.3 下载失败后的重试与断点续传快手视频地址有时是带时效的签名 URL过期后返回 403。面对这种情况工程上常用两个手段一是统一重试二是利用 HTTP Range 实现断点续传。断点续传的意义在于当下载到一半连接断开时不需要从头拉整个文件而是从已写入的字节数继续。import os def download_with_resume(video_url: str, filepath: str, retries: int 3): for attempt in range(retries): existing os.path.getsize(filepath) if os.path.exists(filepath) else 0 headers {Range: fbytes{existing}-} if existing else {} resp requests.get(video_url, streamTrue, headersheaders, timeout30) if resp.status_code 416: print(already complete) return resp.raise_for_status() mode ab if existing else wb with open(filepath, mode) as f: for chunk in resp.iter_content(chunk_size1024 * 256): if chunk: f.write(chunk) if os.path.getsize(filepath) int(resp.headers.get(Content-Length, 0)): return print(give up:, filepath)逻辑说明existing取本地文件已有字节数Range: bytes{existing}-要求服务端从该偏移继续传若服务端返回206追加写入若返回200说明服务端忽略 Range此时应该从头覆盖所以代码用mode区分ab和wb。416表示请求范围超出文件总长度等价于文件已经下完直接退出。参数说明retries3控制最大尝试次数每次重试前应配合退避等待避免连续打到同一个限流接口。下载完成后比较本地文件大小与服务端Content-Length不相等就说明传输异常需要重试。这里有个细节部分服务端不返回Content-Lengthint(resp.headers.get(Content-Length, 0))会取到 0此时条件判断失效稳妥做法是再校验文件尾部是否为完整 FLV/MP4 box但日常采集场景下以上逻辑已经能覆盖九成异常。5. 频控识别的退避策略与合规保存边界5.1 通过日志特征识别限流脚本跑久了最容易出现的不是 IP 被封而是服务端开始对高频接口返回异常。常见的三个信号同一接口连续返回 403、JSON 里的result字段从 1 变为 100、单次响应耗时从 200 毫秒涨到 3 秒以上。把这些信号记录到日志比单纯看状态码更早暴露问题。import time import logging def safe_get(url: str, session: requests.Session, max_retry: int 5): for attempt in range(max_retry): resp session.get(url, timeout15) if resp.status_code 200 and resp.json().get(result) 1: return resp.json() wait min(2 ** attempt * 3, 60) logging.warning(retry %s after %ss, status%s, url, wait, resp.status_code) time.sleep(wait) raise RuntimeError(ffailed after retries: {url})逻辑说明每轮先请求只有状态码 200 且业务码为 1 才认为成功否则按指数退避等待后重试。min(2 ** attempt * 3, 60)计算出的等待序列是 3、6、12、24、48 秒触顶 60 秒后不再增长。日志里同时记录 URL、等待秒数和状态码排查时能直接看出是否集中在某个接口上。5.2 在请求间隔中加入随机抖动纯固定间隔的定时请求在服务端看来节奏过于规整容易被识别。常见做法是在基础间隔上叠加均匀分布的随机值让请求时间点更接近真实用户的滚动节奏。import random def jitter_sleep(base: float 1.5, spread: float 0.8): time.sleep(random.uniform(base - spread, base spread))参数说明base是平均等待秒数spread控制抖动幅度实际等待落在 0.7 到 2.3 秒之间。需要说明的是抖动不是用来对抗反爬而是降低请求在时间轴上碰撞的概率。和上一小节的指数退避搭配使用长时间采集时接口返回成功率会明显提升。5.3 数据保存的合规边界与目录归档采集前查看目标站点的 robots.txt 和平台协议是基本功。快手公开主页的视频元数据、开放平台接口数据与需要登录后可见的完播明细两者的合规性完全不同。kuaishou.py 默认只保存公开视频的标题、作者、播放量和视频文件不涉及非公开数据字段这个范围相对安全。保存时我在每行元数据里追加采集时间戳便于后续审计和数据回溯。rows.append({ video_id: video_id, crawled_at: time.strftime(%Y-%m-%d %H:%M:%S), })另外视频文件建议按日期分目录归档而不是全部平铺在一个目录下。文件数上万之后单目录的索引耗时和删除耗时都会明显增加。归档方式很简单在batch_download里把out_dir拼上当天日期例如videos/20250214/任务的写入范围就限定在了当天目录排查问题也能快速按时间定位。本文还有配套的精品资源点击获取