使用Python实现视频下载的多种方案

pip install requests

import requests

def download_video(url, filename='video.mp4'):
    """
    简单视频下载函数
    :param url: 视频资源的直接URL
    :param filename: 保存的文件名
    """
    try:
        response = requests.get(url, stream=True)
        response.raise_for_status()
        
        with open(filename, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:  # 过滤掉keep-alive新块
                    f.write(chunk)
        print(f"视频已成功下载为 {filename}")
    except requests.exceptions.RequestException as e:
        print(f"下载失败: {e}")

# 使用示例
# download_video("https://example.com/path/to/video.mp4")

import requests
from tqdm import tqdm

def download_video_with_progress(url, filename='video.mp4'):
    """
    带进度条的视频下载
    """
    try:
        response = requests.get(url, stream=True)
        response.raise_for_status()
        
        total_size = int(response.headers.get('content-length', 0))
        block_size = 1024  # 1KB
        
        with open(filename, 'wb') as f, tqdm(
            desc=filename,
            total=total_size,
            unit='iB',
            unit_scale=True,
            unit_divisor=1024,
        ) as bar:
            for data in response.iter_content(block_size):
                f.write(data)
                bar.update(len(data))
        print("\n下载完成!")
    except requests.exceptions.RequestException as e:
        print(f"下载失败: {e}")

# 使用前需要安装: pip install tqdm

import os
import requests
from concurrent.futures import ThreadPoolExecutor

def download_ts_segment(url, index, output_dir):
    """下载单个ts分片"""
    try:
        response = requests.get(url, stream=True)
        with open(f"{output_dir}/segment_{index}.ts", 'wb') as f:
            f.write(response.content)
        return True
    except:
        return False

def download_hls_video(m3u8_url, output_filename='output.mp4'):
    """
    下载HLS格式视频（简化版）
    注意：实际应用中需要更完善的解析和错误处理
    """
    try:
        # 获取m3u8内容
        m3u8_content = requests.get(m3u8_url).text
        
        # 创建临时目录存储分片
        temp_dir = 'temp_segments'
        os.makedirs(temp_dir, exist_ok=True)
        
        # 解析分片URL（简化处理，实际需要更复杂的解析）
        base_url = m3u8_url.rsplit('/', 1)[0] + '/'
        segments = [line for line in m3u8_content.split('\n') if line.endswith('.ts')]
        
        # 多线程下载分片
        with ThreadPoolExecutor(max_workers=8) as executor:
            results = list(executor.map(
                lambda i: download_ts_segment(base_url + segments[i], i, temp_dir),
                range(len(segments))
            ))
        
        if all(results):
            # 合并分片（需要ffmpeg）
            print("所有分片下载完成，正在合并...")
            os.system(f'ffmpeg -f concat -i <(for f in {temp_dir}/*.ts; do echo "file \'$f\'"; done) -c copy {output_filename}')
            print(f"视频已保存为 {output_filename}")
        else:
            print("部分分片下载失败")
            
    except Exception as e:
        print(f"处理过程中出错: {e}")

# 使用前需要安装ffmpeg并添加到PATH

import requests
from time import sleep
from random import uniform

def download_video_robust(url, filename='video.mp4', max_retries=3):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }
    
    for attempt in range(max_retries):
        try:
            sleep(uniform(0.5, 1.5))  # 随机延迟避免被封
            response = requests.get(url, headers=headers, stream=True, timeout=10)
            response.raise_for_status()
            
            with open(filename, 'wb') as f:
                for chunk in response.iter_content(chunk_size=8192):
                    if chunk:
                        f.write(chunk)
            print(f"成功下载: {filename}")
            return True
        except requests.exceptions.RequestException as e:
            print(f"尝试 {attempt + 1} 失败: {e}")
            if attempt == max_retries - 1:
                print("所有尝试均失败")
                return False

# 使用示例
# download_video_robust("https://example.com/video.mp4")