mpd2hls/var/scripts - OneManager

File OriginalPic	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=<channel 配置的原始 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 <your_script.py>` 启
动它，把一个 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 阶段应写成 `"<kid_hex>:<key_hex>"`（允许多个，用 `,` 分隔）。 |
| `pssh` | `list[str]` | 输入（key 阶段） | 来自 MPD 的 `<cenc:pssh>`，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/<random_token>`。只在本机可达，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": "<base64>"}
  resp: {"session_id": "<hex>", "challenge_b64": "<base64>"}

POST {cdm_api_url}/keys
  body: {"session_id": "<hex>", "license_b64": "<base64>"}
  resp: {"keys": [{"kid": "<hex>", "key": "<hex>"}, ...]}
```

典型的 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>`，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://<timezone>` → 动态生成一个"当前时间"测试流
- `vod://<asset_id>` → 从 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、不需要重启服务。当你下次想"我能不能在这里插个什么"的时候，大
概率答案是能。

OneManager

Home / mpd2hls / var / scripts