From 80916ae62611899ee83ec3e577e5c6b16a4679c1 Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Sun, 19 Apr 2026 10:58:39 +0800 Subject: [PATCH] feat: add gimy spider --- py/tests/test_剧迷.py | 228 ++++++++++++++++++++++ py/剧迷.py | 434 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 662 insertions(+) create mode 100644 py/tests/test_剧迷.py create mode 100644 py/剧迷.py diff --git a/py/tests/test_剧迷.py b/py/tests/test_剧迷.py new file mode 100644 index 0000000..52fe859 --- /dev/null +++ b/py/tests/test_剧迷.py @@ -0,0 +1,228 @@ +import unittest +import base64 +from importlib.machinery import SourceFileLoader +from pathlib import Path +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] +MODULE = SourceFileLoader("gimy_spider", str(ROOT / "剧迷.py")).load_module() +Spider = MODULE.Spider + + +class TestGimySpider(unittest.TestCase): + def setUp(self): + Spider._instance = None + self.spider = Spider() + self.spider.init() + + def test_home_content_exposes_expected_categories_and_filters(self): + content = self.spider.homeContent(False) + class_ids = [item["type_id"] for item in content["class"]] + self.assertEqual(class_ids, ["1", "2", "4", "29", "34", "13"]) + self.assertEqual(content["filters"]["2"][0]["key"], "by") + self.assertEqual(content["filters"]["2"][0]["init"], "time") + + def test_parse_cards_extracts_vod_id_title_cover_and_remarks(self): + html = """ +
+ +
更新至第3集
+
+ """ + items = self.spider._extract_cards(html) + self.assertEqual( + items, + [ + { + "vod_id": "123", + "vod_name": "斗罗大陆", + "vod_pic": "https://gimyai.tw/cover.jpg", + "vod_remarks": "更新至第3集", + } + ], + ) + + def test_build_category_url_uses_default_sort_for_page1(self): + self.assertEqual( + self.spider._build_category_url("2", "1", {}), + "https://gimyai.tw/genre/2.html", + ) + + def test_build_category_url_appends_page_and_sort_when_needed(self): + self.assertEqual( + self.spider._build_category_url("2", "3", {"by": "hits"}), + "https://gimyai.tw/genre/2.html?page=3&by=hits", + ) + + @patch.object(Spider, "_request_html") + def test_home_video_content_limits_to_24_items(self, mock_request_html): + cards = [] + for index in range(1, 27): + cards.append( + f'
' + ) + mock_request_html.return_value = "".join(cards) + result = self.spider.homeVideoContent() + self.assertEqual(len(result["list"]), 24) + self.assertEqual(result["list"][0]["vod_id"], "1") + + @patch.object(Spider, "_request_html") + def test_category_content_builds_page_result(self, mock_request_html): + mock_request_html.return_value = """ +
+ +
HD
+
+ """ + result = self.spider.categoryContent("2", "2", False, {"by": "hits"}) + self.assertEqual(mock_request_html.call_args.args[0], "https://gimyai.tw/genre/2.html?page=2&by=hits") + self.assertEqual(result["page"], 2) + self.assertEqual(result["pagecount"], 2) + self.assertEqual(result["list"][0]["vod_name"], "分类影片") + + def test_parse_detail_page_extracts_meta_and_playlists(self): + html = """ + +

機動戰士鋼彈

+ +
一段劇情簡介
+
+ Gimy + 雲播 +
+
+ 第1集 + 第2集 +
+
+ 第1集 +
+ """ + result = self.spider._parse_detail_page(html, "100") + vod = result["list"][0] + self.assertEqual(vod["vod_id"], "100") + self.assertEqual(vod["vod_name"], "机动战士钢弹") + self.assertEqual(vod["vod_pic"], "https://gimyai.tw/poster.jpg") + self.assertEqual(vod["vod_content"], "一段剧情简介") + self.assertEqual(vod["vod_remarks"], "更新至第2集") + self.assertEqual(vod["type_name"], "动画") + self.assertEqual(vod["vod_year"], "2026") + self.assertEqual(vod["vod_area"], "日本") + self.assertEqual(vod["vod_actor"], "演员甲,演员乙") + self.assertEqual(vod["vod_director"], "导演甲") + self.assertEqual(vod["vod_play_from"], "Gimy$$$云播") + self.assertEqual(vod["vod_play_url"], "第1集$100-1-1#第2集$100-1-2$$$第1集$100-2-1") + + @patch.object(Spider, "_request_html") + def test_detail_content_accepts_numeric_id(self, mock_request_html): + mock_request_html.return_value = """ +

详情影片

+
Gimy
+
正片
+ """ + result = self.spider.detailContent(["300"]) + self.assertEqual(mock_request_html.call_args.args[0], "https://gimyai.tw/detail/300.html") + self.assertEqual(result["list"][0]["vod_id"], "300") + + def test_normalize_search_keyword_removes_noise_and_converts_traditional(self): + self.assertEqual(self.spider._normalize_search_keyword("鬥羅大陸 線上看"), "斗罗大陆线上看") + + def test_refine_search_results_prefers_exact_match_after_simplification(self): + result = self.spider._refine_search_results( + [ + {"vod_name": "鬥羅大陸", "vod_id": "1"}, + {"vod_name": "斗罗大陆2绝世唐门", "vod_id": "2"}, + {"vod_name": "海贼王", "vod_id": "3"}, + ], + "斗罗大陆", + ) + self.assertEqual([item["vod_id"] for item in result], ["1", "2"]) + + @patch.object(Spider, "_request_html") + def test_search_content_filters_raw_results(self, mock_request_html): + mock_request_html.return_value = """ +
+
+ +
+
+ +
+
+ """ + result = self.spider.searchContent("斗罗大陆", False, "1") + self.assertEqual([item["vod_id"] for item in result["list"]], ["11"]) + self.assertEqual(result["page"], 1) + self.assertEqual(result["pagecount"], 1) + + def test_extract_player_data_reads_json_block(self): + html = '' + data = self.spider._extract_player_data(html) + self.assertEqual(data["url"], "https://video.example/direct.m3u8") + + def test_decode_play_url_supports_encrypt_0_1_2(self): + self.assertEqual( + self.spider._decode_play_url({"url": "https://video.example/raw.m3u8", "encrypt": "0"}), + "https://video.example/raw.m3u8", + ) + self.assertEqual( + self.spider._decode_play_url({"url": "https%3A%2F%2Fvideo.example%2Fescape.m3u8", "encrypt": "1"}), + "https://video.example/escape.m3u8", + ) + encoded = base64.b64encode("https://video.example/base64.m3u8".encode("utf-8")).decode("utf-8") + self.assertEqual( + self.spider._decode_play_url({"url": encoded, "encrypt": "2"}), + "https://video.example/base64.m3u8", + ) + + @patch.object(Spider, "_request_html") + def test_player_content_returns_direct_media_url(self, mock_request_html): + mock_request_html.return_value = """ + + """ + result = self.spider.playerContent("Gimy", "100-1-1", {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["url"], "https://video.example/final.m3u8") + self.assertEqual(result["header"]["Referer"], "https://gimyai.tw/play/100-1-1.html") + + @patch.object(Spider, "_request_html") + def test_player_content_uses_parse_php_when_available(self, mock_request_html): + mock_request_html.side_effect = [ + '', + '{"url":"https://video.example/parsed.m3u8"}', + ] + result = self.spider.playerContent("Gimy", "100-1-1", {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["url"], "https://video.example/parsed.m3u8") + self.assertEqual(result["header"]["Origin"], "https://play.gimyai.tw") + + @patch.object(Spider, "_request_html") + def test_player_content_falls_back_to_raw_url_when_parse_php_returns_empty(self, mock_request_html): + mock_request_html.side_effect = [ + '', + '{}', + ] + result = self.spider.playerContent("Gimy", "100-1-1", {}) + self.assertEqual(result["parse"], 1) + self.assertEqual(result["jx"], 1) + self.assertEqual(result["url"], "https://middle.example/embed?id=1") + + @patch.object(Spider, "_request_html") + def test_player_content_falls_back_to_play_page_when_player_data_missing(self, mock_request_html): + mock_request_html.return_value = "empty" + result = self.spider.playerContent("Gimy", "100-1-1", {}) + self.assertEqual(result["parse"], 1) + self.assertEqual(result["jx"], 1) + self.assertEqual(result["url"], "https://gimyai.tw/play/100-1-1.html") + + +if __name__ == "__main__": + unittest.main() diff --git a/py/剧迷.py b/py/剧迷.py new file mode 100644 index 0000000..c3f2a2a --- /dev/null +++ b/py/剧迷.py @@ -0,0 +1,434 @@ +# coding=utf-8 +import base64 +import json +import re +import sys +from urllib.parse import quote, unquote + +from base.spider import Spider as BaseSpider + +sys.path.append("..") + + +class Spider(BaseSpider): + def __init__(self): + self.name = "Gimy剧迷" + self.host = "https://gimyai.tw" + self.headers = { + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/146.0.0.0 Safari/537.36" + ), + "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", + "Referer": "https://gimyai.tw/", + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", + } + self.categories = [ + {"type_id": "1", "type_name": "电影"}, + {"type_id": "2", "type_name": "电视剧"}, + {"type_id": "4", "type_name": "动漫"}, + {"type_id": "29", "type_name": "综艺"}, + {"type_id": "34", "type_name": "短剧"}, + {"type_id": "13", "type_name": "陆剧"}, + ] + filter_value = [ + {"n": "最新", "v": "time"}, + {"n": "人气", "v": "hits"}, + {"n": "评分", "v": "score"}, + ] + self.filters = { + key: [{"key": "by", "name": "排序", "init": "time", "value": filter_value}] + for key in ["1", "2", "4", "29", "34", "13"] + } + self.filter_def = {key: {"by": "time"} for key in self.filters} + self.t2s_map = { + "鬥": "斗", + "羅": "罗", + "陸": "陆", + "劇": "剧", + "藝": "艺", + "綜": "综", + "動": "动", + "畫": "画", + "機": "机", + "戰": "战", + "鋼": "钢", + "彈": "弹", + "雲": "云", + "狀": "状", + "態": "态", + "類": "类", + "國": "国", + "導": "导", + "簡": "简", + "線": "线", + "賊": "贼", + } + + def init(self, extend=""): + return None + + def getName(self): + return self.name + + def homeContent(self, filter): + return {"class": self.categories, "filters": self.filters} + + def _build_url(self, href): + raw = str(href or "").strip() + if not raw: + return "" + if raw.startswith(("http://", "https://")): + return raw + if raw.startswith("//"): + return "https:" + raw + return self.host + "/" + raw.lstrip("/") + + def _normalize_text(self, text): + value = re.sub(r"<[^>]+>", " ", str(text or "")) + return re.sub(r"\s+", " ", value.replace(" ", " ")).strip() + + def _to_display_text(self, text): + clean = self._normalize_text(text) + for trad, simp in self.t2s_map.items(): + clean = clean.replace(trad, simp) + return clean + + def _extract_cards(self, html): + root = self.html(html) + if root is None: + return [] + + items = [] + seen = set() + for anchor in root.xpath("//a[contains(@href,'/detail/')]"): + href = (anchor.xpath("./@href") or [""])[0].strip() + matched = re.search(r"/detail/(\d+)\.html", href) + if not matched: + continue + + vod_id = matched.group(1) + if vod_id in seen: + continue + + title = ( + (anchor.xpath("./@title") or [""])[0].strip() + or (anchor.xpath(".//img/@alt") or [""])[0].strip() + or "".join(anchor.xpath(".//text()")).strip() + ) + pic = ( + (anchor.xpath("./@data-original") or [""])[0].strip() + or (anchor.xpath("./@data-src") or [""])[0].strip() + or (anchor.xpath(".//img/@src") or [""])[0].strip() + ) + parent = anchor.getparent() + remarks = "" + if parent is not None: + remarks = "".join( + parent.xpath(".//*[contains(@class,'note') or contains(@class,'text')]//text()") + ).strip() + + seen.add(vod_id) + items.append( + { + "vod_id": vod_id, + "vod_name": self._to_display_text(title), + "vod_pic": self._build_url(pic), + "vod_remarks": self._to_display_text(remarks), + } + ) + return items + + def _build_category_url(self, tid, pg, extend): + page = int(pg) + values = dict(self.filter_def.get(str(tid), {"by": "time"})) + if isinstance(extend, dict): + values.update(extend) + + by = str(values.get("by", "time")) + url = f"{self.host}/genre/{tid}.html" + if page > 1 or by != "time": + url += f"?page={page}&by={by}" + return url + + 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 response.text or "" + + def homeVideoContent(self): + html = self._request_html(self.host) + return {"list": self._extract_cards(html)[:24]} + + def categoryContent(self, tid, pg, filter, extend): + url = self._build_category_url(str(tid), str(pg), extend if isinstance(extend, dict) else {}) + items = self._extract_cards(self._request_html(url)) + page = int(pg) + pagecount = page + 1 if len(items) >= 20 else page + return { + "list": items, + "page": page, + "pagecount": pagecount, + "limit": 20, + "total": page * 20 + len(items), + } + + def _pick_info(self, html, label): + matched = re.search(rf"]*>{label}[：:](.*?)", str(html or ""), re.S) + return self._to_display_text(matched.group(1) if matched else "") + + def _parse_play_sources(self, html): + root = self.html(html) + if root is None: + return [] + + groups = [] + tabs = [] + for tab in root.xpath("//*[@id='playTab']//a[starts-with(@href,'#con_playlist_')]"): + tabs.append( + { + "id": (tab.xpath("./@href") or [""])[0].replace("#", ""), + "name": self._to_display_text("".join(tab.xpath(".//text()"))), + } + ) + + for tab in tabs: + episodes = [] + for anchor in root.xpath(f"//*[@id='{tab['id']}']//a[contains(@href,'/play/')]"): + href = (anchor.xpath("./@href") or [""])[0].strip() + matched = re.search(r"/play/(\d+-\d+-\d+)\.html", href) + if not matched: + continue + title = self._to_display_text("".join(anchor.xpath(".//text()"))) or "正片" + episodes.append(f"{title}${matched.group(1)}") + if episodes: + groups.append({"from": tab["name"], "urls": "#".join(episodes)}) + return groups + + def _parse_detail_page(self, html, vod_id): + root = self.html(html) + if root is None: + return {"list": []} + + title = "".join( + root.xpath("//h1[contains(@class,'text-overflow')][1]//text()") or root.xpath("//h1[1]//text()") + ).strip() + poster = ((root.xpath("//meta[@property='og:image']/@content") or [""])[0]).strip() + content = "".join( + root.xpath("//*[contains(@class,'details-content') or contains(@class,'detail-sketch')][1]//text()") + ).strip() + actors = [self._to_display_text(text) for text in root.xpath("//li[contains(.,'主演')]//a/text()")] + directors = [ + self._to_display_text(text) + for text in root.xpath("//li[contains(.,'導演') or contains(.,'导演')]//a/text()") + ] + play_groups = self._parse_play_sources(html) + + return { + "list": [ + { + "vod_id": str(vod_id), + "vod_name": self._to_display_text(title), + "vod_pic": self._build_url(poster), + "vod_content": self._to_display_text(content), + "vod_remarks": self._pick_info(html, "狀態") or self._pick_info(html, "状态"), + "type_name": self._pick_info(html, "類別") or self._pick_info(html, "类别"), + "vod_year": self._pick_info(html, "年代") or self._pick_info(html, "年份"), + "vod_area": self._pick_info(html, "國家/地區") or self._pick_info(html, "国家/地区"), + "vod_actor": ",".join([item for item in actors if item]), + "vod_director": ",".join([item for item in directors if item]), + "vod_play_from": "$$$".join([item["from"] for item in play_groups]), + "vod_play_url": "$$$".join([item["urls"] for item in play_groups]), + } + ] + } + + def detailContent(self, ids): + raw = ids[0] + url = raw if str(raw).startswith("http") else f"{self.host}/detail/{raw}.html" + html = self._request_html(url) + matched = re.search(r"/detail/(\d+)\.html", url) + vod_id = matched.group(1) if matched else str(raw) + return self._parse_detail_page(html, vod_id) + + def _normalize_search_keyword(self, text): + value = self._to_display_text(text).lower() + return re.sub(r"[\s\-_—–·•:：,，.。!?！？'\"“”‘’()（）\[\]【】{}]", "", value) + + def _build_search_tokens(self, keyword): + base = self._normalize_search_keyword(keyword) + if not base: + return [] + + tokens = [base] + simplified = re.sub( + r"线上看|線上看|全集|連續劇|连续剧|電視劇|电视剧|動漫|动漫|電影|电影|綜藝|综艺", + "", + base, + ).strip() + if simplified and simplified not in tokens: + tokens.append(simplified) + + stripped = re.sub(r"第\d+[集季部篇]?$", "", simplified) + stripped = re.sub(r"\d+$", "", stripped) + if stripped and stripped not in tokens: + tokens.append(stripped) + + return sorted([item for item in tokens if item], key=len, reverse=True) + + def _score_search_result(self, vod_name, keyword): + name = self._normalize_search_keyword(vod_name) + best = 0 + for token in self._build_search_tokens(keyword): + if name == token: + best = max(best, 1000 + len(token)) + elif name.startswith(token): + best = max(best, 800 + len(token)) + elif token in name: + best = max(best, 600 + len(token)) + elif len(name) >= 2 and name in token: + best = max(best, 450 + len(name)) + return best + + def _refine_search_results(self, items, keyword): + scored = [] + for item in items: + score = self._score_search_result(item.get("vod_name", ""), keyword) + if score > 0: + scored.append((score, item)) + + if scored: + scored.sort(key=lambda pair: (-pair[0], pair[1].get("vod_name", ""))) + return [item for _, item in scored] + + loose_key = self._normalize_search_keyword(keyword) + return [ + item + for item in items + if loose_key and loose_key in self._normalize_search_keyword(item.get("vod_name", "")) + ] + + def searchContent(self, key, quick, pg="1"): + url = f"{self.host}/find/-------------.html?wd={quote(str(key))}&page={pg}" + html = self._request_html(url) + raw_items = self._extract_cards(html) + items = self._refine_search_results(raw_items, key) + page = int(pg) + pagecount = page + 1 if len(raw_items) >= 20 else page + return { + "list": items, + "page": page, + "pagecount": pagecount, + "limit": len(items), + "total": len(items), + } + + def _extract_player_data(self, html): + matched = re.search(r"var\s+player_data\s*=\s*(\{[\s\S]*?\})\s*;?\s*", str(html or ""), re.S) + if not matched: + return {} + try: + return json.loads(matched.group(1)) + except Exception: + return {} + + def _decode_play_url(self, player_data): + raw = str(player_data.get("url", "")).strip() + encrypt = str(player_data.get("encrypt", "0")).strip() + if not raw: + return "" + if encrypt == "1": + return unquote(raw) + if encrypt == "2": + try: + decoded = base64.b64decode(raw).decode("utf-8") + return unquote(decoded) + except Exception: + return "" + return raw + + def _is_direct_media_url(self, url): + return bool(re.match(r"^https?://.*\.(m3u8|mp4|flv|m4s)(\?.*)?$", str(url or ""), re.I)) + + def _build_parse_context(self, raw_url, play_from, play_page_url): + base = "https://play.gimyai.tw/v/" + referer = f"https://play.gimyai.tw/v/?url={quote(raw_url)}" + if play_from in ["JD4K", "JD2K", "JDHG", "JDQM"]: + base = "https://play.gimyai.tw/d/" + referer = ( + f"https://play.gimyai.tw/d/?url={quote(raw_url)}" + f"&jctype={play_from}&next={quote(play_page_url)}" + ) + elif play_from == "NSYS": + base = "https://play.gimyai.tw/n/" + referer = ( + f"https://play.gimyai.tw/n/?url={quote(raw_url)}" + f"&jctype={play_from}&next={quote(play_page_url)}" + ) + return { + "parse_url": f"{base}parse.php?url={quote(raw_url)}&_t=1", + "referer": referer, + "origin": "https://play.gimyai.tw", + } + + def playerContent(self, flag, id, vipFlags): + play_id = str(id) + play_page_url = play_id if play_id.startswith("http") else f"{self.host}/play/{play_id}.html" + page_headers = { + "User-Agent": self.headers["User-Agent"], + "Referer": play_page_url, + "Origin": self.host, + } + + html = self._request_html(play_page_url, referer=play_page_url) + player_data = self._extract_player_data(html) + raw_url = self._decode_play_url(player_data) + + if self._is_direct_media_url(raw_url): + return { + "parse": 0, + "jx": 0, + "url": raw_url, + "header": {"User-Agent": self.headers["User-Agent"], "Referer": play_page_url}, + } + + if raw_url: + context = self._build_parse_context(raw_url, str(player_data.get("from", "")).strip(), play_page_url) + parse_text = self._request_html(context["parse_url"], referer=context["referer"]) + try: + parse_json = json.loads(parse_text or "{}") + except Exception: + parse_json = {} + + media_url = str( + parse_json.get("url") or parse_json.get("video") or parse_json.get("playurl") or "" + ).strip() + if media_url: + return { + "parse": 0, + "jx": 0, + "url": media_url, + "header": { + "User-Agent": self.headers["User-Agent"], + "Referer": context["referer"], + "Origin": context["origin"], + }, + } + + if raw_url.startswith("http"): + return { + "parse": 1, + "jx": 1, + "url": raw_url, + "header": { + "User-Agent": self.headers["User-Agent"], + "Referer": context["referer"], + }, + } + + return {"parse": 1, "jx": 1, "url": play_page_url, "header": page_headers}