feat: support SunnyPT API indexer

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,

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]:
         """

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]]:

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()

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

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

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}"

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)