feat: scaffold butailing spider
This commit is contained in:
Harold
@@ -0,0 +1,67 @@
|
||||
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}])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,116 @@
|
||||
# coding=utf-8
|
||||
import json
|
||||
import sys
|
||||
from urllib.parse import urlencode, urljoin
|
||||
|
||||
import requests
|
||||
|
||||
from base.spider import Spider as BaseSpider
|
||||
|
||||
sys.path.append("..")
|
||||
|
||||
|
||||
class Spider(BaseSpider):
|
||||
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 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]}
|
||||
Reference in New Issue
Block a user