yeyinly/tvboxzt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: add tingyoufm spider

Browse Source
This commit is contained in:
Harold
2026-04-24 15:21:19 +08:00
parent 0de607c2ad
commit 0d658a368b
3 changed files with 925 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/docs/superpowers/plans/2026-04-24-tingyoufm-spider.md
+290
View File
@@ -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"
```
py/tests/test_听友FM.py
+182
View File
@@ -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 = """
<html>
<body>
<a href="/categories/46">有声小说</a>
<a href="/categories/11">武侠小说</a>
<a href="/albums/1001">
<img class="cover" src="/cover1.jpg" alt="鬼吹灯">
<p>鬼吹灯 作者：天下霸唱 播音：周建龙 12期 已完结</p>
</a>
</body>
</html>
"""
CATEGORY_HTML_WITH_NUXT = """
<html>
<body>
<script id="__NUXT_DATA__" type="application/json">[null,{"data":{"categoryAlbums-46":{"page":2,"pages":9,"data":[{"id":2001,"title":"盗墓笔记","cover_url":"https://img.test/2001.jpg","count":20,"status":1,"teller":"艾宝良"}]}}}]</script>
</body>
</html>
"""
SEARCH_HTML_WITHOUT_NUXT = """
<html>
<body>
<a href="/albums/3001">
<img class="cover" src="https://img.test/3001.jpg" alt="鬼吹灯之精绝古城">
<p>鬼吹灯之精绝古城 播音：主播甲</p>
</a>
</body>
</html>
"""
DETAIL_HTML = """
<html>
<head>
<meta property="og:title" content="鬼吹灯">
<meta property="og:image" content="https://img.test/detail.jpg">
<meta name="description" content="摸金校尉探险故事">
</head>
<body>
<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">
<li class="chapter-item">
<p>1</p>
<div class="item-content"><span class="title">第1集</span></div>
</li>
<li class="chapter-item">
<p>2</p>
<div class="item-content"><span class="title">第2集</span></div>
</li>
</ul>
</body>
</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()
py/听友FM.py
+453
View File
@@ -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 "<html></html>"
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'<script[^>]*id=["\']__NUXT_DATA__["\'][^>]*>([\s\S]*?)</script>', 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()}