From 3dde995bd7026d2a8dd6452e38a002c9907de414 Mon Sep 17 00:00:00 2001
From: Harold <8866033@gmail.com>
Date: Fri, 24 Apr 2026 17:19:57 +0800
Subject: [PATCH] feat: add nsvod spider
---
py/tests/test_耐视点播.py | 250 +++++++++++++++++++++++++++++++
py/耐视点播.py | 306 ++++++++++++++++++++++++++++++++++++++
2 files changed, 556 insertions(+)
create mode 100644 py/tests/test_耐视点播.py
create mode 100644 py/耐视点播.py
diff --git a/py/tests/test_耐视点播.py b/py/tests/test_耐视点播.py
new file mode 100644
index 0000000..60d8e4d
--- /dev/null
+++ b/py/tests/test_耐视点播.py
@@ -0,0 +1,250 @@
+import unittest
+from importlib.machinery import SourceFileLoader
+from pathlib import Path
+from unittest.mock import patch
+
+
+ROOT = Path(__file__).resolve().parents[1]
+MODULE = SourceFileLoader("nsvod_spider", str(ROOT / "耐视点播.py")).load_module()
+Spider = MODULE.Spider
+
+
+class TestNSVodSpider(unittest.TestCase):
+ def setUp(self):
+ Spider._instance = None
+ self.spider = Spider()
+ self.spider.init()
+
+ def test_home_content_exposes_fixed_classes(self):
+ content = self.spider.homeContent(False)
+ self.assertEqual(
+ [(item["type_id"], item["type_name"]) for item in content["class"]],
+ [
+ ("1", "电影"),
+ ("2", "连续剧"),
+ ("3", "综艺"),
+ ("4", "动漫"),
+ ("37", "Netflix"),
+ ("40", "纪录片"),
+ ],
+ )
+
+ def test_build_url_and_extract_vod_id_handle_relative_paths(self):
+ self.assertEqual(
+ self.spider._build_url("/index.php/vod/detail/id/7.html"),
+ "https://nsvod.me/index.php/vod/detail/id/7.html",
+ )
+ self.assertEqual(
+ self.spider._build_url("//img.example/poster.jpg"),
+ "https://img.example/poster.jpg",
+ )
+ self.assertEqual(self.spider._extract_vod_id("/index.php/vod/detail/id/99.html"), "99")
+ self.assertEqual(self.spider._extract_vod_id("/bad/path"), "")
+
+ def test_parse_cards_extracts_unique_videos(self):
+ html = """
+
+ ![]()
+ 更新至01集
+
+
+ 
+ HD
+
+
+ ![]()
+
+ """
+ self.assertEqual(
+ self.spider._parse_cards(html),
+ [
+ {
+ "vod_id": "11",
+ "vod_name": "示例A",
+ "vod_pic": "https://nsvod.me/a.jpg",
+ "vod_remarks": "更新至01集",
+ },
+ {
+ "vod_id": "12",
+ "vod_name": "示例B",
+ "vod_pic": "https://nsvod.me/b.jpg",
+ "vod_remarks": "HD",
+ },
+ ],
+ )
+
+ @patch.object(Spider, "_request_html")
+ def test_home_video_content_reads_home_page(self, mock_request_html):
+ mock_request_html.return_value = """
+
+ ![]()
+ 热播
+
+ """
+ result = self.spider.homeVideoContent()
+ self.assertEqual(mock_request_html.call_args.args[0], "https://nsvod.me/")
+ self.assertEqual(
+ result,
+ {
+ "list": [
+ {
+ "vod_id": "88",
+ "vod_name": "首页片",
+ "vod_pic": "https://nsvod.me/home.jpg",
+ "vod_remarks": "热播",
+ }
+ ]
+ },
+ )
+
+ @patch.object(Spider, "_request_html")
+ def test_category_content_reads_category_page(self, mock_request_html):
+ mock_request_html.return_value = """
+
+ ![]()
+ 更新中
+
+ """
+ result = self.spider.categoryContent("2", "3", False, {})
+ self.assertEqual(
+ mock_request_html.call_args.args[0],
+ "https://nsvod.me/index.php/vod/show/id/2.html?page=3",
+ )
+ self.assertEqual(result["page"], 3)
+ self.assertEqual(result["limit"], 1)
+ self.assertEqual(result["list"][0]["vod_id"], "21")
+ self.assertNotIn("pagecount", result)
+
+ @patch.object(Spider, "_request_html")
+ def test_category_content_falls_back_to_home_sections(self, mock_request_html):
+ mock_request_html.side_effect = [
+ "",
+ """
+ 最新综艺
+
+ ![]()
+ 第1期
+
+ 最新纪录片
+ """,
+ ]
+ result = self.spider.categoryContent("3", "1", False, {})
+ self.assertEqual([item["vod_id"] for item in result["list"]], ["31"])
+
+ @patch.object(Spider, "_request_html")
+ def test_search_content_short_circuits_blank_keyword_and_parses_results(self, mock_request_html):
+ self.assertEqual(self.spider.searchContent("", False, "1"), {"page": 1, "total": 0, "list": []})
+ mock_request_html.assert_not_called()
+
+ mock_request_html.return_value = """
+
+ """
+ result = self.spider.searchContent("繁花", False, "2")
+ self.assertEqual(
+ mock_request_html.call_args.args[0],
+ "https://nsvod.me/index.php/vod/search.html?wd=%E7%B9%81%E8%8A%B1",
+ )
+ self.assertEqual(result["total"], 1)
+ self.assertEqual(
+ result["list"][0],
+ {
+ "vod_id": "45",
+ "vod_name": "搜索片",
+ "vod_pic": "https://nsvod.me/search.jpg",
+ "vod_remarks": "抢先版",
+ },
+ )
+
+ def test_parse_detail_page_extracts_metadata_and_play_groups(self):
+ html = """
+ 《耐视示例》在线观看
+
+ 年份2025
+ 地区中国香港
+ 导演 导演甲
+ 主演 演员甲 / 演员乙
+
+
+
+ """
+ vod = self.spider._parse_detail_page("70", html)
+ self.assertEqual(vod["vod_name"], "耐视示例")
+ self.assertEqual(vod["vod_pic"], "https://nsvod.me/poster-detail.jpg")
+ self.assertEqual(vod["vod_year"], "2025")
+ self.assertEqual(vod["vod_area"], "中国香港")
+ self.assertEqual(vod["vod_director"], "导演甲")
+ self.assertEqual(vod["vod_actor"], "演员甲 / 演员乙")
+ self.assertEqual(vod["vod_content"], "这是一段简介。")
+ self.assertEqual(vod["vod_play_from"], "线路A$$$线路B")
+ self.assertEqual(
+ vod["vod_play_url"],
+ "第1集$/index.php/vod/play/id/70/sid/1/nid/1.html#第2集$/index.php/vod/play/id/70/sid/1/nid/2.html$$$正片$/index.php/vod/play/id/70/sid/2/nid/1.html",
+ )
+
+ @patch.object(Spider, "_request_html")
+ def test_detail_content_reads_detail_url_and_returns_single_vod(self, mock_request_html):
+ mock_request_html.return_value = """
+ 详情标题
+
+ """
+ result = self.spider.detailContent(["80"])
+ self.assertEqual(
+ mock_request_html.call_args.args[0],
+ "https://nsvod.me/index.php/vod/detail/id/80.html",
+ )
+ self.assertEqual(result["list"][0]["vod_id"], "80")
+ self.assertEqual(result["list"][0]["vod_name"], "详情标题")
+ self.assertEqual(result["list"][0]["vod_play_from"], "线路1")
+ self.assertEqual(result["list"][0]["vod_play_url"], "正片$/index.php/vod/play/id/80/sid/1/nid/1.html")
+
+ @patch.object(Spider, "_request_html")
+ def test_player_content_returns_player_aaaa_url(self, mock_request_html):
+ mock_request_html.return_value = """
+
+ """
+ result = self.spider.playerContent("线路A", "/index.php/vod/play/id/70/sid/1/nid/1.html", {})
+ self.assertEqual(result["parse"], 0)
+ self.assertEqual(result["jx"], 0)
+ self.assertEqual(result["url"], "https://cdn.example/direct.m3u8")
+ self.assertEqual(result["header"]["Referer"], "https://nsvod.me/index.php/vod/play/id/70/sid/1/nid/1.html")
+
+ @patch.object(Spider, "_request_html")
+ def test_player_content_falls_back_to_inline_m3u8(self, mock_request_html):
+ mock_request_html.return_value = ''
+ result = self.spider.playerContent("线路A", "/index.php/vod/play/id/71/sid/1/nid/1.html", {})
+ self.assertEqual(result["url"], "https://cdn.example/fallback.m3u8?token=1")
+ self.assertEqual(result["jx"], 0)
+
+ @patch.object(Spider, "_request_html")
+ def test_player_content_returns_play_page_when_no_media_url_found(self, mock_request_html):
+ mock_request_html.return_value = "empty"
+ result = self.spider.playerContent("线路A", "/index.php/vod/play/id/72/sid/1/nid/1.html", {})
+ self.assertEqual(result["parse"], 0)
+ self.assertEqual(result["jx"], 1)
+ self.assertEqual(result["playUrl"], "")
+ self.assertEqual(result["url"], "https://nsvod.me/index.php/vod/play/id/72/sid/1/nid/1.html")
+ self.assertEqual(result["header"]["Referer"], "https://nsvod.me/")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/py/耐视点播.py b/py/耐视点播.py
new file mode 100644
index 0000000..9d3144f
--- /dev/null
+++ b/py/耐视点播.py
@@ -0,0 +1,306 @@
+# coding=utf-8
+import json
+import re
+import sys
+from urllib.parse import quote, urljoin
+
+from base.spider import Spider as BaseSpider
+
+sys.path.append("..")
+
+
+class Spider(BaseSpider):
+ def __init__(self):
+ self.name = "耐视点播"
+ self.host = "https://nsvod.me"
+ self.headers = {
+ "User-Agent": (
+ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+ "AppleWebKit/537.36 (KHTML, like Gecko) "
+ "Chrome/120.0.0.0 Safari/537.36"
+ ),
+ "Referer": self.host + "/",
+ }
+ self.classes = [
+ {"type_id": "1", "type_name": "电影"},
+ {"type_id": "2", "type_name": "连续剧"},
+ {"type_id": "3", "type_name": "综艺"},
+ {"type_id": "4", "type_name": "动漫"},
+ {"type_id": "37", "type_name": "Netflix"},
+ {"type_id": "40", "type_name": "纪录片"},
+ ]
+
+ def init(self, extend=""):
+ return None
+
+ def getName(self):
+ return self.name
+
+ def homeContent(self, filter):
+ return {"class": self.classes}
+
+ def _stringify(self, value):
+ return "" if value is None else str(value)
+
+ def _clean_text(self, text):
+ raw = self._stringify(text).replace(" ", " ")
+ raw = re.sub(r"<[^>]+>", " ", raw)
+ return re.sub(r"\s+", " ", raw).strip()
+
+ def _build_url(self, path):
+ raw = self._stringify(path).strip()
+ if not raw:
+ return ""
+ if raw.startswith(("http://", "https://")):
+ return raw
+ if raw.startswith("//"):
+ return "https:" + raw
+ return urljoin(self.host + "/", raw)
+
+ def _extract_vod_id(self, href):
+ matched = re.search(r"/index\.php/vod/detail/id/(\d+)\.html", self._stringify(href))
+ return matched.group(1) if matched else ""
+
+ def _build_detail_url(self, vod_id):
+ return self._build_url(f"/index.php/vod/detail/id/{self._stringify(vod_id).strip()}.html")
+
+ def _request_html(self, path_or_url, headers=None, referer=None):
+ target = path_or_url if self._stringify(path_or_url).startswith("http") else self._build_url(path_or_url)
+ merged_headers = dict(self.headers)
+ if headers:
+ merged_headers.update(headers)
+ merged_headers["Referer"] = referer or self.headers["Referer"]
+ response = self.fetch(target, headers=merged_headers, timeout=10)
+ if response.status_code != 200:
+ return ""
+ return response.text or ""
+
+ def _parse_cards(self, html):
+ items = []
+ seen = set()
+ for block, href, vod_id, title in re.findall(
+ r'(]*href="(/index\.php/vod/detail/id/(\d+)\.html)"[^>]*title="([^"]*)"[^>]*>[\s\S]*?)',
+ self._stringify(html),
+ re.I,
+ ):
+ if vod_id in seen or not title:
+ continue
+ seen.add(vod_id)
+ pic_match = re.search(r'data-src="([^"]*)"', block, re.I) or re.search(r'src="([^"]*)"', block, re.I)
+ remarks_match = re.search(r'public-list-prb[^>]*>([^<]*)', block, re.I)
+ if not remarks_match:
+ remarks_match = re.search(r'public-list-subtitle[^>]*>([^<]*)', block, re.I)
+ items.append(
+ {
+ "vod_id": vod_id,
+ "vod_name": title.strip(),
+ "vod_pic": self._build_url(pic_match.group(1) if pic_match else ""),
+ "vod_remarks": self._clean_text(remarks_match.group(1) if remarks_match else ""),
+ }
+ )
+ return items
+
+ def homeVideoContent(self):
+ return {"list": self._parse_cards(self._request_html(self.host + "/"))}
+
+ def _extract_section_cards(self, html, tid):
+ section_map = {
+ "1": "最新电影",
+ "2": "最新连续剧",
+ "3": "最新综艺",
+ "4": "最新动漫",
+ "37": "最新Netflix",
+ "40": "最新纪录片",
+ }
+ section_name = section_map.get(self._stringify(tid))
+ if not section_name:
+ return []
+ markers = [
+ "最新热播",
+ "最新Netflix",
+ "最新电影",
+ "最新连续剧",
+ "最新资讯",
+ "最新动漫",
+ "最新综艺",
+ "最新纪录片",
+ ]
+ body = self._stringify(html)
+ start_tag = f'title-h cor4">{section_name}'
+ start = body.find(start_tag)
+ if start < 0:
+ return []
+ end = len(body)
+ for name in markers:
+ if name == section_name:
+ continue
+ pos = body.find(f'title-h cor4">{name}', start + len(start_tag))
+ if pos > 0 and pos < end:
+ end = pos
+ return self._parse_cards(body[start:end])
+
+ def _parse_search_cards(self, html):
+ items = []
+ seen = set()
+ for block in re.findall(
+ r'(]*class="[^"]*module-search-item[^"]*"[^>]*>[\s\S]*?)',
+ self._stringify(html),
+ re.I,
+ ):
+ href_match = re.search(r'class="[^"]*video-serial[^"]*"[^>]*href="([^"]*)"', block, re.I)
+ title_match = re.search(r'class="[^"]*video-serial[^"]*"[^>]*title="([^"]*)"', block, re.I)
+ pic_match = re.search(r'data-src="([^"]*)"', block, re.I) or re.search(r'src="([^"]*)"', block, re.I)
+ remark_match = re.search(r'class="[^"]*video-serial[^"]*"[^>]*>([^<]*)', block, re.I)
+ vod_id = self._extract_vod_id(href_match.group(1) if href_match else "")
+ if not vod_id or vod_id in seen:
+ continue
+ seen.add(vod_id)
+ items.append(
+ {
+ "vod_id": vod_id,
+ "vod_name": self._clean_text(title_match.group(1) if title_match else ""),
+ "vod_pic": self._build_url(pic_match.group(1) if pic_match else ""),
+ "vod_remarks": self._clean_text(remark_match.group(1) if remark_match else ""),
+ }
+ )
+ return items
+
+ def categoryContent(self, tid, pg, filter, extend):
+ page = int(pg)
+ url = self._build_url(f"/index.php/vod/show/id/{tid}.html?page={page}")
+ items = self._parse_cards(self._request_html(url))
+ if not items:
+ items = self._extract_section_cards(self._request_html(self.host + "/"), tid)
+ return {"page": page, "limit": len(items), "total": page * 20 + len(items), "list": items}
+
+ def searchContent(self, key, quick, pg="1"):
+ page = int(pg)
+ keyword = self._clean_text(key)
+ if not keyword:
+ return {"page": page, "total": 0, "list": []}
+ url = self._build_url(f"/index.php/vod/search.html?wd={quote(keyword)}")
+ items = self._parse_search_cards(self._request_html(url))
+ return {"page": page, "total": len(items), "list": items}
+
+ def _strip_title(self, title):
+ matched = re.search(r"《([^》]+)》", self._stringify(title))
+ if matched:
+ return matched.group(1)
+ return self._clean_text(title)
+
+ def _parse_play_groups(self, html):
+ source_names = []
+ for raw in re.findall(r']*class="[^"]*swiper-slide[^"]*"[^>]*>([\s\S]*?)', self._stringify(html), re.I):
+ cleaned = re.sub(r"\d+\s*集?$", "", self._clean_text(raw)).strip()
+ if cleaned:
+ source_names.append(cleaned)
+
+ groups = []
+ for index, block in enumerate(
+ re.findall(r'(]*class="[^"]*anthology-list-box[^"]*"[^>]*>[\s\S]*?)', self._stringify(html), re.I)
+ ):
+ episodes = []
+ for href, title in re.findall(r']*href="([^"]*/index\.php/vod/play/[^"]*)"[^>]*>([\s\S]*?)', block, re.I):
+ episodes.append(f"{self._clean_text(title)}${href.replace('&', '&').strip()}")
+ if episodes:
+ groups.append((source_names[index] if index < len(source_names) else f"线路{index + 1}", "#".join(episodes)))
+
+ if not groups:
+ episodes = []
+ for href, title in re.findall(
+ r']*href="([^"]*/index\.php/vod/play/[^"]*)"[^>]*>([\s\S]*?)',
+ self._stringify(html),
+ re.I,
+ ):
+ episodes.append(f"{self._clean_text(title)}${href.replace('&', '&').strip()}")
+ if episodes:
+ groups.append(("线路1", "#".join(episodes)))
+ return groups
+
+ def _parse_detail_page(self, vod_id, html):
+ body = self._stringify(html)
+ name_match = re.search(r"([^<]+)", body, re.I)
+ name = self._strip_title(name_match.group(1) if name_match else "")
+ if not name:
+ slide_match = re.search(r'slide-info-title[^>]*>([^<]+)<', body, re.I)
+ name = self._clean_text(slide_match.group(1) if slide_match else "")
+
+ pic_match = re.search(r'detail-pic[\s\S]*?data-src="([^"]*)"', body, re.I) or re.search(
+ r'detail-pic[\s\S]*?src="([^"]*)"',
+ body,
+ re.I,
+ )
+ year_match = re.search(r"年份[\s\S]*?(\d{4})", body, re.I) or re.search(r"年份[\s\S]*?(\d{4})", body, re.I)
+ area_match = re.search(r"地区[\s\S]*?([^<]+)", body, re.I)
+ director_match = re.search(r"导演\s*([^<\n]+)", body, re.I)
+ actor_match = re.search(r"主演\s*([^<\n]+)", body, re.I)
+ content_match = re.search(r'id="height_limit"[^>]*>([\s\S]*?)', body, re.I)
+ groups = self._parse_play_groups(body)
+
+ director = self._clean_text(director_match.group(1) if director_match else "")
+ actor = self._clean_text(actor_match.group(1) if actor_match else "")
+ return {
+ "vod_id": self._stringify(vod_id),
+ "vod_name": name,
+ "vod_pic": self._build_url(pic_match.group(1) if pic_match else ""),
+ "vod_year": year_match.group(1) if year_match else "",
+ "vod_area": self._clean_text(area_match.group(1) if area_match else ""),
+ "vod_director": "" if director == "未知" else director,
+ "vod_actor": "" if actor == "未知" else actor,
+ "vod_content": self._clean_text(content_match.group(1) if content_match else "") or "暂无简介",
+ "vod_play_from": "$$$".join(item[0] for item in groups),
+ "vod_play_url": "$$$".join(item[1] for item in groups),
+ }
+
+ def detailContent(self, ids):
+ vod_id = self._stringify(ids[0] if isinstance(ids, list) and ids else ids).strip()
+ if not vod_id:
+ return {"list": []}
+ vod = self._parse_detail_page(vod_id, self._request_html(self._build_detail_url(vod_id)))
+ if not vod.get("vod_play_from"):
+ return {"list": []}
+ return {"list": [vod]}
+
+ def _extract_player_data(self, html):
+ matched = re.search(r"var\s+player_aaaa\s*=\s*(\{[\s\S]*?\})", self._stringify(html), re.I)
+ if not matched:
+ return {}
+ try:
+ return json.loads(matched.group(1))
+ except Exception:
+ return {}
+
+ def _pick_direct_media_url(self, html):
+ matched = re.search(r'(https?://[^"\'\s]+\.m3u8[^"\'\s]*)', self._stringify(html), re.I)
+ return matched.group(1) if matched else ""
+
+ def playerContent(self, flag, id, vipFlags):
+ play_url = self._build_url(id)
+ html = self._request_html(play_url, headers={"User-Agent": self.headers["User-Agent"]}, referer=self.host + "/")
+ player_data = self._extract_player_data(html)
+ if player_data.get("url"):
+ return {
+ "parse": 0,
+ "jx": 0,
+ "playUrl": "",
+ "url": self._stringify(player_data.get("url")),
+ "header": {"User-Agent": self.headers["User-Agent"], "Referer": play_url},
+ }
+
+ media_url = self._pick_direct_media_url(html)
+ if media_url:
+ return {
+ "parse": 0,
+ "jx": 0,
+ "playUrl": "",
+ "url": media_url,
+ "header": {"User-Agent": self.headers["User-Agent"], "Referer": play_url},
+ }
+
+ return {
+ "parse": 0,
+ "jx": 1,
+ "playUrl": "",
+ "url": play_url,
+ "header": dict(self.headers),
+ }
+ 更新至01集
+ 
+ 
+ 
+ 
+ 
