2026最新Python抓取YouTube数据实战【从零搭建】

Python抓取YouTube数据实战:从零搭建你的视频监控系统

你是否曾经想过这样的场景:想追踪某个Up主的所有视频数据变化,或者想批量分析某个领域的热门视频规律,却苦于手动复制粘贴太麻烦?

今天,我们就用Python搭建一个完整的YouTube视频监控系统,让你能够自动抓取、存储和分析任何视频的数据。这个系统不仅会获取基本信息,还能持续追踪数据变化,帮你发现趋势。

大规模抓取时会遇到IP被封的问题,使用代理IP是必然选择。

kookeey 全球代理IP点击按钮免费试用

一、系统设计:我们要搭建什么?

在开始写代码之前,先来看看我们要构建的系统长什么样:

2026最新Python抓取YouTube数据实战【从零搭建】

这个系统能做什么?

  • 自动抓取:视频标题、观看次数、点赞数、评论数、发布日期
  • 持续监控:每天定时运行,记录数据变化
  • 趋势分析:自动生成观看量增长曲线
  • 异常报警:当数据出现异常波动时提醒你

二、环境搭建:5分钟准备好工具

2.1 安装Python(如果还没装)

  1. 访问 python.org
  2. 下载最新版Python(3.9以上)
  3. 安装时务必勾选“Add Python to PATH”
  4. 安装完成后,打开命令提示符(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
免费获取 kookeey 新用户福利 🎁
200MB 动态流量
100MB 移动代理
288元 优惠大礼包
100%独享IP ISP 支持独享端口 / API 调用

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 代理使用的最佳实践

  1. 按需启用代理:只在需要时使用代理,因为代理会降低速度
  2. 合理轮换IP:不要每次请求都换IP,容易引起怀疑
  3. 监控代理状态:记录哪些代理可用,哪些已失效
  4. 备用方案:当代理失效时,自动切换到下一个

常见问题(FAQ)
能抓取整个频道的所有视频吗?
可以!把频道URL传给yt-dlp,它会自动获取该频道的所有视频列表。本文的监控系统稍作修改就能支持频道监控。
能抓取播放列表吗?
ASIN(Amazon Standard Identification Number)可以!yt-dlp可以提取自动生成的字幕和上传者提供的字幕,支持多种语言。

六、运行你的监控系统

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数据,更重要的是搭建了一个完整的、可扩展的监控系统。可以尝试:

  1. 批量监控任意数量的YouTube视频
  2. 自动记录数据变化,追踪历史趋势
  3. 生成图表直观展示观看量增长
  4. 设置定时让程序每天自动运行
  5. 扩展定制根据需求添加新功能

这个系统的应用场景非常广泛:

  • 自媒体创作者:分析自己的视频表现
  • 市场研究人员:监控竞争对手的爆款内容
  • 数据分析爱好者:研究视频流行规律
  • 品牌运营:追踪合作UP主的数据变化

2026最新Python抓取YouTube数据实战【从零搭建】

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

(0)
kookeeykookeey
上一篇 2026-03-04 18:47
下一篇 2024-01-04 08:41

相关推荐

  • HTTP代理ip对于网络安全是否有贡献?

     使用HTTP代理IP可以在一定程度上增强网络安全,但它并不能单独解决所有安全问题。以下是一些HTTP代理IP对网络安全的贡献:        1,匿名性和隐私保护:使用HTTP代理IP可以隐藏用户的真实IP地址,提高用户的匿名性,从而减少被追踪的风险。这对于保护个人隐私是有帮助的。     &nb…

    2024-02-27
  • TikTok与亚马逊合作下的跨境电商新机遇

    近年来,随着全球化的不断推进和互联网技术的飞速发展,跨境电商行业迎来了前所未有的发展机遇。近日,TikTok与亚马逊宣布达成重要合作,这一消息不仅引发了业界的广泛关注,更为跨境电商行业注入了新的活力。本文将深入分析TikTok与亚马逊合作的背景、影响以及未来跨境电商的新机遇。 一、合作背景分析 1. 亚马逊的增长困境 亚马逊作为全球最大的电商平台之一,近年来…

    2024-08-23
  • 海外IP/VPN/代理节点有什么区别?应该怎么选?

    随着互联网行业、跨境电商、TikTok出海等行业的快速发展,行业里对海外网络的需求越来越多,我们经常会听到几个名词,vpn、节点或者是代理ip等,大家都想要买到实惠适合自己的工具,应该怎么选。在选择之前,我们应该了解下这几个词的区别。VPN 和代理是两种上网方式,都是通过在请求和访问终端之间,对你的网络请求进行(直接或加密)中转,隐藏你的真实 IP 地址,并…

    2023-12-06
  • 海外IP代理应用:亚马逊使用什么代理IP?

    代理IP作为网络活动的有力工具,同时也是跨境电商的必备神器。亚马逊作为跨境电商的头部平台,吸引了大量的跨境电商玩家入驻,想要做好亚马逊,养号、测评都需要代理IP的帮助。那么应该使用什么代理IP呢?如何使用?下面具体介绍。 一、为什么要使用代理IP访问亚马逊隐私保护:使用代理IP可以帮助用户在访问亚马逊时隐藏自己的真实IP地址,增加匿名性,从而保护个人隐私。避…

    2024-03-11
  • 海外代理IP与VPN有何区别?哪个更好? 

    当谈到网络安全和IP变更时,人们会想到VPN和IP代理服务器。很多人很困惑,它们之间有什么区别,应该选择哪一个呢?这取决于您的需求来决定哪一个更好。 一、什么是VPN与IP代理? VPN 是虚拟专用网络 (Virtual Private Networks) 的缩写,它使用计算机设备和互联网之间的虚拟连接来对通过加密隧道的所有数据进行加密。您的数据将通过 12…

    2023-11-22