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

修复

Browse Source
This commit is contained in:
Harold
2026-04-19 21:53:43 +08:00
parent b975ba51d6
commit bcc7d4b7ff
4 changed files with 51 additions and 993 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/4KVM.py
-719
View File
@@ -1,719 +0,0 @@
# coding=utf-8
import json
import re
import subprocess
import sys
from pathlib import Path
from urllib.parse import parse_qsl, quote, urljoin, urlsplit
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 = "4KVM"
self.host = "https://www.4kvm.org"
self.page_limit = 30
self.headers = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/134.0.0.0 Safari/537.36"
),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Referer": self.host + "/",
"Cache-Control": "no-cache",
}
self.play_extensions = (".m3u8", ".mp4", ".flv", ".avi", ".mkv", ".ts")
def init(self, extend=""):
return None
def getName(self):
return self.name
def _build_url(self, path):
return urljoin(self.host + "/", str(path or "").strip())
def _normalize_url(self, value):
raw = str(value or "").strip()
if not raw:
return ""
if raw.startswith("//"):
return "https:" + raw
if raw.startswith(("http://", "https://")):
return raw
return self._build_url(raw.lstrip("/"))
def _encode_site_path(self, value):
raw = str(value or "").strip()
if not raw:
return ""
if raw.startswith("//"):
raw = "https:" + raw
if raw.startswith(("http://", "https://")):
parsed = urlsplit(raw)
path = parsed.path.lstrip("/")
query = parsed.query
else:
parsed = urlsplit(raw)
path = parsed.path.lstrip("/")
query = parsed.query
if not path and not query:
return ""
return path + ("?" + query if query else "")
def _decode_site_path(self, value):
raw = str(value or "").strip()
if not raw:
return ""
if raw.startswith(("http://", "https://")):
return raw
return self._build_url(raw.lstrip("/"))
def _request_html(self, url):
response = self.fetch(url, headers=self.headers, timeout=10, verify=False)
if response.status_code != 200:
return ""
return response.text or ""
def _parse_html(self, html_text):
text = str(html_text or "").strip()
if not text:
return None
return lxml_html.fromstring(text)
def _clean_text(self, value):
return re.sub(r"\s+", " ", str(value or "").replace("\xa0", " ")).strip()
def _first_non_empty(self, values):
for value in values:
clean = self._clean_text(value)
if clean:
return clean
return ""
def _node_text(self, node):
if node is None:
return ""
return self._clean_text("".join(node.xpath(".//text()")))
def _first_xpath_text(self, node, expressions):
for expr in expressions:
values = node.xpath(expr)
if not values:
continue
if isinstance(values[0], str):
text = self._first_non_empty(values)
else:
text = self._first_non_empty([self._node_text(item) for item in values])
if text:
return text
return ""
def _first_xpath_attr(self, node, expressions):
for expr in expressions:
values = node.xpath(expr)
if values:
value = self._clean_text(values[0])
if value:
return value
return ""
def _is_allowed_class_link(self, href):
path = "/" + self._encode_site_path(href).split("?", 1)[0].lstrip("/")
return bool(re.match(r"^/(movie|tv|anime|filter)(/.*)?$", path))
def _default_classes(self):
return [
{"type_id": "movie", "type_name": "电影"},
{"type_id": "tv", "type_name": "电视剧"},
{"type_id": "anime", "type_name": "动漫"},
{"type_id": "filter", "type_name": "筛选"},
]
def _extract_video_basic(self, node):
href = self._first_xpath_attr(
node,
[
".//a[contains(@href,'/play/')]/@href",
".//a[1]/@href",
".//h3//a[1]/@href",
],
)
vod_id = self._encode_site_path(href)
if not vod_id:
return None
title = self._first_xpath_text(
node,
[
".//h3[1]//text()",
".//*[contains(@class,'data')]//h3[1]//text()",
".//img[1]/@alt",
".//a[1]/@title",
],
) or "未知标题"
pic = self._encode_site_path(
self._first_xpath_attr(node, [".//img[1]/@data-src", ".//img[1]/@src"])
)
remarks = self._first_xpath_text(
node,
[
".//*[contains(@class,'absolute') and contains(@class,'bottom-0')]//span[last()]//text()",
".//*[contains(@class,'text-xs') and contains(@class,'text-gray-400')][last()]//text()",
".//*[contains(@class,'rating') or contains(@class,'imdb') or contains(@class,'vote')]//text()",
".//*[contains(@class,'type') or contains(@class,'genre') or contains(@class,'tag')]//text()",
".//span[last()]//text()",
],
)
return {
"vod_id": vod_id,
"vod_name": title,
"vod_pic": self._decode_site_path(pic) if pic else "",
"vod_remarks": remarks,
}
def _card_nodes(self, root):
if root is None:
return []
return root.xpath(
"//*[contains(@class,'movie-card')]"
"|//article"
"|//*[contains(@class,'items')]//article"
"|//*[contains(@class,'content')]//article"
)
def _parse_video_list(self, html_text):
root = self._parse_html(html_text)
seen = set()
videos = []
for node in self._card_nodes(root):
item = self._extract_video_basic(node)
if not item or item["vod_id"] in seen:
continue
seen.add(item["vod_id"])
videos.append(item)
return videos
def _parse_search_video_list(self, html_text):
root = self._parse_html(html_text)
if root is None:
return []
videos = []
seen = set()
for node in root.xpath("//a[starts-with(@href,'/play/')]"):
vod_id = self._encode_site_path(node.get("href"))
if not vod_id or vod_id in seen:
continue
title = self._first_non_empty(
[
self._first_xpath_text(node, [".//h3[1]//text()"]),
node.xpath(".//img[1]/@alt")[0] if node.xpath(".//img[1]/@alt") else "",
node.get("title", ""),
]
)
if not title:
continue
pic = self._first_xpath_attr(node, [".//img[1]/@data-src", ".//img[1]/@src"])
remarks = self._first_xpath_text(
node,
[
".//*[contains(@class,'absolute') and contains(@class,'top-2') and contains(@class,'right-2')]//text()",
".//*[contains(@class,'absolute') and contains(@class,'bottom-0')]//p//text()",
],
)
seen.add(vod_id)
videos.append(
{
"vod_id": vod_id,
"vod_name": title,
"vod_pic": self._decode_site_path(self._encode_site_path(pic)) if pic else "",
"vod_remarks": remarks,
}
)
return videos
def _page_result(self, items, pg):
page = int(pg)
return {
"list": items,
"page": page,
"limit": self.page_limit,
"total": (max(page, 1) - 1) * self.page_limit + len(items),
}
def homeContent(self, filter):
html_text = self._request_html(self.host)
root = self._parse_html(html_text)
classes = []
seen = set()
if root is not None:
for node in root.xpath("//nav//a[contains(@class,'nav-item')]"):
name = self._node_text(node)
href = node.get("href", "")
type_id = self._encode_site_path(href)
if (
not name
or not type_id
or name in ["首页", "影片下载", "片单"]
or not self._is_allowed_class_link(href)
or type_id in seen
):
continue
seen.add(type_id)
classes.append({"type_id": type_id, "type_name": name})
return {"class": classes or self._default_classes()}
def homeVideoContent(self):
html_text = self._request_html(self.host)
return {"list": self._parse_video_list(html_text)}
def _build_category_url(self, tid, pg):
encoded = self._encode_site_path(tid)
if not encoded:
return self.host
url = self._decode_site_path(encoded)
page = int(pg)
if page <= 1:
return url
return url + ("&" if "?" in url else "?") + f"page={page}"
def categoryContent(self, tid, pg, filter, extend):
html_text = self._request_html(self._build_category_url(tid, pg))
return self._page_result(self._parse_video_list(html_text), pg)
def filterSearchResults(self, results, search_key):
if not results or not search_key:
return results
query = self._clean_text(search_key).lower()
words = [item for item in query.split(" ") if item]
scored = []
for item in results:
title = self._clean_text(item.get("vod_name", "")).lower()
score = 0
if query == title:
score = 100
elif title.startswith(query):
score = 80
elif query in title:
score = 70
elif words and all(word in title for word in words):
score = 60
else:
matches = len([word for word in words if word in title])
if matches == 0:
continue
score = 30 + matches * 10
if "剧" in query and ("tvshows" in item.get("vod_id", "") or "/tv/" in item.get("vod_id", "")):
score += 5
if "电影" in query and ("movies" in item.get("vod_id", "") or "/movie/" in item.get("vod_id", "")):
score += 5
scored.append({"score": score, "result": item})
scored.sort(key=lambda item: item["score"], reverse=True)
min_score = 30 if len(words) > 1 else 40
filtered = [item["result"] for item in scored if item["score"] >= min_score]
if len(filtered) < 3 and len(scored) > 3:
return [item["result"] for item in scored[:10]]
return filtered
def searchContent(self, key, quick, pg="1"):
page = int(pg)
url = f"{self.host}/search?q={quote(str(key or ''))}"
if page > 1:
url += f"&page={page}"
html_text = self._request_html(url)
items = self.filterSearchResults(self._parse_search_video_list(html_text), key)
return self._page_result(items, pg)
def _extract_detail_map(self, root):
detail_map = {}
if root is None:
return detail_map
for node in root.xpath("//*[contains(@class,'grid')]//*[contains(@class,'text-gray-500')]"):
key = self._node_text(node)
value_node = node.getnext()
value = self._node_text(value_node)
if key and value:
detail_map[key] = value
return detail_map
def _extract_play_options(self, root, detail_id):
if root is None:
return []
play_links = []
nodes = root.xpath(
"//*[@id='playeroptions']//li"
"|//*[contains(concat(' ', normalize-space(@class), ' '), ' dooplay_player_option ')]"
)
for node in nodes:
title = self._first_xpath_text(node, [".//*[contains(@class,'title')]//text()", ".//span[contains(@class,'title')]//text()"]) or "播放"
server = self._first_xpath_text(node, [".//*[contains(@class,'server')]//text()", ".//span[contains(@class,'server')]//text()"])
if server:
title = f"{title}-{server}"
data_post = self._clean_text(node.get("data-post", ""))
data_nume = self._clean_text(node.get("data-nume", ""))
data_type = self._clean_text(node.get("data-type", "")) or "movie"
if not data_post or not data_nume:
continue
play_links.append(
{
"name": title,
"url": f"{detail_id}?post={data_post}&nume={data_nume}&type={data_type}",
}
)
return play_links
def _extract_episode_sources(self, root, detail_id):
if root is None:
return []
source_map = {}
order_keys = []
for node in root.xpath("//*[contains(concat(' ', normalize-space(@class), ' '), ' episode-link ')][@data-line][@data-episode]"):
href = self._encode_site_path(node.get("href", ""))
if not href:
continue
line = self._clean_text(node.get("data-line", "")) or "1"
source_name = f"线路{line}"
raw_name = self._first_xpath_text(node, [".//span[last()]//text()"]) or self._clean_text(node.get("data-episode", "")) or "播放"
episode_number = self._clean_text(node.get("data-episode", ""))
if raw_name.isdigit():
episode_name = f"第{raw_name}集"
elif episode_number.isdigit():
episode_name = f"第{episode_number}集"
else:
episode_name = raw_name
if source_name not in source_map:
source_map[source_name] = []
order_keys.append(source_name)
try:
order = int(episode_number)
except Exception:
order = len(source_map[source_name]) + 1
source_map[source_name].append({"name": episode_name, "url": href, "order": order})
sources = []
for name in order_keys:
episodes = source_map[name]
episodes.sort(key=lambda item: item["order"])
sources.append({"name": name, "episodes": [{"name": item["name"], "url": item["url"]} for item in episodes]})
if sources:
return sources
play_links = self._extract_play_options(root, detail_id)
if play_links:
return [{"name": "4KVM", "episodes": play_links}]
return []
def _serialize_play_sources(self, sources):
valid = [item for item in sources if item.get("episodes")]
return {
"vod_play_from": "$$$".join([item.get("name", "") for item in valid]) or "4KVM",
"vod_play_url": "$$$".join(
["#".join([f"{ep.get('name', '')}${ep.get('url', '')}" for ep in item.get("episodes", [])]) for item in valid]
),
}
def detailContent(self, ids):
detail_id = self._encode_site_path(ids[0] if ids else "")
html_text = self._request_html(self._decode_site_path(detail_id))
root = self._parse_html(html_text)
detail_map = self._extract_detail_map(root)
meta_desc = self._first_xpath_attr(root, ["//meta[@name='description']/@content"]) if root is not None else ""
meta_pic = self._first_xpath_attr(root, ["//meta[@property='og:image']/@content"]) if root is not None else ""
meta_title = self._first_xpath_attr(root, ["//meta[@property='og:title']/@content"]) if root is not None else ""
meta_keywords = self._first_xpath_attr(root, ["//meta[@name='keywords']/@content"]) if root is not None else ""
vod = {
"vod_id": detail_id,
"vod_name": self._first_xpath_text(root, ["//h1[contains(@class,'text-xl')][1]//text()", "//h1[1]//text()"])
or re.sub(r"\s*-\s*第\d+集.*$", "", meta_title)
or "未知标题",
"vod_pic": self._decode_site_path(
self._encode_site_path(
self._first_xpath_attr(root, ["//*[contains(@class,'video-player')][1]/@data-poster", "//*[contains(@class,'video-player')]//img[1]/@src"])
or meta_pic
)
),
"vod_content": self._first_xpath_text(
root,
["//*[contains(@class,'bg-dark-800')]//p[contains(@class,'text-xs') and contains(@class,'text-gray-300')][1]//text()"],
)
or self._clean_text(meta_desc),
"vod_year": detail_map.get("年份", ""),
"vod_area": detail_map.get("地区", "") or detail_map.get("国家/地区", ""),
"vod_remarks": detail_map.get("状态", ""),
"vod_actor": detail_map.get("主演", ""),
"vod_director": detail_map.get("导演", ""),
}
genres = []
if detail_map.get("类型"):
genres.extend([self._clean_text(item) for item in detail_map["类型"].split("/") if self._clean_text(item)])
elif meta_keywords:
keywords = [self._clean_text(item) for item in meta_keywords.split(",") if self._clean_text(item)]
genres.extend([item for item in keywords if item not in [vod["vod_name"], vod["vod_director"]]][:1])
if genres:
vod["type_name"] = ", ".join(genres)
sources = self._extract_episode_sources(root, detail_id)
if not sources:
sources = [{"name": "4KVM", "episodes": [{"name": "播放", "url": detail_id}]}]
play_data = self._serialize_play_sources(sources)
vod["vod_play_from"] = play_data["vod_play_from"]
vod["vod_play_url"] = play_data["vod_play_url"]
return {"list": [vod]}
def _is_direct_play_url(self, url):
lower_url = str(url or "").lower()
return any(ext in lower_url for ext in self.play_extensions)
def _parse_json(self, text):
try:
return json.loads(text or "{}")
except Exception:
return {}
def _extract_nbmovie_context(self, play_page_html, play_slug):
html_text = str(play_page_html or "")
root = self._parse_html(html_text)
play_path = "/" + str(play_slug or "").lstrip("/")
if not play_path.startswith("/play/"):
play_path = "/play/" + play_path.lstrip("/")
dataid = ""
if root is not None:
for node in root.xpath("//a[@href and @dataid]"):
href = self._clean_text(node.get("href", ""))
if href == play_path:
dataid = self._clean_text(node.get("dataid", ""))
break
userlink = ""
userlink_match = re.search(r"userlink:'([^']+)'", html_text)
if userlink_match:
userlink = self._clean_text(userlink_match.group(1))
nbst = ""
if root is not None:
nbst = self._first_xpath_attr(root, ["//meta[@id='nb-st']/@content"])
wasm_js = ""
wasm_bg = ""
if root is not None:
wasm_js = self._first_xpath_attr(root, ["//link[@id='wasm-cfg']/@data-js"])
wasm_bg = self._first_xpath_attr(root, ["//link[@id='wasm-cfg']/@data-bg"])
if not (dataid and userlink and nbst and wasm_js and wasm_bg):
return {}
return {
"dataid": dataid,
"userlink": userlink,
"nbst": nbst,
"wasm_js": self._normalize_url(wasm_js),
"wasm_bg": self._normalize_url(wasm_bg),
}
def _cache_nbmovie_runtime(self, wasm_js_url, wasm_bg_url):
cache_dir = Path("/tmp/4kvm_nbmovie")
cache_dir.mkdir(parents=True, exist_ok=True)
js_name = re.sub(r"[^a-zA-Z0-9._-]", "_", urlsplit(wasm_js_url).path.rsplit("/", 1)[-1] or "nbmovie_wasm.js")
if not js_name.endswith(".mjs"):
js_name = re.sub(r"\.js$", "", js_name) + ".mjs"
wasm_name = re.sub(r"[^a-zA-Z0-9._-]", "_", urlsplit(wasm_bg_url).path.rsplit("/", 1)[-1] or "nbmovie_wasm_bg.wasm")
js_path = cache_dir / js_name
wasm_path = cache_dir / wasm_name
if not js_path.exists():
response = self.fetch(wasm_js_url, headers=self.headers, timeout=15, verify=False)
if response.status_code != 200:
return None, None
js_path.write_text(response.text or "", encoding="utf-8")
if not wasm_path.exists():
response = self.fetch(wasm_bg_url, headers=self.headers, timeout=15, verify=False)
if response.status_code != 200:
return None, None
content = getattr(response, "content", None)
if content is None:
return None, None
wasm_path.write_bytes(content)
return js_path, wasm_path
def _build_nbmovie_api_url(self, play_page_html, play_slug, dataid, quality="1080"):
context = self._extract_nbmovie_context(play_page_html, play_slug)
if not context:
return ""
js_path, wasm_path = self._cache_nbmovie_runtime(context["wasm_js"], context["wasm_bg"])
if not (js_path and wasm_path):
return ""
script = """
import { readFileSync } from "node:fs";
globalThis.HTMLMetaElement = class HTMLMetaElement { constructor(content){ this.content = content; } };
globalThis.Window = class Window {};
const meta = {
"nb-st": new HTMLMetaElement(process.argv[1]),
"nb-plt": new HTMLMetaElement(String(Date.now()))
};
const w = new Window();
w.document = { getElementById(id){ return meta[id] || null; } };
globalThis.window = w;
const mod = await import(process.argv[2]);
mod.initSync({ module: readFileSync(process.argv[3]) });
process.stdout.write(%s + mod.build_play_url(process.argv[4], process.argv[5], process.argv[6], process.argv[7]));
""" % json.dumps(self.host)
try:
result = subprocess.run(
[
"node",
"--input-type=module",
"-e",
script,
context["nbst"],
js_path.resolve().as_uri(),
str(wasm_path),
str(dataid or context["dataid"]),
str(play_slug),
str(quality),
context["userlink"],
],
capture_output=True,
text=True,
timeout=10,
check=False,
)
except Exception:
return ""
if result.returncode != 0:
return ""
return self._clean_text(result.stdout)
def _pick_nbmovie_play_url(self, api_payload):
data = api_payload.get("data", {}) if isinstance(api_payload, dict) else {}
quality_urls = data.get("quality_urls", []) if isinstance(data, dict) else []
if not isinstance(quality_urls, list) or not quality_urls:
return ""
candidates = []
current_quality = data.get("current_quality")
try:
current_index = int(current_quality)
except Exception:
current_index = -1
if 0 <= current_index < len(quality_urls):
candidates.append(quality_urls[current_index])
candidates.extend(quality_urls)
for item in candidates:
if not isinstance(item, dict):
continue
url = self._clean_text(item.get("url", ""))
if not url or url == "1":
continue
return self._normalize_url(url)
return ""
def playerContent(self, flag, id, vipFlags):
encoded = self._encode_site_path(id)
parsed = urlsplit(self._decode_site_path(encoded))
base_id = parsed.path.lstrip("/")
page_url = self._decode_site_path(base_id)
params = dict(parse_qsl(parsed.query, keep_blank_values=True))
data_post = self._clean_text(params.get("post", ""))
data_nume = self._clean_text(params.get("nume", ""))
data_type = self._clean_text(params.get("type", "")) or "movie"
if data_post and data_nume:
api_url = f"{self.host}/wp-json/dooplayer/v1/post/{data_post}"
response = self.fetch(
api_url,
params={"type": data_type, "source": data_nume},
headers=self.headers,
timeout=10,
verify=False,
)
if response.status_code == 200:
embed_url = self._clean_text(self._parse_json(response.text).get("embed_url", ""))
if embed_url:
return {
"parse": 0 if self._is_direct_play_url(embed_url) else 1,
"url": self._normalize_url(embed_url),
"header": self.headers,
}
html_text = self._request_html(page_url)
play_slug = base_id or encoded.split("?", 1)[0]
nbmovie_api_url = self._build_nbmovie_api_url(
play_page_html=html_text,
play_slug=play_slug,
dataid=self._extract_nbmovie_context(html_text, play_slug).get("dataid", ""),
quality="1080",
)
if nbmovie_api_url:
api_headers = dict(self.headers)
api_headers["Referer"] = page_url
api_headers["Accept"] = "application/json,text/plain,*/*"
response = self.fetch(nbmovie_api_url, headers=api_headers, timeout=10, verify=False)
if response.status_code == 200:
play_url = self._pick_nbmovie_play_url(self._parse_json(response.text))
if play_url:
return {
"parse": 0 if self._is_direct_play_url(play_url) else 1,
"url": play_url,
"header": api_headers,
}
root = self._parse_html(html_text)
iframe = self._first_xpath_attr(
root,
[
"(//iframe[contains(@class,'metaframe')]/@src)",
"(//*[contains(@class,'dooplay_player')]//iframe/@src)",
"(//*[contains(@class,'player')]//iframe/@src)",
"(//iframe/@src)",
],
)
if iframe:
iframe_url = self._normalize_url(iframe)
return {
"parse": 0 if self._is_direct_play_url(iframe_url) else 1,
"url": iframe_url,
"header": self.headers,
}
video_url = self._first_xpath_attr(root, ["(//video/source/@src)", "(//video/@src)"])
if video_url:
full_video_url = self._normalize_url(video_url)
return {"parse": 0, "url": full_video_url, "header": self.headers}
return {"parse": 1, "url": page_url or self._decode_site_path(encoded), "header": self.headers}
py/tests/test_4KVM.py
-269
View File
@@ -1,269 +0,0 @@
import unittest
from importlib.machinery import SourceFileLoader
from pathlib import Path
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
MODULE = SourceFileLoader("k4vm_spider", str(ROOT / "4KVM.py")).load_module()
Spider = MODULE.Spider
class FakeResponse:
def __init__(self, text="", status_code=200, headers=None):
self.text = text
self.status_code = status_code
self.headers = headers or {}
self.encoding = "utf-8"
class Test4KVMSpider(unittest.TestCase):
def setUp(self):
Spider._instance = None
self.spider = Spider()
self.spider.init()
def test_encode_site_path_keeps_short_ids(self):
self.assertEqual(self.spider._encode_site_path("/play/test-slug"), "play/test-slug")
self.assertEqual(self.spider._encode_site_path("https://www.4kvm.org/tv?page=2"), "tv?page=2")
self.assertEqual(self.spider._decode_site_path("play/test-slug"), "https://www.4kvm.org/play/test-slug")
@patch.object(Spider, "fetch")
def test_home_content_and_home_video_content_parse_nav_and_cards(self, mock_fetch):
mock_fetch.return_value = FakeResponse(
"""
<html>
<nav>
<a class="nav-item" href="/">首页</a>
<a class="nav-item" href="/movie">电影</a>
<a class="nav-item" href="/tv">电视剧</a>
<a class="nav-item" href="/anime">动漫</a>
<a class="nav-item" href="/download">影片下载</a>
</nav>
<article class="movie-card">
<a href="/play/home-film">
<img src="/poster.jpg" alt="首页推荐" />
</a>
<h3>首页推荐</h3>
<div class="absolute bottom-0"><span>HD中字</span></div>
</article>
</html>
"""
)
home = self.spider.homeContent(False)
videos = self.spider.homeVideoContent()
self.assertEqual(
home["class"],
[
{"type_id": "movie", "type_name": "电影"},
{"type_id": "tv", "type_name": "电视剧"},
{"type_id": "anime", "type_name": "动漫"},
],
)
self.assertEqual(
videos["list"],
[
{
"vod_id": "play/home-film",
"vod_name": "首页推荐",
"vod_pic": "https://www.4kvm.org/poster.jpg",
"vod_remarks": "HD中字",
}
],
)
@patch.object(Spider, "fetch")
def test_category_content_builds_page_url_and_returns_items_without_pagecount(self, mock_fetch):
mock_fetch.return_value = FakeResponse(
"""
<html>
<article>
<a href="/play/cate-film"><img src="/cate.jpg" alt="分类片" /></a>
<h3>分类片</h3>
<span class="text-xs text-gray-400">更新至10集</span>
</article>
</html>
"""
)
result = self.spider.categoryContent("tv", "2", False, {})
self.assertEqual(result["page"], 2)
self.assertNotIn("pagecount", result)
self.assertEqual(result["list"][0]["vod_id"], "play/cate-film")
self.assertEqual(result["list"][0]["vod_remarks"], "更新至10集")
self.assertEqual(mock_fetch.call_args.args[0], "https://www.4kvm.org/tv?page=2")
@patch.object(Spider, "fetch")
def test_search_content_filters_and_ranks_results(self, mock_fetch):
mock_fetch.return_value = FakeResponse(
"""
<html>
<a href="/play/exact-hit">
<img src="/exact.jpg" alt="繁花" />
<h3>繁花</h3>
<div class="absolute top-2 right-2">完结</div>
</a>
<a href="/play/partial-hit">
<img src="/partial.jpg" alt="繁花前传" />
<h3>繁花前传</h3>
</a>
<a href="/play/other">
<img src="/other.jpg" alt="风起云涌" />
<h3>风起云涌</h3>
</a>
</html>
"""
)
result = self.spider.searchContent("繁花", False, "3")
self.assertEqual(result["page"], 3)
self.assertNotIn("pagecount", result)
self.assertEqual([item["vod_id"] for item in result["list"]], ["play/exact-hit", "play/partial-hit"])
self.assertIn("/search?q=%E7%B9%81%E8%8A%B1&page=3", mock_fetch.call_args.args[0])
@patch.object(Spider, "fetch")
def test_detail_content_maps_meta_and_episode_sources(self, mock_fetch):
mock_fetch.return_value = FakeResponse(
"""
<html>
<head>
<meta name="description" content="站点简介" />
<meta property="og:image" content="/meta.jpg" />
<meta name="keywords" content="详情片,剧情,导演甲" />
</head>
<body>
<h1 class="text-xl">详情片</h1>
<div class="video-player" data-poster="/poster.jpg"></div>
<div class="grid">
<div class="text-gray-500">年份</div><div>2025</div>
<div class="text-gray-500">地区</div><div>大陆</div>
<div class="text-gray-500">状态</div><div>更新至2集</div>
<div class="text-gray-500">主演</div><div>主演甲/主演乙</div>
<div class="text-gray-500">导演</div><div>导演甲</div>
<div class="text-gray-500">类型</div><div>剧情/动作</div>
</div>
<div class="bg-dark-800">
<p class="text-xs text-gray-300 leading-relaxed">正文简介</p>
</div>
<a class="episode-link" data-line="2" data-episode="1" href="/play/detail-slug?ep=1"><span>1</span></a>
<a class="episode-link" data-line="1" data-episode="2" href="/play/detail-slug?ep=2"><span>2</span></a>
<a class="episode-link" data-line="1" data-episode="1" href="/play/detail-slug?ep=1"><span>1</span></a>
</body>
</html>
"""
)
result = self.spider.detailContent(["play/detail-slug"])
vod = result["list"][0]
self.assertEqual(vod["vod_id"], "play/detail-slug")
self.assertEqual(vod["vod_name"], "详情片")
self.assertEqual(vod["vod_pic"], "https://www.4kvm.org/poster.jpg")
self.assertEqual(vod["vod_content"], "正文简介")
self.assertEqual(vod["vod_year"], "2025")
self.assertEqual(vod["vod_area"], "大陆")
self.assertEqual(vod["vod_remarks"], "更新至2集")
self.assertEqual(vod["vod_actor"], "主演甲/主演乙")
self.assertEqual(vod["vod_director"], "导演甲")
self.assertEqual(vod["type_name"], "剧情, 动作")
self.assertEqual(vod["vod_play_from"], "线路2$$$线路1")
self.assertEqual(
vod["vod_play_url"],
"第1集$play/detail-slug?ep=1$$$第1集$play/detail-slug?ep=1#第2集$play/detail-slug?ep=2",
)
@patch.object(Spider, "fetch")
def test_detail_content_falls_back_to_dooplay_options(self, mock_fetch):
mock_fetch.return_value = FakeResponse(
"""
<html>
<body>
<h1>选集片</h1>
<li class="dooplay_player_option" data-post="7788" data-nume="2" data-type="movie">
<span class="title">正片</span>
<span class="server">海外</span>
</li>
</body>
</html>
"""
)
result = self.spider.detailContent(["play/option-slug"])
vod = result["list"][0]
self.assertEqual(vod["vod_play_from"], "4KVM")
self.assertEqual(vod["vod_play_url"], "正片-海外$play/option-slug?post=7788&nume=2&type=movie")
@patch.object(Spider, "fetch")
def test_player_content_prefers_api_embed_url(self, mock_fetch):
def fake_fetch(url, params=None, headers=None, timeout=5, verify=True, stream=False, allow_redirects=True):
self.assertEqual(url, "https://www.4kvm.org/wp-json/dooplayer/v1/post/7788")
self.assertEqual(params, {"type": "movie", "source": "2"})
return FakeResponse('{"embed_url":"https://cdn.example.com/master.m3u8"}')
mock_fetch.side_effect = fake_fetch
result = self.spider.playerContent("", "play/option-slug?post=7788&nume=2&type=movie", [])
self.assertEqual(result["parse"], 0)
self.assertEqual(result["url"], "https://cdn.example.com/master.m3u8")
@patch.object(Spider, "fetch")
def test_player_content_falls_back_to_iframe_then_video(self, mock_fetch):
responses = [
FakeResponse("{}", status_code=500),
FakeResponse('<html><iframe class="metaframe" src="//player.example.com/embed/42"></iframe></html>'),
FakeResponse('<html><video><source src="/stream.m3u8" /></video></html>'),
]
mock_fetch.side_effect = responses
iframe_result = self.spider.playerContent("", "play/fallback?post=11&nume=1&type=tv", [])
video_result = self.spider.playerContent("", "play/video-only", [])
self.assertEqual(iframe_result["parse"], 1)
self.assertEqual(iframe_result["url"], "https://player.example.com/embed/42")
self.assertEqual(video_result["parse"], 0)
self.assertEqual(video_result["url"], "https://www.4kvm.org/stream.m3u8")
@patch.object(Spider, "fetch")
def test_player_content_uses_nbmovie_api_for_current_site_structure(self, mock_fetch):
page_html = """
<html>
<head>
<meta id="nb-st" content="1776604294700" />
</head>
<body>
<nav x-data="{isLoggedIn: false, userlink:'X1VaWUBdUgYJBg4FCgc7IUlfXUlc'}"></nav>
<link id="wasm-cfg" data-js="/static/wasm/nbmovie_wasm.426511b7.js" data-bg="/static/wasm/nbmovie_wasm_bg.d5d51939.wasm" />
<a href="/play/ch43qikut" dataid="33260" data-line="1" data-episode="1">1</a>
</body>
</html>
"""
api_json = """
{
"code": 200,
"data": {
"play_id": "33260",
"current_quality": 1,
"quality_urls": [
{"title": "流畅", "url": "1"},
{"title": "1080P", "url": "https://cdn.example.com/stream/master.m3u8"}
]
}
}
"""
self.spider._build_nbmovie_api_url = lambda **kwargs: "https://www.4kvm.org/video/play?signed=1"
mock_fetch.side_effect = [FakeResponse(page_html), FakeResponse(api_json)]
result = self.spider.playerContent("", "play/ch43qikut", [])
self.assertEqual(result["parse"], 0)
self.assertEqual(result["url"], "https://cdn.example.com/stream/master.m3u8")
if __name__ == "__main__":
unittest.main()
py/tests/test_剧圈圈.py
+33 -3
View File
@@ -2,7 +2,7 @@ import unittest
from importlib.machinery import SourceFileLoader
from pathlib import Path
from requests.exceptions import ConnectionError
from unittest.mock import patch
from unittest.mock import call, patch
ROOT = Path(__file__).resolve().parents[1]
@@ -74,18 +74,48 @@ class TestJuQuanQuanSpider(unittest.TestCase):
self.assertEqual(result["list"][0]["vod_id"], "vod/1")
@patch.object(Spider, "_request_html")
def test_category_content_builds_page_result(self, mock_request_html):
def test_category_content_prefers_vodshow_page_result(self, mock_request_html):
mock_request_html.return_value = """
<a class="module-poster-item module-item" href="/vod/456.html">
<div class="module-poster-item-title">分类影片</div>
</a>
"""
result = self.spider.categoryContent("juji", "2", False, {})
self.assertEqual(mock_request_html.call_args.args[0], "https://www.jqqzx.cc/type/juji/page/2.html")
self.assertEqual(mock_request_html.call_args.args[0], "https://www.jqqzx.cc/vodshow/id/juji/page/2.html")
self.assertEqual(result["page"], 2)
self.assertNotIn("pagecount", result)
self.assertEqual(result["list"][0]["vod_id"], "vod/456")
@patch.object(Spider, "_request_html")
def test_category_content_falls_back_to_type_and_stops_retrying_vodshow(self, mock_request_html):
mock_request_html.side_effect = [
"<title>系统安全验证</title>",
"""
<a class="module-poster-item module-item" href="/vod/456.html">
<div class="module-poster-item-title">分类影片</div>
</a>
""",
"""
<a class="module-poster-item module-item" href="/vod/789.html">
<div class="module-poster-item-title">后续分类影片</div>
</a>
""",
]
first_result = self.spider.categoryContent("juji", "2", False, {})
second_result = self.spider.categoryContent("juji", "3", False, {})
self.assertEqual(
mock_request_html.call_args_list,
[
call("https://www.jqqzx.cc/vodshow/id/juji/page/2.html"),
call("https://www.jqqzx.cc/type/juji/page/2.html"),
call("https://www.jqqzx.cc/type/juji/page/3.html"),
],
)
self.assertEqual(first_result["list"][0]["vod_id"], "vod/456")
self.assertEqual(second_result["list"][0]["vod_id"], "vod/789")
@patch.object(Spider, "_request_html")
def test_search_content_uses_suggest_api(self, mock_request_html):
mock_request_html.return_value = '{"list":[{"id":"777","name":"搜索结果","pic":"/pic.jpg"}]}'
py/剧圈圈.py
+18 -2
View File
@@ -16,10 +16,12 @@ class Spider(BaseSpider):
def __init__(self):
self.name = "剧圈圈"
self.host = "https://www.jqqzx.cc"
self._category_vodshow_unavailable = False
self.headers = {
"User-Agent": "Mozilla/5.0",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Referer": self.host + "/",
"Cookie": "gg_iscookie=1; gg_show_number6122=4; gg_iscookie=1; gg_show_number6122=3; mx_style=white; showBtn=true; mac_history_mxpro=%5B%7B%22vod_name%22%3A%22%E5%81%8F%E5%81%8F%E9%81%87%E8%A7%81%E4%BD%A0%22%2C%22vod_url%22%3A%22https%3A%2F%2Fwww.jqqzx.cc%2Fplay%2F62215-5-1.html%22%2C%22vod_part%22%3A%2201%22%7D%2C%7B%22vod_name%22%3A%22%E6%9C%88%E9%B3%9E%E7%BB%AE%E7%BA%AA%22%2C%22vod_url%22%3A%22https%3A%2F%2Fwww.jqqzx.cc%2Fplay%2F61941-2-1.html%22%2C%22vod_part%22%3A%22%E7%AC%AC1%E9%9B%86%22%7D%5D; PHPSESSID=gtc81g6q1dnisr5f6gldv4aivq",
}
self.categories = [
{"type_id": "dianying", "type_name": "电影"},
@@ -128,12 +130,26 @@ class Spider(BaseSpider):
def homeVideoContent(self):
return {"list": self._parse_cards(self._request_html(self.host))[:40]}
def _build_vodshow_category_url(self, tid, page):
if page <= 1:
return self._build_url(f"/vodshow/id/{tid}.html")
return self._build_url(f"/vodshow/id/{tid}/page/{page}.html")
def _build_type_category_url(self, tid, page):
return self._build_url(f"/type/{tid}/page/{page}.html")
def categoryContent(self, tid, pg, filter, extend):
page = int(pg)
items = self._parse_cards(self._request_html(self._build_url(f"/type/{tid}/page/{page}.html")))
items = []
if not self._category_vodshow_unavailable:
items = self._parse_cards(self._request_html(self._build_vodshow_category_url(tid, page)))
if not items:
self._category_vodshow_unavailable = True
if not items:
items = self._parse_cards(self._request_html(self._build_type_category_url(tid, page)))
return {
"page": page,
"total": page * len(items) + (1 if items else 0),
"total": page * 30 + len(items),
"list": items,
}