273 lines
9.0 KiB
Python
273 lines
9.0 KiB
Python
"""
|
||
工作台 /client/overview：YouTube 频道摘要、微信绑定摘要。
|
||
|
||
多路转播语义：每条线路为「一个抖音直播间 ↔ 一个 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 _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 _relay_url_config_path_for_channel(config_dir: Path, channel_id: str) -> Path:
|
||
return config_dir / "relay_pro" / f"url_config.{_safe_channel_file_id(channel_id)}.ini"
|
||
|
||
|
||
def _display_channel_name(raw_name: str, display_index: int) -> str:
|
||
name = (raw_name or "").strip()
|
||
if not name or re.fullmatch(r"频道\s*\d+", name):
|
||
return f"频道 {display_index}"
|
||
return name
|
||
|
||
|
||
def _extract_douyin_urls(text: str) -> list[str]:
|
||
out: list[str] = []
|
||
seen: set[str] = set()
|
||
for raw in str(text or "").splitlines():
|
||
line = raw.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
m = re.search(r"https?://(?:live|v|www)\.douyin\.com/[^\s,，#;]+", line, re.I)
|
||
if m:
|
||
url = m.group(0).replace("http://", "https://", 1)
|
||
else:
|
||
m = re.search(r"(?:live|v|www)\.douyin\.com/[^\s,，#;]+", line, re.I)
|
||
if not m:
|
||
continue
|
||
url = "https://" + m.group(0)
|
||
key = _normalize_douyin_url(url)
|
||
if key and key not in seen:
|
||
seen.add(key)
|
||
out.append(url)
|
||
return out
|
||
|
||
|
||
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 ""
|
||
urls = _extract_douyin_urls(text)
|
||
return urls[0] if urls else ""
|
||
|
||
|
||
def _parse_youtube_channels_json(path: Path) -> tuple[list[dict[str, str]], list[str]]:
|
||
"""
|
||
解析 youtube_channels.json：空 YouTube key 的占位频道不算线路。
|
||
源站地址以 relay_pro/url_config.<channel_id>.ini 为准，兼容旧字段 douyinUrl。
|
||
"""
|
||
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 数组"]
|
||
|
||
config_dir = path.parent
|
||
parsed: list[tuple[str, list[str], str]] = []
|
||
for item in arr:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
cid = str(item.get("id") or "").strip()
|
||
key = str(item.get("youtubeKey") or item.get("key") or item.get("stream_key") or "").strip()
|
||
if not key:
|
||
continue
|
||
name = _display_channel_name(str(item.get("name") or item.get("id") or ""), len(parsed) + 1)
|
||
source_texts = [
|
||
str(item.get("douyinUrl") or item.get("douyin_url") or item.get("douyin") or item.get("url") or "")
|
||
]
|
||
if cid:
|
||
p = _relay_url_config_path_for_channel(config_dir, cid)
|
||
if p.exists():
|
||
try:
|
||
source_texts.append(p.read_text(encoding="utf-8", errors="ignore"))
|
||
except Exception:
|
||
pass
|
||
urls = _extract_douyin_urls("\n".join(source_texts))
|
||
parsed.append((name, urls, key))
|
||
|
||
seen_d: dict[str, str] = {}
|
||
seen_k: dict[str, str] = {}
|
||
ok_strict: set[str] = set()
|
||
for name, urls, key in parsed:
|
||
nk = _normalize_stream_key(key)
|
||
if not nk:
|
||
errors.append(f"条目「{name}」密钥无效")
|
||
continue
|
||
if nk in seen_k:
|
||
errors.append(
|
||
f"禁止多条抖音线路共用同一 YouTube 串流密钥（冲突：「{seen_k[nk]}」与「{name}」）"
|
||
)
|
||
continue
|
||
for durl in urls:
|
||
nd = _normalize_douyin_url(durl)
|
||
if not nd:
|
||
continue
|
||
if nd in seen_d:
|
||
errors.append(
|
||
f"禁止重复使用同一抖音直播间绑定多条线路（冲突：「{seen_d[nd]}」与「{name}」）"
|
||
)
|
||
continue
|
||
seen_d[nd] = name
|
||
seen_k[nk] = name
|
||
ok_strict.add(name)
|
||
|
||
out: list[dict[str, str]] = []
|
||
for name, urls, key in parsed:
|
||
if urls:
|
||
preview = _short_douyin_hint(urls[0])
|
||
if len(urls) > 1:
|
||
preview = f"{preview}（{len(urls)}条）"
|
||
else:
|
||
preview = "（待填写）"
|
||
out.append(
|
||
{
|
||
"name": name,
|
||
"douyinPreview": preview,
|
||
"keyPreview": _mask_key(key),
|
||
}
|
||
)
|
||
|
||
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,
|
||
}
|