fix: restore butailing files after docs commit

py/tests/test_不太灵.py

@@ -0,0 +1,185 @@
+import unittest
+from importlib.machinery import SourceFileLoader
+from pathlib import Path
+from unittest.mock import patch
+
+
+ROOT = Path(__file__).resolve().parents[1]
+MODULE = SourceFileLoader("butailing_spider", str(ROOT / "不太灵.py")).load_module()
+Spider = MODULE.Spider
+
+
+class FakeResponse:
+    def __init__(self, text, status_code=200):
+        self.text = text
+        self.status_code = status_code
+        self.encoding = "utf-8"
+
+
+class TestBuTaiLingSpider(unittest.TestCase):
+    def setUp(self):
+        Spider._instance = None
+        self.spider = Spider()
+        self.spider.init()
+
+    def test_home_content_returns_fixed_categories_and_filters(self):
+        with patch.object(
+            Spider,
+            "_request_api",
+            return_value={"t1": [{"title": "动作"}], "t2": [{"title": "中国"}], "t3": [], "t4": [], "t5": []},
+        ):
+            content = self.spider.homeContent(False)
+
+        self.assertEqual(
+            [item["type_id"] for item in content["class"]],
+            ["1", "2", "3", "4", "5"],
+        )
+        self.assertIn("1", content["filters"])
+        self.assertIn("2", content["filters"])
+        self.assertEqual(content["filters"]["1"][0]["key"], "sc")
+
+    @patch.object(Spider, "_request_api")
+    def test_home_video_content_requests_recent_hot_list(self, mock_request_api):
+        mock_request_api.return_value = [
+            {"doub_id": 11, "title": "示例电影", "image": "https://img.test/poster.jpg", "ejs": "HD"}
+        ]
+        content = self.spider.homeVideoContent()
+        self.assertEqual(mock_request_api.call_args.args[0], "getVideoList")
+        self.assertEqual(mock_request_api.call_args.args[1]["sc"], "3")
+        self.assertEqual(content["list"][0]["vod_id"], "11")
+        self.assertEqual(content["list"][0]["vod_remarks"], "HD")
+
+    def test_build_api_url_appends_credentials(self):
+        url = self.spider._build_api_url("getVideoList", {"page": 2, "limit": 24})
+        self.assertIn("getVideoList", url)
+        self.assertIn("app_id=83768d9ad4", url)
+        self.assertIn("identity=23734adac0301bccdcb107c4aa21f96c", url)
+        self.assertIn("page=2", url)
+
+    @patch("requests.get")
+    def test_request_api_parses_json_and_callback_payload(self, mock_get):
+        mock_get.return_value = FakeResponse('callback({"success":true,"code":200,"data":{"data":[{"doub_id": 1}]}})')
+        result = self.spider._request_api("getVideoList", {"page": 1})
+        self.assertEqual(result, [{"doub_id": 1}])
+
+    def test_parse_ext_object_supports_plain_urlencoded_and_base64_json(self):
+        plain = self.spider._parse_ext_object('{"sc":"动作"}')
+        encoded = self.spider._parse_ext_object("%7B%22sd%22%3A%22中国%22%7D")
+        wrapped = self.spider._parse_ext_object("eyJzZSI6IjIwMjYifQ==")
+        self.assertEqual(plain["sc"], "动作")
+        self.assertEqual(encoded["sd"], "中国")
+        self.assertEqual(wrapped["se"], "2026")
+
+    @patch.object(Spider, "_request_api")
+    def test_category_content_movie_uses_filter_params(self, mock_request_api):
+        mock_request_api.return_value = [
+            {"doub_id": 21, "title": "电影A", "image": "a.jpg", "ejs": "4K"},
+            {"doub_id": 21, "title": "电影A", "image": "a.jpg", "ejs": "4K"},
+        ]
+        result = self.spider.categoryContent("1", "2", False, '{"sc":"动作","sd":"中国","iswp":"1","status":"更新中"}')
+        self.assertEqual(mock_request_api.call_args.args[0], "getVideoMovieList")
+        self.assertEqual(
+            mock_request_api.call_args.args[1],
+            {"sa": 1, "page": 2, "sc": "动作", "sd": "中国", "se": "", "sf": "", "sh": "", "sg": "1", "iswp": 1},
+        )
+        self.assertEqual(result["page"], 2)
+        self.assertEqual(len(result["list"]), 1)
+        self.assertNotIn("pagecount", result)
+
+    @patch.object(Spider, "_request_api")
+    def test_category_content_hot_uses_local_dedupe_and_pagination(self, mock_request_api):
+        mock_request_api.return_value = [
+            {"doub_id": 1, "title": "A"},
+            {"doub_id": 1, "title": "A"},
+            {"doub_id": 2, "title": "B"},
+        ]
+        result = self.spider.categoryContent("3", "1", False, {})
+        self.assertEqual(mock_request_api.call_args.args[0], "getVideoList")
+        self.assertEqual(result["total"], 2)
+        self.assertEqual(len(result["list"]), 2)
+
+    @patch.object(Spider, "_request_api")
+    def test_search_content_filters_by_name_and_dedupes(self, mock_request_api):
+        mock_request_api.return_value = [
+            {"doub_id": 1, "title": "繁花", "image": "1.jpg"},
+            {"doub_id": 2, "title": "繁花幕后", "image": "2.jpg"},
+            {"doub_id": 2, "title": "繁花幕后", "image": "2.jpg"},
+            {"doub_id": 3, "title": "别的内容", "image": "3.jpg"},
+        ]
+        result = self.spider.searchContent("繁花", False, "1")
+        self.assertEqual(mock_request_api.call_args.args[0], "getVideoList")
+        self.assertEqual(result["total"], 2)
+        self.assertEqual([item["vod_id"] for item in result["list"]], ["1", "2"])
+
+    def test_detect_drive_type_by_url(self):
+        self.assertEqual(self.spider._detect_drive_type_by_url("https://pan.baidu.com/s/abc"), "baidu")
+        self.assertEqual(self.spider._detect_drive_type_by_url("https://pan.quark.cn/s/abc"), "quark")
+        self.assertEqual(self.spider._detect_drive_type_by_url("https://pan.xunlei.com/s/abc"), "xunlei")
+        self.assertEqual(self.spider._detect_drive_type_by_url("https://www.alipan.com/s/abc"), "aliyun")
+        self.assertEqual(self.spider._detect_drive_type_by_url("https://www.123pan.com/s/abc"), "a123")
+        self.assertEqual(self.spider._detect_drive_type_by_url("https://example.com/file"), "other")
+
+    def test_extract_pan_sources_creates_independent_lines_and_ignores_magnet_groups(self):
+        detail = {
+            "title": "示例详情",
+            "movies_online_seed": {
+                "夸克": [
+                    {"seed_name": "夸克资源A", "link": "https://pan.quark.cn/s/q1"},
+                    {"seed_name": "夸克资源A", "link": "https://pan.quark.cn/s/q1"},
+                ],
+                "百度网盘": [{"seed_name": "百度资源", "link": "https://pan.baidu.com/s/b1"}],
+                "未知": [{"seed_name": "未知资源", "link": "https://example.com/file"}],
+            },
+            "ecca": {"WEB-4K": [{"zlink": "magnet:?xt=1"}]},
+        }
+        play_from, play_url = self.spider._extract_pan_sources(detail)
+        self.assertEqual(play_from, "baidu#1$$$quark#1$$$other#1")
+        self.assertEqual(
+            play_url,
+            "baidu#1$https://pan.baidu.com/s/b1$$$quark#1$https://pan.quark.cn/s/q1$$$other#1$https://example.com/file",
+        )
+
+    @patch.object(Spider, "_request_api")
+    def test_detail_content_maps_metadata_and_pan_lines(self, mock_request_api):
+        mock_request_api.return_value = {
+            "doub_id": 88,
+            "title": "详情标题",
+            "image": "https://img.test/detail.jpg",
+            "ejs": "全 12 集",
+            "years": "2026",
+            "abstract": "这是一段剧情简介",
+            "performer": "演员甲/演员乙",
+            "director": "导演甲",
+            "production_area": "中国",
+            "movies_online_seed": {
+                "阿里": [{"seed_name": "阿里资源", "link": "https://www.alipan.com/s/ali1"}]
+            },
+        }
+        result = self.spider.detailContent(["88"])
+        vod = result["list"][0]
+        self.assertEqual(mock_request_api.call_args.args[0], "getVideoDetail")
+        self.assertEqual(mock_request_api.call_args.args[1], {"id": "88"})
+        self.assertEqual(vod["vod_id"], "88")
+        self.assertEqual(vod["vod_name"], "详情标题")
+        self.assertEqual(vod["vod_play_from"], "aliyun#1")
+        self.assertEqual(vod["vod_play_url"], "aliyun#1$https://www.alipan.com/s/ali1")
+
+    def test_player_content_passthroughs_share_links_with_tvbox_shape(self):
+        self.assertEqual(
+            self.spider.playerContent("quark#1", "https://pan.quark.cn/s/q1", {}),
+            {"parse": 0, "playUrl": "", "url": "https://pan.quark.cn/s/q1"},
+        )
+        self.assertEqual(
+            self.spider.playerContent("quark#1", "push://https://pan.quark.cn/s/q1", {}),
+            {"parse": 0, "playUrl": "", "url": "https://pan.quark.cn/s/q1"},
+        )
+
+    def test_player_content_returns_empty_url_for_non_pan_links(self):
+        self.assertEqual(
+            self.spider.playerContent("other#1", "https://example.com/file", {}),
+            {"parse": 0, "playUrl": "", "url": ""},
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

py/不太灵.py

@@ -0,0 +1,308 @@
+# coding=utf-8
+import base64
+import json
+import sys
+from urllib.parse import unquote, urlencode, urljoin
+
+import requests
+
+from base.spider import Spider as BaseSpider
+
+sys.path.append("..")
+
+
+class Spider(BaseSpider):
+    DRIVE_ORDER = {"baidu": 1, "quark": 2, "xunlei": 3, "aliyun": 4, "a123": 5, "other": 99}
+
+    def __init__(self):
+        self.name = "不太灵"
+        self.api_base = "https://web5.mukaku.com/prod/api/v1/"
+        self.app_id = "83768d9ad4"
+        self.identity = "23734adac0301bccdcb107c4aa21f96c"
+        self.timeout = 10
+        self.headers = {
+            "User-Agent": (
+                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                "AppleWebKit/537.36 (KHTML, like Gecko) "
+                "Chrome/124.0.0.0 Safari/537.36"
+            ),
+            "Referer": self.api_base,
+        }
+        self.classes = [
+            {"type_id": "1", "type_name": "电影"},
+            {"type_id": "2", "type_name": "电视剧"},
+            {"type_id": "3", "type_name": "近日热门"},
+            {"type_id": "4", "type_name": "本周热门"},
+            {"type_id": "5", "type_name": "本月热门"},
+        ]
+
+    def init(self, extend=""):
+        return None
+
+    def getName(self):
+        return self.name
+
+    def _build_api_url(self, endpoint, params=None):
+        query = {"app_id": self.app_id, "identity": self.identity}
+        query.update(params or {})
+        return urljoin(self.api_base, endpoint) + "?" + urlencode(query)
+
+    def _parse_payload(self, text):
+        raw = str(text or "").strip()
+        if raw.startswith("callback(") and raw.endswith(")"):
+            raw = raw[9:-1]
+        return json.loads(raw) if raw else {}
+
+    def _request_api(self, endpoint, params=None):
+        response = requests.get(
+            self._build_api_url(endpoint, params),
+            headers=dict(self.headers),
+            timeout=self.timeout,
+            verify=False,
+        )
+        if response.status_code != 200:
+            return [] if endpoint in {"getVideoList", "getVideoMovieList"} else None
+        payload = self._parse_payload(response.text)
+        if not payload or payload.get("success") is not True or payload.get("code") != 200:
+            return [] if endpoint in {"getVideoList", "getVideoMovieList"} else None
+        data = payload.get("data")
+        if endpoint in {"getVideoList", "getVideoMovieList"}:
+            if isinstance(data, dict) and isinstance(data.get("data"), list):
+                return data.get("data")
+            if isinstance(data, dict) and isinstance(data.get("list"), list):
+                return data.get("list")
+            return []
+        return data
+
+    def _map_filter_values(self, items):
+        values = [{"n": "不限", "v": "0"}]
+        for item in items or []:
+            title = str((item or {}).get("title") or "").strip()
+            if title:
+                values.append({"n": title, "v": title})
+        return values
+
+    def _build_filters(self, type_data):
+        if not isinstance(type_data, dict):
+            return {}
+        common = [
+            {"key": "sc", "name": "影视类型", "value": self._map_filter_values(type_data.get("t1"))},
+            {"key": "sd", "name": "制片地区", "value": self._map_filter_values(type_data.get("t2"))},
+            {"key": "se", "name": "上映年份", "value": self._map_filter_values(type_data.get("t3"))},
+            {"key": "sf", "name": "资源画质", "value": self._map_filter_values(type_data.get("t4"))},
+            {"key": "sh", "name": "影视标签", "value": self._map_filter_values(type_data.get("t5"))},
+        ]
+        return {
+            "1": common,
+            "2": common + [{"key": "status", "name": "剧集状态", "value": [{"n": "不限", "v": "0"}]}],
+        }
+
+    def _normalize_video(self, item):
+        data = item or {}
+        return {
+            "vod_id": str(data.get("doub_id") or data.get("id") or ""),
+            "vod_name": data.get("title") or "未知标题",
+            "vod_pic": data.get("image") or data.get("epic") or "",
+            "vod_remarks": data.get("ejs") or data.get("zqxd") or "",
+            "vod_year": data.get("years") or "",
+            "vod_content": data.get("abstract") or "",
+            "vod_actor": data.get("performer") or "",
+            "vod_director": data.get("director") or "",
+            "vod_area": data.get("production_area") or "",
+        }
+
+    def _parse_ext_object(self, ext):
+        candidates = [ext]
+        try:
+            decoded = base64.b64decode(str(ext or "")).decode("utf-8")
+            if decoded and decoded != ext:
+                candidates.append(decoded)
+        except Exception:
+            pass
+        for raw in list(candidates):
+            try:
+                decoded = unquote(str(raw))
+                if decoded != raw:
+                    candidates.append(decoded)
+            except Exception:
+                pass
+        for raw in candidates:
+            try:
+                value = json.loads(raw)
+                if isinstance(value, dict):
+                    return value
+            except Exception:
+                continue
+        return {}
+
+    def _normalize_filter_value(self, value):
+        text = str(value or "").strip()
+        return "" if text in {"", "0", "不限", "all"} else text
+
+    def _to01(self, value, fallback=0):
+        text = str(value if value is not None else fallback).strip().lower()
+        if text in {"1", "true", "yes", "on"}:
+            return 1
+        if text in {"0", "false", "no", "off", ""}:
+            return 0
+        return fallback
+
+    def _dedupe_by_vod_id(self, items):
+        result = []
+        seen = set()
+        for item in items or []:
+            vod_id = str((item or {}).get("doub_id") or (item or {}).get("id") or "")
+            if not vod_id or vod_id in seen:
+                continue
+            seen.add(vod_id)
+            result.append(item)
+        return result
+
+    def _paginate_items(self, items, page, limit):
+        page_num = max(1, int(page))
+        size = max(1, int(limit))
+        total = len(items)
+        start = (page_num - 1) * size
+        return {"page": page_num, "limit": size, "total": total, "list": items[start:start + size]}
+
+    def _build_movie_params(self, tid, pg, ext):
+        ext_obj = self._parse_ext_object(ext)
+        return {
+            "sa": int(tid),
+            "page": int(pg),
+            "sc": self._normalize_filter_value(ext_obj.get("sc")),
+            "sd": self._normalize_filter_value(ext_obj.get("sd")),
+            "se": self._normalize_filter_value(ext_obj.get("se")),
+            "sf": self._normalize_filter_value(ext_obj.get("sf")),
+            "sh": self._normalize_filter_value(ext_obj.get("sh")),
+            "sg": self._normalize_filter_value(ext_obj.get("sg")) or "1",
+            "iswp": self._to01(ext_obj.get("iswp"), 0),
+        }
+
+    def _normalize_drive_key(self, value):
+        text = str(value or "").strip().lower()
+        if text in {"alipan", "aliyun", "aliyundrive", "ali", "阿里"}:
+            return "aliyun"
+        if "quark" in text or "夸克" in text:
+            return "quark"
+        if "baidu" in text or "百度" in text:
+            return "baidu"
+        if "xunlei" in text or "迅雷" in text:
+            return "xunlei"
+        if "123" in text:
+            return "a123"
+        return "other"
+
+    def _detect_drive_type_by_url(self, url):
+        text = str(url or "").lower()
+        if "pan.quark.cn" in text:
+            return "quark"
+        if "pan.baidu.com" in text:
+            return "baidu"
+        if "pan.xunlei.com" in text:
+            return "xunlei"
+        if "alipan.com" in text or "aliyundrive.com" in text:
+            return "aliyun"
+        if "123865.com" in text or "123684.com" in text or "123pan.com" in text:
+            return "a123"
+        return "other"
+
+    def _is_supported_pan_url(self, url):
+        text = str(url or "").strip().lower()
+        return any(
+            token in text
+            for token in [
+                "pan.quark.cn",
+                "pan.baidu.com",
+                "pan.xunlei.com",
+                "alipan.com",
+                "aliyundrive.com",
+                "123865.com",
+                "123684.com",
+                "123pan.com",
+                "drive.uc.cn",
+            ]
+        )
+
+    def _extract_pan_sources(self, detail):
+        groups = (detail or {}).get("movies_online_seed")
+        if not isinstance(groups, dict):
+            return "", ""
+        seen = set()
+        counters = {}
+        entries = []
+        for group_name, items in groups.items():
+            for item in items or []:
+                link = str((item or {}).get("link") or "").strip()
+                if not link or link in seen:
+                    continue
+                seen.add(link)
+                kind = self._detect_drive_type_by_url(link)
+                if kind == "other":
+                    kind = self._normalize_drive_key(group_name)
+                counters[kind] = counters.get(kind, 0) + 1
+                line_name = f"{kind}@{counters[kind]}"
+                entries.append((line_name, link))
+        entries.sort(
+            key=lambda pair: (
+                self.DRIVE_ORDER.get(pair[0].split("@")[0], 999),
+                int(pair[0].split("@")[1]),
+                pair[0],
+            )
+        )
+        return "$$$".join(name for name, _ in entries), "$$$".join(f"{name}${url}" for name, url in entries)
+
+    def homeContent(self, filter):
+        return {"class": self.classes, "filters": self._build_filters(self._request_api("getVideoTypeList", {}))}
+
+    def homeVideoContent(self):
+        items = self._request_api("getVideoList", {"sc": "3", "limit": 20}) or []
+        return {"list": [self._normalize_video(item) for item in items]}
+
+    def categoryContent(self, tid, pg, filter, extend):
+        page = int(pg)
+        if str(tid) in {"1", "2"}:
+            items = self._dedupe_by_vod_id(
+                self._request_api("getVideoMovieList", self._build_movie_params(tid, page, extend)) or []
+            )
+            return {
+                "page": page,
+                "limit": len(items) or 20,
+                "total": page * 20 + len(items),
+                "list": [self._normalize_video(item) for item in items],
+            }
+        items = self._dedupe_by_vod_id(self._request_api("getVideoList", {"sc": str(tid), "page": 1, "limit": 300}) or [])
+        paged = self._paginate_items(items, page, 20)
+        paged["list"] = [self._normalize_video(item) for item in paged["list"]]
+        return paged
+
+    def searchContent(self, key, quick, pg="1"):
+        keyword = str(key or "").strip()
+        page = int(pg)
+        if not keyword:
+            return {"page": page, "total": 0, "list": []}
+        items = self._request_api("getVideoList", {"sb": keyword, "page": 1, "limit": 300}) or []
+        matched = [item for item in items if keyword.lower() in str((item or {}).get("title") or "").lower()]
+        paged = self._paginate_items(self._dedupe_by_vod_id(matched), page, 20)
+        paged["list"] = [self._normalize_video(item) for item in paged["list"]]
+        return paged
+
+    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": []}
+        data = self._request_api("getVideoDetail", {"id": vod_id})
+        if not isinstance(data, dict):
+            return {"list": []}
+        vod = self._normalize_video(data)
+        play_from, play_url = self._extract_pan_sources(data)
+        vod["vod_play_from"] = play_from
+        vod["vod_play_url"] = play_url
+        return {"list": [vod]}
+
+    def playerContent(self, flag, id, vipFlags):
+        raw = str(id or "").strip()
+        url = raw[7:] if raw.startswith("push://") else raw
+        if self._is_supported_pan_url(url):
+            return {"parse": 0, "playUrl": "", "url": url}
+        return {"parse": 0, "playUrl": "", "url": ""}