614 lines
22 KiB
Python
614 lines
22 KiB
Python
"""
|
||
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
|
||
|
||
# 与 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=）。"""
|
||
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
|
||
|
||
|
||
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"
|