批量下载m3u8的ts文件

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
合并 MPEG-TS 分片文件为一个完整的 .ts 文件。

用法:
    python merge_ts.py                      # 合并当前目录 index*.ts -> merged.ts
    python merge_ts.py -o out.ts            # 指定输出文件名
    python merge_ts.py -d D:\\hls -o D:\\out.ts   # 指定输入目录
    python merge_ts.py a001.ts a002.ts ...  # 显式指定要合并的文件（按给定顺序）

特性:
    - 按文件名中的数字自然排序（index2 在 index10 之前）
    - 校验每个分片是否为 188 字节的整数倍（TS 包大小），不一致会警告
    - 流式拷贝，内存占用低，支持大文件
    - 校验文件头同步字节 0x47，可选剥离 ID3 头
"""
import argparse
import os
import re
import sys

TS_PACKET = 188
ID3_MAGIC = b"ID3"


def natural_key(name: str):
    """把文件名按其中出现的数字分段，实现自然排序。"""
    base = os.path.basename(name)
    parts = re.split(r"(\d+)", base)
    return [int(p) if p.isdigit() else p.lower() for p in parts]


def find_parts(directory: str):
    files = [
        os.path.join(directory, f)
        for f in os.listdir(directory)
        if f.lower().endswith(".ts") and os.path.isfile(os.path.join(directory, f))
    ]
    # 排除输出文件本身
    files = [f for f in files if not os.path.basename(f).startswith("merged")]
    return sorted(files, key=natural_key)


def copy_ts(src: str, dst, strip_id3: bool, stats: dict):
    with open(src, "rb") as f:
        head = f.read(3)
        if head == ID3_MAGIC and strip_id3:
            # 跳过 ID3v2 头: 10 字节头 + syncsafe 长度
            f.seek(0)
            hdr = f.read(10)
            if len(hdr) < 10:
                return
            size = (
                (hdr[6] & 0x7F) << 21
                | (hdr[7] & 0x7F) << 14
                | (hdr[8] & 0x7F) << 7
                | (hdr[9] & 0x7F)
            )
            f.seek(10 + size)
            stats["id3_stripped"] += 1
        else:
            f.seek(0)

        first = f.read(1)
        if first and first[0] != 0x47:
            stats["bad_sync"].append(os.path.basename(src))
        f.seek(0 if not (head == ID3_MAGIC and strip_id3) else 10 + size)

        # 流式拷贝
        while True:
            chunk = f.read(1024 * 1024)
            if not chunk:
                break
            dst.write(chunk)


def main():
    ap = argparse.ArgumentParser(description="合并 MPEG-TS 分片文件")
    ap.add_argument("inputs", nargs="*", help="显式指定分片文件（按顺序）。留空则自动查找目录中的 *.ts")
    ap.add_argument("-d", "--dir", default=".", help="输入目录，默认当前目录")
    ap.add_argument("-o", "--output", default="merged.ts", help="输出文件名，默认 merged.ts")
    ap.add_argument("-s", "--sorted", action="store_true",
                    help="使用文件名自然排序（默认即开启，此参数保留兼容）")
    ap.add_argument("--keep-id3", action="store_true",
                    help="保留分片开头的 ID3 标签（默认会剥离，避免污染 TS 流）")
    args = ap.parse_args()

    if args.inputs:
        parts = args.inputs
        for p in parts:
            if not os.path.isfile(p):
                sys.exit(f"错误: 找不到文件 {p}")
    else:
        parts = find_parts(args.dir)
        if not parts:
            sys.exit(f"错误: 在 {os.path.abspath(args.dir)} 中未找到 .ts 文件")

    out_path = args.output
    if os.path.isdir(out_path):
        out_path = os.path.join(out_path, "merged.ts")
    if os.path.abspath(out_path) in [os.path.abspath(p) for p in parts]:
        sys.exit("错误: 输出文件与输入文件相同，请用 -o 指定其它文件名")

    strip_id3 = not args.keep_id3
    stats = {"id3_stripped": 0, "bad_sync": []}

    print(f"共 {len(parts)} 个分片 -> {out_path}")
    print("-" * 60)

    total_bytes = 0
    misaligned = []
    with open(out_path, "wb") as out:
        for i, p in enumerate(parts, 1):
            size = os.path.getsize(p)
            if size % TS_PACKET != 0:
                misaligned.append((os.path.basename(p), size))
            copy_ts(p, out, strip_id3, stats)
            total_bytes += size
            print(f"[{i:>3}/{len(parts)}] {os.path.basename(p):<20} {size:>12,} 字节")

    print("-" * 60)
    print(f"完成: {total_bytes:,} 字节 ({total_bytes / 1024 / 1024:.1f} MiB) -> {out_path}")

    if stats["id3_stripped"]:
        print(f"已剥离 {stats['id3_stripped']} 个分片开头的 ID3 标签")
    if stats["bad_sync"]:
        print(f"警告: 以下分片未以 0x47 同步字节开头（可能已损坏）: {', '.join(stats['bad_sync'])}")
    if misaligned:
        print(f"警告: 以下分片长度不是 {TS_PACKET} 的整数倍，可能导致播放器解码错位:")
        for name, size in misaligned:
            print(f"        {name} ({size} 字节, 余 {size % TS_PACKET})")


if __name__ == "__main__":
    main()