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

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

一、技术选型与前期准备

  1. 核心技术栈说明
    本次爬虫开发采用 Python 作为核心编程语言,搭配以下第三方库,各库的核心作用如下:
    ● requests:用于发送 HTTP/HTTPS 请求,获取抖音平台的接口响应数据,是爬虫的基础网络请求工具。
    ● jsonpath:用于快速解析接口返回的 JSON 格式数据,精准提取视频链接、作者信息、视频标题等核心字段,比原生json库更灵活高效。
    ● aiohttp(可选,异步增强):用于实现异步网络请求,大幅提升批量爬取的效率,适合大规模视频提取场景。
    ● pandas:用于整理爬取到的结构化数据,最终生成 Excel/CSV 文件进行存储,方便后续数据分析与查看。
    ● os:Python 内置库,用于创建文件目录、处理文件路径,实现视频文件的本地持久化存储。
    ● requests_toolbelt:辅助处理大文件下载,避免视频下载过程中出现断连、文件损坏等问题。
  2. 前期环境搭建
    首先需要安装所需的第三方依赖库
  3. 关键前置知识:抖音接口特性与合规声明
  4. 接口特性:抖音的公开内容数据通过 API 接口返回 JSON 格式数据,直接爬取网页端(PC 端)数据难度较大,本文优先针对抖音移动端接口(无登录权限可访问的公开内容)进行开发,避免涉及复杂的登录验证、cookie 加密等问题。
  5. 合规声明:
    ○ 本文爬取的内容仅限抖音平台公开可访问的公开视频,不涉及用户隐私数据、付费内容或未公开内容。
    ○ 爬取行为需遵守《网络爬虫自律公约》及抖音平台的《用户服务协议》,不得高频次请求给平台服务器造成压力,不得将爬取数据用于商业盈利、恶意传播等违规用途。
    ○ 若需爬取个人账号专属内容,需先获得账号所有者授权,并通过正规平台接口申请访问权限。
    二、核心实现步骤
    步骤 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 实现了抖音视频的批量提取与数据存储,完整覆盖了「接口分析→数据爬取→视频下载→元数据存储」的全流程,提供了可直接运行的代码示例与详细的步骤讲解。需要注意的是,抖音平台的接口与反爬机制会持续更新,本文的示例接口仅作技术演示,实际使用时需要通过抓包工具获取最新的有效接口与参数。
相关文章
|
6月前
|
数据采集 JSON JavaScript
Python 抖音爬虫从 0 到 1 实战:环境配置与数据爬取全教程
Python 抖音爬虫从 0 到 1 实战:环境配置与数据爬取全教程
|
9月前
|
数据采集 自然语言处理 数据可视化
时序数据分析:Python爬取新浪财经频道新闻并绘制趋势图
时序数据分析:Python爬取新浪财经频道新闻并绘制趋势图
|
3月前
|
数据采集 JSON 数据挖掘
抖音搜索页数据批量爬取,多关键词同步采集实现
抖音搜索页数据批量爬取,多关键词同步采集实现
|
3月前
|
数据采集 Web App开发 JavaScript
Python 爬虫动态 JS 渲染与无头浏览器实战选型指南
Python 爬虫动态 JS 渲染与无头浏览器实战选型指南
|
3月前
|
数据采集 JSON 数据安全/隐私保护
Python 爬虫爬取应用商店数据:请求构造与数据解析
Python 爬虫爬取应用商店数据:请求构造与数据解析
|
11月前
|
数据采集 Web App开发 JavaScript
应对反爬:使用Selenium模拟浏览器抓取12306动态旅游产品
应对反爬:使用Selenium模拟浏览器抓取12306动态旅游产品
|
数据采集 存储 Web App开发
如何避免爬虫因Cookie过期导致登录失效
如何避免爬虫因Cookie过期导致登录失效
|
7月前
|
数据采集 Web App开发 API
爬虫技术选股:Python 自动化筛选潜力股
爬虫技术选股:Python 自动化筛选潜力股
|
JSON 监控 API
抖音视频列表API秘籍!轻松获取视频列表数据
抖音视频列表API是抖音开放平台提供的核心接口,支持按关键词、分类、排序方式筛选视频,适用于内容推荐、趋势分析等场景。接口返回含视频ID、标题、播放量等50+字段,支持分页获取,通过HTTP GET请求调用,返回JSON格式数据,便于开发者快速集成与处理。需注册平台账号获取访问权限。
2710 56