From 60f54818a8657acf957f7216ec9c54c7f85e365c Mon Sep 17 00:00:00 2001
From: Harold <8866033@gmail.com>
Date: Fri, 24 Apr 2026 16:33:34 +0800
Subject: [PATCH] feat: add jinpai spider

---
 py/tests/test_金牌.py | 214 +++++++++++++++++++++++++++++++++++++
 py/金牌.py            | 240 ++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 454 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..d9ececa
--- /dev/null
+++ b/py/tests/test_金牌.py
@@ -0,0 +1,214 @@
+import hashlib
+import unittest
+from importlib.machinery import SourceFileLoader
+from pathlib import Path
+from unittest.mock import patch
+
+
+ROOT = Path(__file__).resolve().parents[1]
+MODULE = SourceFileLoader("jinpai_spider", str(ROOT / "金牌.py")).load_module()
+Spider = MODULE.Spider
+
+
+class TestJinPaiSpider(unittest.TestCase):
+    def setUp(self):
+        Spider._instance = None
+        self.spider = Spider()
+        self.spider.init()
+
+    @patch("jinpai_spider.time.time", return_value=1713952212.345)
+    def test_signed_headers_generates_expected_hashes(self, _mock_time):
+        headers = self.spider._signed_headers({"pageNum": "1", "type1": "2"})
+        expected_t = "1713952212345"
+        expected_obj = "pageNum=1&type1=2&key=cb808529bae6b6be45ecfab29a4889bc&t=1713952212345"
+        expected_sign = hashlib.sha1(
+            hashlib.md5(expected_obj.encode("utf-8")).hexdigest().encode("utf-8")
+        ).hexdigest()
+        self.assertEqual(headers["t"], expected_t)
+        self.assertEqual(headers["sign"], expected_sign)
+        self.assertEqual(headers["Referer"], "https://m.jiabaide.cn/")
+
+    @patch.object(Spider, "_fetch_json")
+    def test_home_content_returns_class_filters_and_hot_list(self, mock_fetch_json):
+        mock_fetch_json.side_effect = [
+            {"data": [{"typeId": 1, "typeName": "电影"}, {"typeId": 2, "typeName": "剧集"}]},
+            {
+                "data": {
+                    "1": {
+                        "typeList": [{"itemText": "动作", "itemValue": "dongzuo"}],
+                        "plotList": [{"itemText": "剧情"}],
+                        "districtList": [{"itemText": "中国香港"}],
+                        "languageList": [{"itemText": "粤语"}],
+                        "yearList": [{"itemText": "2026"}],
+                    }
+                }
+            },
+            {
+                "data": [
+                    {
+                        "vodId": 9,
+                        "vodName": "热播片",
+                        "vodPic": "https://img/9.jpg",
+                        "vodRemarks": "更新至10集",
+                        "vodDoubanScore": "8.6",
+                        "vodPubdate": "2026-01-01",
+                        "typeId": 1,
+                        "typeName": "电影",
+                    }
+                ]
+            },
+        ]
+        result = self.spider.homeContent(False)
+        self.assertEqual(
+            result["class"],
+            [{"type_id": "1", "type_name": "电影"}, {"type_id": "2", "type_name": "剧集"}],
+        )
+        self.assertEqual(result["filters"]["1"][0]["key"], "type")
+        self.assertEqual(result["filters"]["1"][0]["value"][0], {"n": "全部", "v": ""})
+        self.assertEqual(result["filters"]["1"][0]["value"][1], {"n": "动作", "v": "dongzuo"})
+        self.assertEqual(result["filters"]["1"][-1]["key"], "by")
+        self.assertEqual(result["list"][0]["vod_remarks"], "更新至10集_8.6")
+        self.assertEqual(result["list"][0]["vod_year"], "2026")
+
+    def test_home_video_content_returns_empty_list(self):
+        self.assertEqual(self.spider.homeVideoContent(), {"list": []})
+
+    @patch.object(Spider, "_fetch_json")
+    def test_category_content_maps_filters_to_api_payload(self, mock_fetch_json):
+        mock_fetch_json.return_value = {
+            "data": {
+                "list": [
+                    {
+                        "vodId": 11,
+                        "vodName": "分类片",
+                        "vodPic": "https://img/11.jpg",
+                        "vodRemarks": "完结",
+                        "vodDoubanScore": "7.1",
+                        "vodPubdate": "2025-09-09",
+                        "typeId": 1,
+                        "typeName": "电影",
+                    }
+                ]
+            }
+        }
+        result = self.spider.categoryContent(
+            "1",
+            "2",
+            False,
+            {"area": "中国香港", "lang": "粤语", "year": "2025", "by": "4", "type": "dongzuo", "class": "剧情"},
+        )
+        self.assertEqual(
+            mock_fetch_json.call_args.args,
+            (
+                "/api/mw-movie/anonymous/video/list",
+                {
+                    "area": "中国香港",
+                    "lang": "粤语",
+                    "pageNum": "2",
+                    "pageSize": "30",
+                    "sort": "4",
+                    "sortBy": "1",
+                    "type": "dongzuo",
+                    "type1": "1",
+                    "v_class": "剧情",
+                    "year": "2025",
+                },
+            ),
+        )
+        self.assertEqual(result["page"], 2)
+        self.assertEqual(result["limit"], 30)
+        self.assertEqual(result["list"][0]["vod_id"], "11")
+        self.assertNotIn("pagecount", result)
+
+    def test_search_content_returns_empty_for_blank_keyword(self):
+        self.assertEqual(self.spider.searchContent("", False, "1"), {"page": 1, "limit": 30, "total": 0, "list": []})
+
+    @patch.object(Spider, "_fetch_json")
+    def test_search_content_maps_keyword_and_page(self, mock_fetch_json):
+        mock_fetch_json.return_value = {
+            "data": {
+                "list": [
+                    {
+                        "vodId": 21,
+                        "vodName": "搜索片",
+                        "vodPic": "https://img/21.jpg",
+                        "vodRemarks": "更新",
+                        "vodDoubanScore": "8.0",
+                        "vodPubdate": "2024-01-01",
+                        "typeId": 2,
+                        "typeName": "剧集",
+                    }
+                ]
+            }
+        }
+        result = self.spider.searchContent("繁花", False, "3")
+        self.assertEqual(
+            mock_fetch_json.call_args.args,
+            (
+                "/api/mw-movie/anonymous/video/searchByWordPageable",
+                {"keyword": "繁花", "pageNum": "3", "pageSize": "30"},
+            ),
+        )
+        self.assertEqual(result["page"], 3)
+        self.assertEqual(result["list"][0]["vod_name"], "搜索片")
+        self.assertEqual(result["list"][0]["vod_year"], "2024")
+
+    @patch.object(Spider, "_fetch_json")
+    def test_detail_content_maps_metadata_and_episode_ids(self, mock_fetch_json):
+        mock_fetch_json.return_value = {
+            "data": {
+                "vodId": 88,
+                "vodName": "详情片",
+                "vodPic": "https://img/88.jpg",
+                "vodClass": "动作",
+                "vodRemarks": "更新至12集",
+                "vodYear": "2026",
+                "vodArea": "中国香港",
+                "vodLang": "粤语",
+                "vodDirector": "导演甲",
+                "vodActor": "演员甲/演员乙",
+                "vodContent": "一段简介",
+                "episodeList": [{"name": "第1集", "nid": 9001}, {"name": "第2集", "nid": 9002}],
+            }
+        }
+        result = self.spider.detailContent(["88"])
+        vod = result["list"][0]
+        self.assertEqual(mock_fetch_json.call_args.args, ("/api/mw-movie/anonymous/video/detail", {"id": "88"}))
+        self.assertEqual(vod["vod_id"], "88")
+        self.assertEqual(vod["vod_name"], "详情片")
+        self.assertEqual(vod["vod_play_from"], "金牌线路")
+        self.assertEqual(vod["vod_play_url"], "第1集$88@9001#第2集$88@9002")
+
+    def test_detail_content_returns_empty_for_blank_id(self):
+        self.assertEqual(self.spider.detailContent([""]), {"list": []})
+
+    @patch.object(Spider, "_fetch_json")
+    def test_player_content_returns_first_episode_url_with_headers(self, mock_fetch_json):
+        mock_fetch_json.return_value = {
+            "data": {
+                "list": [
+                    {"name": "1080P", "url": "https://cdn.example/1080.m3u8"},
+                    {"name": "720P", "url": "https://cdn.example/720.m3u8"},
+                ]
+            }
+        }
+        result = self.spider.playerContent("金牌线路", "88@9001", {})
+        self.assertEqual(
+            mock_fetch_json.call_args.args,
+            ("/api/mw-movie/anonymous/v2/video/episode/url", {"clientType": "3", "id": "88", "nid": "9001"}),
+        )
+        self.assertEqual(
+            result,
+            {
+                "parse": 0,
+                "playUrl": "",
+                "url": "https://cdn.example/1080.m3u8",
+                "header": {"User-Agent": self.spider.mobile_ua},
+            },
+        )
+
+    def test_player_content_rejects_broken_play_id(self):
+        self.assertEqual(
+            self.spider.playerContent("金牌线路", "broken", {}),
+            {"parse": 0, "playUrl": "", "url": "", "header": {}},
+        )
diff --git a/py/金牌.py b/py/金牌.py
new file mode 100644
index 0000000..e7834df
--- /dev/null
+++ b/py/金牌.py
@@ -0,0 +1,240 @@
+# coding=utf-8
+import hashlib
+import json
+import sys
+import time
+
+from base.spider import Spider as BaseSpider
+
+sys.path.append("..")
+
+
+class Spider(BaseSpider):
+    def __init__(self):
+        self.name = "金牌"
+        self.host = "https://m.jiabaide.cn"
+        self.app_key = "cb808529bae6b6be45ecfab29a4889bc"
+        self.mobile_ua = (
+            "Mozilla/5.0 (Linux; Android 11; Pixel 5) "
+            "AppleWebKit/537.36 (KHTML, like Gecko) "
+            "Chrome/90.0.4430.91 Mobile Safari/537.36"
+        )
+        self.headers = {
+            "User-Agent": self.mobile_ua,
+            "Referer": self.host + "/",
+        }
+        self.class_name_map = {
+            "typeList": ("type", "类型"),
+            "plotList": ("class", "剧情"),
+            "districtList": ("area", "地区"),
+            "languageList": ("lang", "语言"),
+            "yearList": ("year", "年份"),
+            "serialList": ("by", "排序"),
+        }
+        self.sort_values = [
+            {"n": "最近更新", "v": "1"},
+            {"n": "添加时间", "v": "2"},
+            {"n": "人气高低", "v": "3"},
+            {"n": "评分高低", "v": "4"},
+        ]
+
+    def init(self, extend=""):
+        return None
+
+    def getName(self):
+        return self.name
+
+    def homeVideoContent(self):
+        return {"list": []}
+
+    def _obj_to_form(self, obj=None):
+        pairs = []
+        for key, value in (obj or {}).items():
+            if value in ("", None):
+                continue
+            pairs.append(f"{key}={value}")
+        return "&".join(pairs)
+
+    def _signed_headers(self, params=None):
+        current = str(int(time.time() * 1000))
+        sign_obj = dict(params or {})
+        sign_obj["key"] = self.app_key
+        sign_obj["t"] = current
+        obj_str = self._obj_to_form(sign_obj)
+        md5_value = hashlib.md5(obj_str.encode("utf-8")).hexdigest()
+        sign = hashlib.sha1(md5_value.encode("utf-8")).hexdigest()
+        headers = dict(self.headers)
+        headers["t"] = current
+        headers["sign"] = sign
+        return headers
+
+    def _fetch_json(self, path, params=None):
+        target = self.host + str(path)
+        query = self._obj_to_form(params or {})
+        if query:
+            target = target + "?" + query
+        response = self.fetch(target, headers=self._signed_headers(params or {}), timeout=10, verify=False)
+        if response.status_code != 200:
+            return {}
+        try:
+            data = json.loads(response.text or "{}")
+        except Exception:
+            return {}
+        if str(data.get("code", "200")) not in ("200", "0", ""):
+            return {}
+        return data
+
+    def _map_vod(self, item):
+        item = item or {}
+        pubdate = str(item.get("vodPubdate", "")).strip()
+        year = pubdate.split("-")[0] if pubdate else ""
+        remarks = [str(item.get("vodRemarks", "")).strip(), str(item.get("vodDoubanScore", "")).strip()]
+        return {
+            "vod_id": str(item.get("vodId", "")),
+            "vod_name": str(item.get("vodName", "")),
+            "vod_pic": str(item.get("vodPic", "")),
+            "vod_remarks": "_".join([value for value in remarks if value]),
+            "vod_year": year,
+            "type_id": str(item.get("typeId", "")),
+            "type_name": str(item.get("typeName", "")),
+        }
+
+    def _build_filters(self):
+        classes = []
+        filters = {}
+        type_res = self._fetch_json("/api/mw-movie/anonymous/get/filer/type")
+        for item in type_res.get("data", []):
+            classes.append(
+                {
+                    "type_id": str(item.get("typeId", "")),
+                    "type_name": str(item.get("typeName", "")),
+                }
+            )
+        filter_res = self._fetch_json("/api/mw-movie/anonymous/v1/get/filer/list")
+        filter_data = filter_res.get("data", {})
+        for cls in classes:
+            tid = cls["type_id"]
+            filters[tid] = []
+            current = (filter_data or {}).get(tid, {})
+            for raw_key, (key, name) in self.class_name_map.items():
+                values = [{"n": "全部", "v": ""}]
+                if raw_key == "serialList":
+                    values.extend(self.sort_values)
+                else:
+                    for item in current.get(raw_key, []):
+                        label = str(item.get("itemText", "")).strip()
+                        if not label:
+                            continue
+                        value_key = "itemValue" if raw_key == "typeList" else "itemText"
+                        values.append({"n": label, "v": str(item.get(value_key, "")).strip()})
+                if len(values) > 1:
+                    filters[tid].append({"key": key, "name": name, "value": values})
+        return classes, filters
+
+    def _page_result(self, items, pg, limit=30):
+        page = max(int(str(pg or 1)), 1)
+        return {"page": page, "limit": limit, "total": page * limit + len(items), "list": items}
+
+    def homeContent(self, filter):
+        classes, filters = self._build_filters()
+        home = self._fetch_json("/api/mw-movie/anonymous/home/hotSearch")
+        return {
+            "class": classes,
+            "filters": filters,
+            "list": [self._map_vod(item) for item in home.get("data", [])],
+        }
+
+    def categoryContent(self, tid, pg, filter, extend):
+        page = max(int(str(pg or 1)), 1)
+        values = dict(extend or {})
+        payload = {
+            "area": str(values.get("area", "")),
+            "lang": str(values.get("lang", "")),
+            "pageNum": str(page),
+            "pageSize": "30",
+            "sort": str(values.get("by", "1")),
+            "sortBy": "1",
+            "type": str(values.get("type", "")),
+            "type1": str(tid),
+            "v_class": str(values.get("class", "")),
+            "year": str(values.get("year", "")),
+        }
+        result = self._fetch_json("/api/mw-movie/anonymous/video/list", payload)
+        items = [self._map_vod(item) for item in ((result.get("data") or {}).get("list") or [])]
+        return self._page_result(items, page, 30)
+
+    def searchContent(self, key, quick, pg="1"):
+        page = max(int(str(pg or 1)), 1)
+        keyword = str(key or "").strip()
+        if not keyword:
+            return {"page": page, "limit": 30, "total": 0, "list": []}
+        result = self._fetch_json(
+            "/api/mw-movie/anonymous/video/searchByWordPageable",
+            {"keyword": keyword, "pageNum": str(page), "pageSize": "30"},
+        )
+        items = [self._map_vod(item) for item in ((result.get("data") or {}).get("list") or [])]
+        return self._page_result(items, page, 30)
+
+    def _build_play_id(self, vod_id, nid):
+        return f"{str(vod_id).strip()}@{str(nid).strip()}"
+
+    def _parse_play_id(self, value):
+        parts = str(value or "").split("@", 1)
+        if len(parts) != 2:
+            return {"id": "", "nid": ""}
+        sid = parts[0].strip()
+        nid = parts[1].strip()
+        if not sid or not nid:
+            return {"id": "", "nid": ""}
+        return {"id": sid, "nid": nid}
+
+    def detailContent(self, ids):
+        vod_id = str(ids[0] if isinstance(ids, list) and ids else ids or "").strip()
+        if not vod_id:
+            return {"list": []}
+        result = self._fetch_json("/api/mw-movie/anonymous/video/detail", {"id": vod_id})
+        data = result.get("data") or {}
+        if not data:
+            return {"list": []}
+        episodes = []
+        for item in data.get("episodeList", []):
+            name = str(item.get("name", "")).strip()
+            nid = str(item.get("nid", "")).strip()
+            if not name or not nid:
+                continue
+            episodes.append(f"{name}${self._build_play_id(data.get('vodId', vod_id), nid)}")
+        vod = {
+            "vod_id": str(data.get("vodId", vod_id)),
+            "vod_name": str(data.get("vodName", "")),
+            "vod_pic": str(data.get("vodPic", "")),
+            "type_name": str(data.get("vodClass", "")),
+            "vod_remarks": str(data.get("vodRemarks", "")),
+            "vod_year": str(data.get("vodYear", "")),
+            "vod_area": str(data.get("vodArea", "")),
+            "vod_lang": str(data.get("vodLang", "")),
+            "vod_director": str(data.get("vodDirector", "")),
+            "vod_actor": str(data.get("vodActor", "")),
+            "vod_content": str(data.get("vodContent", "")),
+            "vod_play_from": "金牌线路" if episodes else "",
+            "vod_play_url": "#".join(episodes),
+        }
+        return {"list": [vod]}
+
+    def playerContent(self, flag, id, vipFlags):
+        parsed = self._parse_play_id(id)
+        if not parsed["id"] or not parsed["nid"]:
+            return {"parse": 0, "playUrl": "", "url": "", "header": {}}
+        result = self._fetch_json(
+            "/api/mw-movie/anonymous/v2/video/episode/url",
+            {"clientType": "3", "id": parsed["id"], "nid": parsed["nid"]},
+        )
+        for item in ((result.get("data") or {}).get("list") or []):
+            url = str(item.get("url", "")).strip()
+            if url:
+                return {
+                    "parse": 0,
+                    "playUrl": "",
+                    "url": url,
+                    "header": {"User-Agent": self.mobile_ua},
+                }
+        return {"parse": 0, "playUrl": "", "url": "", "header": {}}