Merge branch 'hongguo-duanju'

py/tests/test_红果短剧.py

@@ -0,0 +1,127 @@
+import unittest
+from importlib.machinery import SourceFileLoader
+from pathlib import Path
+from unittest.mock import patch
+
+
+ROOT = Path(__file__).resolve().parents[1]
+MODULE = SourceFileLoader("hongguo_spider", str(ROOT / "红果短剧.py")).load_module()
+Spider = MODULE.Spider
+
+
+class TestHongGuoDuanJuSpider(unittest.TestCase):
+    def setUp(self):
+        Spider._instance = None
+        self.spider = Spider()
+        self.spider.init()
+
+    def test_home_content_exposes_categories(self):
+        content = self.spider.homeContent(False)
+        self.assertEqual(content["class"][0]["type_id"], "热播")
+        self.assertIn("分类", content["filters"])
+
+    @patch.object(Spider, "_get_json")
+    def test_home_video_content_uses_hot_category(self, mock_json):
+        mock_json.return_value = {
+            "data": [
+                {"id": 1, "title": "【热播】逆袭人生", "cover": "http://img/1.jpg", "totalChapterNum": 88}
+            ]
+        }
+        result = self.spider.homeVideoContent()
+        self.assertEqual(result["list"][0]["vod_id"], "1")
+        self.assertEqual(result["list"][0]["vod_name"], "逆袭人生")
+        self.assertEqual(result["list"][0]["vod_remarks"], "更新至88集")
+        self.assertIn("name=%E7%83%AD%E6%92%AD", mock_json.call_args.args[0])
+
+    @patch.object(Spider, "_get_json")
+    def test_category_content_builds_query_and_maps_items(self, mock_json):
+        mock_json.return_value = {
+            "data": [
+                {"id": "2", "title": "新剧回归", "cover": "http://img/2.jpg", "totalChapterNum": 36}
+            ]
+        }
+        result = self.spider.categoryContent("新剧", "3", False, {})
+        self.assertEqual(result["page"], 3)
+        self.assertEqual(result["list"][0]["vod_id"], "2")
+        self.assertNotIn("pagecount", result)
+        self.assertIn("name=%E6%96%B0%E5%89%A7&page=3", mock_json.call_args.args[0])
+
+    @patch.object(Spider, "_get_json")
+    def test_search_content_uses_keyword_query(self, mock_json):
+        mock_json.return_value = {
+            "data": [
+                {"id": 3, "title": "霸总归来", "cover": "", "totalChapterNum": 12}
+            ]
+        }
+        result = self.spider.searchContent("霸总", False, "1")
+        self.assertEqual(result["list"][0]["vod_name"], "霸总归来")
+        self.assertIn("name=%E9%9C%B8%E6%80%BB&page=1", mock_json.call_args.args[0])
+
+    @patch.object(Spider, "_get_json")
+    def test_search_content_returns_empty_for_blank_keyword(self, mock_json):
+        result = self.spider.searchContent("", False, "1")
+        self.assertEqual(result["list"], [])
+        mock_json.assert_not_called()
+
+    def test_clean_title_removes_hot_and_new_tags(self):
+        self.assertEqual(self.spider._clean_title("【热播】新剧名"), "新剧名")
+        self.assertEqual(self.spider._clean_title("[新剧热播] 第二部"), "第二部")
+
+    @patch.object(Spider, "_get_json")
+    def test_detail_content_builds_single_playlist(self, mock_json):
+        mock_json.return_value = {
+            "data": [
+                {"title": "【热播】逆袭人生 1", "cover": "http://img/d.jpg", "video_id": "v1"},
+                {"title": "【热播】逆袭人生 2", "cover": "http://img/d.jpg", "video_id": "v2"},
+            ]
+        }
+        result = self.spider.detailContent(["100"])
+        vod = result["list"][0]
+        self.assertEqual(vod["vod_id"], "100")
+        self.assertEqual(vod["vod_name"], "逆袭人生")
+        self.assertEqual(vod["vod_pic"], "http://img/d.jpg")
+        self.assertEqual(vod["vod_remarks"], "更新至2集")
+        self.assertEqual(vod["vod_play_from"], "红果短剧")
+        self.assertIn("逆袭人生 1$v1", vod["vod_play_url"])
+        self.assertIn("逆袭人生 2$v2", vod["vod_play_url"])
+
+    @patch.object(Spider, "_get_json")
+    def test_detail_content_returns_empty_when_id_missing(self, mock_json):
+        result = self.spider.detailContent([""])
+        self.assertEqual(result, {"list": []})
+        mock_json.assert_not_called()
+
+    def test_sort_qualities_prefers_1080p_then_sc_then_sd(self):
+        ordered = self.spider._sort_qualities(
+            [
+                {"quality": "sd", "download_url": "sd.mp4"},
+                {"quality": "1080p", "download_url": "1080.mp4"},
+                {"quality": "sc", "download_url": "sc.mp4"},
+            ]
+        )
+        self.assertEqual([item["quality"] for item in ordered], ["1080p", "sc", "sd"])
+
+    @patch.object(Spider, "_get_json")
+    def test_player_content_returns_best_quality_direct_url(self, mock_json):
+        mock_json.return_value = {
+            "data": {
+                "qualities": [
+                    {"quality": "sd", "title": "标清", "download_url": "http://video/sd.mp4"},
+                    {"quality": "1080p", "title": "超清", "download_url": "http://video/1080.mp4"},
+                ]
+            }
+        }
+        result = self.spider.playerContent("红果短剧", "v1", {})
+        self.assertEqual(result["parse"], 0)
+        self.assertEqual(result["url"], "http://video/1080.mp4")
+
+    @patch.object(Spider, "_get_json")
+    def test_player_content_returns_empty_url_on_error(self, mock_json):
+        mock_json.side_effect = Exception("boom")
+        result = self.spider.playerContent("红果短剧", "v2", {})
+        self.assertEqual(result["parse"], 0)
+        self.assertEqual(result["url"], "")
+
+
+if __name__ == "__main__":
+    unittest.main()

py/红果短剧.py

@@ -0,0 +1,206 @@
+# 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_LIST = [
+    {"type_id": "热播", "type_name": "🎬热播"},
+    {"type_id": "新剧", "type_name": "🎬新剧"},
+    {"type_id": "都市", "type_name": "🎬都市"},
+    {"type_id": "穿越", "type_name": "🎬穿越"},
+    {"type_id": "重生", "type_name": "🎬重生"},
+    {"type_id": "赘婿", "type_name": "🎬赘婿"},
+    {"type_id": "逆袭", "type_name": "🎬逆袭"},
+    {"type_id": "霸总", "type_name": "🎬霸总"},
+    {"type_id": "职场", "type_name": "🎬职场"},
+    {"type_id": "异能", "type_name": "🎬异能"},
+    {"type_id": "神医", "type_name": "🎬神医"},
+    {"type_id": "系统", "type_name": "🎬系统"},
+    {"type_id": "总裁", "type_name": "🎬总裁"},
+    {"type_id": "豪门", "type_name": "🎬豪门"},
+    {"type_id": "神豪", "type_name": "🎬神豪"},
+    {"type_id": "校园", "type_name": "🎬校园"},
+    {"type_id": "青春", "type_name": "🎬青春"},
+    {"type_id": "马甲", "type_name": "🎬马甲"},
+    {"type_id": "年代", "type_name": "🎬年代"},
+    {"type_id": "闪婚", "type_name": "🎬闪婚"},
+    {"type_id": "战神", "type_name": "🎬战神"},
+    {"type_id": "女主", "type_name": "🎬女主"},
+    {"type_id": "修仙", "type_name": "🎬修仙"},
+    {"type_id": "亲情", "type_name": "🎬亲情"},
+    {"type_id": "虐恋", "type_name": "🎬虐恋"},
+    {"type_id": "追妻", "type_name": "🎬追妻"},
+    {"type_id": "萌宝", "type_name": "🎬萌宝"},
+    {"type_id": "古风", "type_name": "🎬古风"},
+    {"type_id": "传承", "type_name": "🎬传承"},
+    {"type_id": "甜宠", "type_name": "🎬甜宠"},
+    {"type_id": "奇幻", "type_name": "🎬奇幻"},
+    {"type_id": "爱情", "type_name": "🎬爱情"},
+    {"type_id": "乡村", "type_name": "🎬乡村"},
+    {"type_id": "历史", "type_name": "🎬历史"},
+    {"type_id": "王妃", "type_name": "🎬王妃"},
+    {"type_id": "高手", "type_name": "🎬高手"},
+    {"type_id": "娱乐", "type_name": "🎬娱乐"},
+    {"type_id": "联合", "type_name": "🎬联合"},
+    {"type_id": "破镜", "type_name": "🎬破镜"},
+    {"type_id": "暗恋", "type_name": "🎬暗恋"},
+    {"type_id": "民国", "type_name": "🎬民国"},
+    {"type_id": "冤家", "type_name": "🎬冤家"},
+    {"type_id": "真假", "type_name": "🎬真假"},
+    {"type_id": "龙王", "type_name": "🎬龙王"},
+    {"type_id": "穿书", "type_name": "🎬穿书"},
+    {"type_id": "女帝", "type_name": "🎬女帝"},
+    {"type_id": "团宠", "type_name": "🎬团宠"},
+    {"type_id": "玄幻", "type_name": "🎬玄幻"},
+    {"type_id": "仙侠", "type_name": "🎬仙侠"},
+    {"type_id": "青梅", "type_name": "🎬青梅"},
+    {"type_id": "悬疑", "type_name": "🎬悬疑"},
+    {"type_id": "推理", "type_name": "🎬推理"},
+    {"type_id": "皇后", "type_name": "🎬皇后"},
+    {"type_id": "替身", "type_name": "🎬替身"},
+    {"type_id": "大叔", "type_name": "🎬大叔"},
+    {"type_id": "喜剧", "type_name": "🎬喜剧"},
+    {"type_id": "剧情", "type_name": "🎬剧情"},
+]
+
+
+class Spider(BaseSpider):
+    def __init__(self):
+        self.name = "红果短剧"
+        self.host = "https://api-v2.cenguigui.cn"
+        self.api = self.host + "/api/duanju.html"
+        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"
+            ),
+            "Accept": "application/json, text/plain, */*",
+            "Accept-Language": "zh-CN,zh;q=0.9",
+        }
+        self.classes = CLASS_LIST
+        self.filters = {
+            "分类": [
+                {
+                    "key": "area",
+                    "name": "分类",
+                    "value": CLASS_LIST,
+                }
+            ]
+        }
+
+    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._list_items("热播", 1)}
+
+    def categoryContent(self, tid, pg, filter, extend):
+        page = int(pg)
+        items = self._list_items(tid, page)
+        return {"page": page, "limit": len(items), "total": len(items), "list": items}
+
+    def searchContent(self, key, quick, pg="1"):
+        page = int(pg)
+        keyword = str(key or "").strip()
+        if not keyword:
+            return {"page": page, "limit": 0, "total": 0, "list": []}
+        items = self._list_items(keyword, page)
+        return {"page": page, "limit": len(items), "total": len(items), "list": items}
+
+    def detailContent(self, ids):
+        result = {"list": []}
+        for raw_id in ids:
+            vod_id = str(raw_id or "").strip()
+            if not vod_id:
+                continue
+            payload = self._get_json(f"{self.api}?id={quote(vod_id)}")
+            data = payload.get("data", []) if isinstance(payload, dict) else []
+            if not data:
+                continue
+            first = data[0]
+            base_name = self._clean_title(re.sub(r"\s+\d+\s*$", "", str(first.get("title", ""))))
+            episodes = []
+            for index, item in enumerate(data, start=1):
+                play_id = str(item.get("video_id", "")).strip()
+                if not play_id:
+                    continue
+                episode_name = self._clean_title(str(item.get("title", f"第{index}集")))
+                episodes.append(f"{episode_name}${play_id}")
+            result["list"].append(
+                {
+                    "vod_id": vod_id,
+                    "vod_name": base_name,
+                    "vod_pic": str(first.get("cover", "")),
+                    "vod_area": "中国",
+                    "vod_remarks": f"更新至{len(episodes)}集",
+                    "vod_content": f"{base_name} 短剧，共 {len(episodes)} 集",
+                    "vod_play_from": "红果短剧",
+                    "vod_play_url": "#".join(episodes),
+                }
+            )
+        return result
+
+    def playerContent(self, flag, id, vipFlags):
+        try:
+            payload = self._get_json(f"{self.api}?video_id={quote(str(id or '').strip())}")
+            data = payload.get("data", {}) if isinstance(payload, dict) else {}
+            qualities = self._sort_qualities(data.get("qualities", []))
+            best = next(
+                (item for item in qualities if str(item.get("download_url", "")).strip()),
+                {},
+            )
+            return {
+                "parse": 0,
+                "jx": 0,
+                "url": str(best.get("download_url", "")).strip(),
+                "header": {},
+            }
+        except Exception:
+            return {"parse": 0, "jx": 0, "url": "", "header": {}}
+
+    def _list_items(self, name, page):
+        payload = self._get_json(f"{self.api}?name={quote(str(name))}&page={int(page)}")
+        data = payload.get("data", []) if isinstance(payload, dict) else []
+        return [self._map_vod(item) for item in data]
+
+    def _map_vod(self, item):
+        total = item.get("totalChapterNum")
+        return {
+            "vod_id": str(item.get("id", "")),
+            "vod_name": self._clean_title(item.get("title", "")),
+            "vod_pic": str(item.get("cover", "")),
+            "vod_remarks": f"更新至{total}集" if total else "",
+        }
+
+    def _clean_title(self, text):
+        value = str(text or "")
+        value = re.sub(r"[【\[]热播(?:好剧|短剧)?[】\]]", "", value)
+        value = re.sub(r"[【\[]新剧(?:热播)?[】\]]", "", value)
+        return re.sub(r"\s+", " ", value).strip()
+
+    def _sort_qualities(self, items):
+        priority = {"1080p": 3, "sc": 2, "sd": 1}
+        return sorted(
+            items or [],
+            key=lambda item: priority.get(str(item.get("quality", "")), 0),
+            reverse=True,
+        )
+
+    def _get_json(self, url):
+        response = self.fetch(url, headers=dict(self.headers), timeout=10, verify=False)
+        if response.status_code != 200:
+            return {}
+        return json.loads(response.text or "{}")