#!/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()