Login
OneManager
Home
/
mpd2hls
/
var
/
scripts
File
OriginalPic
Thumbnails
CopyAllDownloadUrl
EditTime
Size
_socks5.py
2026-04-24 22:27:15
5.37 KB
README.md
2026-04-24 22:27:15
13.91 KB
# mpd-hls Python 脚本开发指南 mpd-hls 的 **脚本管线(Scripting Pipeline)** 允许你在解析 MPD、获取 License Key 这两个关键环节插入任意 Python 代码,把 mpd-hls 从"只会处理 KID:KEY 的 DASH→HLS 转封装器"变成一个可以对接**任意鉴权、任意 DRM、任 意登录流程**的通用直播中继。 --- ## 1. 设计总览 每个频道(Channel)在 `channels.json` 里有一个可选字段: ```json "scripts": [ { "stage": "source", "path": "var/scripts/nowtv/source.py" }, { "stage": "key", "path": "var/scripts/nowtv/key.py", "timeout_secs": 20 } ] ``` mpd-hls 启动/刷新一个频道时,按顺序执行: ``` ┌───────────────────────────────────────────────────────────────────┐ │ 1. SOURCE 阶段 │ │ 按声明顺序依次运行所有 stage="source" 的脚本 │ │ 输入: { channel_id, source_url=
, ... } │ │ 目的: 把 source_url 改写为真正可抓取的 MPD URL;把鉴权 │ │ 上下文(token / origin / referer …)塞到 extra │ ├───────────────────────────────────────────────────────────────────┤ │ 2. 系统解析 MPD,提取 PSSH / default KIDs │ ├───────────────────────────────────────────────────────────────────┤ │ 3. KEY 阶段 │ │ 按顺序依次运行所有 stage="key" 的脚本 │ │ 输入: 继承 SOURCE 阶段的 ctx,并追加 pssh / default_kids │ │ 目的: 访问 CDM + license server,把 license_key 写回到 ctx │ ├───────────────────────────────────────────────────────────────────┤ │ 4. 系统拿着 license_key 解密分片 → 封装 fMP4 → 发布 HLS │ └───────────────────────────────────────────────────────────────────┘ ``` 所有阶段都是**可选的**。如果你的流是公开的 ClearKey 流,可以一个脚本 都不挂;如果只是 source 需要登录拿 MPD,就只挂 source。 --- ## 2. 通信协议:ScriptContext (JSON over stdin/stdout) 脚本是一个**独立的子进程**,mpd-hls 通过 `python3
` 启 动它,把一个 JSON 对象写到 stdin,然后从 stdout 读取改写后的 JSON。 stderr 的每一行都会被 mpd-hls 转成 WARN 级别的日志 —— 请尽情 `print()` 调试信息到 stderr。 ### 输入 / 输出的字段 | 字段 | 类型 | 方向 | 说明 | |---|---|---|---| | `channel_id` | str | 输入 | 频道 UUID(稳定不变)。 | | `channel_name` | str | 输入 | 频道显示名,只读参考。 | | `source_url` | str | 输入&输出 | 原始 `source_url`;source 阶段应改写为真正 MPD URL。 | | `license_key` | str | 输入&输出 | key 阶段应写成 `"
:
"`(允许多个,用 `,` 分隔)。 | | `pssh` | `list[str]` | 输入(key 阶段) | 来自 MPD 的 `
`,base64。空列表说明 MPD 没带 —— 你可以下载 init segment 自己抠(见 `nowtv/key.py`)。 | | `default_kids` | `list[str]` | 输入(key 阶段) | 来自 MPD 的 `default_KID`,hex。 | | `cdm_api_url` | str | 输入 | 本进程内嵌的 Widevine CDM 的基地址,形如 `http://127.0.0.1:9100/internal/cdm/
`。只在本机可达,token 每次启动都会变。 | | `extra` | object | 输入&输出 | 任意 JSON。**这是在各脚本之间传递状态的唯一通道**。 | ### 脚本骨架 最小骨架(直通,不做任何事): ```python #!/usr/bin/env python3 import json, sys ctx = json.load(sys.stdin) # TODO: 做你的事 print(f"channel={ctx['channel_name']}", file=sys.stderr) json.dump(ctx, sys.stdout) ``` ### 几条必须遵守的规矩 1. **必须把完整 ctx 写回 stdout**。只打印你改的字段会把其他字段丢掉, 下游脚本会拿不到 `extra` 里的东西。 2. **退出码非 0 = 失败**。mpd-hls 会把这次刷新算作失败,下个周期重 试。`sys.exit(1)` 或抛异常都会被视为失败。 3. **默认超时 30 秒**(`MPD_HLS_SCRIPT_TIMEOUT_SECS`)。超时进程会被 kill。需要更长的,在频道配置里用 `"timeout_secs": 60` 覆盖。 4. **stderr 全部进日志**。别 `print()` 敏感信息到 stderr。 5. **脚本每个刷新周期都会跑一次**。进程间不共享内存;如果要缓存 token,用文件 / 环境变量 / 外部 KV。 --- ## 3. 内部 CDM API(给 key 脚本用) mpd-hls 自带一个 Widevine CDM(需要在 `MPD_HLS_CDM_DEVICE` 指向的目录 下放 `device_client_id_blob` + `device_private_key`)。它以一个临时 HTTP 端点暴露给脚本: ``` POST {cdm_api_url}/challenge body: {"pssh": "
"} resp: {"session_id": "
", "challenge_b64": "
"} POST {cdm_api_url}/keys body: {"session_id": "
", "license_b64": "
"} resp: {"keys": [{"kid": "
", "key": "
"}, ...]} ``` 典型的 key 脚本流程: ``` pssh ──► /challenge ──► (session_id, challenge_b64) │ ▼ POST 到 license server,拿回 license 二进制 │ ▼ (session_id, license_b64) ──► /keys ──► 得到 kid:key ``` - `cdm_api_url` 绑在 `127.0.0.1`,并带一条 64 字节随机 token 的路径 前缀 —— 外部攻击者即使能访问 9100 端口也用不了。 - CDM session 的 TTL 是 5 分钟;用完会自动清理。 --- ## 4. 完整示例:从零写一个对接自研 DRM 的脚本 假设有个流媒体服务 `foo.tv`,channel URL 是 `foo://42`,登录 token 放在 `FOO_TOKEN` 环境变量里,它的 license server 需要一个 HMAC 签名 头。你需要两个脚本: ### `scripts/foo/source.py` ```python #!/usr/bin/env python3 import json, os, sys, urllib.request sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import _socks5 _socks5.install(os.environ.get("SOCKS5_PROXY", "")) ctx = json.load(sys.stdin) token = os.environ["FOO_TOKEN"] channel_no = ctx["source_url"].removeprefix("foo://") api = f"https://api.foo.tv/v1/stream/{channel_no}" req = urllib.request.Request(api, headers={"Authorization": f"Bearer {token}"}) data = json.loads(urllib.request.urlopen(req, timeout=10).read()) ctx["source_url"] = data["mpd_url"] ctx["extra"] = { "foo_token": token, "foo_asset_id": data["asset_id"], "license_server": data["license_url"], } json.dump(ctx, sys.stdout) ``` ### `scripts/foo/key.py` ```python #!/usr/bin/env python3 import base64, hashlib, hmac, json, os, sys, time, urllib.request sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import _socks5 _socks5.install(os.environ.get("SOCKS5_PROXY", "")) ctx = json.load(sys.stdin) cdm_url = ctx["cdm_api_url"] extra = ctx.get("extra", {}) token = extra["foo_token"] asset_id = extra["foo_asset_id"] lic_url = extra["license_server"] if not ctx.get("pssh"): print("MPD has no pssh", file=sys.stderr); sys.exit(1) # 1. 向本地 CDM 要 challenge r = urllib.request.urlopen(urllib.request.Request( f"{cdm_url}/challenge", data=json.dumps({"pssh": ctx["pssh"][0]}).encode(), headers={"Content-Type": "application/json"}, ), timeout=15) r = json.loads(r.read()) session_id, challenge_b64 = r["session_id"], r["challenge_b64"] # 2. 把 challenge 提交给 foo.tv,带上自定义 HMAC ts = str(int(time.time())) sig = hmac.new(token.encode(), f"{asset_id}|{ts}".encode(), hashlib.sha256).hexdigest() lic = urllib.request.urlopen(urllib.request.Request( lic_url, data=base64.b64decode(challenge_b64), headers={ "Content-Type": "application/octet-stream", "X-Asset-Id": asset_id, "X-Signature": sig, "X-Signature-Ts": ts, }, ), timeout=15).read() # 3. 交给 CDM 解出 kid:key r = urllib.request.urlopen(urllib.request.Request( f"{cdm_url}/keys", data=json.dumps({ "session_id": session_id, "license_b64": base64.b64encode(lic).decode(), }).encode(), headers={"Content-Type": "application/json"}, ), timeout=15) keys = json.loads(r.read())["keys"] ctx["license_key"] = f"{keys[0]['kid']}:{keys[0]['key']}" json.dump(ctx, sys.stdout) ``` --- ## 5. 把脚本挂到频道上 ### 方式 A:Web UI 1. 在频道列表点击该频道的 **[脚本]** 按钮。 2. 对话框里点 **"+ 添加脚本"**: - `stage` 选 `source` 或 `key` - `path` 填脚本的**绝对路径**,或相对于 mpd-hls 启动目录的路径 - `timeout_secs`(可选)覆盖默认 30s 3. 保存后立即生效,无需重启。 ### 方式 B:直接编辑 `var/channels.json` ```json { "channel_id": "...", "name": "Foo 42", "source_url": "foo://42", "license_key": "", "scripts": [ { "stage": "source", "path": "scripts/foo/source.py" }, { "stage": "key", "path": "scripts/foo/key.py" } ] } ``` > **注意**:挂了 `stage:"key"` 脚本时,`license_key` 允许留空;此时 > mpd-hls 知道"key 会由脚本产出"。如果留 KID:KEY,脚本产出的会**覆 > 盖**它。 ### 方式 C:API ```bash curl -u admin:xxx -X POST http://127.0.0.1:9100/api/channels \ -d '{"name":"Foo 42","source_url":"foo://42","license_key":"", "scripts":[{"stage":"source","path":"scripts/foo/source.py"}, {"stage":"key","path":"scripts/foo/key.py"}]}' ``` --- ## 6. 调试技巧 | 症状 | 排查 | |---|---| | 脚本一直超时 | 把默认 30s 改大:`MPD_HLS_SCRIPT_TIMEOUT_SECS=60`;或在单个脚本上 `timeout_secs: 60`。 | | 想看脚本输出 | 所有 stderr 会进 mpd-hls 日志(WARN 级别,`script stderr:` 前缀)。直接 tail 服务日志即可。 | | 想本地单独跑 | `cat test_ctx.json | python3 scripts/foo/source.py`,自己构造一个 ctx JSON(`cdm_api_url` 可以临时指向 `http://127.0.0.1:9100/internal/cdm/
`,token 可从运行中的服务启动日志里抓)。 | | CDM 401/403 | `cdm_api_url` 必须用脚本收到的那个,token 每次进程启动都会重新生成。别硬编码。 | | 改了 Python 解释器 | `MPD_HLS_SCRIPT_PYTHON=/opt/py311/bin/python3`。脚本的 shebang 会被忽略。 | | 代理 | `_socks5.install(url)` 是 stdlib 实现,monkey-patch 掉 `http.client`。对 `127.0.0.1` 自动旁路,CDM 调用不走代理。 | --- ## 7. 延伸玩法 🛠️ 这套 "source + key + extra" 模型其实比 DRM 大得多。几个可以直接动手 玩的方向: ### (a) 把任意奇怪的 URL 协议映射成 MPD 只要写 source 脚本,`source_url` 的前缀可以是你发明的任何东西: - `clock://
` → 动态生成一个"当前时间"测试流 - `vod://
` → 从 VOD API 拿 MPD,直接当"7×24 频道"重放 ### (b) 登录态自愈 mpd-hls 在刷新失败时会重跑 source/key 管线。利用这点: - source 脚本维护一个 token 文件;发现 upstream 返回 401 时删掉文件 并重新登录 —— 下次刷新就自动拿到新 token,频道无感恢复。 - 把 `last_login_ts` 塞进 `extra`,key 脚本用它做 rate-limit 决策。 ### (c) 基于时间 / 观众 / 地理的多源切换 - source 脚本根据当前小时数挑不同的上游(白天看 BBC One,晚上看赛事 专用 CDN)。 - 把 `channel_id` 映射到几条备份 MPD,依次探活选能用的那条 —— 穷人版 多源 failover。 - 读 `extra.preferred_region`(由运维手动编辑 `channels.json` 注入) 在 source 阶段选 CDN 节点。 ### (d) 替代 DRM 系统 CDM API 处理的是 Widevine,但 `license_key` 最终只认 `kid:key`。所 以你可以: - 自己实现一个纯 Python 的 PlayReady/FairPlay client,在 key 脚本里 跑完握手,产出 `kid:key` 写回去。 - 做 **ClearKey 的协议转换器**:上游给你一个魔改 JSON,你翻译成标准 `kid:key`。 - 做 **本地密钥库**:key 脚本去读 `keys.json`(按 KID 索引),不走网 络 —— 适合离线调试 / CI。 ### (e) 跟系统的其他特性联动 把脚本和 mpd-hls 的其他能力组合: - source 脚本把 `subtitle_profile_id` 映射到频道 → 配合 PGS + OCR, 就能做"根据节目自动选字幕样式"。 - key 脚本获取 license 成功/失败后,向你的 Prometheus pushgateway 发 一次 metric —— 天然的 DRM 可用性监控。 - source 脚本调 Telegram Bot API,流切换时推送通知给订阅者。 ### (f) 开发辅助 / 测试 - source 脚本在 `source_url` 以 `test://` 开头时,短路到本地 ffmpeg 起的一条循环视频 —— 零依赖的集成测试夹具。 - key 脚本读 `DEBUG_FORCE_FAIL=1` 环境变量就 `sys.exit(1)`,用来压 测上层重试逻辑。 --- ## 8. 一点哲学 "在 MPD → HLS 之间暴露两个 Python 钩子 + 一个 JSON 上下文 + 一个本 地 CDM" 这套设计,本质上把 mpd-hls 变成了 **"以 MPD 为中间表示的 IPTV 运行时"**。只要你能在 Python 里生成一个可被抓取的 MPD URL(和 必要时对应的 kid:key),这台服务器就能把它广播成标准的 HLS。 这意味着接入一个新源的工作量通常就是写 100~200 行 Python,不需要改 Rust、不需要重启服务。当你下次想"我能不能在这里插个什么"的时候,大 概率答案是能。
Close
2026-05-31 10:31:40 Sunday 216.73.217.89 Runningtime:0.657s Mem:496.84 KB