Python 爬虫技术：抖音视频批量提取与数据存储

原创

小白学大数据

发布于 2026-01-19 16:52:56

9340

一、技术选型与前期准备

1. 核心技术栈说明

本次爬虫开发采用 Python 作为核心编程语言，搭配以下第三方库，各库的核心作用如下：

requests：用于发送 HTTP/HTTPS 请求，获取抖音平台的接口响应数据，是爬虫的基础网络请求工具。
jsonpath：用于快速解析接口返回的 JSON 格式数据，精准提取视频链接、作者信息、视频标题等核心字段，比原生json库更灵活高效。
aiohttp（可选，异步增强）：用于实现异步网络请求，大幅提升批量爬取的效率，适合大规模视频提取场景。
pandas：用于整理爬取到的结构化数据，最终生成 Excel/CSV 文件进行存储，方便后续数据分析与查看。
os：Python 内置库，用于创建文件目录、处理文件路径，实现视频文件的本地持久化存储。
requests_toolbelt：辅助处理大文件下载，避免视频下载过程中出现断连、文件损坏等问题。

2. 前期环境搭建

首先需要安装所需的第三方依赖库

3. 关键前置知识：抖音接口特性与合规声明

接口特性：抖音的公开内容数据通过 API 接口返回 JSON 格式数据，直接爬取网页端（PC 端）数据难度较大，本文优先针对抖音移动端接口（无登录权限可访问的公开内容）进行开发，避免涉及复杂的登录验证、cookie 加密等问题。
合规声明：
- 本文爬取的内容仅限抖音平台公开可访问的公开视频，不涉及用户隐私数据、付费内容或未公开内容。
- 爬取行为需遵守《网络爬虫自律公约》及抖音平台的《用户服务协议》，不得高频次请求给平台服务器造成压力，不得将爬取数据用于商业盈利、恶意传播等违规用途。
- 若需爬取个人账号专属内容，需先获得账号所有者授权，并通过正规平台接口申请访问权限。

二、核心实现步骤

步骤 1：分析抖音接口，获取视频核心数据

抖音的视频内容接口具有一定的规律，我们以抖音创作者公开主页的视频列表为例（也可针对热门视频榜单接口进行适配）。通过移动端抓包工具（如 Fiddler、Charles）可捕获到视频列表接口，该接口返回的数据包含所有视频的核心信息，我们需要从中提取以下关键字段：

视频唯一标识（video_id）
视频播放链接（play_url）
视频标题（title）
作者昵称（author_name）
发布时间（publish_time）
视频点赞数（like_count）

以下是获取视频列表核心数据的 Python 代码实现：

python

运行

import requests
from jsonpath import jsonpath
import pandas as pd
import os
from datetime import datetime

# 配置项
HEADERS = {
    "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.0 Mobile/15E148 Safari/604.1",
    "Referer": "https://www.douyin.com/",
    "Accept": "application/json, text/plain, */*"
}

# 代理配置信息（指定的代理参数）
proxyHost = "www.16yun.cn"
proxyPort = "5445"
proxyUser = "16QMSOML"
proxyPass = "280651"

# 构造带认证的代理URL格式：http://用户名:密码@代理地址:端口
proxy_url = f"http://{proxyUser}:{proxyPass}@{proxyHost}:{proxyPort}"

# 配置requests所需的proxies字典（同时支持http和https请求代理）
PROXIES = {
    "http": proxy_url,
    "https": proxy_url
}

# 抖音公开视频列表接口（示例，需通过抓包获取有效接口与参数）
VIDEO_LIST_API = "https://www.douyin.com/aweme/v1/aweme/post/"

def get_video_list(user_id, max_page=5):
    """
    获取指定用户的公开视频列表数据
    :param user_id: 抖音用户ID
    :param max_page: 最大获取页数
    :return: 视频核心数据列表
    """
    video_data_list = []
    page = 1
    cursor = 0  # 分页游标，用于加载下一页数据
    
    while page <= max_page:
        # 构造请求参数
        params = {
            "user_id": user_id,
            "cursor": cursor,
            "count": 20,  # 每页返回20条视频
            "aid": "1128",
            "platform": "ios"
        }
        
        try:
            # 发送GET请求，添加proxies参数配置代理，禁止重定向（避免接口跳转）
            response = requests.get(
                url=VIDEO_LIST_API,
                headers=HEADERS,
                params=params,
                proxies=PROXIES,  # 传入配置好的代理信息
                allow_redirects=False,
                timeout=30
            )
            
            # 校验响应状态码
            if response.status_code != 200:
                print(f"第{page}页请求失败，状态码：{response.status_code}")
                break
            
            # 解析JSON响应数据
            response_json = response.json()
            
            # 提取视频列表数据
            aweme_list = jsonpath(response_json, "$.aweme_list[*]")
            if not aweme_list:
                print("已无更多视频数据")
                break
            
            # 遍历视频列表，提取核心字段
            for aweme in aweme_list:
                video_info = {
                    "video_id": jsonpath(aweme, "$.aweme_id")[0] if jsonpath(aweme, "$.aweme_id") else "",
                    "title": jsonpath(aweme, "$.desc")[0] if jsonpath(aweme, "$.desc") else "无标题",
                    "author_name": jsonpath(aweme, "$.author.nickname")[0] if jsonpath(aweme, "$.author.nickname") else "",
                    "play_url": jsonpath(aweme, "$.video.play_addr.url_list[0]")[0] if jsonpath(aweme, "$.video.play_addr.url_list[0]") else "",
                    "like_count": jsonpath(aweme, "$.statistics.digg_count")[0] if jsonpath(aweme, "$.statistics.digg_count") else 0,
                    "publish_time": datetime.fromtimestamp(jsonpath(aweme, "$.create_time")[0]).strftime("%Y-%m-%d %H:%M:%S") if jsonpath(aweme, "$.create_time") else ""
                }
                video_data_list.append(video_info)
            
            # 更新分页游标（获取下一页数据的标识）
            cursor = jsonpath(response_json, "$.next_cursor")[0] if jsonpath(response_json, "$.next_cursor") else 0
            page += 1
            print(f"第{page-1}页数据获取完成，累计获取{len(video_data_list)}条视频")
        
        except Exception as e:
            print(f"第{page}页请求异常：{str(e)}")
            break
    
    return video_data_list

步骤 2：批量下载视频文件到本地

获取到视频的播放链接（play_url）后，我们需要编写下载函数，将视频文件批量保存到本地指定目录。为了避免文件重名，我们以「视频 ID + 标题」作为文件名，并自动创建以作者昵称为名称的文件夹进行分类存储。

python

运行

from requests_toolbelt import streaming_iterator

def download_single_video(video_info, save_base_path="./douyin_videos"):
    """
    下载单个视频文件到本地
    :param video_info: 单个视频核心信息字典
    :param save_base_path: 视频存储根目录
    :return: 下载结果（成功/失败）
    """
    # 提取必要参数
    video_id = video_info.get("video_id")
    title = video_info.get("title")
    author_name = video_info.get("author_name")
    play_url = video_info.get("play_url")
    
    if not video_id or not play_url:
        print("视频ID或播放链接为空，跳过下载")
        return False
    
    # 处理文件名（去除特殊字符，避免路径错误）
    valid_title = "".join([c for c in title if c not in r'\/:*?"<>|'])[:50]  # 截断过长标题
    file_name = f"{video_id}_{valid_title}.mp4"
    
    # 构造存储目录（按作者分类）
    save_dir = os.path.join(save_base_path, author_name)
    if not os.path.exists(save_dir):
        os.makedirs(save_dir, exist_ok=True)
    
    # 构造完整文件路径
    file_path = os.path.join(save_dir, file_name)
    
    # 避免重复下载
    if os.path.exists(file_path):
        print(f"视频已存在，跳过下载：{file_name}")
        return True
    
    try:
        # 发送GET请求，流式获取视频数据（适合大文件下载）
        with requests.get(
            url=play_url,
            headers=HEADERS,
            stream=True,
            timeout=60
        ) as response:
            if response.status_code != 200:
                print(f"视频下载失败：{file_name}，状态码：{response.status_code}")
                return False
            
            # 写入视频文件到本地
            with open(file_path, "wb") as f:
                for chunk in streaming_iterator(response):
                    f.write(chunk)
        
        print(f"视频下载成功：{file_name}")
        return True
    
    except Exception as e:
        print(f"视频下载异常：{file_name}，错误信息：{str(e)}")
        # 删除损坏的文件
        if os.path.exists(file_path):
            os.remove(file_path)
        return False

def batch_download_videos(video_data_list):
    """
    批量下载视频文件
    :param video_data_list: 视频核心数据列表
    :return: 下载成功数量
    """
    success_count = 0
    for video_info in video_data_list:
        if download_single_video(video_info):
            success_count += 1
    
    print(f"批量下载完成，成功下载{success_count}/{len(video_data_list)}个视频")
    return success_count

步骤 3：结构化数据存储（Excel/CSV 格式）

除了下载视频文件，我们还需要将视频的核心元数据（标题、作者、点赞数等）以结构化格式存储，方便后续数据分析。这里使用pandas库将数据转换为 DataFrame，并导出为 Excel 和 CSV 文件。

python

运行

def save_video_metadata(video_data_list, save_path="./douyin_data"):
    """
    保存视频元数据为Excel和CSV格式
    :param video_data_list: 视频核心数据列表
    :param save_path: 数据存储目录
    :return: 存储结果
    """
    if not video_data_list:
        print("无视频元数据可存储")
        return False
    
    # 确保存储目录存在
    if not os.path.exists(save_path):
        os.makedirs(save_path, exist_ok=True)
    
    # 转换为DataFrame
    df = pd.DataFrame(video_data_list)
    
    # 构造文件名（按当前时间戳命名，避免重名）
    current_time = datetime.now().strftime("%Y%m%d_%H%M%S")
    excel_file_path = os.path.join(save_path, f"抖音视频元数据_{current_time}.xlsx")
    csv_file_path = os.path.join(save_path, f"抖音视频元数据_{current_time}.csv")
    
    try:
        # 导出为Excel文件（包含表头，忽略索引）
        df.to_excel(excel_file_path, index=False, engine="openpyxl")
        
        # 导出为CSV文件（UTF-8编码，避免中文乱码）
        df.to_csv(csv_file_path, index=False, encoding="utf-8-sig")
        
        print(f"元数据存储成功，Excel文件：{excel_file_path}")
        print(f"元数据存储成功，CSV文件：{csv_file_path}")
        return True
    
    except Exception as e:
        print(f"元数据存储失败，错误信息：{str(e)}")
        return False

步骤 4：整合所有功能，实现批量爬取与存储

最后，我们编写主函数，整合上述所有功能，实现「获取视频列表→批量下载视频→存储元数据」的完整流程。

python

运行

def main():
    """
    主函数：实现抖音视频批量提取与数据存储的完整流程
    """
    # 配置参数（替换为有效抖音用户ID）
    target_user_id = "1234567890"
    max_crawl_page = 3
    
    print("========== 开始爬取抖音视频数据 ==========")
    # 步骤1：获取视频列表数据
    video_data_list = get_video_list(
        user_id=target_user_id,
        max_page=max_crawl_page
    )
    
    if not video_data_list:
        print("未获取到有效视频数据，任务终止")
        return
    
    print(f"========== 共获取到{len(video_data_list)}条视频数据，开始批量下载 ==========")
    # 步骤2：批量下载视频文件
    batch_download_videos(video_data_list)
    
    print("========== 视频下载完成，开始存储元数据 ==========")
    # 步骤3：存储视频元数据
    save_video_metadata(video_data_list)
    
    print("========== 所有任务执行完成 ==========")

if __name__ == "__main__":
    main()

三、关键问题与优化方案

1. 常见问题排查

接口请求返回 403/302：抖音接口有严格的反爬机制，User-Agent需模拟移动端设备，同时避免高频次请求；可添加请求间隔（time.sleep(1-3)），降低爬取频率。
JSON 解析失败：部分接口返回数据可能包含乱码或特殊字符，可添加response.encoding = "utf-8"指定编码格式。
视频下载后无法播放：抖音视频链接有有效期，且部分链接需要携带有效cookie，可通过抓包更新play_url的获取方式，或添加cookie到请求头中。

2. 性能优化方向

异步请求优化：使用aiohttp替代requests实现异步网络请求，同时批量下载视频，大幅提升爬取效率，适合大规模爬取场景。
断点续传功能：在视频下载函数中添加文件大小校验，支持断点续传，避免因网络中断导致重复下载。
数据去重：基于video_id建立唯一索引，避免重复爬取同一视频数据。
日志记录：添加logging模块，记录爬取过程中的关键信息与错误日志，方便后续问题排查。

3. 扩展功能建议

多用户批量爬取：配置用户 ID 列表，循环执行爬取任务，实现多账号视频数据提取。
数据库存储：将视频元数据存储到 MySQL、MongoDB 等数据库中，方便后续数据查询与管理。
视频信息补充：扩展提取视频评论、转发数、收藏数等更多字段，丰富数据维度。

四、总结

本文通过 Python 实现了抖音视频的批量提取与数据存储，完整覆盖了「接口分析→数据爬取→视频下载→元数据存储」的全流程，提供了可直接运行的代码示例与详细的步骤讲解。需要注意的是，抖音平台的接口与反爬机制会持续更新，本文的示例接口仅作技术演示，实际使用时需要通过抓包工具获取最新的有效接口与参数。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python

python爬虫

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python

python爬虫

登录后参与评论

0 条评论

热度