diff --git a/py/tests/test_不太灵.py b/py/tests/test_不太灵.py index aab9b44..2b2ac79 100644 --- a/py/tests/test_不太灵.py +++ b/py/tests/test_不太灵.py @@ -62,6 +62,55 @@ class TestBuTaiLingSpider(unittest.TestCase): 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"]) + if __name__ == "__main__": unittest.main() diff --git a/py/不太灵.py b/py/不太灵.py index abc5693..2b8198e 100644 --- a/py/不太灵.py +++ b/py/不太灵.py @@ -1,7 +1,8 @@ # coding=utf-8 +import base64 import json import sys -from urllib.parse import urlencode, urljoin +from urllib.parse import unquote, urlencode, urljoin import requests @@ -108,9 +109,105 @@ class Spider(BaseSpider): "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 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