#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ HLS 拉流下载器(纯标准库,零依赖) 用法: python3 hls_dl.py "https://huangguo.me/h5/m3u8/v3/bktappdata/sp/om/xxx.m3u8" -o out.mp4 python3 hls_dl.py "https://xxx.m3u8" --referer https://huangguo.me/ --proxy http://127.0.0.1:7890 python3 hls_dl.py "https://xxx.m3u8" --ts-url "https://dyum53hre43nj.cloudfront.net/.../xxx.ts?md=..." # 单个分片直接下 说明: - 支持 master 播放列表(自动选最高码率)、多级嵌套 m3u8、相对路径拼接 - AES-128 加密的流会自动尝试解密(需要 pycryptodome,没有则提示改用 ffmpeg) - 合并优先调用 ffmpeg(需已安装);没有 ffmpeg 就二进制拼接 TS """ import argparse import os import re import shutil import subprocess import sys import tempfile import urllib.request import urllib.parse from concurrent.futures import ThreadPoolExecutor, as_completed UA = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36") def build_opener(proxy=None): handlers = [] if proxy: handlers.append(urllib.request.ProxyHandler({"http": proxy, "https": proxy})) else: handlers.append(urllib.request.ProxyHandler({})) # 不走系统代理 return urllib.request.build_opener(*handlers) def http_get(url, referer=None, proxy=None, timeout=30): req = urllib.request.Request(url, headers={ "User-Agent": UA, "Referer": referer or "", "Accept": "*/*", }) opener = build_opener(proxy) with opener.open(req, timeout=timeout) as resp: return resp.read() def parse_m3u8(text, base_url): """返回 (type, items) type='master' -> items=[(bandwidth, uri)] type='media' -> items=[{'uri':..., 'key':..., 'iv':...}, ...] """ lines = [l.strip() for l in text.splitlines() if l.strip()] if not lines or lines[0] != "#EXTM3U": raise ValueError("不是合法的 m3u8") if any(l.startswith("#EXT-X-STREAM-INF") for l in lines): variants = [] for i, l in enumerate(lines): if l.startswith("#EXT-X-STREAM-INF"): m = re.search(r"BANDWIDTH=(\d+)", l) bw = int(m.group(1)) if m else 0 # 下一个非注释行就是 uri for j in range(i + 1, len(lines)): if not lines[j].startswith("#"): variants.append((bw, urllib.parse.urljoin(base_url, lines[j]))) break return "master", variants segments, cur_key, cur_iv = [], None, None seq = 0 # EXT-X-MEDIA-SEQUENCE,无 IV 时用它做 AES IV for l in lines: if l.startswith("#EXT-X-MEDIA-SEQUENCE"): seq = int(l.split(":", 1)[1]) elif l.startswith("#EXT-X-KEY"): m = re.search(r'URI="([^"]+)"', l) iv = re.search(r"IV=(0x[0-9a-fA-F]+)", l) cur_key = urllib.parse.urljoin(base_url, m.group(1)) if m else None cur_iv = iv.group(1) if iv else None elif l.startswith("#EXT-X-MAP"): m = re.search(r'URI="([^"]+)"', l) if m: # fMP4 的 init 段 segments.append({"uri": urllib.parse.urljoin(base_url, m.group(1)), "key": cur_key, "iv": cur_iv, "seq": seq, "init": True}) elif not l.startswith("#"): segments.append({"uri": urllib.parse.urljoin(base_url, l), "key": cur_key, "iv": cur_iv, "seq": seq, "init": False}) seq += 1 return "media", segments def aes_decrypt(data, key, iv_hex=None, seq=0): try: from Cryptodome.Cipher import AES # pycryptodome except ImportError: try: from Crypto.Cipher import AES except ImportError: return None if iv_hex: iv = bytes.fromhex(iv_hex.replace("0x", "")) else: # HLS 规范:未给 IV 时,IV = 分片的 media sequence 号的 16 字节大端 iv = seq.to_bytes(16, "big") cipher = AES.new(key, AES.MODE_CBC, iv) return cipher.decrypt(data) _KEY_CACHE = {} def fetch_key(key_url, referer, proxy): if key_url not in _KEY_CACHE: _KEY_CACHE[key_url] = http_get(key_url, referer, proxy) return _KEY_CACHE[key_url] def download_segment(seg, referer, proxy, retries=3): for attempt in range(retries): try: data = http_get(seg["uri"], referer, proxy) if seg.get("key"): key = fetch_key(seg["key"], referer, proxy) dec = aes_decrypt(data, key, seg.get("iv"), seg.get("seq", 0)) if dec is None: raise RuntimeError("AES 解密需要 pycryptodome,请 pip install pycryptodome") data = dec return data except Exception as e: if attempt == retries - 1: raise RuntimeError(f"分片失败 {seg['uri']}: {e}") from e return b"" def merge(files, out_path): ffmpeg = shutil.which("ffmpeg") if ffmpeg: # 用 concat demuxer,免去 -c copy 拼接时的时间戳问题 list_file = out_path + ".list.txt" with open(list_file, "w") as f: for p in files: f.write(f"file '{os.path.abspath(p)}'\n") cmd = [ffmpeg, "-y", "-f", "concat", "-safe", "0", "-i", list_file, "-c", "copy", out_path] r = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) os.remove(list_file) if r.returncode == 0: return "ffmpeg" print(f"[警告] ffmpeg 合并失败,回退二进制拼接:\n{r.stderr.decode()[-400:]}", file=sys.stderr) # 二进制拼接(TS 天然支持) with open(out_path, "wb") as out: for p in files: with open(p, "rb") as f: shutil.copyfileobj(f, out) return "binary" def main(): ap = argparse.ArgumentParser(description="HLS 拉流下载器") ap.add_argument("url", help="m3u8 地址(或用 --ts-url 下单个分片)") ap.add_argument("-o", "--output", default="output.mp4", help="输出文件,默认 output.mp4") ap.add_argument("--referer", default="", help="防盗链 Referer,如 https://huangguo.me/") ap.add_argument("--proxy", default="none", help="HTTP 代理,默认直连;需要时传如 http://127.0.0.1:7890") ap.add_argument("--workers", type=int, default=8, help="并发下载数,默认 8") ap.add_argument("--ts-url", default="", help="直接下载单个 .ts 分片") args = ap.parse_args() proxy = None if args.proxy.lower() == "none" else args.proxy # ---- 单分片模式 ---- if args.ts_url: data = http_get(args.ts_url, args.referer, proxy) with open(args.output, "wb") as f: f.write(data) print(f"[OK] 分片已保存: {args.output} ({len(data)} 字节)") return # ---- 拉取 m3u8,层层解析 ---- url, text, depth = args.url, None, 0 while True: depth += 1 if depth > 5: sys.exit("嵌套层级过深,疑似异常") print(f"[*] 拉取播放列表: {url}") raw = http_get(url, args.referer, proxy) try: text = raw.decode("utf-8") except UnicodeDecodeError: text = raw.decode("latin-1") kind, items = parse_m3u8(text, url) if kind == "master": items.sort(key=lambda v: v[0], reverse=True) print(f"[+] {len(items)} 个码率,选最高 {items[0][0]} bps") url = items[0][1] continue break segments = items if not segments: sys.exit("播放列表里没有分片") print(f"[+] 共 {len(segments)} 个分片") if any(s.get("key") for s in segments): print("[!] 流是 AES-128 加密的,将尝试解密(需 pycryptodome,否则请改用 ffmpeg)") # ---- 并发下载 ---- tmpdir = tempfile.mkdtemp(prefix="hls_") files, done = [None] * len(segments), 0 def task(i_seg): i, seg = i_seg suffix = ".mp4" if seg.get("init") else ".ts" path = os.path.join(tmpdir, f"{i:05d}{suffix}") if not os.path.exists(path): with open(path, "wb") as f: f.write(download_segment(seg, args.referer, proxy)) return i, path with ThreadPoolExecutor(max_workers=args.workers) as ex: futures = [ex.submit(task, (i, s)) for i, s in enumerate(segments)] for fut in as_completed(futures): i, path = fut.result() files[i] = path done += 1 print(f"\r[*] 进度 {done}/{len(segments)}", end="", flush=True) print() # ---- 合并 ---- mode = merge(files, args.output) size = os.path.getsize(args.output) print(f"[OK] 合并方式: {'ffmpeg (无损)' if mode == 'ffmpeg' else '二进制拼接'}") print(f"[OK] 已保存: {os.path.abspath(args.output)} ({size / 1024 / 1024:.2f} MB)") # 清理 for p in files: try: os.remove(p) except OSError: pass os.rmdir(tmpdir) if __name__ == "__main__": main()