m3u8url解析

Rallan

"""
M3U8视频安全下载合并工具
功能：
1. 全量下载所有视频片段到临时目录
2. 完整性校验并提示缺失片段
3. 按顺序合并存在片段（边合并边清理）
4. 自动处理异常和重试
"""

import os
import re
import requests
from urllib.parse import urljoin, urlparse
from concurrent.futures import ThreadPoolExecutor, as_completed
from tqdm import tqdm

# ████████ 工具函数 ████████
def sanitize_filename(filename):
    """清理非法文件名字符"""
    return re.sub(r'[\\/*?:"<>|]', "_", filename)

def format_index(index, total):
    """生成统一格式的索引文件名"""
    digits = len(str(total))
    return f"{index:0{digits}d}.ts"

# ████████ 阶段1：下载所有片段 ████████
class DownloadManager:
    def __init__(self, m3u8_url, temp_folder="temp_ts"):
        self.m3u8_url = m3u8_url
        self.temp_folder = temp_folder
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Referer': f'{urlparse(m3u8_url).scheme}://{urlparse(m3u8_url).netloc}/'
        }
        os.makedirs(temp_folder, exist_ok=True)

    def _get_retry_session(self):
        """创建带重试机制的Session"""
        session = requests.Session()
        retries = 5
        backoff_factor = 0.3
        status_forcelist = [500, 502, 503, 504]

        retry = requests.packages.urllib3.util.Retry(
            total=retries,
            read=retries,
            connect=retries,
            backoff_factor=backoff_factor,
            status_forcelist=status_forcelist,
        )
        adapter = requests.adapters.HTTPAdapter(max_retries=retry)
        session.mount('http://', adapter)
        session.mount('https://', adapter)
        return session

    def parse_m3u8(self):
        """解析M3U8获取片段列表"""
        try:
            response = requests.get(self.m3u8_url, headers=self.headers, timeout=10)
            response.raise_for_status()
            lines = response.text.splitlines()

            total = len([l for l in lines if l.strip() and not l.startswith('#')])
            segments = []
            for index, line in enumerate(lines):
                line = line.strip()
                if line and not line.startswith('#'):
                    url = urljoin(self.m3u8_url, line)
                    filename = format_index(index, total)
                    segments.append({
                        'url': url,
                        'filename': filename,
                        'save_path': os.path.join(self.temp_folder, filename)
                    })
            return segments
        except Exception as e:
            raise RuntimeError(f"M3U8解析失败: {str(e)}")

    def download_all(self, max_workers=8):
        """多线程下载所有片段"""
        segments = self.parse_m3u8()
        print(f"需要下载总片段数: {len(segments)}")

        with tqdm(total=len(segments), desc="下载进度") as pbar:
            with ThreadPoolExecutor(max_workers=max_workers) as executor:
                futures = []
                session = self._get_retry_session()

                for seg in segments:
                    # 跳过已存在的文件
                    if os.path.exists(seg['save_path']):
                        pbar.update(1)
                        continue

                    future = executor.submit(
                        self._download_single,
                        session=session,
                        url=seg['url'],
                        save_path=seg['save_path']
                    )
                    futures.append(future)

                for future in as_completed(futures):
                    try:
                        future.result()
                    except Exception as e:
                        print(f"\n下载错误: {str(e)}")
                    pbar.update(1)

        return len(segments)

    def _download_single(self, session, url, save_path):
        """下载单个片段"""
        try:
            response = session.get(url, headers=self.headers, timeout=20)
            response.raise_for_status()
            with open(save_path, 'wb') as f:
                f.write(response.content)
        except Exception as e:
            raise RuntimeError(f"下载失败 {os.path.basename(save_path)}: {str(e)}")

# ████████ 阶段2：完整性校验 ████████
def check_missing(total, temp_folder):
    """检查缺失片段"""
    missing = []
    for index in range(total):
        filename = format_index(index, total)
        path = os.path.join(temp_folder, filename)
        if not os.path.exists(path):
            missing.append(index)

    if missing:
        print(f"警告：缺失 {len(missing)} 个片段 (示例: {missing[:5]}...)")
        choice = input("是否继续合并？(y/n): ").lower()
        if choice != 'y':
            exit()
    return missing

# ████████ 阶段3：安全合并 ████████
class SafeMerger:
    def __init__(self, temp_folder, output_file):
        self.temp_folder = temp_folder
        self.output_file = output_file
        self.total_files = len([f for f in os.listdir(temp_folder) if f.endswith('.ts')])

    def merge(self):
        """按顺序合并并删除临时文件"""
        sorted_files = sorted(
            [f for f in os.listdir(self.temp_folder) if f.endswith('.ts')],
            key=lambda x: int(x.split('.')[0])
        )

        with open(self.output_file, 'wb') as outfile, \
             tqdm(total=self.total_files, desc="合并进度") as pbar:

            for filename in sorted_files:
                file_path = os.path.join(self.temp_folder, filename)
                try:
                    with open(file_path, 'rb') as infile:
                        outfile.write(infile.read())
                    os.remove(file_path)  # 立即删除已合并文件
                    pbar.update(1)
                except Exception as e:
                    print(f"\n合并失败 {filename}: {str(e)}")

        # 清理空目录
        try:
            os.rmdir(self.temp_folder)
        except OSError:
            pass

# ████████ 主流程 ████████
def main(m3u8_url, output_file="output.mp4", temp_folder="temp_ts", max_workers=8):
    # 阶段1：下载
    downloader = DownloadManager(m3u8_url, temp_folder)
    total = downloader.download_all(max_workers)

    # 阶段2：校验
    missing = check_missing(total, temp_folder)

    # 阶段3：合并
    merger = SafeMerger(temp_folder, output_file)
    merger.merge()

    print(f"视频合并完成！保存至: {output_file}")
    if missing:
        print(f"注意：缺失 {len(missing)} 个片段，视频可能不完整")

if __name__ == '__main__':
    # 使用示例
    main(
        m3u8_url="https://surrit.com/e6847292-3317-4ca0-ae58-0b02aa5e6bba/1080p/video.m3u8",
        output_file="final_video2.mp4",
        temp_folder="download_cache",
        max_workers=10
    )