""" 工作台 /client/overview：YouTube 频道摘要、微信绑定摘要（EasyTier 节点数由 Electron 主进程补充）。 多路转播语义：每条线路为「一个抖音直播间 ↔ 一个 YouTube 串流」一对一； 禁止同一抖音直播间同时向多个 YouTube 频道推流；禁止多条线路共用同一串流密钥（避免串台）。 """ from __future__ import annotations import configparser import json import re from pathlib import Path from typing import Any from web2_wechat_chatbot import build_client, load_wechat_chatbot_config, load_wechat_state def _mask_key(key: str) -> str: k = (key or "").strip() if not k: return "" if len(k) <= 8: return "****" return f"{k[:4]}…{k[-4:]}" def _short_douyin_hint(url: str) -> str: u = (url or "").strip() if not u: return "" if len(u) <= 56: return u return f"{u[:28]}…{u[-16:]}" def _normalize_douyin_url(url: str) -> str: """用于去重：去空白、小写 host、去尾部斜杠。""" u = (url or "").strip() if not u: return "" u = re.sub(r"#.*$", "", u).strip() return u.rstrip("/").lower() def _normalize_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 _parse_youtube_ini_key(ini_path: Path) -> str: if not ini_path.exists(): return "" cp = configparser.ConfigParser() cp.read(ini_path, encoding="utf-8") if not cp.has_section("youtube"): return "" for name, val in cp.items("youtube"): if name in ("key", "stream_key"): v = (val or "").strip() if v and not v.startswith("#"): return v return "" def _first_douyin_line_from_url_config(config_dir: Path) -> str: """单路 legacy：从 URL_config.ini 取首条抖音链接作展示提示（仍可能多 URL 共单 key，与多路一对一模型不同）。""" p = config_dir / "URL_config.ini" if not p.exists(): return "" try: text = p.read_text(encoding="utf-8", errors="ignore") except Exception: return "" for raw in text.splitlines(): line = raw.strip() if not line or line.startswith("#"): continue if "douyin" not in line.lower(): continue if "," in line: line = line.split(",")[-1].strip() return line.strip() return "" def _parse_youtube_channels_json(path: Path) -> tuple[list[dict[str, str]], list[str]]: """ 解析 youtube_channels.json：展示层包含「待填写」线路；仅对同时填写两项的条目做一对一冲突校验。 不完整条目不标为错误，由客户端自动维护文件，用户只需在录制页填写即可。 """ errors: list[str] = [] try: raw = json.loads(path.read_text(encoding="utf-8")) except Exception as e: return [], [f"youtube_channels.json 解析失败: {e}"] arr = raw.get("channels") if isinstance(raw, dict) else None if not isinstance(arr, list): return [], ["youtube_channels.json 缺少 channels 数组"] parsed: list[tuple[str, str, str]] = [] for item in arr: if not isinstance(item, dict): continue name = str(item.get("name") or item.get("id") or "").strip() or "未命名" durl = str( item.get("douyinUrl") or item.get("douyin_url") or item.get("douyin") or item.get("url") or "" ).strip() key = str(item.get("youtubeKey") or item.get("key") or item.get("stream_key") or "").strip() parsed.append((name, durl, key)) # 仅对「两项均已填写」的条目做冲突与规范化校验 strict: list[tuple[str, str, str]] = [] for name, durl, key in parsed: if durl and key: strict.append((name, durl, key)) seen_d: dict[str, str] = {} seen_k: dict[str, str] = {} ok_strict: set[str] = set() for name, durl, key in strict: nd = _normalize_douyin_url(durl) nk = _normalize_stream_key(key) if not nd or not nk: errors.append(f"条目「{name}」地址或密钥无效") continue if nd in seen_d: errors.append( f"禁止重复使用同一抖音直播间绑定多条线路（冲突：「{seen_d[nd]}」与「{name}」）" ) continue if nk in seen_k: errors.append( f"禁止多条抖音线路共用同一 YouTube 串流密钥（冲突：「{seen_k[nk]}」与「{name}」）" ) continue seen_d[nd] = name seen_k[nk] = name ok_strict.add(name) out: list[dict[str, str]] = [] for name, durl, key in parsed: if not durl and not key: out.append( { "name": name, "douyinPreview": "（待填写）", "keyPreview": "（待填写）", } ) continue if not durl or not key: out.append( { "name": name, "douyinPreview": _short_douyin_hint(durl) if durl else "（待填写）", "keyPreview": _mask_key(key) if key else "（待填写）", } ) continue if name in ok_strict: out.append( { "name": name, "douyinPreview": _short_douyin_hint(durl), "keyPreview": _mask_key(key), } ) else: out.append( { "name": name, "douyinPreview": _short_douyin_hint(durl) if durl else "（待填写）", "keyPreview": _mask_key(key) if key else "（待填写）", } ) return out, errors def _youtube_channels_list(config_dir: Path) -> tuple[list[dict[str, str]], list[str], list[str]]: """ 返回：(频道行, 校验错误, 说明性备注)。 错误用于重复抖音/重复密钥等；备注用于单路 legacy 提示，不应标红为「错误」。 """ path = config_dir / "youtube_channels.json" if path.exists(): pairs, errs = _parse_youtube_channels_json(path) if pairs: return pairs, errs, [] if errs: return [], errs, [] return ( [], [], [ "转播线路文件已就绪，请在客户端「录制」页 Pro 转播中填写抖音地址与 YouTube 密钥，将自动保存。", ], ) key = _parse_youtube_ini_key(config_dir / "youtube.ini") if not key: return [], [], [] hint = _first_douyin_line_from_url_config(config_dir) row: dict[str, str] = { "name": "默认（单路）", "keyPreview": _mask_key(key), } if hint: row["douyinPreview"] = _short_douyin_hint(hint) else: row["douyinPreview"] = "（请在 URL_config.ini 填写抖音地址）" notes = [ "当前为单路 legacy：youtube.ini + URL_config.ini；多路一对一由客户端自动生成 youtube_channels.json。", ] return [row], [], notes def build_client_overview(config_dir: Path, base_dir: Path) -> dict[str, Any]: channels, relay_errors, relay_notes = _youtube_channels_list(config_dir) cfg = load_wechat_chatbot_config(base_dir) st = load_wechat_state(base_dir) wechat: dict[str, Any] = { "configured": bool(cfg.get("proxyUrl") and cfg.get("apiKey")), "nickName": st.get("nickName"), "wcId": st.get("wcId"), "online": False, "proxyError": None, } if wechat["configured"]: try: c = build_client(cfg) ps = c.get_status() wechat["online"] = bool(ps.get("isLoggedIn")) or bool(ps.get("wcId")) if ps.get("nickName"): wechat["nickName"] = ps.get("nickName") if ps.get("wcId"): wechat["wcId"] = ps.get("wcId") except Exception as e: wechat["proxyError"] = str(e)[:300] return { "youtubeChannelCount": len(channels), "youtubeChannels": channels, "relayPairingPolicy": "one_douyin_one_youtube", "relayValidationErrors": relay_errors, "relayNotes": relay_notes, "wechat": wechat, "easyTierPeersNote": "peer 数量由 Electron 调用 easytier-cli peer list 获取", }