diff --git a/py/tests/test_两个BT.py b/py/tests/test_两个BT.py new file mode 100644 index 0000000..24c6fae --- /dev/null +++ b/py/tests/test_两个BT.py @@ -0,0 +1,139 @@ +import unittest +from importlib.machinery import SourceFileLoader +from pathlib import Path +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] +MODULE = SourceFileLoader("lianggebt_spider", str(ROOT / "两个BT.py")).load_module() +Spider = MODULE.Spider + + +class TestLiangGeBTSpider(unittest.TestCase): + def setUp(self): + Spider._instance = None + self.spider = Spider() + self.spider.init() + + def test_home_content_exposes_expected_categories(self): + content = self.spider.homeContent(False) + self.assertEqual( + [item["type_id"] for item in content["class"]], + [ + "zgjun", + "meiju", + "jpsrtv", + "movie_bt_tags/xiju", + "movie_bt_tags/aiqing", + "movie_bt_tags/adt", + "movie_bt_tags/at", + "movie_bt_tags/donghua", + "movie_bt_tags/qihuan", + "movie_bt_tags/xuanni", + "movie_bt_tags/kehuan", + "movie_bt_tags/juqing", + "movie_bt_tags/kongbu", + "gf", + ], + ) + self.assertNotIn("filters", content) + + def test_extract_cards_supports_short_ids_and_relative_images(self): + html = """ + + """ + cards = self.spider._extract_cards(html) + self.assertEqual( + cards, + [ + { + "vod_id": "123", + "vod_name": "示例影片", + "vod_pic": "https://www.bttwoo.com/cover.jpg", + "vod_remarks": "更新至10集", + } + ], + ) + + @patch.object(Spider, "_request_html") + def test_home_video_content_requests_home_page(self, mock_request_html): + mock_request_html.return_value = """ +
  • + + + +

    首页影片

    +
  • + """ + result = self.spider.homeVideoContent() + self.assertEqual(mock_request_html.call_args.args[0], "https://www.bttwoo.com") + self.assertEqual(result["list"][0]["vod_id"], "200") + + @patch.object(Spider, "_request_html") + def test_category_content_builds_expected_urls(self, mock_request_html): + mock_request_html.return_value = """ +
  • + + + +

    分类影片

    + HD +
  • + """ + page_one = self.spider.categoryContent("meiju", "1", False, {}) + page_three = self.spider.categoryContent("movie_bt_tags/xiju", "3", False, {}) + self.assertEqual(mock_request_html.call_args_list[0].args[0], "https://www.bttwoo.com/meiju") + self.assertEqual( + mock_request_html.call_args_list[1].args[0], + "https://www.bttwoo.com/movie_bt_tags/xiju?paged=3", + ) + self.assertEqual(page_one["page"], 1) + self.assertEqual(page_one["list"][0]["vod_name"], "分类影片") + self.assertNotIn("pagecount", page_one) + self.assertEqual(page_three["page"], 3) + + @patch.object(Spider, "_request_html") + def test_search_content_builds_query_and_filters_irrelevant_results(self, mock_request_html): + mock_request_html.return_value = """ +
  • + + + +

    繁花

    +
  • +
  • + + + +

    无关结果

    +
  • + """ + result = self.spider.searchContent("繁花", False, "2") + self.assertEqual( + mock_request_html.call_args.args[0], + "https://www.bttwoo.com/xssssearch?q=%E7%B9%81%E8%8A%B1&p=2", + ) + self.assertEqual(result["page"], 2) + self.assertEqual( + result["list"], + [ + { + "vod_id": "401", + "vod_name": "繁花", + "vod_pic": "https://www.bttwoo.com/match.jpg", + "vod_remarks": "", + } + ], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/py/两个BT.py b/py/两个BT.py new file mode 100644 index 0000000..4c2adce --- /dev/null +++ b/py/两个BT.py @@ -0,0 +1,198 @@ +# coding=utf-8 +import re +import sys +from urllib.parse import quote, urljoin + +from lxml import html as lxml_html + +from base.spider import Spider as BaseSpider + +sys.path.append("..") + + +class Spider(BaseSpider): + def __init__(self): + self.name = "两个BT" + self.host = "https://www.bttwoo.com" + 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" + ), + "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", + "Referer": self.host + "/", + } + self.classes = [ + {"type_id": "zgjun", "type_name": "国产剧"}, + {"type_id": "meiju", "type_name": "美剧"}, + {"type_id": "jpsrtv", "type_name": "日韩剧"}, + {"type_id": "movie_bt_tags/xiju", "type_name": "喜剧"}, + {"type_id": "movie_bt_tags/aiqing", "type_name": "爱情"}, + {"type_id": "movie_bt_tags/adt", "type_name": "冒险"}, + {"type_id": "movie_bt_tags/at", "type_name": "动作"}, + {"type_id": "movie_bt_tags/donghua", "type_name": "动画"}, + {"type_id": "movie_bt_tags/qihuan", "type_name": "奇幻"}, + {"type_id": "movie_bt_tags/xuanni", "type_name": "悬疑"}, + {"type_id": "movie_bt_tags/kehuan", "type_name": "科幻"}, + {"type_id": "movie_bt_tags/juqing", "type_name": "剧情"}, + {"type_id": "movie_bt_tags/kongbu", "type_name": "恐怖"}, + {"type_id": "gf", "type_name": "高分电影"}, + ] + + def init(self, extend=""): + return None + + def getName(self): + return self.name + + def homeContent(self, filter): + return {"class": self.classes} + + def homeVideoContent(self): + return {"list": self._extract_cards(self._request_html(self.host))} + + def categoryContent(self, tid, pg, filter, extend): + page = max(1, self._to_int(pg, 1)) + items = self._extract_cards(self._request_html(self._build_category_url(tid, page))) + return {"page": page, "limit": len(items), "total": len(items), "list": items} + + def searchContent(self, key, quick, pg="1"): + page = max(1, self._to_int(pg, 1)) + keyword = self._clean_text(key) + if not keyword: + return {"page": page, "limit": 0, "total": 0, "list": []} + url = self.host + f"/xssssearch?q={quote(keyword)}" + if page > 1: + url += f"&p={page}" + items = self._extract_cards(self._request_html(url), keyword=keyword) + return {"page": page, "limit": len(items), "total": len(items), "list": items} + + def detailContent(self, ids): + return {"list": []} + + def playerContent(self, flag, id, vipFlags): + return {"parse": 1, "jx": 1, "playUrl": "", "url": "", "header": {}} + + def _request_html(self, url, referer=None): + headers = dict(self.headers) + headers["Referer"] = referer or self.headers["Referer"] + try: + response = self.fetch(url, headers=headers, timeout=10, verify=False) + except Exception: + return "" + if response.status_code != 200: + return "" + return str(response.text or "") + + def _build_category_url(self, tid, page): + path = str(tid or "").strip().lstrip("/") + url = self.host + "/" + path + if page > 1: + url += f"?paged={page}" + return url + + def _extract_cards(self, html, keyword=None): + root = self._parse_html(html) + if root is None: + return [] + results = [] + seen = set() + for node in root.xpath("//li[.//a[contains(@href,'/movie/')]]"): + href = self._first_attr(node, ".//a[contains(@href,'/movie/')][1]", "href") + vod_id = self._extract_vod_id(href) + title = ( + self._first_text(node, ".//h3//a[1]") + or self._first_text(node, ".//h3[1]") + or self._clean_text(self._first_attr(node, ".//a[@title][1]", "title")) + or self._first_text(node, ".//*[contains(@class,'title')][1]") + or self._first_text(node, ".//*[contains(@class,'name')][1]") + ) + if not vod_id or not title or vod_id in seen: + continue + if keyword and not self._is_relevant_search_result(title, keyword): + continue + pic = ( + self._first_attr(node, ".//img[@data-original][1]", "data-original") + or self._first_attr(node, ".//img[@data-src][1]", "data-src") + or self._first_attr(node, ".//img[@src][1]", "src") + ) + remarks = ( + self._first_text(node, ".//*[contains(@class,'rating')][1]") + or self._first_text(node, ".//*[contains(@class,'status')][1]") + ) + seen.add(vod_id) + results.append( + { + "vod_id": vod_id, + "vod_name": title, + "vod_pic": self._abs_url(pic), + "vod_remarks": remarks, + } + ) + return results + + def _is_relevant_search_result(self, title, keyword): + title_text = self._clean_text(title).lower() + keyword_text = self._clean_text(keyword).lower() + if not title_text or not keyword_text: + return False + if keyword_text in title_text: + return True + if len(keyword_text) <= 2: + return False + key_chars = set(keyword_text.replace(" ", "")) + title_chars = set(title_text.replace(" ", "")) + if not key_chars: + return False + return len(key_chars & title_chars) / float(len(key_chars)) >= 0.6 + + def _extract_vod_id(self, href): + matched = re.search(r"/movie/(\d+)\.html", str(href or "").strip()) + return matched.group(1) if matched else "" + + 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.lstrip("/")) + + def _parse_html(self, text): + body = str(text or "").strip() + if not body: + return None + try: + return lxml_html.fromstring(body) + except Exception: + return None + + def _clean_text(self, text): + return re.sub(r"\s+", " ", str(text or "").replace("\xa0", " ")).strip() + + def _first_attr(self, node, xpath, attr): + if node is None: + return "" + for item in node.xpath(xpath): + value = str(item.get(attr) or "").strip() + if value: + return value + return "" + + def _first_text(self, node, xpath): + if node is None: + return "" + for item in node.xpath(xpath): + text = self._clean_text(item.text_content()) + if text: + return text + return "" + + def _to_int(self, value, default=0): + try: + return int(str(value)) + except Exception: + return default