's'

web2_relay_pro.py

@@ -3,6 +3,7 @@ Pro 多路转播：每频道独立 URL 配置文件、跨频道抖音地址去
 """
 from __future__ import annotations
 
+import configparser
 import json
 import re
 import time
@@ -43,6 +44,36 @@ def url_config_path_for_channel(config_dir: Path, channel_id: str) -> Path:
     return relay_pro_dir(config_dir) / f"url_config.{_safe_channel_file_id(channel_id)}.ini"
 
 
+def youtube_ini_path_for_channel(config_dir: Path, channel_id: str) -> Path:
+    return relay_pro_dir(config_dir) / f"youtube.{_safe_channel_file_id(channel_id)}.ini"
+
+
+def materialize_relay_channel_youtube_ini(config_dir: Path, channel_id: str) -> tuple[bool, str]:
+    """为独立子进程写入 relay_pro/youtube.<id>.ini（密钥来自频道；其余项尽量继承全局 youtube.ini）。"""
+    config_dir = Path(config_dir)
+    d = get_channel_detail(config_dir, channel_id)
+    if not d:
+        return False, "频道不存在"
+    key = str(d.get("youtubeKey") or "").strip()
+    if not key:
+        return False, "请先配置该频道的串流密钥"
+    out = youtube_ini_path_for_channel(config_dir, channel_id)
+    relay_pro_dir(config_dir).mkdir(parents=True, exist_ok=True)
+    cp = configparser.ConfigParser()
+    global_ini = config_dir / "youtube.ini"
+    if global_ini.exists():
+        try:
+            cp.read(global_ini, encoding="utf-8")
+        except Exception:
+            pass
+    if not cp.has_section("youtube"):
+        cp.add_section("youtube")
+    cp.set("youtube", "key", key)
+    with open(out, "w", encoding="utf-8") as f:
+        cp.write(f)
+    return True, str(out)
+
+
 def channels_json_path(config_dir: Path) -> Path:
     return config_dir / "youtube_channels.json"
 
@@ -74,7 +105,7 @@ def ensure_default_relay_config(config_dir: Path) -> None:
         "channels": [
             {
                 "id": "ch_default",
-                "name": "默认线路",
+                "name": "频道",
                 "douyinUrl": "",
                 "youtubeKey": "",
             }
@@ -178,43 +209,81 @@ def get_channel_by_id(config_dir: Path, channel_id: str) -> dict[str, Any] | Non
     return None
 
 
+def _all_keys_for_channel(item: dict) -> list[str]:
+    lk = item.get("youtubeKeys")
+    if isinstance(lk, list) and len(lk) > 0:
+        return [str(x).strip() for x in lk if str(x).strip()]
+    k = str(item.get("youtubeKey") or item.get("key") or "").strip()
+    return [k] if k else []
+
+
 def get_channel_detail(config_dir: Path, channel_id: str) -> dict[str, Any] | None:
     ch = get_channel_by_id(config_dir, channel_id)
     if not ch:
         return None
+    keys = _all_keys_for_channel(ch)
+    if not keys:
+        keys = [""]
+    active = str(ch.get("youtubeKey") or ch.get("key") or "").strip()
+    ai = 0
+    if active:
+        for i, k in enumerate(keys):
+            if _norm_stream_key(k) == _norm_stream_key(active):
+                ai = i
+                break
     return {
         "id": str(ch.get("id") or "").strip(),
         "name": str(ch.get("name") or "").strip() or "未命名",
         "douyinUrl": str(ch.get("douyinUrl") or ch.get("douyin_url") or "").strip(),
-        "youtubeKey": str(ch.get("youtubeKey") or ch.get("key") or "").strip(),
+        "youtubeKey": active,
+        "youtubeKeys": keys,
+        "youtubeActiveIndex": ai,
     }
 
 
 def update_channel_key(config_dir: Path, channel_id: str, new_key: str) -> tuple[bool, str]:
+    return update_channel_keys_list(config_dir, channel_id, [new_key.strip()], 0)
+
+
+def update_channel_keys_list(
+    config_dir: Path, channel_id: str, keys: list[str], active_index: int
+) -> tuple[bool, str]:
     doc, errs = load_channels_document(config_dir)
     if doc is None:
         return False, errs[0] if errs else "无法读取配置"
     arr = doc.get("channels")
     if not isinstance(arr, list):
         return False, "channels 无效"
-    nk = _norm_stream_key(new_key)
-    if not nk:
-        return False, "密钥不能为空"
+    cleaned = [str(x).strip() for x in keys]
+    cleaned = [x for x in cleaned if x]
+    if not cleaned:
+        return False, "至少需要一个有效密钥"
+    norms = [_norm_stream_key(k) for k in cleaned]
+    if len(set(norms)) != len(norms):
+        return False, "同一线路内不能重复同一串流密钥"
+    ai = max(0, min(int(active_index), len(cleaned) - 1))
+    active_raw = cleaned[ai]
     for item in arr:
         if not isinstance(item, dict):
             continue
         oid = str(item.get("id") or "").strip()
         if oid == channel_id:
             continue
-        ok = _norm_stream_key(str(item.get("youtubeKey") or item.get("key") or ""))
-        if ok and ok == nk:
-            return False, "禁止多条线路共用同一 YouTube 串流密钥"
+        for ok in _all_keys_for_channel(item):
+            okn = _norm_stream_key(ok)
+            if not okn:
+                continue
+            for ck in cleaned:
+                ckn = _norm_stream_key(ck)
+                if ckn and ckn == okn:
+                    return False, "禁止多条线路共用同一 YouTube 串流密钥"
     for item in arr:
         if not isinstance(item, dict):
             continue
         if str(item.get("id") or "").strip() != channel_id:
             continue
-        item["youtubeKey"] = new_key.strip()
+        item["youtubeKeys"] = cleaned
+        item["youtubeKey"] = active_raw
         channels_json_path(config_dir).write_text(json.dumps(doc, ensure_ascii=False, indent=2), encoding="utf-8")
         return True, "已保存"
     return False, "未找到该频道"
@@ -352,32 +421,48 @@ def parse_relay_meta_lines(log_text: str) -> list[dict[str, Any]]:
     return out
 
 
+def relay_pm2_name_for_channel(channel_id: str) -> str:
+    return f"youtube2__{channel_id}"
+
+
 async def build_relay_live_status(
     config_dir: Path,
     base_dir: Path,
     get_process_info: Callable[[str], Awaitable[dict[str, Any]]],
     read_log_tail: Callable[[str, int], Awaitable[str]],
-    youtube_pm2_name: str,
+    _youtube_pm2_name_legacy: str,
 ) -> dict[str, Any]:
-    """工作台 Pro：解析日志 RELAY_META + 进程状态，匹配 youtube_channels.json 频道名。"""
+    """工作台 Pro：每频道独立进程 youtube2__<channelId>；解析 RELAY_META 日志。"""
     ch_doc = list_channels_for_pro(config_dir)
     id_to_name = {c["id"]: c["name"] for c in ch_doc.get("channels") or []}
-    try:
-        info = await get_process_info(youtube_pm2_name)
-    except Exception:
-        info = {"process_status": "unknown", "out_path": "", "err_path": ""}
-    st = str(info.get("process_status") or "unknown")
-    online = st == "online"
-    out_path = str(info.get("out_path") or "")
-    err_path = str(info.get("err_path") or "")
+    channels = ch_doc.get("channels") or []
     recent = ""
-    try:
-        if out_path:
-            recent += await read_log_tail(out_path, 400)
-        if err_path:
-            recent += "\n" + await read_log_tail(err_path, 400)
-    except Exception:
-        recent = recent or ""
+    any_online = False
+    channel_statuses: list[dict[str, Any]] = []
+    for c in channels:
+        cid = str(c.get("id") or "").strip()
+        if not cid:
+            continue
+        cname = str(c.get("name") or "").strip() or "未命名"
+        pname = relay_pm2_name_for_channel(cid)
+        try:
+            info = await get_process_info(pname)
+        except Exception:
+            info = {"process_status": "unknown", "out_path": "", "err_path": ""}
+        st = str(info.get("process_status") or "unknown")
+        online = st == "online"
+        if online:
+            any_online = True
+        out_path = str(info.get("out_path") or "")
+        err_path = str(info.get("err_path") or "")
+        try:
+            if out_path:
+                recent += await read_log_tail(out_path, 400)
+            if err_path:
+                recent += "\n" + await read_log_tail(err_path, 400)
+        except Exception:
+            pass
+        channel_statuses.append({"channelId": cid, "channelName": cname, "live": online})
     metas = parse_relay_meta_lines(recent)
     last = metas[-1] if metas else None
     rows: list[dict[str, Any]] = []
@@ -397,18 +482,30 @@ async def build_relay_live_status(
                 "streamTitle": last.get("streamTitle"),
                 "youtubeKeySuffix": last.get("youtubeKeySuffix"),
                 "durationSec": duration_sec,
-                "processOnline": online,
+                "processOnline": any_online,
             }
         )
     return {
-        "youtubeProcess": youtube_pm2_name,
-        "processStatus": st,
-        "processOnline": online,
+        "youtubeProcess": "multi",
+        "processStatus": "online" if any_online else "stopped",
+        "processOnline": any_online,
         "relayRows": rows,
         "lastRelayMeta": last,
+        "matchedChannelId": matched_id or None,
+        "channelStatuses": channel_statuses,
     }
 
 
+def _key_tail_for_match(k: str) -> str:
+    k = (k or "").strip()
+    if not k:
+        return ""
+    if k.startswith("rtmp://") or k.startswith("rtmps://"):
+        parts = k.rstrip("/").split("/")
+        k = parts[-1] if parts else k
+    return k[-8:].lower() if len(k) >= 8 else k.lower()
+
+
 def _match_channel_by_key_suffix(config_dir: Path, suffix: str) -> tuple[str, str]:
     """用 RELAY_META 的 youtubeKeySuffix 与 youtube_channels.json 中密钥比对。"""
     if not suffix:
@@ -416,21 +513,98 @@ def _match_channel_by_key_suffix(config_dir: Path, suffix: str) -> tuple[str, st
     doc, _ = load_channels_document(config_dir)
     if not doc or not isinstance(doc.get("channels"), list):
         return "", ""
+    su = suffix.strip().lower()
     for item in doc["channels"]:
         if not isinstance(item, dict):
             continue
-        k = str(item.get("youtubeKey") or item.get("key") or "").strip()
-        if not k:
-            continue
-        if k.startswith("rtmp://") or k.startswith("rtmps://"):
-            parts = k.rstrip("/").split("/")
-            k = parts[-1] if parts else k
-        tail = k[-8:].lower() if len(k) >= 8 else k.lower()
-        if suffix == tail or (len(suffix) >= 4 and tail.endswith(suffix)):
-            return str(item.get("id") or ""), str(item.get("name") or "")
+        for k in _all_keys_for_channel(item):
+            tail = _key_tail_for_match(k)
+            if not tail:
+                continue
+            if su == tail or (len(su) >= 4 and tail.endswith(su)):
+                return str(item.get("id") or ""), str(item.get("name") or "")
     return "", ""
 
 
+def add_relay_channel(
+    config_dir: Path, name: str, youtube_key: str = ""
+) -> tuple[bool, str, str | None]:
+    """新增一条频道并创建对应 url_config 文件。"""
+    config_dir = Path(config_dir)
+    relay_pro_dir(config_dir)
+    doc, errs = load_channels_document(config_dir)
+    if doc is None:
+        return False, errs[0] if errs else "无法读取配置", None
+    arr = doc.get("channels")
+    if not isinstance(arr, list):
+        return False, "channels 无效", None
+    key = (youtube_key or "").strip()
+    if key:
+        nk = _norm_stream_key(key)
+        for item in arr:
+            if not isinstance(item, dict):
+                continue
+            for ok in _all_keys_for_channel(item):
+                if nk and nk == _norm_stream_key(ok):
+                    return False, "禁止多条线路共用同一 YouTube 串流密钥", None
+    new_id = f"ch_{int(time.time())}"
+    while any(str(x.get("id") or "").strip() == new_id for x in arr if isinstance(x, dict)):
+        new_id = f"ch_{int(time.time())}_{len(arr)}"
+    nm = (name or "").strip() or f"线路{len(arr) + 1}"
+    row: dict[str, Any] = {
+        "id": new_id,
+        "name": nm,
+        "douyinUrl": "",
+        "youtubeKey": key,
+    }
+    if key:
+        row["youtubeKeys"] = [key]
+    arr.append(row)
+    channels_json_path(config_dir).write_text(json.dumps(doc, ensure_ascii=False, indent=2), encoding="utf-8")
+    uc = url_config_path_for_channel(config_dir, new_id)
+    if not uc.exists():
+        uc.write_text(
+            "# 每行一个抖音直播间地址，例如：\n# https://live.douyin.com/123456789\n\n",
+            encoding="utf-8",
+        )
+    return True, "已新增频道", new_id
+
+
+def remove_relay_channel(config_dir: Path, channel_id: str) -> tuple[bool, str]:
+    """从 youtube_channels.json 移除一条频道，并删除对应的 url_config / relay_pro youtube.ini（若存在）。"""
+    cid = (channel_id or "").strip()
+    if not cid:
+        return False, "缺少 channelId"
+    config_dir = Path(config_dir)
+    doc, errs = load_channels_document(config_dir)
+    if doc is None:
+        return False, errs[0] if errs else "无法读取配置"
+    arr = doc.get("channels")
+    if not isinstance(arr, list):
+        return False, "channels 无效"
+    if len(arr) <= 1:
+        return False, "至少保留一条频道"
+    idx: int | None = None
+    for i, item in enumerate(arr):
+        if isinstance(item, dict) and str(item.get("id") or "").strip() == cid:
+            idx = i
+            break
+    if idx is None:
+        return False, "未找到该频道"
+    arr.pop(idx)
+    channels_json_path(config_dir).write_text(json.dumps(doc, ensure_ascii=False, indent=2), encoding="utf-8")
+    for p in (
+        url_config_path_for_channel(config_dir, cid),
+        youtube_ini_path_for_channel(config_dir, cid),
+    ):
+        try:
+            if p.exists():
+                p.unlink()
+        except OSError:
+            pass
+    return True, "已删除频道"
+
+
 def first_youtube_pm2_name(script_entries: list[dict[str, Any]]) -> str:
     for e in script_entries:
         script = str(e.get("script") or "")