eric/gitlab-instance-0a899031_d2yexeapp
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

's'

Browse Source
This commit is contained in:
eric
2026-05-20 23:50:43 -05:00
parent 882ea9268c
commit e6e4f7a02e
17 changed files with 570 additions and 188 deletions
Download Patch File Download Diff File Expand all files Collapse all files

154
web2_client_overview.py
View File

@@ -52,6 +52,44 @@ def _normalize_stream_key(key: str) -> str:
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 ""
@@ -76,22 +114,14 @@ def _first_douyin_line_from_url_config(config_dir: Path) -> str:
text = p.read_text(encoding="utf-8", errors="ignore")
except Exception:
return ""
for raw in text.splitlines():
line = raw.strip()
if not line or line.startswith("#"):
continue
if "douyin" not in line.lower():
continue
if "," in line:
line = line.split(",")[-1].strip()
return line.strip()
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_channels.json：空 YouTube key 的占位频道不算线路。
源站地址以 relay_pro/url_config.<channel_id>.ini 为准，兼容旧字段 douyinUrl。
"""
errors: list[str] = []
try:
@@ -103,82 +133,70 @@ def _parse_youtube_channels_json(path: Path) -> tuple[list[dict[str, str]], list
if not isinstance(arr, list):
return [], ["youtube_channels.json 缺少 channels 数组"]
parsed: list[tuple[str, str, str]] = []
config_dir = path.parent
parsed: list[tuple[str, list[str], str]] = []
for item in arr:
if not isinstance(item, dict):
continue
name = str(item.get("name") or item.get("id") or "").strip() or "未命名"
durl = str(
item.get("douyinUrl") or item.get("douyin_url") or item.get("douyin") or item.get("url") or ""
).strip()
cid = str(item.get("id") or "").strip()
key = str(item.get("youtubeKey") or item.get("key") or item.get("stream_key") or "").strip()
parsed.append((name, durl, key))
# 仅对「两项均已填写」的条目做冲突与规范化校验
strict: list[tuple[str, str, str]] = []
for name, durl, key in parsed:
if durl and key:
strict.append((name, durl, key))
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, durl, key in strict:
nd = _normalize_douyin_url(durl)
for name, urls, key in parsed:
nk = _normalize_stream_key(key)
if not nd or not nk:
errors.append(f"条目「{name}」地址或密钥无效")
continue
if nd in seen_d:
errors.append(
f"禁止重复使用同一抖音直播间绑定多条线路（冲突：「{seen_d[nd]}」与「{name}」）"
)
if not nk:
errors.append(f"条目「{name}」密钥无效")
continue
if nk in seen_k:
errors.append(
f"禁止多条抖音线路共用同一 YouTube 串流密钥（冲突：「{seen_k[nk]}」与「{name}」）"
)
continue
seen_d[nd] = name
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, durl, key in parsed:
if not durl and not key:
out.append(
{
"name": name,
"douyinPreview": "（待填写）",
"keyPreview": "（待填写）",
}
)
continue
if not durl or not key:
out.append(
{
"name": name,
"douyinPreview": _short_douyin_hint(durl) if durl else "（待填写）",
"keyPreview": _mask_key(key) if key else "（待填写）",
}
)
continue
if name in ok_strict:
out.append(
{
"name": name,
"douyinPreview": _short_douyin_hint(durl),
"keyPreview": _mask_key(key),
}
)
for name, urls, key in parsed:
if urls:
preview = _short_douyin_hint(urls[0])
if len(urls) > 1:
preview = f"{preview}（{len(urls)}条）"
else:
out.append(
{
"name": name,
"douyinPreview": _short_douyin_hint(durl) if durl else "（待填写）",
"keyPreview": _mask_key(key) if key else "（待填写）",
}
)
preview = "（待填写）"
out.append(
{
"name": name,
"douyinPreview": preview,
"keyPreview": _mask_key(key),
}
)
return out, errors