diff --git a/app/chain/download.py b/app/chain/download.py index fc73de39..bee6c060 100644 --- a/app/chain/download.py +++ b/app/chain/download.py @@ -7,7 +7,7 @@ import shutil import time from pathlib import Path from typing import List, Optional, Tuple, Set, Dict, Union -from urllib.parse import parse_qs, urlparse +from urllib.parse import parse_qs, urljoin, urlparse from app import schemas from app.chain import ChainBase @@ -60,6 +60,28 @@ class DownloadChain(ChainBase): ".rar": "rar", } + @staticmethod + def _normalize_indirect_download_url(url: str, base_url: Optional[str] = None) -> str: + """ + 将两段式下载结果约束到索引器配置的可信 API 地址。 + + :param url: 换票接口返回的临时下载地址 + :param base_url: 索引器配置的可信 API Base URL + :return: 使用可信 API 来源的临时下载地址 + """ + if not url or not base_url: + return url + base_parts = urlparse(base_url) + if not base_parts.scheme or not base_parts.netloc: + return url + url_parts = urlparse(url) + if not url_parts.netloc: + return urljoin(f"{base_url.rstrip('/')}/", url) + return url_parts._replace( + scheme=base_parts.scheme, + netloc=base_parts.netloc, + ).geturl() + @staticmethod def _media_identity_keys(media: Optional[MediaInfo]) -> Set[str]: """返回媒体的统一身份键及全部兼容 ID，用于临时缺失集映射匹配。""" @@ -693,7 +715,11 @@ class DownloadChain(ChainBase): data = data.get(key) if not data: return None - logger.info(f"获取到下载地址：{data}") + data = self._normalize_indirect_download_url( + url=data, + base_url=req_params.get('result_base_url'), + ) + logger.info("已获取到站点临时下载地址") return data return None @@ -704,7 +730,8 @@ class DownloadChain(ChainBase): return torrent.enclosure, "", [] # Cookie site_cookie = torrent.site_cookie - if torrent.enclosure.startswith("["): + indirect_download = torrent.enclosure.startswith("[") + if indirect_download: # 需要解码获取下载地址 torrent_url = __get_redict_url(url=torrent.enclosure, ua=torrent.site_ua, @@ -714,21 +741,22 @@ class DownloadChain(ChainBase): else: torrent_url = torrent.enclosure if not torrent_url: - logger.error(f"{torrent.title} 无法获取下载地址：{torrent.enclosure}！") + logger.error(f"{torrent.title} 无法获取下载地址！") return None, "", [] # 下载种子文件 _, content, download_folder, files, error_msg = TorrentHelper().download_torrent( url=torrent_url, cookie=site_cookie, ua=torrent.site_ua or settings.USER_AGENT, - proxy=torrent.site_proxy) + proxy=torrent.site_proxy, + cache_invalid=not indirect_download) if isinstance(content, str): # 磁力链 return content, "", [] if not content: - logger.error(f"下载种子文件失败：{torrent.title} - {torrent_url}") + logger.error(f"下载种子文件失败：{torrent.title}") self.post_message(Notification( channel=channel, source=source if channel else None, diff --git a/app/chain/site.py b/app/chain/site.py index e2833aa8..032e22b1 100644 --- a/app/chain/site.py +++ b/app/chain/site.py @@ -46,6 +46,7 @@ class SiteChain(ChainBase): _text_page_size = 10 def __init__(self): + """初始化站点管理处理链及特殊站点测试器""" super().__init__() # 特殊站点登录验证 @@ -59,6 +60,7 @@ class SiteChain(ChainBase): "yemapt.org": self.__yema_test, "hddolby.com": self.__hddolby_test, "rousi.pro": self.__rousi_test, + "sunnypt.top": self.__sunnypt_test, } def refresh_userdata(self, site: dict = None) -> Optional[SiteUserData]: @@ -233,6 +235,41 @@ class SiteChain(ChainBase): else: return False, f"错误：{res.status_code} {res.reason}" + @staticmethod + def __sunnypt_test(site: Site) -> Tuple[bool, str]: + """ + 通过 profile 接口测试 SunnyPT API Key 和下载权限 + + :param site: SunnyPT 站点配置 + :return: 是否可用及状态信息 + """ + indexer = SitesHelper().get_indexer(site.domain) or {} + api_url = str( + indexer.get("api_url") or "https://api.sunnypt.top/api/v1/mp" + ).rstrip("/") + res = RequestUtils( + headers={ + "Accept": "application/json", + "User-Agent": site.ua or settings.USER_AGENT, + "X-API-Key": site.apikey, + }, + proxies=settings.PROXY if site.proxy else None, + timeout=site.timeout or 15, + ).get_res(url=f"{api_url}/profile") + if res is None: + return False, "无法连接 SunnyPT API 服务" + if res.status_code != 200: + return False, f"错误：{res.status_code} {res.reason}" + try: + payload = res.json() or {} + except (TypeError, ValueError): + return False, "SunnyPT API 响应不是有效 JSON" + if str(payload.get("code")) != "0" or not isinstance(payload.get("data"), dict): + return False, payload.get("msg") or "API Key 已过期或无效" + if payload["data"].get("download_allowed") is False: + return False, "当前账号没有下载权限" + return True, "连接成功" + @staticmethod def __yema_test(site: Site) -> Tuple[bool, str]: """ diff --git a/app/helper/torrent.py b/app/helper/torrent.py index 7a61e84f..ad993d1a 100644 --- a/app/helper/torrent.py +++ b/app/helper/torrent.py @@ -60,16 +60,24 @@ class TorrentHelper: """ def __init__(self): + """初始化种子失败地址缓存""" self._invalid_torrents = TTLCache(region="invalid_torrents", maxsize=128, ttl=3600 * 24) def download_torrent(self, url: str, cookie: Optional[str] = None, ua: Optional[str] = None, referer: Optional[str] = None, - proxy: Optional[bool] = False) \ + proxy: Optional[bool] = False, + cache_invalid: bool = True) \ -> Tuple[Optional[Path], Optional[Union[str, bytes]], Optional[str], Optional[list], Optional[str]]: """ 把种子下载到本地 + :param url: 种子下载地址 + :param cookie: 站点 Cookie + :param ua: 请求 User-Agent + :param referer: 请求来源地址 + :param proxy: 是否使用系统代理 + :param cache_invalid: 是否缓存失败地址；短时凭证地址必须关闭 :return: 种子缓存相对路径【用于索引缓存】, 种子内容、种子主目录、种子文件清单、错误信息 """ if url.startswith("magnet:"): @@ -142,16 +150,16 @@ class TorrentHelper: # 检查是不是种子文件，如果不是抛出异常 Torrent.from_string(req.content) # 跳过成功 - logger.info(f"触发了站点首次种子下载，已自动跳过：{url}") + logger.info("触发了站点首次种子下载，已自动跳过") skip_flag = True elif req is not None: logger.warn(f"触发了站点首次种子下载，且无法自动跳过，" f"返回码：{req.status_code}，错误原因：{req.reason}") else: - logger.warn(f"触发了站点首次种子下载，且无法自动跳过：{url}") + logger.warn("触发了站点首次种子下载，且无法自动跳过") break except Exception as err: - logger.warn(f"触发了站点首次种子下载，尝试自动跳过时出现错误：{str(err)}，链接：{url}") + logger.warn(f"触发了站点首次种子下载，尝试自动跳过时出现错误：{str(err)}") if not skip_flag: return cache_path, None, "", [], "种子数据有误，请确认链接是否正确，如为PT站点则需手工在站点下载一次种子" # 种子内容 @@ -177,7 +185,8 @@ class TorrentHelper: return cache_path, None, "", [], "触发站点流控，请稍后重试" else: # 把错误的种子记下来，避免重复使用 - self.add_invalid(url) + if cache_invalid: + self.add_invalid(url) return cache_path, None, "", [], f"下载种子出错，状态码：{req.status_code}" def get_torrent_info(self, torrent_path: Path) -> Tuple[str, List[str]]: diff --git a/app/modules/indexer/__init__.py b/app/modules/indexer/__init__.py index fbf7ed9c..5c062286 100644 --- a/app/modules/indexer/__init__.py +++ b/app/modules/indexer/__init__.py @@ -13,6 +13,7 @@ from app.modules.indexer.spider.haidan import HaiDanSpider from app.modules.indexer.spider.hddolby import HddolbySpider from app.modules.indexer.spider.mtorrent import MTorrentSpider from app.modules.indexer.spider.rousi import RousiSpider +from app.modules.indexer.spider.sunnypt import SunnyPTSpider from app.modules.indexer.spider.tnode import TNodeSpider from app.modules.indexer.spider.torrentleech import TorrentLeech from app.modules.indexer.spider.yema import YemaSpider @@ -28,6 +29,7 @@ SPIDER_PARSER_CLASSES = { "Haidan": HaiDanSpider, "HDDolby": HddolbySpider, "RousiPro": RousiSpider, + "SunnyPT": SunnyPTSpider, } @@ -39,6 +41,7 @@ class IndexerModule(_ModuleBase): _site_schemas = [] def init_module(self) -> None: + """加载站点用户数据解析器""" # 加载模块 self._site_schemas = ModuleHelper.load( 'app.modules.indexer.parser', @@ -47,6 +50,7 @@ class IndexerModule(_ModuleBase): @staticmethod def get_name() -> str: + """获取模块名称""" return "站点索引" @staticmethod @@ -71,6 +75,7 @@ class IndexerModule(_ModuleBase): return 0 def stop(self): + """停止索引模块""" pass def test(self) -> Tuple[bool, str]: @@ -83,6 +88,7 @@ class IndexerModule(_ModuleBase): return True, "" def init_setting(self) -> Tuple[str, Union[str, bool]]: + """索引模块无需独立开关配置""" pass @staticmethod @@ -240,6 +246,13 @@ class IndexerModule(_ModuleBase): mtype=mtype, page=page ) + elif site.get('parser') == "SunnyPT": + error_flag, result = SunnyPTSpider(site).search( + keyword=search_word, + mtype=mtype, + cat=cat, + page=page + ) elif site.get('parser') == "Yema": error_flag, result = YemaSpider(site).search( keyword=search_word, @@ -376,6 +389,13 @@ class IndexerModule(_ModuleBase): mtype=mtype, page=page ) + elif site.get('parser') == "SunnyPT": + error_flag, result = await SunnyPTSpider(site).async_search( + keyword=search_word, + mtype=mtype, + cat=cat, + page=page + ) elif site.get('parser') == "Yema": error_flag, result = await YemaSpider(site).async_search( keyword=search_word, @@ -573,7 +593,8 @@ class IndexerModule(_ModuleBase): apikey=site.get("apikey"), token=site.get("token"), ua=site.get("ua"), - proxy=site.get("proxy")) + proxy=site.get("proxy"), + api_url=site.get("api_url")) return None site_obj = __get_site_obj() diff --git a/app/modules/indexer/parser/__init__.py b/app/modules/indexer/parser/__init__.py index b1805d76..bc9dc3d0 100644 --- a/app/modules/indexer/parser/__init__.py +++ b/app/modules/indexer/parser/__init__.py @@ -18,6 +18,8 @@ from app.utils.string import StringUtils # 站点框架 class SiteSchema(Enum): + """站点用户数据解析框架类型""" + DiscuzX = "DiscuzX" Gazelle = "Gazelle" Ipt = "IPTorrents" @@ -37,9 +39,12 @@ class SiteSchema(Enum): Zhixing = "Zhixing" Bitpt = "Bitpt" RousiPro = "RousiPro" + SunnyPT = "SunnyPT" class SiteParserBase(metaclass=ABCMeta): + """站点用户数据解析器基类""" + # 站点模版 schema = None # 请求模式 cookie/apikey @@ -53,7 +58,22 @@ class SiteParserBase(metaclass=ABCMeta): session: Session = None, ua: Optional[str] = None, emulate: bool = False, - proxy: bool = None): + proxy: bool = None, + api_url: Optional[str] = None): + """ + 初始化站点用户数据解析器 + + :param site_name: 站点名称 + :param url: 站点前端地址 + :param site_cookie: 站点 Cookie + :param apikey: 站点 API Key + :param token: 站点 Token + :param session: 可复用的 HTTP 会话 + :param ua: 请求 User-Agent + :param emulate: 是否使用浏览器仿真 + :param proxy: 是否使用系统代理 + :param api_url: 站点独立 API Base URL + """ super().__init__() # 站点信息 @@ -65,6 +85,7 @@ class SiteParserBase(metaclass=ABCMeta): self._site_domain = __split_url.netloc self._base_url = f"{__split_url.scheme}://{__split_url.netloc}" self._site_cookie = site_cookie + self._api_url = api_url self._session = session if session else None self._ua = ua self._emulate = emulate diff --git a/app/modules/indexer/parser/sunnypt.py b/app/modules/indexer/parser/sunnypt.py new file mode 100644 index 00000000..35fc3c2b --- /dev/null +++ b/app/modules/indexer/parser/sunnypt.py @@ -0,0 +1,168 @@ +# -*- coding: utf-8 -*- +import json +from typing import Optional, Tuple +from urllib.parse import urlencode, urljoin + +from app.log import logger +from app.modules.indexer.parser import SiteParserBase, SiteSchema +from app.utils.string import StringUtils + + +class SunnyPTSiteUserInfo(SiteParserBase): + """ + SunnyPT MoviePilot API 用户数据解析器 + """ + + schema = SiteSchema.SunnyPT + request_mode = "apikey" + + def _parse_site_page(self, html_text: str) -> None: + """ + 配置 SunnyPT 用户数据接口地址和认证请求头 + + :param html_text: API Key 模式下的空首页数据 + """ + self._base_url = f"{str(self._api_url or 'https://api.sunnypt.top/api/v1/mp').rstrip('/')}/" + self._user_basic_page = "profile" + self._user_detail_page = None + self._user_traffic_page = None + self._torrent_seeding_page = None + self._user_mail_unread_page = "messages" + self._sys_mail_unread_page = None + self._addition_headers = { + "Accept": "application/json", + "User-Agent": self._ua, + "X-API-Key": self.apikey, + } + + def _load_api_payload(self, html_text: str, operation: str) -> Optional[dict]: + """ + 解析并校验 SunnyPT 通用 JSON 响应 + + :param html_text: API 响应文本 + :param operation: 错误信息中的操作名称 + :return: 成功时返回响应 data，失败时返回 None + """ + if not html_text: + self.err_msg = f"{operation}失败，未收到 API 响应" + return None + try: + payload = json.loads(html_text) + except (TypeError, json.JSONDecodeError) as err: + self.err_msg = f"{operation}失败，API 响应不是有效 JSON" + logger.warning(f"{self._site_name} {self.err_msg}：{str(err)}") + return None + if not isinstance(payload, dict): + self.err_msg = f"{operation}失败，API 响应结构无效" + logger.warning(f"{self._site_name} {self.err_msg}") + return None + if str(payload.get("code")) != "0": + self.err_msg = payload.get("msg") or f"{operation}失败" + logger.warning(f"{self._site_name} {operation}失败：{self.err_msg}") + return None + data = payload.get("data") + return data if isinstance(data, dict) else None + + def _parse_user_base_info(self, html_text: str) -> None: + """ + 解析 SunnyPT 用户资料、流量和做种统计 + + :param html_text: profile 接口响应文本 + """ + user_info = self._load_api_payload(html_text, "获取用户信息") + if not user_info: + return + self.userid = user_info.get("id") + self.username = user_info.get("username") + self.user_level = user_info.get("level") or str(user_info.get("class") or "") + self.join_at = StringUtils.unify_datetime_str(user_info.get("registered_at")) + self.upload = int(user_info.get("uploaded") or 0) + self.download = int(user_info.get("downloaded") or 0) + self.ratio = float(user_info.get("ratio") or 0) + self.bonus = float(user_info.get("bonus") or 0) + self.seeding = int(user_info.get("seeding_count") or 0) + self.seeding_size = int(user_info.get("seeding_size") or 0) + self.leeching = int(user_info.get("leeching_count") or 0) + self.leeching_size = int(user_info.get("leeching_size") or 0) + self.message_unread = int(user_info.get("unread_messages") or 0) + + def _pase_unread_msgs(self) -> None: + """ + 分页读取 SunnyPT 未读消息；解析阶段不标记已读，避免投递失败后丢失通知 + """ + page = 1 + while True: + query = urlencode({ + "unread_only": "true", + "page": page, + "page_size": 100, + }) + html_text = self._get_page_content( + url=urljoin(self._base_url, f"messages?{query}") + ) + has_more = self._parse_message_unread_links(html_text, []) + if not has_more: + return + page += 1 + + def _parse_message_unread_links(self, html_text: str, msg_links: list) -> Optional[str]: + """ + 解析 SunnyPT 未读消息列表并直接保存消息正文 + + :param html_text: messages 接口响应文本 + :param msg_links: 兼容解析器基类签名的消息链接容器 + :return: 存在下一页时返回占位字符串，否则返回 None + """ + messages_data = self._load_api_payload(html_text, "获取未读消息") + if not messages_data: + return None + self.message_unread = int(messages_data.get("unread_count") or 0) + for message in messages_data.get("items") or []: + if not isinstance(message, dict) or not message.get("unread"): + continue + title = message.get("title") + content = message.get("content") + created_at = StringUtils.unify_datetime_str(message.get("created_at")) + if title and content and created_at: + self.message_unread_contents.append((title, created_at, content)) + return "next" if messages_data.get("has_more") else None + + def _parse_user_traffic_info(self, html_text: str) -> None: + """ + 跳过独立流量页面，profile 接口已经返回完整统计 + + :param html_text: 未使用的页面文本 + """ + + def _parse_user_detail_info(self, html_text: str) -> None: + """ + 跳过独立用户详情页面，profile 接口已经返回完整资料 + + :param html_text: 未使用的页面文本 + """ + + def _parse_user_torrent_seeding_info( + self, + html_text: str, + multi_page: bool = False, + ) -> Optional[str]: + """ + 跳过独立做种列表，profile 接口已经返回做种数量和体积 + + :param html_text: 未使用的页面文本 + :param multi_page: 是否为后续分页 + :return: 始终返回 None + """ + return None + + def _parse_message_content( + self, + html_text: str, + ) -> Tuple[Optional[str], Optional[str], Optional[str]]: + """ + 跳过消息详情请求，messages 接口已经返回完整正文 + + :param html_text: 未使用的页面文本 + :return: 三个空值 + """ + return None, None, None diff --git a/app/modules/indexer/spider/sunnypt.py b/app/modules/indexer/spider/sunnypt.py new file mode 100644 index 00000000..b5d23154 --- /dev/null +++ b/app/modules/indexer/spider/sunnypt.py @@ -0,0 +1,385 @@ +import base64 +import json +import time +from typing import List, Optional, Tuple + +from app.core.config import settings +from app.log import logger +from app.schemas import MediaType +from app.utils.http import AsyncRequestUtils, RequestUtils +from app.utils.string import StringUtils + + +class SunnyPTSpider: + """ + SunnyPT MoviePilot API 索引器 + """ + + _size = 100 + _category_cache_ttl = 3600 + _category_cache = {} + + def __init__(self, indexer: dict): + """ + 初始化 SunnyPT API 索引器 + + :param indexer: 合并站点认证信息后的索引配置 + """ + indexer = indexer or {} + self._indexer_id = indexer.get("id") + self._name = indexer.get("name") or "SunnyPT" + self._site_url = indexer.get("domain") or "https://sunnypt.top/" + self._api_url = str( + indexer.get("api_url") or "https://api.sunnypt.top/api/v1/mp" + ).rstrip("/") + self._proxy = settings.PROXY if indexer.get("proxy") else None + self._use_proxy = bool(indexer.get("proxy")) + self._user_agent = indexer.get("ua") or settings.USER_AGENT + self._api_key = indexer.get("apikey") + self._timeout = indexer.get("timeout") or 15 + self._configured_categories = self._parse_configured_categories( + indexer.get("category") or {} + ) + + @classmethod + def get_search_page_size(cls, keyword: Optional[str] = None) -> Optional[int]: + """ + 获取搜索接口单页容量 + + :param keyword: 搜索关键字，SunnyPT 不按关键字改变分页容量 + :return: 搜索接口单页容量 + """ + return cls._size + + @staticmethod + def _parse_configured_categories(category_config: dict) -> dict: + """ + 从站点索引配置提取电影和电视剧分类 ID，作为分类接口不可用时的兜底 + + :param category_config: Build 站点配置中的分类段 + :return: 按 API media_type 索引的分类 ID + """ + category_map = {"movie": [], "tv": []} + for media_type in category_map: + for item in category_config.get(media_type) or []: + category_id = item.get("id") if isinstance(item, dict) else item + if category_id is not None: + category_map[media_type].append(str(category_id)) + return category_map + + @staticmethod + def _parse_category_items(items: list) -> dict: + """ + 将分类接口结果转换为按媒体类型索引的分类 ID + + :param items: SunnyPT 分类接口 data 数组 + :return: 按 API media_type 索引的分类 ID + """ + category_map = {"movie": [], "tv": []} + for item in items or []: + if not isinstance(item, dict) or item.get("id") is None: + continue + category_id = str(item["id"]) + for media_type in item.get("media_types") or []: + if media_type in category_map and category_id not in category_map[media_type]: + category_map[media_type].append(category_id) + return category_map + + @classmethod + def _get_cached_categories(cls, api_url: str) -> Optional[dict]: + """ + 获取未过期的站点分类缓存 + + :param api_url: SunnyPT API Base URL + :return: 分类映射，缓存不存在或过期时返回 None + """ + cached = cls._category_cache.get(api_url) + if not cached: + return None + cached_at, category_map = cached + if time.monotonic() - cached_at >= cls._category_cache_ttl: + cls._category_cache.pop(api_url, None) + return None + return category_map + + @classmethod + def _set_cached_categories(cls, api_url: str, category_map: dict) -> None: + """ + 缓存站点分类映射 + + :param api_url: SunnyPT API Base URL + :param category_map: 按媒体类型索引的分类 ID + """ + cls._category_cache[api_url] = (time.monotonic(), category_map) + + def _request_headers(self) -> dict: + """ + 构造 SunnyPT API 请求头 + + :return: 不包含 Cookie 和 Authorization 的 API Key 请求头 + """ + return { + "Accept": "application/json", + "User-Agent": self._user_agent, + "X-API-Key": self._api_key, + } + + def _response_data(self, response, operation: str): + """ + 校验 SunnyPT 通用响应并返回 data 字段 + + :param response: RequestUtils 返回的响应对象 + :param operation: 日志中使用的操作名称 + :return: 接口成功时返回 data，失败时返回 None + """ + if response is None: + logger.warning(f"{self._name} {operation}失败，无法连接 API 服务") + return None + if response.status_code != 200: + logger.warning(f"{self._name} {operation}失败，HTTP 错误码：{response.status_code}") + return None + try: + payload = response.json() or {} + except (TypeError, ValueError) as err: + logger.warning(f"{self._name} {operation}响应不是有效 JSON：{str(err)}") + return None + if not isinstance(payload, dict): + logger.warning(f"{self._name} {operation}响应结构无效") + return None + if str(payload.get("code")) != "0": + logger.warning(f"{self._name} {operation}失败：{payload.get('msg') or '未知错误'}") + return None + return payload.get("data") + + def _load_categories(self) -> dict: + """ + 同步读取并缓存 SunnyPT 分类映射，接口失败时使用 Build 配置兜底 + + :return: 按媒体类型索引的分类 ID + """ + cached = self._get_cached_categories(self._api_url) + if cached is not None: + return cached + response = RequestUtils( + headers=self._request_headers(), + proxies=self._proxy, + timeout=self._timeout, + ).get_res(url=f"{self._api_url}/categories") + items = self._response_data(response, "获取分类") + category_map = self._parse_category_items(items) if isinstance(items, list) else {} + if not category_map or not any(category_map.values()): + return self._configured_categories + self._set_cached_categories(self._api_url, category_map) + return category_map + + async def _async_load_categories(self) -> dict: + """ + 异步读取并缓存 SunnyPT 分类映射，接口失败时使用 Build 配置兜底 + + :return: 按媒体类型索引的分类 ID + """ + cached = self._get_cached_categories(self._api_url) + if cached is not None: + return cached + response = await AsyncRequestUtils( + headers=self._request_headers(), + proxies=self._proxy, + timeout=self._timeout, + ).get_res(url=f"{self._api_url}/categories") + items = self._response_data(response, "获取分类") + category_map = self._parse_category_items(items) if isinstance(items, list) else {} + if not category_map or not any(category_map.values()): + return self._configured_categories + self._set_cached_categories(self._api_url, category_map) + return category_map + + @staticmethod + def _media_type_value(media_type: MediaType) -> Optional[str]: + """ + 将 MoviePilot 媒体类型转换为 SunnyPT API 枚举 + + :param media_type: MoviePilot 媒体类型 + :return: movie、tv 或 None + """ + if media_type == MediaType.MOVIE: + return "movie" + if media_type == MediaType.TV: + return "tv" + return None + + def _build_params( + self, + keyword: Optional[str], + media_type: MediaType, + category: Optional[str], + page: Optional[int], + category_map: dict, + ) -> dict: + """ + 构造 SunnyPT 种子搜索参数 + + :param keyword: 搜索关键字或完整 IMDb ID + :param media_type: MoviePilot 媒体类型 + :param category: 用户显式选择的分类 ID + :param page: MoviePilot 从 0 开始的页码 + :param category_map: SunnyPT 分类接口返回的映射 + :return: SunnyPT API 查询参数 + """ + params = { + "page": int(page or 0) + 1, + "page_size": self._size, + "sort": "created_at", + "order": "desc", + } + if keyword: + params["keyword"] = keyword + api_media_type = self._media_type_value(media_type) + if api_media_type: + params["media_type"] = api_media_type + categories = category + if not categories and api_media_type: + categories = ",".join(category_map.get(api_media_type) or []) + if categories: + params["categories"] = categories + return params + + def _parse_result(self, results: List[dict]) -> List[dict]: + """ + 将 SunnyPT API 种子数据转换为 MoviePilot 标准字段 + + :param results: SunnyPT 种子接口 data.items 数组 + :return: MoviePilot 标准种子字典列表 + """ + torrents = [] + for result in results or []: + if not isinstance(result, dict): + continue + media_type = result.get("media_type") + if media_type == "movie": + category = MediaType.MOVIE.value + elif media_type == "tv": + category = MediaType.TV.value + else: + category = MediaType.UNKNOWN.value + + promotion = result.get("promotion") or {} + promotion_active = bool(promotion.get("is_active")) + download_factor = float(promotion.get("down_multiplier", 1.0)) \ + if promotion_active else 1.0 + upload_factor = float(promotion.get("up_multiplier", 1.0)) \ + if promotion_active else 1.0 + freedate = StringUtils.unify_datetime_str(promotion.get("until")) \ + if promotion_active and promotion.get("until") else None + torrent_id = result.get("id") + torrents.append({ + "title": result.get("title"), + "description": result.get("subtitle"), + "enclosure": self._build_download_url(torrent_id), + "pubdate": StringUtils.unify_datetime_str(result.get("created_at")), + "size": int(result.get("size") or 0), + "seeders": int(result.get("seeders") or 0), + "peers": int(result.get("leechers") or 0), + "grabs": int(result.get("completed") or 0), + "downloadvolumefactor": download_factor, + "uploadvolumefactor": upload_factor, + "freedate": freedate, + "page_url": result.get("details_url"), + "imdbid": result.get("imdb_id"), + "labels": result.get("tags") or [], + "hit_and_run": bool(result.get("hit_and_run")), + "category": category, + }) + return torrents + + def _process_search_response(self, response) -> Tuple[bool, List[dict]]: + """ + 处理 SunnyPT 种子搜索响应 + + :param response: RequestUtils 返回的响应对象 + :return: 是否失败及标准种子列表 + """ + data = self._response_data(response, "搜索") + if not isinstance(data, dict): + return True, [] + return False, self._parse_result(data.get("items") or []) + + def search( + self, + keyword: Optional[str], + mtype: MediaType = None, + cat: Optional[str] = None, + page: Optional[int] = 0, + ) -> Tuple[bool, List[dict]]: + """ + 同步搜索 SunnyPT 种子 + + :param keyword: 搜索关键字或完整 IMDb ID + :param mtype: MoviePilot 媒体类型 + :param cat: 用户显式选择的分类 ID + :param page: MoviePilot 从 0 开始的页码 + :return: 是否失败及标准种子列表 + """ + if not self._api_key: + logger.warning(f"{self._name} 未配置 API Key") + return True, [] + category_map = self._load_categories() if mtype and not cat else self._configured_categories + params = self._build_params(keyword, mtype, cat, page, category_map) + response = RequestUtils( + headers=self._request_headers(), + proxies=self._proxy, + timeout=self._timeout, + ).get_res(url=f"{self._api_url}/torrents", params=params) + return self._process_search_response(response) + + async def async_search( + self, + keyword: Optional[str], + mtype: MediaType = None, + cat: Optional[str] = None, + page: Optional[int] = 0, + ) -> Tuple[bool, List[dict]]: + """ + 异步搜索 SunnyPT 种子 + + :param keyword: 搜索关键字或完整 IMDb ID + :param mtype: MoviePilot 媒体类型 + :param cat: 用户显式选择的分类 ID + :param page: MoviePilot 从 0 开始的页码 + :return: 是否失败及标准种子列表 + """ + if not self._api_key: + logger.warning(f"{self._name} 未配置 API Key") + return True, [] + category_map = await self._async_load_categories() \ + if mtype and not cat else self._configured_categories + params = self._build_params(keyword, mtype, cat, page, category_map) + response = await AsyncRequestUtils( + headers=self._request_headers(), + proxies=self._proxy, + timeout=self._timeout, + ).get_res(url=f"{self._api_url}/torrents", params=params) + return self._process_search_response(response) + + def _build_download_url(self, torrent_id: int) -> str: + """ + 构造先换取短时下载地址再下载种子文件的两段式链接 + + :param torrent_id: SunnyPT 种子 ID + :return: Base64 请求配置与 download-token 接口 URL + """ + request_config = { + "method": "post", + "cookie": False, + "header": { + "X-API-Key": self._api_key, + "Accept": "application/json", + }, + "proxy": self._use_proxy, + "result": "data.download_url", + "result_base_url": self._api_url, + } + encoded_config = base64.b64encode( + json.dumps(request_config).encode("utf-8") + ).decode("ascii") + token_url = f"{self._api_url}/torrents/{torrent_id}/download-token" + return f"[{encoded_config}]{token_url}" diff --git a/tests/test_sunnypt_indexer.py b/tests/test_sunnypt_indexer.py new file mode 100644 index 00000000..64c7a504 --- /dev/null +++ b/tests/test_sunnypt_indexer.py @@ -0,0 +1,438 @@ +import asyncio +import base64 +import json +from types import SimpleNamespace + +import pytest + +from app.chain.download import DownloadChain +from app.chain.site import SiteChain +from app.core.config import settings +from app.core.context import TorrentInfo +from app.modules.indexer import IndexerModule +from app.modules.indexer.parser.sunnypt import SunnyPTSiteUserInfo +from app.modules.indexer.spider.sunnypt import SunnyPTSpider +from app.schemas import MediaType + + +class _FakeResponse: + """构造站点 API 测试使用的最小响应对象。""" + + def __init__(self, payload: dict, status_code: int = 200): + """保存响应数据和状态码。""" + self._payload = payload + self.status_code = status_code + self.reason = "OK" + + def __bool__(self) -> bool: + """按 HTTP 成功状态模拟 requests.Response 的布尔值。""" + return self.status_code < 400 + + def json(self) -> dict: + """返回预设 JSON 数据。""" + return self._payload + + +@pytest.fixture(autouse=True) +def clear_sunnypt_category_cache(): + """在用例前后清理 SunnyPT 分类缓存，避免进程级状态互相污染。""" + SunnyPTSpider._category_cache.clear() + yield + SunnyPTSpider._category_cache.clear() + + +def _build_indexer() -> dict: + """构造 SunnyPT API Spider 所需的最小站点配置。""" + return { + "id": "sunnypt", + "name": "Sunny", + "domain": "https://sunnypt.top/", + "api_url": "https://api.sunnypt.top/api/v1/mp", + "apikey": "sunny-secret", + "ua": "MoviePilot-Test", + "proxy": False, + "category": { + "movie": [{"id": 401}, {"id": 404}, {"id": 405}], + "tv": [{"id": 402}, {"id": 403}, {"id": 404}, {"id": 405}], + }, + } + + +def _category_response() -> _FakeResponse: + """构造 SunnyPT 分类接口响应。""" + return _FakeResponse({ + "code": 0, + "msg": "ok", + "data": [ + {"id": 401, "name": "电影", "media_types": ["movie"]}, + {"id": 402, "name": "电视剧", "media_types": ["tv"]}, + {"id": 404, "name": "纪录片", "media_types": ["movie", "tv"]}, + ], + }) + + +def _torrent_response() -> _FakeResponse: + """构造 SunnyPT 种子搜索接口响应。""" + return _FakeResponse({ + "code": 0, + "msg": "ok", + "data": { + "items": [{ + "id": 123, + "title": "Movie.2026.2160p.WEB-DL.H.265-GROUP", + "subtitle": "电影中文副标题", + "media_type": "movie", + "size": 21474836480, + "created_at": "2026-07-21T14:00:00+08:00", + "seeders": 15, + "leechers": 2, + "completed": 30, + "imdb_id": "tt1234567", + "tags": ["中字", "HDR"], + "hit_and_run": True, + "promotion": { + "is_active": True, + "up_multiplier": 2.0, + "down_multiplier": 0.0, + "until": "2026-07-23T14:00:00+08:00", + }, + "details_url": "https://sunnypt.top/torrent/123", + }], + }, + }) + + +def test_sunnypt_search_maps_api_fields_and_caches_categories(monkeypatch): + """SunnyPT 搜索应映射标准字段、编码下载请求并复用分类缓存。""" + calls = [] + + def fake_get_res(request, url: str, params: dict = None, **_kwargs): + """按请求路径回放分类和种子响应。""" + calls.append((url, params, request._headers)) + if url.endswith("/categories"): + return _category_response() + return _torrent_response() + + monkeypatch.setattr( + "app.modules.indexer.spider.sunnypt.RequestUtils.get_res", + fake_get_res, + ) + spider = SunnyPTSpider(_build_indexer()) + + error, torrents = spider.search( + keyword="tt1234567", + mtype=MediaType.MOVIE, + page=0, + ) + second_error, _ = SunnyPTSpider(_build_indexer()).search( + keyword="Movie", + mtype=MediaType.MOVIE, + page=1, + ) + + assert not error + assert not second_error + assert [call[0] for call in calls].count( + "https://api.sunnypt.top/api/v1/mp/categories" + ) == 1 + search_url, search_params, search_headers = calls[1] + assert search_url == "https://api.sunnypt.top/api/v1/mp/torrents" + assert search_params == { + "page": 1, + "page_size": 100, + "sort": "created_at", + "order": "desc", + "keyword": "tt1234567", + "media_type": "movie", + "categories": "401,404", + } + assert search_headers["X-API-Key"] == "sunny-secret" + assert torrents == [{ + "title": "Movie.2026.2160p.WEB-DL.H.265-GROUP", + "description": "电影中文副标题", + "enclosure": torrents[0]["enclosure"], + "pubdate": "2026-07-21 14:00:00", + "size": 21474836480, + "seeders": 15, + "peers": 2, + "grabs": 30, + "downloadvolumefactor": 0.0, + "uploadvolumefactor": 2.0, + "freedate": "2026-07-23 14:00:00", + "page_url": "https://sunnypt.top/torrent/123", + "imdbid": "tt1234567", + "labels": ["中字", "HDR"], + "hit_and_run": True, + "category": MediaType.MOVIE.value, + }] + + encoded_config, token_url = torrents[0]["enclosure"].split("]", 1) + request_config = json.loads(base64.b64decode(encoded_config[1:]).decode("utf-8")) + assert token_url == "https://api.sunnypt.top/api/v1/mp/torrents/123/download-token" + assert request_config == { + "method": "post", + "cookie": False, + "header": { + "X-API-Key": "sunny-secret", + "Accept": "application/json", + }, + "proxy": False, + "result": "data.download_url", + "result_base_url": "https://api.sunnypt.top/api/v1/mp", + } + + +def test_sunnypt_async_search_uses_api_contract(monkeypatch): + """SunnyPT 异步搜索应使用同一套 GET 参数和响应映射。""" + calls = [] + + async def fake_get_res(request, url: str, params: dict = None, **_kwargs): + """异步回放分类和种子响应。""" + calls.append((url, params, request._headers)) + if url.endswith("/categories"): + return _category_response() + return _torrent_response() + + monkeypatch.setattr( + "app.modules.indexer.spider.sunnypt.AsyncRequestUtils.get_res", + fake_get_res, + ) + + error, torrents = asyncio.run( + SunnyPTSpider(_build_indexer()).async_search( + keyword="Movie", + mtype=MediaType.TV, + cat="402,404", + page=2, + ) + ) + + assert not error + assert len(torrents) == 1 + assert len(calls) == 1 + assert calls[0][1]["page"] == 3 + assert calls[0][1]["media_type"] == "tv" + assert calls[0][1]["categories"] == "402,404" + assert calls[0][2]["X-API-Key"] == "sunny-secret" + + +def test_sunnypt_user_parser_reads_profile_and_messages_without_marking_read(monkeypatch): + """SunnyPT 用户解析器应读取统计和未读消息，但不能在解析阶段标记已读。""" + requested_urls = [] + profile = { + "code": 0, + "msg": "ok", + "data": { + "id": 1001, + "username": "sunny", + "level": "Power User", + "registered_at": "2025-01-01T12:00:00+08:00", + "uploaded": 1099511627776, + "downloaded": 536870912000, + "ratio": 2.048, + "bonus": 12345.6, + "seeding_count": 30, + "seeding_size": 2147483648000, + "leeching_count": 1, + "leeching_size": 10737418240, + "unread_messages": 1, + }, + } + messages = { + "code": 0, + "msg": "ok", + "data": { + "items": [{ + "id": 9001, + "title": "种子审核通过", + "content": "你发布的种子已审核通过。", + "created_at": "2026-07-21T16:30:00+08:00", + "unread": True, + }], + "unread_count": 1, + "has_more": False, + }, + } + + def fake_get_page_content(_self, url: str, **_kwargs): + """按请求地址回放用户信息和消息列表。""" + requested_urls.append(url) + return json.dumps(messages if "/messages" in url else profile) + + monkeypatch.setattr(SunnyPTSiteUserInfo, "_get_page_content", fake_get_page_content) + monkeypatch.setattr(settings, "SITE_MESSAGE", True) + parser = SunnyPTSiteUserInfo( + site_name="Sunny", + url="https://sunnypt.top/", + site_cookie="", + apikey="sunny-secret", + token=None, + api_url="https://api.sunnypt.top/api/v1/mp", + ) + + parser.parse() + + assert parser.userid == 1001 + assert parser.username == "sunny" + assert parser.user_level == "Power User" + assert parser.join_at == "2025-01-01 12:00:00" + assert parser.upload == 1099511627776 + assert parser.download == 536870912000 + assert parser.ratio == 2.048 + assert parser.bonus == 12345.6 + assert parser.seeding == 30 + assert parser.seeding_size == 2147483648000 + assert parser.leeching == 1 + assert parser.leeching_size == 10737418240 + assert parser.message_unread == 1 + assert parser.message_unread_contents == [ + ("种子审核通过", "2026-07-21 16:30:00", "你发布的种子已审核通过。") + ] + assert not any(url.endswith("/read") or url.endswith("/read-all") for url in requested_urls) + + +def test_indexer_module_dispatches_sunnypt_search(monkeypatch): + """IndexerModule 应把 SunnyPT 同步搜索参数交给专用 API Spider。""" + captured = {} + + def fake_search(_self, keyword, mtype, cat, page): + """记录 IndexerModule 传给 SunnyPT Spider 的搜索参数。""" + captured.update({ + "keyword": keyword, + "mtype": mtype, + "cat": cat, + "page": page, + }) + return False, [{"title": "Movie.2026", "enclosure": "https://example.com/1.torrent"}] + + monkeypatch.setattr(SunnyPTSpider, "search", fake_search) + monkeypatch.setattr( + IndexerModule, + "_IndexerModule__search_check", + staticmethod(lambda _site, _keyword=None: True), + ) + monkeypatch.setattr( + IndexerModule, + "_IndexerModule__indexer_statistic", + staticmethod(lambda **_kwargs: None), + ) + site = { + **_build_indexer(), + "parser": "SunnyPT", + "pri": 1, + } + + torrents = object.__new__(IndexerModule).search_torrents( + site=site, + keyword="Movie", + mtype=MediaType.MOVIE, + cat="401", + page=2, + ) + + assert captured == { + "keyword": "Movie", + "mtype": MediaType.MOVIE, + "cat": "401", + "page": 2, + } + assert len(torrents) == 1 + assert torrents[0].title == "Movie.2026" + + +def test_sunnypt_site_test_uses_profile_api(monkeypatch): + """SunnyPT 连接测试应使用 Build 配置的 profile API 和 X-API-Key。""" + captured = {} + + def fake_get_indexer(domain: str): + """返回包含独立 API 地址的 SunnyPT 索引配置。""" + assert domain == "sunnypt.top" + return {"api_url": "https://api.sunnypt.top/api/v1/mp"} + + def fake_sites_helper(): + """构造不触发动态资源保护元类的站点帮助器替身。""" + return SimpleNamespace(get_indexer=fake_get_indexer) + + def fake_get_res(request, url: str, **_kwargs): + """记录站点连接测试请求并返回有效用户资料。""" + captured.update({"url": url, "headers": request._headers}) + return _FakeResponse({ + "code": 0, + "msg": "ok", + "data": {"download_allowed": True}, + }) + + monkeypatch.setattr("app.chain.site.SitesHelper", fake_sites_helper) + monkeypatch.setattr("app.chain.site.RequestUtils.get_res", fake_get_res) + site = SimpleNamespace( + domain="sunnypt.top", + ua="MoviePilot-Test", + apikey="sunny-secret", + proxy=0, + timeout=15, + ) + + state, message = SiteChain._SiteChain__sunnypt_test(site) + + assert state + assert message == "连接成功" + assert captured["url"] == "https://api.sunnypt.top/api/v1/mp/profile" + assert captured["headers"]["X-API-Key"] == "sunny-secret" + assert "Authorization" not in captured["headers"] + + +def test_indirect_download_does_not_log_or_cache_temporary_url(monkeypatch): + """两段式下载不得记录或缓存包含短时凭证的真实下载地址。""" + captured = {} + log_messages = [] + + def fake_post_res(_request, url: str, params: dict = None, **_kwargs): + """回放 download-token 接口返回的短时下载地址。""" + captured["token_url"] = url + captured["token_params"] = params + return _FakeResponse({ + "code": 0, + "msg": "ok", + "data": { + "download_url": "https://sunnypt.top/api/v1/mp/download/temporary-credential", + }, + }) + + def fake_download_torrent(_helper, **kwargs): + """记录 TorrentHelper 的失败缓存开关并返回有效种子内容。""" + captured.update(kwargs) + return None, b"torrent-content", "Movie", ["Movie.mkv"], "" + + def capture_log(message: str): + """收集下载链日志以校验敏感地址不会泄露。""" + log_messages.append(message) + + monkeypatch.setattr("app.chain.download.RequestUtils.post_res", fake_post_res) + monkeypatch.setattr( + "app.chain.download.TorrentHelper.download_torrent", + fake_download_torrent, + ) + monkeypatch.setattr( + "app.chain.download.logger", + SimpleNamespace(info=capture_log, error=capture_log), + ) + enclosure = SunnyPTSpider(_build_indexer())._build_download_url(123) + torrent = TorrentInfo( + title="Movie.2026", + enclosure=enclosure, + site_ua="MoviePilot-Test", + site_proxy=False, + ) + + content, folder, files = object.__new__(DownloadChain).download_torrent(torrent) + + assert content == b"torrent-content" + assert folder == "Movie" + assert files == ["Movie.mkv"] + assert captured["cache_invalid"] is False + assert captured["cookie"] is None + assert captured["url"] == ( + "https://api.sunnypt.top/api/v1/mp/download/temporary-credential" + ) + assert not any("sunny-secret" in message for message in log_messages) + assert not any("temporary-credential" in message for message in log_messages)