's'

web2_relay_pro.py

@@ -0,0 +1,439 @@
+"""
+Pro 多路转播：每频道独立 URL 配置文件、跨频道抖音地址去重、同步到 legacy youtube.ini + URL_config.ini。
+"""
+from __future__ import annotations
+
+import json
+import re
+import time
+from pathlib import Path
+from typing import Any, Callable, Awaitable
+
+# 与 web2_client_overview 一致的去重键
+def _norm_douyin_url(url: str) -> str:
+    u = (url or "").strip()
+    if not u:
+        return ""
+    u = re.sub(r"#.*$", "", u).strip()
+    return u.rstrip("/").lower()
+
+
+def _norm_stream_key(key: str) -> str:
+    k = (key 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.strip().lower()
+
+
+def relay_pro_dir(config_dir: Path) -> Path:
+    p = config_dir / "relay_pro"
+    p.mkdir(parents=True, exist_ok=True)
+    return p
+
+
+def _safe_channel_file_id(channel_id: str) -> str:
+    s = re.sub(r"[^a-zA-Z0-9_-]", "_", (channel_id or "").strip())
+    return s or "default"
+
+
+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 channels_json_path(config_dir: Path) -> Path:
+    return config_dir / "youtube_channels.json"
+
+
+def ensure_default_relay_config(config_dir: Path) -> None:
+    """若缺少 youtube_channels.json 或 channels 为空，则写入默认 ch_default 与 url_config（便于独立运行 uvicorn）。"""
+    config_dir = Path(config_dir)
+    config_dir.mkdir(parents=True, exist_ok=True)
+    relay_pro_dir(config_dir)
+    p = channels_json_path(config_dir)
+    need_write = False
+    if not p.exists():
+        need_write = True
+    else:
+        try:
+            raw = json.loads(p.read_text(encoding="utf-8"))
+        except Exception:
+            need_write = True
+            raw = None
+        else:
+            arr = raw.get("channels") if isinstance(raw, dict) else None
+            if not isinstance(arr, list) or len(arr) == 0:
+                need_write = True
+    if not need_write:
+        return
+    default_doc: dict[str, Any] = {
+        "version": 1,
+        "autoManaged": True,
+        "channels": [
+            {
+                "id": "ch_default",
+                "name": "默认线路",
+                "douyinUrl": "",
+                "youtubeKey": "",
+            }
+        ],
+    }
+    p.write_text(json.dumps(default_doc, ensure_ascii=False, indent=2), encoding="utf-8")
+    uc = url_config_path_for_channel(config_dir, "ch_default")
+    if not uc.exists():
+        uc.write_text(
+            "# 软件已自动创建。每行一个抖音直播间地址，例如：\n# https://live.douyin.com/123456789\n\n",
+            encoding="utf-8",
+        )
+
+
+def load_channels_document(config_dir: Path) -> tuple[dict[str, Any] | None, list[str]]:
+    p = channels_json_path(config_dir)
+    if not p.exists():
+        return None, ["缺少 config/youtube_channels.json，请先添加（见 youtube_channels.example.json）"]
+    try:
+        raw = json.loads(p.read_text(encoding="utf-8"))
+    except Exception as e:
+        return None, [f"youtube_channels.json 解析失败: {e}"]
+    if not isinstance(raw, dict):
+        return None, ["youtube_channels.json 格式错误"]
+    arr = raw.get("channels")
+    if not isinstance(arr, list):
+        return None, ["youtube_channels.json 缺少 channels 数组"]
+    return raw, []
+
+
+def ensure_channel_ids(config_dir: Path, raw: dict[str, Any]) -> bool:
+    """为缺 id 的条目补 ch_0, ch_1 并写回文件。返回是否写回。"""
+    arr = raw.get("channels")
+    if not isinstance(arr, list):
+        return False
+    changed = False
+    for i, item in enumerate(arr):
+        if not isinstance(item, dict):
+            continue
+        cid = str(item.get("id") or "").strip()
+        if not cid:
+            item["id"] = f"ch_{i}"
+            changed = True
+    if changed:
+        channels_json_path(config_dir).write_text(json.dumps(raw, ensure_ascii=False, indent=2), encoding="utf-8")
+    return changed
+
+
+def list_channels_for_pro(config_dir: Path) -> dict[str, Any]:
+    doc, errs = load_channels_document(config_dir)
+    if doc is None:
+        return {"ok": False, "error": errs[0] if errs else "未知错误", "channels": []}
+    ensure_channel_ids(config_dir, doc)
+    doc, _ = load_channels_document(config_dir)
+    assert doc is not None
+    arr = doc.get("channels")
+    out: list[dict[str, Any]] = []
+    if isinstance(arr, list):
+        for item in arr:
+            if not isinstance(item, dict):
+                continue
+            cid = str(item.get("id") or "").strip()
+            name = str(item.get("name") or "").strip() or "未命名"
+            durl = str(item.get("douyinUrl") or item.get("douyin_url") or "").strip()
+            key = str(item.get("youtubeKey") or item.get("key") or "").strip()
+            if not cid:
+                continue
+            kp = _mask_key_preview(key)
+            out.append(
+                {
+                    "id": cid,
+                    "name": name,
+                    "douyinUrl": durl,
+                    "keyPreview": kp,
+                }
+            )
+    return {"ok": True, "channels": out}
+
+
+def _mask_key_preview(key: str) -> str:
+    k = (key or "").strip()
+    if not k:
+        return ""
+    if len(k) <= 8:
+        return "****"
+    return f"{k[:4]}…{k[-4:]}"
+
+
+def get_channel_by_id(config_dir: Path, channel_id: str) -> dict[str, Any] | None:
+    doc, _ = load_channels_document(config_dir)
+    if doc is None:
+        return None
+    arr = doc.get("channels")
+    if not isinstance(arr, list):
+        return None
+    for item in arr:
+        if not isinstance(item, dict):
+            continue
+        if str(item.get("id") or "").strip() == channel_id:
+            return item
+    return None
+
+
+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
+    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(),
+    }
+
+
+def update_channel_key(config_dir: Path, channel_id: str, new_key: str) -> 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, "密钥不能为空"
+    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 item in arr:
+        if not isinstance(item, dict):
+            continue
+        if str(item.get("id") or "").strip() != channel_id:
+            continue
+        item["youtubeKey"] = new_key.strip()
+        channels_json_path(config_dir).write_text(json.dumps(doc, ensure_ascii=False, indent=2), encoding="utf-8")
+        return True, "已保存"
+    return False, "未找到该频道"
+
+
+def extract_url_lines_from_ini(text: str) -> list[str]:
+    lines: list[str] = []
+    for raw in text.splitlines():
+        t = raw.strip()
+        if not t or t.startswith("#"):
+            continue
+        if "," in t:
+            t = t.split(",")[-1].strip()
+        if "douyin" in t.lower() or "v.douyin.com" in t.lower() or "live.douyin.com" in t.lower():
+            lines.append(t)
+    return lines
+
+
+def collect_normalized_urls_from_all_other_channels(
+    config_dir: Path, exclude_channel_id: str
+) -> dict[str, set[str]]:
+    """channel_id -> set of normalized douyin urls from their url_config file."""
+    doc, _ = load_channels_document(config_dir)
+    out: dict[str, set[str]] = {}
+    if doc is None:
+        return out
+    arr = doc.get("channels")
+    if not isinstance(arr, list):
+        return out
+    for item in arr:
+        if not isinstance(item, dict):
+            continue
+        cid = str(item.get("id") or "").strip()
+        if not cid or cid == exclude_channel_id:
+            continue
+        p = url_config_path_for_channel(config_dir, cid)
+        if not p.exists():
+            out[cid] = set()
+            continue
+        try:
+            text = p.read_text(encoding="utf-8", errors="ignore")
+        except Exception:
+            out[cid] = set()
+            continue
+        s: set[str] = set()
+        for line in extract_url_lines_from_ini(text):
+            nd = _norm_douyin_url(line)
+            if nd:
+                s.add(nd)
+        out[cid] = s
+    return out
+
+
+def validate_url_config_no_cross_duplicate(
+    config_dir: Path, channel_id: str, new_content: str
+) -> list[str]:
+    """保存本频道 URL 列表时，与其它频道的 URL 去重（禁止同一抖音直播间出现在多条线路）。"""
+    errors: list[str] = []
+    mine: set[str] = set()
+    for line in extract_url_lines_from_ini(new_content):
+        nd = _norm_douyin_url(line)
+        if nd:
+            mine.add(nd)
+    others = collect_normalized_urls_from_all_other_channels(config_dir, channel_id)
+    for ocid, oset in others.items():
+        inter = mine & oset
+        if inter:
+            errors.append(
+                f"与频道「{ocid}」的地址列表冲突：以下抖音链接已出现在其它线路（禁止重复）："
+                + "；".join(list(inter)[:5])
+            )
+    return errors
+
+
+def read_url_config_for_channel(config_dir: Path, channel_id: str) -> str:
+    p = url_config_path_for_channel(config_dir, channel_id)
+    if not p.exists():
+        return "# 每行一个抖音直播间地址\n"
+    try:
+        return p.read_text(encoding="utf-8")
+    except Exception:
+        return "# 读取失败\n"
+
+
+def write_url_config_for_channel(config_dir: Path, channel_id: str, content: str) -> tuple[bool, str]:
+    errs = validate_url_config_no_cross_duplicate(config_dir, channel_id, content)
+    if errs:
+        return False, errs[0]
+    p = url_config_path_for_channel(config_dir, channel_id)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_text(content, encoding="utf-8")
+    return True, "保存成功"
+
+
+def write_youtube_ini_single_key(config_dir: Path, key: str) -> None:
+    """写入单路 youtube.ini（[youtube] key=）。"""
+    lines = ["[youtube]", f"key = {key.strip()}", ""]
+    (config_dir / "youtube.ini").write_text("\n".join(lines), encoding="utf-8")
+
+
+def sync_channel_to_legacy_files(config_dir: Path, channel_id: str) -> tuple[bool, str]:
+    """将选中频道的密钥与 URL 列表同步到 youtube.ini + URL_config.ini，供当前单进程 youtube 脚本使用。"""
+    d = get_channel_detail(config_dir, channel_id)
+    if not d:
+        return False, "频道不存在"
+    key = d.get("youtubeKey") or ""
+    if not str(key).strip():
+        return False, "该频道未配置 youtubeKey"
+    url_path = url_config_path_for_channel(config_dir, channel_id)
+    if not url_path.exists():
+        return False, f"缺少该频道的地址文件：{url_path.name}"
+    try:
+        url_body = url_path.read_text(encoding="utf-8")
+    except Exception as e:
+        return False, f"读取地址文件失败: {e}"
+    write_youtube_ini_single_key(config_dir, str(key))
+    (config_dir / "URL_config.ini").write_text(url_body, encoding="utf-8")
+    return True, "已同步到 youtube.ini 与 URL_config.ini"
+
+
+def parse_relay_meta_lines(log_text: str) -> list[dict[str, Any]]:
+    out: list[dict[str, Any]] = []
+    for line in log_text.splitlines():
+        if "[RELAY_META]" not in line:
+            continue
+        idx = line.find("{")
+        if idx < 0:
+            continue
+        try:
+            j = json.loads(line[idx:])
+            if isinstance(j, dict):
+                out.append(j)
+        except Exception:
+            continue
+    return out
+
+
+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,
+) -> dict[str, Any]:
+    """工作台 Pro：解析日志 RELAY_META + 进程状态，匹配 youtube_channels.json 频道名。"""
+    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 "")
+    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 ""
+    metas = parse_relay_meta_lines(recent)
+    last = metas[-1] if metas else None
+    rows: list[dict[str, Any]] = []
+    suffix = str(last.get("youtubeKeySuffix") or "").strip().lower() if last else ""
+    matched_id, matched_name = _match_channel_by_key_suffix(config_dir, suffix)
+    if last:
+        ts = last.get("ts")
+        duration_sec: float | None = None
+        if isinstance(ts, (int, float)):
+            duration_sec = max(0.0, time.time() - float(ts))
+        rows.append(
+            {
+                "channelId": matched_id or None,
+                "channelName": matched_name or id_to_name.get(matched_id, "") or "当前推流",
+                "anchor": last.get("anchor"),
+                "douyinHint": last.get("douyinHint"),
+                "streamTitle": last.get("streamTitle"),
+                "youtubeKeySuffix": last.get("youtubeKeySuffix"),
+                "durationSec": duration_sec,
+                "processOnline": online,
+            }
+        )
+    return {
+        "youtubeProcess": youtube_pm2_name,
+        "processStatus": st,
+        "processOnline": online,
+        "relayRows": rows,
+        "lastRelayMeta": last,
+    }
+
+
+def _match_channel_by_key_suffix(config_dir: Path, suffix: str) -> tuple[str, str]:
+    """用 RELAY_META 的 youtubeKeySuffix 与 youtube_channels.json 中密钥比对。"""
+    if not suffix:
+        return "", ""
+    doc, _ = load_channels_document(config_dir)
+    if not doc or not isinstance(doc.get("channels"), list):
+        return "", ""
+    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 "")
+    return "", ""
+
+
+def first_youtube_pm2_name(script_entries: list[dict[str, Any]]) -> str:
+    for e in script_entries:
+        script = str(e.get("script") or "")
+        if script.endswith(".py") and Path(script).stem.startswith("youtube"):
+            return Path(script).stem
+    return "youtube2"