From 51d2ed1200084f8d2de1a9e62453b9f8a39b9faf Mon Sep 17 00:00:00 2001 From: mustangpt <168743846+mustangpt@users.noreply.github.com> Date: Mon, 3 Aug 2026 07:01:21 +0800 Subject: [PATCH] feat: support YemaPT Open API (#6227) --- app/chain/download.py | 13 +- app/modules/indexer/parser/__init__.py | 22 +- app/modules/indexer/parser/yema.py | 146 ++++++----- app/modules/indexer/spider/yema.py | 340 +++++++++++++++---------- tests/test_search_ai_recommend.py | 10 +- tests/test_yemapt_open_api.py | 284 +++++++++++++++++++++ 6 files changed, 598 insertions(+), 217 deletions(-) create mode 100644 tests/test_yemapt_open_api.py diff --git a/app/chain/download.py b/app/chain/download.py index 36fc8d74..553e0cde 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, urljoin, urlparse +from urllib.parse import parse_qs, urlencode, urljoin, urlparse from app import schemas from app.chain import ChainBase @@ -713,10 +713,21 @@ class DownloadChain(ChainBase): return res.text else: data = res.json() + success_key = req_params.get('success') + if success_key and not data.get(success_key): + return None for key in str(req_params.get('result')).split("."): data = data.get(key) if not data: return None + result_path = req_params.get('result_path') + result_query_param = req_params.get('result_query_param') + if result_path and result_query_param: + result_url = urljoin( + f"{str(req_params.get('result_base_url')).rstrip('/')}/", + str(result_path).lstrip('/'), + ) + return f"{result_url}?{urlencode({result_query_param: data})}" data = self._normalize_indirect_download_url( url=data, base_url=req_params.get('result_base_url'), diff --git a/app/modules/indexer/parser/__init__.py b/app/modules/indexer/parser/__init__.py index bc9dc3d0..5529da84 100644 --- a/app/modules/indexer/parser/__init__.py +++ b/app/modules/indexer/parser/__init__.py @@ -129,6 +129,8 @@ class SiteParserBase(metaclass=ABCMeta): self._user_basic_page = None # 用户基础信息参数 self._user_basic_params = None + # 用户基础信息请求方法 + self._user_basic_method = None # 用户基础信息请求头 self._user_basic_headers = None @@ -208,7 +210,8 @@ class SiteParserBase(metaclass=ABCMeta): self._get_page_content( url=urljoin(self._base_url, self._user_basic_page), params=self._user_basic_params, - headers=self._user_basic_headers + headers=self._user_basic_headers, + **({"method": self._user_basic_method} if self._user_basic_method else {}), ) ) else: @@ -325,12 +328,19 @@ class SiteParserBase(metaclass=ABCMeta): """ pass - def _get_page_content(self, url: str, params: dict = None, headers: dict = None): + def _get_page_content( + self, + url: str, + params: dict = None, + headers: dict = None, + method: Optional[str] = None, + ): """ 获取页面内容 :param url: 网页地址 :param params: post参数 :param headers: 额外的请求头 + :param method: 强制使用的 HTTP 请求方法 :return: """ req_headers = None @@ -363,19 +373,19 @@ class SiteParserBase(metaclass=ABCMeta): cookie = self._site_cookie session = self._session - if params: - if req_headers.get("Content-Type") == "application/json": + if method == "post" or params: + if (req_headers or {}).get("Content-Type") == "application/json": res = RequestUtils(cookies=cookie, session=session, timeout=60, proxies=proxies, - headers=req_headers).post_res(url=url, json=params) + headers=req_headers).post_res(url=url, json=params or {}) else: res = RequestUtils(cookies=cookie, session=session, timeout=60, proxies=proxies, - headers=req_headers).post_res(url=url, data=params) + headers=req_headers).post_res(url=url, data=params or {}) else: res = RequestUtils(cookies=cookie, session=session, diff --git a/app/modules/indexer/parser/yema.py b/app/modules/indexer/parser/yema.py index 79dec31c..f79d2bef 100644 --- a/app/modules/indexer/parser/yema.py +++ b/app/modules/indexer/parser/yema.py @@ -2,109 +2,121 @@ import json from typing import Optional, Tuple +from app.log import logger from app.modules.indexer.parser import SiteParserBase, SiteSchema from app.utils.string import StringUtils -class TYemaSiteUserInfo(SiteParserBase): - schema = SiteSchema.Yema +class YemaSiteUserInfo(SiteParserBase): + """ + YemaPT 开放 API 用户数据解析器 + """ - def _parse_site_page(self, html_text: str): + schema = SiteSchema.Yema + request_mode = "apikey" + + def _parse_site_page(self, html_text: str) -> None: """ - 获取站点页面地址 + 配置 YemaPT 用户基本信息接口和认证请求头 + + :param html_text: API AuthKey 模式下的空首页数据 """ - self._user_traffic_page = None - self._user_detail_page = None - self._user_basic_page = "api/consumer/fetchSelfDetail" + self._user_basic_page = "openApi/user/fetchBasicInfo.json" self._user_basic_params = {} + self._user_basic_method = "post" + self._user_detail_page = None + self._user_traffic_page = None + self._torrent_seeding_page = None self._sys_mail_unread_page = None self._user_mail_unread_page = None - self._mail_unread_params = {} - self._torrent_seeding_page = "/api/userTorrent/fetchSeedTorrentInfo" - self._torrent_seeding_params = { - # 虽然这个参数是无意义的，但这个 API 必须用 POST - "status": "seeding" - } - self._torrent_seeding_headers = {} self._addition_headers = { + "Authorization": self.apikey, "Content-Type": "application/json", "Accept": "application/json, text/plain, */*", + "User-Agent": self._ua, } - def _parse_logged_in(self, html_text): + def _parse_user_base_info(self, html_text: str) -> None: """ - 判断是否登录成功, 通过判断是否存在用户信息 - 暂时跳过检测，待后续优化 - :param html_text: - :return: - """ - return True + 解析开放 API 返回的用户基本信息和促销流量 - def _parse_user_base_info(self, html_text: str): - """ - 解析用户基本信息，这里把_parse_user_traffic_info和_parse_user_detail_info合并到这里 + :param html_text: fetchBasicInfo 接口响应文本 """ if not html_text: - return None - detail = json.loads(html_text) - if not detail or not detail.get("success"): + self.err_msg = "获取用户信息失败，未收到开放 API 响应" return - user_info = detail.get("data", {}) + try: + payload = json.loads(html_text) + except (TypeError, json.JSONDecodeError) as err: + self.err_msg = "获取用户信息失败，开放 API 响应不是有效 JSON" + logger.warning(f"{self._site_name} {self.err_msg}：{str(err)}") + return + if not isinstance(payload, dict): + self.err_msg = "获取用户信息失败，开放 API 响应结构无效" + logger.warning(f"{self._site_name} {self.err_msg}") + return + if not payload.get("success") or not isinstance(payload.get("data"), dict): + self.err_msg = payload.get("errorMessage") or "获取用户信息失败" + logger.warning(f"{self._site_name} 获取用户信息失败：{self.err_msg}") + return + + user_info = payload["data"] self.userid = user_info.get("id") self.username = user_info.get("name") - self.user_level = str(user_info.get("level")) if user_info.get("level") is not None else None + self.user_level = str(user_info.get("level")) \ + if user_info.get("level") is not None else None self.join_at = StringUtils.unify_datetime_str(user_info.get("registerTime")) - - self.upload = user_info.get('uploadSize') - # 使用 promotionDownloadSize 获取真实下载量（考虑促销因素） - if "promotionDownloadSize" in user_info: - self.download = user_info.get('promotionDownloadSize') - else: - self.download = user_info.get('downloadSize') + self.upload = int(user_info.get("promotionUploadSize") or 0) + self.download = int(user_info.get("promotionDownloadSize") or 0) self.ratio = round(self.upload / (self.download or 1), 2) - self.bonus = user_info.get("bonus") - self.message_unread = 0 + self.bonus = float(user_info.get("bonus") or 0) - def _parse_user_traffic_info(self, html_text: str): + def _parse_user_traffic_info(self, html_text: str) -> None: """ - 解析用户流量信息 + 跳过独立流量页面，用户基本信息接口已经返回促销流量 + + :param html_text: 未使用的页面文本 """ - pass - def _parse_user_detail_info(self, html_text: str): + def _parse_user_detail_info(self, html_text: str) -> None: """ - 解析用户详细信息 + 跳过独立用户详情页面，开放 API 未提供该接口 + + :param html_text: 未使用的页面文本 """ - pass - def _parse_user_torrent_seeding_info(self, html_text: str, multi_page: Optional[bool] = False) -> Optional[str]: + def _parse_user_torrent_seeding_info( + self, + html_text: str, + multi_page: bool = False, + ) -> Optional[str]: """ - 解析用户做种信息 + 跳过做种统计，开放 API 未提供该接口 + + :param html_text: 未使用的页面文本 + :param multi_page: 是否为后续分页 + :return: 始终返回 None """ - if not html_text: - return None - seeding_info = json.loads(html_text) - if not seeding_info or not seeding_info.get("success") or not seeding_info.get("data"): - return None - - torrents = seeding_info.get("data") - - self.seeding += torrents.get("num") - self.seeding_size += torrents.get("fileSize") - - # 是否存在下页数据 - next_page = None - - return next_page + return None def _parse_message_unread_links(self, html_text: str, msg_links: list) -> Optional[str]: """ - 解析未读消息链接，这里直接读出详情 - """ - pass + 跳过站内消息，开放 API 未提供该接口 - def _parse_message_content(self, html_text) -> Tuple[Optional[str], Optional[str], Optional[str]]: + :param html_text: 未使用的页面文本 + :param msg_links: 未使用的消息链接容器 + :return: 始终返回 None """ - 解析消息内容 + return None + + def _parse_message_content( + self, + html_text: str, + ) -> Tuple[Optional[str], Optional[str], Optional[str]]: """ - pass + 跳过消息详情，开放 API 未提供该接口 + + :param html_text: 未使用的页面文本 + :return: 三个空值 + """ + return None, None, None diff --git a/app/modules/indexer/spider/yema.py b/app/modules/indexer/spider/yema.py index 2d6c958b..c2a67c99 100644 --- a/app/modules/indexer/spider/yema.py +++ b/app/modules/indexer/spider/yema.py @@ -1,34 +1,23 @@ -from typing import Tuple, List, Optional +import base64 +import json +from typing import List, Optional, Tuple from app.core.config import settings -from app.db.systemconfig_oper import SystemConfigOper from app.log import logger from app.schemas import MediaType -from app.utils.http import RequestUtils, AsyncRequestUtils +from app.utils.http import AsyncRequestUtils, RequestUtils from app.utils.string import StringUtils class YemaSpider: """ - YemaPT API + YemaPT 开放 API 索引器 """ - _indexerid = None - _domain = None - _name = "" - _proxy = None - _cookie = None - _ua = None - _size = 40 - _searchurl = "%sapi/torrent/fetchOpenTorrentList" - _downloadurl = "%sapi/torrent/download?id=%s" - _pageurl = "%s#/torrent/detail/%s/" - _timeout = 15 - # 分类 + _size = 100 _movie_category = [4] - _tv_category = [5, 13, 14, 17, 15, 6, 16] + _tv_category = [5, 6, 13, 14, 15, 16, 17] - # 标签 https://wiki.yemapt.org/developer/constants _labels = { "1": "禁转", "2": "首发", @@ -44,173 +33,248 @@ class YemaSpider: "12": "完结", } + def __init__(self, indexer: dict): + """ + 初始化 YemaPT 开放 API 索引器 + + :param indexer: 合并站点认证信息后的索引配置 + """ + indexer = indexer or {} + self._name = indexer.get("name") or "YemaPT" + self._site_url = str(indexer.get("domain") or "https://www.yemapt.org/").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._search_url = f"{self._site_url}/openApi/torrent/fetchOpenTorrentList.json" + self._download_key_url = f"{self._site_url}/openApi/torrent/generateDownloadKey.json" + @classmethod def get_search_page_size(cls, keyword: Optional[str] = None) -> Optional[int]: """ - 获取搜索接口单页容量。 + 获取搜索接口单页容量 + + :param keyword: 搜索关键字，YemaPT 不按关键字改变分页容量 + :return: 搜索接口单页容量 """ return cls._size - def __init__(self, indexer: dict): - self.systemconfig = SystemConfigOper() - if indexer: - self._indexerid = indexer.get('id') - self._domain = indexer.get('domain') - self._searchurl = self._searchurl % self._domain - self._name = indexer.get('name') - if indexer.get('proxy'): - self._proxy = settings.PROXY - self._cookie = indexer.get('cookie') - self._ua = indexer.get('ua') - self._timeout = indexer.get('timeout') or 15 - - def __get_params(self, keyword: str = None, page: Optional[int] = 0) -> dict: + def _request_headers(self) -> dict: """ - 获取搜索参数 + 构造开放 API 请求头 + + :return: 不包含 Cookie 的 AuthKey 请求头 + """ + return { + "Authorization": self._api_key, + "Content-Type": "application/json", + "Accept": "application/json, text/plain, */*", + "User-Agent": self._user_agent, + } + + def _build_params( + self, + keyword: Optional[str], + page: Optional[int], + ) -> dict: + """ + 构造公开种子列表查询参数 + + :param keyword: 搜索关键字 + :param page: MoviePilot 从 0 开始的页码 + :return: YemaPT 开放 API 请求体 """ params = { "pageParam": { - "current": page + 1, + "current": int(page or 0) + 1, "pageSize": self._size, - "total": self._size }, - "sorter": {} + "sorter": {}, } if keyword: - params.update({ - "keyword": keyword, - }) + params["keyword"] = keyword return params - def __parse_result(self, results: List[dict]) -> List[dict]: + def _parse_result(self, results: List[dict]) -> List[dict]: """ - 解析搜索结果 + 将开放 API 种子数据转换为 MoviePilot 标准字段 + + :param results: 公开种子列表接口 data 数组 + :return: MoviePilot 标准种子字典列表 """ torrents = [] - if not results: - return torrents - - for result in results: - category_value = result.get('categoryId') + for result in results or []: + if not isinstance(result, dict): + continue + category_value = result.get("categoryId") if category_value in self._tv_category: category = MediaType.TV.value elif category_value in self._movie_category: category = MediaType.MOVIE.value else: category = MediaType.UNKNOWN.value - pass - - torrentLabelIds = result.get('tagList', []) or [] - torrentLabels = [] - for labelId in torrentLabelIds: - if self._labels.get(labelId) is not None: - torrentLabels.append(self._labels.get(labelId)) - pass - pass - torrent = { - 'title': result.get('showName'), - 'description': result.get('shortDesc'), - 'enclosure': self.__get_download_url(result.get('id')), - 'pubdate': StringUtils.unify_datetime_str(result.get('listingTime')), - 'size': result.get('fileSize'), - 'seeders': result.get('seedNum'), - 'peers': result.get('leechNum'), - 'grabs': result.get('completedNum'), - 'downloadvolumefactor': self.__get_downloadvolumefactor(result.get('downloadPromotion')), - 'uploadvolumefactor': self.__get_uploadvolumefactor(result.get('uploadPromotion')), - 'freedate': StringUtils.unify_datetime_str(result.get('downloadPromotionEndTime')), - 'page_url': self._pageurl % (self._domain, result.get('id')), - 'labels': torrentLabels, - 'category': category - } - torrents.append(torrent) + labels = [ + self._labels[label_id] + for label_id in result.get("tagList") or [] + if label_id in self._labels + ] + torrent_id = result.get("id") + torrents.append({ + "title": result.get("showName"), + "description": result.get("shortDesc"), + "enclosure": self._build_download_url(torrent_id), + "pubdate": StringUtils.unify_datetime_str(result.get("listingTime")), + "size": result.get("fileSize"), + "seeders": result.get("seedNum"), + "peers": result.get("leechNum"), + "grabs": result.get("completedNum"), + "downloadvolumefactor": self._download_factor(result.get("downloadPromotion")), + "uploadvolumefactor": self._upload_factor(result.get("uploadPromotion")), + "freedate": StringUtils.unify_datetime_str(result.get("downloadPromotionEndTime")), + "page_url": f"{self._site_url}/#/torrent/detail/{torrent_id}/", + "labels": labels, + "hit_and_run": bool(result.get("hrPunishEnable")), + "category": category, + }) return torrents - def search(self, keyword: str, - mtype: MediaType = None, page: Optional[int] = 0) -> Tuple[bool, List[dict]]: - """ - 搜索 + def _process_search_response(self, response) -> Tuple[bool, List[dict]]: """ + 校验开放 API 通用响应并解析搜索结果 - res = RequestUtils( - headers={ - "Content-Type": "application/json", - "User-Agent": f"{self._ua}", - "Accept": "application/json, text/plain, */*" - }, - cookies=self._cookie, + :param response: RequestUtils 返回的响应对象 + :return: 是否失败及标准种子列表 + """ + if response is None: + logger.warning(f"{self._name} 搜索失败，无法连接开放 API") + return True, [] + if response.status_code != 200: + logger.warning(f"{self._name} 搜索失败，HTTP 错误码：{response.status_code}") + return True, [] + try: + payload = response.json() or {} + except (TypeError, ValueError) as err: + logger.warning(f"{self._name} 搜索响应不是有效 JSON：{str(err)}") + return True, [] + if not isinstance(payload, dict): + logger.warning(f"{self._name} 搜索响应结构无效") + return True, [] + if not payload.get("success"): + logger.warning(f"{self._name} 搜索失败：{payload.get('errorMessage') or '未知错误'}") + return True, [] + results = payload.get("data") + if not isinstance(results, list): + logger.warning(f"{self._name} 搜索响应 data 不是数组") + return True, [] + return False, self._parse_result(results) + + def search( + self, + keyword: Optional[str], + mtype: MediaType = None, + page: Optional[int] = 0, + ) -> Tuple[bool, List[dict]]: + """ + 同步搜索 YemaPT 公开种子 + + :param keyword: 搜索关键字 + :param mtype: MoviePilot 媒体类型，开放 API 不支持直接按媒体类型查询 + :param page: MoviePilot 从 0 开始的页码 + :return: 是否失败及标准种子列表 + """ + if not self._api_key: + logger.warning(f"{self._name} 未配置 API AuthKey") + return True, [] + response = RequestUtils( + headers=self._request_headers(), proxies=self._proxy, - referer=f"{self._domain}", - timeout=self._timeout - ).post_res(url=self._searchurl, json=self.__get_params(keyword, page)) - if res and res.status_code == 200: - results = res.json().get('data', []) or [] - return False, self.__parse_result(results) - elif res is not None: - logger.warn(f"{self._name} 搜索失败，错误码：{res.status_code}") - return True, [] - else: - logger.warn(f"{self._name} 搜索失败，无法连接 {self._domain}") - return True, [] + timeout=self._timeout, + ).post_res( + url=self._search_url, + json=self._build_params(keyword, page), + ) + return self._process_search_response(response) - async def async_search(self, keyword: str, - mtype: MediaType = None, page: Optional[int] = 0) -> Tuple[bool, List[dict]]: + async def async_search( + self, + keyword: Optional[str], + mtype: MediaType = None, + page: Optional[int] = 0, + ) -> Tuple[bool, List[dict]]: """ - 异步搜索 + 异步搜索 YemaPT 公开种子 + + :param keyword: 搜索关键字 + :param mtype: MoviePilot 媒体类型，开放 API 不支持直接按媒体类型查询 + :param page: MoviePilot 从 0 开始的页码 + :return: 是否失败及标准种子列表 """ - res = await AsyncRequestUtils( - headers={ - "Content-Type": "application/json", - "User-Agent": f"{self._ua}", - "Accept": "application/json, text/plain, */*" - }, - cookies=self._cookie, + if not self._api_key: + logger.warning(f"{self._name} 未配置 API AuthKey") + return True, [] + response = await AsyncRequestUtils( + headers=self._request_headers(), proxies=self._proxy, - referer=f"{self._domain}", - timeout=self._timeout - ).post_res(url=self._searchurl, json=self.__get_params(keyword, page)) - - if res and res.status_code == 200: - results = res.json().get('data', []) or [] - return False, self.__parse_result(results) - elif res is not None: - logger.warn(f"{self._name} 搜索失败，错误码：{res.status_code}") - return True, [] - else: - logger.warn(f"{self._name} 搜索失败，无法连接 {self._domain}") - return True, [] + timeout=self._timeout, + ).post_res( + url=self._search_url, + json=self._build_params(keyword, page), + ) + return self._process_search_response(response) @staticmethod - def __get_downloadvolumefactor(discount: str) -> float: + def _download_factor(promotion: str) -> float: """ - 获取下载系数 + 转换下载促销类型 + + :param promotion: 开放 API 下载促销枚举 + :return: MoviePilot 下载系数 """ - discount_dict = { + return { "free": 0, "half": 0.5, - "none": 1 - } - if discount: - return discount_dict.get(discount, 1) - return 1 + "none": 1, + }.get(promotion, 1) @staticmethod - def __get_uploadvolumefactor(discount: str) -> float: + def _upload_factor(promotion: str) -> float: """ - 获取上传系数 + 转换上传促销类型 + + :param promotion: 开放 API 上传促销枚举 + :return: MoviePilot 上传系数 """ - discount_dict = { + return { "none": 1, "one_half": 1.5, - "double_upload": 2 - } - if discount: - return discount_dict.get(discount, 1) - return 1 + "double_upload": 2, + }.get(promotion, 1) - def __get_download_url(self, torrent_id: str) -> str: + def _build_download_url(self, torrent_id: int) -> str: """ - 获取下载链接 + 构造先生成下载凭证再获取种子文件的两段式链接 + + :param torrent_id: YemaPT 种子 ID + :return: Base64 请求配置与下载凭证接口 URL """ - return self._downloadurl % (self._domain, torrent_id) + request_config = { + "method": "post", + "cookie": False, + "header": { + "Authorization": self._api_key, + "Accept": "application/json", + }, + "params": {"id": torrent_id}, + "proxy": self._use_proxy, + "success": "success", + "result": "data", + "result_base_url": self._site_url, + "result_path": "api/torrent/download1", + "result_query_param": "token", + } + encoded_config = base64.b64encode( + json.dumps(request_config).encode("utf-8") + ).decode("ascii") + return f"[{encoded_config}]{self._download_key_url}" diff --git a/tests/test_search_ai_recommend.py b/tests/test_search_ai_recommend.py index 5fea1d83..6a59dd61 100644 --- a/tests/test_search_ai_recommend.py +++ b/tests/test_search_ai_recommend.py @@ -252,18 +252,18 @@ class SearchChainAIRecommendTest(unittest.IsolatedAsyncioTestCase): def test_search_all_sites_uses_parser_page_size_for_yema(self): """ - 验证专用解析器按自身页容量判断，避免 Yema 的 40 条分页被误停。 + 验证专用解析器按自身页容量判断，避免 Yema 的 100 条分页被误停。 """ chain = self._make_chain() requested_pages = [] def search_torrents(**kwargs): """ - 模拟 Yema 第一页满 40 条，第二页不足 40 条后停止。 + 模拟 Yema 第一页满 100 条，第二页不足 100 条后停止。 """ page = kwargs["page"] requested_pages.append(page) - count = 40 if page == 0 else 39 + count = 100 if page == 0 else 99 return [ SimpleNamespace(title=f"Result Page {page}-{index}", description="") for index in range(count) @@ -294,14 +294,14 @@ class SearchChainAIRecommendTest(unittest.IsolatedAsyncioTestCase): ) self.assertEqual([0, 1], requested_pages) - self.assertEqual(79, len(results)) + self.assertEqual(199, len(results)) def test_indexer_module_search_page_size_uses_spider_metadata(self): """ 验证站点单页容量由索引器模块统一读取，避免搜索链写死 parser 容量。 """ self.assertEqual( - 40, + 100, IndexerModule.get_search_page_size({"parser": "Yema"}, keyword="keyword") ) self.assertEqual( diff --git a/tests/test_yemapt_open_api.py b/tests/test_yemapt_open_api.py new file mode 100644 index 00000000..299c26e7 --- /dev/null +++ b/tests/test_yemapt_open_api.py @@ -0,0 +1,284 @@ +import asyncio +import base64 +import json + +from app.chain.download import DownloadChain +from app.core.context import TorrentInfo +from app.modules.indexer.parser.yema import YemaSiteUserInfo +from app.modules.indexer.spider.yema import YemaSpider +from app.schemas import MediaType + + +class _FakeResponse: + """构造 YemaPT 开放 API 测试使用的最小响应对象。""" + + def __init__(self, payload: dict, status_code: int = 200): + """保存响应数据和状态码。""" + self._payload = payload + self.status_code = status_code + self.reason = "OK" + self.text = json.dumps(payload) + + def __bool__(self) -> bool: + """按 HTTP 成功状态模拟 requests.Response 的布尔值。""" + return self.status_code < 400 + + def json(self) -> dict: + """返回预设 JSON 数据。""" + return self._payload + + +def _build_indexer() -> dict: + """构造 YemaPT 开放 API Spider 所需的最小站点配置。""" + return { + "id": "yemapt", + "name": "YemaPT", + "domain": "https://www.yemapt.org/", + "apikey": "yema-auth-key", + "ua": "MoviePilot-Test", + "proxy": False, + } + + +def _torrent_response() -> _FakeResponse: + """构造 YemaPT 公开种子列表响应。""" + return _FakeResponse({ + "success": True, + "showType": 0, + "data": [{ + "id": 100, + "showName": "Movie.2026.2160p.WEB-DL.H.265-GROUP", + "shortDesc": "电影中文副标题", + "categoryId": 4, + "fileSize": 21474836480, + "seedNum": 15, + "leechNum": 2, + "completedNum": 30, + "listingTime": "2026-08-02T14:00:00+08:00", + "uploadPromotion": "double_upload", + "downloadPromotion": "free", + "downloadPromotionEndTime": "2026-08-04T14:00:00+08:00", + "tagList": ["6", "9"], + "hrPunishEnable": True, + }], + }) + + +def test_yemapt_search_uses_open_api_auth_and_maps_fields(monkeypatch): + """YemaPT 搜索应使用 AuthKey、开放接口及标准字段映射。""" + captured = {} + + def fake_post_res(request, url: str, json: dict = None, **_kwargs): + """记录开放 API 搜索请求并回放种子数据。""" + captured.update({ + "url": url, + "json": json, + "headers": request._headers, + "cookies": request._cookies, + }) + return _torrent_response() + + monkeypatch.setattr( + "app.modules.indexer.spider.yema.RequestUtils.post_res", + fake_post_res, + ) + + error, torrents = YemaSpider(_build_indexer()).search( + keyword="Movie", + mtype=MediaType.MOVIE, + page=2, + ) + + assert not error + assert captured == { + "url": "https://www.yemapt.org/openApi/torrent/fetchOpenTorrentList.json", + "json": { + "keyword": "Movie", + "pageParam": {"current": 3, "pageSize": 100}, + "sorter": {}, + }, + "headers": { + "Authorization": "yema-auth-key", + "Content-Type": "application/json", + "Accept": "application/json, text/plain, */*", + "User-Agent": "MoviePilot-Test", + }, + "cookies": None, + } + assert torrents[0] == { + "title": "Movie.2026.2160p.WEB-DL.H.265-GROUP", + "description": "电影中文副标题", + "enclosure": torrents[0]["enclosure"], + "pubdate": "2026-08-02 14:00:00", + "size": 21474836480, + "seeders": 15, + "peers": 2, + "grabs": 30, + "downloadvolumefactor": 0, + "uploadvolumefactor": 2, + "freedate": "2026-08-04 14:00:00", + "page_url": "https://www.yemapt.org/#/torrent/detail/100/", + "labels": ["中字", "HDR10"], + "hit_and_run": True, + "category": MediaType.MOVIE.value, + } + + encoded_config, credential_url = torrents[0]["enclosure"].split("]", 1) + request_config = json.loads(base64.b64decode(encoded_config[1:]).decode("utf-8")) + assert credential_url == ( + "https://www.yemapt.org/openApi/torrent/generateDownloadKey.json" + ) + assert request_config["header"]["Authorization"] == "yema-auth-key" + assert request_config["params"] == {"id": 100} + assert request_config["success"] == "success" + assert request_config["result_path"] == "api/torrent/download1" + assert request_config["result_query_param"] == "token" + + +def test_yemapt_search_rejects_business_failure(monkeypatch): + """HTTP 成功但业务失败时，YemaPT 搜索必须返回错误。""" + monkeypatch.setattr( + "app.modules.indexer.spider.yema.RequestUtils.post_res", + lambda *_args, **_kwargs: _FakeResponse({ + "success": False, + "errorCode": 403, + "errorMessage": "need api auth", + }), + ) + + error, torrents = YemaSpider(_build_indexer()).search(keyword="Movie") + + assert error + assert torrents == [] + + +def test_yemapt_async_search_uses_open_api(monkeypatch): + """YemaPT 异步搜索应使用与同步搜索相同的开放 API 契约。""" + captured = {} + + async def fake_post_res(request, url: str, json: dict = None, **_kwargs): + """记录异步开放 API 请求并回放种子数据。""" + captured.update({ + "url": url, + "json": json, + "headers": request._headers, + "cookies": request._cookies, + }) + return _torrent_response() + + monkeypatch.setattr( + "app.modules.indexer.spider.yema.AsyncRequestUtils.post_res", + fake_post_res, + ) + + error, torrents = asyncio.run( + YemaSpider(_build_indexer()).async_search(keyword="Movie", page=0) + ) + + assert not error + assert len(torrents) == 1 + assert captured["url"].endswith("/openApi/torrent/fetchOpenTorrentList.json") + assert captured["headers"]["Authorization"] == "yema-auth-key" + assert captured["cookies"] is None + + +def test_yemapt_user_parser_uses_basic_info_only(monkeypatch): + """YemaPT 用户解析器应仅调用开放 API 已提供的基本信息接口。""" + captured = [] + payload = { + "success": True, + "showType": 0, + "data": { + "id": 10, + "name": "yema-user", + "bonus": 1000000, + "level": 7, + "registerTime": "2024-05-01T00:00:00+08:00", + "promotionUploadSize": 2000000, + "promotionDownloadSize": 1000000, + }, + } + + def fake_post_res(request, url: str, json: dict = None, **_kwargs): + """记录用户基本信息请求并回放开放 API 响应。""" + captured.append((url, json, request._headers, request._cookies)) + return _FakeResponse(payload) + + monkeypatch.setattr( + "app.modules.indexer.parser.RequestUtils.post_res", + fake_post_res, + ) + parser = YemaSiteUserInfo( + site_name="YemaPT", + url="https://www.yemapt.org/", + site_cookie="legacy-cookie", + apikey="yema-auth-key", + token=None, + ua="MoviePilot-Test", + proxy=False, + ) + + parser.parse() + + assert len(captured) == 1 + assert captured[0][0] == "https://www.yemapt.org/openApi/user/fetchBasicInfo.json" + assert captured[0][1] == {} + assert captured[0][2]["Authorization"] == "yema-auth-key" + assert captured[0][3] is None + assert parser.userid == 10 + assert parser.username == "yema-user" + assert parser.user_level == "7" + assert parser.upload == 2000000 + assert parser.download == 1000000 + assert parser.ratio == 2.0 + + +def test_yemapt_download_generates_and_urlencodes_temporary_key(monkeypatch): + """YemaPT 下载应临时生成凭证并在下载 URL 中安全编码。""" + captured = {} + + def fake_post_res(request, url: str, params: dict = None, **_kwargs): + """回放下载凭证接口响应。""" + captured.update({ + "credential_url": url, + "credential_params": params, + "credential_headers": request._headers, + "credential_cookies": request._cookies, + }) + return _FakeResponse({"success": True, "data": "abc+/="}) + + def fake_download_torrent(_helper, **kwargs): + """记录最终种子文件下载地址并返回有效种子内容。""" + captured.update(kwargs) + return None, b"torrent-content", "Movie", ["Movie.mkv"], "" + + monkeypatch.setattr("app.chain.download.RequestUtils.post_res", fake_post_res) + monkeypatch.setattr( + "app.chain.download.TorrentHelper.download_torrent", + fake_download_torrent, + ) + enclosure = YemaSpider(_build_indexer())._build_download_url(100) + torrent = TorrentInfo( + title="Movie.2026", + enclosure=enclosure, + site_cookie="legacy-cookie", + 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["credential_url"] == ( + "https://www.yemapt.org/openApi/torrent/generateDownloadKey.json" + ) + assert captured["credential_params"] == {"id": 100} + assert captured["credential_headers"]["Authorization"] == "yema-auth-key" + assert captured["credential_cookies"] is None + assert captured["url"] == ( + "https://www.yemapt.org/api/torrent/download1?token=abc%2B%2F%3D" + ) + assert captured["cookie"] is None + assert captured["cache_invalid"] is False