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

feat: add nsvod spider

Browse Source
This commit is contained in:
Harold
2026-04-24 17:19:57 +08:00
parent 67bd67435b
commit 3dde995bd7
2 changed files with 556 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/tests/test_耐视点播.py
+250
View File
@@ -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 = """
<a href="/index.php/vod/detail/id/11.html" title="示例A">
<img data-src="/a.jpg" />
<span class="public-list-prb">更新至01集</span>
</a>
<a href="/index.php/vod/detail/id/12.html" title="示例B">
<img src="/b.jpg" />
<div class="public-list-subtitle">HD</div>
</a>
<a href="/index.php/vod/detail/id/11.html" title="示例A">
<img data-src="/dup.jpg" />
</a>
"""
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 = """
<a href="/index.php/vod/detail/id/88.html" title="首页片">
<img data-src="/home.jpg" />
<span class="public-list-prb">热播</span>
</a>
"""
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 = """
<a href="/index.php/vod/detail/id/21.html" title="分类片">
<img data-src="/cate.jpg" />
<span class="public-list-prb">更新中</span>
</a>
"""
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 = [
"<html></html>",
"""
<div class="title-h cor4">最新综艺</div>
<a href="/index.php/vod/detail/id/31.html" title="综艺回退">
<img data-src="/variety.jpg" />
<span class="public-list-prb">第1期</span>
</a>
<div class="title-h cor4">最新纪录片</div>
""",
]
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 = """
<div class="module-search-item">
<a class="video-serial" href="/index.php/vod/detail/id/45.html" title="搜索片">抢先版</a>
<div class="module-item-pic"><img data-src="/search.jpg" alt="搜索片" /></div>
</div>
"""
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 = """
<title>《耐视示例》在线观看</title>
<div class="detail-pic"><img data-src="/poster-detail.jpg" /></div>
<div>年份</em><span>2025</span></div>
<div>地区</em><span>中国香港</span></div>
<div>导演</em> 导演甲 </div>
<div>主演</em> 演员甲 / 演员乙 </div>
<div id="height_limit"><p>这是一段简介。</p></div>
<div class="anthology-tab">
<div class="swiper-slide">线路A 2集</div>
<div class="swiper-slide">线路B 1集</div>
</div>
<div class="anthology-list">
<div class="anthology-list-box">
<a href="/index.php/vod/play/id/70/sid/1/nid/1.html">第1集</a>
<a href="/index.php/vod/play/id/70/sid/1/nid/2.html">第2集</a>
</div>
<div class="anthology-list-box">
<a href="/index.php/vod/play/id/70/sid/2/nid/1.html">正片</a>
</div>
</div>
"""
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 = """
<div class="slide-info-title">详情标题</div>
<div class="anthology-list">
<div class="anthology-list-box">
<a href="/index.php/vod/play/id/80/sid/1/nid/1.html">正片</a>
</div>
</div>
"""
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 = """
<script>
var player_aaaa={"url":"https://cdn.example/direct.m3u8","encrypt":"0"}
</script>
"""
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 = '<script>var now="https://cdn.example/fallback.m3u8?token=1"</script>'
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 = "<html><body>empty</body></html>"
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()
py/耐视点播.py
+306
View File
@@ -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("&nbsp;", " ")
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'(<a[^>]*href="(/index\.php/vod/detail/id/(\d+)\.html)"[^>]*title="([^"]*)"[^>]*>[\s\S]*?</a>)',
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[^>]*>([^<]*)</span>', block, re.I)
if not remarks_match:
remarks_match = re.search(r'public-list-subtitle[^>]*>([^<]*)</div>', 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'(<div[^>]*class="[^"]*module-search-item[^"]*"[^>]*>[\s\S]*?</div>)',
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[^"]*"[^>]*>([^<]*)</a>', 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'<div[^>]*class="[^"]*swiper-slide[^"]*"[^>]*>([\s\S]*?)</div>', 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'(<div[^>]*class="[^"]*anthology-list-box[^"]*"[^>]*>[\s\S]*?</div>)', self._stringify(html), re.I)
):
episodes = []
for href, title in re.findall(r'<a[^>]*href="([^"]*/index\.php/vod/play/[^"]*)"[^>]*>([\s\S]*?)</a>', block, re.I):
episodes.append(f"{self._clean_text(title)}${href.replace('&amp;', '&').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'<a[^>]*href="([^"]*/index\.php/vod/play/[^"]*)"[^>]*>([\s\S]*?)</a>',
self._stringify(html),
re.I,
):
episodes.append(f"{self._clean_text(title)}${href.replace('&amp;', '&').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"<title>([^<]+)</title>", 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"年份</em>[\s\S]*?<span>(\d{4})</span>", body, re.I) or re.search(r"年份[\s\S]*?(\d{4})", body, re.I)
area_match = re.search(r"地区</em>[\s\S]*?<span>([^<]+)</span>", body, re.I)
director_match = re.search(r"导演</em>\s*([^<\n]+)", body, re.I)
actor_match = re.search(r"主演</em>\s*([^<\n]+)", body, re.I)
content_match = re.search(r'id="height_limit"[^>]*>([\s\S]*?)</div>', 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),
}