feat: add docker match live stream crawler

docker/crawler/config/crawler_tasks.yaml

@@ -154,3 +154,8 @@ tasks:
     action: ai_comment_dispatch
     cron: "*/10 * * * *"
     active: true
+  - key: match_live_stream
+    name: 赛事直播流抓取
+    action: match_live_stream
+    cron: "*/1 * * * *"
+    active: true

docker/crawler/main.py

@@ -30,6 +30,7 @@
   python main.py nba_news           # 采集NBA新闻资讯
   python main.py cba_news           # 采集CBA新闻资讯
   python main.py ai_comment_dispatch # Docker AI 评论调度
+  python main.py match_live_stream  # 抓取赛事直播线路播放流
   python main.py article_content_fetch # 抓取资讯文章正文内容
   python main.py all                # 采集积分榜 + 赛程
   python main.py single <season_id> # 采集单个联赛积分榜
@@ -59,6 +60,7 @@ TASK_ALERT_POLICY = {
     "lottery_draw": "saved_if_candidates",
     "lottery_draw_force": "saved_if_candidates",
     "ai_comment_dispatch": "saved_if_candidates",
+    "match_live_stream": "saved_if_candidates",
 }
 
 DEFAULT_ACTION_TIMEOUT = 600
@@ -86,6 +88,7 @@ ACTION_TIMEOUTS = {
     "fifa_worldcup_news": 600,
     "dqd_worldcup": 600,
     "ai_comment_dispatch": 1800,
+    "match_live_stream": 180,
 }
 
 
@@ -522,6 +525,24 @@ async def cmd_ai_comment_dispatch():
     return await asyncio.to_thread(run, db_config)
 
 
+async def cmd_match_live_stream():
+    import asyncio
+    from match_live_stream import run
+    from src.core.config import load_config
+
+    cfg = load_config().database
+    db_config = {
+        "host": cfg.host,
+        "port": cfg.port,
+        "user": cfg.username,
+        "password": cfg.password,
+        "database": cfg.database,
+        "charset": cfg.charset,
+        "prefix": cfg.prefix,
+    }
+    return await asyncio.to_thread(run, db_config)
+
+
 async def cmd_match_data():
     from src.scheduler.task_runner import TaskRunner
     async with TaskRunner() as runner:
@@ -837,6 +858,7 @@ def main():
         "nba_news": cmd_nba_news,
         "cba_news": cmd_cba_news,
         "ai_comment_dispatch": cmd_ai_comment_dispatch,
+        "match_live_stream": cmd_match_live_stream,
         "article_content_fetch": cmd_article_content_fetch,
         "match_finish": cmd_match_finish,
         "live_detail": cmd_live_detail,

docker/crawler/match_live_stream.py

@@ -0,0 +1,260 @@
+#!/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}"
+FETCH_BATCH_SIZE = 30
+FETCH_TIMEOUT = 15
+FETCH_ERROR_LIMIT = 500
+ROOM_ID_RE = re.compile(r"/room/(\d+)")
+DETAIL_JSONP_RE = re.compile(r"detail\((\{.*\})\)\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://yyzb1.tv/",
+}
+
+
+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 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 _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)
+
+
+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 "
+            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 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 = 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
+    rows: List[Dict[str, Any]] = []
+
+    try:
+        rows = repo.load_due_rows(now_ts, limit=FETCH_BATCH_SIZE)
+        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 {
+            "success": True,
+            "candidate_count": candidate_count,
+            "saved_count": success_count,
+            "count": success_count,
+            "failed_count": failed_count,
+            "summary_text": (
+                f"赛事直播流抓取完成: 待处理 {candidate_count} 条, "
+                f"成功 {success_count} 条, 失败 {failed_count} 条"
+            ),
+        }
+    finally:
+        repo.close()
+        lock.release()

docker/crawler/tests/test_match_live_stream.py

@@ -0,0 +1,44 @@
+import json
+import sys
+import unittest
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from match_live_stream import choose_next_fetch_at, parse_match_live_detail_html
+
+
+class MatchLiveStreamTest(unittest.TestCase):
+    def test_parse_json_ncctrials_detail_extracts_four_urls(self):
+        payload = {
+            "code": 200,
+            "msg": "ok",
+            "data": {
+                "stream": {
+                    "flv": "http://a/flv",
+                    "hdFlv": "http://a/hdflv",
+                    "m3u8": "http://a/m3u8",
+                    "hdM3u8": "http://a/hdm3u8",
+                }
+            },
+        }
+
+        parsed = parse_match_live_detail_html(f"detail({json.dumps(payload, ensure_ascii=False)})")
+
+        self.assertEqual(parsed["play_url_m3u8"], "http://a/m3u8")
+        self.assertEqual(parsed["play_url_hd_m3u8"], "http://a/hdm3u8")
+        self.assertEqual(parsed["play_url_flv"], "http://a/flv")
+        self.assertEqual(parsed["play_url_hd_flv"], "http://a/hdflv")
+
+    def test_choose_next_fetch_at_uses_60s_for_empty_streams(self):
+        self.assertEqual(choose_next_fetch_at(now_ts=1000, has_existing_stream=False), 1060)
+
+    def test_choose_next_fetch_at_uses_600s_for_existing_streams(self):
+        self.assertEqual(choose_next_fetch_at(now_ts=1000, has_existing_stream=True), 1600)
+
+
+if __name__ == "__main__":
+    unittest.main()

docs/sql/insert_crontab_match_live_stream.sql

@@ -0,0 +1,20 @@
+INSERT INTO `la_dev_crontab`
+(`name`, `type`, `system`, `remark`, `command`, `params`, `status`, `expression`, `create_time`, `update_time`)
+SELECT
+    '赛事直播流抓取',
+    1,
+    0,
+    'Docker crawler 每分钟抓取赛事直播线路播放地址',
+    'crawler',
+    'match_live_stream',
+    1,
+    '* * * * *',
+    UNIX_TIMESTAMP(),
+    UNIX_TIMESTAMP()
+WHERE NOT EXISTS (
+    SELECT 1
+    FROM `la_dev_crontab`
+    WHERE `command` = 'crawler'
+      AND `params` = 'match_live_stream'
+      AND `delete_time` IS NULL
+);