feat: add tengxun search

This commit is contained in:
Harold
2026-04-23 21:11:05 +08:00
parent 06dd147c6e
commit c02edac650
2 changed files with 117 additions and 0 deletions
+53
View File
@@ -156,6 +156,59 @@ class TestTencentSpider(unittest.TestCase):
"第1集$https://v.qq.com/x/cover/cid001/vid001.html#终极预告$https://v.qq.com/x/cover/cid001/vid002.html",
)
@patch.object(Spider, "post")
def test_search_content_collects_normal_and_area_results(self, mock_post):
mock_post.return_value = SimpleNamespace(
json=lambda: {
"data": {
"normalList": {
"itemList": [
{
"doc": {"id": "mzc001234567890"},
"videoInfo": {
"title": "<em>庆余年</em>",
"imgUrl": "https://img.test/1.jpg",
"firstLine": "热播",
},
}
]
},
"areaBoxList": [
{
"itemList": [
{
"doc": {"id": "mzc009876543210"},
"videoInfo": {
"title": "雪中悍刀行",
"imgUrl": "https://img.test/2.jpg",
"secondLine": "完结",
},
}
]
}
],
}
}
)
result = self.spider.searchContent("庆余年", False, "2")
self.assertEqual(result["page"], 2)
self.assertEqual(result["limit"], 30)
self.assertEqual(result["list"][0]["vod_name"], "庆余年")
self.assertEqual(result["list"][1]["vod_remarks"], "完结")
payload = mock_post.call_args.kwargs["json"]
self.assertEqual(payload["query"], "庆余年")
self.assertEqual(payload["pagenum"], 1)
@patch.object(Spider, "post")
def test_search_content_returns_empty_page_on_error(self, mock_post):
mock_post.side_effect = RuntimeError("network error")
self.assertEqual(
self.spider.searchContent("失败", False, "1"),
{"list": [], "page": 1, "pagecount": 1, "limit": 30, "total": 0},
)
if __name__ == "__main__":
unittest.main()
+64
View File
@@ -184,3 +184,67 @@ class Spider(BaseSpider):
vod["vod_play_url"] = "#".join(play_items)
result["list"].append(vod)
return result
def searchContent(self, key, quick, pg="1"):
url = (
"https://pbaccess.video.qq.com/trpc.videosearch.mobile_search."
"MultiTerminalSearch/MbSearch?vplatform=2"
)
headers = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/98.0.4758.139 Safari/537.36"
),
"Content-Type": "application/json",
"Origin": "https://v.qq.com",
"Referer": "https://v.qq.com/",
}
payload = {
"version": "25042201",
"clientType": 1,
"query": key,
"pagenum": int(pg) - 1,
"pagesize": 30,
"extraInfo": {
"isNewMarkLabel": "1",
"multi_terminal_pc": "1",
"themeType": "1",
},
}
try:
response = self.post(url, headers=headers, json=payload)
data = response.json()
except Exception:
return {"list": [], "page": 1, "pagecount": 1, "limit": 30, "total": 0}
videos = []
def process_item(item):
if not item or not item.get("doc") or not item["doc"].get("id") or not item.get("videoInfo"):
return
if len(item["doc"]["id"]) <= 11:
return
info = item["videoInfo"]
videos.append(
{
"vod_id": item["doc"]["id"],
"vod_name": re.sub(r"</?em>", "", info.get("title", "")),
"vod_pic": info.get("imgUrl", ""),
"vod_remarks": info.get("firstLine") or info.get("secondLine") or "",
}
)
for item in (((data.get("data") or {}).get("normalList") or {}).get("itemList") or []):
process_item(item)
for area in ((data.get("data") or {}).get("areaBoxList") or []):
for item in (area.get("itemList") or []):
process_item(item)
page = int(pg)
return {
"list": videos,
"page": page,
"pagecount": page + 1 if len(videos) >= 20 else page,
"limit": 30,
"total": 999 if videos else 0,
}