Files
gitlab-instance-0a899031_d2…/web2_relay_pro.py
2026-05-20 21:13:19 -05:00

631 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
Pro 多路转播:每频道独立 URL 配置文件、跨频道抖音地址去重、同步到 legacy youtube.ini + URL_config.ini。
"""
from __future__ import annotations
import configparser
import json
import re
import time
from pathlib import Path
from typing import Any, Callable, Awaitable
from web2_youtube_ini import parse_youtube_ini, serialize_youtube_ini
# 与 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 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"
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 _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": 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 无效"
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
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["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, "未找到该频道"
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=)。"""
config_dir = Path(config_dir)
ini = config_dir / "youtube.ini"
options: dict[str, str] = {}
header = ""
if ini.exists():
try:
parsed = parse_youtube_ini(ini.read_text(encoding="utf-8-sig"))
raw_options = parsed.get("options")
if isinstance(raw_options, dict):
options = {str(k): str(v) for k, v in raw_options.items()}
raw_header = parsed.get("header")
if isinstance(raw_header, str):
header = raw_header
except Exception:
options = {}
header = ""
ini.write_text(serialize_youtube_ini([key.strip()], 0, options, header), 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
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_legacy: str,
) -> dict[str, Any]:
"""工作台 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 []}
channels = ch_doc.get("channels") or []
recent = ""
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]] = []
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": any_online,
}
)
return {
"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:
return "", ""
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
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 "")
if script.endswith(".py") and Path(script).stem.startswith("youtube"):
return Path(script).stem
return "youtube2"