令牌会过期？给 IPTV 直播链路装上「全自动续期」：Docker 化 24 小时无感更新

rtsp://27.148.240.185/PLTV/.../xxx.smil?accountinfo=%7E%7EV2.0%7E...%7E...:20261005120930,10500002078800,27.152.x.x,...&...

strings /usr/bin/rtp2httpd | grep -i reload
# Only reload M3U/EPG, skipping configuration reload (config file path not set)
# Configuration reloaded successfully from %s

每 5 分钟一轮检查：
├─ ① 出口 IP ≠ 上次续期时的 IP   → 续期（IP 漂移，令牌已死）
├─ ② 距上次续期 > 24h            → 续期（定时兜底，防服务端会话老化）
└─ ③ 试播 CCTV-1 连续 2 次失败   → 续期（自愈：不管什么原因，重认证准没错）

/root/iptv/
├── docker-compose.yml
├── renewer/
│   ├── Dockerfile
│   └── renewer.py        # 续期核心逻辑（附录有完整版）
└── data/                 # 产物: state.json + 三份 m3u

# 备份现有配置（容器会接管这个文件）
cp /etc/rtp2httpd.conf /etc/rtp2httpd.conf.bak-$(date +%F)
mkdir -p /root/iptv/renewer /root/iptv/data

docker pull docker.1ms.run/library/python:3.12-alpine
docker tag docker.1ms.run/library/python:3.12-alpine python:3.12-alpine

FROM python:3.12-alpine
# 国内源: alpine apk + pypi
RUN sed -i 's#https://dl-cdn.alpinelinux.org#https://mirrors.tuna.tsinghua.edu.cn#g' /etc/apk/repositories \
    && apk add --no-cache tzdata \
    && pip install --no-cache-dir -i https://pypi.tuna.tsinghua.edu.cn/simple requests
WORKDIR /app
COPY renewer.py .
CMD ["python", "-u", "renewer.py"]

services:
  renewer:
    build: ./renewer
    container_name: iptv-renewer
    restart: unless-stopped
    network_mode: host    # 关键: 与机顶盒同出口 IP, 认证才会被接受
    pid: host             # 关键: 让容器看到宿主机进程, 才能发 SIGHUP
    privileged: true      # 配合 pid:host, 允许向宿主进程发信号
    environment:
      TZ: Asia/Shanghai
      CHECK_INTERVAL: "300"        # 每 5 分钟检查一轮
      RENEW_MAX_AGE_H: "24"        # 定时兜底周期
      STREAM_CHECK_URL: "http://127.0.0.1:5140/%E5%A4%AE%E8%A7%86/CCTV-1"
      HTTP_M3U_BASE: "http://10.0.0.1:5140"
    volumes:
      - ./data:/data
      # 单文件挂载, 容器内原地覆写(同 inode), 不破坏挂载
      - /etc/rtp2httpd.conf:/conf/rtp2httpd.conf

cd /root/iptv && docker compose up -d --build

# 看首轮续期日志
docker logs -f iptv-renewer

[12:35:36] iptv-renewer 启动: 检查间隔 300s, 最长续期周期 24.0h
[12:35:36] 需要续期: 出口IP漂移: None -> 27.152.x.x
[12:35:36] ===== 开始续期 (出口IP=27.152.x.x) =====
[12:35:36] 认证成功: UserToken=MyumOM****DscD
[12:35:36] 频道列表: HTTP 200, 698892B
[12:35:36] 解析频道: 235
[12:35:36] 已写入 /conf/rtp2httpd.conf (116 RTSP 频道)
[12:35:36] 已向 rtp2httpd (pid=13154) 发送 SIGHUP 热重载
[12:35:38] 试播验证: 成功 (65536B)
[12:35:38] HLS 频道: 103
[12:35:38] ===== 续期完成, 试播正常 =====

[12:40:39] 试播验证: 成功 (65536B)
[12:40:39] 令牌有效 (IP=27.152.x.x, 已 0.1h), 跳过

curl -o /dev/null -w "%{http_code} %{size_download}B\n" \
  "http://10.0.0.1:5140/%E5%A4%AE%E8%A7%86/CCTV-1" --max-time 8
# 200 262144B  ✅ 全家设备的播放地址, 一个字都没改过

def find_master_pid():
    best, best_start = None, None
    for d in os.listdir('/proc'):
        if not d.isdigit():
            continue
        try:
            with open(f'/proc/{d}/comm') as f:
                if f.read().strip() != 'rtp2httpd':
                    continue
            with open(f'/proc/{d}/stat') as f:
                start = int(f.read().split(') ', 1)[1].split()[19])
            if best_start is None or start < best_start:
                best, best_start = int(d), start
        except (OSError, IndexError, ValueError):
            continue
    return best   # 取启动最早的 = master

机顶盒(已收进抽屉)
                ┌─ 每5min: IP探测 + 试播探活
电信 EPG ◄──────┤─ 触发时: 三步重放认证 → 拉全量频道表
                └─ 更新 conf → SIGHUP 热重载(不断流) → 试播验证
                        │
OpenWrt rtp2httpd :5140 │  ← 令牌只活在这一段
                        ▼
   全屋设备 http://路由器:5140/分组/频道名  ← 永远不变

# -*- coding: utf-8 -*-
"""IPTV 令牌自动续期容器 (免盒子, 24小时无感)

运行于 OpenWrt 路由器 Docker, host 网络 (与机顶盒同 CGNAT 出口)。
触发条件(满足其一):
  1. 出口公网 IP 与上次续期时不一致 (令牌绑定出口 IP, 漂移即失效)
  2. 距上次续期超过 RENEW_MAX_AGE_H 小时 (定时兜底)
  3. 连续两次试播失败 (自愈)
动作: 三步重放认证 -> 拉取全量频道 -> 原地覆写 /etc/rtp2httpd.conf
      -> SIGHUP 热重载 rtp2httpd (不断流) -> 试播验证 -> 产出 m3u
"""
import gzip
import json
import os
import re
import signal
import sys
import time
import zlib
from urllib.parse import quote

import requests

# ============ 设备身份凭证 (来自抓包, 长期有效) ============
HOST = 'http://27.148.x.x:33200'          # EPG 服务器
USERID = '<你的UserID>'                    # 抓包中 UserID= 参数
AUTHENTICATOR = '<128位hex,抓包获得>'       # ValidAuthenticationHWCTC 提交的凭证
TEMPKEY = '<32位hex,抓包获得>'              # userToken/tempKey
STB = dict(STBType='EC6108V9U_pub_fjmdx', STBVersion='19.2.0-LFJD04.B015',
           conntype='2', STBID='<盒子的STBID>',
           templateName='StbEpg2023Group', areaId='1111',
           userGroupId='<你的userGroupId>', mac='<盒子MAC>',
           UserField='2', SoftwareVersion='19.2.0-LFJD04.B015',
           IsSmartStb='0')

UA = {'User-Agent': 'Mozilla/5.0 (X11; U; Linux i686; en-US) AppleWebKit/534.0 (KHTML, like Gecko)'}

# ============ 运行配置 ============
CONF_PATH = '/conf/rtp2httpd.conf'      # 宿主机 /etc/rtp2httpd.conf 单文件挂载
DATA_DIR = '/data'
CHECK_INTERVAL = int(os.environ.get('CHECK_INTERVAL', '300'))
RENEW_MAX_AGE_H = float(os.environ.get('RENEW_MAX_AGE_H', '24'))
STREAM_CHECK_URL = os.environ.get('STREAM_CHECK_URL', 'http://127.0.0.1:5140/%E5%A4%AE%E8%A7%86/CCTV-1')
HTTP_M3U_BASE = os.environ.get('HTTP_M3U_BASE', 'http://10.0.0.1:5140')
IP_ECHO_URLS = ['http://members.3322.org/dyndns/getip', 'https://myip.ipip.net',
                'http://ip.3322.net', 'https://api.ipify.org', 'https://icanhazip.com']
STATE_FILE = os.path.join(DATA_DIR, 'state.json')


def log(msg):
    print(time.strftime('[%Y-%m-%d %H:%M:%S] ') + msg, flush=True)


# ---------------- 出口 IP 探测 ----------------
def get_egress_ip():
    """从文本中提取首个 IPv4 (myip.ipip.net 等返回带中文的句子)"""
    for u in IP_ECHO_URLS:
        try:
            r = requests.get(u, timeout=8)
            m = re.search(r'(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})', r.text)
            if m:
                return m.group(1)
        except Exception as e:
            log(f'IP探测失败 {u}: {e}')
    return None


# ---------------- 认证链 (三步重放) ----------------
def authenticate():
    s = requests.Session()
    s.headers.update(UA)
    s.get(HOST + '/EPG/jsp/AuthenticationURL',
          params={'UserID': USERID, 'Action': 'Login'}, timeout=15)
    s.post(HOST + '/EPG/jsp/authLoginHWCTC.jsp', data=f'UserID={USERID}&VIP=',
           headers={'Referer': HOST + f'/EPG/jsp/AuthenticationURL?UserID={USERID}&Action=Login',
                    'Content-Type': 'application/x-www-form-urlencoded'}, timeout=15)
    body = (f'UserID={USERID}&Lang=1&SupportHD=1&NetUserID=&Authenticator={AUTHENTICATOR}'
            f'&STBType={STB["STBType"]}&STBVersion={STB["STBVersion"]}&conntype={STB["conntype"]}'
            f'&STBID={STB["STBID"]}&templateName={STB["templateName"]}&areaId={STB["areaId"]}'
            f'&userToken={TEMPKEY}&userGroupId={STB["userGroupId"]}&productPackageId='
            f'&mac={STB["mac"].replace(":", "%3A")}&UserField={STB["UserField"]}'
            f'&SoftwareVersion={STB["SoftwareVersion"]}&IsSmartStb={STB["IsSmartStb"]}'
            f'&desktopId=&stbmaker=&VIP=')
    r = s.post(HOST + '/EPG/jsp/ValidAuthenticationHWCTC.jsp', data=body,
               headers={'Referer': HOST + '/EPG/jsp/authLoginHWCTC.jsp',
                        'Content-Type': 'application/x-www-form-urlencoded'}, timeout=15)
    ut = re.search(r'"UserToken" value="(.+?)"', r.text)
    jsid = s.cookies.get('JSESSIONID')
    if not ut or not jsid:
        raise RuntimeError('认证失败, 响应片段: ' + r.text[:300])
    log(f'认证成功: UserToken={ut.group(1)}')
    return ut.group(1), jsid


def fetch_channels(user_token, jsid):
    body = (f'conntype={STB["conntype"]}&UserToken={user_token}&tempKey={TEMPKEY}'
            f'&stbid=990060&SupportHD=1&UserID={USERID}&Lang=1')
    r = requests.post(HOST + '/EPG/jsp/getchannellistHWCTC.jsp', data=body,
                      headers={**UA, 'Referer': HOST + '/EPG/index.jsp',
                               'Content-Type': 'application/x-www-form-urlencoded',
                               'Cookie': f'JSESSIONID={jsid}'}, timeout=60)
    content = r.content
    stripped = content.lstrip(b'\r\n \t')
    if stripped[:2] == b'\x1f\x8b':
        content = gzip.decompress(stripped)
    elif 'deflate' in (r.headers.get('Content-Encoding') or '').lower():
        try:
            content = zlib.decompress(stripped)
        except zlib.error:
            content = zlib.decompress(stripped, -zlib.MAX_WBITS)
    log(f'频道列表: HTTP {r.status_code}, {len(content)}B')
    return content.decode('utf-8', 'replace')


def parse_channels(text):
    pat = re.compile(
        r'ChannelName="([^"]+)".*?UserChannelID="(\d+)".*?ChannelURL="([^"]+)".*?ChannelSDP="([^"]+)"', re.S)
    chans, seen = [], set()
    for m in pat.finditer(text):
        no = m.group(2)
        if no in seen:
            continue
        seen.add(no)
        chans.append({'name': m.group(1), 'no': no, 'url': m.group(3), 'sdp': m.group(4)})
    log(f'解析频道: {len(chans)}')
    return chans


# ---------------- 频道清洗/分组 ----------------
def clean(n):
    n = re.sub(r'[（(][^（）()]*[）)]', '', n)
    n = re.sub(r'-?RTSP.*$', '', n, flags=re.I)
    n = re.sub(r'\(0222\)|\(现网使用\)|\(测试\)|-2311|-2212', '', n)
    n = n.replace('RSTP', '').strip()
    return re.sub(r'[-\s]+$', '', n).strip()


def group(name):
    if '未知' in name:
        return '其他'
    if '4K' in name or '超高清' in name:
        return '4K专区'
    if name.startswith('CCTV') or 'CGTN' in name or '中国教育' in name:
        return '央视'
    if any(k in name for k in ('福建', '东南卫视', '海峡', '厦门')):
        return '本地福建'
    if '卡通' in name or '少儿' in name or '中学生' in name:
        return '少儿'
    if '卫视' in name:
        return '卫视'
    return '其他'


ORDER = ['央视', '卫视', '本地福建', '4K专区', '少儿', '其他']


def build_channel_list(chans):
    """RTSP 与 HLS 是两组不相交的频道, 分开产出。
    返回 (rtsp_items, hls_items): 各为 [(group, name, url)]"""
    rtsp_items, hls_items = [], []
    for c in sorted(chans, key=lambda x: int(x['no']) if x['no'].isdigit() else 99999):
        name = clean(c['name'])
        if not name or '直播室' in name or '马赛克' in name:
            continue
        g = group(name)
        if c['sdp'].startswith('rtsp'):
            rtsp_items.append((g, name, c['sdp']))
        elif c['url'].startswith('rtsp'):
            rtsp_items.append((g, name, c['url']))
        if c['url'].startswith('http'):
            hls_items.append((g, name, c['url']))
    return rtsp_items, hls_items


def gen_conf(items):
    groups = {}
    for g, n, u in items:
        groups.setdefault(g, []).append((n, u))
    cfg = ['#RTP2HTTPD Config file - auto renewed by iptv-renewer',
           time.strftime('# last renewal: %Y-%m-%d %H:%M:%S'), '',
           '[global]', 'verbosity = 1', 'maxclients = 20', 'workers = 2',
           'cors-allow-origin = *', '', '[bind]', '* 5140', '', '[services]', '#EXTM3U']
    for g in ORDER:
        lst = groups.get(g)
        if not lst:
            continue
        cfg.append(f'# ---- {g} ----')
        for n, u in lst:
            cfg.append(f'#EXTINF:-1 group-title="{g}",{n}')
            cfg.append(u)
    return '\n'.join(cfg) + '\n'


def gen_http_m3u(items):
    """客户端播放列表: 地址永远不变 -> 全屋无感"""
    out = ['#EXTM3U', '']
    for g, n, _ in items:
        out.append(f'# ---- {g} ----')
        out.append(f'#EXTINF:-1 group-title="{g}",{n}')
        out.append(f'{HTTP_M3U_BASE}/{quote(g, safe="")}/{quote(n, safe="")}')
        out.append('')
    return '\n'.join(out) + '\n'


def gen_rtsp_m3u(items):
    out = ['#EXTM3U']
    for g, n, u in items:
        out.append(f'#EXTINF:-1 tvg-name="{n}" group-title="{g}",{n}')
        out.append(u)
    return '\n'.join(out) + '\n'


def gen_hls_m3u(items):
    out = ['#EXTM3U']
    for g, n, u in items:
        out.append(f'#EXTINF:-1 tvg-name="{n}" group-title="{g}",{n}')
        out.append(u)
    return '\n'.join(out) + '\n'


# ---------------- rtp2httpd 热重载 ----------------
def find_master_pid():
    """pid:host 模式下容器可直接看到宿主机进程, 取最早启动的 master"""
    best, best_start = None, None
    for d in os.listdir('/proc'):
        if not d.isdigit():
            continue
        try:
            with open(f'/proc/{d}/comm') as f:
                if f.read().strip() != 'rtp2httpd':
                    continue
            with open(f'/proc/{d}/stat') as f:
                start = int(f.read().split(') ', 1)[1].split()[19])
            if best_start is None or start < best_start:
                best, best_start = int(d), start
        except (OSError, IndexError, ValueError):
            continue
    return best


def hot_reload():
    pid = find_master_pid()
    if pid is None:
        raise RuntimeError('未找到宿主机 rtp2httpd 进程')
    os.kill(pid, signal.SIGHUP)
    log(f'已向 rtp2httpd (pid={pid}) 发送 SIGHUP 热重载')
    time.sleep(2)
    if find_master_pid() is None:
        raise RuntimeError('SIGHUP 后 rtp2httpd 消失了!')
    return pid


def verify_stream():
    """试播验证: 拉一段流即认为令牌有效"""
    try:
        with requests.get(STREAM_CHECK_URL, stream=True, timeout=15) as r:
            n = len(r.raw.read(64 * 1024))
        log(f'试播验证: {"成功" if n > 0 else "无数据"} ({n}B)')
        return n > 0
    except Exception as e:
        log(f'试播验证失败: {e}')
        return False


# ---------------- 状态与主循环 ----------------
def load_state():
    try:
        return json.load(open(STATE_FILE, encoding='utf-8'))
    except Exception:
        return None


def save_state(ip, n_ch):
    json.dump({'ts': time.time(), 'ts_str': time.strftime('%Y-%m-%d %H:%M:%S'),
               'egress_ip': ip, 'channels': n_ch},
              open(STATE_FILE, 'w', encoding='utf-8'), ensure_ascii=False, indent=1)


def write_atomic(path, content):
    tmp = path + '.tmp'
    with open(tmp, 'w', encoding='utf-8', newline='\n') as f:
        f.write(content)
    os.replace(tmp, path)


def do_renew(current_ip):
    log(f'===== 开始续期 (出口IP={current_ip}) =====')
    user_token, jsid = authenticate()
    text = fetch_channels(user_token, jsid)
    chans = parse_channels(text)
    rtsp_items, hls_items = build_channel_list(chans)
    if len(rtsp_items) < 50:
        raise RuntimeError(f'频道数异常 ({len(rtsp_items)}), 拒绝覆写配置')
    conf = gen_conf(rtsp_items)
    # 原地覆写挂载文件 (保持同一 inode, bind mount 有效)
    with open(CONF_PATH, 'w', encoding='utf-8', newline='\n') as f:
        f.write(conf)
    log(f'已写入 {CONF_PATH} ({len(rtsp_items)} RTSP 频道)')
    hot_reload()
    ok = verify_stream()
    # 产出 m3u 副本
    write_atomic(os.path.join(DATA_DIR, 'channels.m3u'), gen_rtsp_m3u(rtsp_items))
    write_atomic(os.path.join(DATA_DIR, 'channels-hls.m3u'), gen_hls_m3u(hls_items))
    write_atomic(os.path.join(DATA_DIR, 'channels-http.m3u'), gen_http_m3u(rtsp_items))
    save_state(current_ip, len(rtsp_items))
    log(f'HLS 频道: {len(hls_items)}')
    log(f'===== 续期完成, 试播{"正常" if ok else "异常(请人工检查)"} =====')
    return ok


def need_renew(state, current_ip):
    if state is None:
        return '首次运行'
    if current_ip and state.get('egress_ip') != current_ip:
        return f'出口IP漂移: {state.get("egress_ip")} -> {current_ip}'
    if time.time() - state.get('ts', 0) > RENEW_MAX_AGE_H * 3600:
        return f'令牌超过 {RENEW_MAX_AGE_H}h 定时刷新'
    return None


def main():
    log(f'iptv-renewer 启动: 检查间隔 {CHECK_INTERVAL}s, 最长续期周期 {RENEW_MAX_AGE_H}h')
    probe_fails = 0
    while True:
        try:
            state = load_state()
            ip = get_egress_ip()
            reason = need_renew(state, ip)
            if not reason and state is not None:
                # 令牌名义上仍有效 -> 试播探活, 失败即自愈续期
                if not verify_stream():
                    probe_fails += 1
                    if probe_fails >= 2:
                        reason = '连续试播失败, 疑似令牌失效 (自动自愈)'
                        probe_fails = 0
                    else:
                        log('试播失败一次, 下轮复检')
                        time.sleep(15)
                else:
                    probe_fails = 0
            if reason:
                log(f'需要续期: {reason}')
                try:
                    do_renew(ip)
                except Exception as e:
                    log(f'!! 续期失败 (保留旧配置继续服务): {e}')
            elif state is not None:
                log(f'令牌有效 (IP={state.get("egress_ip")}, '
                    f'已 {(time.time()-state["ts"])/3600:.1f}h), 跳过')
        except Exception as e:
            log(f'!! 循环异常: {e}')
        time.sleep(CHECK_INTERVAL)


if __name__ == '__main__':
    sys.exit(main())