From 2be71f7c20543828a4ead085053aa7932e9cb13c Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Mon, 20 Apr 2026 10:58:25 +0800 Subject: [PATCH] feat: add zxzj list and search parsing --- py/tests/test_在线之家.py | 62 +++++++++++++++++++++++++++++++++ py/在线之家.py | 72 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 134 insertions(+) diff --git a/py/tests/test_在线之家.py b/py/tests/test_在线之家.py index b499602..cf02424 100644 --- a/py/tests/test_在线之家.py +++ b/py/tests/test_在线之家.py @@ -1,6 +1,7 @@ import unittest from importlib.machinery import SourceFileLoader from pathlib import Path +from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] @@ -36,6 +37,67 @@ class TestZXZJSpider(unittest.TestCase): url = self.spider._build_category_url("2", "1", {}) self.assertEqual(url, "https://www.zxzjhd.com/vodshow/2--------1---.html") + def test_fix_json_wrapped_html_unwraps_html_string(self): + wrapped = "\"ok\"" + self.assertEqual(self.spider._fix_json_wrapped_html(wrapped), "ok") + + def test_parse_cards_extracts_short_vod_id_title_cover_and_remarks(self): + html = """ + + """ + cards = self.spider._parse_cards(html) + self.assertEqual( + cards, + [ + { + "vod_id": "voddetail/12345.html", + "vod_name": "示例影片", + "vod_pic": "https://www.zxzjhd.com/cover.jpg", + "vod_remarks": "更新至10集", + } + ], + ) + + @patch.object(Spider, "_request_html") + def test_category_content_uses_built_url_and_returns_page_result(self, mock_request_html): + mock_request_html.return_value = """ + + """ + result = self.spider.categoryContent("1", "2", False, {"area": "欧美"}) + self.assertEqual(mock_request_html.call_args.args[0], "https://www.zxzjhd.com/vodshow/1-欧美-------2---.html") + self.assertEqual(result["page"], 2) + self.assertEqual(result["limit"], 24) + self.assertEqual(result["list"][0]["vod_id"], "voddetail/23456.html") + self.assertNotIn("pagecount", result) + + @patch.object(Spider, "_request_html") + def test_search_content_parses_search_cards(self, mock_request_html): + mock_request_html.return_value = """ + + """ + result = self.spider.searchContent("繁花", False, "1") + self.assertEqual( + mock_request_html.call_args.args[0], + "https://www.zxzjhd.com/vodsearch/%E7%B9%81%E8%8A%B1-------------.html", + ) + self.assertEqual(result["list"][0]["vod_name"], "搜索命中") + self.assertNotIn("pagecount", result) + if __name__ == "__main__": unittest.main() diff --git a/py/在线之家.py b/py/在线之家.py index 64a5803..533568c 100644 --- a/py/在线之家.py +++ b/py/在线之家.py @@ -1,6 +1,8 @@ # coding=utf-8 import json +import re import sys +from urllib.parse import quote from base.spider import Spider as BaseSpider @@ -153,3 +155,73 @@ class Spider(BaseSpider): f"{values.get('year', '')}" ) return self._build_url(f"/vodshow/{path}.html") + + def _clean_text(self, text): + return re.sub(r"\s+", " ", str(text or "").replace("\xa0", " ")).strip() + + def _fix_json_wrapped_html(self, html): + value = str(html or "").strip() + if value.startswith("<"): + return value + if value.startswith('"') and value.endswith('"'): + try: + parsed = json.loads(value) + if isinstance(parsed, str) and parsed.startswith("<"): + return parsed.strip() + except Exception: + return value + return value + + def _request_html(self, path_or_url, referer=None): + target = path_or_url if str(path_or_url).startswith("http") else self._build_url(path_or_url) + headers = dict(self.headers) + headers["Referer"] = referer or self.headers["Referer"] + response = self.fetch(target, headers=headers, timeout=10) + if response.status_code != 200: + return "" + return self._fix_json_wrapped_html(response.text or "") + + def _extract_vod_id(self, href): + raw = str(href or "").strip() + matched = re.search(r"/?(voddetail/\d+\.html)", raw) + return matched.group(1) if matched else "" + + def _parse_cards(self, html): + root = self.html(html) + if root is None: + return [] + items = [] + seen = set() + for node in root.xpath("//ul[contains(@class,'stui-vodlist')]//li"): + href = ((node.xpath(".//a[@href][1]/@href") or [""])[0]).strip() + vod_id = self._extract_vod_id(href) + title = ((node.xpath(".//a[@title][1]/@title") or [""])[0]).strip() + pic = ( + ((node.xpath(".//a[@data-original][1]/@data-original") or [""])[0]).strip() + or ((node.xpath(".//img[@data-original][1]/@data-original") or [""])[0]).strip() + or ((node.xpath(".//img[@src][1]/@src") or [""])[0]).strip() + ) + remarks = self._clean_text("".join(node.xpath(".//*[contains(@class,'pic-text')][1]//text()"))) + if not vod_id or not title or vod_id in seen: + continue + seen.add(vod_id) + items.append( + { + "vod_id": vod_id, + "vod_name": title, + "vod_pic": self._build_url(pic), + "vod_remarks": remarks, + } + ) + return items + + def categoryContent(self, tid, pg, filter, extend): + page = int(pg) + items = self._parse_cards(self._request_html(self._build_category_url(tid, pg, extend))) + return {"list": items, "page": page, "limit": 24, "total": page * 30 + len(items)} + + def searchContent(self, key, quick, pg="1"): + page = int(pg) + url = self._build_url("/vodsearch/{0}-------------.html".format(quote(str(key or "").strip()))) + items = self._parse_cards(self._request_html(url)) + return {"list": items, "page": page, "limit": len(items), "total": len(items)}