feat: add juquanquan list and search flows

This commit is contained in:
Harold
2026-04-19 19:01:11 +08:00
parent 4da4d980a1
commit f6c847e871
2 changed files with 123 additions and 1 deletions
+55
View File
@@ -1,6 +1,7 @@
import unittest import unittest
from importlib.machinery import SourceFileLoader from importlib.machinery import SourceFileLoader
from pathlib import Path from pathlib import Path
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1] ROOT = Path(__file__).resolve().parents[1]
@@ -41,6 +42,60 @@ class TestJuQuanQuanSpider(unittest.TestCase):
], ],
) )
def test_parse_cards_extracts_compact_vod_ids(self):
html = """
<a class="module-poster-item module-item" href="/vod/123.html">
<img data-original="/cover.jpg" />
<div class="module-poster-item-title">示例影片</div>
<div class="module-item-note">更新至1集</div>
</a>
"""
self.assertEqual(
self.spider._parse_cards(html),
[
{
"vod_id": "vod/123",
"vod_name": "示例影片",
"vod_pic": "https://www.jqqzx.cc/cover.jpg",
"vod_remarks": "更新至1集",
}
],
)
@patch.object(Spider, "_request_html")
def test_home_video_content_limits_recommendations(self, mock_request_html):
mock_request_html.return_value = "".join(
f'<a class="module-poster-item module-item" href="/vod/{index}.html"><div class="module-poster-item-title">影片{index}</div></a>'
for index in range(1, 45)
)
result = self.spider.homeVideoContent()
self.assertEqual(len(result["list"]), 40)
self.assertEqual(result["list"][0]["vod_id"], "vod/1")
@patch.object(Spider, "_request_html")
def test_category_content_builds_page_result(self, mock_request_html):
mock_request_html.return_value = """
<a class="module-poster-item module-item" href="/vod/456.html">
<div class="module-poster-item-title">分类影片</div>
</a>
"""
result = self.spider.categoryContent("juji", "2", False, {})
self.assertEqual(mock_request_html.call_args.args[0], "https://www.jqqzx.cc/type/juji/page/2.html")
self.assertEqual(result["page"], 2)
self.assertEqual(result["pagecount"], 3)
self.assertEqual(result["list"][0]["vod_id"], "vod/456")
@patch.object(Spider, "_request_html")
def test_search_content_uses_suggest_api(self, mock_request_html):
mock_request_html.return_value = '{"list":[{"id":"777","name":"搜索结果","pic":"/pic.jpg"}]}'
result = self.spider.searchContent("繁花", False, "1")
self.assertEqual(
mock_request_html.call_args.args[0],
"https://www.jqqzx.cc/index.php/ajax/suggest?mid=1&wd=%E7%B9%81%E8%8A%B1",
)
self.assertEqual(result["list"][0]["vod_id"], "vod/777")
self.assertEqual(result["pagecount"], 1)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+68 -1
View File
@@ -2,7 +2,7 @@
import json import json
import re import re
import sys import sys
from urllib.parse import urljoin from urllib.parse import quote, urljoin
from base.spider import Spider as BaseSpider from base.spider import Spider as BaseSpider
@@ -77,3 +77,70 @@ class Spider(BaseSpider):
} }
) )
return items return items
def _request_html(self, path_or_url, headers=None):
target = path_or_url if str(path_or_url).startswith("http") else self._build_url(path_or_url)
request_headers = dict(self.headers)
if headers:
request_headers.update(headers)
response = self.fetch(target, headers=request_headers, timeout=10)
if response.status_code != 200:
return ""
return response.text or ""
def _parse_cards(self, html):
root = self.html(html)
if root is None:
return []
items = []
seen = set()
for anchor in root.xpath("//a[contains(@class,'module-poster-item') and contains(@class,'module-item')]"):
vod_id = self._encode_vod_id((anchor.xpath("./@href") or [""])[0])
if not vod_id or vod_id in seen:
continue
title = self._clean_text(
"".join(anchor.xpath(".//*[contains(@class,'module-poster-item-title')][1]//text()"))
or (anchor.xpath("./@title") or [""])[0]
or (anchor.xpath(".//img[1]/@alt") or [""])[0]
)
if not title:
continue
seen.add(vod_id)
pic = (
anchor.xpath(".//img[1]/@data-original")
or anchor.xpath(".//img[1]/@src")
or [""]
)[0]
note = self._clean_text("".join(anchor.xpath(".//*[contains(@class,'module-item-note')][1]//text()")))
items.append(
{
"vod_id": vod_id,
"vod_name": title,
"vod_pic": self._build_url(pic),
"vod_remarks": note,
}
)
return items
def homeVideoContent(self):
return {"list": self._parse_cards(self._request_html(self.host))[:40]}
def categoryContent(self, tid, pg, filter, extend):
page = int(pg)
items = self._parse_cards(self._request_html(self._build_url(f"/type/{tid}/page/{page}.html")))
return {
"page": page,
"pagecount": page + 1 if items else page,
"total": page * len(items) + (1 if items else 0),
"list": items,
}
def searchContent(self, key, quick, pg="1"):
page = int(pg)
keyword = self._clean_text(key)
if not keyword:
return {"page": page, "pagecount": 1, "total": 0, "list": []}
items = self._parse_search_list(
self._request_html(self._build_url(f"/index.php/ajax/suggest?mid=1&wd={quote(keyword)}"))
)
return {"page": page, "pagecount": 1, "total": len(items), "list": items}