1. 项目概述：从“宅舞”热潮到数据获取的自动化需求

最近几年，一种被称为“宅舞”的舞蹈视频内容在各大视频平台持续火爆。这类视频通常由创作者在特定场景下，伴随着动漫、游戏或流行音乐进行舞蹈表演，因其充满活力的表现形式和贴近特定文化圈层的审美，吸引了海量观众。作为一个对流行文化和数据技术都感兴趣的从业者，我观察到这些视频的更新频率极高，每天都有大量新作品涌现。手动一个个去收藏、下载，效率低下且容易遗漏精彩内容。于是，一个很自然的需求就产生了：能否通过技术手段，自动化地、批量地将这些每日更新的“宅舞”视频采集下来，构建一个属于自己的离线资源库，方便随时、反复观看与研究？

这不仅仅是一个简单的“下载”动作，其背后涉及对特定内容源的持续监控、结构化数据的提取、媒体文件的获取与存储，以及整个流程的自动化调度。从技术角度看，这是一个典型的网络爬虫与数据采集项目，但其对象是动态更新的流媒体内容，因此比爬取静态网页文本或图片要复杂一些。它考验的是对目标网站结构的逆向分析、网络请求的模拟处理、以及应对反爬策略的灵活能力。对于想学习网络数据采集，尤其是涉及视频内容抓取的开发者来说，这是一个非常具有实践价值的练手项目。当然，一切操作的前提是严格遵守相关平台的服务条款，仅将技术用于个人学习与研究，并尊重内容创作者的版权。

2. 核心思路与技术选型解析

要实现“批量爬取每日更新的宅舞视频”这个目标，我们需要将任务拆解成几个清晰的步骤，并为每个步骤选择合适的技术工具。整个流程的核心思路是：发现目标 -> 解析列表 -> 提取链接 -> 下载媒体 -> 组织存储。

2.1 目标平台分析与请求模拟

首先，我们需要确定视频来源。国内主流视频平台是这类内容的主要集散地。我们的程序需要能模拟浏览器访问这些平台，获取页面数据。这里不能使用简单的requests.get()，因为现代网站大量依赖JavaScript动态渲染内容，直接请求得到的HTML可能是空的或者不完整的。

技术选型：Selenium 或 Playwright为了获取完整渲染后的页面内容，我们需要一个能控制真实浏览器的工具。早期常用Selenium，它支持多种浏览器。但近年来，微软开源的Playwright因其更快的速度、更强大的API和更好的异步支持，成为了许多开发者的新宠。它能够可靠地录制和回放用户操作，完美应对动态加载的页面。在本项目中，我选择使用Playwright for Python，因为它能更轻松地处理无限滚动加载的视频列表页。

为什么不用简单的API？有些平台提供了公开API，但通常有严格的调用频率限制，且不一定能获取到所有我们需要的字段（如高清视频地址）。而通过模拟浏览器访问，我们获取的是和普通用户一样的页面数据，虽然需要解析HTML，但数据更全面，也更稳定（只要网站前端不改版）。

2.2 页面解析与数据提取策略

获取到页面HTML后，下一步是从中提取我们关心的信息：视频标题、UP主名称、播放链接、封面图、以及最重要的——视频文件的真实下载地址。

技术选型：BeautifulSoup 或 lxmlBeautifulSoup是Python里最流行的HTML/XML解析库，语法友好，适合快速开发。lxml的解析速度更快，但语法稍显复杂。对于视频平台这种结构相对规整的页面，两者皆可。我习惯用BeautifulSoup，配合css selector或find_all方法，可以精准定位到包含视频信息的DOM节点。

核心挑战：视频地址的获取这是本项目最大的技术难点。网页上播放器中的视频地址（src）通常不是指向一个直接的.mp4或.flv文件，而是一个经过加密或动态生成的m3u8索引文件。m3u8是一种基于HTTP Live Streaming (HLS) 协议的流媒体播放列表，里面包含了视频分片（ts文件）的地址。我们需要：

1. 从页面中找到m3u8文件的链接。
2. 下载并解析这个m3u8文件，得到所有ts分片文件的地址列表。
3. 批量下载所有ts分片。
4. 使用工具（如ffmpeg）将这些ts分片合并成一个完整的视频文件。

这个过程要求我们能分析网页的网络请求（通过浏览器开发者工具的Network面板），找到关键的媒体请求。

2.3 下载与存储架构设计

确定了视频的真实地址后，就需要高效、稳定地下载。对于直接的文件链接，可以用requests流式下载。对于m3u8，则需要一个能处理并发下载分片的模块。

技术选型：aiohttp 与 asyncio由于需要下载大量文件（无论是多个视频，还是一个视频的多个ts分片），同步下载会非常慢。Python的asyncio异步IO框架配合aiohttp库，可以同时发起多个网络请求，极大提升下载效率。我们需要构建一个异步下载器，并合理控制并发连接数，避免对目标服务器造成过大压力或被封禁IP。

存储设计：本地文件系统 + 元数据库下载的视频文件可以按日期或UP主分类存储在本地文件夹。但为了更好地管理，我建议引入一个轻量级数据库（如SQLite），用来记录每条视频的元数据：平台来源、视频ID、标题、UP主、发布时间、本地存储路径、下载状态等。这样，我们可以方便地查询、去重，以及实现断点续传的功能。

2.4 自动化与调度

“每日更新”意味着我们的爬虫需要定期运行。我们不可能一直手动启动脚本。

技术选型：系统任务调度器在Windows上，可以使用“任务计划程序”；在Linux/Mac上，cron是标准选择。我们可以将爬虫脚本设置为每天在固定时间（例如凌晨2点）自动运行一次，检查并下载过去24小时内发布的新视频。脚本本身应该具备良好的日志功能，记录每次运行下载了哪些视频，遇到了什么错误，方便后续排查。

注意：法律与道德边界在开始任何爬取项目前，必须仔细阅读目标网站的robots.txt文件和服务条款。批量下载视频可能违反其使用协议。本技术讨论仅限用于个人学习、研究及对公开数据的合法分析，严禁用于商业用途、盗版传播或任何对网站正常运营造成干扰的行为。务必设置合理的请求间隔（如每请求一次休眠1-2秒），体现“善意爬虫”的原则。

3. 实操步骤：构建你的宅舞视频爬虫

下面，我将以模拟一个主流视频平台的宅舞频道为例，详细拆解实现步骤。请注意，以下代码为技术原理演示，实际网站结构可能不同，需要你根据实际情况调整选择器。

3.1 环境准备与依赖安装

首先，确保你的Python环境在3.7以上。我们使用venv创建虚拟环境是一个好习惯。

# 创建并激活虚拟环境（Linux/Mac） python3 -m venv dance_spider_env source dance_spider_env/bin/activate # 创建并激活虚拟环境（Windows） python -m venv dance_spider_env dance_spider_env\Scripts\activate

安装必要的库：

pip install playwright beautifulsoup4 aiohttp aiosqlite # 安装Playwright所需的浏览器内核 playwright install chromium

这里我们选择Chromium作为浏览器内核，它足够轻量且兼容性好。aiosqlite是SQLite的异步版本，配合我们的异步爬虫更高效。

3.2 分析目标页面与编写解析器

假设我们要爬取的频道列表页URL模式为：https://example.com/channel/dance?page={page}。我们的第一步是使用Playwright打开页面，等待内容加载完成。

import asyncio from playwright.async_api import async_playwright from bs4 import BeautifulSoup import aiohttp import aiosqlite from urllib.parse import urljoin import re import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s: %(message)s') logger = logging.getLogger(__name__) class DanceVideoSpider: def __init__(self, base_url, start_page=1, max_page=5): self.base_url = base_url self.start_page = start_page self.max_page = max_page self.video_list = [] # 用于存储解析到的视频信息字典 async def fetch_page(self, page_num): """使用Playwright获取指定页码的页面HTML""" url = self.base_url.format(page=page_num) async with async_playwright() as p: # 启动无头浏览器，headless=False可以让你看到浏览器操作，调试时有用 browser = await p.chromium.launch(headless=True) context = await browser.new_context( viewport={'width': 1920, 'height': 1080}, user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ...' ) page = await context.new_page() try: logger.info(f"正在访问页面: {url}") await page.goto(url, wait_until='networkidle') # 等待网络空闲，确保内容加载完 # 有些页面是滚动加载，这里可以模拟滚动 for i in range(3): # 滚动3次，每次等待1秒 await page.evaluate('window.scrollTo(0, document.body.scrollHeight)') await page.wait_for_timeout(1000) page_content = await page.content() await browser.close() return page_content except Exception as e: logger.error(f"获取页面 {url} 失败: {e}") await browser.close() return None

获取到HTML后，我们需要编写解析函数。打开浏览器开发者工具，检查视频列表项的HTML结构。假设每个视频卡片都有一个类名为video-card的<div>，里面包含了我们需要的信息。

def parse_video_list(self, html): """从列表页HTML中解析视频基本信息""" if not html: return [] soup = BeautifulSoup(html, 'html.parser') video_items = soup.find_all('div', class_='video-card') # 根据实际结构修改选择器 videos = [] for item in video_items: try: title_elem = item.find('a', class_='title') title = title_elem.get('title') if title_elem else title_elem.text.strip() link = urljoin(self.base_url, title_elem.get('href')) if title_elem else None up_elem = item.find('span', class_='up-name') up_name = up_elem.text.strip() if up_elem else '未知' cover_elem = item.find('img', class_='cover') cover_url = cover_elem.get('src') if cover_elem else None # 视频ID可以从链接中提取 video_id = None if link: match = re.search(r'/video/([a-zA-Z0-9]+)', link) video_id = match.group(1) if match else None if title and link and video_id: videos.append({ 'video_id': video_id, 'title': title, 'url': link, 'up_name': up_name, 'cover_url': cover_url, 'downloaded': False }) except Exception as e: logger.warning(f"解析单个视频卡片时出错: {e}") continue logger.info(f"本页解析到 {len(videos)} 个视频") return videos

3.3 获取视频真实播放地址（关键步骤）

这是最具挑战性的一步。我们需要进入每个视频的详情页，找到m3u8文件地址。通常，这个地址不会直接写在HTML里，而是由JavaScript脚本动态加载。我们可以通过监听页面网络请求来捕获它。

async def get_video_play_url(self, video_page_url): """访问视频详情页，尝试获取m3u8播放地址""" async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context() page = await context.new_page() m3u8_url = None # 监听所有网络响应，寻找包含.m3u8的请求 def handle_response(response): nonlocal m3u8_url if '.m3u8' in response.url: m3u8_url = response.url logger.info(f"捕获到m3u8地址: {m3u8_url}") page.on('response', handle_response) try: await page.goto(video_page_url, wait_until='networkidle') # 可能需要点击播放按钮触发视频加载 # await page.click('.play-button') # 根据实际页面调整 await page.wait_for_timeout(5000) # 等待足够时间让视频请求发出 except Exception as e: logger.error(f"访问视频页 {video_page_url} 失败: {e}") finally: await browser.close() return m3u8_url

实操心得：

1. 请求过滤：在开发者工具的Network面板中，使用m3u8或ts关键字过滤请求，能快速定位目标。
2. 多方案备用：有些网站的视频地址可能藏在页面某个<script>标签的变量里，需要用正则表达式去匹配。例如，搜索"play_url":"(.*?\.m3u8)"。所以，一个健壮的get_video_play_url函数可能需要结合网络监听和页面脚本解析两种方式。
3. 清晰度选择：m3u8文件可能有多个，对应不同清晰度（如720p.m3u8,1080p.m3u8）。你需要分析捕获到的URL规律，选择你想要的清晰度。

3.4 下载m3u8并合并视频

获取到m3u8地址后，我们下载它并解析出所有ts分片链接。

async def download_m3u8_video(self, m3u8_url, video_id, title): """下载m3u8流媒体视频并合并""" import aiofiles import os from pathlib import Path # 创建临时目录存储ts文件 temp_dir = Path(f"./temp_{video_id}") temp_dir.mkdir(exist_ok=True) output_path = Path(f"./videos/{title}_{video_id}.mp4") output_path.parent.mkdir(parents=True, exist_ok=True) async with aiohttp.ClientSession() as session: try: # 1. 下载m3u8文件 async with session.get(m3u8_url) as resp: m3u8_content = await resp.text() # 2. 解析出所有ts文件链接 ts_urls = [] base_url = m3u8_url[:m3u8_url.rfind('/')+1] # 获取m3u8文件的基础路径 for line in m3u8_content.split('\n'): line = line.strip() if line and not line.startswith('#'): if line.startswith('http'): ts_urls.append(line) else: ts_urls.append(urljoin(base_url, line)) logger.info(f"视频 {title} 共有 {len(ts_urls)} 个ts分片") # 3. 异步下载所有ts文件 semaphore = asyncio.Semaphore(10) # 控制并发数，避免被封 async def download_one_ts(index, ts_url): async with semaphore: ts_filename = temp_dir / f"segment_{index:04d}.ts" try: async with session.get(ts_url) as resp: if resp.status == 200: content = await resp.read() async with aiofiles.open(ts_filename, 'wb') as f: await f.write(content) logger.debug(f"已下载分片 {index+1}/{len(ts_urls)}") else: logger.error(f"下载分片失败: {ts_url}, 状态码: {resp.status}") except Exception as e: logger.error(f"下载分片 {ts_url} 时出错: {e}") tasks = [download_one_ts(i, url) for i, url in enumerate(ts_urls)] await asyncio.gather(*tasks, return_exceptions=True) # 4. 合并ts文件为mp4 (使用ffmpeg) ts_file_list = sorted(temp_dir.glob('*.ts')) if ts_file_list: list_file = temp_dir / "file_list.txt" with open(list_file, 'w', encoding='utf-8') as f: for ts in ts_file_list: f.write(f"file '{ts.absolute()}'\n") import subprocess # 调用ffmpeg合并 cmd = [ 'ffmpeg', '-f', 'concat', '-safe', '0', '-i', str(list_file.absolute()), '-c', 'copy', str(output_path.absolute()) ] subprocess.run(cmd, capture_output=True, check=True) logger.info(f"视频合并完成: {output_path}") else: logger.error(f"未找到任何ts文件，合并失败") except Exception as e: logger.error(f"下载或合并视频 {title} 时发生错误: {e}") finally: # 5. 清理临时ts文件 import shutil if temp_dir.exists(): shutil.rmtree(temp_dir)

重要提示：ffmpeg依赖上述代码需要系统已安装ffmpeg命令行工具。你需要先去ffmpeg官网下载并安装，并确保其路径已添加到系统环境变量中，使得在命令行中可以直接调用ffmpeg命令。

3.5 数据存储与任务调度

我们将爬取到的视频元数据存入SQLite数据库，便于管理和去重。

async def init_database(self): """初始化数据库，创建表""" self.db = await aiosqlite.connect('dance_videos.db') await self.db.execute(''' CREATE TABLE IF NOT EXISTS videos ( id INTEGER PRIMARY KEY AUTOINCREMENT, video_id TEXT UNIQUE, title TEXT, url TEXT, up_name TEXT, cover_url TEXT, m3u8_url TEXT, local_path TEXT, downloaded BOOLEAN DEFAULT 0, publish_date TEXT, create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') await self.db.commit() async def save_video_info(self, video_info): """保存或更新视频信息到数据库""" try: await self.db.execute(''' INSERT OR IGNORE INTO videos (video_id, title, url, up_name, cover_url, downloaded) VALUES (?, ?, ?, ?, ?, ?) ''', (video_info['video_id'], video_info['title'], video_info['url'], video_info['up_name'], video_info['cover_url'], video_info.get('downloaded', False))) await self.db.commit() except Exception as e: logger.error(f"保存视频信息到数据库失败: {e}") async def mark_as_downloaded(self, video_id, m3u8_url, local_path): """标记视频已下载""" try: await self.db.execute(''' UPDATE videos SET downloaded = 1, m3u8_url = ?, local_path = ? WHERE video_id = ? ''', (m3u8_url, local_path, video_id)) await self.db.commit() except Exception as e: logger.error(f"更新视频下载状态失败: {e}")

最后，编写主函数，将所有步骤串联起来，并加入简单的定时循环逻辑（实际生产环境建议用系统cron）。

async def run(self): """主运行函数""" await self.init_database() for page in range(self.start_page, self.start_page + self.max_page): logger.info(f"开始处理第 {page} 页") html = await self.fetch_page(page) if not html: break videos = self.parse_video_list(html) for video in videos: # 检查是否已下载过 cursor = await self.db.execute('SELECT downloaded FROM videos WHERE video_id = ?', (video['video_id'],)) row = await cursor.fetchone() if row and row[0]: logger.info(f"视频 {video['title']} 已下载，跳过") continue await self.save_video_info(video) logger.info(f"开始处理视频: {video['title']}") m3u8_url = await self.get_video_play_url(video['url']) if m3u8_url: output_path = f"./videos/{video['title']}_{video['video_id']}.mp4" await self.download_m3u8_video(m3u8_url, video['video_id'], video['title']) await self.mark_as_downloaded(video['video_id'], m3u8_url, output_path) else: logger.warning(f"未能获取视频 {video['title']} 的播放地址") await asyncio.sleep(2) # 处理完一个视频后休眠，避免请求过快 await asyncio.sleep(5) # 处理完一页后休眠更长时间 await self.db.close() async def main(): # 示例基础URL，请替换为实际目标地址 base_url = "https://example.com/channel/dance?page={page}" spider = DanceVideoSpider(base_url, start_page=1, max_page=3) # 先爬3页测试 await spider.run() if __name__ == '__main__': asyncio.run(main())

4. 常见问题、反爬策略与优化技巧

在实际操作中，你几乎一定会遇到各种问题。下面是我在类似项目中踩过的坑和总结的经验。

4.1 常见问题排查表

4.2 应对反爬策略的进阶技巧

1. User-Agent与请求头伪装：Playwright启动浏览器时已经带有真实的UA，但通过aiohttp直接下载ts文件时，需要手动设置UA等请求头，模拟浏览器行为。

   headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) ...', 'Referer': video_page_url, # 关键！很多服务器会校验Referer } async with session.get(ts_url, headers=headers) as resp: ...

2. 使用代理IP：当单个IP被限制后，代理是必须的。可以将代理列表存入文件或数据库，在创建aiohttp.ClientSession或Playwright浏览器上下文时配置。

   # aiohttp 使用代理 connector = aiohttp.TCPConnector(limit=100) proxy = "http://your_proxy:port" async with aiohttp.ClientSession(connector=connector, headers=headers) as session: async with session.get(url, proxy=proxy) as resp: ... # Playwright 使用代理 browser = await p.chromium.launch(proxy={"server": "http://your_proxy:port"})

3. 模拟人类行为：在关键操作间增加随机延迟和鼠标移动。Playwright可以录制脚本，生成包含点击、滚动等操作的代码，让行为更像真人。

   await page.wait_for_timeout(random.randint(1000, 3000)) # 随机等待1-3秒 # 模拟鼠标移动到某个元素上 await page.hover('.some-element')

4. 处理验证码：如果遇到验证码，项目复杂度会急剧上升。可以考虑：a) 手动处理（不适用于全自动）；b) 使用付费的打码平台API；c) 尝试用机器学习库识别简单验证码（成功率有限）。

4.3 项目优化与扩展方向

1. 增量爬取与去重：我们的数据库设计了video_id唯一索引和downloaded字段，就是为了实现增量爬取。每天运行脚本时，只处理downloaded=0的记录。还可以根据publish_date字段，只爬取特定日期之后的视频。

2. 分布式爬虫：如果视频量巨大，可以考虑使用Scrapy框架结合Scrapy-Redis实现分布式爬取，提升效率。但Scrapy与Playwright的集成需要一些额外配置。

3. 元数据丰富与分类：除了下载视频，还可以进一步解析视频的标签、分类、弹幕、评论等，存入数据库。利用这些数据可以做简单的分析，比如最受欢迎的UP主、热门标签等。

4. 图形化界面与通知：使用PyQt或Tkinter为爬虫做一个简单的图形界面，方便配置和启动。或者，在爬虫完成下载后，通过邮件、Server酱等工具发送一条通知给你。

5. 容器化部署：使用Docker将整个爬虫环境（包括Python、Chromium、ffmpeg）打包成一个镜像，可以方便地在任何支持Docker的服务器上部署和运行。

最后一点个人体会：这类项目最大的价值不在于最终下载了多少视频，而在于解决问题的过程中，你对网络协议、浏览器自动化、异步编程、数据解析和反爬对抗的理解会深入很多。每一个报错都是一个学习的机会。务必保持耐心，从简单的目标开始，逐步增加复杂度，并始终将伦理和法律约束放在首位。