216 lines
7.9 KiB
Python
216 lines
7.9 KiB
Python
"""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,
|
||
}
|