haha/sbnews
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: sync yuyan anchor live stream task

Browse Source
This commit is contained in:
hajimi
2026-07-03 21:10:05 +08:00
parent 8c358dc376
commit 33664adbe0
4 changed files with 376 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docker/crawler/config/crawler_tasks.yaml
+1 -1
View File
@@ -155,7 +155,7 @@ tasks:
cron: "*/10 * * * *"
active: true
- key: match_live_stream
name: 赛事直播流抓取
name: 赛事直播流与主播信息抓取
action: match_live_stream
cron: "*/1 * * * *"
active: true
docker/crawler/match_live_stream.py
+219 -8
View File
@@ -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: