128 lines
3.7 KiB
Python
128 lines
3.7 KiB
Python
"""
|
|
youtube.ini 结构化读写：多行 key=（注释行为备用密钥）、其它键值对保留。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
|
|
def norm_stream_key(key: str) -> str:
|
|
k = (key or "").strip()
|
|
if not k:
|
|
return ""
|
|
if k.lower().startswith(("rtmp://", "rtmps://")):
|
|
parts = k.rstrip("/").split("/")
|
|
k = parts[-1] if parts else k
|
|
return k.strip().lower()
|
|
|
|
|
|
def validate_youtube_keys(keys: list[str]) -> tuple[bool, str]:
|
|
cleaned = [str(x).strip() for x in keys if str(x).strip()]
|
|
if not cleaned:
|
|
return False, "至少需要一个有效的 YouTube 串流密钥"
|
|
seen: set[str] = set()
|
|
for key in cleaned:
|
|
nk = norm_stream_key(key)
|
|
if not nk:
|
|
continue
|
|
if nk in seen:
|
|
return False, "同一配置内不能重复使用相同的 YouTube 串流密钥"
|
|
seen.add(nk)
|
|
return True, ""
|
|
|
|
|
|
def parse_youtube_ini(content: str) -> dict[str, Any]:
|
|
if not (content or "").strip():
|
|
return {
|
|
"keys": [""],
|
|
"activeIndex": 0,
|
|
"options": {},
|
|
"header": "",
|
|
}
|
|
m = re.search(r"(?im)^\s*\[youtube\]\s*$", content)
|
|
if not m:
|
|
return {
|
|
"keys": [""],
|
|
"activeIndex": 0,
|
|
"options": {},
|
|
"header": content,
|
|
}
|
|
header = content[: m.start()]
|
|
body = content[m.end() :]
|
|
keys: list[str] = []
|
|
key_commented: list[bool] = []
|
|
options: dict[str, str] = {}
|
|
active_index = 0
|
|
found_active = False
|
|
for raw in body.splitlines():
|
|
stripped = raw.strip()
|
|
if not stripped:
|
|
continue
|
|
if stripped.startswith("[") and stripped.endswith("]"):
|
|
break
|
|
km = re.match(r"^;\s*(key|stream_key)\s*=\s*(.+)$", stripped, re.I)
|
|
if km:
|
|
val = km.group(2).strip().split("#")[0].strip()
|
|
keys.append(val)
|
|
key_commented.append(True)
|
|
continue
|
|
km2 = re.match(r"^(key|stream_key)\s*=\s*(.+)$", stripped, re.I)
|
|
if km2:
|
|
val = km2.group(2).strip().split("#")[0].strip()
|
|
keys.append(val)
|
|
key_commented.append(False)
|
|
if not found_active:
|
|
active_index = len(keys) - 1
|
|
found_active = True
|
|
continue
|
|
if stripped.startswith(";"):
|
|
continue
|
|
om = re.match(r"^(\w+)\s*=\s*(.+)$", stripped)
|
|
if om:
|
|
k = om.group(1).strip()
|
|
v = om.group(2).strip().split("#")[0].strip()
|
|
if k.lower() not in ("key", "stream_key"):
|
|
options[k] = v
|
|
if not keys:
|
|
keys = [""]
|
|
active_index = 0
|
|
elif not found_active:
|
|
active_index = 0
|
|
return {
|
|
"keys": keys,
|
|
"activeIndex": max(0, min(active_index, len(keys) - 1)),
|
|
"options": options,
|
|
"header": header,
|
|
}
|
|
|
|
|
|
def serialize_youtube_ini(
|
|
keys: list[str],
|
|
active_index: int,
|
|
options: dict[str, str],
|
|
header: str,
|
|
) -> str:
|
|
lines: list[str] = []
|
|
h = (header or "").rstrip()
|
|
if h:
|
|
lines.append(h.rstrip("\n"))
|
|
if not lines[-1].endswith("\n"):
|
|
lines[-1] = lines[-1] + "\n"
|
|
lines.append("[youtube]")
|
|
if not keys:
|
|
keys = [""]
|
|
ai = max(0, min(int(active_index), len(keys) - 1))
|
|
for i, k in enumerate(keys):
|
|
val = (k or "").strip()
|
|
if i == ai:
|
|
lines.append(f"key = {val}")
|
|
else:
|
|
lines.append(f"; key = {val}")
|
|
for opt_k, opt_v in options.items():
|
|
if opt_k.lower() in ("key", "stream_key"):
|
|
continue
|
|
lines.append(f"{opt_k} = {opt_v}")
|
|
lines.append("")
|
|
return "\n".join(lines)
|