From 94897aeb660b9ca452a499c46a07e40fb9ee2845 Mon Sep 17 00:00:00 2001
From: Harold <8866033@gmail.com>
Date: Fri, 24 Apr 2026 17:53:13 +0800
Subject: [PATCH] =?UTF-8?q?=E5=88=A0=E9=99=A4?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
py/tests/test_电影人生.py | 163 ----------------
py/电影人生.py | 383 --------------------------------------
2 files changed, 546 deletions(-)
delete mode 100644 py/tests/test_电影人生.py
delete mode 100644 py/电影人生.py
diff --git a/py/tests/test_电影人生.py b/py/tests/test_电影人生.py
deleted file mode 100644
index d5bd070..0000000
--- a/py/tests/test_电影人生.py
+++ /dev/null
@@ -1,163 +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("dyrs_spider", str(ROOT / "电影人生.py")).load_module()
-Spider = MODULE.Spider
-
-SAMPLE_LIST_HTML = """
-
- ![]()
- 更新至10集
-
-2025剧情
-
- 
-
-"""
-
-DETAIL_HTML = """
-
-
-电影人生示例
-年份2025
-地区大陆
-语言国语
-导演
-主演
-标签
-剧情简介
-
-"""
-
-LINE1_HTML = """
-
-"""
-
-LINE2_HTML = """
-
-"""
-
-SEARCH_HTML = """
-
-
-
-"""
-
-
-class TestDYRSSpider(unittest.TestCase):
- def setUp(self):
- Spider._instance = None
- self.spider = Spider()
- self.spider.init()
-
- def test_home_content_exposes_classes_and_filters(self):
- content = self.spider.homeContent(False)
- self.assertEqual(
- [item["type_id"] for item in content["class"]],
- ["dianying", "dianshiju", "dongman", "zongyi"],
- )
- self.assertEqual(
- [item["key"] for item in content["filters"]["dianying"]],
- ["class", "sort_field"],
- )
-
- def test_parse_cards_extracts_short_id_title_pic_and_remarks(self):
- cards = self.spider._parse_vod_cards(SAMPLE_LIST_HTML)
- self.assertEqual(cards[0]["vod_id"], "dyrscom-foo.html")
- self.assertEqual(cards[0]["vod_name"], "片名A")
- self.assertEqual(cards[0]["vod_pic"], "https://dyrsok.com/cover/a.jpg")
- self.assertEqual(cards[0]["vod_year"], "2025")
- self.assertEqual(cards[0]["type_name"], "剧情")
- self.assertEqual(cards[0]["vod_remarks"], "更新至10集")
-
- @patch.object(Spider, "_request_html")
- def test_category_content_builds_query_string(self, mock_html):
- mock_html.return_value = SAMPLE_LIST_HTML
- result = self.spider.categoryContent(
- "dianying",
- "2",
- False,
- {"class": "动作", "sort_field": "update_time"},
- )
- self.assertEqual(result["page"], 2)
- self.assertEqual(result["list"][0]["vod_id"], "dyrscom-foo.html")
- mock_html.assert_called_with(
- "https://dyrsok.com/dianying.html?class=%E5%8A%A8%E4%BD%9C&sort_field=update_time&page=2"
- )
-
- @patch.object(Spider, "_request_html")
- def test_home_video_content_reads_home_page(self, mock_html):
- mock_html.return_value = SAMPLE_LIST_HTML
- result = self.spider.homeVideoContent()
- self.assertEqual(len(result["list"]), 2)
- mock_html.assert_called_with("https://dyrsok.com/")
-
- @patch.object(Spider, "_request_html")
- def test_detail_content_extracts_meta_and_lines(self, mock_html):
- mock_html.side_effect = [DETAIL_HTML, LINE1_HTML, LINE2_HTML]
- result = self.spider.detailContent(["dyrscom-foo.html"])
- vod = result["list"][0]
- self.assertEqual(vod["vod_id"], "dyrscom-foo.html")
- self.assertEqual(vod["vod_name"], "电影人生示例")
- self.assertEqual(vod["vod_pic"], "https://dyrsok.com/poster.jpg")
- self.assertEqual(vod["vod_year"], "2025")
- self.assertEqual(vod["vod_area"], "大陆")
- self.assertEqual(vod["vod_lang"], "国语")
- self.assertEqual(vod["vod_director"], "张三")
- self.assertEqual(vod["vod_actor"], "李四,王五")
- self.assertEqual(vod["type_name"], "剧情 / 悬疑")
- self.assertEqual(vod["vod_play_from"], "线路一$$$线路二")
- self.assertIn("第1集$", vod["vod_play_url"])
- self.assertIn("正片$", vod["vod_play_url"])
-
- @patch.object(Spider, "_request_html")
- def test_search_content_refines_exact_match_first(self, mock_html):
- mock_html.return_value = SEARCH_HTML
- result = self.spider.searchContent("繁花", False, "1")
- self.assertEqual([item["vod_name"] for item in result["list"][:2]], ["繁花", "繁花2023"])
- mock_html.assert_called_with("https://dyrsok.com/s.html?name=%E7%B9%81%E8%8A%B1")
-
- @patch.object(Spider, "_request_html")
- @patch.object(Spider, "_request_response")
- def test_player_content_resolves_api_m3u8_and_raw_playlist(self, mock_response, mock_html):
- mock_html.side_effect = [
- '',
- "#EXTM3U\n/api/m3u8?id=xyz&raw=1\n",
- ]
-
- class FakeResponse:
- status_code = 302
- headers = {"Location": "https://media.example/master.m3u8"}
- text = ""
-
- mock_response.return_value = FakeResponse()
- play_id = (
- '{"title":"第1集","origin":"线路一","page":"https://dyrsok.com/dyrscom-foo.html?play=1",'
- '"vodName":"电影人生示例","pic":"https://dyrsok.com/poster.jpg"}'
- )
- result = self.spider.playerContent("线路一", play_id, {})
- self.assertEqual(result["parse"], 0)
- self.assertEqual(result["url"], "https://media.example/api/m3u8?id=xyz&raw=1")
-
- @patch.object(Spider, "_request_html")
- def test_player_content_falls_back_to_parse_when_no_api_link(self, mock_html):
- mock_html.return_value = "empty"
- result = self.spider.playerContent("线路一", '{"page":"https://dyrsok.com/dyrscom-foo.html?play=1"}', {})
- self.assertEqual(result["parse"], 1)
- self.assertEqual(result["url"], "https://dyrsok.com/dyrscom-foo.html?play=1")
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/py/电影人生.py b/py/电影人生.py
deleted file mode 100644
index e1d201f..0000000
--- a/py/电影人生.py
+++ /dev/null
@@ -1,383 +0,0 @@
-# coding=utf-8
-import json
-import re
-import sys
-from urllib.parse import quote, urlencode, urljoin
-
-from base.spider import Spider as BaseSpider
-
-sys.path.append("..")
-
-
-class Spider(BaseSpider):
- def __init__(self):
- self.name = "电影人生"
- self.host = "https://dyrsok.com"
- self.headers = {
- "User-Agent": (
- "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
- "AppleWebKit/537.36 (KHTML, like Gecko) "
- "Chrome/146.0.0.0 Safari/537.36"
- ),
- "Referer": self.host + "/",
- }
- self.classes = [
- {"type_id": "dianying", "type_name": "电影"},
- {"type_id": "dianshiju", "type_name": "电视剧"},
- {"type_id": "dongman", "type_name": "动漫"},
- {"type_id": "zongyi", "type_name": "综艺"},
- ]
- common_movie_tv = [
- {
- "key": "class",
- "name": "类型",
- "init": "",
- "value": [
- {"n": "全部", "v": ""},
- {"n": "剧情", "v": "剧情"},
- {"n": "喜剧", "v": "喜剧"},
- {"n": "动作", "v": "动作"},
- ],
- },
- {
- "key": "sort_field",
- "name": "排序",
- "init": "",
- "value": [
- {"n": "默认", "v": ""},
- {"n": "热度", "v": "play_hot"},
- {"n": "更新时间", "v": "update_time"},
- ],
- },
- ]
- common_dm = [
- {
- "key": "class",
- "name": "类型",
- "init": "",
- "value": [
- {"n": "全部", "v": ""},
- {"n": "冒险", "v": "冒险"},
- {"n": "热血", "v": "热血"},
- {"n": "搞笑", "v": "搞笑"},
- ],
- },
- {
- "key": "sort_field",
- "name": "排序",
- "init": "",
- "value": [
- {"n": "默认", "v": ""},
- {"n": "热度", "v": "play_hot"},
- {"n": "更新时间", "v": "update_time"},
- ],
- },
- ]
- common_zy = [
- {
- "key": "class",
- "name": "类型",
- "init": "",
- "value": [
- {"n": "全部", "v": ""},
- {"n": "真人秀", "v": "真人秀"},
- {"n": "脱口秀", "v": "脱口秀"},
- {"n": "音乐", "v": "音乐"},
- ],
- },
- {
- "key": "sort_field",
- "name": "排序",
- "init": "",
- "value": [
- {"n": "默认", "v": ""},
- {"n": "热度", "v": "play_hot"},
- {"n": "更新时间", "v": "update_time"},
- ],
- },
- ]
- self.filters = {
- "dianying": common_movie_tv,
- "dianshiju": common_movie_tv,
- "dongman": common_dm,
- "zongyi": common_zy,
- }
-
- def init(self, extend=""):
- return None
-
- def getName(self):
- return self.name
-
- def homeContent(self, filter):
- return {"class": self.classes, "filters": self.filters}
-
- def homeVideoContent(self):
- return {"list": self._parse_vod_cards(self._request_html(self.host + "/"))[:24]}
-
- def _abs_url(self, value):
- raw = str(value or "").strip()
- if not raw:
- return ""
- return urljoin(self.host + "/", raw)
-
- def _clean_text(self, text):
- return re.sub(r"\s+", " ", str(text or "").replace("\xa0", " ")).strip()
-
- def _request_html(self, url, headers=None):
- request_headers = dict(self.headers)
- if headers:
- request_headers.update(headers)
- response = self.fetch(url, headers=request_headers, timeout=10, verify=False)
- return response.text if response.status_code == 200 else ""
-
- def _request_response(self, url, headers=None, allow_redirects=False):
- request_headers = dict(self.headers)
- if headers:
- request_headers.update(headers)
- return self.fetch(
- url,
- headers=request_headers,
- timeout=10,
- verify=False,
- allow_redirects=allow_redirects,
- )
-
- def _parse_vod_cards(self, html):
- root = self.html(html)
- if root is None:
- return []
- cards = []
- seen = set()
- for node in root.xpath("//a[@data-url and @title]"):
- href = ((node.xpath("./@href") or [""])[0]).strip()
- title = ((node.xpath("./@title") or [""])[0]).strip()
- if not href or not title or "/dyrscom-" not in href:
- continue
- vod_id = href.lstrip("/")
- if vod_id in seen:
- continue
- seen.add(vod_id)
- parent = node.getparent()
- year = ""
- type_name = ""
- if parent is not None:
- year = self._clean_text(
- "".join(parent.xpath(".//*[contains(@class,'items-center')][1]/span[1]//text()"))
- )
- type_name = self._clean_text(
- "".join(parent.xpath(".//*[contains(@class,'items-center')][1]/span[last()]//text()"))
- )
- remarks = self._clean_text("".join(node.xpath(".//*[contains(@class,'text-[10px]')][1]//text()")))
- pic = ((node.xpath(".//img[1]/@data-src") or node.xpath(".//img[1]/@src") or [""])[0]).strip()
- cards.append(
- {
- "vod_id": vod_id,
- "vod_name": title,
- "vod_pic": self._abs_url(pic),
- "vod_url": self._abs_url(href),
- "vod_year": year,
- "type_name": type_name,
- "vod_remarks": remarks,
- }
- )
- return cards
-
- def categoryContent(self, tid, pg, filter, extend):
- page = int(pg)
- values = extend if isinstance(extend, dict) else {}
- query = {}
- if values.get("class"):
- query["class"] = values["class"]
- if values.get("sort_field"):
- query["sort_field"] = values["sort_field"]
- if page > 1:
- query["page"] = page
- url = self.host + f"/{tid}.html"
- if query:
- url += "?" + urlencode(query)
- return {
- "page": page,
- "total": 0,
- "list": self._parse_vod_cards(self._request_html(url)),
- }
-
- def _pick_meta_value(self, html, label):
- patterns = [
- rf"{label}\s*\s*]*>([^<]+)",
- rf">{label}\s*<[^>]*>\s*]*>([^<]+)",
- rf"]*>{label}\s*]*>([^<]+)",
- ]
- for pattern in patterns:
- matched = re.search(pattern, html)
- if matched:
- return self._clean_text(matched.group(1))
- root = self.html(html)
- if root is None:
- return ""
- nodes = root.xpath(f"//span[normalize-space(text())='{label}']/following-sibling::span[1]")
- if nodes:
- return self._clean_text("".join(nodes[0].xpath(".//text()")))
- return ""
-
- def _extract_link_texts(self, html, title):
- root = self.html(html)
- if root is None:
- return []
- values = []
- for node in root.xpath(f"//h3[contains(normalize-space(.), '{title}')]/following-sibling::*[1]//a//span"):
- text = self._clean_text("".join(node.xpath(".//text()"))).replace("#", "")
- if text:
- values.append(text)
- return values
-
- def detailContent(self, ids):
- vod_id = str(ids[0] if isinstance(ids, list) and ids else ids).strip()
- if not vod_id:
- return {"list": []}
- url = self._abs_url(vod_id)
- html = self._request_html(url)
- root = self.html(html)
- if root is None:
- return {"list": []}
-
- vod_name = self._clean_text("".join(root.xpath("(//h1|//h2)[1]//text()")))
- vod_pic = self._abs_url(((root.xpath("//meta[@property='og:image']/@content") or [""])[0]).strip())
- vod_content = self._clean_text(
- "".join(
- root.xpath(
- "//h3[contains(normalize-space(.), '剧情简介')]/following-sibling::*[1]"
- "//*[contains(@class,'text-sm')][1]//text()"
- )
- )
- ) or self._clean_text(((root.xpath("//meta[@name='description']/@content") or [""])[0]))
- vod_year = self._pick_meta_value(html, "年份")
- vod_area = self._pick_meta_value(html, "地区")
- vod_lang = self._pick_meta_value(html, "语言")
- vod_director = ",".join(self._extract_link_texts(html, "导演"))
- vod_actor = ",".join(self._extract_link_texts(html, "主演"))
- tags = self._extract_link_texts(html, "标签")
-
- from_list = []
- url_list = []
- for line_node in root.xpath("//*[@id='originTabs']//a[@href]"):
- href = ((line_node.xpath("./@href") or [""])[0]).strip()
- name = self._clean_text(
- ((line_node.xpath(".//button[@data-origin]/@data-origin") or [""])[0])
- or "".join(line_node.xpath(".//text()"))
- )
- line_url = self._abs_url(href)
- line_html = html if line_url == url else self._request_html(line_url)
- line_root = self.html(line_html)
- episodes = []
- if line_root is not None:
- for episode in line_root.xpath("//*[contains(@class,'seqlist')]//a[@href]"):
- ep_href = ((episode.xpath("./@href") or [""])[0]).strip()
- ep_title = self._clean_text(
- ((episode.xpath("./@data-title") or [""])[0]) or "".join(episode.xpath(".//text()"))
- ) or "播放"
- payload = json.dumps(
- {
- "title": ep_title,
- "origin": name,
- "page": self._abs_url(ep_href),
- "vodName": vod_name,
- "pic": vod_pic,
- },
- ensure_ascii=False,
- )
- episodes.append(f"{ep_title}${payload}")
- if episodes:
- from_list.append(name)
- url_list.append("#".join(episodes))
-
- return {
- "list": [
- {
- "vod_id": vod_id,
- "vod_name": vod_name,
- "vod_pic": vod_pic,
- "vod_content": vod_content,
- "vod_year": vod_year,
- "vod_area": vod_area,
- "vod_lang": vod_lang,
- "vod_director": vod_director,
- "vod_actor": vod_actor,
- "type_name": " / ".join(tags),
- "vod_play_from": "$$$".join(from_list),
- "vod_play_url": "$$$".join(url_list),
- }
- ]
- }
-
- def _normalize_keyword(self, value):
- return re.sub(r"[\s\-_—–·•:：,，.。!?！？'\"“”‘’()（）\[\]【】{}]", "", str(value or "")).lower()
-
- def _search_score(self, vod_name, keyword):
- name = self._normalize_keyword(vod_name)
- key = self._normalize_keyword(keyword)
- if not name or not key:
- return 0
- if name == key:
- return 1000
- if name.startswith(key):
- return 800
- if key in name:
- return 600
- return 0
-
- def _refine_search_results(self, items, keyword):
- scored = [(self._search_score(item.get("vod_name"), keyword), item) for item in items]
- matched = [
- item
- for score, item in sorted(scored, key=lambda pair: (-pair[0], pair[1].get("vod_name", "")))
- if score > 0
- ]
- return matched or 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.host + "/s.html?name=" + quote(keyword)
- if page > 1:
- url += "&page=" + str(page - 1)
- items = self._parse_vod_cards(self._request_html(url))
- return {"page": page, "total": len(items), "list": self._refine_search_results(items, keyword)}
-
- def playerContent(self, flag, id, vipFlags):
- raw = str(id or "").strip()
- if not raw:
- return {"parse": 1, "playUrl": "", "url": "", "header": {}, "jx": 1}
- try:
- meta = json.loads(raw)
- except Exception:
- meta = {"page": raw}
- page_url = self._abs_url(meta.get("page", ""))
- headers = {
- "User-Agent": self.headers["User-Agent"],
- "Referer": page_url or self.host + "/",
- "Origin": self.host,
- }
- html = self._request_html(page_url, headers={"Referer": self.host + "/"}) if page_url else ""
- matched = re.search(r"/api/m3u8\?origin=([^\"'\\\s&]+|[^\"'\\\s]+?)(&|&)url=([a-zA-Z0-9]+)", html)
- if not matched:
- return {"parse": 1, "playUrl": "", "url": page_url, "header": headers, "jx": 1}
-
- api_url = self.host + "/api/m3u8?origin=" + matched.group(1) + "&url=" + matched.group(3)
- probe = self._request_response(api_url, headers=headers, allow_redirects=False)
- final_url = api_url
- location = ""
- if hasattr(probe, "headers"):
- location = probe.headers.get("Location") or probe.headers.get("location") or ""
- if location:
- final_url = self._abs_url(location)
- playlist = self._request_html(final_url, headers={"Referer": self.host + "/"})
- raw_line = next(
- (line.strip() for line in playlist.splitlines() if "/api/m3u8?id=" in line and "raw=1" in line),
- "",
- )
- if raw_line:
- final_url = urljoin(final_url, raw_line)
- return {"parse": 0, "playUrl": "", "url": final_url, "header": headers, "jx": 0}
-