From 0d658a368bcf4b1d23a27ff50cee3ac4806158f7 Mon Sep 17 00:00:00 2001
From: Harold <8866033@gmail.com>
Date: Fri, 24 Apr 2026 15:21:19 +0800
Subject: [PATCH] feat: add tingyoufm spider
---
.../plans/2026-04-24-tingyoufm-spider.md | 290 +++++++++++
py/tests/test_听友FM.py | 182 +++++++
py/听友FM.py | 453 ++++++++++++++++++
3 files changed, 925 insertions(+)
create mode 100644 py/docs/superpowers/plans/2026-04-24-tingyoufm-spider.md
create mode 100644 py/tests/test_听友FM.py
create mode 100644 py/听友FM.py
diff --git a/py/docs/superpowers/plans/2026-04-24-tingyoufm-spider.md b/py/docs/superpowers/plans/2026-04-24-tingyoufm-spider.md
new file mode 100644
index 0000000..7007ec3
--- /dev/null
+++ b/py/docs/superpowers/plans/2026-04-24-tingyoufm-spider.md
@@ -0,0 +1,290 @@
+# 听友FM Spider Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** 为 `https://tingyou.fm` 新增一个可离线测试的 Python Spider，覆盖首页、分类、搜索、详情和播放主链路。
+
+**Architecture:** 站点文件 `py/听友FM.py` 采用“页面解析为主、播放接口为主链”的结构。列表和详情统一复用 HTML/Nuxt 解析 helper，播放链路单独收敛到 payload 加解密和直链提取 helper，并在失败时回退到播放页 URL。
+
+**Tech Stack:** Python, `unittest`, `unittest.mock`, `json`, `re`, `base.spider.Spider`, `Crypto.Cipher.AES`
+
+---
+
+### Task 1: Scaffold Spider And Red Tests
+
+**Files:**
+- Create: `py/听友FM.py`
+- Create: `py/tests/test_听友FM.py`
+- Test: `py/tests/test_听友FM.py`
+
+- [ ] **Step 1: Write the failing test**
+
+```python
+def test_home_content_extracts_categories_and_album_cards(self):
+ result = self.spider.homeContent(False)
+ self.assertEqual(result["class"][0], {"type_id": "46", "type_name": "有声小说"})
+ self.assertEqual(result["list"][0]["vod_id"], "1001")
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run python -m unittest tests.test_听友FM.TestTingYouFMSpider.test_home_content_extracts_categories_and_album_cards -v`
+Expected: FAIL with missing module `听友FM.py` or missing `homeContent`
+
+- [ ] **Step 3: Write minimal implementation**
+
+```python
+class Spider(BaseSpider):
+ def __init__(self):
+ self.name = "听友FM"
+ self.host = "https://tingyou.fm"
+ self.classes = [{"type_id": "46", "type_name": "有声小说"}]
+
+ def init(self, extend=""):
+ return None
+
+ def getName(self):
+ return self.name
+
+ def homeContent(self, filter):
+ return {"class": list(self.classes), "list": []}
+```
+
+- [ ] **Step 4: Run test to verify it passes**
+
+Run: `uv run python -m unittest tests.test_听友FM.TestTingYouFMSpider.test_home_content_extracts_categories_and_album_cards -v`
+Expected: PASS
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add py/听友FM.py py/tests/test_听友FM.py
+git commit -m "feat: scaffold tingyoufm spider"
+```
+
+### Task 2: Implement HTML And Nuxt Parsing
+
+**Files:**
+- Modify: `py/听友FM.py`
+- Modify: `py/tests/test_听友FM.py`
+- Test: `py/tests/test_听友FM.py`
+
+- [ ] **Step 1: Write the failing test**
+
+```python
+@patch.object(Spider, "fetch")
+def test_category_and_search_prefer_nuxt_and_fallback_to_dom(self, mock_fetch):
+ mock_fetch.side_effect = [
+ SimpleNamespace(status_code=200, text=CATEGORY_HTML_WITH_NUXT),
+ SimpleNamespace(status_code=200, text=SEARCH_HTML_WITHOUT_NUXT),
+ ]
+ category = self.spider.categoryContent("46", "2", False, {})
+ search = self.spider.searchContent("鬼吹灯", False, "1")
+ self.assertEqual(category["page"], 2)
+ self.assertEqual(category["list"][0]["vod_id"], "2001")
+ self.assertEqual(search["list"][0]["vod_id"], "3001")
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run python -m unittest tests.test_听友FM.TestTingYouFMSpider.test_category_and_search_prefer_nuxt_and_fallback_to_dom -v`
+Expected: FAIL with missing `categoryContent`, `searchContent`, or wrong parsed fields
+
+- [ ] **Step 3: Write minimal implementation**
+
+```python
+def _decode_nuxt_value(self, table, node, seen=None):
+ if seen is None:
+ seen = {}
+ return node
+
+def _parse_album_card(self, html):
+ return {"vod_id": "2001", "vod_name": "示例专辑", "vod_pic": "", "vod_remarks": ""}
+
+def categoryContent(self, tid, pg, filter, extend):
+ html = self._get_html(f"/categories/{tid}?sort=comprehensive&page={pg}")
+ data = self._parse_category_nuxt(html, tid)
+ items = data or []
+ return {"page": int(pg), "limit": len(items), "total": len(items), "list": items}
+
+def searchContent(self, key, quick, pg="1"):
+ if not str(key).strip():
+ return {"page": 1, "limit": 0, "total": 0, "list": []}
+ html = self._get_html(f"/search?q={quote(key)}")
+ items = self._parse_search_nuxt(html) or self._parse_search_dom(html)
+ return {"page": int(pg), "limit": len(items), "total": len(items), "list": items}
+```
+
+- [ ] **Step 4: Run test to verify it passes**
+
+Run: `uv run python -m unittest tests.test_听友FM.TestTingYouFMSpider.test_category_and_search_prefer_nuxt_and_fallback_to_dom -v`
+Expected: PASS
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add py/听友FM.py py/tests/test_听友FM.py
+git commit -m "feat: add tingyoufm list and search parsing"
+```
+
+### Task 3: Implement Detail Parsing
+
+**Files:**
+- Modify: `py/听友FM.py`
+- Modify: `py/tests/test_听友FM.py`
+- Test: `py/tests/test_听友FM.py`
+
+- [ ] **Step 1: Write the failing test**
+
+```python
+@patch.object(Spider, "fetch")
+def test_detail_content_extracts_album_and_playlist(self, mock_fetch):
+ mock_fetch.return_value = SimpleNamespace(status_code=200, text=DETAIL_HTML)
+ result = self.spider.detailContent(["1001"])
+ vod = result["list"][0]
+ self.assertEqual(vod["vod_name"], "鬼吹灯")
+ self.assertEqual(vod["vod_play_from"], "听友FM")
+ self.assertEqual(vod["vod_play_url"], "第1集$1001|1#第2集$1001|2")
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run python -m unittest tests.test_听友FM.TestTingYouFMSpider.test_detail_content_extracts_album_and_playlist -v`
+Expected: FAIL with missing `detailContent` or empty playlist
+
+- [ ] **Step 3: Write minimal implementation**
+
+```python
+def detailContent(self, ids):
+ album_id = str((ids or [""])[0])
+ html = self._get_html(f"/albums/{album_id}")
+ vod = self._parse_detail_page(html, album_id)
+ return {"list": [vod] if vod else []}
+```
+
+- [ ] **Step 4: Run test to verify it passes**
+
+Run: `uv run python -m unittest tests.test_听友FM.TestTingYouFMSpider.test_detail_content_extracts_album_and_playlist -v`
+Expected: PASS
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add py/听友FM.py py/tests/test_听友FM.py
+git commit -m "feat: add tingyoufm detail parsing"
+```
+
+### Task 4: Implement Payload Crypto Helpers
+
+**Files:**
+- Modify: `py/听友FM.py`
+- Modify: `py/tests/test_听友FM.py`
+- Test: `py/tests/test_听友FM.py`
+
+- [ ] **Step 1: Write the failing test**
+
+```python
+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_response('{"url":"https://a.test/1.m4a"}'))
+ self.assertEqual(plain, '{"url":"https://a.test/1.m4a"}')
+
+def test_decrypt_v2_payload_reverses_cipher_before_decoder(self):
+ self.spider._xchacha_decrypt = lambda key, nonce, cipher: self.assertEqual(cipher, b"abc") or b'{"url":"https://a.test/2.m4a"}'
+ raw_hex = self._build_v2_response_with_reversed_cipher()
+ plain = self.spider._decrypt_payload(raw_hex)
+ self.assertIn("2.m4a", plain)
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run python -m unittest tests.test_听友FM.TestTingYouFMSpider.test_encrypt_payload_prefixes_version_and_decrypts_v1_payload tests.test_听友FM.TestTingYouFMSpider.test_decrypt_v2_payload_reverses_cipher_before_decoder -v`
+Expected: FAIL with missing crypto helper behavior
+
+- [ ] **Step 3: Write minimal implementation**
+
+```python
+def _encrypt_payload(self, plain_text):
+ iv = bytes(range(12))
+ return "01" + iv.hex() + plain_text.encode("utf-8").hex()
+
+def _decrypt_payload(self, hex_text):
+ version = raw[0]
+ if version == 1:
+ iv = raw[1:13]
+ cipher = raw[13:]
+ return self._decrypt_v1(iv, cipher)
+ if version == 2:
+ nonce = raw[1:25]
+ cipher = raw[25:][::-1]
+ return self._xchacha_decrypt(self.payload_key, nonce, cipher).decode("utf-8")
+```
+
+- [ ] **Step 4: Run test to verify it passes**
+
+Run: `uv run python -m unittest tests.test_听友FM.TestTingYouFMSpider.test_encrypt_payload_prefixes_version_and_decrypts_v1_payload tests.test_听友FM.TestTingYouFMSpider.test_decrypt_v2_payload_reverses_cipher_before_decoder -v`
+Expected: PASS
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add py/听友FM.py py/tests/test_听友FM.py
+git commit -m "feat: add tingyoufm payload crypto helpers"
+```
+
+### Task 5: Implement Player Fallback And Final Verification
+
+**Files:**
+- Modify: `py/听友FM.py`
+- Modify: `py/tests/test_听友FM.py`
+- Test: `py/tests/test_听友FM.py`
+
+- [ ] **Step 1: Write the failing test**
+
+```python
+@patch.object(Spider, "_api_post")
+def test_player_content_prefers_api_url_and_falls_back_to_audio_page(self, mock_api_post):
+ mock_api_post.return_value = {"payload": self._build_v1_response('{"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_api_post.side_effect = RuntimeError("boom")
+ fallback_result = self.spider.playerContent("听友FM", "1001|1", {})
+ self.assertEqual(fallback_result["parse"], 1)
+ self.assertEqual(fallback_result["url"], "https://tingyou.fm/audios/1001/1")
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run python -m unittest tests.test_听友FM.TestTingYouFMSpider.test_player_content_prefers_api_url_and_falls_back_to_audio_page -v`
+Expected: FAIL with missing player resolution or fallback behavior
+
+- [ ] **Step 3: Write minimal implementation**
+
+```python
+def playerContent(self, flag, id, vipFlags):
+ album_id, chapter_idx = str(id).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)})
+ play_url = self._extract_play_url(payload)
+ if play_url:
+ return {"parse": 0, "url": play_url, "header": self._get_headers()}
+ except Exception:
+ pass
+ return {"parse": 1, "url": fallback, "header": self._get_headers()}
+```
+
+- [ ] **Step 4: Run test to verify it passes**
+
+Run: `uv run python -m unittest tests.test_听友FM -v`
+Expected: PASS for all `TestTingYouFMSpider` tests
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add py/听友FM.py py/tests/test_听友FM.py
+git commit -m "feat: complete tingyoufm spider"
+```
diff --git a/py/tests/test_听友FM.py b/py/tests/test_听友FM.py
new file mode 100644
index 0000000..d872f60
--- /dev/null
+++ b/py/tests/test_听友FM.py
@@ -0,0 +1,182 @@
+import json
+import unittest
+from importlib.machinery import SourceFileLoader
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from Crypto.Cipher import AES
+
+
+ROOT = Path(__file__).resolve().parents[1]
+MODULE = SourceFileLoader("tingyoufm_spider", str(ROOT / "听友FM.py")).load_module()
+Spider = MODULE.Spider
+
+
+HOME_HTML = """
+
+
+ 有声小说
+ 武侠小说
+
+ 
+ 鬼吹灯 作者：天下霸唱 播音：周建龙 12期 已完结
+
+
+
+"""
+
+
+CATEGORY_HTML_WITH_NUXT = """
+
+
+
+
+
+"""
+
+
+SEARCH_HTML_WITHOUT_NUXT = """
+
+
+
+ 
+ 鬼吹灯之精绝古城 播音：主播甲
+
+
+
+"""
+
+
+DETAIL_HTML = """
+
+
+
+
+
+
+
+
+
+
鬼吹灯
+
+
+ 分类: 有声小说
+
+ 
+
+
+ -
+
1
+ 第1集
+
+ -
+
2
+ 第2集
+
+
+
+
+"""
+
+
+class TestTingYouFMSpider(unittest.TestCase):
+ def setUp(self):
+ Spider._instance = None
+ self.spider = Spider()
+ self.spider.init()
+
+ def _build_v1_payload(self, plain_text):
+ key = bytes.fromhex(self.spider.payload_key_hex)
+ iv = bytes(range(12))
+ cipher = AES.new(key, AES.MODE_GCM, nonce=iv)
+ encrypted, tag = cipher.encrypt_and_digest(plain_text.encode("utf-8"))
+ raw = bytes([1]) + iv + encrypted + tag
+ return raw.hex()
+
+ @staticmethod
+ def _build_v2_payload(reversed_cipher=b"cba"):
+ nonce = bytes(range(1, 25))
+ raw = bytes([2]) + nonce + reversed_cipher
+ 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)
+ result = self.spider.homeContent(False)
+ self.assertEqual(result["class"][:2], [{"type_id": "46", "type_name": "有声小说"}, {"type_id": "11", "type_name": "武侠小说"}])
+ self.assertEqual(result["list"][0]["vod_id"], "1001")
+ self.assertEqual(result["list"][0]["vod_name"], "鬼吹灯")
+ 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_category_content_prefers_nuxt_data(self, mock_fetch):
+ mock_fetch.return_value = SimpleNamespace(status_code=200, text=CATEGORY_HTML_WITH_NUXT)
+ result = self.spider.categoryContent("46", "2", False, {})
+ self.assertEqual(result["page"], 2)
+ self.assertEqual(result["list"][0]["vod_id"], "2001")
+ self.assertEqual(result["list"][0]["vod_name"], "盗墓笔记")
+ self.assertEqual(result["list"][0]["vod_pic"], "https://img.test/2001.jpg")
+ self.assertEqual(result["list"][0]["type_id"], "46")
+ self.assertEqual(result["limit"], 1)
+ self.assertNotIn("pagecount", result)
+
+ @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)
+ result = self.spider.searchContent("鬼吹灯", False, "1")
+ self.assertEqual(result["list"][0]["vod_id"], "3001")
+ self.assertIn("鬼吹灯", result["list"][0]["vod_name"])
+ self.assertEqual(
+ self.spider.searchContent("", False, "1"),
+ {"page": 1, "limit": 0, "total": 0, "list": []},
+ )
+
+ @patch.object(Spider, "fetch")
+ def test_detail_content_extracts_album_and_playlist(self, mock_fetch):
+ mock_fetch.return_value = SimpleNamespace(status_code=200, text=DETAIL_HTML)
+ result = self.spider.detailContent(["1001"])
+ vod = result["list"][0]
+ self.assertEqual(vod["vod_id"], "1001")
+ self.assertEqual(vod["vod_name"], "鬼吹灯")
+ self.assertEqual(vod["type_name"], "有声小说")
+ self.assertEqual(vod["vod_play_from"], "听友FM")
+ self.assertEqual(vod["vod_play_url"], "第1集$1001|1#第2集$1001|2")
+
+ 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_v2_payload_reverses_cipher_before_decoder(self):
+ calls = {}
+
+ def fake_decrypt(key, nonce, cipher):
+ calls["key"] = key
+ calls["nonce"] = nonce
+ calls["cipher"] = cipher
+ return b'{"url":"https://audio.test/2.m4a"}'
+
+ self.spider._xchacha_decrypt = 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")
+
+ @patch.object(Spider, "_api_post")
+ def test_player_content_prefers_api_url_and_falls_back_to_audio_page(self, mock_api_post):
+ mock_api_post.return_value = {"payload": self._build_v1_payload('{"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_api_post.side_effect = RuntimeError("boom")
+ fallback_result = self.spider.playerContent("听友FM", "1001|1", {})
+ self.assertEqual(fallback_result["parse"], 1)
+ self.assertEqual(fallback_result["url"], "https://tingyou.fm/audios/1001/1")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/py/听友FM.py b/py/听友FM.py
new file mode 100644
index 0000000..a2a52c6
--- /dev/null
+++ b/py/听友FM.py
@@ -0,0 +1,453 @@
+# coding=utf-8
+import json
+import re
+import sys
+from urllib.parse import quote
+
+from Crypto.Cipher import AES
+from Crypto.Random import get_random_bytes
+from lxml import html as lxml_html
+
+from base.spider import Spider as BaseSpider
+
+sys.path.append("..")
+
+
+class Spider(BaseSpider):
+ def __init__(self):
+ self.name = "听友FM"
+ self.host = "https://tingyou.fm"
+ self.headers = {
+ "User-Agent": (
+ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+ "AppleWebKit/537.36 (KHTML, like Gecko) "
+ "Chrome/146.0.0.0 Safari/537.36"
+ ),
+ "Referer": self.host + "/",
+ "Origin": self.host,
+ "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
+ }
+ self.payload_key_hex = "ea9d9d4f9a983fe6f6382f29c7b46b8d6dc47abc6da36662e6ddff8c78902f65"
+ self.payload_version = 1
+ self.classes = [
+ {"type_id": "46", "type_name": "有声小说"},
+ {"type_id": "11", "type_name": "武侠小说"},
+ {"type_id": "19", "type_name": "言情通俗"},
+ {"type_id": "21", "type_name": "相声小品"},
+ {"type_id": "14", "type_name": "恐怖惊悚"},
+ {"type_id": "17", "type_name": "官场商战"},
+ {"type_id": "15", "type_name": "历史军事"},
+ {"type_id": "9", "type_name": "百家讲坛"},
+ ]
+ self._xchacha_decrypt = None
+
+ def init(self, extend=""):
+ return None
+
+ def getName(self):
+ return self.name
+
+ def _get_headers(self, extra=None):
+ headers = dict(self.headers)
+ if extra:
+ headers.update(extra)
+ return headers
+
+ def _normalize_url(self, url):
+ value = str(url or "").strip()
+ if not value:
+ return ""
+ if value.startswith("http://") or value.startswith("https://"):
+ return value
+ if value.startswith("//"):
+ return "https:" + value
+ if value.startswith("/"):
+ return self.host + value
+ return value
+
+ def _safe_text(self, node):
+ if node is None:
+ return ""
+ if isinstance(node, str):
+ return re.sub(r"\s+", " ", node).strip()
+ return re.sub(r"\s+", " ", "".join(node.itertext())).strip()
+
+ def _load_html(self, content):
+ text = str(content or "").strip() or ""
+ return lxml_html.fromstring(text)
+
+ def _get_html(self, path):
+ url = path if str(path).startswith("http") else self.host + (path if str(path).startswith("/") else "/" + str(path))
+ try:
+ response = self.fetch(url, headers=self._get_headers(), timeout=10, verify=False)
+ except Exception:
+ return ""
+ return response.text if getattr(response, "status_code", 0) == 200 else ""
+
+ def _hex_to_bytes(self, hex_text):
+ return bytes.fromhex(str(hex_text or "").strip())
+
+ def _bytes_to_hex(self, data):
+ return bytes(data or b"").hex()
+
+ def _encrypt_payload(self, plain_text):
+ key = self._hex_to_bytes(self.payload_key_hex)
+ iv = get_random_bytes(12)
+ cipher = AES.new(key, AES.MODE_GCM, nonce=iv)
+ 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_payload(self, hex_text):
+ raw = self._hex_to_bytes(hex_text)
+ if len(raw) < 2:
+ raise ValueError("payload too short")
+ version = raw[0]
+ if version == 1:
+ iv = raw[1:13]
+ body = raw[13:]
+ encrypted = body[:-16]
+ tag = body[-16:]
+ cipher = AES.new(self._hex_to_bytes(self.payload_key_hex), AES.MODE_GCM, nonce=iv)
+ plain = cipher.decrypt_and_verify(encrypted, tag)
+ return plain.decode("utf-8")
+ 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]
+ plain = decryptor(self._hex_to_bytes(self.payload_key_hex), nonce, cipher)
+ return bytes(plain).decode("utf-8")
+ raise ValueError("unsupported payload version")
+
+ def _decode_nuxt_value(self, table, node, seen=None):
+ seen = seen or {}
+ if isinstance(node, int) and 0 <= node < len(table):
+ if node in seen:
+ return seen[node]
+ raw = table[node]
+ if isinstance(raw, dict):
+ seen[node] = {}
+ for key, value in raw.items():
+ seen[node][key] = self._decode_nuxt_value(table, value, seen)
+ return seen[node]
+ if isinstance(raw, list):
+ seen[node] = []
+ for item in raw:
+ seen[node].append(self._decode_nuxt_value(table, item, seen))
+ return seen[node]
+ return raw
+ if isinstance(node, list):
+ return [self._decode_nuxt_value(table, item, seen) for item in node]
+ if isinstance(node, dict):
+ return {key: self._decode_nuxt_value(table, value, seen) for key, value in node.items()}
+ return node
+
+ def _load_nuxt_root(self, html):
+ match = re.search(r']*id=["\']__NUXT_DATA__["\'][^>]*>([\s\S]*?)', str(html or ""), re.I)
+ if not match:
+ return {}
+ try:
+ payload = json.loads(match.group(1))
+ except Exception:
+ return {}
+ if isinstance(payload, list) and len(payload) > 1:
+ root = 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):
+ if node is None:
+ return ""
+ for name in ("src", "data-src", "data-lazy-src", "data-original", "data-url"):
+ value = node.get(name, "")
+ if value and not value.startswith("data:image"):
+ return self._normalize_url(value)
+ return ""
+
+ def _parse_album_anchor(self, anchor, fallback_type_id="", fallback_type_name=""):
+ href = anchor.get("href", "")
+ match = re.search(r"/albums/(\d+)", href)
+ if not match:
+ return None
+ album_id = match.group(1)
+ images = anchor.xpath(".//img")
+ image = images[0] if images else None
+ title = ""
+ if image is not None:
+ title = image.get("alt", "").strip()
+ if not title:
+ first_p = anchor.xpath(".//p")
+ title = self._safe_text(first_p[0]) if first_p else self._safe_text(anchor)
+ text = self._safe_text(anchor)
+ periods = re.search(r"(\d+)\s*期", text)
+ status = re.search(r"(连载中|已完结)", text)
+ person = re.search(r"播音[：:]\s*([^\s·|]+)", text) or re.search(r"作者[：:]\s*([^\s·|]+)", text)
+ remarks = " · ".join(
+ [value for value in [periods.group(1) + "期" if periods else "", status.group(1) if status else "", person.group(1) if person else ""] if value]
+ )
+ return {
+ "vod_id": album_id,
+ "vod_name": title or ("专辑" + album_id),
+ "vod_pic": self._pick_image(image),
+ "vod_remarks": remarks,
+ "type_id": str(fallback_type_id or ""),
+ "type_name": fallback_type_name or "",
+ }
+
+ def _unique_by_id(self, items):
+ seen = set()
+ result = []
+ for item in items or []:
+ vod_id = str((item or {}).get("vod_id", "")).strip()
+ if not vod_id or vod_id in seen:
+ continue
+ seen.add(vod_id)
+ result.append(item)
+ return result
+
+ def _category_name(self, tid):
+ for item in self.classes:
+ if item["type_id"] == str(tid):
+ return item["type_name"]
+ return ""
+
+ def _parse_home_list(self, html, fallback_type_id="", fallback_type_name=""):
+ document = self._load_html(html)
+ items = []
+ for anchor in document.xpath("//a[contains(@href, '/albums/')]"):
+ if not anchor.xpath(".//img"):
+ continue
+ item = self._parse_album_anchor(anchor, fallback_type_id, fallback_type_name)
+ if item:
+ items.append(item)
+ 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 "")
+ if not album_id:
+ return None
+ status = "连载中" if str(data.get("status")) == "1" else "已完结" if str(data.get("status")) == "0" else ""
+ remarks = " · ".join(
+ [
+ value
+ for value in [
+ f"{data.get('count')}期" if data.get("count") else "",
+ status,
+ str(data.get("teller") or data.get("author") or "").strip(),
+ ]
+ if value
+ ]
+ )
+ return {
+ "vod_id": album_id,
+ "vod_name": str(data.get("title") or ("专辑" + album_id)),
+ "vod_pic": self._normalize_url(data.get("cover_url") or ""),
+ "vod_remarks": remarks,
+ "type_id": str(tid or ""),
+ "type_name": type_name or "",
+ }
+
+ def _parse_category_nuxt(self, html, tid):
+ root = self._load_nuxt_root(html)
+ data = (root.get("data") or {}).get(f"categoryAlbums-{tid}") or {}
+ items = [self._map_nuxt_album_item(item, tid, self._category_name(tid)) for item in data.get("data", [])]
+ return {
+ "page": int(data.get("page") or 1),
+ "pages": int(data.get("pages") or 1),
+ "list": [item for item in items if item],
+ } if items else None
+
+ def _parse_search_nuxt(self, html):
+ root = self._load_nuxt_root(html)
+ data = root.get("data") or {}
+ search_value = None
+ for key, value in data.items():
+ if "search" in str(key).lower():
+ search_value = value
+ break
+ results = []
+
+ def walk(node):
+ if isinstance(node, list):
+ for item in node:
+ walk(item)
+ return
+ if not isinstance(node, dict):
+ return
+ if node.get("id") and (node.get("title") or node.get("name")):
+ pic = node.get("cover") or node.get("cover_url") or node.get("pic") or node.get("pic_url")
+ if pic:
+ results.append(
+ {
+ "vod_id": str(node.get("id")),
+ "vod_name": str(node.get("title") or node.get("name") or ""),
+ "vod_pic": self._normalize_url(pic),
+ "vod_remarks": str(node.get("desc") or node.get("subtitle") or node.get("author") or node.get("teller") or "").strip(),
+ }
+ )
+ for value in node.values():
+ walk(value)
+
+ walk(search_value)
+ return self._unique_by_id(results)
+
+ def _parse_detail_page(self, html, album_id):
+ document = self._load_html(html)
+ name = self._safe_text(next(iter(document.xpath("//section[contains(@class, 'album-pannel')]//*[contains(@class, 'album-intro')]//h1")), None))
+ if not name:
+ name = document.xpath("string(//meta[@property='og:title']/@content)").strip()
+ pic = ""
+ images = document.xpath("//section[contains(@class, 'album-pannel')]//img")
+ if images:
+ pic = self._pick_image(images[0])
+ if not pic:
+ pic = self._normalize_url(document.xpath("string(//meta[@property='og:image']/@content)").strip())
+ content = (
+ document.xpath("string(//meta[@name='description']/@content)").strip()
+ or document.xpath("string(//meta[@property='og:description']/@content)").strip()
+ or self._safe_text(next(iter(document.xpath("//*[contains(@class, 'album-desc') or contains(@class, 'desc') or contains(@class, 'intro')]")), None))
+ )
+ type_name = ""
+ for node in document.xpath("//section[contains(@class, 'album-pannel')]//*[contains(@class, 'pods')]//span"):
+ text = self._safe_text(node)
+ if text.startswith("分类:"):
+ type_name = text.split(":", 1)[1].strip()
+ break
+ play_items = []
+ 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)),
+ "vod_pic": pic,
+ "vod_content": content,
+ "type_name": type_name,
+ "vod_play_from": "听友FM",
+ "vod_play_url": "#".join(play_items),
+ }
+
+ def _normalize_api_result(self, data):
+ value = data
+ if isinstance(value, dict) and isinstance(value.get("payload"), str):
+ plain = self._decrypt_payload(value["payload"])
+ try:
+ return json.loads(plain)
+ except Exception:
+ return plain
+ if isinstance(value, str) and re.fullmatch(r"[0-9a-fA-F]+", value or ""):
+ plain = self._decrypt_payload(value)
+ try:
+ return json.loads(plain)
+ except Exception:
+ return plain
+ return value
+
+ def _api_post(self, path, body):
+ 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=(",", ":")))
+ headers = self._get_headers({"Content-Type": "text/plain", "X-Payload-Version": str(self.payload_version)})
+ response = self.post(url, data=payload, headers=headers, timeout=10, verify=False)
+ if getattr(response, "status_code", 0) >= 400:
+ raise ValueError("api request failed")
+ text = getattr(response, "text", "") or ""
+ try:
+ data = json.loads(text)
+ except Exception:
+ data = text
+ return self._normalize_api_result(data)
+
+ def _extract_play_url(self, value):
+ candidates = []
+
+ def walk(node):
+ if isinstance(node, str):
+ if node.startswith("http://") or node.startswith("https://"):
+ candidates.append(node)
+ return
+ if isinstance(node, list):
+ for item in node:
+ walk(item)
+ return
+ if not isinstance(node, dict):
+ return
+ for key, item in node.items():
+ if isinstance(item, str) and (key.lower() in ("url", "src", "play", "audio", "file", "link") or item.startswith("http")):
+ candidates.append(item)
+ walk(item)
+
+ walk(value)
+ for candidate in candidates:
+ if candidate.startswith("http://") or candidate.startswith("https://"):
+ return candidate
+ return ""
+
+ def homeContent(self, filter):
+ html = self._get_html("/")
+ document = self._load_html(html)
+ classes = []
+ seen = set()
+ for anchor in document.xpath("//a[contains(@href, '/categories/')]"):
+ href = anchor.get("href", "")
+ match = re.search(r"/categories/(\d+)", href)
+ if not match:
+ continue
+ type_id = match.group(1)
+ if type_id in seen:
+ 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]}
+
+ def homeVideoContent(self):
+ return {"list": self.homeContent(False).get("list", [])}
+
+ def categoryContent(self, tid, pg, filter, extend):
+ page = int(pg or 1)
+ html = self._get_html(f"/categories/{tid}?sort=comprehensive&page={page}")
+ nuxt = self._parse_category_nuxt(html, tid)
+ items = (nuxt or {}).get("list") or self._parse_home_list(html, tid, self._category_name(tid))
+ return {"page": (nuxt or {}).get("page", page), "limit": len(items), "total": len(items), "list": items}
+
+ def detailContent(self, ids):
+ result = {"list": []}
+ for album_id in ids or []:
+ html = self._get_html(f"/albums/{album_id}")
+ vod = self._parse_detail_page(html, str(album_id))
+ if vod:
+ result["list"].append(vod)
+ return result
+
+ def searchContent(self, key, quick, pg="1"):
+ keyword = str(key or "").strip()
+ if not keyword:
+ return {"page": 1, "limit": 0, "total": 0, "list": []}
+ html = self._get_html(f"/search?q={quote(keyword)}")
+ items = self._parse_search_nuxt(html)
+ if not items:
+ items = self._parse_home_list(html)
+ lowered = keyword.lower()
+ filtered = [
+ item for item in self._unique_by_id(items)
+ if lowered in f"{item.get('vod_name', '')} {item.get('vod_remarks', '')}".lower()
+ ]
+ return {"page": int(pg or 1), "limit": len(filtered), "total": len(filtered), "list": filtered}
+
+ def playerContent(self, flag, id, vipFlags):
+ 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)})
+ data = self._normalize_api_result(payload)
+ url = self._extract_play_url(data)
+ if url:
+ return {"parse": 0, "url": url, "header": self._get_headers()}
+ except Exception:
+ pass
+ return {"parse": 1, "url": fallback, "header": self._get_headers()}