From 734dfcb8135b22220c5c2b20dc06cd3d6afd9a04 Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Mon, 20 Apr 2026 14:55:59 +0800 Subject: [PATCH 1/3] feat: scaffold ouge spider --- py/tests/test_欧歌.py | 60 +++++++++++++++++++++++++++ py/欧歌.py | 94 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 154 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..8b5c674 --- /dev/null +++ b/py/tests/test_欧歌.py @@ -0,0 +1,60 @@ +import unittest +from importlib.machinery import SourceFileLoader +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +MODULE = SourceFileLoader("ouge_spider", str(ROOT / "欧歌.py")).load_module() +Spider = MODULE.Spider + + +class TestOuGeSpider(unittest.TestCase): + def setUp(self): + Spider._instance = None + self.spider = Spider() + self.spider.init() + + def test_home_content_exposes_reference_categories(self): + content = self.spider.homeContent(False) + self.assertEqual( + [(item["type_id"], item["type_name"]) for item in content["class"]], + [ + ("1", "欧歌电影"), + ("2", "欧哥剧集"), + ("3", "欧歌动漫"), + ("4", "欧歌综艺"), + ("5", "欧歌短剧"), + ("21", "欧歌综合"), + ], + ) + + def test_home_video_content_returns_empty_list(self): + self.assertEqual(self.spider.homeVideoContent(), {"list": []}) + + def test_build_url_joins_relative_paths_against_host(self): + self.assertEqual( + self.spider._build_url("/index.php/vod/detail/id/1.html"), + "https://woog.nxog.eu.org/index.php/vod/detail/id/1.html", + ) + self.assertEqual( + self.spider._build_url("https://cdn.example.com/poster.jpg"), + "https://cdn.example.com/poster.jpg", + ) + + def test_detect_pan_type_returns_expected_type_and_label(self): + self.assertEqual( + self.spider._detect_pan_type("https://pan.baidu.com/s/demo"), + ("baidu", "百度资源"), + ) + self.assertEqual( + self.spider._detect_pan_type("https://pan.quark.cn/s/demo"), + ("quark", "夸克资源"), + ) + self.assertEqual( + self.spider._detect_pan_type("https://example.com/video"), + ("", ""), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/py/欧歌.py b/py/欧歌.py new file mode 100644 index 0000000..24021af --- /dev/null +++ b/py/欧歌.py @@ -0,0 +1,94 @@ +# coding=utf-8 +import re +import sys +from urllib.parse import urljoin + +from base.spider import Spider as BaseSpider + +sys.path.append("..") + + +class Spider(BaseSpider): + def __init__(self): + self.name = "欧歌" + self.host = "https://woog.nxog.eu.org" + self.headers = { + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/136.0.0.0 Safari/537.36" + ), + "Referer": self.host + "/", + } + self.classes = [ + {"type_id": "1", "type_name": "欧歌电影"}, + {"type_id": "2", "type_name": "欧哥剧集"}, + {"type_id": "3", "type_name": "欧歌动漫"}, + {"type_id": "4", "type_name": "欧歌综艺"}, + {"type_id": "5", "type_name": "欧歌短剧"}, + {"type_id": "21", "type_name": "欧歌综合"}, + ] + self.pan_patterns = [ + ("baidu", "百度资源", r"pan\.baidu\.com|yun\.baidu\.com"), + ("a139", "139资源", r"yun\.139\.com"), + ("a189", "天翼资源", r"cloud\.189\.cn"), + ("a123", "123资源", r"123pan\.com|123684\.com|123865\.com|123912\.com"), + ("a115", "115资源", r"115\.com"), + ("quark", "夸克资源", r"pan\.quark\.cn"), + ("xunlei", "迅雷资源", r"pan\.xunlei\.com"), + ("aliyun", "阿里资源", r"aliyundrive\.com|alipan\.com"), + ("uc", "UC资源", r"drive\.uc\.cn"), + ] + self.pan_priority = { + "baidu": 1, + "a139": 2, + "a189": 3, + "a123": 4, + "a115": 5, + "quark": 6, + "xunlei": 7, + "aliyun": 8, + "uc": 9, + } + + 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": []} + + def _stringify(self, value): + return "" if value is None else str(value) + + def _clean_text(self, text): + return re.sub(r"\s+", " ", self._stringify(text).replace("\xa0", " ")).strip() + + def _build_url(self, path): + raw = self._stringify(path).strip() + if not raw: + return "" + return urljoin(self.host + "/", raw) + + def _request_html(self, path_or_url, headers=None, referer=None): + target = path_or_url if self._stringify(path_or_url).startswith("http") else self._build_url(path_or_url) + merged_headers = dict(self.headers) + if headers: + merged_headers.update(headers) + merged_headers["Referer"] = referer or self.headers["Referer"] + response = self.fetch(target, headers=merged_headers, timeout=10) + if response.status_code != 200: + return "" + return response.text or "" + + def _detect_pan_type(self, url): + raw = self._stringify(url).strip() + for pan_type, title, pattern in self.pan_patterns: + if re.search(pattern, raw, re.I): + return pan_type, title + return "", "" From 7b1449c7ab81e77e02528b1288b2412c30cdd63f Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Mon, 20 Apr 2026 14:57:18 +0800 Subject: [PATCH 2/3] feat: add ouge list and search parsing --- py/tests/test_欧歌.py | 80 +++++++++++++++++++++++++++++++++++++++ py/欧歌.py | 87 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 167 insertions(+) diff --git a/py/tests/test_欧歌.py b/py/tests/test_欧歌.py index 8b5c674..85b940a 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] @@ -55,6 +56,85 @@ class TestOuGeSpider(unittest.TestCase): ("", ""), ) + def test_parse_cards_extracts_short_path_id_and_cover(self): + html = """ +
+
+
+ + 示例影片 +
+
HD
+
+
+ """ + self.assertEqual( + self.spider._parse_cards(html), + [ + { + "vod_id": "/index.php/vod/detail/id/123.html", + "vod_name": "示例影片", + "vod_pic": "https://woog.nxog.eu.org/poster.jpg", + "vod_remarks": "HD", + } + ], + ) + + @patch.object(Spider, "_request_html") + def test_category_content_builds_reference_url_and_returns_page_payload(self, mock_request_html): + mock_request_html.return_value = """ +
+
+
+ + 分类影片 +
+
更新至10集
+
+
+ """ + result = self.spider.categoryContent("2", "3", False, {}) + self.assertEqual( + mock_request_html.call_args.args[0], + "https://woog.nxog.eu.org/index.php/vod/show/id/2/page/3.html", + ) + self.assertEqual(result["page"], 3) + self.assertEqual(result["limit"], 1) + self.assertEqual(result["list"][0]["vod_name"], "分类影片") + self.assertNotIn("pagecount", result) + + @patch.object(Spider, "_request_html") + def test_search_content_builds_reference_url_and_parses_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://woog.nxog.eu.org/index.php/vod/search/page/2/wd/%E7%B9%81%E8%8A%B1.html", + ) + self.assertEqual( + result["list"][0], + { + "vod_id": "/index.php/vod/detail/id/789.html", + "vod_name": "搜索影片", + "vod_pic": "https://woog.nxog.eu.org/search.jpg", + "vod_remarks": "抢先版", + }, + ) + self.assertNotIn("pagecount", result) + + @patch.object(Spider, "_request_html") + def test_search_content_short_circuits_blank_keyword(self, mock_request_html): + result = self.spider.searchContent("", False, "1") + self.assertEqual(result, {"page": 1, "total": 0, "list": []}) + mock_request_html.assert_not_called() + if __name__ == "__main__": unittest.main() diff --git a/py/欧歌.py b/py/欧歌.py index 24021af..965f9b6 100644 --- a/py/欧歌.py +++ b/py/欧歌.py @@ -1,6 +1,7 @@ # coding=utf-8 import re import sys +from urllib.parse import quote from urllib.parse import urljoin from base.spider import Spider as BaseSpider @@ -92,3 +93,89 @@ class Spider(BaseSpider): if re.search(pattern, raw, re.I): return pan_type, title return "", "" + + def _fix_img_url(self, img_url): + raw = self._stringify(img_url).strip() + if not raw: + return "" + if raw.startswith(("http://", "https://")): + return raw + return self._build_url(raw) + + def _build_category_url(self, tid, pg): + return self._build_url(f"/index.php/vod/show/id/{self._stringify(tid).strip()}/page/{int(pg)}.html") + + def _build_search_url(self, key, pg): + return self._build_url(f"/index.php/vod/search/page/{int(pg)}/wd/{quote(self._stringify(key).strip())}.html") + + def _parse_cards(self, html): + root = self.html(html) + if root is None: + return [] + items = [] + seen = set() + for card in root.xpath("//*[@id='main']//*[contains(@class,'module-item')]"): + href = ((card.xpath(".//*[contains(@class,'module-item-pic')]//a[@href][1]/@href") or [""])[0]).strip() + title = ((card.xpath(".//*[contains(@class,'module-item-pic')]//img[@alt][1]/@alt") or [""])[0]).strip() + pic = ( + ((card.xpath(".//*[contains(@class,'module-item-pic')]//img[@data-src][1]/@data-src") or [""])[0]).strip() + or ((card.xpath(".//*[contains(@class,'module-item-pic')]//img[@src][1]/@src") or [""])[0]).strip() + ) + remarks = self._clean_text("".join(card.xpath(".//*[contains(@class,'module-item-text')][1]//text()"))) + if not href or not title or href in seen: + continue + seen.add(href) + items.append( + { + "vod_id": href, + "vod_name": title, + "vod_pic": self._fix_img_url(pic), + "vod_remarks": remarks, + } + ) + return items + + def _parse_search_cards(self, html): + root = self.html(html) + if root is None: + return [] + items = [] + seen = set() + for card in root.xpath("//*[contains(@class,'module-search-item')]"): + href = ((card.xpath(".//*[contains(@class,'video-serial')][1]/@href") or [""])[0]).strip() + title = ( + ((card.xpath(".//*[contains(@class,'video-serial')][1]/@title") or [""])[0]).strip() + or ((card.xpath(".//*[contains(@class,'module-item-pic')]//img[@alt][1]/@alt") or [""])[0]).strip() + ) + pic = ( + ((card.xpath(".//*[contains(@class,'module-item-pic')]//img[@data-src][1]/@data-src") or [""])[0]).strip() + or ((card.xpath(".//*[contains(@class,'module-item-pic')]//img[@src][1]/@src") or [""])[0]).strip() + ) + remarks = self._clean_text("".join(card.xpath(".//*[contains(@class,'video-serial')][1]//text()"))) + if not remarks: + remarks = self._clean_text("".join(card.xpath(".//*[contains(@class,'module-item-text')][1]//text()"))) + if not href or not title or href in seen: + continue + seen.add(href) + items.append( + { + "vod_id": href, + "vod_name": title, + "vod_pic": self._fix_img_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))) + return {"list": items, "page": page, "limit": len(items), "total": page * 20 + len(items)} + + def searchContent(self, key, quick, pg="1"): + page = int(pg) + keyword = self._stringify(key).strip() + if not keyword: + return {"page": page, "total": 0, "list": []} + items = self._parse_search_cards(self._request_html(self._build_search_url(keyword, pg))) + return {"page": page, "total": len(items), "list": items} From 59e15248105066345aaae7e6b47416981a81402a Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Mon, 20 Apr 2026 14:59:12 +0800 Subject: [PATCH 3/3] feat: add ouge detail and player support --- py/tests/test_欧歌.py | 74 ++++++++++++++++++++++++++++++++++ py/欧歌.py | 94 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 168 insertions(+) diff --git a/py/tests/test_欧歌.py b/py/tests/test_欧歌.py index 85b940a..07625c3 100644 --- a/py/tests/test_欧歌.py +++ b/py/tests/test_欧歌.py @@ -135,6 +135,80 @@ class TestOuGeSpider(unittest.TestCase): self.assertEqual(result, {"page": 1, "total": 0, "list": []}) mock_request_html.assert_not_called() + def test_parse_detail_page_extracts_metadata_and_deduplicated_pan_groups(self): + html = """ +
欧歌示例
+
+ +
+
2025
+
导演
+
导演甲导演乙
+
主演
+
演员甲演员乙
+
剧情
+

这是一段剧情简介。

+

https://pan.quark.cn/s/q-demo

+

百度合集 https://pan.baidu.com/s/b-demo

+

https://pan.quark.cn/s/q-demo

+ """ + vod = self.spider._parse_detail_page(html, "/index.php/vod/detail/id/999.html") + self.assertEqual(vod["vod_id"], "/index.php/vod/detail/id/999.html") + self.assertEqual(vod["vod_name"], "欧歌示例") + self.assertEqual(vod["vod_pic"], "https://woog.nxog.eu.org/poster-detail.jpg") + self.assertEqual(vod["vod_year"], "2025") + self.assertEqual(vod["vod_director"], "导演甲,导演乙") + self.assertEqual(vod["vod_actor"], "演员甲,演员乙") + self.assertEqual(vod["vod_content"], "这是一段剧情简介。") + self.assertEqual(vod["vod_play_from"], "baidu$$$quark") + self.assertEqual( + vod["vod_play_url"], + "百度资源$https://pan.baidu.com/s/b-demo$$$夸克资源$https://pan.quark.cn/s/q-demo", + ) + + @patch.object(Spider, "_request_html") + def test_detail_content_reads_detail_page_and_returns_single_vod(self, mock_request_html): + mock_request_html.return_value = """ +
详情标题
+

https://pan.quark.cn/s/detail-demo

+ """ + result = self.spider.detailContent(["/index.php/vod/detail/id/1000.html"]) + self.assertEqual( + mock_request_html.call_args.args[0], + "https://woog.nxog.eu.org/index.php/vod/detail/id/1000.html", + ) + self.assertEqual(result["list"][0]["vod_id"], "/index.php/vod/detail/id/1000.html") + self.assertEqual(result["list"][0]["vod_name"], "详情标题") + self.assertEqual(result["list"][0]["vod_play_from"], "quark") + self.assertEqual(result["list"][0]["vod_play_url"], "夸克资源$https://pan.quark.cn/s/detail-demo") + + def test_parse_detail_page_returns_empty_shell_for_blank_html(self): + self.assertEqual( + self.spider._parse_detail_page("", "/index.php/vod/detail/id/404.html"), + { + "vod_id": "/index.php/vod/detail/id/404.html", + "vod_name": "", + "vod_pic": "", + "vod_year": "", + "vod_director": "", + "vod_actor": "", + "vod_content": "", + "vod_play_from": "", + "vod_play_url": "", + }, + ) + + def test_player_content_transparently_returns_supported_pan_link(self): + result = self.spider.playerContent("quark", "https://pan.quark.cn/s/demo", {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["jx"], 0) + self.assertEqual(result["url"], "https://pan.quark.cn/s/demo") + self.assertEqual(result["header"], {}) + + def test_player_content_returns_empty_url_for_unknown_link(self): + result = self.spider.playerContent("unknown", "/index.php/vod/play/id/1.html", {}) + self.assertEqual(result, {"parse": 0, "jx": 0, "playUrl": "", "url": "", "header": {}}) + if __name__ == "__main__": unittest.main() diff --git a/py/欧歌.py b/py/欧歌.py index 965f9b6..7b34734 100644 --- a/py/欧歌.py +++ b/py/欧歌.py @@ -179,3 +179,97 @@ class Spider(BaseSpider): return {"page": page, "total": 0, "list": []} items = self._parse_search_cards(self._request_html(self._build_search_url(keyword, pg))) return {"page": page, "total": len(items), "list": items} + + def _blank_detail(self, vod_id): + return { + "vod_id": vod_id, + "vod_name": "", + "vod_pic": "", + "vod_year": "", + "vod_director": "", + "vod_actor": "", + "vod_content": "", + "vod_play_from": "", + "vod_play_url": "", + } + + def _extract_share_url(self, text): + matched = re.search(r"https?://[^\s'\"<>]+", self._stringify(text), re.I) + return matched.group(0).strip() if matched else "" + + def _join_people(self, node): + values = [self._clean_text(text) for text in node.xpath(".//a//text()")] + if not any(values): + values = [self._clean_text(text) for text in node.xpath(".//text()")] + values = [value for value in values if value] + return ",".join(values) + + def _extract_pan_groups(self, root): + groups = [] + seen = set() + for node in root.xpath("//*[contains(@class,'module-row-info')]//p"): + text = self._clean_text("".join(node.xpath(".//text()"))) + share_url = self._extract_share_url(text) + pan_type, title = self._detect_pan_type(share_url) + if not share_url or not pan_type or share_url in seen: + continue + seen.add(share_url) + groups.append((pan_type, f"{title}${share_url}")) + groups.sort(key=lambda item: self.pan_priority.get(item[0], 999)) + return groups + + def _parse_detail_page(self, html, vod_id): + if not self._stringify(html).strip(): + return self._blank_detail(vod_id) + root = self.html(html) + if root is None: + return self._blank_detail(vod_id) + + vod = self._blank_detail(vod_id) + vod["vod_name"] = self._clean_text("".join(root.xpath("(//*[contains(@class,'page-title')])[1]//text()"))) + pic = ( + ((root.xpath("//*[contains(@class,'mobile-play')]//*[contains(@class,'lazyload')][1]/@data-src") or [""])[0]).strip() + or ((root.xpath("//*[contains(@class,'mobile-play')]//*[contains(@class,'lazyload')][1]/@src") or [""])[0]).strip() + ) + vod["vod_pic"] = self._fix_img_url(pic) + vod["vod_year"] = self._clean_text( + ((root.xpath("(//*[contains(@class,'module-item-caption')]//span[1]/text())") or [""])[0]) + ) + + for label_node in root.xpath("//*[contains(@class,'video-info-itemtitle')]"): + label = self._clean_text("".join(label_node.xpath(".//text()"))) + siblings = label_node.xpath("./following-sibling::*[1]") + if not siblings: + continue + block = siblings[0] + if "导演" in label: + vod["vod_director"] = self._join_people(block) + elif "主演" in label: + vod["vod_actor"] = self._join_people(block) + elif "剧情" in label: + texts = block.xpath(".//p//text()") or block.xpath(".//text()") + vod["vod_content"] = self._clean_text("".join(texts)) + + pan_groups = self._extract_pan_groups(root) + vod["vod_play_from"] = "$$$".join(item[0] for item in pan_groups) + vod["vod_play_url"] = "$$$".join(item[1] for item in pan_groups) + return vod + + def detailContent(self, ids): + result = {"list": []} + for raw_id in ids: + vod_id = self._stringify(raw_id).strip() + if not vod_id: + continue + html = self._request_html(self._build_url(vod_id)) + result["list"].append(self._parse_detail_page(html, vod_id)) + return result + + def _is_supported_pan_link(self, url): + return bool(self._detect_pan_type(url)[0]) + + def playerContent(self, flag, id, vipFlags): + target = self._stringify(id).strip() + if target and self._is_supported_pan_link(target): + return {"parse": 0, "jx": 0, "playUrl": "", "url": target, "header": {}} + return {"parse": 0, "jx": 0, "playUrl": "", "url": "", "header": {}}