听友FM

2026-04-24 16:17:49 +08:00
parent 0ef3c10767
commit d7aa38154e
3 changed files with 303 additions and 35 deletions
@@ -6,6 +6,7 @@ from types import SimpleNamespace
 from unittest.mock import patch
 from Crypto.Cipher import AES
 from Crypto.Cipher import ChaCha20_Poly1305
 ROOT = Path(__file__).resolve().parents[1]
@@ -27,6 +28,20 @@ HOME_HTML = """
 """
 HOME_HTML_WITH_NUXT_COVERS = """
 <html>
  <body>
    <a href="/categories/46">有声小说</a>
    <a href="/albums/1001">
      <img class="cover" alt="鬼吹灯">
      <p>鬼吹灯 作者：天下霸唱 播音：周建龙 12期 已完结</p>
    </a>
    <script id="__NUXT_DATA__" type="application/json">[null,{"data":{"index-home-tabs":{"latest":{"items":[{"id":1001,"title":"鬼吹灯","cover":"https://img.test/1001.jpg","desc":"摸金探险","chapterTotal":12,"status":0,"teller":"周建龙","author":"天下霸唱"}]}}}}]</script>
  </body>
 </html>
 """
 CATEGORY_HTML_WITH_NUXT = """
 <html>
  <body>
@@ -36,6 +51,23 @@ CATEGORY_HTML_WITH_NUXT = """
 """
 CATEGORY_HTML_WITH_NUXT_REFERENCES = """
 <html>
  <body>
    <script id="__NUXT_DATA__" type="application/json">[
      null,
      {"data":2},
      ["ShallowReactive",{"categoryAlbums-46":3}],
      {"page":"1","pages":"7","data":4},
      [
        {"id":2002,"title":"沙海","cover_url":"https://img.test/2002.jpg","count":88,"status":"1","teller":"青雪"}
      ]
    ]</script>
  </body>
 </html>
 """
 SEARCH_HTML_WITHOUT_NUXT = """
 <html>
  <body>
@@ -100,6 +132,14 @@ class TestTingYouFMSpider(unittest.TestCase):
        raw = bytes([2]) + nonce + reversed_cipher
        return raw.hex()
    def _build_xchacha_v1_payload(self, plain_text):
        key = bytes.fromhex(self.spider.payload_key_hex)
        nonce = bytes(range(1, 25))
        cipher = ChaCha20_Poly1305.new(key=key, nonce=nonce)
        encrypted, tag = cipher.encrypt_and_digest(plain_text.encode("utf-8"))
        raw = bytes([1]) + nonce + encrypted + tag
        return raw.hex()
    @patch.object(Spider, "fetch")
    def test_home_content_extracts_categories_and_album_cards(self, mock_fetch):
        mock_fetch.return_value = SimpleNamespace(status_code=200, text=HOME_HTML)
@@ -110,6 +150,14 @@ class TestTingYouFMSpider(unittest.TestCase):
        self.assertEqual(result["list"][0]["vod_pic"], "https://tingyou.fm/cover1.jpg")
        self.assertIn("12期", result["list"][0]["vod_remarks"])
    @patch.object(Spider, "fetch")
    def test_home_content_prefers_nuxt_cover_when_dom_img_has_no_src(self, mock_fetch):
        mock_fetch.return_value = SimpleNamespace(status_code=200, text=HOME_HTML_WITH_NUXT_COVERS)
        result = self.spider.homeContent(False)
        self.assertEqual(result["list"][0]["vod_id"], "1001")
        self.assertEqual(result["list"][0]["vod_pic"], "https://img.test/1001.jpg")
        self.assertIn("12期", result["list"][0]["vod_remarks"])
    @patch.object(Spider, "fetch")
    def test_category_content_prefers_nuxt_data(self, mock_fetch):
        mock_fetch.return_value = SimpleNamespace(status_code=200, text=CATEGORY_HTML_WITH_NUXT)
@@ -122,6 +170,17 @@ class TestTingYouFMSpider(unittest.TestCase):
        self.assertEqual(result["limit"], 1)
        self.assertNotIn("pagecount", result)
    @patch.object(Spider, "fetch")
    def test_category_content_decodes_nuxt_reference_table(self, mock_fetch):
        mock_fetch.return_value = SimpleNamespace(status_code=200, text=CATEGORY_HTML_WITH_NUXT_REFERENCES)
        result = self.spider.categoryContent("46", "1", False, {})
        self.assertEqual(result["page"], 1)
        self.assertEqual(result["limit"], 1)
        self.assertEqual(result["list"][0]["vod_id"], "2002")
        self.assertEqual(result["list"][0]["vod_name"], "沙海")
        self.assertEqual(result["list"][0]["vod_pic"], "https://img.test/2002.jpg")
        self.assertIn("88期", result["list"][0]["vod_remarks"])
    @patch.object(Spider, "fetch")
    def test_search_content_falls_back_to_dom_and_filters_blank_keyword(self, mock_fetch):
        mock_fetch.return_value = SimpleNamespace(status_code=200, text=SEARCH_HTML_WITHOUT_NUXT)
@@ -144,33 +203,111 @@ class TestTingYouFMSpider(unittest.TestCase):
        self.assertEqual(vod["vod_play_from"], "听友FM")
        self.assertEqual(vod["vod_play_url"], "第1集$1001|1#第2集$1001|2")
    @patch.object(Spider, "fetch")
    def test_detail_content_prefers_full_nuxt_chapter_list_over_truncated_dom(self, mock_fetch):
        dom_items = "".join(
            f"""
      <li class="chapter-item">
        <p>{index}</p>
        <div class="item-content"><span class="title">第{index}集</span></div>
      </li>
 """
            for index in range(1, 41)
        )
        nuxt_payload = [
            None,
            {
                "data": {
                    "album-detail-1001": {
                        "title": "鬼吹灯",
                        "cover_url": "https://img.test/detail.jpg",
                        "synopsis": "摸金校尉探险故事",
                    },
                    "album-chapters-1001": {
                        "id": 1001,
                        "available": True,
                        "count": 42,
                        "detail": 30,
                        "chapters": [
                            {"id": 5000 + index, "index": str(index), "title": f"{index:03d}.第{index}集"}
                            for index in range(1, 43)
                        ],
                    },
                }
            },
        ]
        html = f"""
 <html>
  <head>
    <meta property="og:title" content="鬼吹灯">
    <meta property="og:image" content="https://img.test/detail.jpg">
    <meta name="description" content="摸金校尉探险故事">
  </head>
  <body>
    <script id="__NUXT_DATA__" type="application/json">{json.dumps(nuxt_payload, ensure_ascii=False)}</script>
    <section class="album-pannel">
      <div class="album-intro">
        <h1>鬼吹灯</h1>
      </div>
      <div class="pods">
        <span>分类: 有声小说</span>
      </div>
      <img src="https://img.test/detail.jpg">
    </section>
    <ul class="chapter-list">{dom_items}
    </ul>
  </body>
 </html>
 """
        mock_fetch.return_value = SimpleNamespace(status_code=200, text=html)
        result = self.spider.detailContent(["1001"])
        vod = result["list"][0]
        play_urls = vod["vod_play_url"].split("#")
        self.assertEqual(len(play_urls), 42)
        self.assertEqual(play_urls[0], "001.第1集$1001|1")
        self.assertEqual(play_urls[-1], "042.第42集$1001|42")
    def test_encrypt_payload_prefixes_version_and_decrypts_v1_payload(self):
        payload = self.spider._encrypt_payload('{"album_id":1001,"chapter_idx":1}')
        self.assertTrue(payload.startswith("01"))
        plain = self.spider._decrypt_payload(self._build_v1_payload('{"url":"https://audio.test/1.m4a"}'))
        self.assertEqual(plain, '{"url":"https://audio.test/1.m4a"}')
    def test_decrypt_payload_supports_xchacha_v1_response(self):
        plain = self.spider._decrypt_payload(self._build_xchacha_v1_payload('{"auth_token":"guest-token"}'))
        self.assertEqual(plain, '{"auth_token":"guest-token"}')
    def test_decrypt_v2_payload_reverses_cipher_before_decoder(self):
        calls = {}
-        def fake_decrypt(key, nonce, cipher):
+        def fake_decrypt(key, nonce, body):
            calls["key"] = key
            calls["nonce"] = nonce
-            calls["cipher"] = cipher
+            calls["body"] = body
            return b'{"url":"https://audio.test/2.m4a"}'
-        self.spider._xchacha_decrypt = fake_decrypt
+        self.spider._decrypt_xchacha_body = fake_decrypt
        plain = self.spider._decrypt_payload(self._build_v2_payload())
        self.assertEqual(plain, '{"url":"https://audio.test/2.m4a"}')
        self.assertEqual(calls["nonce"], bytes(range(1, 25)))
-        self.assertEqual(calls["cipher"], b"abc")
+        self.assertEqual(calls["body"], b"abc")
    @patch.object(Spider, "_anonymous_auth")
    @patch.object(Spider, "_api_post")
-    def test_player_content_prefers_api_url_and_falls_back_to_audio_page(self, mock_api_post):
+    def test_player_content_prefers_api_url_and_falls_back_to_audio_page(self, mock_api_post, mock_anonymous_auth):
-        mock_api_post.return_value = {"payload": self._build_v1_payload('{"url":"https://audio.test/play.m4a"}')}
+        mock_anonymous_auth.return_value = {
            "auth_token": "guest-token",
            "cookie": "dfp=f-demo:f-token",
        }
        mock_api_post.return_value = {"payload": self._build_xchacha_v1_payload('{"play_url":"https://audio.test/play.m4a"}')}
        api_result = self.spider.playerContent("听友FM", "1001|1", {})
        self.assertEqual(api_result["parse"], 0)
        self.assertEqual(api_result["url"], "https://audio.test/play.m4a")
        mock_anonymous_auth.assert_called_once()
        self.assertEqual(mock_api_post.call_args.args[0], "/api/play_token")
        self.assertEqual(mock_api_post.call_args.args[1], {"album_id": 1001, "chapter_idx": 1})
        self.assertEqual(mock_api_post.call_args.kwargs["extra_headers"]["Authorization"], "Bearer guest-token")
        self.assertEqual(mock_api_post.call_args.kwargs["extra_headers"]["Cookie"], "dfp=f-demo:f-token")
        mock_api_post.side_effect = RuntimeError("boom")
        fallback_result = self.spider.playerContent("听友FM", "1001|1", {})
@@ -2,9 +2,12 @@
 import json
 import re
 import sys
 import time
 from hashlib import sha256
 from urllib.parse import quote
 from Crypto.Cipher import AES
 from Crypto.Cipher import ChaCha20_Poly1305
 from Crypto.Random import get_random_bytes
 from lxml import html as lxml_html
@@ -39,7 +42,6 @@ class Spider(BaseSpider):
            {"type_id": "15", "type_name": "历史军事"},
            {"type_id": "9", "type_name": "百家讲坛"},
        ]
        self._xchacha_decrypt = None
    def init(self, extend=""):
        return None
@@ -97,35 +99,85 @@ class Spider(BaseSpider):
        encrypted, tag = cipher.encrypt_and_digest(str(plain_text or "").encode("utf-8"))
        return self._bytes_to_hex(bytes([self.payload_version]) + iv + encrypted + tag)
    def _decrypt_xchacha_body(self, key, nonce, body):
        cipher = ChaCha20_Poly1305.new(key=key, nonce=nonce)
        return cipher.decrypt_and_verify(body[:-16], body[-16:])
    def _decrypt_aes_gcm_payload(self, raw, attempts):
        key = self._hex_to_bytes(self.payload_key_hex)
        last_error = None
        for iv_start, cipher_start in attempts:
            try:
                iv = raw[iv_start:iv_start + 12]
                body = raw[cipher_start:]
                encrypted = body[:-16]
                tag = body[-16:]
                cipher = AES.new(key, AES.MODE_GCM, nonce=iv)
                return cipher.decrypt_and_verify(encrypted, tag).decode("utf-8")
            except Exception as exc:
                last_error = exc
        if last_error:
            raise last_error
        raise ValueError("aes decrypt failed")
    def _decrypt_payload(self, hex_text):
        raw = self._hex_to_bytes(hex_text)
        if len(raw) < 2:
            raise ValueError("payload too short")
        version = raw[0]
        key = self._hex_to_bytes(self.payload_key_hex)
        if version == 1:
-            iv = raw[1:13]
+            if len(raw) >= 41:
-            body = raw[13:]
+                try:
-            encrypted = body[:-16]
+                    plain = self._decrypt_xchacha_body(key, raw[1:25], raw[25:])
-            tag = body[-16:]
+                    return plain.decode("utf-8")
-            cipher = AES.new(self._hex_to_bytes(self.payload_key_hex), AES.MODE_GCM, nonce=iv)
+                except Exception:
-            plain = cipher.decrypt_and_verify(encrypted, tag)
+                    pass
-            return plain.decode("utf-8")
+            return self._decrypt_aes_gcm_payload(raw, [(1, 13), (0, 12), (2, 14)])
        if version == 2:
            decryptor = getattr(self, "_xchacha_decrypt", None)
            if not callable(decryptor):
                raise ValueError("xchacha decryptor unavailable")
            nonce = raw[1:25]
-            cipher = raw[25:][::-1]
+            body = raw[25:][::-1]
-            plain = decryptor(self._hex_to_bytes(self.payload_key_hex), nonce, cipher)
+            plain = self._decrypt_xchacha_body(key, nonce, body)
            return bytes(plain).decode("utf-8")
        raise ValueError("unsupported payload version")
    def _make_dfp_cookie(self):
        today_hex = format(int(time.strftime("%Y%m%d")), "x")
        seed = f"{time.time()}|tingyou|{get_random_bytes(8).hex()}".encode("utf-8")
        return f"dfp=f-{today_hex}:f-{sha256(seed).hexdigest()}"
    def _decode_nuxt_value(self, table, node, seen=None):
        seen = seen or {}
        markers = {
            "ShallowReactive",
            "Reactive",
            "Ref",
            "EmptyRef",
            "Set",
            "Map",
            "Date",
            "RegExp",
            "BigInt",
            "null",
            "undefined",
            "NaN",
            "-0",
            "Infinity",
            "-Infinity",
        }
        if isinstance(node, int) and 0 <= node < len(table):
            if node in seen:
                return seen[node]
            raw = table[node]
            if isinstance(raw, list) and raw and isinstance(raw[0], str) and raw[0] in markers:
                marker = raw[0]
                if marker in ("ShallowReactive", "Reactive", "Ref"):
                    value = self._decode_nuxt_value(table, raw[1] if len(raw) > 1 else None, seen)
                    seen[node] = value
                    return value
                if marker in ("EmptyRef", "null", "undefined", "NaN"):
                    seen[node] = None
                    return None
            if isinstance(raw, dict):
                seen[node] = {}
                for key, value in raw.items():
@@ -152,10 +204,7 @@ class Spider(BaseSpider):
        except Exception:
            return {}
        if isinstance(payload, list) and len(payload) > 1:
-            root = payload[1]
+            return self._decode_nuxt_value(payload, 1)
            if isinstance(root, dict):
                return root
            return self._decode_nuxt_value(payload, root)
        return payload if isinstance(payload, dict) else {}
    def _pick_image(self, node):
@@ -225,6 +274,42 @@ class Spider(BaseSpider):
                items.append(item)
        return self._unique_by_id(items)
    def _parse_home_nuxt(self, html):
        root = self._load_nuxt_root(html)
        tabs = ((root.get("data") or {}).get("index-home-tabs") or {})
        items = []
        for bucket in tabs.values():
            tab_items = bucket.get("items") if isinstance(bucket, dict) else None
            if not isinstance(tab_items, list):
                continue
            for item in tab_items:
                album_id = str((item or {}).get("id") or "").strip()
                if not album_id:
                    continue
                status = "连载中" if str(item.get("status")) == "1" else "已完结" if str(item.get("status")) == "0" else ""
                remarks = " · ".join(
                    [
                        value
                        for value in [
                            f"{item.get('chapterTotal')}期" if item.get("chapterTotal") else "",
                            status,
                            str(item.get("teller") or item.get("author") or "").strip(),
                        ]
                        if value
                    ]
                )
                items.append(
                    {
                        "vod_id": album_id,
                        "vod_name": str(item.get("title") or ("专辑" + album_id)),
                        "vod_pic": self._normalize_url(item.get("cover") or item.get("cover_url") or ""),
                        "vod_remarks": remarks,
                        "type_id": "",
                        "type_name": str(item.get("categoryName") or "").strip(),
                    }
                )
        return self._unique_by_id(items)
    def _map_nuxt_album_item(self, item, tid, type_name):
        data = item or {}
        album_id = str(data.get("id") or "")
@@ -317,12 +402,30 @@ class Spider(BaseSpider):
            if text.startswith("分类:"):
                type_name = text.split(":", 1)[1].strip()
                break
        root = self._load_nuxt_root(html)
        nuxt_data = root.get("data") or {}
        album_detail = nuxt_data.get(f"album-detail-{album_id}") or {}
        album_chapters = nuxt_data.get(f"album-chapters-{album_id}") or {}
        if not name:
            name = str(album_detail.get("title") or "").strip()
        if not pic:
            pic = self._normalize_url(album_detail.get("cover_url") or "")
        if not content:
            content = str(album_detail.get("synopsis") or "").strip()
        play_items = []
-        for index, item in enumerate(document.xpath("//ul[contains(@class, 'chapter-list')]/li[contains(@class, 'chapter-item')]"), start=1):
+        chapters = album_chapters.get("chapters") or []
-            num_text = self._safe_text(next(iter(item.xpath("./p")), None))
+        if isinstance(chapters, list) and chapters:
-            title = self._safe_text(next(iter(item.xpath(".//*[contains(@class, 'title')]")), None)) or f"第{index}集"
+            for index, item in enumerate(chapters, start=1):
-            chapter_idx = int(num_text) if str(num_text).isdigit() else index
+                raw_index = item.get("index")
-            play_items.append(f"{title}${album_id}|{chapter_idx}")
+                chapter_idx = int(raw_index) if isinstance(raw_index, (int, str)) and str(raw_index).isdigit() else index
                title = str(item.get("title") or f"第{chapter_idx}集").strip()
                play_items.append(f"{title}${album_id}|{chapter_idx}")
        else:
            for index, item in enumerate(document.xpath("//ul[contains(@class, 'chapter-list')]/li[contains(@class, 'chapter-item')]"), start=1):
                num_text = self._safe_text(next(iter(item.xpath("./p")), None))
                title = self._safe_text(next(iter(item.xpath(".//*[contains(@class, 'title')]")), None)) or f"第{index}集"
                chapter_idx = int(num_text) if str(num_text).isdigit() else index
                play_items.append(f"{title}${album_id}|{chapter_idx}")
        return {
            "vod_id": str(album_id),
            "vod_name": name or ("专辑" + str(album_id)),
@@ -349,10 +452,16 @@ class Spider(BaseSpider):
                return plain
        return value
-    def _api_post(self, path, body):
+    def _api_post(self, path, body=None, extra_headers=None):
        url = path if str(path).startswith("http") else self.host + (path if str(path).startswith("/") else "/" + str(path))
-        payload = self._encrypt_payload(json.dumps(body or {}, ensure_ascii=False, separators=(",", ":")))
+        payload = None
-        headers = self._get_headers({"Content-Type": "text/plain", "X-Payload-Version": str(self.payload_version)})
+        if body is not None:
            payload = self._encrypt_payload(json.dumps(body, ensure_ascii=False, separators=(",", ":")))
        headers = self._get_headers({"X-Payload-Version": str(self.payload_version)})
        if payload is not None:
            headers["Content-Type"] = "text/plain"
        if extra_headers:
            headers.update(extra_headers)
        response = self.post(url, data=payload, headers=headers, timeout=10, verify=False)
        if getattr(response, "status_code", 0) >= 400:
            raise ValueError("api request failed")
@@ -388,6 +497,16 @@ class Spider(BaseSpider):
                return candidate
        return ""
    def _anonymous_auth(self):
        cookie = self._make_dfp_cookie()
        data = self._api_post("/api/me", None, extra_headers={"Accept": "application/json", "Cookie": cookie})
        auth_token = str((data or {}).get("auth_token") or "").strip()
        return {
            "auth_token": auth_token,
            "cookie": cookie,
            "data": data or {},
        }
    def homeContent(self, filter):
        html = self._get_html("/")
        document = self._load_html(html)
@@ -403,7 +522,8 @@ class Spider(BaseSpider):
                continue
            seen.add(type_id)
            classes.append({"type_id": type_id, "type_name": self._safe_text(anchor) or self._category_name(type_id)})
-        return {"class": classes or list(self.classes), "list": self._parse_home_list(html)[:20]}
+        items = self._parse_home_nuxt(html) or self._parse_home_list(html)
        return {"class": classes or list(self.classes), "list": items[:20]}
    def homeVideoContent(self):
        return {"list": self.homeContent(False).get("list", [])}
@@ -443,11 +563,22 @@ class Spider(BaseSpider):
        album_id, chapter_idx = str(id or "").split("|", 1)
        fallback = f"{self.host}/audios/{album_id}/{chapter_idx}"
        try:
-            payload = self._api_post("/api/play_token", {"album_id": int(album_id), "chapter_idx": int(chapter_idx)})
+            auth = self._anonymous_auth()
            extra_headers = {
                "Accept": "application/json",
                "Cookie": auth.get("cookie", ""),
            }
            if auth.get("auth_token"):
                extra_headers["Authorization"] = f"Bearer {auth['auth_token']}"
            payload = self._api_post(
                "/api/play_token",
                {"album_id": int(album_id), "chapter_idx": int(chapter_idx)},
                extra_headers=extra_headers,
            )
            data = self._normalize_api_result(payload)
            url = self._extract_play_url(data)
            if url:
-                return {"parse": 0, "url": url, "header": self._get_headers()}
+                return {"parse": 0, "url": url, "header": self._get_headers(extra_headers)}
        except Exception:
            pass
        return {"parse": 1, "url": fallback, "header": self._get_headers()}
@@ -356,7 +356,7 @@ class Spider(BaseSpider):
                "corp": "kuwo",
                "albumid": album_id,
                "pn": 0,
-                "rn": 5000,
+                "rn": 2000,
                "show_copyright_off": 1,
                "vipver": "MUSIC_8.2.0.0_BCS17",
                "mobi": 1,