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

优化直播控制台：引入进程监控轻量轮询，并拆分 Live UI 组件

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
eric
2026-07-09 02:47:33 -05:00
parent 5f1b9db2dc
commit a0104f6517
16 changed files with 1969 additions and 461 deletions
Download Patch File Download Diff File Expand all files Collapse all files

215
web2_douyin_source_display.py Normal file
View File

@@ -0,0 +1,215 @@
"""Douyin source display from URL list + relay log (ported from sh2/src/douyin_source_display.py)."""
from __future__ import annotations
import re
from typing import Any
_DOUYIN_LIVE_URL_RE = re.compile(r"https?://(?:www\.)?live\.douyin\.com/([a-zA-Z0-9_-]+)", re.IGNORECASE)
_DOUYIN_LIVE_ROOM_RE = re.compile(r"(?:https?://(?:www\.)?)?live\.douyin\.com/([a-zA-Z0-9_-]+)", re.IGNORECASE)
_STATUS_TITLE_RE = re.compile(
r"(?:序号\s*\d+\s*)?(.{1,80}?)\s*(?:等待直播|正在直播|直播中|已开播|开始推流|主播已下播|下播|录制中|推流中)",
re.IGNORECASE,
)
_RECORDING_TITLE_RE = re.compile(r"(?:正在录制|录制中)\s*\d*\s*个?\s*[:：]\s*([^\[\r\n]{1,80})", re.IGNORECASE)
_LIVE_TITLE_RE = re.compile(
r"(?:序号\s*\d+\s*)?(.{1,80}?)\s*(?:正在直播中|正在直播|直播中|已开播|开始推流|录制中|推流中)",
re.IGNORECASE,
)
_WAIT_TITLE_RE = re.compile(r"(?:序号\s*\d+\s*)?(.{1,80}?)\s*(?:等待直播|主播已下播|下播)", re.IGNORECASE)
_LABEL_TITLE_RE = re.compile(r"(?:主播|房间名|直播间|抖音源|源站|标题)\s*[:：]\s*(.{1,80})", re.IGNORECASE)
_BAD_TITLE_RE = re.compile(r"^\s*(?:-|—|_|无|未知|unknown|null|undefined|none|nan)?\s*$", re.IGNORECASE)
_ERROR_CONTEXT_RE = re.compile(r"(?:获取失败|错误|异常|error|failed|exception)", re.IGNORECASE)
def _clean_source_title(value: str) -> str:
text = (value or "").strip().strip("\"'`[]()（）【】")
text = re.sub(r"^\[[A-Z]+\]\s*", "", text, flags=re.IGNORECASE)
text = re.sub(r"^序号\s*\d+\s*", "", text)
text = re.sub(r"^(?:主播|房间名|直播间|抖音源|源站|标题)\s*[:：]\s*", "", text, flags=re.IGNORECASE)
text = _DOUYIN_LIVE_ROOM_RE.sub("", text)
text = re.sub(r"(?:等待直播|正在直播|直播中|已开播|开始推流|主播已下播|下播|录制中|推流中).*$", "", text).strip()
text = text.strip(" ,，;；|｜-—:：")
if _BAD_TITLE_RE.match(text):
return ""
if "live.douyin.com" in text.lower():
return ""
if re.fullmatch(r"\d{5,}", text):
return ""
if text.startswith(("监测", "格式", "错误", "无录制任务")):
return ""
return text[:48]
def _url_entries_from_lines(url_lines: str) -> list[dict[str, str]]:
entries: list[dict[str, str]] = []
for line in (url_lines or "").splitlines():
raw = line.strip()
if not raw or raw.startswith("#"):
continue
match = _DOUYIN_LIVE_ROOM_RE.search(raw)
if not match:
continue
room_id = match.group(1)
tail = raw[match.end() :].strip(" ,，;；|｜-—:：")
title = _clean_source_title(tail)
if not title:
parts = re.split(r"[,，|｜]", raw, maxsplit=1)
if len(parts) > 1:
title = _clean_source_title(parts[1])
entries.append(
{
"room_id": room_id,
"source_url": f"https://live.douyin.com/{room_id}",
"source_title": title,
}
)
return entries
def _log_indicates_active_relay(recent_log: str) -> bool:
text = (recent_log or "").lower()
if "frame=" in text and "speed=" in text:
return True
return bool(_RECORDING_TITLE_RE.search(recent_log or "")) or bool(_LIVE_TITLE_RE.search(recent_log or ""))
def _title_from_waiting_log_only(recent_log: str) -> str:
lines = [line.strip() for line in (recent_log or "").splitlines() if line.strip()]
for line in reversed(lines[-240:]):
if _ERROR_CONTEXT_RE.search(line):
continue
match = _WAIT_TITLE_RE.search(line) or _STATUS_TITLE_RE.search(line)
if not match:
continue
title = _clean_source_title(match.group(1))
if title:
return title
return ""
def _source_title_from_log(recent_log: str) -> str:
lines = [line.strip() for line in (recent_log or "").splitlines() if line.strip()]
tail = lines[-240:]
for pattern in (_RECORDING_TITLE_RE, _LIVE_TITLE_RE):
for line in reversed(tail):
match = pattern.search(line)
if not match:
continue
title = _clean_source_title(match.group(1))
if title:
return title
for line in reversed(tail):
if _ERROR_CONTEXT_RE.search(line):
continue
match = _LABEL_TITLE_RE.search(line)
if not match:
continue
title = _clean_source_title(match.group(1))
if title:
return title
for line in reversed(tail):
match = _WAIT_TITLE_RE.search(line) or _STATUS_TITLE_RE.search(line)
if not match:
continue
title = _clean_source_title(match.group(1))
if title:
return title
return ""
def _current_room_id_from_log(recent_log: str) -> str:
room_id = ""
for match in _DOUYIN_LIVE_ROOM_RE.finditer(recent_log or ""):
room_id = match.group(1)
return room_id
def _normalize_title_for_match(value: str) -> str:
return re.sub(r"\s+", "", _clean_source_title(value)).casefold()
def _entry_for_room(entries: list[dict[str, str]], room_id: str) -> dict[str, str] | None:
if not room_id:
return None
for entry in entries:
if entry.get("room_id") == room_id:
return entry
return None
def _entry_for_title(entries: list[dict[str, str]], title: str) -> dict[str, str] | None:
needle = _normalize_title_for_match(title)
if not needle:
return None
for entry in entries:
candidate = _normalize_title_for_match(entry.get("source_title", ""))
if not candidate:
continue
if candidate == needle:
return entry
if len(candidate) >= 3 and len(needle) >= 3 and (candidate in needle or needle in candidate):
return entry
return None
def extract_douyin_source_meta(url_lines: str = "", recent_log: str = "") -> dict[str, Any]:
entries = _url_entries_from_lines(url_lines)
log_room_id = _current_room_id_from_log(recent_log)
log_title = _source_title_from_log(recent_log)
matched_entry = _entry_for_title(entries, log_title) or _entry_for_room(entries, log_room_id)
source_ambiguous = False
source_current = False
if matched_entry:
room_id = matched_entry["room_id"]
source_url = matched_entry["source_url"]
title = log_title or matched_entry.get("source_title", "")
source_current = bool(log_room_id or log_title)
elif log_room_id:
room_id = log_room_id
source_url = f"https://live.douyin.com/{room_id}"
title = log_title
source_current = True
elif len(entries) == 1:
room_id = entries[0]["room_id"]
source_url = entries[0]["source_url"]
title = entries[0].get("source_title", "") or log_title
elif log_title:
room_id = ""
source_url = ""
title = log_title
source_ambiguous = bool(entries)
source_current = True
elif entries:
room_id = entries[0]["room_id"]
source_url = entries[0]["source_url"]
title = entries[0].get("source_title", "")
source_ambiguous = len(entries) > 1
else:
room_id = ""
source_url = ""
title = ""
if (
len(entries) > 1
and title
and not _log_indicates_active_relay(recent_log)
and title == _title_from_waiting_log_only(recent_log)
):
pinned = entries[0]
room_id = pinned["room_id"]
source_url = pinned["source_url"]
title = pinned.get("source_title", "") or title
source_ambiguous = True
source_current = False
display = title or (f"抖音直播间 {room_id}" if room_id else "")
return {
"source_title": title,
"source_room_id": room_id,
"source_url": source_url,
"source_display": display,
"source_ambiguous": source_ambiguous,
"source_current": source_current,
}