Python抓取YouTube数据实战:从零搭建你的视频监控系统
你是否曾经想过这样的场景:想追踪某个Up主的所有视频数据变化,或者想批量分析某个领域的热门视频规律,却苦于手动复制粘贴太麻烦?
今天,我们就用Python搭建一个完整的YouTube视频监控系统,让你能够自动抓取、存储和分析任何视频的数据。这个系统不仅会获取基本信息,还能持续追踪数据变化,帮你发现趋势。
大规模抓取时会遇到IP被封的问题,使用代理IP是必然选择。
一、系统设计:我们要搭建什么?
在开始写代码之前,先来看看我们要构建的系统长什么样:

这个系统能做什么?
- ✅ 自动抓取:视频标题、观看次数、点赞数、评论数、发布日期
- ✅ 持续监控:每天定时运行,记录数据变化
- ✅ 趋势分析:自动生成观看量增长曲线
- ✅ 异常报警:当数据出现异常波动时提醒你
二、环境搭建:5分钟准备好工具
2.1 安装Python(如果还没装)
- 访问 python.org
- 下载最新版Python(3.9以上)
- 安装时务必勾选“Add Python to PATH”
- 安装完成后,打开命令提示符(cmd),输入:
python --version
看到版本号说明成功!
2.2 安装需要的工具包
在命令提示符中逐条运行以下命令:
pip install yt-dlp
pip install pandas
pip install matplotlib
pip install schedule
每个工具的作用:
- yt-dlp:核心抓取工具,YouTube数据的瑞士军刀
- pandas:数据处理,像Excel一样操作表格
- matplotlib:画图工具,生成趋势图表
- schedule:定时任务,让程序自动运行
三、核心代码:一步一步搭建系统
3.1 第一步:创建项目文件夹
在桌面新建一个文件夹,取名 youtube_monitor,里面创建两个文件:
monitor.py:主程序文件videos.txt:存放要监控的视频链接
3.2 编写抓取函数(最核心的部分)
打开 monitor.py,我们先写一个函数来抓取单个视频的数据:
import csv
import json
from datetime import datetime
from yt_dlp import YoutubeDL
import pandas as pd
import os
def fetch_video_data(video_url):
"""
抓取单个视频的完整数据
参数 video_url: YouTube视频链接
返回: 包含视频数据的字典
"""
print(f"正在抓取: {video_url}")
# 配置yt-dlp选项
ydl_opts = {
'quiet': True, # 不输出多余信息
'no_warnings': True, # 忽略警告
'extract_flat': False, # 提取完整信息
'force_generic_extractor': False,
}
try:
with YoutubeDL(ydl_opts) as ydl:
# 提取视频信息
info = ydl.extract_info(video_url, download=False)
# 提取我们需要的数据
video_data = {
'抓取时间': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'视频ID': info.get('id', ''),
'标题': info.get('title', ''),
'频道': info.get('uploader', ''),
'频道ID': info.get('channel_id', ''),
'发布日期': info.get('upload_date', ''),
'时长(秒)': info.get('duration', 0),
'观看次数': info.get('view_count', 0),
'点赞数': info.get('like_count', 0),
'评论数': info.get('comment_count', 0),
'视频链接': video_url,
}
print(f"✓ 抓取成功: {video_data['标题'][:30]}...")
return video_data
except Exception as e:
print(f"✗ 抓取失败: {e}")
return None
# 测试一下
if __name__ == "__main__":
test_url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
data = fetch_video_data(test_url)
if data:
print("\n抓取结果预览:")
for key, value in data.items():
print(f"{key}: {value}")
3.3 第二步:数据存储功能
抓取的数据需要保存下来,方便以后分析。添加保存和读取功能:
def save_to_csv(data, filename='youtube_data.csv'):
"""
将抓取的数据追加保存到CSV文件
"""
file_exists = os.path.isfile(filename)
with open(filename, 'a', newline='', encoding='utf-8-sig') as f:
writer = csv.DictWriter(f, fieldnames=data.keys())
# 如果是新文件,先写表头
if not file_exists:
writer.writeheader()
writer.writerow(data)
print(f"数据已保存到 {filename}")
def read_history(video_id=None, filename='youtube_data.csv'):
"""
读取历史数据
如果指定video_id,只返回该视频的数据
"""
if not os.path.isfile(filename):
print("暂无历史数据")
return None
df = pd.read_csv(filename)
if video_id:
df = df[df['视频ID'] == video_id]
return df
# 测试保存功能
def test_save():
test_url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
data = fetch_video_data(test_url)
if data:
save_to_csv(data)
# 读取并显示
df = read_history()
print("\n历史数据预览:")
print(df.head())
3.4 第三步:批量监控功能(同时监控多个视频)
通常我们需要监控多个视频,而不是一个。从文件读取视频列表:
def load_video_urls(filename='videos.txt'):
"""
从文本文件加载要监控的视频链接
每行一个链接
"""
urls = []
try:
with open(filename, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if line and not line.startswith('#'): # 忽略空行和注释
urls.append(line)
print(f"从 {filename} 加载了 {len(urls)} 个视频")
return urls
except FileNotFoundError:
print(f"找不到 {filename} 文件,将使用默认测试链接")
return ["https://www.youtube.com/watch?v=dQw4w9WgXcQ"]
def monitor_all_videos(filename='videos.txt'):
"""
监控所有视频
"""
urls = load_video_urls(filename)
for url in urls:
data = fetch_video_data(url)
if data:
save_to_csv(data)
print("-" * 50)
3.5 第四步:数据分析功能(生成趋势图表)
抓取数据的最终目的是分析。添加数据分析功能:
import matplotlib.pyplot as plt
import matplotlib
matplotlib.rcParams['font.sans-serif'] = ['SimHei'] # 支持中文显示
matplotlib.rcParams['axes.unicode_minus'] = False
def analyze_video(video_id):
"""
分析单个视频的历史数据
"""
df = read_history(video_id)
if df is None or len(df) == 0:
print("没有找到该视频的历史数据")
return
# 将抓取时间转换为datetime类型
df['抓取时间'] = pd.to_datetime(df['抓取时间'])
df = df.sort_values('抓取时间')
# 获取视频标题
title = df.iloc[0]['标题']
print(f"\n===== 视频分析: {title} =====")
print(f"监控时间段: {df['抓取时间'].min()} 到 {df['抓取时间'].max()}")
print(f"数据点数量: {len(df)}")
# 计算变化
first_views = df.iloc[0]['观看次数']
last_views = df.iloc[-1]['观看次数']
growth = last_views - first_views
growth_rate = (growth / first_views * 100) if first_views > 0 else 0
print(f"初始观看: {first_views:,}")
print(f"当前观看: {last_views:,}")
print(f"增长量: {growth:,} ({growth_rate:.1f}%)")
# 绘制趋势图
fig, axes = plt.subplots(2, 1, figsize=(12, 10))
# 观看次数趋势
axes[0].plot(df['抓取时间'], df['观看次数'], marker='o', linewidth=2, color='#FF0000')
axes[0].set_title('观看次数变化趋势', fontsize=14)
axes[0].set_xlabel('抓取时间')
axes[0].set_ylabel('观看次数')
axes[0].grid(True, alpha=0.3)
axes[0].ticklabel_format(style='plain', axis='y')
# 点赞数趋势
axes[1].plot(df['抓取时间'], df['点赞数'], marker='s', linewidth=2, color='#0066CC')
axes[1].set_title('点赞数变化趋势', fontsize=14)
axes[1].set_xlabel('抓取时间')
axes[1].set_ylabel('点赞数')
axes[1].grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig(f'{video_id}_trend.png', dpi=150, bbox_inches='tight')
plt.show()
print(f"\n趋势图已保存为 {video_id}_trend.png")
def generate_report():
"""
生成所有监控视频的报告
"""
df = read_history()
if df is None:
return
# 获取最新的数据(每个视频的最新记录)
latest = df.sort_values('抓取时间').groupby('视频ID').last().reset_index()
print("\n" + "="*70)
print("当前监控报告")
print("="*70)
for _, row in latest.iterrows():
# 获取该视频的历史数据
video_df = df[df['视频ID'] == row['视频ID']].sort_values('抓取时间')
first = video_df.iloc[0]
views_growth = row['观看次数'] - first['观看次数']
views_rate = (views_growth / first['观看次数'] * 100) if first['观看次数'] > 0 else 0
print(f"\n▶ {row['标题'][:50]}")
print(f" 频道: {row['频道']}")
print(f" 当前观看: {row['观看次数']:,} (增长: {views_growth:+,}, {views_rate:.1f}%)")
print(f" 点赞: {row['点赞数']:,} 评论: {row['评论数']:,}")
print(f" 最后更新: {row['抓取时间']}")
3.6 第五步:自动定时运行
让程序每天自动运行,不需要手动启动:
import schedule
import time
def job():
"""
定时执行的任务
"""
print(f"\n{'='*50}")
print(f"开始执行监控任务: {datetime.now()}")
print('='*50)
monitor_all_videos('videos.txt')
print(f"\n任务完成: {datetime.now()}")
print('='*50)
def run_scheduler():
"""
启动定时调度器
"""
# 设置每天在指定时间运行
schedule.every().day.at("09:00").do(job) # 每天上午9点
schedule.every().day.at("21:00").do(job) # 每天晚上9点
print("定时监控已启动...")
print("每天 09:00 和 21:00 会自动抓取数据")
print("按 Ctrl+C 停止运行")
# 立即执行一次
job()
while True:
schedule.run_pending()
time.sleep(60) # 每分钟检查一次
# 如果直接运行本文件,启动定时任务
if __name__ == "__main__":
import sys
if len(sys.argv) > 1:
command = sys.argv[1]
if command == "run":
# 立即运行一次
monitor_all_videos('videos.txt')
elif command == "report":
# 生成报告
generate_report()
elif command == "analyze":
# 分析指定视频
video_id = sys.argv[2] if len(sys.argv) > 2 else None
if video_id:
analyze_video(video_id)
else:
print("请指定视频ID")
elif command == "schedule":
# 启动定时任务
run_scheduler()
else:
print("""
使用方法:
python monitor.py run - 立即运行一次抓取
python monitor.py report - 生成监控报告
python monitor.py analyze [视频ID] - 分析视频趋势
python monitor.py schedule - 启动定时任务
""")
四、配置文件
在项目文件夹中创建 videos.txt 文件:
# 在这里添加你要监控的YouTube视频链接
# 每行一个,以#开头的行会被忽略
https://www.youtube.com/watch?v=dQw4w9WgXcQ
https://www.youtube.com/watch?v=kJQP7kiw5Fk
https://www.youtube.com/watch?v=9bZkp7q19f0
# 可以添加更多视频...
五、配置代理IP解决封禁问题
当你开始大规模抓取或者长时间运行时,很可能会遇到一个问题:IP被YouTube封禁。你会看到类似这样的错误:
text
✗ 抓取失败: HTTP Error 429: Too Many Requests
这是因为YouTube有反爬虫机制,当同一个IP请求过于频繁时,就会返回429错误甚至直接封禁IP。解决方法就是使用代理IP。
5.1 什么是代理IP?
代理IP就像你的网络“替身”。当你通过代理访问网站时,对方看到的是代理的IP地址,而不是你真实的IP。这样你就可以:
- ✅ 轮换IP:每次请求用不同IP,避免被识别
- ✅ 突破限制:绕过地区限制和访问频率限制
- ✅ 提高稳定性:即使某个IP被封,自动切换到其他IP
5.2 配置kookeey代理
下面以kookeey动态住宅代理IP为例,教你如何配置代理。假设你已经在kookeey平台开通了服务,获得了以下信息:
- 代理地址:
gate.kookeey.io:15959 - 账号:
kookeeey - 密码:
12345678
测试代理是否可用
先写一个简单的测试脚本,验证代理配置是否正确:
python
import urllib.request
# 配置代理
proxy = urllib.request.ProxyHandler({
'https': 'http://kookeeey:12345678@gate.kookeey.io:15959'
})
opener = urllib.request.build_opener(proxy)
urllib.request.install_opener(opener)
# 测试代理(访问IP检测网站)
try:
content = urllib.request.urlopen('https://lumtest.com/myip.json').read()
print("代理测试成功!返回的IP信息:")
print(content.decode())
except Exception as e:
print(f"代理测试失败: {e}")
运行这个脚本,如果成功,你会看到类似下面的输出,显示的是代理服务器的IP地址:
json
{"ip": "123.456.78.90", "country": "US", "asn": {"code": 12345}}
5.3 将代理集成到YouTube监控系统
现在,我们把代理功能集成到之前的监控系统中。修改fetch_video_data函数,添加代理支持:
python
import urllib.request
def setup_proxy(proxy_url=None):
"""
配置全局代理
参数 proxy_url: 代理地址,格式如 'http://账号:密码@地址:端口'
如果不传参数,则不使用代理
"""
if proxy_url:
proxy = urllib.request.ProxyHandler({
'https': proxy_url,
'http': proxy_url
})
opener = urllib.request.build_opener(proxy)
urllib.request.install_opener(opener)
print(f"已配置代理: {proxy_url.split('@')[-1]}") # 只显示地址部分,隐藏密码
else:
print("未使用代理,直接连接")
def fetch_video_data(video_url, use_proxy=True):
"""
抓取单个视频的完整数据(带代理选项)
参数 video_url: YouTube视频链接
参数 use_proxy: 是否使用代理
返回: 包含视频数据的字典
"""
print(f"正在抓取: {video_url}")
# 如果需要使用代理,先配置
if use_proxy:
# 这里配置你的Kookee代理信息
proxy_url = 'http://kookeeey:12345678@gate.kookeey.io:15959'
setup_proxy(proxy_url)
else:
setup_proxy() # 不使用代理
# yt-dlp配置(添加代理设置)
ydl_opts = {
'quiet': True,
'no_warnings': True,
'extract_flat': False,
'force_generic_extractor': False,
}
# 如果使用代理,添加到yt-dlp配置中
if use_proxy:
ydl_opts['proxy'] = proxy_url
try:
with YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(video_url, download=False)
video_data = {
'抓取时间': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'视频ID': info.get('id', ''),
'标题': info.get('title', ''),
'频道': info.get('uploader', ''),
'频道ID': info.get('channel_id', ''),
'发布日期': info.get('upload_date', ''),
'时长(秒)': info.get('duration', 0),
'观看次数': info.get('view_count', 0),
'点赞数': info.get('like_count', 0),
'评论数': info.get('comment_count', 0),
'视频链接': video_url,
}
print(f"✓ 抓取成功: {video_data['标题'][:30]}...")
return video_data
except Exception as e:
print(f"✗ 抓取失败: {e}")
return None
5.4 智能代理轮换
如果监控大量视频,单代理也可能被封。我们可以实现一个代理池,自动轮换IP:
import random
class ProxyManager:
"""代理管理器,自动轮换IP"""
def __init__(self):
# 定义多个代理(实际使用中可以从文件读取)
self.proxies = [
'http://kookeeey:12345678@gate.kookee.io:15959',
'http://kookeeey2:87654321@gate2.kookee.io:15959',
# 可以添加更多代理...
]
self.current_index = 0
def get_next_proxy(self):
"""获取下一个代理(轮询方式)"""
proxy = self.proxies[self.current_index]
self.current_index = (self.current_index + 1) % len(self.proxies)
return proxy
def get_random_proxy(self):
"""随机获取一个代理"""
return random.choice(self.proxies)
# 在监控函数中使用代理池
def monitor_all_videos_with_proxy(filename='videos.txt', use_proxy_pool=True):
"""
使用代理池监控所有视频
"""
urls = load_video_urls(filename)
if use_proxy_pool:
proxy_manager = ProxyManager()
for i, url in enumerate(urls):
if use_proxy_pool:
# 每个视频使用不同代理
proxy = proxy_manager.get_next_proxy()
print(f"使用代理: {proxy.split('@')[-1]}")
# 这里需要修改fetch_video_data,让它接收proxy参数
data = fetch_video_data_with_proxy(url, proxy)
else:
data = fetch_video_data(url, use_proxy=False)
if data:
save_to_csv(data)
# 添加延时,避免请求过快
time.sleep(random.uniform(2, 5))
print("-" * 50)
5.5 其他常用代理配置方案
市面上代理服务商,配置方式大同小异:
# 方案1:BrightData代理
brightdata_proxy = 'http://brd-customer-账号-zone-区域:密码@zproxy.lum-superproxy.io:22225'
# 方案2:Crawlbase智能代理
crawlbase_proxy = 'http://API_TOKEN:@smartproxy.crawlbase.com:8012'
# 方案3:自建代理池(使用免费代理)
def get_free_proxies():
"""从免费代理网站获取代理列表(不保证稳定)"""
# 这里可以写爬虫抓取免费代理
pass
建议:生产环境建议使用付费代理,稳定性有保障。
5.6 代理使用的最佳实践
- 按需启用代理:只在需要时使用代理,因为代理会降低速度
- 合理轮换IP:不要每次请求都换IP,容易引起怀疑
- 监控代理状态:记录哪些代理可用,哪些已失效
- 备用方案:当代理失效时,自动切换到下一个
六、运行你的监控系统
6.1 立即运行一次
python monitor.py run
你会看到类似下面的输出:
正在抓取: https://www.youtube.com/watch?v=dQw4w9WgXcQ
✓ 抓取成功: Rick Astley - Never Gonna Give You Up...
数据已保存到 youtube_data.csv
6.2 查看监控报告
python monitor.py report
输出示例:
======================================================================
当前监控报告
======================================================================
▶ Rick Astley - Never Gonna Give You Up (Video)
频道: Rick Astley
当前观看: 1,450,234,567 (增长: +123,456, +9.2%)
点赞: 12,345,678 评论: 2,345,678
最后更新: 2026-03-11 09:00:23
6.3 分析单个视频趋势
python monitor.py analyze dQw4w9WgXcQ
这会生成趋势图并保存为 dQw4w9WgXcQ_trend.png。
6.4 启动定时监控(让程序每天自动运行)
python monitor.py schedule
程序会一直运行,每天自动抓取两次。即使关闭命令提示符也会停止,如果你想让它在后台永久运行:
- Windows:使用任务计划程序
- Mac/Linux:使用cron或nohup
总结:
通过这个以上,你不仅学会了如何用Python抓取YouTube数据,更重要的是搭建了一个完整的、可扩展的监控系统。可以尝试:
- ✅ 批量监控任意数量的YouTube视频
- ✅ 自动记录数据变化,追踪历史趋势
- ✅ 生成图表直观展示观看量增长
- ✅ 设置定时让程序每天自动运行
- ✅ 扩展定制根据需求添加新功能
这个系统的应用场景非常广泛:
- 自媒体创作者:分析自己的视频表现
- 市场研究人员:监控竞争对手的爆款内容
- 数据分析爱好者:研究视频流行规律
- 品牌运营:追踪合作UP主的数据变化

本文来自网络投稿,不代表kookeey立场,如有问题请联系我们