GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-09-08 17:08:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: 支持站点音乐分类搜索

Browse Source
This commit is contained in:
jxxghp
2026-08-13 19:45:09 +08:00
parent c86307ddd1
commit 23a90991fc
14 changed files with 464 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/modules/indexer/__init__.py
+2
View File
@@ -252,6 +252,7 @@ class IndexerModule(_ModuleBase):
elif site.get('parser') == "TorrentLeech":
error_flag, result = TorrentLeech(site).search(
keyword=search_word,
mtype=mtype,
page=page
)
elif site.get('parser') == "mTorrent":
@@ -395,6 +396,7 @@ class IndexerModule(_ModuleBase):
elif site.get('parser') == "TorrentLeech":
error_flag, result = await TorrentLeech(site).async_search(
keyword=search_word,
mtype=mtype,
page=page
)
elif site.get('parser') == "mTorrent":
app/modules/indexer/spider/__init__.py
+32 -2
View File
@@ -121,6 +121,10 @@ class SiteSpider:
self.list = self.browse.get('list') or self.list
self.fields = self.browse.get('fields') or self.fields
result_num = indexer.get('result_num')
self.result_media_type_from_request = (
(self.search or {}).get("result_media_type") == "requested"
)
self.requested_result_media_type = None
self._field_templates = self.__build_field_templates()
self.domain = indexer.get('domain')
self.result_num = int(result_num or self.default_result_num())
@@ -167,6 +171,7 @@ class SiteSpider:
torrentspath = ""
# 是否选中了媒体类型专用路径，浏览模式下专用路径优先于 browse 配置
typed_path_selected = False
category_filter_selected = False
if len(paths) == 1:
torrentspath = paths[0].get('path', '')
else:
@@ -243,6 +248,7 @@ class SiteSpider:
if allowed_cats and str(cat.get('id')) not in allowed_cats:
continue
if self.category.get("field"):
category_filter_selected = True
value = params.get(self.category.get("field"), "")
params.update({
"%s" % self.category.get("field"): value + self.category.get("delimiter",
@@ -251,6 +257,7 @@ class SiteSpider:
else:
category_param = cat.get("param") or self.category.get("param")
if category_param:
category_filter_selected = True
# 某些站点（例如憨憨）使用重复的 cat[] 参数，字典值列表可由
# UrlUtils.combine_url 以 doseq=True 正确展开。
category_id = cat.get("value", cat.get("id"))
@@ -262,6 +269,7 @@ class SiteSpider:
else:
params[category_param] = [current_value, category_id]
else:
category_filter_selected = True
params.update({
"cat%s" % cat.get("id"): 1
})
@@ -300,6 +308,12 @@ class SiteSpider:
# 搜索Url
searchurl = self.domain + str(torrentspath).format(**inputs_dict)
if self.result_media_type_from_request \
and self.mtype \
and (typed_path_selected or category_filter_selected):
self.requested_result_media_type = self.mtype
else:
self.requested_result_media_type = None
return searchurl
def __format_search_word(self, search_word: str) -> str:
@@ -774,6 +788,9 @@ class SiteSpider:
def __get_category(self, torrent: Any):
# category 电影/电视剧/音乐
if self.requested_result_media_type:
self.torrents_info['category'] = self.requested_result_media_type.value
return
if 'category' not in self.fields:
if self.site_media_type:
self.torrents_info['category'] = self.site_media_type.value
@@ -786,6 +803,19 @@ class SiteSpider:
resolved_type = self.site_media_type
self.torrents_info['category'] = resolved_type.value
def __apply_requested_result_media_type(self, torrents: Optional[List[dict]]) -> List[dict]:
"""
为已由站点查询条件约束类型的结果补充统一媒体类型。
仅在站点配置显式声明 result_media_type=requested 时生效，避免把普通混合搜索结果误分类。
"""
results = torrents or []
if not self.requested_result_media_type:
return results
for torrent in results:
torrent["category"] = self.requested_result_media_type.value
return results
def __get_subtitle_field(self, torrent: Any, field_name: str):
"""
按配置读取字幕字段。
@@ -1080,7 +1110,7 @@ class SiteSpider:
result_num=self.result_num
)
if rust_torrents is not None:
return rust_torrents
return self.__apply_requested_result_media_type(rust_torrents)
# 清空旧结果
self.torrents_info_array = []
@@ -1111,7 +1141,7 @@ class SiteSpider:
torrent_query.clear()
del torrent_query
# 返回数组的副本，防止被后续清理操作影响
return self.torrents_info_array.copy()
return self.__apply_requested_result_media_type(self.torrents_info_array.copy())
except Exception as err:
self.is_error = True
logger.warn(f"错误：{self.indexername} {str(err)}")
app/modules/indexer/spider/haidan.py
+2 -2
View File
@@ -28,8 +28,8 @@ class HaiDanSpider:
# 电影分类
_movie_category = ['401', '404', '405']
_tv_category = ['402', '403', '404', '405']
# 音乐分类：408 为海胆 HQ Audio(音乐) 分区；406 MV 属于视频，不计入音乐
_music_category = ['408']
# 音乐搜索同时覆盖高品质音频和音乐视频/演唱会分区。
_music_category = ['406', '408']
# 足销状态 1-普通，2-免费，3-2X，4-2X免费，5-50%，6-2X50%，7-30%
_dl_state = {
app/modules/indexer/spider/hddolby.py
+2 -2
View File
@@ -28,8 +28,8 @@ class HddolbySpider:
# 分类
_movie_category = [401, 405]
_tv_category = [402, 403, 404, 405]
# 音乐分类：408 为杜比音乐唱片分区；406 演唱会蓝光属于视频，不计入音乐
_music_category = [408]
# 音乐搜索同时覆盖高品质音频和音乐视频/演唱会分区。
_music_category = [406, 408]
# 标签
_labels = {
app/modules/indexer/spider/mtorrent.py
+2 -2
View File
@@ -36,8 +36,8 @@ class MTorrentSpider:
_movie_category = ['401', '419', '420', '421', '439', '405', '404']
# 电视剧分类
_tv_category = ['403', '402', '435', '438', '404', '405']
# 音乐分类：406 为馒头 Music(音樂) 分区
_music_category = ['406']
# 音乐分类：434 为无损音乐，406 为演唱分区
_music_category = ['434', '406']
# API KEY
_apikey = None
app/modules/indexer/spider/torrentleech.py
+55 -7
View File
@@ -3,11 +3,14 @@ from urllib.parse import quote
from app.core.config import settings
from app.log import logger
from app.schemas import MediaType
from app.utils.http import RequestUtils, AsyncRequestUtils
from app.utils.string import StringUtils
class TorrentLeech:
"""TorrentLeech JSON 搜索接口索引器。"""
_indexer = None
_proxy = None
_size = 100
@@ -25,12 +28,40 @@ class TorrentLeech:
return None if keyword else cls._size
def __init__(self, indexer: dict):
"""初始化站点认证信息和媒体分类配置。"""
self._indexer = indexer
if indexer.get('proxy'):
self._proxy = settings.PROXY
self._timeout = indexer.get('timeout') or 15
def __parse_result(self, results: List[dict]) -> List[dict]:
def __category_ids(self, mtype: MediaType = None) -> List[str]:
"""读取资源包中与请求媒体类型对应的 TorrentLeech 分类 ID。"""
category_key = {
MediaType.MOVIE: "movie",
MediaType.TV: "tv",
MediaType.MUSIC: "music",
}.get(mtype)
if not category_key:
return []
return [
str(item.get("id"))
for item in ((self._indexer.get("category") or {}).get(category_key) or [])
if isinstance(item, dict) and item.get("id") is not None
]
def __get_search_url(self, keyword: str, mtype: MediaType = None) -> str:
"""按媒体类型构造 TorrentLeech 搜索接口地址。"""
domain = self._indexer.get('domain')
encoded_keyword = quote(keyword)
category_ids = self.__category_ids(mtype)
if category_ids:
return (
f"{domain}torrents/browse/list/categories/{','.join(category_ids)}"
f"/exact/1/query/{encoded_keyword}"
)
return self._searchurl % (domain, encoded_keyword)
def __parse_result(self, results: List[dict], mtype: MediaType = None) -> List[dict]:
"""
解析搜索结果
"""
@@ -38,6 +69,7 @@ class TorrentLeech:
if not results:
return torrents
requested_category_ids = set(self.__category_ids(mtype))
for result in results:
torrent = {
'title': result.get('name'),
@@ -54,10 +86,21 @@ class TorrentLeech:
'page_url': self._pageurl % (self._indexer.get('domain'), result.get('fid')),
'imdbid': result.get('imdbID')
}
if requested_category_ids:
torrent['category'] = (
mtype.value
if str(result.get('categoryID')) in requested_category_ids
else MediaType.UNKNOWN.value
)
torrents.append(torrent)
return torrents
def search(self, keyword: str, page: Optional[int] = 0) -> Tuple[bool, List[dict]]:
def search(
self,
keyword: str,
mtype: MediaType = None,
page: Optional[int] = 0,
) -> Tuple[bool, List[dict]]:
"""
搜索种子
"""
@@ -66,7 +109,7 @@ class TorrentLeech:
return True, []
if keyword:
url = self._searchurl % (self._indexer.get('domain'), quote(keyword))
url = self.__get_search_url(keyword, mtype)
else:
url = self._browseurl % (self._indexer.get('domain'), int(page) + 1)
@@ -81,7 +124,7 @@ class TorrentLeech:
).get_res(url)
if res and res.status_code == 200:
results = res.json().get('torrentList') or []
return False, self.__parse_result(results)
return False, self.__parse_result(results, mtype)
elif res is not None:
logger.warn(f"{self._indexer.get('name')} 搜索失败，错误码：{res.status_code}")
return True, []
@@ -89,7 +132,12 @@ class TorrentLeech:
logger.warn(f"{self._indexer.get('name')} 搜索失败，无法连接 {self._indexer.get('domain')}")
return True, []
async def async_search(self, keyword: str, page: Optional[int] = 0) -> Tuple[bool, List[dict]]:
async def async_search(
self,
keyword: str,
mtype: MediaType = None,
page: Optional[int] = 0,
) -> Tuple[bool, List[dict]]:
"""
异步搜索种子
"""
@@ -98,7 +146,7 @@ class TorrentLeech:
return True, []
if keyword:
url = self._searchurl % (self._indexer.get('domain'), quote(keyword))
url = self.__get_search_url(keyword, mtype)
else:
url = self._browseurl % (self._indexer.get('domain'), int(page) + 1)
@@ -113,7 +161,7 @@ class TorrentLeech:
).get_res(url)
if res and res.status_code == 200:
results = res.json().get('torrentList') or []
return False, self.__parse_result(results)
return False, self.__parse_result(results, mtype)
elif res is not None:
logger.warn(f"{self._indexer.get('name')} 搜索失败，错误码：{res.status_code}")
return True, []
app/modules/indexer/spider/yema.py
+65 -13
View File
@@ -17,7 +17,8 @@ class YemaSpider:
# YemaPT 开放 API 使用更大的分页容量时会返回空结果。
_size = 40
_movie_category = [4]
_tv_category = [5, 6, 13, 14, 15, 16, 17]
_tv_category = [5, 6, 13, 14, 15, 17]
_music_category = [8, 16]
_labels = {
"1": "禁转",
@@ -78,12 +79,14 @@ class YemaSpider:
self,
keyword: Optional[str],
page: Optional[int],
category_id: Optional[int] = None,
) -> dict:
"""
构造公开种子列表查询参数
:param keyword: 搜索关键字
:param page: MoviePilot 从 0 开始的页码
:param category_id: 可选的站点分类 ID
:return: YemaPT 开放 API 请求体
"""
params = {
@@ -95,8 +98,39 @@ class YemaSpider:
}
if keyword:
params["keyword"] = keyword
if category_id is not None:
params["categoryId"] = category_id
return params
def _search_category_ids(self, mtype: MediaType = None) -> List[Optional[int]]:
"""
返回当前媒体类型需要分别查询的站点分类。
YemaPT 分类参数不支持数组，音乐需要分别查询音频音乐和 MV/演唱会分类。
"""
if mtype == MediaType.MOVIE:
return list(self._movie_category)
if mtype == MediaType.TV:
return list(self._tv_category)
if mtype == MediaType.MUSIC:
return list(self._music_category)
return [None]
@staticmethod
def _merge_search_results(result_groups: List[List[dict]]) -> List[dict]:
"""按种子详情链接合并多个分类查询结果并保持首次出现顺序。"""
merged = []
seen = set()
for results in result_groups:
for torrent in results:
identity = torrent.get("page_url") or torrent.get("enclosure")
if identity and identity in seen:
continue
if identity:
seen.add(identity)
merged.append(torrent)
return merged
def _parse_result(self, results: List[dict]) -> List[dict]:
"""
将开放 API 种子数据转换为 MoviePilot 标准字段
@@ -113,6 +147,8 @@ class YemaSpider:
category = MediaType.TV.value
elif category_value in self._movie_category:
category = MediaType.MOVIE.value
elif category_value in self._music_category:
category = MediaType.MUSIC.value
else:
category = MediaType.UNKNOWN.value
@@ -181,22 +217,30 @@ class YemaSpider:
同步搜索 YemaPT 公开种子
:param keyword: 搜索关键字
:param mtype: MoviePilot 媒体类型，开放 API 不支持直接按媒体类型查询
:param mtype: MoviePilot 媒体类型，指定后按真实站点分类分别查询
:param page: MoviePilot 从 0 开始的页码
:return: 是否失败及标准种子列表
"""
if not self._api_key:
logger.warning(f"{self._name} 未配置 API AuthKey")
return True, []
response = RequestUtils(
request = RequestUtils(
headers=self._request_headers(),
proxies=self._proxy,
timeout=self._timeout,
).post_res(
url=self._search_url,
json=self._build_params(keyword, page),
)
return self._process_search_response(response)
result_groups = []
errors = []
for category_id in self._search_category_ids(mtype):
response = request.post_res(
url=self._search_url,
json=self._build_params(keyword, page, category_id),
)
error, results = self._process_search_response(response)
errors.append(error)
if not error:
result_groups.append(results)
return all(errors), self._merge_search_results(result_groups)
async def async_search(
self,
@@ -208,22 +252,30 @@ class YemaSpider:
异步搜索 YemaPT 公开种子
:param keyword: 搜索关键字
:param mtype: MoviePilot 媒体类型，开放 API 不支持直接按媒体类型查询
:param mtype: MoviePilot 媒体类型，指定后按真实站点分类分别查询
:param page: MoviePilot 从 0 开始的页码
:return: 是否失败及标准种子列表
"""
if not self._api_key:
logger.warning(f"{self._name} 未配置 API AuthKey")
return True, []
response = await AsyncRequestUtils(
request = AsyncRequestUtils(
headers=self._request_headers(),
proxies=self._proxy,
timeout=self._timeout,
).post_res(
url=self._search_url,
json=self._build_params(keyword, page),
)
return self._process_search_response(response)
result_groups = []
errors = []
for category_id in self._search_category_ids(mtype):
response = await request.post_res(
url=self._search_url,
json=self._build_params(keyword, page, category_id),
)
error, results = self._process_search_response(response)
errors.append(error)
if not error:
result_groups.append(results)
return all(errors), self._merge_search_results(result_groups)
@staticmethod
def _download_factor(promotion: str) -> float:
tests/test_haidan_spider.py
+6 -4
View File
@@ -26,10 +26,10 @@ def haidan_spider(monkeypatch):
def test_music_search_uses_music_categories(haidan_spider):
"""音乐搜索应只提交海胆 HQ Audio 分区分类。"""
"""音乐搜索应提交海胆 HQ Audio 和音乐视频分区。"""
params = haidan_spider._HaiDanSpider__get_params("张学友", MediaType.MUSIC)
assert "cat=408" in params
assert "cat=406%2C408" in params or "cat=406,408" in params
assert "401" not in params
@@ -46,14 +46,16 @@ def test_parse_result_reads_item_category(haidan_spider):
"code": 0,
"data": {
"1": {"name": "张学友 - 他在那里 FLAC", "category": 408, "size": "1024"},
"2": {"name": "流浪地球", "category": 401, "size": "1024"},
"3": {"name": "剧集 S01", "category": 402, "size": "1024"},
"2": {"name": "张学友演唱会", "category": 406, "size": "1024"},
"3": {"name": "流浪地球", "category": 401, "size": "1024"},
"4": {"name": "剧集 S01", "category": 402, "size": "1024"},
},
}
torrents = haidan_spider._HaiDanSpider__parse_result(result)
assert [torrent["category"] for torrent in torrents] == [
MediaType.MUSIC.value,
MediaType.MUSIC.value,
MediaType.MOVIE.value,
MediaType.TV.value,
tests/test_hddolby_spider.py
+3 -3
View File
@@ -26,7 +26,7 @@ def hddolby_spider(monkeypatch):
def test_music_search_uses_music_categories(hddolby_spider):
"""音乐搜索应只提交杜比音乐唱片分区分类。"""
"""音乐搜索应提交杜比高品质音频和音乐视频分区。"""
params = hddolby_spider._HddolbySpider__get_params("张学友", MediaType.MUSIC, 0)
assert params["categories"] == HddolbySpider._music_category
@@ -40,7 +40,7 @@ def test_movie_search_keeps_movie_categories(hddolby_spider):
def test_parse_result_marks_music_torrents(hddolby_spider):
"""音乐唱片分区种子应标记为音乐媒体类型，演唱会视频仍按原逻辑处理。"""
"""高品质音频和音乐视频分区都应标记为音乐媒体类型。"""
results = hddolby_spider._HddolbySpider__parse_result([
{"id": 1, "name": "VA - Chillout 2022 FLAC", "category": 408, "size": 1024},
{"id": 2, "name": "Epica Live 1080p Blu-ray", "category": 406, "size": 1024},
@@ -50,7 +50,7 @@ def test_parse_result_marks_music_torrents(hddolby_spider):
assert [torrent["category"] for torrent in results] == [
MediaType.MUSIC.value,
MediaType.UNKNOWN.value,
MediaType.MUSIC.value,
MediaType.MOVIE.value,
MediaType.TV.value,
]
tests/test_indexer_spider_search_url.py
+24
View File
@@ -225,6 +225,30 @@ def test_typed_search_path_falls_back_to_all_path():
assert parsed_url.path == "/torrents.php"
def test_video_search_uses_dedicated_path_without_synthesized_category_params():
"""专属影视路径已携带分类时，不应再追加站点不支持的默认 cat 参数。"""
indexer = _build_indexer(
domain="https://iptorrents.com/",
search={
"paths": [
{"path": "t?q={keyword}", "type": "all"},
{"path": "t?72&q={keyword}", "type": "movie"},
{"path": "t?73&q={keyword}", "type": "tv"},
],
},
category={
"movie": [{"id": 72, "cat": "Movies"}],
"tv": [{"id": 73, "cat": "TV"}],
},
)
movie_url = _get_search_url(indexer, "Movie 2026", MediaType.MOVIE)
tv_url = _get_search_url(indexer, "Series S01", MediaType.TV)
assert movie_url == "https://iptorrents.com/t?72&q=Movie%202026"
assert tv_url == "https://iptorrents.com/t?73&q=Series%20S01"
def test_music_browse_uses_dedicated_music_entry():
"""
订阅刷新浏览音乐资源时应使用站点的音乐专用入口，而不是默认首页。
tests/test_mtorrent_spider.py
+5 -3
View File
@@ -42,12 +42,14 @@ def test_movie_search_keeps_movie_categories(mteam_spider):
def test_parse_result_marks_music_torrents(mteam_spider):
"""音乐分区种子应标记为音乐媒体类型，供音乐搜索链路筛选。"""
results = mteam_spider._MTorrentSpider__parse_result([
{"id": "1", "name": "周杰伦 - 七里香 [FLAC]", "category": "406", "size": "1024", "status": {}},
{"id": "2", "name": "流浪地球 2160p", "category": "419", "size": "1024", "status": {}},
{"id": "3", "name": "其他资源", "category": "999", "size": "1024", "status": {}},
{"id": "1", "name": "周杰伦 - 七里香 [FLAC]", "category": "434", "size": "1024", "status": {}},
{"id": "2", "name": "周杰伦演唱会", "category": "406", "size": "1024", "status": {}},
{"id": "3", "name": "流浪地球 2160p", "category": "419", "size": "1024", "status": {}},
{"id": "4", "name": "其他资源", "category": "999", "size": "1024", "status": {}},
])
assert [torrent["category"] for torrent in results] == [
MediaType.MUSIC.value,
MediaType.MUSIC.value,
MediaType.MOVIE.value,
MediaType.UNKNOWN.value,
tests/test_music_indexer_category.py
+56
View File
@@ -61,3 +61,59 @@ def test_site_level_music_type_fills_missing_torrent_category():
spider._SiteSpider__get_category(None)
assert spider.torrents_info["category"] == MediaType.MUSIC.value
def test_requested_result_media_type_overrides_unrepresentable_site_category():
"""音乐专属查询使用非主分类筛选时，应按显式契约把结果标记为音乐。"""
spider = SiteSpider(
indexer={
"id": "typed-music",
"name": "Typed Music",
"domain": "https://music.example/",
"search": {
"paths": [
{"path": "torrents?q={keyword}", "type": "all"},
{"path": "music?q={keyword}", "type": "music"},
],
"result_media_type": "requested",
},
"category": {
"movie": [{"id": "1", "name": "电影"}],
"music": [{"id": "20", "name": "音乐规格"}],
},
"torrents": {
"fields": {
"category": {"selector": "a.category"},
}
},
},
mtype=MediaType.MUSIC,
)
search_url = spider._SiteSpider__get_search_url()
spider._SiteSpider__get_category(None)
assert search_url == "https://music.example/music?q="
assert spider.torrents_info["category"] == MediaType.MUSIC.value
def test_requested_result_media_type_requires_an_active_type_filter():
"""未命中专属路径或分类参数时，不得把混合搜索结果强制标记为请求类型。"""
spider = SiteSpider(
indexer={
"id": "mixed",
"name": "Mixed",
"domain": "https://mixed.example/",
"search": {
"paths": [{"path": "torrents?q={keyword}", "type": "all"}],
"result_media_type": "requested",
},
"torrents": {"fields": {}},
},
mtype=MediaType.MOVIE,
)
spider._SiteSpider__get_search_url()
spider._SiteSpider__get_category(None)
assert "category" not in spider.torrents_info
tests/test_torrentleech_music.py
+135
View File
@@ -0,0 +1,135 @@
import asyncio
from app.modules.indexer.spider.torrentleech import TorrentLeech
from app.schemas import MediaType
class _FakeResponse:
"""构造 TorrentLeech 搜索测试使用的最小 JSON 响应。"""
status_code = 200
def json(self) -> dict:
"""返回包含音乐分类种子的固定响应。"""
return {
"torrentList": [{
"fid": 100,
"filename": "album.torrent",
"name": "Artist Album FLAC",
"categoryID": 31,
"addedTimestamp": 1767225600,
"size": 1024,
}]
}
def _build_indexer() -> dict:
"""构造 TorrentLeech 音乐搜索所需的最小站点配置。"""
return {
"id": "torrentleech",
"name": "TorrentLeech",
"domain": "https://www.torrentleech.org/",
"ua": "MoviePilot-Test",
"category": {
"movie": [
{"id": 8, "cat": "Movies"},
{"id": 9, "cat": "Movies"},
],
"tv": [
{"id": 26, "cat": "TV"},
{"id": 32, "cat": "TV"},
],
"music": [
{"id": 31, "cat": "Music"},
{"id": 16, "cat": "Music"},
]
},
}
def test_torrentleech_music_search_filters_categories_and_maps_result(monkeypatch):
"""TorrentLeech 同步音乐搜索应提交 Audio/MV 分类并标记结果类型。"""
captured = {}
def fake_get_res(_request, url: str, **_kwargs):
"""记录同步搜索地址并回放 JSON 响应。"""
captured["url"] = url
return _FakeResponse()
monkeypatch.setattr(
"app.modules.indexer.spider.torrentleech.RequestUtils.get_res",
fake_get_res,
)
error, torrents = TorrentLeech(_build_indexer()).search(
keyword="Artist Album",
mtype=MediaType.MUSIC,
)
assert not error
assert captured["url"] == (
"https://www.torrentleech.org/torrents/browse/list/categories/31,16/"
"exact/1/query/Artist%20Album"
)
assert torrents[0]["category"] == MediaType.MUSIC.value
def test_torrentleech_video_search_uses_requested_category_contract(monkeypatch):
"""TorrentLeech 影视搜索应使用资源配置中的分类并标记请求类型。"""
captured = {}
def fake_get_res(_request, url: str, **_kwargs):
"""记录影视搜索地址并返回与电影分类匹配的固定响应。"""
captured["url"] = url
response = _FakeResponse()
response.json = lambda: {
"torrentList": [{
"fid": 101,
"filename": "movie.torrent",
"name": "Movie 2026",
"categoryID": 8,
"addedTimestamp": 1767225600,
"size": 2048,
}]
}
return response
monkeypatch.setattr(
"app.modules.indexer.spider.torrentleech.RequestUtils.get_res",
fake_get_res,
)
error, torrents = TorrentLeech(_build_indexer()).search(
keyword="Movie 2026",
mtype=MediaType.MOVIE,
)
assert not error
assert "/categories/8,9/" in captured["url"]
assert torrents[0]["category"] == MediaType.MOVIE.value
def test_torrentleech_async_music_search_uses_same_contract(monkeypatch):
"""TorrentLeech 异步音乐搜索应复用同步搜索的分类契约。"""
captured = {}
async def fake_get_res(_request, url: str, **_kwargs):
"""记录异步搜索地址并回放 JSON 响应。"""
captured["url"] = url
return _FakeResponse()
monkeypatch.setattr(
"app.modules.indexer.spider.torrentleech.AsyncRequestUtils.get_res",
fake_get_res,
)
error, torrents = asyncio.run(
TorrentLeech(_build_indexer()).async_search(
keyword="Artist Album",
mtype=MediaType.MUSIC,
)
)
assert not error
assert "/categories/31,16/" in captured["url"]
assert torrents[0]["category"] == MediaType.MUSIC.value
tests/test_yemapt_open_api.py
+75
View File
@@ -93,6 +93,7 @@ def test_yemapt_search_uses_open_api_auth_and_maps_fields(monkeypatch):
assert captured == {
"url": "https://www.yemapt.org/openApi/torrent/fetchOpenTorrentList.json",
"json": {
"categoryId": 4,
"keyword": "Movie",
"pageParam": {"current": 3, "pageSize": 40},
"sorter": {},
@@ -152,6 +153,80 @@ def test_yemapt_search_rejects_business_failure(monkeypatch):
assert torrents == []
def test_yemapt_maps_audio_and_music_video_categories_to_music():
"""YemaPT 音乐和 MV/演唱会分类都应映射为统一音乐类型。"""
results = YemaSpider(_build_indexer())._parse_result([
{"id": 101, "showName": "Album FLAC", "categoryId": 8},
{"id": 102, "showName": "Live Concert", "categoryId": 16},
])
assert [item["category"] for item in results] == [
MediaType.MUSIC.value,
MediaType.MUSIC.value,
]
def test_yemapt_music_search_queries_both_music_categories(monkeypatch):
"""YemaPT 音乐搜索应分别查询音乐和 MV/演唱会分类。"""
request_payloads = []
def fake_post_res(_request, url: str, json: dict = None, **_kwargs):
"""记录音乐分类请求并按分类返回一条种子。"""
request_payloads.append(json)
category_id = json["categoryId"]
return _FakeResponse({
"success": True,
"data": [{
"id": category_id,
"showName": f"Music {category_id}",
"categoryId": category_id,
}],
})
monkeypatch.setattr(
"app.modules.indexer.spider.yema.RequestUtils.post_res",
fake_post_res,
)
error, torrents = YemaSpider(_build_indexer()).search(
keyword="Artist",
mtype=MediaType.MUSIC,
)
assert not error
assert [payload["categoryId"] for payload in request_payloads] == [8, 16]
assert [torrent["category"] for torrent in torrents] == [
MediaType.MUSIC.value,
MediaType.MUSIC.value,
]
def test_yemapt_tv_search_queries_all_video_categories(monkeypatch):
"""YemaPT 剧集搜索应覆盖剧集、短剧、综艺、动漫、纪录片和体育。"""
request_payloads = []
def fake_post_res(_request, url: str, json: dict = None, **_kwargs):
"""记录剧集分类请求并返回空的成功响应。"""
request_payloads.append(json)
return _FakeResponse({"success": True, "data": []})
monkeypatch.setattr(
"app.modules.indexer.spider.yema.RequestUtils.post_res",
fake_post_res,
)
error, torrents = YemaSpider(_build_indexer()).search(
keyword="Series",
mtype=MediaType.TV,
)
assert not error
assert torrents == []
assert [payload["categoryId"] for payload in request_payloads] == [
5, 6, 13, 14, 15, 17,
]
def test_yemapt_async_search_uses_open_api(monkeypatch):
"""YemaPT 异步搜索应使用与同步搜索相同的开放 API 契约。"""
captured = {}