nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-11-15 12:50:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(indexer): 支持 FluxTorrent 兼容层站点搜索与用户数据 (#7355)

Browse Source 
* feat(indexer): 支持 FluxTorrent 兼容层站点搜索与用户数据

好学（hxpt.org）已由 NexusPHP 迁移到 FluxTorrent，原有页面刮削全部失效。
FluxTorrent 提供 /api/v1/compat/nexusphp/* 只读 JSON 兼容层，本提交按该契约
新增索引器与用户数据解析器，并在分发层加入能力探测，使未在站点资源中登记
专站协议的站点也能被自动识别（站点清单是闭源发布资源，主仓无法新增条目）。

- 新增 FluxTorrentSpider：调用兼容层 torrents.json 搜索，按 category_newznab
  映射媒体类型，按 promotion 换算下载与上传量系数，下载地址携带开放 API 令牌
- 新增 FluxTorrentSiteUserInfo：解析兼容层 user.json 的用户资料与流量
- 新增同步与异步能力探测：站点未登记专站协议时，用站点令牌请求一次
  user.json，响应符合 {code, message, data, request_id} 信封即判定为 FluxTorrent
- 探测要求站点已配置令牌，并用命中 24 小时、未命中 30 分钟的双层 TTL 缓存
- 不下发分类与媒体类型筛选：兼容层的 category 只接受站内分类 ID，且客户端
  过滤会破坏分页续拉契约，分类筛选待站点资源发布链路补齐后再放开

* fix(indexer): 抽出用户数据模型解析，消除复杂度增长

CI 复杂度门禁报 IndexerModule.refresh_userdata 圈复杂度 15->16（上限 15）。
把 FluxTorrent 能力探测与模型选择抽成模块级函数 _resolve_userdata_schema，
refresh_userdata 的方法体不再因新增分支而增长。

* refactor(indexer): FluxTorrent 模块文件改为单词命名

命名门禁 test_new_host_production_modules_use_single_word_filenames 要求
app 下的模块文件名满足 [a-z][a-z0-9]*，flux_torrent.py 不合规 ⇒ 两个模块
改名为 fluxtorrent.py，同步更新导入与测试打桩路径。

* test(indexer): 修正发布时间断言并同步架构依赖基线

- 发布时间断言原先写死本机时区渲染结果，在 UTC 的 CI 上必然失败 ⇒ 改为对同一
  转换函数取期望值，只验证 Unix 时间戳经由 format_timestamp 输出
- dependency-baseline.json 新增 app.modules.indexer.spider.fluxtorrent 与
  app.modules.indexer.parser.fluxtorrent 的模块与导入边（edge_count 9973→9990、
  module_count 1151→1153），结构与既有 haidan 等专站模块一致，边界未越权

* docs(architecture): 同步架构快照的模块与导入边数量

架构依赖基线新增两个索引器模块后，docs/architecture-overview.md 的
快照表需同步：Python 模块 1151→1153、内部导入边 9,973→9,990。

* test(indexer): 补齐异步探测与失败路径用例，满足新文件覆盖率门槛

新增文件的覆盖率低水位要求 80%，补齐后 spider 90.81%、parser 92.21%：
- 异步能力探测命中与请求异常两条路径
- 站点无响应、响应不是 JSON 时的降级判定
- 异步搜索缺少令牌时的直接失败
- 用户接口空响应 / 非 JSON / 结构异常 / 业务错误码的错误信息
- 站点地址缺协议与空白地址的归一

* test(indexer): 覆盖协议解析的三条分支

存量文件覆盖率低水位要求未覆盖语句数不增长，补齐协议解析的已登记协议、
探测命中、未识别三种情况（同步与异步），以及用户数据模型名的两条分支。

* refactor(indexer): 按站点资源路由 FluxTorrent 协议

删除运行时能力探测与 TTL 缓存，搜索和用户数据直接使用站点资源声明的 parser/schema；保留 FluxTorrent spider 与用户数据解析器的注册，由站点构建配置决定路由。

---------

Co-authored-by: gntv456 <gntv456@users.noreply.github.com>
Co-authored-by: jxxghp <jxxghp@gmail.com>
This commit is contained in:
gntv456
2026-11-11 10:27:52 +08:00
committed by GitHub
parent dcdebe366a
commit 6166968f4c
7 changed files with 978 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
app/modules/indexer/__init__.py
View File

@@ -22,6 +22,7 @@ from app.foundation.reflection import ModuleHelper
from app.modules import _ModuleBase
from app.modules.indexer.parser import SiteParserBase, SiteSchema
from app.modules.indexer.spider import SiteSpider
from app.modules.indexer.spider.fluxtorrent import FluxTorrentSpider
from app.modules.indexer.spider.haidan import HaiDanSpider
from app.modules.indexer.spider.hddolby import HddolbySpider
from app.modules.indexer.spider.milkie import MilkieSpider
@@ -74,6 +75,7 @@ SPIDER_PARSER_CLASSES: dict[str, type[_PagedSpider]] = {
"RousiPro": RousiSpider,
"SunnyPT": SunnyPTSpider,
"Milkie": MilkieSpider,
"FluxTorrent": FluxTorrentSpider,
}
# 仅传递各专站搜索入口支持的参数，避免同步与异步调用契约分叉。
@@ -87,6 +89,7 @@ _SPECIALIZED_SEARCH_ARGUMENTS = {
"HDDolby": ("keyword", "mtype", "cat", "page"),
"RousiPro": ("keyword", "mtype", "cat", "page"),
"Milkie": ("keyword", "mtype", "page"),
"FluxTorrent": ("keyword", "page"),
}
# 专站字幕入口支持的参数顺序，其余站点使用通用页面解析。

2
app/modules/indexer/parser/__init__.py
View File

@@ -67,6 +67,8 @@ class SiteSchema(Enum):
RousiPro = "RousiPro"
# SunnyPT 的 MoviePilot API。
SunnyPT = "SunnyPT"
# FluxTorrent 兼容层站点。
FluxTorrent = "FluxTorrent"
class SiteParserBase(metaclass=ABCMeta):

145
app/modules/indexer/parser/fluxtorrent.py Normal file
View File

@@ -0,0 +1,145 @@
"""
解析 FluxTorrent 兼容层的用户资料与流量信息。
"""
import json
from typing import Any, Dict, Optional, Tuple
from app.modules.indexer.parser import SitePage, SiteParserBase, SiteSchema
from app.runtime.log import logger
class FluxTorrentSiteUserInfo(SiteParserBase):
"""
FluxTorrent 兼容层用户数据解析器。
兼容层按 NexusPHP userdetails 口径返回用户资料，因此不需要解析站点页面。
"""
schema = SiteSchema.FluxTorrent
request_mode = "apikey"
def _parse_site_page(self, html_text: SitePage) -> None:
"""
配置兼容层用户信息接口地址和认证请求头。
:param html_text: API Key 模式下的空首页数据
"""
self._user_basic_page = "api/v1/compat/nexusphp/user.json"
self._user_basic_params = {}
self._user_detail_page = None
self._user_traffic_page = None
self._torrent_seeding_page = None
self._user_mail_unread_page = None
self._sys_mail_unread_page = None
token = self.apikey or self.token or ""
headers: Dict[str, Optional[str]] = {
"Authorization": f"Token {token}",
"Accept": "application/json",
}
if self._ua:
headers["User-Agent"] = self._ua
self._addition_headers = headers
def _load_api_payload(self, html_text: SitePage, operation: str) -> Optional[Dict[str, Any]]:
"""
解析并校验兼容层统一响应信封。
:param html_text: 兼容层响应文本
:param operation: 错误信息中的操作名称
:return: 成功时返回响应 data，失败时返回 None
"""
if not html_text:
self.err_msg = f"{operation}失败，未收到接口响应"
return None
payload: Any
if isinstance(html_text, dict):
payload = html_text
else:
try:
payload = json.loads(html_text)
except (TypeError, json.JSONDecodeError) as err:
self.err_msg = f"{operation}失败，接口响应不是有效 JSON"
logger.warning(f"{self._site_name} {self.err_msg}：{str(err)}")
return None
if not isinstance(payload, dict):
self.err_msg = f"{operation}失败，接口响应结构无效"
logger.warning(f"{self._site_name} {self.err_msg}")
return None
if payload.get("code") != 0:
self.err_msg = payload.get("message") 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: SitePage) -> None:
"""
解析兼容层返回的用户资料与流量统计。
:param html_text: user.json 接口响应文本
"""
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("class_name") or str(user_info.get("class") or "")
self.upload = int(user_info.get("uploaded") or 0)
self.download = int(user_info.get("downloaded") or 0)
# 站点在只上传未下载时把分享率记为无穷大，此时没有可用的有限数值。
if user_info.get("ratio_infinite") is True:
self.ratio = 0.0
else:
self.ratio = float(user_info.get("ratio") or 0)
self.bonus = float(user_info.get("seedbonus") or 0)
def _parse_user_traffic_info(self, html_text: str) -> None:
"""
跳过独立流量页面，兼容层用户接口已经返回完整统计。
:param html_text: 未使用的页面文本
"""
def _parse_user_detail_info(self, html_text: str) -> None:
"""
跳过独立用户详情页面，兼容层用户接口已经返回完整资料。
:param html_text: 未使用的页面文本
"""
def _parse_user_torrent_seeding_info(
self,
html_text: str,
multi_page: bool = False,
) -> Optional[str]:
"""
跳过独立做种列表，兼容层用户接口未提供做种统计。
:param html_text: 未使用的页面文本
:param multi_page: 是否为后续分页
:return: 始终返回 None
"""
return None
def _parse_message_unread_links(self, html_text: str, msg_links: list[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]]:
"""
兼容层未提供站内消息接口，跳过消息内容解析。
:param html_text: 未使用的页面文本
:return: 三个空值
"""
return None, None, None

328
app/modules/indexer/spider/fluxtorrent.py Normal file
View File

@@ -0,0 +1,328 @@
"""
通过 FluxTorrent 兼容层搜索种子并转换为通用索引结果。
"""
from typing import Any, Dict, List, Optional, Tuple
from app.adapters.network.http import AsyncRequestUtils, RequestUtils
from app.foundation import temporal as time_tools
from app.runtime.log import logger
from app.runtime.settings import get_runtime_setting
from app.schemas.types import MediaType
# FluxTorrent 兼容层的只读接口前缀。
_COMPAT_PREFIX = "/api/v1/compat/nexusphp"
# 兼容层促销档位到下载、上传量系数的映射，与站点计费口径一致：
# free 下载 0 / x2 上传 2 / x2free 0+2 / half 下载 0.5 / x2half 0.5+2 / p30 下载 0.3。
_PROMOTION_FACTORS: Dict[str, Tuple[float, float]] = {
"free": (0.0, 1.0),
"x2free": (0.0, 2.0),
"half": (0.5, 1.0),
"x2half": (0.5, 2.0),
"x2": (1.0, 2.0),
"p30": (0.3, 1.0),
}
# Newznab 标准分类号到 MoviePilot 媒体类型的映射。站点按标准号登记分类，
# 因此同一套映射对所有 FluxTorrent 站点都成立；软件与游戏没有对应媒体类型，
# 有声书与电子书归入系列。
_NEWZNAB_MEDIA_TYPES: Dict[int, MediaType] = {
2000: MediaType.MOVIE,
2010: MediaType.MOVIE,
2030: MediaType.MOVIE,
2040: MediaType.MOVIE,
2045: MediaType.MOVIE,
3000: MediaType.MUSIC,
3030: MediaType.MUSIC,
3040: MediaType.MUSIC,
5000: MediaType.TV,
5030: MediaType.TV,
5040: MediaType.TV,
5060: MediaType.TV,
5070: MediaType.TV,
5080: MediaType.TV,
5090: MediaType.TV,
7000: MediaType.COLLECTION,
7010: MediaType.COLLECTION,
7020: MediaType.COLLECTION,
7030: MediaType.COLLECTION,
}
# 站点分类名到 MoviePilot 媒体类型的兜底映射，标准号缺失时使用。
_CATEGORY_NAME_MEDIA_TYPES: Dict[str, MediaType] = {
"电影": MediaType.MOVIE,
"电视剧": MediaType.TV,
"短剧": MediaType.TV,
"综艺": MediaType.TV,
"动漫": MediaType.TV,
"纪录片": MediaType.TV,
"教育影音": MediaType.TV,
"音乐": MediaType.MUSIC,
"有声书": MediaType.COLLECTION,
"电子书": MediaType.COLLECTION,
}
def site_base_url(site: Dict[str, Any]) -> str:
"""
返回站点根地址。
:param site: 站点索引配置
:return: 去掉末尾斜杠的站点根地址；没有可用地址时为空字符串
"""
raw = str(site.get("url") or site.get("domain") or "").strip()
if not raw:
return ""
if not raw.startswith(("http://", "https://")):
raw = f"https://{raw.lstrip('/')}"
return raw.rstrip("/")
def site_api_token(site: Dict[str, Any]) -> str:
"""
返回站点配置的开放 API 令牌。
:param site: 站点索引配置
:return: 开放 API 令牌；未配置时为空字符串
"""
return str(site.get("apikey") or site.get("token") or "").strip()
class FluxTorrentSpider:
"""
FluxTorrent 兼容层索引器。
FluxTorrent 是通用 PT 建站系统，其兼容层按 NexusPHP 字段口径暴露只读 JSON，
因此本索引器不解析站点页面，只调用兼容层接口。
"""
# 兼容层单页容量上限为 50。
_size = 50
def __init__(self, indexer: dict[str, Any]):
"""
初始化 FluxTorrent 兼容层索引器。
:param indexer: 合并站点认证信息后的索引配置
"""
indexer = indexer or {}
self._name = indexer.get("name") or "FluxTorrent"
self._site_url = site_base_url(indexer)
self._api_token = site_api_token(indexer)
self._proxy = get_runtime_setting("PROXY") if indexer.get("proxy") else None
self._user_agent = indexer.get("ua") or get_runtime_setting("USER_AGENT")
self._timeout = indexer.get("timeout") or 15
@classmethod
def get_search_page_size(cls, keyword: Optional[str] = None) -> Optional[int]:
"""
获取搜索接口单页容量。
:param keyword: 搜索关键字，兼容层不按关键字改变分页容量
:return: 搜索接口单页容量
"""
return cls._size
@property
def _base_url(self) -> str:
"""
返回兼容层接口基址。
:return: 站点根路径拼接兼容层前缀后的地址
"""
return f"{self._site_url}{_COMPAT_PREFIX}"
def _request_headers(self) -> Dict[str, Any]:
"""
构造兼容层请求头。
:return: 带开放 API 令牌的请求头
"""
return {
"Authorization": f"Token {self._api_token}",
"Accept": "application/json",
"User-Agent": self._user_agent,
}
def _response_data(self, response: Any, operation: str) -> Any:
"""
解析兼容层统一响应信封并返回业务数据。
:param response: 请求返回的响应对象
:param operation: 错误信息中的操作名称
:return: 成功时返回响应 data，失败时返回 None
"""
if response is None:
logger.warning(f"{self._name} {operation}失败：站点无响应")
return None
status_code = getattr(response, "status_code", 0)
try:
body = response.json()
except Exception:
logger.warning(f"{self._name} {operation}失败：响应不是 JSON（状态码 {status_code}）")
return None
if not isinstance(body, dict):
logger.warning(f"{self._name} {operation}失败：响应结构异常")
return None
code = body.get("code")
if code != 0:
logger.warning(f"{self._name} {operation}失败：接口返回 {code} {body.get('message') or ''}")
return None
return body.get("data")
def _build_params(
self,
keyword: Optional[str],
page: Optional[int],
) -> Dict[str, Any]:
"""
构造兼容层搜索参数。
兼容层的分类筛选只接受单个站内分类 ID，而 MoviePilot 的媒体类型与分类
筛选都是站点自定义口径，因此这里只按关键字与分页检索，分类交由结果的
分类字段表达，避免用错口径的分类 ID 把结果筛空。
:param keyword: 搜索关键字
:param page: MoviePilot 从 0 开始的页码
:return: 兼容层查询参数
"""
params: Dict[str, Any] = {
"page": (page or 0) + 1,
"pagesize": self._size,
}
if keyword:
params["keyword"] = keyword
return params
@staticmethod
def _media_type(item: Dict[str, Any]) -> str:
"""
将兼容层种子分类映射为 MoviePilot 媒体类型。
:param item: 兼容层返回的种子记录
:return: MoviePilot 媒体类型，未识别时为未知类型
"""
newznab_id = item.get("category_newznab")
media_type = None
if isinstance(newznab_id, int):
media_type = _NEWZNAB_MEDIA_TYPES.get(newznab_id)
if media_type is None:
media_type = _CATEGORY_NAME_MEDIA_TYPES.get(str(item.get("category_name") or "").strip())
return (media_type or MediaType.UNKNOWN).value
@staticmethod
def _volume_factors(promotion: Optional[str]) -> Tuple[float, float]:
"""
将兼容层促销档位换算为下载、上传量系数。
:param promotion: 兼容层返回的促销档位
:return: 下载量系数与上传量系数
"""
return _PROMOTION_FACTORS.get(str(promotion or "").strip(), (1.0, 1.0))
def _build_download_url(self, torrent_id: object) -> str:
"""
构造种子文件下载地址。
兼容层下载端点接受开放 API 令牌作为查询参数，下载器取种子时不会携带
请求头，因此令牌必须落在地址里。
:param torrent_id: 兼容层返回的种子 ID
:return: 兼容层下载接口地址
"""
return f"{self._base_url}/download.php?id={torrent_id}&apikey={self._api_token}"
def _parse_result(self, items: List[Any]) -> List[Dict[str, Any]]:
"""
将兼容层种子数据转换为 MoviePilot 标准字段。
:param items: 兼容层 data.items 数组
:return: MoviePilot 标准种子字典列表
"""
torrents: List[Dict[str, Any]] = []
for item in items or []:
if not isinstance(item, dict):
continue
torrent_id = item.get("id")
download_factor, upload_factor = self._volume_factors(item.get("promotion"))
labels: List[str] = []
promotion_name = str(item.get("promotion_name") or "").strip()
if promotion_name:
labels.append(promotion_name)
if item.get("official") is True:
labels.append("官方")
torrents.append({
"title": item.get("name"),
"description": item.get("small_descr"),
"enclosure": self._build_download_url(torrent_id),
"pubdate": time_tools.format_timestamp(item.get("added")),
"size": int(item.get("size") or 0),
"seeders": int(item.get("seeders") or 0),
"peers": int(item.get("leechers") or 0),
"grabs": int(item.get("completed") or 0),
"downloadvolumefactor": download_factor,
"uploadvolumefactor": upload_factor,
"page_url": f"{self._site_url}/torrent/{torrent_id}",
"labels": labels,
"category": self._media_type(item),
})
return torrents
def _process_search_response(self, response: Any) -> Tuple[bool, List[Dict[str, Any]]]:
"""
处理兼容层种子搜索响应。
:param response: 请求返回的响应对象
: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],
page: Optional[int] = 0,
) -> Tuple[bool, List[Dict[str, Any]]]:
"""
同步搜索 FluxTorrent 站点种子。
:param keyword: 搜索关键字
:param page: MoviePilot 从 0 开始的页码
:return: 是否失败及标准种子列表
"""
if not self._api_token:
logger.warning(f"{self._name} 未配置开放 API 令牌")
return True, []
params = self._build_params(keyword, page)
response = RequestUtils(
headers=self._request_headers(),
proxies=self._proxy,
timeout=self._timeout,
).get_res(url=f"{self._base_url}/torrents.json", params=params)
return self._process_search_response(response)
async def async_search(
self,
keyword: Optional[str],
page: Optional[int] = 0,
) -> Tuple[bool, List[Dict[str, Any]]]:
"""
异步搜索 FluxTorrent 站点种子。
:param keyword: 搜索关键字
:param page: MoviePilot 从 0 开始的页码
:return: 是否失败及标准种子列表
"""
if not self._api_token:
logger.warning(f"{self._name} 未配置开放 API 令牌")
return True, []
params = self._build_params(keyword, page)
response = await AsyncRequestUtils(
headers=self._request_headers(),
proxies=self._proxy,
timeout=self._timeout,
).get_res(url=f"{self._base_url}/torrents.json", params=params)
return self._process_search_response(response)

4
docs/architecture-overview.md
View File

@@ -824,8 +824,8 @@ flowchart LR
| 指标 | 当前值 |
|---|---:|
| Python 模块 | 1151 |
| 内部导入边 | 9,973 |
| Python 模块 | 1153 |
| 内部导入边 | 9,989 |
| 非平凡 SCC | 1（29 个模块的 TMDB vendor 精确 containment） |
| Application / Chain 具体 Adapter 直连 | 0 / 0 |
| Direct egress | 51（精确 policy 准入，Application / Chain 均为 0） |

24
tests/fixtures/architecture/dependency-baseline.json vendored
View File

@@ -962,8 +962,8 @@
"runtime_only": true
}
},
"edge_count": 9973,
"edge_sha256": "176287ee92a7a64f9f1ded862ad7b408f426fae401e5ecdc26dfa1811bfd4494",
"edge_count": 9989,
"edge_sha256": "49d778c3ba091205a2f6d3ed7de6dac11ff3dc112e06e16e48b50279b220f366",
"edges": [
"app -> app.foundation",
"app -> app.foundation.environment",
@@ -7839,6 +7839,7 @@
"app.modules.indexer -> app.modules",
"app.modules.indexer -> app.modules.indexer.parser",
"app.modules.indexer -> app.modules.indexer.spider",
"app.modules.indexer -> app.modules.indexer.spider.fluxtorrent",
"app.modules.indexer -> app.modules.indexer.spider.haidan",
"app.modules.indexer -> app.modules.indexer.spider.hddolby",
"app.modules.indexer -> app.modules.indexer.spider.milkie",
@@ -7888,6 +7889,11 @@
"app.modules.indexer.parser.file_list -> app.modules",
"app.modules.indexer.parser.file_list -> app.modules.indexer",
"app.modules.indexer.parser.file_list -> app.modules.indexer.parser",
"app.modules.indexer.parser.fluxtorrent -> app.modules",
"app.modules.indexer.parser.fluxtorrent -> app.modules.indexer",
"app.modules.indexer.parser.fluxtorrent -> app.modules.indexer.parser",
"app.modules.indexer.parser.fluxtorrent -> app.runtime",
"app.modules.indexer.parser.fluxtorrent -> app.runtime.log",
"app.modules.indexer.parser.gazelle -> app.foundation",
"app.modules.indexer.parser.gazelle -> app.foundation.dom",
"app.modules.indexer.parser.gazelle -> app.foundation.size",
@@ -8042,6 +8048,16 @@
"app.modules.indexer.spider -> app.runtime.settings",
"app.modules.indexer.spider -> app.schemas",
"app.modules.indexer.spider -> app.schemas.types",
"app.modules.indexer.spider.fluxtorrent -> app.adapters",
"app.modules.indexer.spider.fluxtorrent -> app.adapters.network",
"app.modules.indexer.spider.fluxtorrent -> app.adapters.network.http",
"app.modules.indexer.spider.fluxtorrent -> app.foundation",
"app.modules.indexer.spider.fluxtorrent -> app.foundation.temporal",
"app.modules.indexer.spider.fluxtorrent -> app.runtime",
"app.modules.indexer.spider.fluxtorrent -> app.runtime.log",
"app.modules.indexer.spider.fluxtorrent -> app.runtime.settings",
"app.modules.indexer.spider.fluxtorrent -> app.schemas",
"app.modules.indexer.spider.fluxtorrent -> app.schemas.types",
"app.modules.indexer.spider.haidan -> app.adapters",
"app.modules.indexer.spider.haidan -> app.adapters.network",
"app.modules.indexer.spider.haidan -> app.adapters.network.http",
@@ -10939,7 +10955,7 @@
"app.workflow.actions.transfer_file -> app.workflow",
"app.workflow.actions.transfer_file -> app.workflow.actions"
],
"module_count": 1151,
"module_count": 1153,
"modules": [
"app",
"app.adapters",
@@ -11723,6 +11739,7 @@
"app.modules.indexer.parser.bitpt",
"app.modules.indexer.parser.discuz",
"app.modules.indexer.parser.file_list",
"app.modules.indexer.parser.fluxtorrent",
"app.modules.indexer.parser.gazelle",
"app.modules.indexer.parser.hddolby",
"app.modules.indexer.parser.ipt",
@@ -11741,6 +11758,7 @@
"app.modules.indexer.parser.yema",
"app.modules.indexer.parser.zhixing",
"app.modules.indexer.spider",
"app.modules.indexer.spider.fluxtorrent",
"app.modules.indexer.spider.haidan",
"app.modules.indexer.spider.hddolby",
"app.modules.indexer.spider.milkie",

477
tests/test_flux_torrent_indexer.py Normal file
View File

@@ -0,0 +1,477 @@
"""
验证 FluxTorrent 兼容层站点的搜索、用户数据解析与站点资源协议路由。
"""
import asyncio
import json
import app.modules.indexer as indexer_module
from app.foundation import temporal as time_tools
from app.modules.indexer import IndexerModule
from app.modules.indexer.parser import SiteSchema
from app.modules.indexer.parser.fluxtorrent import FluxTorrentSiteUserInfo
from app.modules.indexer.spider import fluxtorrent as flux_torrent_module
from app.modules.indexer.spider.fluxtorrent import FluxTorrentSpider
from app.schemas.types import MediaType
_SITE_URL = "https://flux.example/"
_API_TOKEN = "fxo_test_token"
class _FakeResponse:
"""
提供 FluxTorrent 兼容层响应判定所需的最小 HTTP 响应契约。
"""
def __init__(self, payload=None, status_code: int = 200):
"""
保存固定状态、JSON 数据和响应文本。
"""
self._payload = payload
self.status_code = status_code
self.reason = "OK"
self.text = json.dumps(payload) if payload is not None else ""
def __bool__(self) -> bool:
"""
按 requests.Response 的成功状态布尔语义返回结果。
"""
return self.status_code < 400
def json(self):
"""
返回预设 JSON 数据。
"""
return self._payload
def _build_indexer(**overrides) -> dict:
"""
构造 FluxTorrent 站点索引配置。
"""
indexer = {
"id": "flux",
"name": "FluxTorrent",
"url": _SITE_URL,
"domain": _SITE_URL,
"apikey": _API_TOKEN,
"ua": "MoviePilot-Test",
"proxy": False,
}
indexer.update(overrides)
return indexer
def _envelope(data, code: int = 0, message: str = "ok") -> dict:
"""
构造 FluxTorrent 统一响应信封。
"""
return {
"code": code,
"message": message,
"data": data,
"request_id": "req-1",
}
def _torrent_items() -> list:
"""
构造兼容层种子列表数据。
"""
return [
{
"id": 123,
"name": "Example.Movie.2026.2160p.WEB-DL.H.265-GROUP",
"small_descr": "电影中文副标题",
"seeders": 15,
"leechers": 2,
"completed": 30,
"size": 21474836480,
"added": 1753084800,
"category": 1,
"category_name": "电影",
"category_newznab": 2000,
"promotion": "free",
"promotion_name": "Free",
"official": True,
"sticky": False,
}
]
def test_flux_torrent_spider_maps_compat_items(monkeypatch):
"""
兼容层种子应映射为标准字段，并按促销档位给出计费倍率。
"""
captured = {}
def fake_get_res(request, url: str, params: dict = None, **_kwargs):
"""
回放兼容层种子列表响应并记录请求参数。
"""
captured["url"] = url
captured["params"] = params
captured["headers"] = request._headers
return _FakeResponse(_envelope({"items": _torrent_items()}))
monkeypatch.setattr("app.modules.indexer.spider.fluxtorrent.RequestUtils.get_res", fake_get_res)
error, torrents = FluxTorrentSpider(_build_indexer()).search(keyword="Example", page=0)
assert not error
assert captured["url"] == "https://flux.example/api/v1/compat/nexusphp/torrents.json"
assert captured["params"] == {"page": 1, "pagesize": 50, "keyword": "Example"}
assert captured["headers"]["Authorization"] == f"Token {_API_TOKEN}"
assert torrents == [{
"title": "Example.Movie.2026.2160p.WEB-DL.H.265-GROUP",
"description": "电影中文副标题",
"enclosure": (
"https://flux.example/api/v1/compat/nexusphp/download.php"
f"?id=123&apikey={_API_TOKEN}"
),
# 兼容层给的是 Unix 时间戳，发布时间按运行环境时区渲染，这里对同一转换取期望值。
"pubdate": time_tools.format_timestamp(1753084800),
"size": 21474836480,
"seeders": 15,
"peers": 2,
"grabs": 30,
"downloadvolumefactor": 0.0,
"uploadvolumefactor": 1.0,
"page_url": "https://flux.example/torrent/123",
"labels": ["Free", "官方"],
"category": MediaType.MOVIE.value,
}]
def test_flux_torrent_spider_maps_promotion_factors():
"""
促销档位应换算为站点计费口径的下载与上传倍率。
"""
assert FluxTorrentSpider._volume_factors("free") == (0.0, 1.0)
assert FluxTorrentSpider._volume_factors("x2free") == (0.0, 2.0)
assert FluxTorrentSpider._volume_factors("half") == (0.5, 1.0)
assert FluxTorrentSpider._volume_factors("x2half") == (0.5, 2.0)
assert FluxTorrentSpider._volume_factors("x2") == (1.0, 2.0)
assert FluxTorrentSpider._volume_factors("p30") == (0.3, 1.0)
assert FluxTorrentSpider._volume_factors(None) == (1.0, 1.0)
assert FluxTorrentSpider._volume_factors("unknown") == (1.0, 1.0)
def test_flux_torrent_spider_maps_media_types():
"""
分类应按 Newznab 标准号映射媒体类型，缺标准号时按分类名兜底。
"""
assert FluxTorrentSpider._media_type({"category_newznab": 2000}) == MediaType.MOVIE.value
assert FluxTorrentSpider._media_type({"category_newznab": 5000}) == MediaType.TV.value
assert FluxTorrentSpider._media_type({"category_newznab": 5070}) == MediaType.TV.value
assert FluxTorrentSpider._media_type({"category_newznab": 3000}) == MediaType.MUSIC.value
assert FluxTorrentSpider._media_type({"category_newznab": 7020}) == MediaType.COLLECTION.value
assert FluxTorrentSpider._media_type({"category_newznab": 4000}) == MediaType.UNKNOWN.value
assert FluxTorrentSpider._media_type({"category_name": "短剧"}) == MediaType.TV.value
assert FluxTorrentSpider._media_type({"category_name": "电子书"}) == MediaType.COLLECTION.value
assert FluxTorrentSpider._media_type({}) == MediaType.UNKNOWN.value
def test_flux_torrent_spider_requires_api_token(monkeypatch):
"""
未配置开放 API 令牌时不应发起搜索请求。
"""
def fake_get_res(*_args, **_kwargs):
"""
记录不应发生的请求。
"""
raise AssertionError("未配置令牌时不应发起请求")
monkeypatch.setattr("app.modules.indexer.spider.fluxtorrent.RequestUtils.get_res", fake_get_res)
error, torrents = FluxTorrentSpider(_build_indexer(apikey="")).search(keyword="Example")
assert error
assert torrents == []
def test_flux_torrent_spider_reports_business_error(monkeypatch):
"""
兼容层返回业务错误码时应判定搜索失败。
"""
monkeypatch.setattr(
"app.modules.indexer.spider.fluxtorrent.RequestUtils.get_res",
lambda *_args, **_kwargs: _FakeResponse(
_envelope(None, code=2001, message="令牌无效"), status_code=401
),
)
error, torrents = FluxTorrentSpider(_build_indexer()).search(keyword="Example")
assert error
assert torrents == []
def test_flux_torrent_spider_sync_async_parity(monkeypatch):
"""
同步与异步搜索入口应返回一致结果。
"""
def sync_get_res(_request, url: str, params: dict = None, **_kwargs):
"""
回放兼容层种子列表响应。
"""
return _FakeResponse(_envelope({"items": _torrent_items()}))
async def async_get_res(_request, url: str, params: dict = None, **_kwargs):
"""
回放兼容层种子列表响应。
"""
return _FakeResponse(_envelope({"items": _torrent_items()}))
monkeypatch.setattr("app.modules.indexer.spider.fluxtorrent.RequestUtils.get_res", sync_get_res)
monkeypatch.setattr(
"app.modules.indexer.spider.fluxtorrent.AsyncRequestUtils.get_res", async_get_res
)
spider = FluxTorrentSpider(_build_indexer())
assert spider.search(keyword="Example", page=0) == asyncio.run(
spider.async_search(keyword="Example", page=0)
)
def test_flux_torrent_parser_reads_user_profile(monkeypatch):
"""
兼容层用户接口应解析出账号资料与流量统计。
"""
payload = _envelope({
"id": 42,
"username": "tester",
"uploaded": 21474836480,
"downloaded": 1073741824,
"seedbonus": 1024,
"class": 9,
"class_name": "VIP",
"ratio": 20.0,
"ratio_infinite": False,
"ratio_display": "20.0000",
})
monkeypatch.setattr(
FluxTorrentSiteUserInfo,
"_get_page_content",
lambda self, url, **_kwargs: json.dumps(payload),
)
parser = FluxTorrentSiteUserInfo(
site_name="FluxTorrent",
url=_SITE_URL,
site_cookie=None,
apikey=_API_TOKEN,
token=None,
ua="MoviePilot-Test",
)
parser.parse()
assert parser.userid == 42
assert parser.username == "tester"
assert parser.user_level == "VIP"
assert parser.upload == 21474836480
assert parser.download == 1073741824
assert parser.ratio == 20.0
assert parser.bonus == 1024.0
assert parser.err_msg is None
def test_flux_torrent_parser_handles_infinite_ratio(monkeypatch):
"""
只上传未下载时站点标记分享率为无穷大，此时不应给出有限数值。
"""
payload = _envelope({
"id": 42,
"username": "tester",
"uploaded": 21474836480,
"downloaded": 0,
"seedbonus": 0,
"class": 1,
"class_name": "普通用户",
"ratio": 0.0,
"ratio_infinite": True,
})
monkeypatch.setattr(
FluxTorrentSiteUserInfo,
"_get_page_content",
lambda self, url, **_kwargs: json.dumps(payload),
)
parser = FluxTorrentSiteUserInfo(
site_name="FluxTorrent",
url=_SITE_URL,
site_cookie=None,
apikey=_API_TOKEN,
token=None,
ua="MoviePilot-Test",
)
parser.parse()
assert parser.userid == 42
assert parser.ratio == 0.0
def test_flux_torrent_parser_reports_business_error(monkeypatch):
"""
兼容层返回业务错误码时应给出错误信息且不产出用户 ID。
"""
monkeypatch.setattr(
FluxTorrentSiteUserInfo,
"_get_page_content",
lambda self, url, **_kwargs: json.dumps(
_envelope(None, code=2001, message="令牌无效或已撤销")
),
)
parser = FluxTorrentSiteUserInfo(
site_name="FluxTorrent",
url=_SITE_URL,
site_cookie=None,
apikey=_API_TOKEN,
token=None,
ua="MoviePilot-Test",
)
parser.parse()
assert parser.userid is None
assert parser.err_msg == "令牌无效或已撤销"
def test_flux_torrent_registrations():
"""
FluxTorrent 应登记为专站搜索协议并声明支持的搜索参数。
"""
assert SiteSchema.FluxTorrent.value == "FluxTorrent"
assert indexer_module.SPIDER_PARSER_CLASSES["FluxTorrent"] is FluxTorrentSpider
assert indexer_module._SPECIALIZED_SEARCH_ARGUMENTS["FluxTorrent"] == ("keyword", "page")
assert FluxTorrentSpider.get_search_page_size() == 50
def test_flux_torrent_routes_by_declared_parser():
"""
站点资源声明 FluxTorrent 后，搜索应直接路由到兼容层索引器。
"""
assert IndexerModule.get_search_page_size(_build_indexer(parser="FluxTorrent")) == 50
request = IndexerModule._IndexerModule__create_search_request(
site=_build_indexer(parser="FluxTorrent"),
keyword="Example",
mtype=MediaType.MOVIE,
page=1,
)
assert request is not None
assert request.parser_class is FluxTorrentSpider
assert dict(request.arguments) == {"keyword": "Example", "page": 1}
def test_refresh_userdata_uses_declared_flux_torrent_schema(monkeypatch):
"""
站点资源声明 FluxTorrent 用户数据模型后，刷新应直接使用兼容层解析器。
"""
payload = _envelope({
"id": 42,
"username": "tester",
"uploaded": 1024,
"downloaded": 512,
"seedbonus": 10,
"class": 3,
"class_name": "VIP",
"ratio": 2.0,
"ratio_infinite": False,
})
monkeypatch.setattr(
FluxTorrentSiteUserInfo,
"_get_page_content",
lambda self, url, **_kwargs: json.dumps(payload),
)
monkeypatch.setattr(IndexerModule, "_site_schemas", [FluxTorrentSiteUserInfo])
module = object.__new__(IndexerModule)
result = module.refresh_userdata({
"name": "FluxTorrent",
"url": _SITE_URL,
"schema": SiteSchema.FluxTorrent.value,
"apikey": _API_TOKEN,
"public": False,
})
assert result.userid == 42
assert result.username == "tester"
assert result.user_level == "VIP"
assert result.ratio == 2.0
class _BrokenResponse:
"""
提供响应体不是 JSON 时的最小 HTTP 响应契约。
"""
status_code = 200
def __bool__(self) -> bool:
"""
按 requests.Response 的成功状态布尔语义返回结果。
"""
return True
def json(self):
"""
模拟响应体解析失败。
"""
raise ValueError("not json")
def test_flux_torrent_site_base_url_normalizes_address():
"""
站点地址缺少协议时补 https，末尾斜杠应去掉，空地址返回空串。
"""
assert flux_torrent_module.site_base_url({"url": "https://flux.example/"}) == "https://flux.example"
assert flux_torrent_module.site_base_url({"url": "flux.example/"}) == "https://flux.example"
assert flux_torrent_module.site_base_url({"url": " "}) == ""
assert flux_torrent_module.site_api_token({"apikey": " fxo_a "}) == "fxo_a"
def test_flux_torrent_spider_tolerates_unusable_responses(monkeypatch):
"""
站点无响应或响应不是 JSON 时应判定为失败而不是抛异常。
"""
spider = FluxTorrentSpider(_build_indexer())
monkeypatch.setattr(
"app.modules.indexer.spider.fluxtorrent.RequestUtils.get_res",
lambda *_args, **_kwargs: None,
)
assert spider.search(keyword="Example", page=0) == (True, [])
monkeypatch.setattr(
"app.modules.indexer.spider.fluxtorrent.RequestUtils.get_res",
lambda *_args, **_kwargs: _BrokenResponse(),
)
assert spider.search(keyword="Example", page=0) == (True, [])
def test_flux_torrent_async_search_requires_api_token():
"""
异步搜索在没有开放 API 令牌时应直接判定失败。
"""
spider = FluxTorrentSpider(_build_indexer(apikey="", token=""))
assert asyncio.run(spider.async_search(keyword="Example", page=0)) == (True, [])
def test_flux_torrent_parser_rejects_unusable_payloads():
"""
用户接口返回空响应、非 JSON、结构异常或业务错误码时都应给出错误信息。
"""
parser = FluxTorrentSiteUserInfo(
site_name="FluxTorrent",
url=_SITE_URL,
site_cookie=None,
apikey=_API_TOKEN,
token=None,
ua="",
)
assert parser._load_api_payload("", "获取用户信息") is None
assert "未收到接口响应" in parser.err_msg
assert parser._load_api_payload("not-json", "获取用户信息") is None
assert "不是有效 JSON" in parser.err_msg
assert parser._load_api_payload("[1, 2]", "获取用户信息") is None
assert "结构无效" in parser.err_msg
payload = _envelope(None, code=2001, message="invalid token")
assert parser._load_api_payload(json.dumps(payload), "获取用户信息") is None
assert parser.err_msg == "invalid token"