488 lines
18 KiB
Python
488 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
import pymysql
|
|
import requests
|
|
|
|
|
|
DETAIL_URL_TEMPLATE = "https://json.ncctrials.com/room/{room_id}/detail.json?v={ts}"
|
|
LIVE_ROOMS_URL_TEMPLATE = "https://json.ncctrials.com/all_live_rooms.json?v={ts}"
|
|
YUYAN_SOURCE_SITE = "yuyan"
|
|
YUYAN_SOURCE_DOMAIN = "https://yuyantv.cn"
|
|
YUYAN_ROOM_URL_TEMPLATE = YUYAN_SOURCE_DOMAIN + "/pages/liveRoom.html?roomNum={room_num}"
|
|
FETCH_BATCH_SIZE = 30
|
|
FETCH_TIMEOUT = 15
|
|
FETCH_ERROR_LIMIT = 500
|
|
ROOM_ID_RE = re.compile(r"(?:/room/|[?&]roomNum=)(\d+)")
|
|
DETAIL_JSONP_RE = re.compile(r"detail\((\{.*\})\)\s*;?\s*$", re.S)
|
|
LIVE_ROOMS_JSONP_RE = re.compile(r"all_live_rooms\((\{.*\})\)\s*;?\s*$", re.S)
|
|
DEFAULT_HEADERS = {
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/137.0.0.0 Safari/537.36"
|
|
),
|
|
"Accept": "application/json,text/javascript,*/*;q=0.1",
|
|
"Referer": "https://yuyantv.cn/liveType.html",
|
|
}
|
|
|
|
|
|
def _should_use_internal_lock() -> bool:
|
|
flag = str(os.environ.get("DQD_RUNNING_UNDER_TASK_RUNNER") or "").strip().lower()
|
|
return flag not in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def parse_match_live_detail_html(body: str) -> dict:
|
|
match = DETAIL_JSONP_RE.search((body or "").strip())
|
|
if not match:
|
|
raise ValueError("detail jsonp not found")
|
|
|
|
payload = json.loads(match.group(1))
|
|
stream = payload.get("data", {}).get("stream", {})
|
|
parsed = {
|
|
"play_url_flv": str(stream.get("flv") or "").strip(),
|
|
"play_url_hd_flv": str(stream.get("hdFlv") or "").strip(),
|
|
"play_url_m3u8": str(stream.get("m3u8") or "").strip(),
|
|
"play_url_hd_m3u8": str(stream.get("hdM3u8") or "").strip(),
|
|
}
|
|
if not any(parsed.values()):
|
|
raise ValueError("stream urls empty")
|
|
return parsed
|
|
|
|
|
|
def parse_yuyan_live_rooms_jsonp(body: str) -> List[Dict[str, Any]]:
|
|
match = LIVE_ROOMS_JSONP_RE.search((body or "").strip())
|
|
if not match:
|
|
raise ValueError("live rooms jsonp not found")
|
|
|
|
payload = json.loads(match.group(1))
|
|
if _safe_int(payload.get("code"), default=0) != 200:
|
|
raise ValueError(str(payload.get("msg") or "live rooms request failed"))
|
|
|
|
data = payload.get("data") or {}
|
|
groups = data.values() if isinstance(data, dict) else data
|
|
rooms: List[Dict[str, Any]] = []
|
|
seen_room_nums = set()
|
|
for group in groups:
|
|
if not isinstance(group, list):
|
|
continue
|
|
for room in group:
|
|
if not isinstance(room, dict):
|
|
continue
|
|
room_num = str(room.get("roomNum") or "").strip()
|
|
if not room_num or room_num in seen_room_nums:
|
|
continue
|
|
if _safe_int(room.get("liveStatus"), default=1) != 1:
|
|
continue
|
|
normalized = dict(room)
|
|
normalized["roomNum"] = room_num
|
|
rooms.append(normalized)
|
|
seen_room_nums.add(room_num)
|
|
return rooms
|
|
|
|
|
|
def choose_next_fetch_at(now_ts: int, has_existing_stream: bool) -> int:
|
|
return int(now_ts) + (600 if has_existing_stream else 60)
|
|
|
|
|
|
def _safe_int(value: Any, default: int = 0) -> int:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def _trim_error(message: Any) -> str:
|
|
return str(message or "").strip()[:FETCH_ERROR_LIMIT]
|
|
|
|
|
|
def _normalize_text(value: Any) -> str:
|
|
return re.sub(r"\s+", " ", str(value or "")).strip()
|
|
|
|
|
|
def _limit_text(value: Any, limit: int) -> str:
|
|
return _normalize_text(value)[:limit]
|
|
|
|
|
|
def _extract_anchor_info(room: Dict[str, Any]) -> Dict[str, Any]:
|
|
anchor = room.get("anchor") if isinstance(room.get("anchor"), dict) else {}
|
|
grow_dto = anchor.get("growDto") if isinstance(anchor.get("growDto"), dict) else {}
|
|
avatar = str(anchor.get("cutOutIcon") or anchor.get("icon") or "").strip()
|
|
return {
|
|
"anchor_id": _safe_int(anchor.get("uid"), default=0),
|
|
"anchor_name": _limit_text(anchor.get("nickName"), 100),
|
|
"anchor_avatar": avatar[:500],
|
|
"anchor_level": _limit_text(grow_dto.get("name"), 50),
|
|
"room_view_count": _safe_int(room.get("viewCount"), default=0),
|
|
"room_focus_count": _safe_int(room.get("focusCount"), default=0),
|
|
"room_notice": _limit_text(room.get("notice"), 500),
|
|
"room_detail": _limit_text(room.get("detail"), 255),
|
|
}
|
|
|
|
|
|
def _build_yuyan_room_source_url(room_num: Any) -> str:
|
|
return YUYAN_ROOM_URL_TEMPLATE.format(room_num=str(room_num or "").strip())
|
|
|
|
|
|
def _derive_room_id(source_url: str) -> str:
|
|
match = ROOM_ID_RE.search(str(source_url or "").strip())
|
|
if not match:
|
|
raise ValueError("room id not found in source_url")
|
|
return match.group(1)
|
|
|
|
|
|
def _build_detail_url(source_url: str, now_ts: int) -> str:
|
|
room_id = _derive_room_id(source_url)
|
|
return DETAIL_URL_TEMPLATE.format(room_id=room_id, ts=int(now_ts))
|
|
|
|
|
|
def _has_existing_stream(row: Dict[str, Any]) -> bool:
|
|
return any(
|
|
str(row.get(field) or "").strip()
|
|
for field in ("play_url_m3u8", "play_url_hd_m3u8", "play_url_flv", "play_url_hd_flv")
|
|
)
|
|
|
|
|
|
def _fetch_stream_payload(source_url: str, now_ts: int) -> Dict[str, str]:
|
|
detail_url = _build_detail_url(source_url, now_ts)
|
|
response = requests.get(detail_url, timeout=FETCH_TIMEOUT, headers=DEFAULT_HEADERS)
|
|
response.raise_for_status()
|
|
return parse_match_live_detail_html(response.text)
|
|
|
|
|
|
def _fetch_yuyan_live_rooms(now_ts: int) -> List[Dict[str, Any]]:
|
|
url = LIVE_ROOMS_URL_TEMPLATE.format(ts=int(now_ts))
|
|
response = requests.get(url, timeout=FETCH_TIMEOUT, headers=DEFAULT_HEADERS)
|
|
response.raise_for_status()
|
|
return parse_yuyan_live_rooms_jsonp(response.text)
|
|
|
|
|
|
def _build_run_result(
|
|
candidate_count: int,
|
|
success_count: int,
|
|
failed_count: int,
|
|
room_count: int = 0,
|
|
upserted_room_count: int = 0,
|
|
stale_room_count: int = 0,
|
|
) -> Dict[str, Any]:
|
|
summary_text = (
|
|
f"赛事直播流与主播信息抓取完成: 同步直播间 {room_count} 个, 入库 {upserted_room_count} 个, 下播 {stale_room_count} 个, 待处理 {candidate_count} 条, "
|
|
f"成功 {success_count} 条, 失败 {failed_count} 条"
|
|
)
|
|
result = {
|
|
"success": failed_count == 0,
|
|
"room_count": room_count,
|
|
"upserted_room_count": upserted_room_count,
|
|
"stale_room_count": stale_room_count,
|
|
"candidate_count": candidate_count,
|
|
"saved_count": success_count,
|
|
"count": success_count,
|
|
"failed_count": failed_count,
|
|
"summary_text": summary_text,
|
|
}
|
|
if failed_count > 0:
|
|
result["error"] = f"赛事直播流与主播信息抓取存在失败线路: 失败 {failed_count} 条"
|
|
return result
|
|
|
|
|
|
class TaskFileLock:
|
|
def __init__(self, name: str, lock_dir: Path | None = None):
|
|
safe_name = "".join(ch if ch.isalnum() or ch in ("-", "_") else "_" for ch in str(name or "task"))
|
|
self.lock_dir = Path(lock_dir or os.environ.get("DQD_LOCK_DIR") or (Path(__file__).resolve().parent / "data" / "locks"))
|
|
self.path = self.lock_dir / f"{safe_name}.lock"
|
|
self._fp = None
|
|
self._locked = False
|
|
|
|
def acquire(self) -> bool:
|
|
self.lock_dir.mkdir(parents=True, exist_ok=True)
|
|
self._fp = open(self.path, "a+b")
|
|
try:
|
|
self._fp.seek(0)
|
|
if os.name == "nt":
|
|
import msvcrt
|
|
|
|
msvcrt.locking(self._fp.fileno(), msvcrt.LK_NBLCK, 1)
|
|
else:
|
|
import fcntl
|
|
|
|
fcntl.flock(self._fp.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
self._locked = True
|
|
self._fp.seek(0)
|
|
self._fp.truncate()
|
|
self._fp.write(f"pid={os.getpid()} started_at={int(time.time())}\n".encode("utf-8"))
|
|
self._fp.flush()
|
|
return True
|
|
except (BlockingIOError, OSError):
|
|
self.release()
|
|
return False
|
|
|
|
def release(self) -> None:
|
|
if not self._fp:
|
|
return
|
|
try:
|
|
if self._locked and os.name == "nt":
|
|
import msvcrt
|
|
|
|
self._fp.seek(0)
|
|
msvcrt.locking(self._fp.fileno(), msvcrt.LK_UNLCK, 1)
|
|
elif self._locked:
|
|
import fcntl
|
|
|
|
fcntl.flock(self._fp.fileno(), fcntl.LOCK_UN)
|
|
finally:
|
|
self._fp.close()
|
|
self._fp = None
|
|
self._locked = False
|
|
|
|
|
|
@dataclass
|
|
class MatchLiveStreamRepository:
|
|
db_config: Dict[str, Any]
|
|
|
|
def __post_init__(self) -> None:
|
|
self.prefix = str(self.db_config.get("prefix") or "la_")
|
|
self.conn = pymysql.connect(
|
|
host=self.db_config["host"],
|
|
port=int(self.db_config["port"]),
|
|
user=self.db_config["user"],
|
|
password=self.db_config["password"],
|
|
database=self.db_config["database"],
|
|
charset=self.db_config.get("charset") or "utf8mb4",
|
|
cursorclass=pymysql.cursors.DictCursor,
|
|
autocommit=False,
|
|
)
|
|
|
|
def close(self) -> None:
|
|
self.conn.close()
|
|
|
|
def load_due_rows(self, now_ts: int, limit: int = FETCH_BATCH_SIZE) -> List[Dict[str, Any]]:
|
|
sql = (
|
|
f"SELECT id, match_id, title, source_url, play_url_flv, play_url_hd_flv, play_url_m3u8, play_url_hd_m3u8 "
|
|
f"FROM `{self.prefix}match_live` "
|
|
f"WHERE `delete_time` IS NULL AND `next_fetch_at` <= %s AND `play_type` <> 'html' "
|
|
f"ORDER BY `next_fetch_at` ASC, `id` ASC LIMIT %s"
|
|
)
|
|
with self.conn.cursor() as cur:
|
|
cur.execute(sql, (int(now_ts), int(limit)))
|
|
return list(cur.fetchall() or [])
|
|
|
|
def upsert_live_rooms(self, rooms: List[Dict[str, Any]], now_ts: int) -> int:
|
|
sql = (
|
|
f"INSERT INTO `{self.prefix}match_live` ("
|
|
"`match_id`, `title`, `source_url`, `play_type`, `iframe_url`, "
|
|
"`source_site`, `source_domain`, `source_match_key`, `source_line_key`, `source_is_hot`, `source_match_time`, "
|
|
"`anchor_id`, `anchor_name`, `anchor_avatar`, `anchor_level`, "
|
|
"`room_view_count`, `room_focus_count`, `room_notice`, `room_detail`, "
|
|
"`play_url_m3u8`, `play_url_hd_m3u8`, `play_url_flv`, `play_url_hd_flv`, "
|
|
"`fetch_status`, `fetch_error`, `last_fetch_at`, `next_fetch_at`, `create_time`, `update_time`, `delete_time`"
|
|
") VALUES ("
|
|
+ ", ".join(["%s"] * 30)
|
|
+ ") ON DUPLICATE KEY UPDATE "
|
|
"`title`=VALUES(`title`), "
|
|
"`play_type`=VALUES(`play_type`), "
|
|
"`iframe_url`=VALUES(`iframe_url`), "
|
|
"`source_site`=VALUES(`source_site`), "
|
|
"`source_domain`=VALUES(`source_domain`), "
|
|
"`source_match_key`=VALUES(`source_match_key`), "
|
|
"`source_line_key`=VALUES(`source_line_key`), "
|
|
"`source_is_hot`=VALUES(`source_is_hot`), "
|
|
"`source_match_time`=VALUES(`source_match_time`), "
|
|
"`anchor_id`=VALUES(`anchor_id`), "
|
|
"`anchor_name`=VALUES(`anchor_name`), "
|
|
"`anchor_avatar`=VALUES(`anchor_avatar`), "
|
|
"`anchor_level`=VALUES(`anchor_level`), "
|
|
"`room_view_count`=VALUES(`room_view_count`), "
|
|
"`room_focus_count`=VALUES(`room_focus_count`), "
|
|
"`room_notice`=VALUES(`room_notice`), "
|
|
"`room_detail`=VALUES(`room_detail`), "
|
|
"`next_fetch_at`=VALUES(`next_fetch_at`), "
|
|
"`update_time`=VALUES(`update_time`), "
|
|
"`delete_time`=NULL"
|
|
)
|
|
saved_count = 0
|
|
with self.conn.cursor() as cur:
|
|
for room in rooms:
|
|
room_num = str(room.get("roomNum") or "").strip()
|
|
if not room_num:
|
|
continue
|
|
title = _limit_text(room.get("title"), 255) or f"雨燕直播间 {room_num}"
|
|
anchor_info = _extract_anchor_info(room)
|
|
params = (
|
|
0,
|
|
title,
|
|
_build_yuyan_room_source_url(room_num),
|
|
"stream",
|
|
"",
|
|
YUYAN_SOURCE_SITE,
|
|
YUYAN_SOURCE_DOMAIN,
|
|
f"room:{room_num}",
|
|
room_num[:100],
|
|
1 if _safe_int(room.get("markType"), default=0) > 0 else 0,
|
|
0,
|
|
anchor_info["anchor_id"],
|
|
anchor_info["anchor_name"],
|
|
anchor_info["anchor_avatar"],
|
|
anchor_info["anchor_level"],
|
|
anchor_info["room_view_count"],
|
|
anchor_info["room_focus_count"],
|
|
anchor_info["room_notice"],
|
|
anchor_info["room_detail"],
|
|
"",
|
|
"",
|
|
"",
|
|
"",
|
|
0,
|
|
"",
|
|
0,
|
|
int(now_ts),
|
|
int(now_ts),
|
|
int(now_ts),
|
|
None,
|
|
)
|
|
cur.execute(sql, params)
|
|
saved_count += 1
|
|
if saved_count > 0:
|
|
self.conn.commit()
|
|
return saved_count
|
|
|
|
def cleanup_stale_live_rooms(self, active_room_nums: List[str], now_ts: int) -> int:
|
|
active_room_nums = [str(room_num).strip() for room_num in active_room_nums if str(room_num or "").strip()]
|
|
params: List[Any] = [
|
|
"直播间已下播或不在雨燕直播列表",
|
|
int(now_ts),
|
|
choose_next_fetch_at(now_ts, has_existing_stream=True),
|
|
int(now_ts),
|
|
int(now_ts),
|
|
YUYAN_SOURCE_SITE,
|
|
]
|
|
where = [
|
|
"`delete_time` IS NULL",
|
|
"`source_site`=%s",
|
|
"`play_type`='stream'",
|
|
]
|
|
if active_room_nums:
|
|
placeholders = ", ".join(["%s"] * len(active_room_nums))
|
|
where.append(f"`source_line_key` NOT IN ({placeholders})")
|
|
params.extend(active_room_nums)
|
|
|
|
sql = (
|
|
f"UPDATE `{self.prefix}match_live` SET "
|
|
"`fetch_status`=2, `fetch_error`=%s, `last_fetch_at`=%s, `next_fetch_at`=%s, "
|
|
"`update_time`=%s, `delete_time`=%s "
|
|
"WHERE " + " AND ".join(where)
|
|
)
|
|
with self.conn.cursor() as cur:
|
|
cur.execute(sql, tuple(params))
|
|
stale_count = int(getattr(cur, "rowcount", 0) or 0)
|
|
if stale_count > 0:
|
|
self.conn.commit()
|
|
return stale_count
|
|
|
|
def mark_success(self, row_id: int, stream_urls: Dict[str, str], now_ts: int) -> None:
|
|
sql = (
|
|
f"UPDATE `{self.prefix}match_live` SET "
|
|
"`play_url_flv`=%s, `play_url_hd_flv`=%s, `play_url_m3u8`=%s, `play_url_hd_m3u8`=%s, "
|
|
"`fetch_status`=1, `fetch_error`='', `last_fetch_at`=%s, `next_fetch_at`=%s, `update_time`=%s "
|
|
"WHERE `id`=%s"
|
|
)
|
|
next_fetch_at = choose_next_fetch_at(now_ts, has_existing_stream=True)
|
|
with self.conn.cursor() as cur:
|
|
cur.execute(
|
|
sql,
|
|
(
|
|
stream_urls["play_url_flv"],
|
|
stream_urls["play_url_hd_flv"],
|
|
stream_urls["play_url_m3u8"],
|
|
stream_urls["play_url_hd_m3u8"],
|
|
int(now_ts),
|
|
next_fetch_at,
|
|
int(now_ts),
|
|
int(row_id),
|
|
),
|
|
)
|
|
self.conn.commit()
|
|
|
|
def mark_failed(self, row_id: int, error_message: str, now_ts: int, has_existing_stream: bool) -> None:
|
|
sql = (
|
|
f"UPDATE `{self.prefix}match_live` SET "
|
|
"`fetch_status`=2, `fetch_error`=%s, `last_fetch_at`=%s, `next_fetch_at`=%s, `update_time`=%s "
|
|
"WHERE `id`=%s"
|
|
)
|
|
with self.conn.cursor() as cur:
|
|
cur.execute(
|
|
sql,
|
|
(
|
|
_trim_error(error_message),
|
|
int(now_ts),
|
|
choose_next_fetch_at(now_ts, has_existing_stream=has_existing_stream),
|
|
int(now_ts),
|
|
int(row_id),
|
|
),
|
|
)
|
|
self.conn.commit()
|
|
|
|
|
|
def run(db_config: Dict[str, Any]) -> Dict[str, Any]:
|
|
lock = None
|
|
if _should_use_internal_lock():
|
|
lock = TaskFileLock("match_live_stream")
|
|
if not lock.acquire():
|
|
return {
|
|
"success": True,
|
|
"candidate_count": 0,
|
|
"saved_count": 0,
|
|
"summary_text": "match_live_stream 已在执行中，跳过本次",
|
|
}
|
|
|
|
repo = MatchLiveStreamRepository(db_config)
|
|
now_ts = int(time.time())
|
|
success_count = 0
|
|
failed_count = 0
|
|
room_count = 0
|
|
upserted_room_count = 0
|
|
stale_room_count = 0
|
|
rows: List[Dict[str, Any]] = []
|
|
|
|
try:
|
|
live_rooms = _fetch_yuyan_live_rooms(now_ts)
|
|
room_count = len(live_rooms)
|
|
upserted_room_count = repo.upsert_live_rooms(live_rooms, now_ts)
|
|
stale_room_count = repo.cleanup_stale_live_rooms(
|
|
[str(room.get("roomNum") or "").strip() for room in live_rooms],
|
|
now_ts,
|
|
)
|
|
fetch_limit = max(FETCH_BATCH_SIZE, room_count)
|
|
rows = repo.load_due_rows(now_ts, limit=fetch_limit)
|
|
for row in rows:
|
|
fetch_ts = int(time.time())
|
|
has_existing_stream = _has_existing_stream(row)
|
|
try:
|
|
stream_urls = _fetch_stream_payload(str(row.get("source_url") or ""), fetch_ts)
|
|
repo.mark_success(_safe_int(row.get("id")), stream_urls, fetch_ts)
|
|
success_count += 1
|
|
except Exception as exc: # noqa: BLE001
|
|
repo.mark_failed(_safe_int(row.get("id")), str(exc), fetch_ts, has_existing_stream)
|
|
failed_count += 1
|
|
|
|
candidate_count = len(rows)
|
|
return _build_run_result(
|
|
candidate_count,
|
|
success_count,
|
|
failed_count,
|
|
room_count=room_count,
|
|
upserted_room_count=upserted_room_count,
|
|
stale_room_count=stale_room_count,
|
|
)
|
|
finally:
|
|
repo.close()
|
|
if lock is not None:
|
|
lock.release()
|