用Flask+Python搞定m3u8视频下载与Cloudflare R2上传，保姆级配置避坑指南

Flask+Python实现m3u8视频下载与Cloudflare R2云存储全流程实战

当我们需要从网络获取视频资源时，m3u8格式的流媒体文件是常见的选择。本文将带你从零开始，构建一个完整的Flask应用，实现m3u8视频的自动下载、本地存储以及Cloudflare R2云存储的双重备份方案。

1. 环境准备与基础配置

在开始编码前，我们需要确保开发环境已经准备就绪。首先创建一个干净的Python虚拟环境：

python -m venv m3u8_downloader source m3u8_downloader/bin/activate # Linux/Mac m3u8_downloader\Scripts\activate # Windows

安装必要的依赖包：

pip install flask boto3 requests certifi

接下来配置Cloudflare R2的访问凭证。在Cloudflare控制面板中创建R2存储桶后，获取以下信息：

• 端点URL (Endpoint URL)
• 访问密钥ID (Access Key ID)
• 秘密访问密钥 (Secret Access Key)
• 存储桶名称 (Bucket Name)

建议将这些敏感信息存储在环境变量中，而非直接硬编码在脚本里：

import os from flask import Flask app = Flask(__name__) app.config['R2_ENDPOINT_URL'] = os.getenv('R2_ENDPOINT_URL') app.config['R2_ACCESS_KEY_ID'] = os.getenv('R2_ACCESS_KEY_ID') app.config['R2_SECRET_ACCESS_KEY'] = os.getenv('R2_SECRET_ACCESS_KEY') app.config['R2_BUCKET_NAME'] = os.getenv('R2_BUCKET_NAME')

2. 构建m3u8下载核心功能

m3u8文件本质上是一个播放列表，包含了多个.ts分片视频的地址。我们需要先下载m3u8文件，解析出所有.ts分片，然后逐个下载。

首先创建一个下载器类，封装HTTP请求逻辑：

import requests import certifi class VideoDownloader: def __init__(self): self.session = requests.Session() self.session.verify = certifi.where() def set_headers(self, headers): """设置自定义请求头""" self.session.headers.update(headers) def download(self, url, timeout=30): """下载文件内容""" try: response = self.session.get(url, timeout=timeout) response.raise_for_status() return response.content except requests.exceptions.RequestException as e: print(f"下载失败: {url}, 错误: {e}") return None

接下来实现m3u8解析器：

import re from urllib.parse import urljoin class M3U8Parser: @staticmethod def parse_m3u8(content, base_url): """解析m3u8内容，返回所有ts文件URL""" if not content: return [] text = content.decode('utf-8') lines = text.split('\n') ts_urls = [] for line in lines: line = line.strip() if line and not line.startswith('#') and line.endswith('.ts'): ts_urls.append(urljoin(base_url, line)) return ts_urls

3. 实现本地与云存储双重备份

为了确保数据安全，我们将同时保存视频文件到本地和Cloudflare R2存储。首先实现本地存储功能：

import os from pathlib import Path class LocalStorage: @staticmethod def save(content, directory, filename): """保存文件到本地""" try: Path(directory).mkdir(parents=True, exist_ok=True) filepath = os.path.join(directory, filename) with open(filepath, 'wb') as f: f.write(content) return True except Exception as e: print(f"本地保存失败: {e}") return False

然后是Cloudflare R2存储的实现。由于R2兼容S3 API，我们可以使用boto3库：

import boto3 from botocore.exceptions import ClientError class R2Storage: def __init__(self, app): self.client = boto3.client( 's3', endpoint_url=app.config['R2_ENDPOINT_URL'], aws_access_key_id=app.config['R2_ACCESS_KEY_ID'], aws_secret_access_key=app.config['R2_SECRET_ACCESS_KEY'] ) self.bucket = app.config['R2_BUCKET_NAME'] def upload(self, content, key): """上传文件到R2""" try: self.client.put_object( Bucket=self.bucket, Key=key, Body=content ) return True except ClientError as e: print(f"R2上传失败: {e}") return False

4. 构建Flask API与错误处理

现在我们将各个模块整合到Flask应用中，并添加完善的错误处理机制。

首先定义API路由：

from flask import request, jsonify import tempfile @app.route('/api/download', methods=['POST']) def download_video(): data = request.json m3u8_url = data.get('m3u8_url') base_url = data.get('base_url') # ts文件的基础URL output_dir = data.get('output_dir', 'downloads') if not m3u8_url or not base_url: return jsonify({'error': '缺少必要参数'}), 400 downloader = VideoDownloader() if 'headers' in data: downloader.set_headers(data['headers']) # 下载m3u8文件 m3u8_content = downloader.download(m3u8_url) if not m3u8_content: return jsonify({'error': 'm3u8文件下载失败'}), 500 # 解析ts文件列表 ts_urls = M3U8Parser.parse_m3u8(m3u8_content, base_url) if not ts_urls: return jsonify({'error': '未找到有效的ts文件'}), 500 # 保存m3u8文件 LocalStorage.save(m3u8_content, output_dir, 'index.m3u8') r2_storage = R2Storage(app) r2_storage.upload(m3u8_content, f'{output_dir}/index.m3u8') # 下载所有ts文件 results = [] for ts_url in ts_urls: ts_content = downloader.download(ts_url) if ts_content: filename = ts_url.split('/')[-1] # 本地保存 LocalStorage.save(ts_content, output_dir, filename) # R2上传 r2_key = f'{output_dir}/{filename}' r2_storage.upload(ts_content, r2_key) results.append({ 'url': ts_url, 'filename': filename, 'status': 'success' }) else: results.append({ 'url': ts_url, 'filename': '', 'status': 'failed' }) return jsonify({ 'message': '下载任务完成', 'results': results, 'success_count': len([r for r in results if r['status'] == 'success']) })

添加必要的错误处理中间件：

@app.errorhandler(404) def not_found(error): return jsonify({'error': '资源不存在'}), 404 @app.errorhandler(500) def internal_error(error): return jsonify({'error': '服务器内部错误'}), 500

5. 性能优化与实用技巧

基础功能实现后，我们可以进行多方面的优化来提升系统的性能和可靠性。

5.1 多线程下载

使用Python的concurrent.futures实现多线程下载：

from concurrent.futures import ThreadPoolExecutor, as_completed def download_all_ts(ts_urls, output_dir, max_workers=5): with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = [] for ts_url in ts_urls: futures.append(executor.submit( download_single_ts, ts_url, output_dir )) results = [] for future in as_completed(futures): results.append(future.result()) return results

5.2 断点续传与重试机制

实现一个带重试机制的下载函数：

def download_with_retry(url, max_retries=3, timeout=30): for attempt in range(max_retries): try: response = requests.get(url, timeout=timeout) response.raise_for_status() return response.content except Exception as e: if attempt == max_retries - 1: raise time.sleep(2 ** attempt) # 指数退避

5.3 进度显示

添加下载进度显示功能：

from tqdm import tqdm def download_with_progress(url, output_dir, filename): response = requests.get(url, stream=True) total_size = int(response.headers.get('content-length', 0)) filepath = os.path.join(output_dir, filename) with open(filepath, 'wb') as f, tqdm( desc=filename, total=total_size, unit='B', unit_scale=True, unit_divisor=1024, ) as bar: for data in response.iter_content(chunk_size=1024): f.write(data) bar.update(len(data))

6. 部署与监控

完成开发后，我们需要考虑如何部署应用并监控其运行状态。

6.1 使用Gunicorn部署

pip install gunicorn gunicorn -w 4 -b :5000 app:app

6.2 添加健康检查端点

@app.route('/health') def health_check(): return jsonify({'status': 'healthy'})

6.3 日志配置

import logging from logging.handlers import RotatingFileHandler handler = RotatingFileHandler('app.log', maxBytes=10000, backupCount=3) handler.setLevel(logging.INFO) app.logger.addHandler(handler)

7. 安全加固措施

最后，我们需要考虑应用的安全性，防止常见的安全威胁。

7.1 请求速率限制

from flask_limiter import Limiter from flask_limiter.util import get_remote_address limiter = Limiter( app, key_func=get_remote_address, default_limits=["200 per day", "50 per hour"] ) @app.route('/api/download') @limiter.limit("10 per minute") def download_video(): # ...

7.2 输入验证

def validate_url(url): """验证URL格式""" pattern = re.compile( r'^(https?://)?' # http:// or https:// r'([a-zA-Z0-9.-]+)' # domain r'(\.[a-zA-Z]{2,63})' # .com, .org etc r'(:[0-9]{1,5})?' # optional port r'(/.*)?$' # optional path ) return bool(pattern.match(url))

7.3 敏感信息保护

from flask_talisman import Talisman Talisman(app, force_https=True)