feat: scaffold ouge spider
This commit is contained in:
@@ -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()
|
||||
@@ -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 "", ""
|
||||
Reference in New Issue
Block a user