From 33664adbe020d51b5e60be02b7ab0d89b40f538f Mon Sep 17 00:00:00 2001 From: hajimi Date: Fri, 3 Jul 2026 21:10:05 +0800 Subject: [PATCH] fix: sync yuyan anchor live stream task --- docker/crawler/config/crawler_tasks.yaml | 2 +- docker/crawler/match_live_stream.py | 227 +++++++++++++++++- docs/sql/alter_match_live_anchor_fields.sql | 143 +++++++++++ docs/sql/insert_crontab_match_live_stream.sql | 15 +- 4 files changed, 376 insertions(+), 11 deletions(-) create mode 100644 docs/sql/alter_match_live_anchor_fields.sql diff --git a/docker/crawler/config/crawler_tasks.yaml b/docker/crawler/config/crawler_tasks.yaml index eba8150..595781f 100644 --- a/docker/crawler/config/crawler_tasks.yaml +++ b/docker/crawler/config/crawler_tasks.yaml @@ -155,7 +155,7 @@ tasks: cron: "*/10 * * * *" active: true - key: match_live_stream - name: 赛事直播流抓取 + name: 赛事直播流与主播信息抓取 action: match_live_stream cron: "*/1 * * * *" active: true diff --git a/docker/crawler/match_live_stream.py b/docker/crawler/match_live_stream.py index 1d1bc9f..89588c1 100644 --- a/docker/crawler/match_live_stream.py +++ b/docker/crawler/match_live_stream.py @@ -14,11 +14,16 @@ 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/(\d+)") +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) " @@ -26,7 +31,7 @@ DEFAULT_HEADERS = { "Chrome/137.0.0.0 Safari/537.36" ), "Accept": "application/json,text/javascript,*/*;q=0.1", - "Referer": "https://yyzb1.tv/", + "Referer": "https://yuyantv.cn/liveType.html", } @@ -53,6 +58,37 @@ def parse_match_live_detail_html(body: str) -> dict: 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) @@ -68,6 +104,34 @@ 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: @@ -94,13 +158,30 @@ def _fetch_stream_payload(source_url: str, now_ts: int) -> Dict[str, str]: return parse_match_live_detail_html(response.text) -def _build_run_result(candidate_count: int, success_count: int, failed_count: int) -> Dict[str, Any]: +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"赛事直播流抓取完成: 待处理 {candidate_count} 条, " + 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, @@ -108,7 +189,7 @@ def _build_run_result(candidate_count: int, success_count: int, failed_count: in "summary_text": summary_text, } if failed_count > 0: - result["error"] = f"赛事直播流抓取存在失败线路: 失败 {failed_count} 条" + result["error"] = f"赛事直播流与主播信息抓取存在失败线路: 失败 {failed_count} 条" return result @@ -186,13 +267,125 @@ class MatchLiveStreamRepository: 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 " + 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 " @@ -253,10 +446,21 @@ def run(db_config: Dict[str, Any]) -> Dict[str, Any]: 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: - rows = repo.load_due_rows(now_ts, limit=FETCH_BATCH_SIZE) + 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) @@ -269,7 +473,14 @@ def run(db_config: Dict[str, Any]) -> Dict[str, Any]: failed_count += 1 candidate_count = len(rows) - return _build_run_result(candidate_count, success_count, failed_count) + 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: diff --git a/docs/sql/alter_match_live_anchor_fields.sql b/docs/sql/alter_match_live_anchor_fields.sql new file mode 100644 index 0000000..68a2bdd --- /dev/null +++ b/docs/sql/alter_match_live_anchor_fields.sql @@ -0,0 +1,143 @@ +SET @column_exists := ( + SELECT COUNT(*) + FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'la_match_live' + AND COLUMN_NAME = 'anchor_id' +); +SET @sql := IF( + @column_exists = 0, + 'ALTER TABLE `la_match_live` ADD COLUMN `anchor_id` bigint UNSIGNED NOT NULL DEFAULT 0 COMMENT ''主播ID'' AFTER `source_match_time`', + 'SELECT 1' +); +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @column_exists := ( + SELECT COUNT(*) + FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'la_match_live' + AND COLUMN_NAME = 'anchor_name' +); +SET @sql := IF( + @column_exists = 0, + 'ALTER TABLE `la_match_live` ADD COLUMN `anchor_name` varchar(100) NOT NULL DEFAULT '''' COMMENT ''主播昵称'' AFTER `anchor_id`', + 'SELECT 1' +); +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @column_exists := ( + SELECT COUNT(*) + FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'la_match_live' + AND COLUMN_NAME = 'anchor_avatar' +); +SET @sql := IF( + @column_exists = 0, + 'ALTER TABLE `la_match_live` ADD COLUMN `anchor_avatar` varchar(500) NOT NULL DEFAULT '''' COMMENT ''主播头像'' AFTER `anchor_name`', + 'SELECT 1' +); +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @column_exists := ( + SELECT COUNT(*) + FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'la_match_live' + AND COLUMN_NAME = 'anchor_level' +); +SET @sql := IF( + @column_exists = 0, + 'ALTER TABLE `la_match_live` ADD COLUMN `anchor_level` varchar(50) NOT NULL DEFAULT '''' COMMENT ''主播等级'' AFTER `anchor_avatar`', + 'SELECT 1' +); +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @column_exists := ( + SELECT COUNT(*) + FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'la_match_live' + AND COLUMN_NAME = 'room_view_count' +); +SET @sql := IF( + @column_exists = 0, + 'ALTER TABLE `la_match_live` ADD COLUMN `room_view_count` int UNSIGNED NOT NULL DEFAULT 0 COMMENT ''直播间观看数'' AFTER `anchor_level`', + 'SELECT 1' +); +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @column_exists := ( + SELECT COUNT(*) + FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'la_match_live' + AND COLUMN_NAME = 'room_focus_count' +); +SET @sql := IF( + @column_exists = 0, + 'ALTER TABLE `la_match_live` ADD COLUMN `room_focus_count` int UNSIGNED NOT NULL DEFAULT 0 COMMENT ''直播间关注数'' AFTER `room_view_count`', + 'SELECT 1' +); +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @column_exists := ( + SELECT COUNT(*) + FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'la_match_live' + AND COLUMN_NAME = 'room_notice' +); +SET @sql := IF( + @column_exists = 0, + 'ALTER TABLE `la_match_live` ADD COLUMN `room_notice` varchar(500) NOT NULL DEFAULT '''' COMMENT ''直播间公告'' AFTER `room_focus_count`', + 'SELECT 1' +); +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @column_exists := ( + SELECT COUNT(*) + FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'la_match_live' + AND COLUMN_NAME = 'room_detail' +); +SET @sql := IF( + @column_exists = 0, + 'ALTER TABLE `la_match_live` ADD COLUMN `room_detail` varchar(255) NOT NULL DEFAULT '''' COMMENT ''直播间简介'' AFTER `room_notice`', + 'SELECT 1' +); +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @index_exists := ( + SELECT COUNT(*) + FROM information_schema.STATISTICS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'la_match_live' + AND INDEX_NAME = 'idx_anchor_id' +); +SET @sql := IF( + @index_exists = 0, + 'ALTER TABLE `la_match_live` ADD KEY `idx_anchor_id` (`anchor_id`) USING BTREE', + 'SELECT 1' +); +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; diff --git a/docs/sql/insert_crontab_match_live_stream.sql b/docs/sql/insert_crontab_match_live_stream.sql index 8311b69..7fca2e0 100644 --- a/docs/sql/insert_crontab_match_live_stream.sql +++ b/docs/sql/insert_crontab_match_live_stream.sql @@ -1,10 +1,21 @@ +UPDATE `la_dev_crontab` +SET + `name` = '赛事直播流与主播信息抓取', + `remark` = 'Docker crawler 每分钟同步雨燕直播间、主播信息并抓取直播线路播放地址', + `status` = 1, + `expression` = '* * * * *', + `update_time` = UNIX_TIMESTAMP() +WHERE `command` = 'crawler' + AND `params` = 'match_live_stream' + AND `delete_time` IS NULL; + INSERT INTO `la_dev_crontab` (`name`, `type`, `system`, `remark`, `command`, `params`, `status`, `expression`, `create_time`, `update_time`) SELECT - '赛事直播流抓取', + '赛事直播流与主播信息抓取', 1, 0, - 'Docker crawler 每分钟抓取赛事直播线路播放地址', + 'Docker crawler 每分钟同步雨燕直播间、主播信息并抓取直播线路播放地址', 'crawler', 'match_live_stream', 1,