From 0f119a89cd757e2be8f20590dbf379ef1ea1a115 Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Fri, 24 Apr 2026 20:14:29 +0800 Subject: [PATCH] feat: add FKTV list and search parsing --- py/tests/test_凡客TV.py | 53 ++++++++++++++++++++++ py/凡客TV.py | 99 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 152 insertions(+) diff --git a/py/tests/test_凡客TV.py b/py/tests/test_凡客TV.py index 975b50e..04d8e73 100644 --- a/py/tests/test_凡客TV.py +++ b/py/tests/test_凡客TV.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] @@ -49,6 +50,58 @@ class TestFKTVSpider(unittest.TestCase): self.assertEqual(payload["episode_name"], "第1集") self.assertEqual(payload["page"], "https://fktv.me/movie/detail/9001") + @patch.object(Spider, "_request_html") + def test_category_content_builds_url_and_parses_cards(self, mock_request_html): + mock_request_html.return_value = """ +
+
+ 示例电影 + + 电影 + 更新中 +
+
+ """ + result = self.spider.categoryContent("1", "2", False, {}) + self.assertEqual( + mock_request_html.call_args.args[0], + "https://fktv.me/channel?page=2&cat_id=1&page_size=32&order=new", + ) + self.assertEqual( + result["list"], + [ + { + "vod_id": "abc123", + "vod_name": "示例电影", + "vod_pic": "https://fktv.me/poster.jpg", + "vod_remarks": "电影 | 更新中", + "type_name": "电影", + } + ], + ) + self.assertNotIn("pagecount", result) + + @patch.object(Spider, "_request_html") + def test_search_content_builds_url_and_handles_blank_keyword(self, mock_request_html): + blank = self.spider.searchContent("", False, "1") + self.assertEqual(blank, {"page": 1, "limit": 0, "total": 0, "list": []}) + mock_request_html.assert_not_called() + + mock_request_html.return_value = """ +
+ 搜索影片 + + 剧集 +
+ """ + result = self.spider.searchContent("繁花", False, "3") + self.assertEqual( + mock_request_html.call_args.args[0], + "https://fktv.me/search?keyword=%E7%B9%81%E8%8A%B1", + ) + self.assertEqual(result["page"], 3) + self.assertEqual(result["list"][0]["vod_id"], "xyz789") + if __name__ == "__main__": unittest.main() diff --git a/py/凡客TV.py b/py/凡客TV.py index 886cf2b..49ae220 100644 --- a/py/凡客TV.py +++ b/py/凡客TV.py @@ -1,6 +1,8 @@ # coding=utf-8 import json +import re import sys +from urllib.parse import quote, urljoin from base.spider import Spider as BaseSpider @@ -37,6 +39,103 @@ class Spider(BaseSpider): def homeContent(self, filter): return {"class": self.classes} + def homeVideoContent(self): + return {"list": []} + + def _clean_text(self, value): + return re.sub(r"\s+", " ", str(value or "").replace("\xa0", " ")).strip() + + def _abs_url(self, value): + raw = str(value or "").strip() + if not raw: + return "" + if raw.startswith(("http://", "https://")): + return raw + if raw.startswith("//"): + return "https:" + raw + return urljoin(self.host + "/", raw) + + def _page_headers(self, referer=""): + headers = {"User-Agent": self.user_agent, "Referer": referer or self.host + "/"} + if self.cookie: + headers["Cookie"] = self.cookie + return headers + + def _request_html(self, url, headers=None): + response = self.fetch(url, headers=headers or self._page_headers(), timeout=10, verify=False) + if response.status_code != 200: + return "" + return str(response.text or "") + + def _extract_cards(self, html): + root = self.html(html or "") + if root is None: + return [] + items = [] + seen = set() + nodes = root.xpath("//*[contains(@class,'meta-wrap')]/.. | //*[contains(@class,'hover-wrap')]") + for node in nodes: + href = self._clean_text( + "".join( + node.xpath( + ".//a[contains(@class,'normal-title') or contains(@class,'hover-title')][1]/@href" + ) + ) + ) + matched = re.search(r"/movie/detail/([0-9A-Za-z]+)", href) + if not matched: + continue + vod_id = matched.group(1) + if vod_id in seen: + continue + seen.add(vod_id) + title = self._clean_text( + "".join( + node.xpath( + ".//a[contains(@class,'normal-title') or contains(@class,'hover-title')][1]/@title" + ) + ) + ) or self._clean_text( + "".join( + node.xpath( + ".//a[contains(@class,'normal-title') or contains(@class,'hover-title')][1]//text()" + ) + ) + ) + pic = self._clean_text("".join(node.xpath(".//*[contains(@class,'lazy-load')][1]/@data-src"))) + tags = [ + self._clean_text("".join(tag.xpath(".//text()"))) + for tag in node.xpath(".//*[contains(@class,'tag')]") + ] + tags = [tag for tag in tags if tag] + if not title: + continue + items.append( + { + "vod_id": vod_id, + "vod_name": title, + "vod_pic": self._abs_url(pic), + "vod_remarks": " | ".join(tags), + "type_name": tags[0] if tags else "", + } + ) + return items + + def categoryContent(self, tid, pg, filter, extend): + page = max(1, int(pg)) + url = self.host + f"/channel?page={page}&cat_id={tid}&page_size=32&order=new" + items = self._extract_cards(self._request_html(url)) + return {"page": page, "limit": len(items), "total": len(items), "list": items} + + def searchContent(self, key, quick, pg="1"): + page = max(1, int(pg)) + keyword = self._clean_text(key) + if not keyword: + return {"page": page, "limit": 0, "total": 0, "list": []} + url = self.host + "/search?keyword=" + quote(keyword) + items = self._extract_cards(self._request_html(url)) + return {"page": page, "limit": len(items), "total": len(items), "list": items} + def _encode_play_id(self, payload): return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))