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 = """ + + + + + + + +
+
+

鬼吹灯

+
+
+ 分类: 有声小说 +
+ +
+ + + +""" + + +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()}