b77a8cadb2
Some checks failed
Upstream Sync / Sync latest commits from upstream repo (push) Has been cancelled
256 lines
8.5 KiB
Python
256 lines
8.5 KiB
Python
"""
|
||
工作台 /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 获取",
|
||
}
|