toor
eaacacc8e0
- hls_dl.py: 通用 HLS 下载器(m3u8 解析/AES-128 解密/ffmpeg 合并) - reqable_mcp.py: Reqable MCP 客户端(stdio, --port 9002) - huangguo_video_dl.user.js: 油猴全集下载脚本 v1.1.2 * 详情页一键下载全部集数,解密转封装后打包 ZIP 存 ~/Downloads * 手写 STORE ZIP(逐集 Blob 卸堆外),修复 JSZip 大文件卡死 * @connect 覆盖 rlkfus.cn / cloudfront.net CDN - README.md: 用法与踩坑记录
246 lines
9.0 KiB
Python
246 lines
9.0 KiB
Python
#!/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()
|