From b319bfc785cb931bcd65813b31cdfb34cc4029d2 Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Sat, 18 Apr 2026 13:13:49 +0800 Subject: [PATCH] feat: add libvio spider --- .../plans/2026-04-18-libvio-spider.md | 524 ++++++++++++++++++ .../specs/2026-04-18-libvio-spider-design.md | 280 ++++++++++ py/libvio.py | 292 ++++++++++ py/tests/test_libvio.py | 156 ++++++ 4 files changed, 1252 insertions(+) create mode 100644 py/docs/superpowers/plans/2026-04-18-libvio-spider.md create mode 100644 py/docs/superpowers/specs/2026-04-18-libvio-spider-design.md create mode 100644 py/libvio.py create mode 100644 py/tests/test_libvio.py diff --git a/py/docs/superpowers/plans/2026-04-18-libvio-spider.md b/py/docs/superpowers/plans/2026-04-18-libvio-spider.md new file mode 100644 index 0000000..83ff6a7 --- /dev/null +++ b/py/docs/superpowers/plans/2026-04-18-libvio-spider.md @@ -0,0 +1,524 @@ +# LibVIO 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:** 在当前 Python 仓库中新增一个符合 `base.spider.Spider` 接口的 LibVIO 爬虫,支持 `home/category/detail/search/player` 全链路。 + +**Architecture:** 采用单文件站点脚本 `libvio.py` 承担 LibVIO 规则,内部拆分为 URL 归一化、列表卡片解析、详情字段解析、播放配置提取与播放器 API 解析几组辅助方法。测试沿用当前仓库 `unittest + SourceFileLoader + mock` 风格,优先覆盖纯解析函数和高层方法的 mock 网络流程,不依赖真实站点网络。 + +**Tech Stack:** Python 3, `requests`, `lxml`, `unittest`, `unittest.mock`, `json`, `re`, `urllib.parse` + +--- + +## File Structure + +- Create: `libvio.py` + - LibVIO 站点实现,继承 `base.spider.Spider` + - 暴露 `init`、`homeContent`、`homeVideoContent`、`categoryContent`、`detailContent`、`searchContent`、`playerContent` + - 私有方法负责 URL/ID 归一化、列表卡片解析、详情解析、播放配置与 API 解析 +- Create: `tests/test_libvio.py` + - 用 `SourceFileLoader` 加载 `libvio.py` + - 用 HTML/JS 片段与 mock response 测试首页、分类、搜索、详情和播放器解析 + +### Task 1: Scaffold Spider, Home Flow, And List/Search Parsing + +**Files:** +- Create: `tests/test_libvio.py` +- Create: `libvio.py` + +- [ ] **Step 1: Write the failing test** + +```python +import unittest +from importlib.machinery import SourceFileLoader +from pathlib import Path +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] +MODULE = SourceFileLoader("libvio_spider", str(ROOT / "libvio.py")).load_module() +Spider = MODULE.Spider + + +class TestLibVioSpider(unittest.TestCase): + def setUp(self): + self.spider = Spider() + self.spider.init() + + def test_home_content_exposes_expected_categories(self): + content = self.spider.homeContent(False) + class_ids = [item["type_id"] for item in content["class"]] + self.assertEqual(class_ids, ["index", "movie", "series", "anime", "jpandkr", "euandus"]) + + def test_parse_list_cards_extracts_compact_vod_id(self): + html = """ +
+ + 更新至10集 +
+ """ + cards = self.spider._parse_list_cards(html) + self.assertEqual( + cards, + [{ + "vod_id": "456", + "vod_name": "示例影片", + "vod_pic": "https://libvio.site/cover.jpg", + "vod_remarks": "更新至10集", + }], + ) + + @patch.object(Spider, "_request_html") + def test_home_video_content_reuses_list_card_parser(self, mock_request_html): + mock_request_html.return_value = """ +
+ + HD +
+ """ + result = self.spider.homeVideoContent() + self.assertEqual(result["list"][0]["vod_id"], "111") + self.assertEqual(result["list"][0]["vod_name"], "最近更新") + + @patch.object(Spider, "_request_html") + def test_category_content_builds_page_result(self, mock_request_html): + mock_request_html.return_value = """ +
+ + 完结 +
+ """ + result = self.spider.categoryContent("movie", "2", False, {}) + self.assertEqual(result["page"], 2) + self.assertEqual(result["list"][0]["vod_id"], "222") + + @patch.object(Spider, "_request_html") + def test_search_content_reuses_card_parser(self, mock_request_html): + mock_request_html.return_value = """ +
+ + 抢先版 +
+ """ + result = self.spider.searchContent("繁花", False, "1") + self.assertEqual(result["list"][0]["vod_id"], "333") + self.assertEqual(result["list"][0]["vod_name"], "搜索影片") +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `python -m unittest tests.test_libvio.TestLibVioSpider -v` +Expected: FAIL with `FileNotFoundError` for `libvio.py` or missing methods. + +- [ ] **Step 3: Write minimal implementation** + +```python +# coding=utf-8 +import re +import sys +from urllib.parse import quote + +from base.spider import Spider as BaseSpider + +sys.path.append("..") + + +class Spider(BaseSpider): + def __init__(self): + self.name = "LibVIO" + self.host = "https://libvio.site" + self.headers = { + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/124.0.0.0 Safari/537.36" + ) + } + self.categories = [ + {"type_name": "最近更新", "type_id": "index"}, + {"type_name": "电影", "type_id": "movie"}, + {"type_name": "电视剧", "type_id": "series"}, + {"type_name": "动漫", "type_id": "anime"}, + {"type_name": "日韩剧", "type_id": "jpandkr"}, + {"type_name": "欧美剧", "type_id": "euandus"}, + ] + self.category_paths = { + "index": "/", + "movie": "/type/1-{pg}.html", + "series": "/type/2-{pg}.html", + "anime": "/type/4-{pg}.html", + "jpandkr": "/type/15-{pg}.html", + "euandus": "/type/16-{pg}.html", + } + + def init(self, extend=""): + return None + + def getName(self): + return self.name + + def homeContent(self, filter): + return {"class": self.categories} + + def _build_url(self, href): + raw = str(href or "").strip() + if not raw: + return "" + if raw.startswith(("http://", "https://")): + return raw + if raw.startswith("//"): + return "https:" + raw + return self.host + "/" + raw.lstrip("/") + + def _extract_vod_id(self, href): + raw = str(href or "").strip() + matched = re.search(r"/detail/(\d+)\.html", raw) + if matched: + return matched.group(1) + if re.fullmatch(r"\d+", raw): + return raw + return "" + + def _parse_list_cards(self, html): + root = self.html(html) + results = [] + if root is None: + return results + for card in root.xpath("//*[contains(@class,'stui-vodlist__box')]"): + href = ((card.xpath(".//a[@href][1]/@href") or [""])[0]).strip() + vod_id = self._extract_vod_id(href) + title = ((card.xpath(".//a[@title][1]/@title") or [""])[0]).strip() + pic = ( + (card.xpath(".//a[@data-original][1]/@data-original") or [""])[0].strip() + or (card.xpath(".//img[@data-original][1]/@data-original") or [""])[0].strip() + or (card.xpath(".//img[@src][1]/@src") or [""])[0].strip() + ) + remarks = "".join(card.xpath(".//*[contains(@class,'pic-text')][1]//text()")).strip() + if not vod_id or not title: + continue + results.append({ + "vod_id": vod_id, + "vod_name": title, + "vod_pic": self._build_url(pic), + "vod_remarks": remarks, + }) + return results + + def _request_html(self, path_or_url, expect_xpath=None, referer=None): + target = path_or_url if str(path_or_url).startswith("http") else self._build_url(path_or_url) + headers = dict(self.headers) + headers["Referer"] = referer or (self.host + "/") + response = self.fetch(target, headers=headers, timeout=10) + if response.status_code != 200: + return "" + return response.text or "" + + def _page_result(self, items, pg): + page = int(pg) + pagecount = page + 1 if items else page + return {"list": items, "page": page, "pagecount": pagecount, "limit": len(items), "total": pagecount * max(len(items), 1)} + + def homeVideoContent(self): + html = self._request_html("/", expect_xpath="//*[contains(@class,'stui-vodlist__box')]") + return {"list": self._parse_list_cards(html)} + + def categoryContent(self, tid, pg, filter, extend): + path = self.category_paths.get(tid, self.category_paths["movie"]).format(pg=pg) + html = self._request_html(path, expect_xpath="//*[contains(@class,'stui-vodlist__box')]") + return self._page_result(self._parse_list_cards(html), pg) + + def searchContent(self, key, quick, pg="1"): + path = "/search/-------------.html?wd={0}".format(quote(key)) + html = self._request_html(path, expect_xpath="//*[contains(@class,'stui-vodlist__box')]") + return self._page_result(self._parse_list_cards(html), pg) +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `python -m unittest tests.test_libvio.TestLibVioSpider -v` +Expected: PASS for the five new tests. + +- [ ] **Step 5: Commit** + +```bash +git add tests/test_libvio.py libvio.py +git commit -m "feat: scaffold libvio list parsing" +``` + +### Task 2: Add Detail Parsing And Compact Play IDs + +**Files:** +- Modify: `tests/test_libvio.py` +- Modify: `libvio.py` + +- [ ] **Step 1: Write the failing test** + +```python +class TestLibVioSpider(unittest.TestCase): + def test_parse_detail_page_extracts_fields_and_filters_pan_sources(self): + html = """ +
+ +
+
+

示例剧

+

类型:剧情

+

地区:大陆

+

年份:2026

+

导演:张三

+

主演:李四王五

+

简介:一段剧情简介

+
+

在线播放

+ +

夸克资源

+ + """ + result = self.spider._parse_detail_page(html, "999") + vod = result["list"][0] + self.assertEqual(vod["vod_id"], "999") + self.assertEqual(vod["path"], "https://libvio.site/detail/999.html") + self.assertEqual(vod["vod_name"], "示例剧") + self.assertEqual(vod["type_name"], "剧情") + self.assertEqual(vod["vod_area"], "大陆") + self.assertEqual(vod["vod_year"], "2026") + self.assertEqual(vod["vod_director"], "张三") + self.assertEqual(vod["vod_actor"], "李四,王五") + self.assertEqual(vod["vod_content"], "一段剧情简介") + self.assertEqual(vod["vod_play_from"], "LibVIO") + self.assertEqual(vod["vod_play_url"], "第1集$999-1-1#第2集$999-1-2") + + @patch.object(Spider, "_request_html") + def test_detail_content_builds_detail_request_url_from_vod_id(self, mock_request_html): + mock_request_html.return_value = '

详情影片

' + result = self.spider.detailContent(["123"]) + self.assertEqual(mock_request_html.call_args.args[0], "https://libvio.site/detail/123.html") + self.assertEqual(result["list"][0]["vod_id"], "123") +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `python -m unittest tests.test_libvio.TestLibVioSpider.test_parse_detail_page_extracts_fields_and_filters_pan_sources tests.test_libvio.TestLibVioSpider.test_detail_content_builds_detail_request_url_from_vod_id -v` +Expected: FAIL with missing `_parse_detail_page`, `detailContent`, or incorrect field values. + +- [ ] **Step 3: Write minimal implementation** + +```python + def _extract_play_id(self, href): + raw = str(href or "").strip() + matched = re.search(r"/play/([^./]+-\d+-\d+)\.html", raw) + if matched: + return matched.group(1) + if re.fullmatch(r"[^./]+-\d+-\d+", raw): + return raw + return "" + + def _build_detail_request_url(self, vod_id): + return f"{self.host}/detail/{self._extract_vod_id(vod_id)}.html" + + def _build_play_request_url(self, play_id): + return f"{self.host}/play/{self._extract_play_id(play_id)}.html" + + def _clean_text(self, text): + return re.sub(r"\s+", " ", str(text or "").replace("\xa0", " ")).strip() + + def _extract_detail_field(self, root, label, joiner=""): + nodes = root.xpath(f'.//span[contains(normalize-space(.), "{label}:")]') + if not nodes: + return "" + values = [] + for sibling in nodes[0].itersiblings(): + if sibling.tag == "span" and "text-muted" in " ".join(sibling.xpath("./@class")): + break + text = self._clean_text("".join(sibling.xpath(".//text()"))) + if text: + values.append(text) + values = [value for index, value in enumerate(values) if value and value not in values[:index]] + return joiner.join(values) if joiner else "".join(values) + + def _parse_detail_page(self, html, vod_id): + root = self.html(html) + detail_root = (root.xpath("//*[contains(@class,'stui-content__detail')][1]") or [root])[0] + title = ((detail_root.xpath(".//*[contains(@class,'title')][1]//text()") or [""])[0]).strip() + pic = ( + (root.xpath("//*[contains(@class,'stui-content__thumb')]//img/@data-original") or [""])[0].strip() + or (root.xpath("//*[contains(@class,'stui-content__thumb')]//img/@src") or [""])[0].strip() + ) + episodes = [] + for playlist in root.xpath("//*[contains(@class,'stui-content__playlist')]"): + heading = self._clean_text("".join(playlist.xpath("./preceding-sibling::*[1]//text()"))) + if any(keyword in heading for keyword in ("夸克", "UC", "网盘")): + continue + for anchor in playlist.xpath(".//a[@href]"): + play_id = self._extract_play_id((anchor.xpath("./@href") or [""])[0]) + name = self._clean_text("".join(anchor.xpath(".//text()"))) + if play_id and name: + episodes.append(f"{name}${play_id}") + vod = { + "vod_id": vod_id, + "path": self._build_detail_request_url(vod_id), + "vod_name": title, + "vod_pic": self._build_url(pic), + "vod_tag": "", + "vod_time": "", + "vod_remarks": "", + "vod_play_from": "LibVIO", + "vod_play_url": "#".join(episodes), + "type_name": self._extract_detail_field(detail_root, "类型"), + "vod_content": self._extract_detail_field(detail_root, "简介"), + "vod_year": self._extract_detail_field(detail_root, "年份"), + "vod_area": self._extract_detail_field(detail_root, "地区"), + "vod_lang": "", + "vod_director": self._extract_detail_field(detail_root, "导演", joiner=","), + "vod_actor": self._extract_detail_field(detail_root, "主演", joiner=","), + } + return {"list": [vod]} + + def detailContent(self, ids): + vod_id = ids[0] + html = self._request_html(self._build_detail_request_url(vod_id), expect_xpath="//*[contains(@class,'stui-content__playlist')]") + return self._parse_detail_page(html, vod_id) +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `python -m unittest tests.test_libvio.TestLibVioSpider.test_parse_detail_page_extracts_fields_and_filters_pan_sources tests.test_libvio.TestLibVioSpider.test_detail_content_builds_detail_request_url_from_vod_id -v` +Expected: PASS for the two new tests. + +- [ ] **Step 5: Commit** + +```bash +git add tests/test_libvio.py libvio.py +git commit -m "feat: add libvio detail parsing" +``` + +### Task 3: Add Player Config Parsing And Final URL Resolution + +**Files:** +- Modify: `tests/test_libvio.py` +- Modify: `libvio.py` + +- [ ] **Step 1: Write the failing test** + +```python +class TestLibVioSpider(unittest.TestCase): + def test_extract_player_config_reads_json_assignment(self): + html = '' + data = self.spider._parse_player_config(html) + self.assertEqual(data["from"], "line") + + def test_extract_play_api_base_reads_player_js(self): + body = 'var player={}; src="/player/api.php?url=";' + self.assertEqual(self.spider._extract_play_api_base(body), "https://libvio.site/player/api.php?url=") + + @patch.object(Spider, "_request_html") + def test_player_content_resolves_direct_api_url(self, mock_request_html): + mock_request_html.side_effect = [ + '', + 'src="/player/api.php?url="', + 'var urls="https://video.example/final.m3u8";', + ] + result = self.spider.playerContent("LibVIO", "999-1-1", {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["url"], "https://video.example/final.m3u8") + self.assertEqual(result["header"]["Referer"], "https://libvio.site/") + + @patch.object(Spider, "_request_html") + def test_player_content_returns_empty_for_pan_source(self, mock_request_html): + mock_request_html.return_value = '' + self.assertEqual(self.spider.playerContent("LibVIO", "999-1-1", {}), {"parse": 0, "playUrl": "", "url": ""}) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `python -m unittest tests.test_libvio.TestLibVioSpider.test_extract_player_config_reads_json_assignment tests.test_libvio.TestLibVioSpider.test_extract_play_api_base_reads_player_js tests.test_libvio.TestLibVioSpider.test_player_content_resolves_direct_api_url tests.test_libvio.TestLibVioSpider.test_player_content_returns_empty_for_pan_source -v` +Expected: FAIL with missing player helpers or unresolved final URL. + +- [ ] **Step 3: Write minimal implementation** + +```python +import json + + def _parse_player_config(self, html): + matched = re.search(r"player_[a-z0-9_]+\s*=\s*(\{[\s\S]*?\})\s*;?", html, re.I) + if not matched: + return None + try: + return json.loads(matched.group(1)) + except Exception: + return None + + def _extract_play_api_base(self, body): + matched = re.search(r'src\s*=\s*["\']([^"\']+)["\']', body, re.I) + if not matched: + return "" + return self._build_url(matched.group(1)) + + def _extract_playable_url(self, body): + patterns = [ + r'["\']?urls?["\']?\s*:\s*["\']([^"\']+)["\']', + r'(?:var|let|const)\s+urls?\s*=\s*["\']([^"\']+)["\']', + r'["\']url["\']\s*:\s*["\']([^"\']+\.(?:m3u8|mp4)[^"\']*)["\']', + r'url\s*=\s*["\']([^"\']+\.(?:m3u8|mp4)[^"\']*)["\']', + ] + for pattern in patterns: + matched = re.search(pattern, body, re.I) + if matched: + return self._build_url(matched.group(1).replace("\\/", "/").replace("&", "&")) + return "" + + def _request_player_js(self, source): + return self._request_html(f"/static/player/{source}.js", referer=self.host + "/") + + def playerContent(self, flag, id, vipFlags): + play_page_url = self._build_play_request_url(id) + detail_html = self._request_html(play_page_url, referer=self.host + "/") + config = self._parse_player_config(detail_html) + if not config or config.get("from") in ("kuake", "uc"): + return {"parse": 0, "playUrl": "", "url": ""} + if config.get("from") == "ty_new1": + api_body = self._request_html(f"/vid/ty4.php?url={config.get('url', '')}", referer=self.host + "/") + final_url = self._extract_playable_url(api_body) + else: + player_js = self._request_player_js(config.get("from", "")) + api_base = self._extract_play_api_base(player_js) + api_url = "{base}{url}&next={next}&id={id}&nid={nid}".format( + base=api_base, + url=config.get("url", ""), + next=config.get("link_next", ""), + id=config.get("id", ""), + nid=config.get("nid", ""), + ) + api_body = self._request_html(api_url, referer=self.host + "/") + final_url = self._extract_playable_url(api_body) + if not final_url: + return {"parse": 0, "playUrl": "", "url": ""} + return { + "parse": 0, + "playUrl": "", + "url": final_url, + "header": {"User-Agent": self.headers["User-Agent"], "Referer": self.host + "/"}, + } +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `python -m unittest tests.test_libvio.TestLibVioSpider.test_extract_player_config_reads_json_assignment tests.test_libvio.TestLibVioSpider.test_extract_play_api_base_reads_player_js tests.test_libvio.TestLibVioSpider.test_player_content_resolves_direct_api_url tests.test_libvio.TestLibVioSpider.test_player_content_returns_empty_for_pan_source -v` +Expected: PASS for the four new tests. + +- [ ] **Step 5: Run the full suite** + +Run: `python -m unittest tests.test_libvio -v` +Expected: PASS with all `libvio` tests green. + +- [ ] **Step 6: Commit** + +```bash +git add tests/test_libvio.py libvio.py +git commit -m "feat: add libvio player parsing" +``` diff --git a/py/docs/superpowers/specs/2026-04-18-libvio-spider-design.md b/py/docs/superpowers/specs/2026-04-18-libvio-spider-design.md new file mode 100644 index 0000000..2d1a30e --- /dev/null +++ b/py/docs/superpowers/specs/2026-04-18-libvio-spider-design.md @@ -0,0 +1,280 @@ +# LibVIO Python 爬虫设计 + +## 目标 + +在当前 Python 仓库中新增一个符合 `base.spider.Spider` 接口的 LibVIO 站点爬虫,覆盖以下能力: + +- 首页分类 +- 首页近期列表 +- 分类列表 +- 详情页 +- 搜索 +- 播放解析 + +实现基于网页 DOM 抓取与播放页脚本解析,不依赖 Playwright,不修改 `base/` 公共层。 + +## 范围 + +本次实现包含: + +- 新增独立脚本,暂定文件名为 `libvio.py` +- 使用单一站点主域:`https://libvio.site` +- 支持首页、分类、搜索、详情和站内播放解析 + +本次实现不包含: + +- 多域名自动回退 +- 网盘资源解析 +- 大规模并发探测可播线路 +- 通用资源站抽象框架 + +## 方案选择 + +采用混合方案: + +- 列表、详情、搜索使用 `requests + lxml` 直接解析 HTML +- 播放解析吸收参考插件 `plugin_libvio` 中已验证的 `player_*` 配置、播放器 JS 和中间 API 解析逻辑 + +不直接逐行平移参考 JS,原因是: + +- 当前仓库已有 Python 爬虫风格,宜保持单文件站点实现 +- 参考 JS 中包含探测候选源、并发验证等偏重逻辑,Python 版先只保留稳定主链路 +- 先满足常见播放线路的可维护实现,避免把站点特例扩散到整个仓库 + +## 模块边界 + +新增脚本只在站点文件内部维护逻辑,不修改 `base/`。 + +脚本内部职责拆分如下: + +- `init` + - 初始化主域、请求头、分类映射 +- `homeContent` + - 返回固定分类 +- `homeVideoContent` + - 抓取首页最近更新列表 +- `categoryContent` + - 请求分类页并解析媒体卡片 +- `searchContent` + - 请求搜索页并解析搜索结果 +- `detailContent` + - 提取影片基础信息和剧集列表 +- `playerContent` + - 解析剧集页中的 `player_*` 配置并拿到最终播放地址 +- 私有辅助函数 + - URL 归一化 + - 列表卡片解析 + - 详情字段解析 + - 播放页配置提取 + - 播放器 JS 基址提取 + - API 播放地址提取 + +## Host 与分类策略 + +本次只实现单域: + +- `https://libvio.site` + +请求失败时不做多域切换,只返回空结果或空播放地址。 + +分类沿用参考插件配置: + +- `index` +- `movie` +- `series` +- `anime` +- `jpandkr` +- `euandus` + +分类 URL 映射如下: + +- `index -> /` +- `movie -> /type/1-{pg}.html` +- `series -> /type/2-{pg}.html` +- `anime -> /type/4-{pg}.html` +- `jpandkr -> /type/15-{pg}.html` +- `euandus -> /type/16-{pg}.html` + +其中: + +- `homeContent` 返回上述固定分类 +- `homeVideoContent` 抓取首页并提取最近更新条目 + +## 列表与搜索解析 + +### 首页与分类页 + +首页和分类页主要解析 `stui-vodlist__box` 卡片。 + +每张卡片提取规则: + +- 链接:选择 `href` 含详情页路径的主链接 +- 标题:优先链接 `title`,回退到节点文本 +- 封面:优先 `data-original`,其次 `src` +- 描述:优先类名含 `pic-text` 的文本 + +输出字段: + +- `vod_id` + - 返回紧凑数字 id,不返回完整详情 URL +- `vod_name` +- `vod_pic` +- `vod_remarks` + +列表解析时跳过明显网盘提示项,避免把不可直播资源混进常规列表。 + +### 搜索页 + +搜索 URL: + +- `/search/-------------.html?wd=` + +搜索结果优先按首页同类卡片结构解析,保持输出字段与分类页一致。 + +### 分页策略 + +返回分页字段: + +- `page = 当前页` +- `pagecount = pg + 1`,若当页无内容则为当前页 +- `limit = 实际条目数` +- `total = 近似值` + +不依赖站点总数统计。 + +## 详情页设计 + +`detailContent` 使用 `vod_id` 在内部组装详情页 URL 并请求页面。 + +提取字段: + +- `vod_id` +- `path` +- `vod_name` +- `vod_pic` +- `vod_tag` +- `vod_time` +- `vod_remarks` +- `vod_play_from` +- `vod_play_url` +- `type_name` +- `vod_content` +- `vod_year` +- `vod_area` +- `vod_lang` +- `vod_director` +- `vod_actor` + +不返回: + +- `dbid` +- `type` + +详情字段优先从详情主块中按标签解析,兼容纯文本行与结构化节点两种形式。 + +### 播放线路 + +播放列表只保留站内可播源: + +- 解析 `stui-content__playlist` +- 跳过标题或分组中明显标识为网盘、夸克、UC 的资源 +- `vod_play_from` 用 `$$$` 拼接线路名 +- `vod_play_url` 用 `$$$` 对齐对应线路剧集 + +单个剧集项格式: + +- `标题$播放id` + +其中 `播放id` 为紧凑值,由 `playerContent` 再组装完整播放页 URL。 + +## 播放解析设计 + +LibVIO 播放解析核心不是详情页直链,而是播放页里的 `player_*` 配置和播放器脚本。 + +实现步骤: + +1. 组装播放页 URL 并请求 +2. 提取页面脚本中的 `player_*` JSON +3. 读取关键字段: + - `url` + - `from` + - `link_next` + - `id` + - `nid` +4. 如果 `from` 属于网盘类(如 `kuake`、`uc`),直接返回空 +5. 如果 `from` 是站点特殊源(如 `ty_new1`),按固定 API 模式请求 +6. 否则请求 `/static/player/.js`,提取播放器 API 基址 +7. 根据不同源拼接 API 地址,再从 API 响应中抽取最终 `m3u8/mp4` + +### API 响应解析 + +优先支持以下模式: + +- JSON 或脚本内的 `url/urls` 字段 +- 变量赋值中的 `m3u8/mp4` 直链 +- `tweb` 之类需要二次解码的源,按参考实现做最小必要解码 + +如果最终 URL 仍是站内中间页,则允许一层站内跳转继续解析。 + +返回格式: + +- `parse = 0` +- `playUrl = ""` +- `url = 最终播放地址` +- `header = {"User-Agent": "...", "Referer": "..."}` + +## 请求与兼容性 + +统一请求头至少包含: + +- 浏览器 `User-Agent` +- 需要时补 `Referer` + +播放器 API 解析需要稳定 Referer,因此播放页与播放器 API 请求都应显式带站点 Referer。 + +本次优先采用无状态请求;如验证发现 LibVIO 对 cookie 敏感,再补最小 cookie 维护。 + +## 错误处理 + +实现遵循“失败返回空,不抛异常中断”的原则: + +- 页面请求失败时返回空列表或空播放地址 +- DOM 节点缺失时字段回退为空字符串 +- 播放配置提取失败时尝试回退到直链正则 +- 最终失败则返回 `{"parse": 0, "playUrl": "", "url": ""}` + +日志只保留必要调试信息,主要用于播放解析链路。 + +## 测试设计 + +采用测试优先方式实现,先给纯解析函数写测试,再补生产代码。 + +测试重点: + +1. 首页/分类卡片解析 + - 断言能从 `stui-vodlist__box` 提取详情 id、标题、封面和备注 +2. 首页与分类高层流程 + - 断言 `homeContent` 返回固定分类 + - 断言 `homeVideoContent` 与 `categoryContent` 组装分页结果正确 +3. 搜索解析 + - 断言能解析搜索列表并复用卡片解析 +4. 详情解析 + - 断言基础字段提取正确 + - 断言会过滤网盘线路 + - 断言剧集列表输出为紧凑播放 id +5. 播放解析 + - 断言可从播放页提取 `player_*` 配置 + - 断言可从播放器 JS 提取 API 基址 + - 断言可从 API 或脚本提取最终 `m3u8/mp4` + - 断言特殊源与空结果路径行为正确 + +优先使用 `unittest` 和 mock,避免测试依赖真实站点网络。 + +## 实施顺序 + +1. 新增 `tests/test_libvio.py`,先覆盖首页/分类/搜索/详情/播放器核心解析 +2. 新增 `libvio.py` 基本骨架与分类映射 +3. 实现首页、分类和搜索列表抓取 +4. 实现详情字段和线路提取 +5. 实现播放页配置解析与播放器 API 解析 +6. 运行测试并补必要的站内跳转与特殊源兼容 diff --git a/py/libvio.py b/py/libvio.py new file mode 100644 index 0000000..568212c --- /dev/null +++ b/py/libvio.py @@ -0,0 +1,292 @@ +# coding=utf-8 +import json +import re +import sys +from urllib.parse import quote + +from base.spider import Spider as BaseSpider + +sys.path.append("..") + + +class Spider(BaseSpider): + def __init__(self): + self.name = "LibVIO" + self.host = "https://libvio.site" + self.headers = { + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/124.0.0.0 Safari/537.36" + ) + } + self.categories = [ + {"type_name": "最近更新", "type_id": "index"}, + {"type_name": "电影", "type_id": "movie"}, + {"type_name": "电视剧", "type_id": "series"}, + {"type_name": "动漫", "type_id": "anime"}, + {"type_name": "日韩剧", "type_id": "jpandkr"}, + {"type_name": "欧美剧", "type_id": "euandus"}, + ] + self.category_paths = { + "index": "/", + "movie": "/type/1-{pg}.html", + "series": "/type/2-{pg}.html", + "anime": "/type/4-{pg}.html", + "jpandkr": "/type/15-{pg}.html", + "euandus": "/type/16-{pg}.html", + } + + def init(self, extend=""): + return None + + def getName(self): + return self.name + + def homeContent(self, filter): + return {"class": self.categories} + + def _build_url(self, href): + raw = str(href or "").strip() + if not raw: + return "" + if raw.startswith(("http://", "https://")): + return raw + if raw.startswith("//"): + return "https:" + raw + return self.host + "/" + raw.lstrip("/") + + def _extract_vod_id(self, href): + raw = str(href or "").strip() + matched = re.search(r"/detail/(\d+)\.html", raw) + if matched: + return matched.group(1) + if re.fullmatch(r"\d+", raw): + return raw + return "" + + def _extract_play_id(self, href): + raw = str(href or "").strip() + matched = re.search(r"/play/([^./]+-\d+-\d+)\.html", raw) + if matched: + return matched.group(1) + if re.fullmatch(r"[^./]+-\d+-\d+", raw): + return raw + return "" + + def _build_detail_request_url(self, vod_id): + return f"{self.host}/detail/{self._extract_vod_id(vod_id)}.html" + + def _build_play_request_url(self, play_id): + return f"{self.host}/play/{self._extract_play_id(play_id)}.html" + + def _parse_list_cards(self, html): + root = self.html(html) + results = [] + if root is None: + return results + + seen = set() + for card in root.xpath("//*[contains(@class,'stui-vodlist__box')]"): + href = ((card.xpath(".//a[@href][1]/@href") or [""])[0]).strip() + vod_id = self._extract_vod_id(href) + title = ((card.xpath(".//a[@title][1]/@title") or [""])[0]).strip() + pic = ( + (card.xpath(".//a[@data-original][1]/@data-original") or [""])[0].strip() + or (card.xpath(".//img[@data-original][1]/@data-original") or [""])[0].strip() + or (card.xpath(".//img[@src][1]/@src") or [""])[0].strip() + ) + remarks = "".join(card.xpath(".//*[contains(@class,'pic-text')][1]//text()")).strip() + if not vod_id or vod_id in seen or not title: + continue + seen.add(vod_id) + results.append( + { + "vod_id": vod_id, + "vod_name": title, + "vod_pic": self._build_url(pic), + "vod_remarks": remarks, + } + ) + return results + + def _request_html(self, path_or_url, expect_xpath=None, referer=None): + target = path_or_url if str(path_or_url).startswith("http") else self._build_url(path_or_url) + headers = dict(self.headers) + headers["Referer"] = referer or (self.host + "/") + response = self.fetch(target, headers=headers, timeout=10) + if response.status_code != 200: + return "" + return response.text or "" + + def _page_result(self, items, pg): + page = int(pg) + pagecount = page + 1 if items else page + return { + "list": items, + "page": page, + "pagecount": pagecount, + "limit": len(items), + "total": pagecount * max(len(items), 1), + } + + def _clean_text(self, text): + return re.sub(r"\s+", " ", str(text or "").replace("\xa0", " ")).strip() + + def _extract_detail_field(self, root, label, joiner=""): + if root is None: + return "" + nodes = root.xpath(f'.//span[contains(normalize-space(.), "{label}:")]') + if not nodes: + return "" + values = [] + label_node = nodes[0] + if self._clean_text(label_node.tail): + values.append(self._clean_text(label_node.tail)) + for sibling in label_node.itersiblings(): + if sibling.tag == "span" and "text-muted" in " ".join(sibling.xpath("./@class")): + break + text = self._clean_text("".join(sibling.xpath(".//text()"))) + if text: + values.append(text) + tail = self._clean_text(sibling.tail) + if tail: + values.append(tail) + cleaned = [] + for value in values: + if value and value not in cleaned: + cleaned.append(value) + return joiner.join(cleaned) if joiner else "".join(cleaned) + + def _parse_detail_page(self, html, vod_id): + root = self.html(html) + detail_root = (root.xpath("//*[contains(@class,'stui-content__detail')][1]") or [root])[0] + title = ((detail_root.xpath(".//*[contains(@class,'title')][1]//text()") or [""])[0]).strip() + pic = ( + (root.xpath("//*[contains(@class,'stui-content__thumb')]//img/@data-original") or [""])[0].strip() + or (root.xpath("//*[contains(@class,'stui-content__thumb')]//img/@src") or [""])[0].strip() + ) + + episodes = [] + seen = set() + for playlist in root.xpath("//*[contains(@class,'stui-content__playlist')]"): + heading = self._clean_text("".join(playlist.xpath("./preceding-sibling::*[1]//text()"))) + if any(keyword in heading for keyword in ("夸克", "UC", "网盘")): + continue + for anchor in playlist.xpath(".//a[@href]"): + play_id = self._extract_play_id((anchor.xpath("./@href") or [""])[0]) + name = self._clean_text("".join(anchor.xpath(".//text()"))) + if not play_id or not name or play_id in seen: + continue + seen.add(play_id) + episodes.append(f"{name}${play_id}") + + vod = { + "vod_id": vod_id, + "path": self._build_detail_request_url(vod_id), + "vod_name": title, + "vod_pic": self._build_url(pic), + "vod_tag": "", + "vod_time": "", + "vod_remarks": "", + "vod_play_from": "LibVIO", + "vod_play_url": "#".join(episodes), + "type_name": self._extract_detail_field(detail_root, "类型"), + "vod_content": self._extract_detail_field(detail_root, "简介"), + "vod_year": self._extract_detail_field(detail_root, "年份"), + "vod_area": self._extract_detail_field(detail_root, "地区"), + "vod_lang": "", + "vod_director": self._extract_detail_field(detail_root, "导演", joiner=","), + "vod_actor": self._extract_detail_field(detail_root, "主演", joiner=","), + } + return {"list": [vod]} + + def homeVideoContent(self): + html = self._request_html("/", expect_xpath="//*[contains(@class,'stui-vodlist__box')]") + return {"list": self._parse_list_cards(html)} + + def categoryContent(self, tid, pg, filter, extend): + path = self.category_paths.get(tid, self.category_paths["movie"]).format(pg=pg) + html = self._request_html(path, expect_xpath="//*[contains(@class,'stui-vodlist__box')]") + return self._page_result(self._parse_list_cards(html), pg) + + def searchContent(self, key, quick, pg="1"): + path = "/search/-------------.html?wd={0}".format(quote(key)) + html = self._request_html(path, expect_xpath="//*[contains(@class,'stui-vodlist__box')]") + return self._page_result(self._parse_list_cards(html), pg) + + def detailContent(self, ids): + vod_id = ids[0] + html = self._request_html( + self._build_detail_request_url(vod_id), + expect_xpath="//*[contains(@class,'stui-content__playlist')]", + ) + return self._parse_detail_page(html, vod_id) + + def _parse_player_config(self, html): + matched = re.search(r"player_[a-z0-9_]+\s*=\s*(\{[\s\S]*?\})\s*;?", html, re.I) + if not matched: + return None + try: + return json.loads(matched.group(1)) + except Exception: + return None + + def _extract_play_api_base(self, body): + matched = re.search(r'src\s*=\s*["\']([^"\']+)["\']', body, re.I) + if not matched: + return "" + return self._build_url(matched.group(1)) + + def _extract_playable_url(self, body): + patterns = [ + r'["\']?urls?["\']?\s*:\s*["\']([^"\']+)["\']', + r'(?:var|let|const)\s+urls?\s*=\s*["\']([^"\']+)["\']', + r'["\']url["\']\s*:\s*["\']([^"\']+\.(?:m3u8|mp4)[^"\']*)["\']', + r'url\s*=\s*["\']([^"\']+\.(?:m3u8|mp4)[^"\']*)["\']', + ] + text = str(body or "") + for pattern in patterns: + matched = re.search(pattern, text, re.I) + if matched: + return self._build_url(matched.group(1).replace("\\/", "/").replace("&", "&").strip()) + return "" + + def _request_player_js(self, source): + return self._request_html(f"/static/player/{source}.js", referer=self.host + "/") + + def playerContent(self, flag, id, vipFlags): + play_page_url = self._build_play_request_url(id) + detail_html = self._request_html(play_page_url, referer=self.host + "/") + config = self._parse_player_config(detail_html) + if not config or config.get("from") in ("kuake", "uc"): + return {"parse": 0, "playUrl": "", "url": ""} + + if config.get("from") == "ty_new1": + api_body = self._request_html(f"/vid/ty4.php?url={config.get('url', '')}", referer=self.host + "/") + final_url = self._extract_playable_url(api_body) + else: + player_js = self._request_player_js(config.get("from", "")) + api_base = self._extract_play_api_base(player_js) + api_url = "{base}{url}&next={next}&id={id}&nid={nid}".format( + base=api_base, + url=config.get("url", ""), + next=config.get("link_next", ""), + id=config.get("id", ""), + nid=config.get("nid", ""), + ) + api_body = self._request_html(api_url, referer=self.host + "/") + final_url = self._extract_playable_url(api_body) + + if not final_url: + return {"parse": 0, "playUrl": "", "url": ""} + + return { + "parse": 0, + "playUrl": "", + "url": final_url, + "header": { + "User-Agent": self.headers["User-Agent"], + "Referer": self.host + "/", + }, + } diff --git a/py/tests/test_libvio.py b/py/tests/test_libvio.py new file mode 100644 index 0000000..050ab1b --- /dev/null +++ b/py/tests/test_libvio.py @@ -0,0 +1,156 @@ +import unittest +from importlib.machinery import SourceFileLoader +from pathlib import Path +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] +MODULE = SourceFileLoader("libvio_spider", str(ROOT / "libvio.py")).load_module() +Spider = MODULE.Spider + + +class TestLibVioSpider(unittest.TestCase): + def setUp(self): + self.spider = Spider() + self.spider.init() + + def test_home_content_exposes_expected_categories(self): + content = self.spider.homeContent(False) + class_ids = [item["type_id"] for item in content["class"]] + self.assertEqual(class_ids, ["index", "movie", "series", "anime", "jpandkr", "euandus"]) + + def test_parse_list_cards_extracts_compact_vod_id(self): + html = """ +
+ + 更新至10集 +
+ """ + cards = self.spider._parse_list_cards(html) + self.assertEqual( + cards, + [ + { + "vod_id": "456", + "vod_name": "示例影片", + "vod_pic": "https://libvio.site/cover.jpg", + "vod_remarks": "更新至10集", + } + ], + ) + + @patch.object(Spider, "_request_html") + def test_home_video_content_reuses_list_card_parser(self, mock_request_html): + mock_request_html.return_value = """ +
+ + HD +
+ """ + result = self.spider.homeVideoContent() + self.assertEqual(result["list"][0]["vod_id"], "111") + self.assertEqual(result["list"][0]["vod_name"], "最近更新") + + @patch.object(Spider, "_request_html") + def test_category_content_builds_page_result(self, mock_request_html): + mock_request_html.return_value = """ +
+ + 完结 +
+ """ + result = self.spider.categoryContent("movie", "2", False, {}) + self.assertEqual(result["page"], 2) + self.assertEqual(result["list"][0]["vod_id"], "222") + + @patch.object(Spider, "_request_html") + def test_search_content_reuses_card_parser(self, mock_request_html): + mock_request_html.return_value = """ +
+ + 抢先版 +
+ """ + result = self.spider.searchContent("繁花", False, "1") + self.assertEqual(result["list"][0]["vod_id"], "333") + self.assertEqual(result["list"][0]["vod_name"], "搜索影片") + + def test_parse_detail_page_extracts_fields_and_filters_pan_sources(self): + html = """ +
+ +
+
+

示例剧

+

类型:剧情

+

地区:大陆

+

年份:2026

+

导演:张三

+

主演:李四王五

+

简介:一段剧情简介

+
+

在线播放

+ +

夸克资源

+ + """ + result = self.spider._parse_detail_page(html, "999") + vod = result["list"][0] + self.assertEqual(vod["vod_id"], "999") + self.assertEqual(vod["path"], "https://libvio.site/detail/999.html") + self.assertEqual(vod["vod_name"], "示例剧") + self.assertEqual(vod["type_name"], "剧情") + self.assertEqual(vod["vod_area"], "大陆") + self.assertEqual(vod["vod_year"], "2026") + self.assertEqual(vod["vod_director"], "张三") + self.assertEqual(vod["vod_actor"], "李四,王五") + self.assertEqual(vod["vod_content"], "一段剧情简介") + self.assertEqual(vod["vod_play_from"], "LibVIO") + self.assertEqual(vod["vod_play_url"], "第1集$999-1-1#第2集$999-1-2") + + @patch.object(Spider, "_request_html") + def test_detail_content_builds_detail_request_url_from_vod_id(self, mock_request_html): + mock_request_html.return_value = """ +

详情影片

+ + """ + result = self.spider.detailContent(["123"]) + self.assertEqual(mock_request_html.call_args.args[0], "https://libvio.site/detail/123.html") + self.assertEqual(result["list"][0]["vod_id"], "123") + + def test_extract_player_config_reads_json_assignment(self): + html = '' + data = self.spider._parse_player_config(html) + self.assertEqual(data["from"], "line") + + def test_extract_play_api_base_reads_player_js(self): + body = 'var player={}; src="/player/api.php?url=";' + self.assertEqual(self.spider._extract_play_api_base(body), "https://libvio.site/player/api.php?url=") + + @patch.object(Spider, "_request_html") + def test_player_content_resolves_direct_api_url(self, mock_request_html): + mock_request_html.side_effect = [ + '', + 'src="/player/api.php?url="', + 'var urls="https://video.example/final.m3u8";', + ] + result = self.spider.playerContent("LibVIO", "999-1-1", {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["url"], "https://video.example/final.m3u8") + self.assertEqual(result["header"]["Referer"], "https://libvio.site/") + + @patch.object(Spider, "_request_html") + def test_player_content_returns_empty_for_pan_source(self, mock_request_html): + mock_request_html.return_value = '' + self.assertEqual(self.spider.playerContent("LibVIO", "999-1-1", {}), {"parse": 0, "playUrl": "", "url": ""}) + + +if __name__ == "__main__": + unittest.main()