背景分析

AcFun作为国内知名视频平台,其视频存储机制有几个特点:

  1. 分段存储:视频通常被切割成多个ts文件,通过m3u8索引文件组织
  2. 动态加密:部分视频流采用AES-128加密,需要获取密钥才能解密
  3. 反爬措施:包括请求头验证、IP频率限制等

视频分段示意图

技术方案对比

| 方法 | 优点 | 缺点 | |--------------|-----------------------|-----------------------| | 直接抓包 | 无需官方API | 易被反爬,稳定性差 | | 官方API调用 | 稳定可靠 | 需要逆向分析接口 | | 浏览器自动化 | 绕过部分反爬 | 性能低下,资源占用高 |

核心实现

1. 环境准备

需要安装以下依赖:

pip install requests pycryptodome

2. 获取视频元数据

import requests

def get_video_info(aid):
    """通过视频AV号获取基本信息"""
    api_url = f"https://www.acfun.cn/rest/pc-direct/video/videoInfo?videoId={aid}"
    headers = {
        "User-Agent": "Mozilla/5.0"
    }

    try:
        resp = requests.get(api_url, headers=headers)
        return resp.json()
    except Exception as e:
        print(f"获取视频信息失败: {e}")
        return None

3. 解析m3u8文件

m3u8文件结构

def parse_m3u8(m3u8_url):
    """解析m3u8文件获取ts片段列表"""
    resp = requests.get(m3u8_url)
    ts_list = []

    for line in resp.text.split('\n'):
        if line.endswith('.ts'):
            ts_list.append(line)

    return ts_list

4. 下载并合并视频

import subprocess

def download_and_merge(video_info, output_path):
    """下载并合并视频片段"""
    m3u8_url = video_info['m3u8_url']
    ts_list = parse_m3u8(m3u8_url)

    # 下载所有ts片段
    for i, ts_url in enumerate(ts_list):
        print(f"正在下载第{i+1}/{len(ts_list)}个片段")
        ts_data = requests.get(ts_url).content
        with open(f"temp_{i}.ts", 'wb') as f:
            f.write(ts_data)

    # 使用ffmpeg合并
    cmd = f"ffmpeg -i concat:{'|'.join([f'temp_{i}.ts' for i in range(len(ts_list))])} -c copy {output_path}"
    subprocess.run(cmd, shell=True)

    # 清理临时文件
    for i in range(len(ts_list)):
        os.remove(f"temp_{i}.ts")

反爬对策

  1. 请求头伪装:包含完整的浏览器特征
  2. 请求频率控制:添加随机延迟
  3. 代理池轮换:使用多个IP地址
import time
import random

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    "Referer": "https://www.acfun.cn/"
}

def safe_request(url):
    """带防护的请求方法"""
    time.sleep(random.uniform(0.5, 1.5))
    return requests.get(url, headers=headers)

性能优化

  1. 多线程下载:使用ThreadPoolExecutor加速片段下载
  2. 本地缓存:避免重复下载相同视频
from concurrent.futures import ThreadPoolExecutor

def parallel_download(ts_list):
    """并行下载ts片段"""
    with ThreadPoolExecutor(max_workers=4) as executor:
        executor.map(download_ts, ts_list)

避坑指南

  • 404错误:检查视频是否下架或需要登录
  • 解密失败:确认密钥是否正确获取
  • 合并错误:检查ffmpeg版本和参数

完整代码示例

完整项目代码 包含异常处理和进度显示

扩展思考

  1. 如何实现断点续传功能?
  2. 怎样自动识别视频清晰度选项?
  3. 如何将下载工具封装为浏览器插件?

通过这个方案,我成功下载了上百个AcFun视频用于研究分析。整个过程最耗时的部分是处理各种反爬机制,但只要保持耐心,总能找到解决方案。希望这篇指南对你有帮助!

Logo

音视频技术社区,一个全球开发者共同探讨、分享、学习音视频技术的平台,加入我们,与全球开发者一起创造更加优秀的音视频产品!

更多推荐