快捷導(dǎo)航

使用Python實(shí)現(xiàn)視頻下載的多種方案

更新時(shí)間：2026年05月11日 08:24:42 作者：detayun

在數(shù)字時(shí)代,我們經(jīng)常需要下載視頻用于離線觀看或個(gè)人學(xué)習(xí),本文將介紹如何使用Python編寫一個(gè)簡(jiǎn)單的視頻下載工具,不涉及任何特定視頻網(wǎng)站,而是聚焦于通用技術(shù)原理和實(shí)現(xiàn)方法,需要的朋友可以參考下

基本原理

視頻下載的核心流程通常包括：

獲取視頻資源的真實(shí)URL
發(fā)起HTTP請(qǐng)求下載數(shù)據(jù)
將數(shù)據(jù)保存為本地文件

準(zhǔn)備工作

首先安裝必要的庫(kù)：

pip install requests

基礎(chǔ)實(shí)現(xiàn)方案

方案1：直接下載（適用于已知URL的情況）

import requests

def download_video(url, filename='video.mp4'):
    """
    簡(jiǎn)單視頻下載函數(shù)
    :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:  # 過(guò)濾掉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")

方案2：帶進(jìn)度條的下載

import requests
from tqdm import tqdm

def download_video_with_progress(url, filename='video.mp4'):
    """
    帶進(jìn)度條的視頻下載
    """
    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

高級(jí)技巧：處理分塊視頻

許多視頻服務(wù)采用分塊傳輸（如HLS或DASH格式），這時(shí)需要：

解析網(wǎng)頁(yè)獲取m3u8播放列表
下載所有ts分片
合并為完整視頻

import os
import requests
from concurrent.futures import ThreadPoolExecutor

def download_ts_segment(url, index, output_dir):
    """下載單個(gè)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格式視頻（簡(jiǎn)化版）
    注意：實(shí)際應(yīng)用中需要更完善的解析和錯(cuò)誤處理
    """
    try:
        # 獲取m3u8內(nèi)容
        m3u8_content = requests.get(m3u8_url).text
        
        # 創(chuàng)建臨時(shí)目錄存儲(chǔ)分片
        temp_dir = 'temp_segments'
        os.makedirs(temp_dir, exist_ok=True)
        
        # 解析分片URL（簡(jiǎn)化處理，實(shí)際需要更復(fù)雜的解析）
        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"處理過(guò)程中出錯(cuò): {e}")

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

注意事項(xiàng)

合法性：確保你有權(quán)下載和使用目標(biāo)視頻內(nèi)容
速率限制：添加適當(dāng)?shù)难舆t避免被封禁
錯(cuò)誤處理：完善網(wǎng)絡(luò)請(qǐng)求的錯(cuò)誤處理機(jī)制
資源清理：下載失敗時(shí)清理臨時(shí)文件
用戶代理：設(shè)置合理的User-Agent頭

完整示例（帶用戶代理和重試機(jī)制）

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))  # 隨機(jī)延遲避免被封
            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")