Python 视频爬取教程

2025-10-16 19:34:29 | 世界杯足球队 | admin | 7266°c

文章目录

前言一、视频爬取基本原理二、必备工具与库三、基础视频爬取示例(以 B 站为例)四、处理动态加载视频(以抖音为例)五、高级技巧:多线程 / 异步下载六、法律风险与道德准则七、常见问题与解决方案

前言

以下是一个完整的 Python 视频爬取教程,包含基础原理、工具选择、代码实现和法律风险提示。

一、视频爬取基本原理

网页结构分析 视频网站通常使用 HTML5 标签或 Flash 播放器嵌入视频 真实视频地址可能隐藏在 JavaScript 代码或 API 请求中 需要通过浏览器开发者工具(F12)分析网络请求 反爬机制 验证码、IP 封禁、User-Agent 校验 动态加载、加密视频地址 登录验证、Cookie/Session 跟踪

二、必备工具与库

请求库 requests:发送 HTTP 请求获取网页内容 aiohttp:异步请求,提高爬取效率 解析库 BeautifulSoup:解析 HTML/XML lxml:高性能 XML/HTML 解析器 Selenium:自动化浏览器操作 视频处理 FFmpeg:视频下载、转码(需单独安装) moviepy:视频剪辑与处理

三、基础视频爬取示例(以 B 站为例)

以下是一个爬取 B 站视频的完整代码:

import requests

import json

import re

import os

from bs4 import BeautifulSoup

from urllib.parse import urljoin

class BilibiliCrawler:

def __init__(self):

self.headers = {

'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',

'Referer': 'https://www.bilibili.com/'

}

self.session = requests.Session()

def get_video_info(self, bvid):

"""获取视频信息和真实播放地址"""

url = f"https://api.bilibili.com/x/web-interface/view?bvid={bvid}"

response = self.session.get(url, headers=self.headers)

data = response.json()

if data['code'] != 0:

print(f"获取视频信息失败: {data['message']}")

return None

video_info = data['data']

title = video_info['title']

title = re.sub(r'[\\/:*?"<>|]', '_', title) # 处理文件名非法字符

# 获取视频播放地址

cid = video_info['cid']

play_url = f"https://api.bilibili.com/x/player/playurl?bvid={bvid}&cid={cid}&fnval=16"

play_response = self.session.get(play_url, headers=self.headers)

play_data = play_response.json()

if play_data['code'] != 0:

print(f"获取播放地址失败: {play_data['message']}")

return None

# 提取最高质量视频地址

video_url = play_data['data']['durl'][0]['url']

return {

'title': title,

'video_url': video_url

}

def download_video(self, video_info, save_path='./videos'):

"""下载视频"""

if not os.path.exists(save_path):

os.makedirs(save_path)

title = video_info['title']

video_url = video_info['video_url']

file_path = os.path.join(save_path, f"{title}.mp4")

print(f"开始下载: {title}")

print(f"视频地址: {video_url}")

try:

response = self.session.get(video_url, headers=self.headers, stream=True)

response.raise_for_status()

with open(file_path, 'wb') as f:

for chunk in response.iter_content(chunk_size=8192):

if chunk:

f.write(chunk)

print(f"下载完成: {file_path}")

return True

except Exception as e:

print(f"下载失败: {e}")

return False

# 使用示例

if __name__ == "__main__":

crawler = BilibiliCrawler()

video_info = crawler.get_video_info("BV1xx411c7mz") # 替换为实际BV号

if video_info:

crawler.download_video(video_info)

四、处理动态加载视频(以抖音为例)

对于使用 JavaScript 动态加载的视频,需要使用 Selenium 模拟浏览器行为:

from selenium import webdriver

from selenium.webdriver.chrome.service import Service

from selenium.webdriver.chrome.options import Options

from selenium.webdriver.common.by import By

from selenium.webdriver.support.ui import WebDriverWait

from selenium.webdriver.support import expected_conditions as EC

import time

import os

def download_douyin_video(url, save_path='./douyin_videos'):

if not os.path.exists(save_path):

os.makedirs(save_path)

# 配置Chrome浏览器

chrome_options = Options()

chrome_options.add_argument('--headless') # 无头模式

chrome_options.add_argument('--disable-gpu')

chrome_options.add_argument('--no-sandbox')

chrome_options.add_argument(f'user-agent={USER_AGENT}')

service = Service('path/to/chromedriver') # 替换为你的chromedriver路径

driver = webdriver.Chrome(service=service, options=chrome_options)

try:

driver.get(url)

print(f"访问页面: {url}")

# 等待视频元素加载

video_element = WebDriverWait(driver, 10).until(

EC.presence_of_element_located((By.TAG_NAME, 'video'))

)

# 获取视频源地址

video_url = video_element.get_attribute('src')

if not video_url:

# 尝试从source标签获取

source_elements = video_element.find_elements(By.TAG_NAME, 'source')

if source_elements:

video_url = source_elements[0].get_attribute('src')

if not video_url:

print("未找到视频地址")

return False

# 处理相对URL

if not video_url.startswith('http'):

video_url = urljoin(url, video_url)

# 获取视频标题

title = driver.title

title = title.replace(' - 抖音', '').strip()

title = re.sub(r'[\\/:*?"<>|]', '_', title)

# 下载视频

file_path = os.path.join(save_path, f"{title}.mp4")

print(f"开始下载视频: {title}")

print(f"视频地址: {video_url}")

# 使用requests下载视频

response = requests.get(video_url, stream=True)

with open(file_path, 'wb') as f:

for chunk in response.iter_content(chunk_size=8192):

if chunk:

f.write(chunk)

print(f"视频下载完成: {file_path}")

return True

except Exception as e:

print(f"下载失败: {e}")

return False

finally:

driver.quit()

# 使用示例

if __name__ == "__main__":

video_url = "https://www.douyin.com/video/7012345678901234567" # 替换为实际抖音视频URL

download_douyin_video(video_url)

五、高级技巧:多线程 / 异步下载

使用ThreadPoolExecutor实现多线程下载:

from concurrent.futures import ThreadPoolExecutor

def download_multiple_videos(bvid_list, max_workers=5):

crawler = BilibiliCrawler()

def download_task(bvid):

video_info = crawler.get_video_info(bvid)

if video_info:

crawler.download_video(video_info)

with ThreadPoolExecutor(max_workers=max_workers) as executor:

executor.map(download_task, bvid_list)

# 使用示例

if __name__ == "__main__":

bvid_list = ["BV1xx411c7mz", "BV1JL4y1S7VG", "BV1mQ4y1d7th"] # 替换为实际BV号列表

download_multiple_videos(bvid_list)

六、法律风险与道德准则

遵守网站条款 大多数视频网站禁止未经授权的爬取行为 检查网站robots.txt文件,避免爬取禁止的内容 版权问题 下载受版权保护的内容可能违反法律 仅用于个人学习研究,避免商业传播 合理使用 设置合理的请求间隔(如 1-3 秒) 控制并发数量,避免对目标服务器造成压力

七、常见问题与解决方案

IP 封禁 使用代理 IP 池(如requests-proxies) 控制请求频率 验证码 手动识别或使用第三方验证码服务(如打码平台) 尝试使用 Cookie 保持会话状态 加密视频地址 逆向分析 JavaScript 代码 使用浏览器开发者工具监控网络请求 视频合并 对于分段视频,使用FFmpeg合并: bash ffmpeg -i “concat:part1.ts|part2.ts|part3.ts” -c copy output.mp4