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

fix: preserve season zero semantics (#6150)

Browse Source
This commit is contained in:
InfinityPacer
2026-07-20 06:55:23 +08:00
committed by GitHub
parent 2056aa0b2c
commit 5588e37c6d
31 changed files with 461 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/agent/tools/impl/add_subscribe.py
View File

@@ -97,7 +97,7 @@ class AddSubscribeTool(MoviePilotTool):
message += f" ({year})"
if media_type:
message += f" [{media_type}]"
if season:
if season is not None:
message += f" 第{season}季"
elif media_type == "tv":
message += " 第1季(默认)"

2
app/agent/tools/impl/query_popular_subscribes.py
View File

@@ -118,7 +118,7 @@ class QueryPopularSubscribesTool(MoviePilotTool):
# 处理标题
title = sub.get("name")
season = sub.get("season")
if season and int(season) > 1 and media.tmdb_id:
if season not in (None, "") and int(season) != 1 and media.tmdb_id:
# 小写数据转大写
season_str = cn2an.an2cn(season, "low")
title = f"{title} 第{season_str}季"

2
app/agent/tools/impl/search_media.py
View File

@@ -43,7 +43,7 @@ class SearchMediaTool(MoviePilotTool):
message += f" ({year})"
if media_type:
message += f" [{media_type}]"
if season:
if season is not None:
message += f" 第{season}季"
return message

34
app/api/endpoints/search.py
View File

@@ -40,6 +40,16 @@ def _parse_media_type(mtype: Optional[str]) -> Optional[MediaType]:
return MediaType.from_agent(mtype) or MediaType(mtype)
def _resolve_media_season(
explicit_season: Optional[int],
recognized_season: Optional[int],
) -> Optional[int]:
"""
合并显式季号与识别结果，显式值优先且季 0 属于有效业务值。
"""
return explicit_season if explicit_season is not None else recognized_season
def _sse_event(data: dict, locale: Optional[str] = None) -> str:
"""
转换为SSE事件
@@ -298,8 +308,10 @@ async def search_by_id_stream(
doubanid=doubanid, mtype=media_type
)
if tmdbinfo:
if tmdbinfo.get("season") and not media_season:
media_season = tmdbinfo.get("season")
media_season = _resolve_media_season(
explicit_season=media_season,
recognized_season=tmdbinfo.get("season"),
)
torrents = search_chain.async_search_by_id_stream(
tmdbid=tmdbinfo.get("id"),
mtype=media_type,
@@ -404,7 +416,7 @@ async def search_by_id_stream(
meta.year = year
if media_type:
meta.type = media_type
if media_season:
if media_season is not None:
meta.type = MediaType.TV
meta.begin_season = media_season
mediainfo = await media_chain.async_recognize_by_meta(
@@ -505,8 +517,10 @@ async def search_by_id(
doubanid=doubanid, mtype=media_type
)
if tmdbinfo:
if tmdbinfo.get("season") and not media_season:
media_season = tmdbinfo.get("season")
media_season = _resolve_media_season(
explicit_season=media_season,
recognized_season=tmdbinfo.get("season"),
)
torrents = await search_chain.async_search_by_id(
tmdbid=tmdbinfo.get("id"),
mtype=media_type,
@@ -598,7 +612,7 @@ async def search_by_id(
meta.year = year
if media_type:
meta.type = media_type
if media_season:
if media_season is not None:
meta.type = MediaType.TV
meta.begin_season = media_season
mediainfo = await media_chain.async_recognize_by_meta(
@@ -770,8 +784,10 @@ async def _build_subtitle_search_source(
)
if not tmdbinfo:
return None, "未识别到TMDB媒体信息"
if tmdbinfo.get("season") and not media_season:
media_season = tmdbinfo.get("season")
media_season = _resolve_media_season(
explicit_season=media_season,
recognized_season=tmdbinfo.get("season"),
)
return call_search(tmdbid=tmdbinfo.get("id")), ""
return call_search(doubanid=doubanid), ""
@@ -813,7 +829,7 @@ async def _build_subtitle_search_source(
meta.year = year
if media_type:
meta.type = media_type
if media_season:
if media_season is not None:
meta.type = MediaType.TV
meta.begin_season = media_season
mediainfo = await media_chain.async_recognize_by_meta(

2
app/api/endpoints/subscribe.py
View File

@@ -639,7 +639,7 @@ async def popular_subscribes(
# 处理标题
title = sub.get("name")
season = sub.get("season")
if season and int(season) > 1 and media.tmdb_id:
if season not in (None, "") and int(season) != 1 and media.tmdb_id:
# 小写数据转大写
season_str = cn2an.an2cn(season, "low")
title = f"{title} 第{season_str}季"

7
app/chain/download.py
View File

@@ -488,10 +488,13 @@ class DownloadChain(ChainBase):
)
meta = getattr(context, "meta_info", None)
site = getattr(torrent, "site", None) or getattr(torrent, "site_name", None)
meta_season = getattr(meta, "season", None)
media_season = getattr(media, "season", None)
season = meta_season if meta_season is not None else media_season
payload = {
"media_type": str(media_type or ""),
"media_key": str(media_key or ""),
"season": str(getattr(meta, "season", None) or getattr(media, "season", None) or ""),
"season": str(season) if season is not None else "",
"episodes": cls._format_failure_episodes(meta) or "",
"site": str(site or ""),
"resource": cls._torrent_resource_key(torrent),
@@ -1177,7 +1180,7 @@ class DownloadChain(ChainBase):
if not tv.episodes:
if not need_seasons.get(need_mid):
need_seasons[need_mid] = []
need_seasons[need_mid].append(tv.season or 1)
need_seasons[need_mid].append(tv.season if tv.season is not None else 1)
logger.info(f"缺失整季：{need_seasons}")
# 查找整季包含的种子，只处理整季没集的种子或者是集数超过季的种子
for need_mid, need_season in need_seasons.items():

24
app/chain/media.py
View File

@@ -653,6 +653,16 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
self.obtain_images(mediainfo=mediainfo)
return mediainfo
@staticmethod
def _parse_recognize_event_number(value) -> Optional[int]:
"""
解析辅助识别返回的季集号，兼容整数和数字字符串并保留数值 0。
"""
if value is None:
return None
text = str(value).strip()
return int(text) if text.isdigit() else None
def recognize_help(
self,
title: str,
@@ -686,10 +696,8 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
title = str(event_data["name"]).split("/")[0].strip().replace(".", " ")
if event_data.get("year"):
year = str(event_data["year"]).split("/")[0].strip()
if event_data.get("season") and str(event_data["season"]).isdigit():
season_number = int(event_data["season"])
if event_data.get("episode") and str(event_data["episode"]).isdigit():
episode_number = int(event_data["episode"])
season_number = self._parse_recognize_event_number(event_data.get("season"))
episode_number = self._parse_recognize_event_number(event_data.get("episode"))
if not title:
return None
if title == "Unknown":
@@ -1635,10 +1643,8 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
title = str(event_data["name"]).split("/")[0].strip().replace(".", " ")
if event_data.get("year"):
year = str(event_data["year"]).split("/")[0].strip()
if event_data.get("season") and str(event_data["season"]).isdigit():
season_number = int(event_data["season"])
if event_data.get("episode") and str(event_data["episode"]).isdigit():
episode_number = int(event_data["episode"])
season_number = self._parse_recognize_event_number(event_data.get("season"))
episode_number = self._parse_recognize_event_number(event_data.get("episode"))
if not title:
return None
if title == "Unknown":
@@ -1654,7 +1660,7 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
org_meta.year = year
org_meta.begin_season = season_number
org_meta.begin_episode = episode_number
if org_meta.begin_season or org_meta.begin_episode:
if org_meta.begin_season is not None or org_meta.begin_episode is not None:
org_meta.type = MediaType.TV
# 重新识别
return await self.async_recognize_media(

2
app/chain/message.py
View File

@@ -2087,7 +2087,7 @@ class MediaInteractionChain(ChainBase):
mediakey = mediainfo.tmdb_id or mediainfo.douban_id
no_exists = {mediakey: {}}
if meta.begin_season:
if meta.begin_season is not None:
episodes = mediainfo.seasons.get(meta.begin_season)
if not episodes:
return {}

2
app/chain/search.py
View File

@@ -203,7 +203,7 @@ class SearchChain(ChainBase):
"area": str(params.get("area") or ""),
"title": str(params.get("title") or ""),
"year": str(params.get("year") or ""),
"season": str(params.get("season") or ""),
"season": str(params["season"]) if params.get("season") is not None else "",
"episode": str(params.get("episode") or ""),
"sites": str(params.get("sites") or ""),
"result_type": str(params.get("result_type") or "torrent"),

33
app/core/meta/metaanime.py
View File

@@ -38,6 +38,14 @@ class MetaAnime(MetaBase):
_name_nostring_pattern = re.compile(_name_nostring_re, re.IGNORECASE)
_fps_pattern = re.compile(r"(%s)" % _fps_re, re.IGNORECASE)
@staticmethod
def _parse_season_number(value):
"""解析第三方动漫季号，仅接受整数或纯数字字符串并保留数值 0。"""
if value is None:
return None
text = str(value).strip()
return int(text) if text.isdigit() else None
def __init__(self, title: str, subtitle: str = None, isfile: bool = False):
super().__init__(title, subtitle, isfile)
if not title:
@@ -111,22 +119,19 @@ class MetaAnime(MetaBase):
# 季号
anime_season = anitopy_info.get("anime_season")
if isinstance(anime_season, list):
if len(anime_season) == 1:
begin_season = anime_season[0]
end_season = None
else:
begin_season = anime_season[0]
end_season = anime_season[-1]
elif anime_season:
begin_season = anime_season
end_season = None
seasons = [
season for item in anime_season
if (season := self._parse_season_number(item)) is not None
]
begin_season = seasons[0] if seasons else None
end_season = seasons[-1] if len(seasons) > 1 else None
else:
begin_season = None
begin_season = self._parse_season_number(anime_season)
end_season = None
if begin_season:
self.begin_season = int(begin_season)
if end_season and int(end_season) != self.begin_season:
self.end_season = int(end_season)
if begin_season is not None:
self.begin_season = begin_season
if end_season is not None and end_season != self.begin_season:
self.end_season = end_season
self.total_season = (self.end_season - self.begin_season) + 1
else:
self.total_season = 1

12
app/core/meta/metavideo.py
View File

@@ -251,7 +251,7 @@ class MetaVideo(MetaBase):
if name.isdecimal() \
and int(name) < 1800 \
and not self.year \
and not self.begin_season \
and self.begin_season is None \
and not self.resource_pix \
and not self.resource_type \
and not self.audio_encode \
@@ -259,7 +259,7 @@ class MetaVideo(MetaBase):
if self.begin_episode is None:
self.begin_episode = int(name)
name = None
elif self.is_in_episode(int(name)) and not self.begin_season:
elif self.is_in_episode(int(name)) and self.begin_season is None:
name = None
return name
@@ -366,7 +366,7 @@ class MetaVideo(MetaBase):
if not self.name:
return
if not self.year \
and not self.begin_season \
and self.begin_season is None \
and not self.begin_episode \
and not self.resource_pix \
and not self.resource_type:
@@ -690,7 +690,7 @@ class MetaVideo(MetaBase):
if not self.year \
and not self.resource_pix \
and not self.resource_type \
and not self.begin_season \
and self.begin_season is None \
and not self.begin_episode:
return
re_res = self._video_encode_pattern.search(token)
@@ -738,7 +738,7 @@ class MetaVideo(MetaBase):
if not self.year \
and not self.resource_pix \
and not self.resource_type \
and not self.begin_season \
and self.begin_season is None \
and not self.begin_episode:
return
video_bit = self.extract_video_bit(token)
@@ -759,7 +759,7 @@ class MetaVideo(MetaBase):
if not self.year \
and not self.resource_pix \
and not self.resource_type \
and not self.begin_season \
and self.begin_season is None \
and not self.begin_episode:
return
re_res = self._audio_encode_pattern.search(token)

3
app/helper/server.py
View File

@@ -1464,7 +1464,8 @@ class MoviePilotServerHelper:
params["type"] = media_type
if year := cls._extract_year(meta=meta):
params["year"] = year
if season := cls._extract_season(media_type=media_type, meta=meta):
season = cls._extract_season(media_type=media_type, meta=meta)
if season is not None:
params["season"] = season
return params

6
app/modules/douban/__init__.py
View File

@@ -98,7 +98,7 @@ class DoubanModule(_ModuleBase):
continue
ret_medias.append(MediaInfo(douban_info=item_obj.get("target")))
# 将搜索词中的季写入标题中
if ret_medias and meta.begin_season:
if ret_medias and meta.begin_season is not None:
# 小写数据转大写
season_str = cn2an.an2cn(meta.begin_season, "low")
for media in ret_medias:
@@ -155,7 +155,7 @@ class DoubanModule(_ModuleBase):
elif meta:
info = {}
for name in self._prepare_search_names(meta):
if meta.begin_season:
if meta.begin_season is not None:
logger.info(f"正在识别 {name} 第{meta.begin_season}季 ...")
else:
logger.info(f"正在识别 {name} ...")
@@ -255,7 +255,7 @@ class DoubanModule(_ModuleBase):
elif meta:
info = {}
for name in self._prepare_search_names(meta):
if meta.begin_season:
if meta.begin_season is not None:
logger.info(f"正在识别 {name} 第{meta.begin_season}季 ...")
else:
logger.info(f"正在识别 {name} ...")

2
app/modules/douban/douban_cache.py
View File

@@ -147,7 +147,7 @@ class DoubanCache(metaclass=WeakSingleton):
mtype = MediaType.MOVIE if info.get("type") == "movie" else MediaType.TV
else:
meta = MetaInfo(cache_title)
if meta.begin_season:
if meta.begin_season is not None:
mtype = MediaType.TV
else:
mtype = MediaType.MOVIE

2
app/modules/filemanager/__init__.py
View File

@@ -632,7 +632,7 @@ class FileManagerModule(_ModuleBase):
seasons: Dict[int, list] = {}
for fileitem in fileitems:
file_meta = MetaInfo(fileitem.basename)
season_index = file_meta.begin_season or 1
season_index = file_meta.begin_season if file_meta.begin_season is not None else 1
episode_index = file_meta.begin_episode
if not episode_index:
continue

22
app/modules/themoviedb/__init__.py
View File

@@ -162,7 +162,7 @@ class TheMovieDbModule(_ModuleBase):
if not results:
return []
medias = [MediaInfo(tmdb_info=info) for info in results]
if meta.begin_season:
if meta.begin_season is not None:
# 小写数据转大写
season_str = cn2an.an2cn(meta.begin_season, "low")
for media in medias:
@@ -267,7 +267,7 @@ class TheMovieDbModule(_ModuleBase):
"""
根据名称搜索媒体信息
"""
if meta.begin_season:
if meta.begin_season is not None:
logger.info(f"正在识别 {name} 第{meta.begin_season}季 ...")
else:
logger.info(f"正在识别 {name} ...")
@@ -303,7 +303,7 @@ class TheMovieDbModule(_ModuleBase):
"""
根据名称搜索媒体信息（异步版本）
"""
if meta.begin_season:
if meta.begin_season is not None:
logger.info(f"正在识别 {name} 第{meta.begin_season}季 ...")
else:
logger.info(f"正在识别 {name} ...")
@@ -630,7 +630,7 @@ class TheMovieDbModule(_ModuleBase):
:param name: 名称
:param mtype: 类型
:param year: 年份
:param season: 季号
:param season: 用于匹配指定季，0 表示特别季
"""
# 搜索
logger.info(f"开始使用 名称：{name} 年份：{year} 匹配TMDB信息 ...")
@@ -651,7 +651,7 @@ class TheMovieDbModule(_ModuleBase):
:param name: 名称
:param mtype: 类型
:param year: 年份
:param season: 季号
:param season: 用于匹配指定季，0 表示特别季
"""
# 搜索
logger.info(f"开始使用 名称：{name} 年份：{year} 匹配TMDB信息 ...")
@@ -670,10 +670,10 @@ class TheMovieDbModule(_ModuleBase):
获取TMDB信息
:param tmdbid: int
:param mtype: 媒体类型
:param season: 季号
:return: TVDB信息
:param season: 季号；TV 的显式值（含 0）读取季详情，None 或电影的 0 读取媒体详情
:return: TMDB信息
"""
if not season:
if season is None or (season == 0 and mtype != MediaType.TV):
return self.tmdb.get_info(mtype=mtype, tmdbid=tmdbid)
else:
return self.tmdb.get_tv_season_detail(tmdbid=tmdbid, season=season)
@@ -683,10 +683,10 @@ class TheMovieDbModule(_ModuleBase):
异步获取TMDB信息
:param tmdbid: int
:param mtype: 媒体类型
:param season: 季号
:return: TVDB信息
:param season: 季号；TV 的显式值（含 0）读取季详情，None 或电影的 0 读取媒体详情
:return: TMDB信息
"""
if not season:
if season is None or (season == 0 and mtype != MediaType.TV):
return await self.tmdb.async_get_info(mtype=mtype, tmdbid=tmdbid)
else:
return await self.tmdb.async_get_tv_season_detail(tmdbid=tmdbid, season=season)

2
app/workflow/actions/fetch_torrents.py
View File

@@ -74,7 +74,7 @@ class FetchTorrentsAction(BaseAction):
continue
if params.type and torrent.media_info and torrent.media_info.type != MediaType(params.type):
continue
if params.season and torrent.meta_info.begin_season != params.season:
if params.season is not None and torrent.meta_info.begin_season != params.season:
continue
# 识别媒体信息
if params.match_media:

9
tests/test_agent_add_subscribe_tool.py
View File

@@ -7,6 +7,15 @@ from app.schemas.types import MessageChannel
class TestAgentAddSubscribeTool(unittest.TestCase):
def test_tool_message_displays_special_season_zero(self):
"""Agent 提示必须把显式季 0 显示为特别季，而不是默认第一季。"""
tool = AddSubscribeTool(session_id="session-1", user_id="10001")
message = tool.get_tool_message(title="测试剧", media_type="tv", season=0)
self.assertIn("第0季", message)
self.assertNotIn("第1季(默认)", message)
def test_tv_subscription_without_season_reports_default_first_season(self):
tool = AddSubscribeTool(session_id="session-1", user_id="10001")
tool.set_message_attr(

29
tests/test_agent_query_popular_subscribes_tool.py Normal file
View File

@@ -0,0 +1,29 @@
import asyncio
import json
from app.agent.tools.impl.query_popular_subscribes import QueryPopularSubscribesTool
def test_popular_subscribe_title_distinguishes_special_season_zero(monkeypatch):
"""热门订阅结果应在标题中明确标识特别季，同时保留数值季号。"""
async def fake_statistics(**_kwargs):
return [{
"type": "tv",
"name": "Demo Show",
"season": 0,
"tmdbid": 1,
"count": 5,
}]
monkeypatch.setattr(
"app.agent.tools.impl.query_popular_subscribes."
"MoviePilotServerHelper.async_get_subscribe_statistic",
fake_statistics,
)
tool = QueryPopularSubscribesTool(session_id="session-1", user_id="10001")
result = asyncio.run(tool.run(media_type="tv"))
payload = json.loads(result.split("\n\n", 1)[1])
assert payload[0]["title"] == "Demo Show 第零季"
assert payload[0]["season"] == 0

10
tests/test_agent_search_media_tool.py Normal file
View File

@@ -0,0 +1,10 @@
from app.agent.tools.impl.search_media import SearchMediaTool
def test_tool_message_displays_special_season_zero():
"""媒体搜索提示应展示显式季 0。"""
tool = SearchMediaTool(session_id="session-1", user_id="10001")
message = tool.get_tool_message(title="测试剧", media_type="tv", season=0)
assert "第0季" in message

17
tests/test_douban_cache_management.py
View File

@@ -26,6 +26,10 @@ class _MemoryCacheStub:
"""删除指定缓存条目。"""
self.data.pop(key, None)
def set(self, key: str, value):
"""写入指定缓存条目。"""
self.data[key] = value
def clear(self):
"""清空全部缓存条目。"""
self.data.clear()
@@ -79,6 +83,19 @@ def test_douban_cache_list_items_normalizes_media_type_and_sorting():
assert items[1]["douban_id"] == 0
def test_douban_cache_infers_special_season_title_as_tv():
"""缺少显式类型时，S00 标题仍应按电视剧写入缓存。"""
cache = _build_douban_cache({})
cache.update(
meta=None,
info={"id": "special", "title": "测试剧 S00", "year": "2024"},
)
cached = next(iter(cache._cache.data.values()))
assert cached["type"] == MediaType.TV
def test_douban_cache_delete_and_clear_persist_immediately(monkeypatch):
"""豆瓣管理操作应修改运行时缓存并立即触发本地持久化。"""
cache = _build_douban_cache({"first": {"id": "1"}, "second": {"id": "2"}})

61
tests/test_download_chain.py
View File

@@ -444,6 +444,40 @@ def test_batch_download_rejects_complete_coverage_when_files_do_not_cover_target
chain.download_single.assert_not_called()
def test_batch_download_preserves_special_season_zero(monkeypatch):
"""特别季整季需求必须以季 0 匹配候选，不能回退成第 1 季。"""
_FakeBatchTorrentHelper.episodes = list(range(1, 7))
monkeypatch.setattr(download_module, "TorrentHelper", _FakeBatchTorrentHelper)
monkeypatch.setattr(download_module.eventmanager, "send_event", lambda *args, **kwargs: None)
chain = DownloadChain.__new__(DownloadChain)
chain.download_torrent = MagicMock(return_value=(b"torrent-content", "", ["demo.mkv"]))
chain.download_single = MagicMock(return_value="hash")
context = _build_tv_context()
context.meta_info.season_list = [0]
context.meta_info.season_episode = "S00"
context.meta_info.org_string = "Test Show S00 2160p"
context.torrent_info.title = "Test Show S00 2160p"
no_exists = {
1: {
0: NotExistMediaInfo(
season=0,
episodes=[],
total_episode=6,
start_episode=1,
require_complete_coverage=True,
)
}
}
downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
assert downloads == [context]
assert lefts == {}
chain.download_single.assert_called_once()
def test_batch_download_rejects_complete_coverage_when_only_missing_episodes_match(monkeypatch):
"""
完整覆盖要求目标范围全集，不能只覆盖当前缺口集。
@@ -634,6 +668,33 @@ def test_download_single_records_failure_cooldown_when_downloader_rejects(monkey
assert captured["next_retry_at"] > captured["now_time"]
def test_download_failure_fingerprint_distinguishes_special_season_zero():
"""失败冷却指纹应区分特别季与未指定季，避免错误共享冷却状态。"""
def build_context(season):
return SimpleNamespace(
media_info=SimpleNamespace(
type=MediaType.TV,
title="Demo Show",
year="2026",
tmdb_id=1,
season=None,
),
meta_info=SimpleNamespace(season=season, episode=None, episode_list=[]),
torrent_info=SimpleNamespace(
site=12,
title="Demo Show Specials",
torrent_id="484660",
),
)
special_fingerprint = DownloadChain._build_download_failure_fingerprint(build_context(0))
unspecified_fingerprint = DownloadChain._build_download_failure_fingerprint(build_context(None))
assert special_fingerprint
assert unspecified_fingerprint
assert special_fingerprint != unspecified_fingerprint
def test_batch_download_skips_failed_subscription_resource_and_tries_next(monkeypatch):
"""
订阅自动下载应跳过冷却中的失败资源，但继续尝试同媒体的后续候选。

18
tests/test_filemanager_season_zero.py Normal file
View File

@@ -0,0 +1,18 @@
from types import SimpleNamespace
from unittest.mock import patch
from app.core.context import MediaInfo
from app.modules.filemanager import FileManagerModule
from app.schemas.types import MediaType
def test_local_media_exists_keeps_special_season_zero():
"""本地 S00 文件必须归入特别季，不能计入第一季。"""
module = FileManagerModule()
module.media_files = lambda _mediainfo: [SimpleNamespace(basename="Test.Show.S00E01.mkv")]
mediainfo = MediaInfo(title="Test Show", type=MediaType.TV)
with patch("app.modules.filemanager.settings.LOCAL_EXISTS_SEARCH", True):
exists = module.media_exists(mediainfo)
assert exists.seasons == {0: [1]}

18
tests/test_media_interaction.py
View File

@@ -138,6 +138,24 @@ def _build_single_download_dir() -> list[TransferDirectoryConf]:
]
def test_rebuild_download_scope_keeps_special_season_zero():
"""重新下载特别季时只能重建季 0 范围，不能扩展为整部剧的所有季。"""
meta = _build_meta("测试剧")
meta.begin_season = 0
mediainfo = MediaInfo(
type=MediaType.TV,
title="测试剧",
year="2026",
tmdb_id=1,
seasons={0: [1, 2], 1: [1, 2, 3]},
)
no_exists = MediaInteractionChain._get_noexits_info(meta, mediainfo)
assert list(no_exists[1]) == [0]
assert no_exists[1][0].total_episode == 2
def test_message_routes_text_reply_to_media_interaction_before_ai():
"""已有传统媒体交互时，用户回复应优先交给传统交互处理。"""
chain = MessageChain()

56
tests/test_media_recognize_modules.py
View File

@@ -254,3 +254,59 @@ class MediaRecognizeModulesTest(TestCase):
)
self.assertEqual(matched["id"], "201")
def test_search_result_builders_preserve_special_season_zero(self):
"""TMDB 与豆瓣搜索结果都必须携带显式特别季。"""
meta = MetaBase("测试剧")
meta.name = "测试剧"
meta.begin_season = 0
meta.type = MediaType.TV
tmdb_results = TheMovieDbModule._build_search_medias_result(
meta,
[{"id": 100, "name": "测试剧", "media_type": "tv", "first_air_date": "2024-01-01"}],
)
douban_results = DoubanModule._build_search_medias_result(
meta,
[{
"type_name": MediaType.TV.value,
"target": {"id": "200", "title": "测试剧", "type": "tv", "year": "2024"},
}],
)
self.assertEqual(tmdb_results[0].season, 0)
self.assertEqual(douban_results[0].season, 0)
def test_tmdb_info_treats_special_season_zero_as_season_detail(self):
"""TV 的显式季 0 读取特别季，None 及电影的 0 读取媒体详情。"""
module = TheMovieDbModule()
module.tmdb = Mock()
module.tmdb.get_info.return_value = {"scope": "series"}
module.tmdb.get_tv_season_detail.return_value = {"scope": "season", "season_number": 0}
special = module.tmdb_info(tmdbid=100, mtype=MediaType.TV, season=0)
series = module.tmdb_info(tmdbid=100, mtype=MediaType.TV, season=None)
movie = module.tmdb_info(tmdbid=200, mtype=MediaType.MOVIE, season=0)
self.assertEqual(special["scope"], "season")
self.assertEqual(series["scope"], "series")
self.assertEqual(movie["scope"], "series")
module.tmdb.get_tv_season_detail.assert_called_once_with(tmdbid=100, season=0)
def test_async_tmdb_info_treats_special_season_zero_as_season_detail(self):
"""异步 TMDB 接口必须与同步接口保持相同的季 0 契约。"""
module = TheMovieDbModule()
module.tmdb = Mock()
module.tmdb.async_get_info = AsyncMock(return_value={"scope": "series"})
module.tmdb.async_get_tv_season_detail = AsyncMock(
return_value={"scope": "season", "season_number": 0}
)
special = asyncio.run(module.async_tmdb_info(tmdbid=100, mtype=MediaType.TV, season=0))
series = asyncio.run(module.async_tmdb_info(tmdbid=100, mtype=MediaType.TV, season=None))
movie = asyncio.run(module.async_tmdb_info(tmdbid=200, mtype=MediaType.MOVIE, season=0))
self.assertEqual(special["scope"], "season")
self.assertEqual(series["scope"], "series")
self.assertEqual(movie["scope"], "series")
module.tmdb.async_get_tv_season_detail.assert_awaited_once_with(tmdbid=100, season=0)

22
tests/test_media_recognize_share.py
View File

@@ -243,6 +243,28 @@ class TestMediaRecognizeShare(unittest.TestCase):
self.assertEqual(report_payload["year"], "2024")
self.assertEqual(report_payload["season"], 2)
def test_query_and_report_preserve_special_season_zero(self):
"""共享识别查询和上报都必须保留显式特别季。"""
meta = self._build_meta("测试剧特别篇", MediaType.TV)
meta.begin_season = 0
mediainfo = MediaInfo(title="测试剧", tmdb_id=402, type=MediaType.TV, season=0)
query_params = MoviePilotServerHelper._build_recognize_query_params(meta=meta)
report_payload = MoviePilotServerHelper._build_recognize_report_payload(
meta=meta,
mediainfo=mediainfo,
)
self.assertEqual(query_params["season"], 0)
self.assertEqual(report_payload["season"], 0)
def test_plugin_recognize_number_parser_preserves_zero(self):
"""插件辅助识别应同时接受整数和字符串形式的季 0。"""
self.assertEqual(self.media_chain._parse_recognize_event_number(0), 0)
self.assertEqual(self.media_chain._parse_recognize_event_number("0"), 0)
self.assertIsNone(self.media_chain._parse_recognize_event_number(None))
self.assertIsNone(self.media_chain._parse_recognize_event_number("invalid"))
def test_report_shared_result_with_distinct_keyword_meta(self):
"""
辅助识别成功后应按辅助前名称上报共享结果。

55
tests/test_metainfo.py
View File

@@ -4,6 +4,7 @@ from types import SimpleNamespace
from unittest.mock import patch
from app.core.metainfo import MetaInfo, MetaInfoPath, find_metainfo
from app.core.meta.metaanime import MetaAnime
from app.helper.torrent import TorrentHelper
from app.schemas.types import MediaType
from tests.cases.meta import meta_cases
@@ -357,6 +358,60 @@ def test_video_bit_extracted_for_video_title():
assert meta.video_bit == "10bit"
def test_special_season_zero_enables_whole_season_resource_parsing():
"""只有 S00、没有集号的整季标题仍应识别后续编码信息。"""
with patch("app.core.metainfo.rust_accel.parse_metainfo", return_value=None):
meta = MetaInfo(title="Demo Show S00 X265 AAC")
assert meta.begin_season == 0
assert meta.video_encode == "x265"
assert meta.audio_encode == "AAC"
def test_anime_parser_preserves_numeric_special_season_zero():
"""第三方动漫解析器返回整数 0 时也应保留特别季。"""
parsed = {
"anime_title": "Demo Anime",
"anime_season": 0,
"episode_number": "1",
}
with patch("app.core.meta.metaanime.anitopy.parse", return_value=parsed):
meta = MetaAnime(title="Demo Anime S00E01")
assert meta.begin_season == 0
assert meta.begin_episode == 1
assert meta.type == MediaType.TV
parsed["anime_season"] = [0, "1"]
with patch("app.core.meta.metaanime.anitopy.parse", return_value=parsed):
ranged_meta = MetaAnime(title="Demo Anime S00-S01")
assert ranged_meta.begin_season == 0
assert ranged_meta.end_season == 1
def test_anime_parser_ignores_empty_and_invalid_season_values():
"""第三方动漫季号的空值和非法列表项应按未指定处理且不得抛错。"""
empty = {
"anime_title": "Demo Anime",
"anime_season": "",
"episode_number": "1",
}
invalid_list = {
"anime_title": "Demo Anime",
"anime_season": ["", "invalid"],
"episode_number": "1",
}
with patch("app.core.meta.metaanime.anitopy.parse", return_value=empty):
empty_meta = MetaAnime(title="Demo Anime E01")
with patch("app.core.meta.metaanime.anitopy.parse", return_value=invalid_list):
invalid_meta = MetaAnime(title="Demo Anime E01")
assert empty_meta.begin_season is None
assert invalid_meta.begin_season is None
def test_hdr_vivid_effect_extracted_for_video_title():
"""测试合并写法 HDRVivid 可识别为资源效果。"""
with patch("app.core.metainfo.rust_accel.parse_metainfo", return_value=None):

8
tests/test_search_ai_recommend.py
View File

@@ -447,6 +447,14 @@ class SearchChainAIRecommendTest(unittest.IsolatedAsyncioTestCase):
)
self.assertTrue(any(filename == "__search_result__" for filename, _ in cached))
def test_search_params_preserve_special_season_zero(self):
"""最近搜索参数必须把显式季 0 保存为字符串 0，供页面刷新后重放。"""
params = SearchChain._normalize_search_params(
{"keyword": "tmdb:123", "season": 0}
)
self.assertEqual(params["season"], "0")
def test_tool_factory_excludes_message_tools_when_disabled(self):
with patch(
"app.agent.tools.factory.PluginManager.get_plugin_agent_tools",

8
tests/test_subtitle_search.py
View File

@@ -1,6 +1,6 @@
import pytest
from app.api.endpoints.search import _parse_media_type
from app.api.endpoints.search import _parse_media_type, _resolve_media_season
from app.chain.search import SearchChain
from app.core.context import MediaInfo, SubtitleInfo
from app.modules.indexer import IndexerModule
@@ -24,6 +24,12 @@ AUDIENCES_SUBTITLE_HTML = """
</tbody></table>
"""
def test_explicit_special_season_zero_overrides_recognized_season():
"""精确搜索中显式季 0 必须优先于跨源识别返回的其它季号。"""
assert _resolve_media_season(explicit_season=0, recognized_season=1) == 0
assert _resolve_media_season(explicit_season=None, recognized_season=1) == 1
HHANCLUB_SUBTITLE_HTML = """
<div class="flex flex-col w-full items-center mt-[25px] gap-y-[10px] bg-[#F1F3F5] !rounded-md p-5" id="subtitles-table">
<div class="grid grid-cols-[10%_60%_10%_10%_10%] w-[95%] !rounded-md py-1 items-center bg-[#FFFFFF]/[0.7]">

20
tests/test_template_context_builder.py
View File

@@ -10,7 +10,10 @@ TemplateContextBuilder 的并发安全单元测试。
"""
import threading
from app.core.context import MediaInfo
from app.core.metainfo import MetaInfo
from app.helper.message import TemplateContextBuilder
from app.schemas.types import MediaType
from app.schemas.tmdb import TmdbEpisode
@@ -118,3 +121,20 @@ def test_build_exposes_total_episodes_from_current_season() -> None:
)
assert context.get("total_episodes") == 3
def test_build_preserves_special_season_context() -> None:
"""显式 S00 必须优先于媒体回退季，并使用特别季年份。"""
meta = MetaInfo("Test Show S00E01")
mediainfo = MediaInfo(
title="Test Show",
type=MediaType.TV,
season=1,
season_years={0: "2024", 1: "2025"},
)
context = TemplateContextBuilder().build(meta=meta, mediainfo=mediainfo)
assert context["season"] == "0"
assert context["season_fmt"] == "S00"
assert context["season_year"] == "2024"

36
tests/test_workflow_actions.py
View File

@@ -4,8 +4,10 @@ from app.schemas import ActionContext, DownloadTask, FileItem
from app.schemas.workflow import ActionResult
from app.workflow.actions import BaseAction
from app.workflow.actions import fetch_downloads as fetch_downloads_module
from app.workflow.actions import fetch_torrents as fetch_torrents_module
from app.workflow.actions import scrape_file as scrape_file_module
from app.workflow.actions.fetch_downloads import FetchDownloadsAction
from app.workflow.actions.fetch_torrents import FetchTorrentsAction
from app.workflow.actions.scrape_file import ScrapeFileAction
from app.workflow.actions.fetch_rss import FetchRssAction
from app.workflow import WorkFlowManager
@@ -42,6 +44,40 @@ def test_fetch_downloads_updates_context_downloads(monkeypatch):
assert result.downloads[0].path == "/downloads/movie.mkv"
def test_fetch_torrents_filters_special_season_zero(monkeypatch):
"""工作流显式选择季 0 时只能保留特别季资源。"""
class FakeSearchChain:
"""返回特别季和第一季候选，验证动作层季过滤。"""
def search_by_title(self, **_kwargs):
return [
SimpleNamespace(
meta_info=SimpleNamespace(year=None, begin_season=0),
media_info=None,
torrent_info=SimpleNamespace(title="Test S00"),
),
SimpleNamespace(
meta_info=SimpleNamespace(year=None, begin_season=1),
media_info=None,
torrent_info=SimpleNamespace(title="Test S01"),
),
]
monkeypatch.setattr(fetch_torrents_module, "SearchChain", FakeSearchChain)
monkeypatch.setattr(fetch_torrents_module.global_vars, "is_workflow_stopped", lambda _workflow_id: False)
action = FetchTorrentsAction("fetch-torrents")
action.job_done = lambda *_args, **_kwargs: None
result = action.execute(
workflow_id=1,
params={"search_type": "keyword", "name": "Test", "season": 0},
context=ActionContext(),
)
assert [item.meta_info.begin_season for item in result.torrents] == [0]
def test_scrape_file_keeps_workflow_action_context(monkeypatch):
"""刮削文件动作不应将工作流上下文替换为媒体识别上下文。"""
scraped = []