240 lines
8.0 KiB
Python
240 lines
8.0 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any, Iterable, Optional
|
|
|
|
_SAFE_PM2_SEGMENT = re.compile(r"[^a-zA-Z0-9_.-]+")
|
|
_YOUTUBE_PRO_PM2_PREFIX = "youtube2__"
|
|
_RESERVED_PM2_NAMES = frozenset({"douyin_youtube", "tiktok", "obs", "web"})
|
|
_YOUTUBE_HEADER = re.compile(r"^\s*\[youtube\]\s*$", re.IGNORECASE)
|
|
_ANY_HEADER = re.compile(r"^\s*\[[^\]]+\]\s*$")
|
|
_KEY_LINE = re.compile(r"^\s*(?:key|stream_key)\s*=", re.IGNORECASE)
|
|
_BROKEN_LANE_NAME_RE = re.compile(r"(?:\?|\uff1f){2,}|\ufffd|锟|绾胯矾|çº|è·|Â", re.IGNORECASE)
|
|
|
|
|
|
def sanitize_pm2_segment(value: str) -> str:
|
|
return _SAFE_PM2_SEGMENT.sub("_", (value or "").strip())
|
|
|
|
|
|
def default_youtube_pro_pm2_name(channel_id: str) -> str:
|
|
safe = sanitize_pm2_segment(channel_id)
|
|
return f"{_YOUTUBE_PRO_PM2_PREFIX}{safe or 'channel'}"
|
|
|
|
|
|
def normalize_youtube_pro_pm2_name(raw_value: str, channel_id: str) -> str:
|
|
raw = sanitize_pm2_segment(raw_value)
|
|
if not raw or raw in _RESERVED_PM2_NAMES or raw == _YOUTUBE_PRO_PM2_PREFIX:
|
|
return default_youtube_pro_pm2_name(channel_id)
|
|
if raw == "youtube":
|
|
return raw
|
|
if raw.startswith(_YOUTUBE_PRO_PM2_PREFIX):
|
|
return raw
|
|
return default_youtube_pro_pm2_name(raw)
|
|
|
|
|
|
def channel_pm2_name(channel: dict[str, Any]) -> str:
|
|
return normalize_youtube_pro_pm2_name(
|
|
str(channel.get("pm2Name") or channel.get("pm2_name") or "").strip(),
|
|
str(channel.get("id") or "").strip(),
|
|
)
|
|
|
|
|
|
def is_broken_youtube_pro_lane_name(value: str) -> bool:
|
|
s = (value or "").strip()
|
|
return not s or s.lower() in {"undefined", "null", "nan"} or bool(_BROKEN_LANE_NAME_RE.search(s))
|
|
|
|
|
|
def clean_youtube_pro_lane_name(value: str, index: int) -> str:
|
|
raw = (value or "").strip()
|
|
if not is_broken_youtube_pro_lane_name(raw):
|
|
return raw
|
|
match = re.search(r"\d+", raw)
|
|
lane_no = match.group(0) if match else str(max(0, index) + 1)
|
|
return f"线路 {lane_no}"
|
|
|
|
|
|
def normalize_youtube_pro_channels(channels: Iterable[Any]) -> list[dict[str, Any]]:
|
|
out: list[dict[str, Any]] = []
|
|
seen_ids: set[str] = set()
|
|
seen_pm2: set[str] = set()
|
|
for index, item in enumerate(channels):
|
|
if not isinstance(item, dict):
|
|
raise ValueError("each channel must be an object")
|
|
row = dict(item)
|
|
cid = str(row.get("id", "")).strip()
|
|
if not cid:
|
|
raise ValueError("channel.id required")
|
|
if cid in seen_ids:
|
|
raise ValueError(f"duplicate channel.id: {cid}")
|
|
seen_ids.add(cid)
|
|
|
|
row["id"] = cid
|
|
row["name"] = clean_youtube_pro_lane_name(str(row.get("name", "")), index)
|
|
pm2 = channel_pm2_name(row)
|
|
if not pm2:
|
|
raise ValueError(f"channel.pm2Name invalid for {cid}")
|
|
if pm2 in _RESERVED_PM2_NAMES:
|
|
raise ValueError(f"channel.pm2Name reserved: {pm2}")
|
|
if pm2 in seen_pm2:
|
|
raise ValueError(f"duplicate channel.pm2Name: {pm2}")
|
|
seen_pm2.add(pm2)
|
|
row["pm2Name"] = pm2
|
|
|
|
for key in ("streamKeySuffix", "streamKey", "urlLines", "notes"):
|
|
if key in row and row[key] is not None and not isinstance(row[key], str):
|
|
row[key] = str(row[key])
|
|
|
|
out.append(row)
|
|
return out
|
|
|
|
|
|
def youtube_pro_channel_for_pm2(channels: Iterable[dict[str, Any]], pm2: str) -> Optional[dict[str, Any]]:
|
|
target = (pm2 or "").strip()
|
|
if not target:
|
|
return None
|
|
for channel in channels:
|
|
if not isinstance(channel, dict):
|
|
continue
|
|
if channel_pm2_name(channel) == target:
|
|
return dict(channel)
|
|
return None
|
|
|
|
|
|
def channel_id_for_pm2_name(channels: Iterable[dict[str, Any]], pm2: str) -> str:
|
|
ch = youtube_pro_channel_for_pm2(channels, pm2)
|
|
return str(ch.get("id") or "").strip() if ch else ""
|
|
|
|
|
|
def _read_text_if_file(path: Path) -> Optional[str]:
|
|
try:
|
|
if path.is_file():
|
|
return path.read_text(encoding="utf-8")
|
|
except OSError:
|
|
return None
|
|
return None
|
|
|
|
|
|
def _normalize_url_lines(text: str) -> str:
|
|
raw = (text or "").replace("\r\n", "\n").replace("\r", "\n").lstrip("\ufeff")
|
|
if not raw.strip():
|
|
return ""
|
|
lines = [line.rstrip() for line in raw.split("\n")]
|
|
while lines and not lines[-1].strip():
|
|
lines.pop()
|
|
body = "\n".join(lines).strip()
|
|
return f"{body}\n" if body else ""
|
|
|
|
|
|
def _has_noncomment_url(text: str) -> bool:
|
|
return any(line.strip() and not line.strip().startswith("#") for line in (text or "").splitlines())
|
|
|
|
|
|
def _extract_stream_key(text: str) -> str:
|
|
in_youtube = False
|
|
for raw_line in (text or "").splitlines():
|
|
line = raw_line.strip()
|
|
if _YOUTUBE_HEADER.match(line):
|
|
in_youtube = True
|
|
continue
|
|
if _ANY_HEADER.match(line):
|
|
in_youtube = False
|
|
continue
|
|
if not in_youtube:
|
|
continue
|
|
if not line or line.startswith("#") or line.startswith(";"):
|
|
continue
|
|
if _KEY_LINE.match(line):
|
|
_, _, value = raw_line.partition("=")
|
|
return value.strip().split("#", 1)[0].split(";", 1)[0].strip()
|
|
return ""
|
|
|
|
|
|
def ensure_youtube_pro_channel_runtime_files(
|
|
config_dir: Path,
|
|
process: str,
|
|
channels: Iterable[dict[str, Any]],
|
|
) -> tuple[bool, list[str]]:
|
|
channel = youtube_pro_channel_for_pm2(channels, process)
|
|
if channel is None:
|
|
return False, []
|
|
|
|
cid = str(channel.get("id") or "").strip()
|
|
if not cid:
|
|
return False, []
|
|
|
|
pm2 = channel_pm2_name(channel)
|
|
from web2_relay_pro import (
|
|
materialize_relay_channel_youtube_ini,
|
|
url_config_path_for_channel,
|
|
youtube_ini_path_for_channel,
|
|
)
|
|
|
|
ok, msg = materialize_relay_channel_youtube_ini(config_dir, cid)
|
|
if not ok:
|
|
return True, [f"{pm2}: {msg}"]
|
|
|
|
url_path = url_config_path_for_channel(config_dir, cid)
|
|
seed_url = _read_text_if_file(url_path) or ""
|
|
effective_url = _normalize_url_lines(str(channel.get("urlLines") or "")) or _normalize_url_lines(seed_url)
|
|
douyin = str(channel.get("douyinUrl") or channel.get("douyin_url") or "").strip()
|
|
if not _has_noncomment_url(effective_url) and douyin:
|
|
effective_url = _normalize_url_lines(douyin)
|
|
|
|
errors: list[str] = []
|
|
yt_path = youtube_ini_path_for_channel(config_dir, cid)
|
|
youtube_text = _read_text_if_file(yt_path) or ""
|
|
if not _extract_stream_key(youtube_text):
|
|
errors.append(f"{pm2}: stream key missing")
|
|
if not _has_noncomment_url(effective_url):
|
|
errors.append(f"{pm2}: source URL missing")
|
|
if errors:
|
|
return True, errors
|
|
|
|
try:
|
|
url_path.parent.mkdir(parents=True, exist_ok=True)
|
|
url_path.write_text(effective_url, encoding="utf-8")
|
|
except OSError as exc:
|
|
return True, [f"{pm2}: failed to write url config ({exc})"]
|
|
return True, []
|
|
|
|
|
|
def validate_youtube_runtime_files(
|
|
config_dir: Path,
|
|
process: str,
|
|
channels: Iterable[dict[str, Any]],
|
|
) -> list[str]:
|
|
handled, errors = ensure_youtube_pro_channel_runtime_files(config_dir, process, channels)
|
|
if handled:
|
|
return errors
|
|
|
|
youtube_path = config_dir / "youtube.ini"
|
|
url_path = config_dir / "URL_config.ini"
|
|
youtube_text = _read_text_if_file(youtube_path) or ""
|
|
url_text = _read_text_if_file(url_path) or ""
|
|
|
|
fallback_errors: list[str] = []
|
|
if not _extract_stream_key(youtube_text):
|
|
fallback_errors.append(f"{process}: stream key missing")
|
|
if not _has_noncomment_url(url_text):
|
|
fallback_errors.append(f"{process}: source URL missing")
|
|
return fallback_errors
|
|
|
|
|
|
def sync_all_youtube_pro_runtime_files(
|
|
config_dir: Path,
|
|
channels: Iterable[dict[str, Any]],
|
|
) -> list[str]:
|
|
errors: list[str] = []
|
|
seen: set[str] = set()
|
|
for channel in channels:
|
|
if not isinstance(channel, dict):
|
|
continue
|
|
pm2 = channel_pm2_name(channel)
|
|
if not pm2 or pm2 in seen:
|
|
continue
|
|
seen.add(pm2)
|
|
_, row_errors = ensure_youtube_pro_channel_runtime_files(config_dir, pm2, channels)
|
|
errors.extend(row_errors)
|
|
return errors
|