fix(media): preserve recognition and rule mutation identities (#7144)

* fix(media): preserve recognition and rule mutation identities * test: cover typed classification facts with alias policies --------- Co-authored-by: jxxghp <jxxghp@gmail.com>
2026-10-31 04:14:38 +08:00 · 2026-10-23 07:33:27 +08:00
parent 598d1663b1
commit bcbf79f030
34 changed files with 672 additions and 133 deletions
--- a/app/api/endpoints/media.py
+++ b/app/api/endpoints/media.py
@@ -393,7 +393,7 @@ async def search(
        music_types = (music_type,) if music_type else (MUSIC_ENTITY_RECORDING, MUSIC_ENTITY_ALBUM)
        filtered_music_results = await media_chain.async_search_music(
            query=title,
-            limit=count,
+            limit=page * count,
            media_source=source_selection,
            music_types=music_types,
        )
@@ -534,13 +534,13 @@ def _scrape_impl(
        if not Path(fileitem.path).exists():
            return _SchemaResponse(success=False, message="刮削路径不存在")
    # 手动刮削 (暂时使用同步版本，可以后续优化为异步)
-    ScrapingChain().scrape_metadata(
+    success, message = ScrapingChain().scrape_metadata(
        fileitem=fileitem,
        meta=meta_info,
        mediainfo=media_info,
        overwrite=True,
    )
-    return _SchemaResponse(success=True, message=f"{fileitem.path} 刮削完成")
+    return _SchemaResponse(success=success, message=message)


@router.post("/scrape/{storage}", summary="刮削媒体信息", response_model=_SchemaResponse[None])
--- a/app/api/endpoints/recommend.py
+++ b/app/api/endpoints/recommend.py
@@ -236,7 +236,11 @@ async def _fetch_media_recommendations(
    operation = _media_source_operations(chain, page=page, count=count).get(source)
    if operation is None:
        return False, []
-    return True, await operation()
+    results = await operation()
+    if media_type != "all":
+        results = [item for item in results or []
+                   if isinstance(item, dict) and media_type_to_agent(item.get("type")) == media_type]
+    return True, results


 def _project_agent_recommendations(results: List[Any], count: int) -> List[dict[str, Any]]:
--- a/app/api/endpoints/rule.py
+++ b/app/api/endpoints/rule.py
@@ -14,6 +14,7 @@ from app.api.dependencies.auth import (
 from app.api.principal import ApiPrincipal
 from app.api.response import ResponseAPIRouter
 from app.application.filtering import FilterRuleService
+from app.application.rules import RuleGroupMutationConflictError
 from app.schemas.common import JsonObject as _SchemaJsonObject
 from app.schemas.response import Response as _SchemaResponse
 from app.schemas.rule import (
@@ -149,7 +150,7 @@ async def add_custom_rule(
    """
    try:
        data = await _service(runtime).add_custom(**payload.model_dump())
-    except ValueError as error:
+    except (ValueError, RuleGroupMutationConflictError) as error:
        return _SchemaResponse(success=False, message=str(error))
    return _SchemaResponse(success=True, message=data.get("message") or "", data=data)

@@ -172,7 +173,7 @@ async def reorder_custom_rules(
            payload.rule_ids,
            expected_rule_ids=payload.expected_rule_ids,
        )
-    except ValueError as error:
+    except (ValueError, RuleGroupMutationConflictError) as error:
        return _SchemaResponse(success=False, message=str(error))
    return _SchemaResponse(success=True, message=data.get("message") or "", data=data)

@@ -196,7 +197,7 @@ async def update_custom_rule(
            current_rule_id=rule_id,
            **payload.model_dump(),
        )
-    except ValueError as error:
+    except (ValueError, RuleGroupMutationConflictError) as error:
        return _SchemaResponse(success=False, message=str(error))
    return _SchemaResponse(success=True, message=data.get("message") or "", data=data)

@@ -216,7 +217,7 @@ async def delete_custom_rule(
    """
    try:
        data = await _service(runtime).delete_custom(rule_id)
-    except ValueError as error:
+    except (ValueError, RuleGroupMutationConflictError) as error:
        return _SchemaResponse(success=False, message=str(error))
    return _SchemaResponse(success=True, message=data.get("message") or "", data=data)

@@ -236,7 +237,7 @@ async def add_rule_group(
    """
    try:
        data = await _service(runtime).add_group(**payload.model_dump())
-    except ValueError as error:
+    except (ValueError, RuleGroupMutationConflictError) as error:
        return _SchemaResponse(success=False, message=str(error))
    return _SchemaResponse(success=True, message=data.get("message") or "", data=data)

@@ -259,7 +260,7 @@ async def reorder_rule_groups(
            payload.group_names,
            expected_group_names=payload.expected_group_names,
        )
-    except ValueError as error:
+    except (ValueError, RuleGroupMutationConflictError) as error:
        return _SchemaResponse(success=False, message=str(error))
    return _SchemaResponse(success=True, message=data.get("message") or "", data=data)

@@ -283,7 +284,7 @@ async def update_rule_group(
            current_name=name,
            **payload.model_dump(),
        )
-    except ValueError as error:
+    except (ValueError, RuleGroupMutationConflictError) as error:
        return _SchemaResponse(success=False, message=str(error))
    return _SchemaResponse(success=True, message=data.get("message") or "", data=data)

@@ -303,6 +304,6 @@ async def delete_rule_group(
    """
    try:
        data = await _service(runtime).delete_group(name)
-    except ValueError as error:
+    except (ValueError, RuleGroupMutationConflictError) as error:
        return _SchemaResponse(success=False, message=str(error))
    return _SchemaResponse(success=True, message=data.get("message") or "", data=data)
--- a/app/application/filtering.py
+++ b/app/application/filtering.py
@@ -557,17 +557,29 @@ class FilterRuleService:
        """
        rules = get_custom_rules()
        new_rule = normalize_custom_rule(existing_rules=rules, **values)
-        rules.append(new_rule)
-        await save_system_config(
-            SystemConfigKey.CustomFilterRules,
-            [rule.model_dump(exclude_none=True) for rule in rules],
-        )
+        final_rules = [*rules, new_rule]
+        await self._save_custom_rules(rules, final_rules)
        return {
            "message": f"已新增自定义过滤规则 {new_rule.id}",
            "custom_rule": serialize_custom_rule(new_rule),
-            "count": len(rules),
+            "count": len(final_rules),
        }

+    async def _save_custom_rules(self, expected: list[CustomRule], final: list[CustomRule]) -> None:
+        """
+        通过现有原子作用域核对规则快照，拒绝并发新增、修改或删除覆盖。
+
+        :param expected: 本次操作读取的旧规则集合
+        :param final: 本次操作希望保存的完整规则集合
+        :raises RuleGroupMutationConflictError: 配置已被另一请求修改，需要重新读取
+        """
+        groups = [group.model_dump(exclude_none=True) for group in get_rule_groups()]
+        definitions = [rule.model_dump(exclude_none=True) for rule in final]
+        async with self._mutation_scope() as mutation:
+            await mutation.apply(groups, expected_rule_groups=groups, custom_rules=definitions,
+                                 expected_custom_rules=[rule.model_dump(exclude_none=True) for rule in expected])
+        await self._publish_config_changed(SystemConfigKey.CustomFilterRules, definitions)
+
    async def reorder_custom(
        self,
        rule_ids: list[str],
@@ -669,10 +681,7 @@ class FilterRuleService:
            await self._publish_config_changed(SystemConfigKey.CustomFilterRules, rule_definitions)
            await self._publish_config_changed(SystemConfigKey.UserFilterRuleGroups, group_definitions)
        else:
-            await save_system_config(
-                SystemConfigKey.CustomFilterRules,
-                rule_definitions,
-            )
+            await self._save_custom_rules(rules, final_rules)
        if updated.id is None:
            raise ValueError("更新后的自定义过滤规则缺少 id")
        refs = collect_custom_rule_group_refs(updated_groups, [updated.id]).get(updated.id, [])
@@ -693,10 +702,7 @@ class FilterRuleService:
        if refs:
            raise ValueError(f"自定义过滤规则 '{rule_id}' 仍被规则组引用: {', '.join(refs)}")
        remaining = [rule for rule in rules if rule.id != rule_id]
-        await save_system_config(
-            SystemConfigKey.CustomFilterRules,
-            [rule.model_dump(exclude_none=True) for rule in remaining],
-        )
+        await self._save_custom_rules(rules, remaining)
        return {
            "message": f"已删除自定义过滤规则 {rule_id}",
            "count": len(remaining),
--- a/app/application/rules.py
+++ b/app/application/rules.py
@@ -773,11 +773,9 @@ class RuleParser:
        """
        解析给定的表达式。

-        参数:
-        expression -- 要解析的表达式
-
-        返回:
-        解析结果
+        :param expression: 待解析的完整规则表达式
+        :return: 逻辑表达式解析结果
+        :raises ParseException: 包含不合法语法或无法消费的尾部内容
        """
        if _filter_rule_parser is not None:
            try:
@@ -786,7 +784,7 @@ class RuleParser:
                accelerated = None
            if accelerated is not None:
                return _RustParseResults(accelerated)
-        return self.expr.parse_string(expression)
+        return self.expr.parse_string(expression, parse_all=True)


 class _RustParseResults(list[Any]):
--- a/app/chain/media/album.py
+++ b/app/chain/media/album.py
@@ -313,11 +313,11 @@ class MediaAlbumOwner(_MediaOwnerBase):
            try:
                stat = path.stat()
            except OSError:
-                signature.append((str(path.relative_to(directory)).casefold(), -1, -1))
+                signature.append((str(path.relative_to(directory)), -1, -1))
            else:
                signature.append(
                    (
-                        str(path.relative_to(directory)).casefold(),
+                        str(path.relative_to(directory)),
                        stat.st_size,
                        stat.st_mtime_ns,
                    )
--- a/app/chain/scraping.py
+++ b/app/chain/scraping.py
@@ -470,6 +470,7 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
        :param fileitem: 关联的媒体文件项
        :param path: 元数据文件路径
        :param content: 文件内容
+        :raises OSError: 存储拒绝上传或本地临时文件写入失败
        """
        if not fileitem or not content or not path:
            return
@@ -494,6 +495,7 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
                logger.info(f"已保存文件：{item.path}")
            else:
                logger.warn(f"文件保存失败：{path}")
+                raise OSError(f"文件保存失败：{path}")
        finally:
            self._cleanup_temp_file(tmp_file_path)

@@ -1455,29 +1457,17 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):

        logger.info(f"开始刮削：{filepath} ...")

-        if mediainfo.type == MediaType.MOVIE:
-            self._handle_movie_scraping(
-                fileitem=fileitem,
-                meta=meta,
-                mediainfo=mediainfo,
-                init_folder=init_folder,
-                parent=parent,
-                overwrite=overwrite,
-                recursive=recursive,
-            )
-        elif mediainfo.type == MediaType.TV:
-            self._handle_tv_scraping(
-                fileitem=fileitem,
-                meta=meta,
-                mediainfo=mediainfo,
-                init_folder=init_folder,
-                parent=parent,
-                overwrite=overwrite,
-                recursive=recursive,
-            )
-        else:
+        handlers = {MediaType.MOVIE: self._handle_movie_scraping, MediaType.TV: self._handle_tv_scraping}
+        handler = handlers.get(mediainfo.type) if mediainfo.type is not None else None
+        if handler is None:
            logger.warn(f"{filepath} 媒体类型不支持刮削：{mediainfo.type}")
            return False, "媒体类型不支持刮削"
+        try:
+            handler(fileitem=fileitem, meta=meta, mediainfo=mediainfo, init_folder=init_folder,
+                    parent=parent, overwrite=overwrite, recursive=recursive)
+        except OSError as error:
+            logger.warning(f"{filepath} 刮削失败：{error}")
+            return False, str(error)

        logger.info(f"{filepath.name} 刮削完成")
        return True, f"{filepath.name} 刮削完成"
@@ -2357,15 +2347,17 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):

        if recursive and not is_bluray_folder:
            for file in files:
-                if file.type == "dir":
+                if file.type == "dir" or not file.path or Path(file.path).suffix.lower() not in self.runtime_config.video_extensions:
                    continue
-                self.scrape_metadata(
+                success, message = self.scrape_metadata(
                    fileitem=file,
                    mediainfo=mediainfo,
                    init_folder=False,
                    parent=fileitem,
                    overwrite=overwrite,
                )
+                if not success:
+                    raise OSError(message)

        if init_folder:
            if is_bluray_folder:
@@ -2494,6 +2486,8 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
        if recursive:
            files = self.storagechain.list_files(fileitem=fileitem) or []
            for file in files:
+                if file.type == "file" and (not file.path or Path(file.path).suffix.lower() not in self.runtime_config.video_extensions):
+                    continue
                if (
                        file.type == "dir"
                        and file.name not in self.runtime_config.season_zero_names
@@ -2501,13 +2495,15 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
                ):
                    # 电视剧不处理非季子目录
                    continue
-                self.scrape_metadata(
+                success, message = self.scrape_metadata(
                    fileitem=file,
                    mediainfo=mediainfo,
                    parent=fileitem if file.type == "file" else None,
                    init_folder=True if file.type == "dir" else False,
                    overwrite=overwrite,
                )
+                if not success:
+                    raise OSError(message)

        if init_folder:
            self._initialize_tv_directory_metadata(
--- a/app/chain/tmdb.py
+++ b/app/chain/tmdb.py
@@ -169,13 +169,8 @@ class TmdbChain(ChainBase):
        """
        获取随机壁纸，缓存1个小时
        """
-        infos = self.tmdb_trending()
-        if infos:
-            while True:
-                info = random.choice(infos)
-                if info and info.backdrop_path:
-                    return info.backdrop_path
-        return None
+        wallpapers = [info.backdrop_path for info in self.tmdb_trending() or [] if info and info.backdrop_path]
+        return random.choice(wallpapers) if wallpapers else None

    def get_trending_wallpapers(self, num: Optional[int] = 10) -> List[str]:
        """
--- a/app/db/adapters/configuration.py
+++ b/app/db/adapters/configuration.py
@@ -8,7 +8,7 @@ import copy
 from collections.abc import Callable
 from typing import Optional, Union

-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session

@@ -50,10 +50,15 @@ class SessionSystemConfigurationRepository:
        """
        同步锁定配置行并返回独立值副本。
        """
-        record = self._sync_session().execute(
+        session = self._sync_session()
+        if session.get_bind().dialect.name == "sqlite":
+            session.execute(update(SystemConfig).where(SystemConfig.key == key.value)
+                            .values(id=SystemConfig.id).execution_options(synchronize_session=False))
+        record = session.execute(
            select(SystemConfig)
            .where(SystemConfig.key == key.value)
            .with_for_update()
+            .execution_options(populate_existing=True)
        ).scalar_one_or_none()
        return copy.deepcopy(record.value if record is not None else None)

@@ -61,10 +66,15 @@ class SessionSystemConfigurationRepository:
        """
        异步锁定配置行并返回独立值副本。
        """
-        result = await self._async_session().execute(
+        session = self._async_session()
+        if session.get_bind().dialect.name == "sqlite":
+            await session.execute(update(SystemConfig).where(SystemConfig.key == key.value)
+                                  .values(id=SystemConfig.id).execution_options(synchronize_session=False))
+        result = await session.execute(
            select(SystemConfig)
            .where(SystemConfig.key == key.value)
            .with_for_update()
+            .execution_options(populate_existing=True)
        )
        record = result.scalar_one_or_none()
        return copy.deepcopy(record.value if record is not None else None)
--- a/app/domain/classification/evaluator.py
+++ b/app/domain/classification/evaluator.py
@@ -476,10 +476,16 @@ def _apply_aliases(value: Any, aliases: Mapping[str, str]) -> Any:
    """
    递归将字段别名投影为策略声明的规范值。
    """
+    if not aliases:
+        return value
    if _is_collection(value):
        return [_apply_aliases(item, aliases) for item in value]
    if isinstance(value, str):
-        return aliases.get(value, value)
+        seen: set[str] = set()
+        while value in aliases and value not in seen:
+            seen.add(value)
+            value = aliases[value]
+        return value
    return value


--- a/app/domain/classification/facts.py
+++ b/app/domain/classification/facts.py
@@ -13,6 +13,9 @@ from app.domain.classification.vocabulary import (
 from app.domain.classification.vocabulary import (
    GENRE_KEY_ALIASES as _GENRE_KEY_ALIASES,
 )
+from app.domain.classification.vocabulary import (
+    TMDB_COMPOSITE_GENRE_KEYS as _TMDB_COMPOSITE_GENRE_KEYS,
+)
 from app.domain.classification.vocabulary import (
    TMDB_GENRE_KEYS as _TMDB_GENRE_KEYS,
 )
@@ -153,6 +156,8 @@ def _classification_genre_keys(media: ClassificationMedia) -> list[str] | None:
    for item in _optional_string_list(getattr(media, "classification_genre_keys", None)) or []:
        _append_unique(values, item)
    for item in getattr(media, "genre_ids", None) or []:
+        for genre_key in _TMDB_COMPOSITE_GENRE_KEYS.get(str(item), ()):
+            _append_unique(values, genre_key)
        if key := _TMDB_GENRE_KEYS.get(str(item)):
            _append_unique(values, key)
    for item in _genre_names(getattr(media, "genres", None)) or []:
--- a/app/domain/classification/validation.py
+++ b/app/domain/classification/validation.py
@@ -516,13 +516,11 @@ class ClassificationPolicyValidator:
                )
            else:
                members = cast(list[ClassificationFactScalar], value)
-                if definition.value_type not in {"string", "enum"} or all(
-                    isinstance(item, str) for item in members
-                ):
+                if all(_valid_scalar_type(definition.value_type, item) for item in members):
                    return
                collector.error(
                    "invalid_condition_value",
-                    f"字段 {condition.field} 的成员值必须是字符串",
+                    f"字段 {condition.field} 的成员值必须符合 {definition.value_type} 类型",
                    [*path, "value"],
                )
            return
@@ -558,14 +556,7 @@ class ClassificationPolicyValidator:
                    [*path, "value"],
                )
            return
-        if definition.value_type in {"integer", "year"}:
-            valid = isinstance(value, int) and not isinstance(value, bool)
-        elif definition.value_type == "number":
-            valid = _is_number(value)
-        elif definition.value_type in {"string", "enum"}:
-            valid = isinstance(value, str)
-        else:
-            valid = value is not None
+        valid = _valid_scalar_type(definition.value_type, value)
        if not valid:
            collector.error(
                "invalid_condition_value",
@@ -867,6 +858,23 @@ def _normalize_path_segment(segment: str) -> str:
    return unicodedata.normalize("NFC", segment).casefold()


+def _valid_scalar_type(value_type: str, value: Any) -> bool:
+    """
+    为标量条件与成员列表复用严格类型规则，布尔值不能冒充数字。
+
+    :param value_type: 字段目录声明的值类型
+    :param value: 待校验的标量值
+    :return: 值符合字段类型时返回 True
+    """
+    if value_type in {"integer", "year"}:
+        return isinstance(value, int) and not isinstance(value, bool)
+    if value_type == "number":
+        return _is_number(value)
+    if value_type in {"string", "enum"}:
+        return isinstance(value, str)
+    return value is not None
+
+
 def _is_number(value: Any) -> bool:
    """
    判断值是否为分类规则支持的整数或浮点数，布尔值不属于数字。
--- a/app/domain/classification/vocabulary.py
+++ b/app/domain/classification/vocabulary.py
@@ -29,6 +29,13 @@ TMDB_GENRE_KEYS = {
    "10770": "tv_movie",
 }

+# 复合电视剧类型只在事实归一化时展开；旧分类迁移仍保留原始 Genre ID 的精确语义。
+TMDB_COMPOSITE_GENRE_KEYS = {
+    "10759": ("action", "adventure"),
+    "10765": ("science_fiction", "fantasy"),
+    "10768": ("war",),
+}
+
 GENRE_KEY_ALIASES = {
    "action": "action",
    "动作": "action",
--- a/app/domain/context.py
+++ b/app/domain/context.py
@@ -1939,7 +1939,7 @@ class MediaInfo:
    :param genre_ids: 风格ID
    :param names: 所有别名和译名
    :param seasons: 各季的剧集清单信息
-    :param season_info: 各季详情
+    :param season_info: 各季详情；IMDb 的 episodes 保存单集 id 与 episode_number，用于独立 NFO 身份
    :param season_years: 各季的年份
    :param category: 兼容字段，过渡期始终与 library_category 同步
    :param library_category: 自动分类或人工覆盖后的媒体库相对目录分类
--- a/app/domain/meta/metabase.py
+++ b/app/domain/meta/metabase.py
@@ -475,7 +475,7 @@ class MetaBase(object):
        if self.begin_episode is None:
            self.begin_episode = begin_episode
            self.end_episode = end_episode
-            self.total_episode = end_episode
+            self.total_episode = end_episode - begin_episode + 1
            self.type = MediaType.TV
            self._subtitle_flag = True

--- a/app/domain/meta/words.py
+++ b/app/domain/meta/words.py
@@ -254,35 +254,16 @@ class WordsMatcher(metaclass=Singleton):
            offset_word_info_re = _compile_custom_word_regex(
                r'(?<=%s.*?)[0-9一二三四五六七八九十]+(?=.*?%s)' % (front, back)
            )
-            episode_nums_str = offset_word_info_re.findall(title)
-            if not episode_nums_str:
-                return title, "", False
-            episode_nums_offset_str = []
-            offset_order_flag = False
-            for episode_num_str in episode_nums_str:
+            def replace_episode(match: re.Match[str]) -> str:
+                """
+                仅按原始匹配文本计算偏移，已生成数字不再参加后续匹配。
+                """
+                episode_num_str = match.group()
                episode_num_int = int(cn2an.cn2an(episode_num_str, "smart"))
                episode_num_offset_int = calculate_episode_offset(offset, episode_num_int)
-                if episode_num_int > episode_num_offset_int:
-                    offset_order_flag = True
-                elif episode_num_int < episode_num_offset_int:
-                    offset_order_flag = False
-                episode_num_offset_str = _format_episode_offset(
-                    episode_num_str, episode_num_offset_int
-                )
-                episode_nums_offset_str.append(episode_num_offset_str)
-            episode_nums_dict = dict(zip(episode_nums_str, episode_nums_offset_str))
-            # 集数向前偏移，集数按升序处理
-            if offset_order_flag:
-                episode_nums_list = sorted(episode_nums_dict.items(), key=lambda x: x[1])
-            # 集数向后偏移，集数按降序处理
-            else:
-                episode_nums_list = sorted(episode_nums_dict.items(), key=lambda x: x[1], reverse=True)
-            for episode_num in episode_nums_list:
-                episode_offset_re = _compile_custom_word_regex(
-                    r'(?<=%s.*?)%s(?=.*?%s)' % (front, episode_num[0], back)
-                )
-                title = episode_offset_re.sub(r'%s' % episode_num[1], title)
-            return title, "", True
+                return _format_episode_offset(episode_num_str, episode_num_offset_int)
+            result, count = offset_word_info_re.subn(replace_episode, title)
+            return result, "", count > 0
        except Exception as err:
            logger.warning(f"自定义识别词集数偏移失败：{str(err)} - 标题：{title}，前定位词：{front}，后定位词：{back}，偏移量：{offset}")
            return title, str(err), False
--- a/app/domain/scraper.py
+++ b/app/domain/scraper.py
@@ -48,13 +48,25 @@ class MediaScraperHelper:
    @staticmethod
    def _media_identity(
        mediainfo: MediaInfo,
+        season: Optional[int] = None,
+        episode: Optional[int] = None,
    ) -> tuple[Optional[MediaSource], Optional[str]]:
        """
        获取媒体信息中的来源与来源原生 ID。

        :param mediainfo: 统一媒体信息
-        :return: 数据源名称与原生 ID
+        :param season: 单集所属季号；主媒体或季元数据时为空
+        :param episode: 单集集号；为空时使用主媒体身份
+        :return: 数据源名称与原生 ID；IMDb 单集缺少独立身份时 ID 为 None，不使用剧集主 ID
        """
+        if mediainfo.media_source == MediaSource.IMDb and season is not None and episode is not None:
+            for detail in mediainfo.season_info:
+                if detail.get("season_number") != season:
+                    continue
+                for item in detail.get("episodes") or []:
+                    if item.get("episode_number") == episode:
+                        return MediaSource.IMDb, item.get("id")
+            return MediaSource.IMDb, None
        return mediainfo.media_source, mediainfo.media_id

    @staticmethod
@@ -74,6 +86,8 @@ class MediaScraperHelper:
        mediainfo: MediaInfo,
        doc: minidom.Document,
        root: minidom.Element,
+        season: Optional[int] = None,
+        episode: Optional[int] = None,
    ) -> None:
        """
        向 NFO 根节点写入各媒体类型共享的标准字段。
@@ -81,6 +95,8 @@ class MediaScraperHelper:
        :param mediainfo: 统一媒体信息
        :param doc: XML 文档
        :param root: NFO 根节点
+        :param season: 单集季号，用于选择来源原生身份
+        :param episode: 单集集号，不使用剧集主身份代替单集
        """
        DomUtils.add_node(doc, root, "title", mediainfo.title or "")
        DomUtils.add_node(doc, root, "originaltitle", mediainfo.original_title or "")
@@ -93,7 +109,7 @@ class MediaScraperHelper:
        outline = DomUtils.add_node(doc, root, "outline")
        outline.appendChild(doc.createCDATASection(mediainfo.overview or ""))

-        source, media_id = cls._media_identity(mediainfo)
+        source, media_id = cls._media_identity(mediainfo, season=season, episode=episode)
        if source and media_id:
            unique_id = DomUtils.add_node(doc, root, "uniqueid", media_id)
            unique_id.setAttribute("type", str(source))
@@ -158,7 +174,7 @@ class MediaScraperHelper:
            cls._append_common_nodes(mediainfo, doc, root)
        elif season is not None and episode is not None:
            root = DomUtils.add_node(doc, doc, "episodedetails")
-            cls._append_common_nodes(mediainfo, doc, root)
+            cls._append_common_nodes(mediainfo, doc, root, season=season, episode=episode)
            DomUtils.add_node(doc, root, "season", str(season))
            DomUtils.add_node(doc, root, "episode", str(episode))
            DomUtils.add_node(
--- a/app/modules/douban/apiv2.py
+++ b/app/modules/douban/apiv2.py
@@ -368,13 +368,17 @@ class DoubanApi(metaclass=WeakSingleton):
        resp: Optional[_JsonResponse]
    ) -> Optional[dict[str, Any]]:
        """
-        读取响应 JSON；未取得 HTTP 响应时返回 None。
+        仅接受成功 HTTP 和业务响应，避免把错误对象保存为长期详情缓存。

-        不在本层校验业务状态；JSON 解码异常交由调用方处理。
+        JSON 解码异常仍交由调用方处理；合法空列表响应保持原合同。
        """
-        if resp is None:
+        if resp is None or resp.status_code != 200:
            return None
        payload: dict[str, Any] = resp.json()
+        if not isinstance(payload, (dict, list)):
+            return None
+        if isinstance(payload, dict) and payload.get("code") not in (None, 0, "0"):
+            return None
        return payload

    @_cached_api_request(shared_key="get")
--- a/app/modules/imdb/init.py
+++ b/app/modules/imdb/init.py
@@ -5,7 +5,7 @@ IMDb 原生媒体数据源 Module。
 import asyncio
 import re
 from dataclasses import dataclass
-from typing import Any, Optional, Tuple, Union
+from typing import Any, Optional, Tuple, Union, cast

 from app.domain.context import MediaInfo
 from app.domain.media import is_media_source_enabled
@@ -583,6 +583,8 @@ class ImdbModule(MediaAuxiliaryProviderMixin, _ModuleBase):
                    "name": str(season_number),
                },
            )
+            episode_details = cast(list[dict[str, object]], season_info[season_number].setdefault("episodes", []))
+            episode_details.append({"id": episode.id, "episode_number": episode.episode_number})
        media_info.season_info = list(season_info.values())
        if media_type == MediaType.TV:
            media_info.number_of_seasons = len(season_info)
--- a/app/modules/musicbrainz/init.py
+++ b/app/modules/musicbrainz/init.py
@@ -2166,7 +2166,7 @@ class MusicBrainzModule(_ModuleBase):
        r"MP3|AAC|M4A|OGG|VORBIS|OPUS|WMA|WEB-?DL|WEBRip|WEB)\b|"
        r"\b\d{1,3}\s*-?\s*bits?\b|\b\d{2,4}(?:\.\d)?\s*k(?:hz|bps?)\b|"
        # 流媒体发行实体标记（- Single / - EP），不是曲名的一部分
-        r"\b(?:single|ep|album)\b",
+        r"\s+[-–—]\s*(?:single|ep|album)\s*$",
        re.IGNORECASE,
    )

--- a/app/modules/themoviedb/tmdbapi.py
+++ b/app/modules/themoviedb/tmdbapi.py
@@ -232,13 +232,11 @@ class TmdbApi:
    @staticmethod
    def _project_search_results(title: str, items: List[_MediaRecord], mtype: MediaType) -> List[_MediaRecord]:
        """
-        按媒体类型过滤标题并补齐统一类型字段。
+        为来源已完成搜索匹配的候选补齐类型，不按本地化展示标题二次排除。
        """
-        title_field = "title" if mtype == MediaType.MOVIE else "name"
+        del title
        ret_infos = []
        for item in items:
-            if title not in (item.get(title_field) or ""):
-                continue
            item["media_type"] = mtype
            ret_infos.append(item)
        return ret_infos
--- a/docs/architecture-overview.md
+++ b/docs/architecture-overview.md
@@ -817,7 +817,7 @@ flowchart LR
 | 指标 | 当前值 |
 |---|---:|
 | Python 模块 | 1129 |
-| 内部导入边 | 9,706 |
+| 内部导入边 | 9,707 |
 | 非平凡 SCC | 1（29 个模块的 TMDB vendor 精确 containment） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 51（精确 policy 准入，Application / Chain 均为 0） |
--- a/docs/mcp-api.md
+++ b/docs/mcp-api.md
@@ -340,6 +340,7 @@ Emby、Jellyfin、Plex、Navidrome 的目录、条目、搜索、季集和活动
 - 设置更新优先回传 `expected_revision`。发生冲突时重新读取并比较，不直接覆盖；`generic_write_allowed=false` 时使用返回的专用操作。
 - 插件存在分身时，显式选择目标实例或使用已经配置的默认目标。遇到实例歧义、停用目标或插件门禁拒绝时，按错误处理，不猜测另一个实例。
 - 参数省略、空字符串与 `null` 的含义由具体操作决定。例如更新订阅时，省略字段通常保留原值，允许清空的字段可使用空值；不要把整份读取结果不加筛选地回写。
+- 自定义过滤规则的新增、修改或删除遇到并发快照冲突时返回业务失败。先重新读取规则集合，再重新确认修改，不用旧集合覆盖已成功提交的配置。
 - `search.torrents` 可能执行多轮站点搜索，工具等待上限为 300 秒；客户端应为此设置足够的超时。分页只限制返回结果，不会缩短搜索过程。

 ## 6. 处理结果、分页与失败恢复
@@ -354,6 +355,8 @@ Emby、Jellyfin、Plex、Navidrome 的目录、条目、搜索、季集和活动

 `execution_outcome: "pending"` 表示已提交但尚未确认最终完成；`"unknown"` 表示效果未知，先用只读操作核验实际状态。没有该字段时，依据具体业务响应判断，不能自动推断异步任务完成。

+`media.scrape` 遇到不支持的媒体文件或 NFO 保存失败时返回 `success: false` 及原因，目录中的媒体保存失败也会向上传递。不要将这种响应报告为刮削完成。
+
 ### 分页与截断

 - 按选定操作的合同使用分页字段。例如 `subscription.list` 使用 `page` / `count`，`subscription.execution.list` 使用 `limit`，原生服务列表常用 `offset` / `limit`。
--- a/scripts/generate_agent_skill_docs.py
+++ b/scripts/generate_agent_skill_docs.py
@@ -17,7 +17,7 @@ PROJECT_ROOT = Path(__file__).resolve().parents[1]
 if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

-MOVIEPILOT_API_SKILL_VERSION = "59"
+MOVIEPILOT_API_SKILL_VERSION = "60"
 DATABASE_OPERATION_SKILL_VERSION = "12"


--- a/skills/moviepilot-api/SKILL.md
+++ b/skills/moviepilot-api/SKILL.md
@@ -1,6 +1,6 @@
 ---
 name: moviepilot-api
-version: 59
+version: 60
 description: >-
  Use this skill for MoviePilot product operations such as media search, torrent
  search, downloads, subscriptions, library checks, sites, storage, workflows,
@@ -325,6 +325,9 @@ token exchange downloads retain their existing credential rules.
  `music_type=recording|album|artist`; an artist is browse-only.
 - Treat `success=false`, HTTP error data, empty results, and validation errors as
  real outcomes. Do not claim success without checking the response.
+- `media.scrape` reports rejected media and failed NFO writes as business failures,
+  including failures within a directory. For concurrent custom-filter changes,
+  re-read the rules and confirm the intended change before retrying a conflict.
 - Respect an explicit `execution_outcome`: `pending` is accepted but unfinished,
  while `unknown` means a write may have happened. Do not repeat an unknown write
  or change defaults merely to evade duplicate protection. In the built-in Agent,
--- a/tests/fixtures/architecture/complexity-baseline.json
+++ b/tests/fixtures/architecture/complexity-baseline.json
@@ -72,7 +72,6 @@
      "app/chain/transfer/workflow.py:TransferWorkflowOwner._collect_transfer_candidates": 16,
      "app/chain/transfer/workflow.py:TransferWorkflowOwner._execute_transfer_tasks": 18,
      "app/chain/workflow.py:WorkflowExecutor.evaluate_target_state": 16,
-      "app/domain/classification/validation.py:ClassificationPolicyValidator._validate_condition_value": 18,
      "app/domain/meta/metaanime.py:MetaAnime.__init__": 46,
      "app/domain/meta/metaanime.py:MetaAnime.__prepare_title": 16,
      "app/domain/meta/metabase.py:MetaBase.init_subtitle": 41,
--- a/tests/fixtures/architecture/dependency-baseline.json
+++ b/tests/fixtures/architecture/dependency-baseline.json
@@ -961,8 +961,8 @@
      "runtime_only": true
    }
  },
-  "edge_count": 9706,
-  "edge_sha256": "17f7a368a31cf6d8ade3d18e6a5219249c8b82c436e290ac3173cbf3e5759b13",
+  "edge_count": 9707,
+  "edge_sha256": "eb91f16e795a22fafba1b5742c8f4c591c9e254e6995799b2ce1b4d838275bc5",
  "edges": [
    "app -> app.foundation",
    "app -> app.foundation.environment",
@@ -2965,6 +2965,7 @@
    "app.api.endpoints.rule -> app.api.response",
    "app.api.endpoints.rule -> app.application",
    "app.api.endpoints.rule -> app.application.filtering",
+    "app.api.endpoints.rule -> app.application.rules",
    "app.api.endpoints.rule -> app.schemas",
    "app.api.endpoints.rule -> app.schemas.common",
    "app.api.endpoints.rule -> app.schemas.response",
--- a/tests/test_builtin_skill_boundaries.py
+++ b/tests/test_builtin_skill_boundaries.py
@@ -130,7 +130,7 @@ def test_modified_builtin_skills_have_incremented_versions() -> None:
        "database-operation": "12",
        "feedback-issue": "12",
        "anysearch": "3",
-        "moviepilot-api": "59",
+        "moviepilot-api": "60",
        "moviepilot-update": "7",
        "organize-files": "6",
        "transfer-failed-retry": "6",
--- a/tests/test_douban_apiv2_types.py
+++ b/tests/test_douban_apiv2_types.py
@@ -28,7 +28,7 @@ def test_sync_async_keep_response_and_cache_contract(
    """
    api = object.__new__(DoubanApi)
    payload = {"id": "123", "nested": {"items": []}}
-    response = None if missing else Mock(json=Mock(return_value=payload))
+    response = None if missing else Mock(status_code=200, json=Mock(return_value=payload))
    sync_request = Mock(return_value=response)
    async_request = AsyncMock(return_value=response)
    api._request = Mock(**{f"{verb}_res": sync_request})
--- a/tests/test_douban_recommendations.py
+++ b/tests/test_douban_recommendations.py
@@ -34,7 +34,7 @@ def test_recommendations_preserve_native_lists_and_compatible_mappings(
    }
    payload = payloads[shape]
    original = deepcopy(payload)
-    response = None if payload is None else Mock(json=Mock(return_value=payload))
+    response = None if payload is None else Mock(status_code=200, json=Mock(return_value=payload))
    api = object.__new__(DoubanApi)
    api._request = Mock(get_res=Mock(return_value=response))
    async_request = AsyncMock(return_value=response)
--- a/tests/test_filter_rule_review_concurrency.py
+++ b/tests/test_filter_rule_review_concurrency.py
@@ -0,0 +1,154 @@
+"""
+通过真实配置仓储和事务作用域验证自定义过滤规则并发保存。
+"""
+
+import asyncio
+from functools import partial
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+
+from app.api.endpoints import rule as rule_endpoint
+from app.application import filtering, rules
+from app.application.configuration import SystemConfigService
+from app.application.rules import AsyncRuleGroupMutationService, RuleGroupMutationConflictError
+from app.db.adapters.configuration import SessionSystemConfigurationRepository
+from app.db.adapters.subscription import SessionSubscriptionRepository
+from app.db.models.subscribe import Subscribe
+from app.db.models.systemconfig import SystemConfig
+from app.db.oper.systemconfig import SystemConfigOper
+from app.db.uow import SqlAlchemyAsyncUnitOfWork
+from app.schemas.rule import CustomFilterRuleCreateRequest, CustomFilterRuleUpdateRequest
+from app.schemas.types import SystemConfigKey
+from app.startup.composition import subscription as composition
+
+
+@pytest.mark.parametrize("operation", ["add", "update", "delete"])
+def test_custom_rule_mutations_reject_stale_snapshots(db, monkeypatch, operation) -> None:
+    """
+    两个请求都读取旧快照后，冲突请求明确失败，不覆盖另一已确认成功的规则。
+    """
+    db.watermark(SystemConfig, Subscribe)
+    original = [{"id": "A", "name": "Rule A"}, {"id": "B", "name": "Rule B"}]
+    config = SystemConfigOper()
+    db.add(SystemConfig(key=SystemConfigKey.CustomFilterRules.value, value=original),
+           SystemConfig(key=SystemConfigKey.UserFilterRuleGroups.value, value=[]))
+    config.load_snapshot(db.session)
+    system_service = SystemConfigService(config, async_executor=SimpleNamespace(run=asyncio.to_thread))
+    monkeypatch.setattr(filtering, "get_configured_system_config", lambda: system_service)
+    monkeypatch.setattr(rules, "get_configured_system_config", lambda: system_service)
+    scope = partial(composition.async_rule_group_mutation_scope, config.publish_many)
+    service = filtering.FilterRuleService(None, scope, AsyncMock())
+
+    async def execute():
+        """
+        在首次持久化边界暂停，确保两个异步用例都基于相同旧配置构造意图。
+        """
+        arrived = 0
+        ready = asyncio.Event()
+        original_scope = service._mutation_scope
+        from contextlib import asynccontextmanager
+
+        @asynccontextmanager
+        async def synchronized_scope():
+            """
+            让两个修改同时到达持久化边界，验证旧快照只能提交一次。
+            """
+            nonlocal arrived
+            arrived += 1
+            if arrived == 2:
+                ready.set()
+            await asyncio.wait_for(ready.wait(), 5)
+            async with original_scope() as mutation:
+                yield mutation
+
+        service._mutation_scope = synchronized_scope
+        if operation == "add":
+            tasks = [service.add_custom(rule_id=key, name=f"Rule {key}", include=None, exclude=None,
+                                       size_range=None, seeders=None, publish_time=None) for key in ("C", "D")]
+        elif operation == "update":
+            tasks = [service.update_custom(current_rule_id=key, name=f"Updated {key}") for key in ("A", "B")]
+        else:
+            tasks = [service.delete_custom(key) for key in ("A", "B")]
+        return await asyncio.wait_for(asyncio.gather(*tasks, return_exceptions=True), 10)
+
+    outcomes = asyncio.run(execute())
+    conflicts = [value for value in outcomes if isinstance(value, RuleGroupMutationConflictError)]
+    assert len(conflicts) == 1, outcomes
+    assert sum(isinstance(value, dict) for value in outcomes) == 1, outcomes
+    db.session.expire_all()
+    stored = db.session.query(SystemConfig).filter_by(key=SystemConfigKey.CustomFilterRules.value).one().value
+    if operation == "add":
+        assert {item["id"] for item in stored} in ({"A", "B", "C"}, {"A", "B", "D"})
+    elif operation == "update":
+        assert sum(item["name"].startswith("Updated") for item in stored) == 1
+    else:
+        assert len(stored) == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["add", "update", "delete"])
+async def test_custom_rule_api_returns_explicit_conflict(monkeypatch, operation) -> None:
+    """
+    过期快照作为可重读后重试的业务失败返回，不能退化为未处理的服务器异常。
+    """
+    service = SimpleNamespace(**{f"{operation}_custom": AsyncMock(
+        side_effect=RuleGroupMutationConflictError("规则已更新，请重新读取"))})
+    monkeypatch.setattr(rule_endpoint, "_service", lambda _runtime: service)
+    if operation == "add":
+        result = await rule_endpoint.add_custom_rule(CustomFilterRuleCreateRequest(rule_id="R", name="Rule"), runtime=None)
+    elif operation == "update":
+        result = await rule_endpoint.update_custom_rule("R", CustomFilterRuleUpdateRequest(name="Updated"), runtime=None)
+    else:
+        result = await rule_endpoint.delete_custom_rule("R", runtime=None)
+    assert result.success is False
+    assert result.message == "规则已更新，请重新读取"
+
+
+@pytest.mark.asyncio
+async def test_independent_sqlite_scopes_reject_rule_snapshot_conflict(tmp_path) -> None:
+    """
+    不共享进程锁的两个真实会话也不能同时提交基于旧规则集合的不同变更。
+    """
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'rules.db'}")
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(SystemConfig.__table__.create)
+            await connection.run_sync(Subscribe.__table__.create)
+        async with AsyncSession(engine) as session:
+            session.add_all([SystemConfig(key=key.value, value=[]) for key in (
+                SystemConfigKey.UserFilterRuleGroups, SystemConfigKey.CustomFilterRules)])
+            await session.commit()
+        ready = asyncio.Event()
+        arrived = 0
+
+        async def mutate(rule_id):
+            """
+            每个命令拥有独占会话和事务，只通过数据库竞争协调。
+            """
+            nonlocal arrived
+            async with AsyncSession(engine) as session:
+                service = AsyncRuleGroupMutationService(
+                    SessionSystemConfigurationRepository(session), SessionSubscriptionRepository(session),
+                    SqlAlchemyAsyncUnitOfWork(session), AsyncMock())
+                arrived += 1
+                if arrived == 2:
+                    ready.set()
+                await ready.wait()
+                try:
+                    return await service.apply([], expected_rule_groups=[],
+                                               custom_rules=[{"id": rule_id, "name": rule_id}], expected_custom_rules=[])
+                except RuleGroupMutationConflictError as error:
+                    return error
+
+        outcomes = await asyncio.wait_for(asyncio.gather(mutate("A"), mutate("B")), 10)
+        assert sum(isinstance(value, RuleGroupMutationConflictError) for value in outcomes) == 1
+        async with AsyncSession(engine) as session:
+            row = (await session.execute(select(SystemConfig).where(
+                SystemConfig.key == SystemConfigKey.CustomFilterRules.value))).scalar_one()
+            assert len(row.value) == 1
+    finally:
+        await engine.dispose()
--- a/tests/test_media_scrape_endpoint.py
+++ b/tests/test_media_scrape_endpoint.py
@@ -29,6 +29,7 @@ def test_scrape_uses_explicit_media_source_and_id() -> None:
    chain.recognize_media.return_value = media_info

    scraping_chain = Mock()
+    scraping_chain.scrape_metadata.return_value = (True, "刮削完成")
    with patch("app.api.endpoints.media.MediaChain", return_value=chain) as mock_chain, \
            patch("app.api.endpoints.media.ScrapingChain", return_value=scraping_chain):
        # mkv 非音频文件，需显式关闭 Mock 的 is_audio_path 避免误入音乐分支
@@ -69,6 +70,7 @@ def test_scrape_keeps_automatic_recognition_compatible() -> None:
    chain.recognize_by_path.return_value = Context(meta_info=meta_info, media_info=media_info)

    scraping_chain = Mock()
+    scraping_chain.scrape_metadata.return_value = (True, "刮削完成")
    with patch("app.api.endpoints.media.MediaChain", return_value=chain) as mock_chain, \
            patch("app.api.endpoints.media.ScrapingChain", return_value=scraping_chain):
        # mkv 非音频文件，需显式关闭 Mock 的 is_audio_path 避免误入音乐分支
@@ -150,6 +152,7 @@ def _post_scrape_as(monkeypatch, current_user):
        media_info=MediaInfo(title="测试剧集", type=MediaType.TV),
    )
    scraping_chain = Mock()
+    scraping_chain.return_value.scrape_metadata.return_value = (True, "刮削完成")
    monkeypatch.setattr(media_endpoint, "MediaChain", media_chain)
    monkeypatch.setattr(media_endpoint, "ScrapingChain", scraping_chain)

--- a/tests/test_recognition_rules_review_regressions.py
+++ b/tests/test_recognition_rules_review_regressions.py
@@ -0,0 +1,336 @@
+"""
+识别、分类、搜索和刮削审查发现的真实行为回归。
+"""
+
+import copy
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock, PropertyMock
+from xml.dom import minidom
+
+import pytest
+from pyparsing import ParseBaseException
+
+from app.api.endpoints import media as media_endpoint
+from app.api.endpoints import recommend
+from app.application import rules
+from app.application.classification.configuration import build_default_classification_policy
+from app.chain.media import MediaChain
+from app.chain.scraping import ScrapingChain, ScrapingConfig
+from app.chain.tmdb import TmdbChain
+from app.domain.classification.evaluator import ClassificationEvaluator
+from app.domain.classification.facts import build_classification_facts
+from app.domain.classification.validation import ClassificationPolicyValidator
+from app.domain.context import MediaInfo, TorrentInfo
+from app.domain.meta.words import WordsMatcher
+from app.domain.metainfo import MetaInfo
+from app.domain.scraper import MediaScraperHelper
+from app.modules.douban.apiv2 import DoubanApi
+from app.modules.filter import FilterModule
+from app.modules.imdb import ImdbModule
+from app.modules.imdb.api import ImdbEpisode, ImdbTitle
+from app.modules.musicbrainz import MusicBrainzModule
+from app.modules.themoviedb.tmdbapi import TmdbApi
+from app.schemas.category import ClassificationCategory, ClassificationRule
+from app.schemas.file import FileItem
+from app.schemas.types import MediaSource, MediaType
+
+
+def test_album_signature_preserves_case_sensitive_track_paths(tmp_path) -> None:
+    """
+    同大小、同时间戳的大小写重命名必须使旧路径缓存失效。
+    """
+    old = tmp_path / "track.flac"
+    old.write_bytes(b"isolated-audio")
+    first = MediaChain._album_directory_signature(tmp_path, [old])
+    new = old.rename(tmp_path / "Track.flac")
+    second = MediaChain._album_directory_signature(tmp_path, [new])
+    assert first != second
+
+
+def test_fin_partial_range_uses_count_in_real_size_filter(monkeypatch) -> None:
+    """
+    非首集开头的完结合集按实际集数计算每集大小。
+    """
+    monkeypatch.setattr("app.domain.metainfo.get_metainfo_accelerator", lambda: None)
+    monkeypatch.setattr("app.modules.filter.rust_accel.filter_torrents", lambda *args, **kwargs: None)
+    torrent = TorrentInfo(title="Review Show S01 2026 1080p WEB-DL H264",
+                          description="Review Show [11-26Fin]", size=16 * 1024**3)
+    meta = MetaInfo(torrent.title, torrent.description)
+    module = object.__new__(FilterModule)
+    module.rule_set = {"R": {"size_range": ">800"}}
+    module.rulehelper = SimpleNamespace(get_rule_group_by_media=lambda **_kw: [
+        SimpleNamespace(name="review size", rule_string="R")])
+    assert meta.total_episode == 16
+    assert module.filter_torrents(["review size"], [torrent]) == [torrent]
+
+
+@pytest.mark.parametrize("title,expected", [
+    ("Review E1-E11", "Review E2-E12"),
+    ("Review E9-E99", "Review E10-E100"),
+    ("Review E09-E10", "Review E10-E11"),
+])
+def test_episode_offset_never_rewrites_its_output(title, expected) -> None:
+    """
+    不同位数及重叠前缀的集号均只基于原始匹配计算一次。
+    """
+    prepared, _ = WordsMatcher().prepare(title, ["E <> $ >> EP+1"])
+    assert prepared == expected
+
+
+def test_wallpaper_without_any_backdrop_is_bounded(monkeypatch) -> None:
+    """
+    全部候选缺少背景图时不再无限抽取同一批结果。
+    """
+    chain = object.__new__(TmdbChain)
+    info = MediaInfo(title="No artwork")
+    chain.tmdb_trending = lambda: [info]
+    monkeypatch.setattr("app.chain.tmdb.random.choice", Mock(side_effect=[info, AssertionError("重复抽取空背景图")]))
+    assert chain.get_random_wallpager() is None
+
+
+@pytest.mark.parametrize("title", ["Single Ladies", "The Album", "EP Collection"])
+def test_musicbrainz_preserves_natural_release_title_words(title) -> None:
+    """
+    真实作品名称中的发行词不被当作资源规格全局剥离。
+    """
+    assert MusicBrainzModule._search_title(title) == title
+
+
+def _classification_policy(field, operator, value, *, aliases=None):
+    """
+    构造经过标准 Schema 的最小分类规则和别名策略。
+    """
+    policy = build_default_classification_policy()
+    policy.categories.append(ClassificationCategory(id="review.hit", name="命中", path=["命中"], media_type="电影"))
+    policy.rules = [ClassificationRule.model_validate({
+        "id": "review.rule", "name": "审查", "kind": "category", "media_types": ["电影"],
+        "when": {"field": field, "operator": operator, "value": value}, "target": {"category_id": "review.hit"}})]
+    policy.field_aliases = {field: aliases} if aliases else {}
+    return policy
+
+
+def test_classification_resolves_valid_alias_chain() -> None:
+    """
+    发布允许的无环多级别名在实际事实求值中命中规范值。
+    """
+    policy = _classification_policy("media.language", "equals", "zh", aliases={"普通话": "中文", "中文": "zh"})
+    assert ClassificationPolicyValidator().validate(policy).valid
+    media = MediaInfo(media_source=MediaSource.TMDB, media_id="1", type=MediaType.MOVIE,
+                     title="Review", original_language="普通话")
+    result = ClassificationEvaluator().evaluate(policy, build_classification_facts(media))
+    assert result.result.effective.category_id == "review.hit"
+
+
+@pytest.mark.parametrize("field,operator,value,aliases", [
+    ("media.year", "in", [2026], {"2026": "错误字符串替换"}),
+    ("media.adult", "is_true", None, {"True": "错误字符串替换"}),
+])
+def test_classification_aliases_preserve_numeric_and_boolean_facts(field, operator, value, aliases) -> None:
+    """
+    有效别名策略只映射字符串，数字与布尔事实不得被字符串化或改变比较类型。
+    """
+    policy = _classification_policy(field, operator, value, aliases=aliases)
+    assert ClassificationPolicyValidator().validate(policy).valid
+    facts = build_classification_facts(MediaInfo(
+        media_source=MediaSource.TMDB, media_id="1", type=MediaType.MOVIE,
+        title="Review", year="2026", adult=True,
+    ))
+    before = facts.model_dump()
+    result = ClassificationEvaluator().evaluate(policy, facts, trace=True)
+    assert result.result.effective.category_id == "review.hit"
+    assert facts.model_dump() == before
+    actual = result.trace[0].conditions[0].actual
+    assert type(actual) is (int if field == "media.year" else bool)
+
+
+def test_classification_collection_alias_chains_preserve_original_trace() -> None:
+    """
+    集合事实和期望成员均按多级别名命中，同时保留原始事实与解释轨迹。
+    """
+    policy = _classification_policy("media.countries", "contains_any", ["中国"], aliases={
+        "CHN": "PRC", "PRC": "CN", "中国": "中华", "中华": "CN",
+    })
+    assert ClassificationPolicyValidator().validate(policy).valid
+    facts = build_classification_facts(MediaInfo(
+        media_source=MediaSource.TMDB, media_id="1", type=MediaType.MOVIE,
+        title="Review", origin_country=["CHN", "US"],
+    ))
+    result = ClassificationEvaluator().evaluate(policy, facts, trace=True)
+    assert result.result.effective.category_id == "review.hit"
+    assert facts.media.countries == ["CHN", "US"]
+    assert result.trace[0].conditions[0].actual == ["CHN", "US"]
+    assert result.trace[0].conditions[0].expected == ["中国"]
+
+
+@pytest.mark.parametrize("value", [["2024"], [True], [2024.5]])
+def test_classification_rejects_wrong_numeric_member_types(value) -> None:
+    """
+    数值年份成员列表在发布前拒绝字符串、布尔值和非整数。
+    """
+    policy = _classification_policy("media.year", "in", value)
+    assert not ClassificationPolicyValidator().validate(policy).valid
+
+
+@pytest.mark.parametrize("genre,expected", [
+    (10759, {"action", "adventure"}), (10765, {"science_fiction", "fantasy"}),
+    (10768, {"war"}),
+])
+def test_classification_normalizes_composite_tmdb_tv_genres(genre, expected) -> None:
+    """
+    电视剧复合来源类型展开为可跨来源匹配的标准键。
+    """
+    media = MediaInfo(tmdb_info={"id": 2, "media_type": MediaType.TV, "name": "Review", "genre_ids": [genre]})
+    assert expected <= set(build_classification_facts(media).media.genre_keys or [])
+
+
+@pytest.mark.parametrize("mtype,title_field,original_field", [
+    (MediaType.MOVIE, "title", "original_title"), (MediaType.TV, "name", "original_name")])
+def test_tmdb_localized_candidates_keep_original_title_matches(mtype, title_field, original_field) -> None:
+    """
+    已由来源搜索命中的原文结果不能因本地化展示标题被排除。
+    """
+    candidate = {"id": 603, title_field: "黑客帝国", original_field: "The Matrix"}
+    assert TmdbApi._project_search_results("the matrix", [candidate], mtype) == [candidate]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize("status,payload", [(404, {"code": 1001, "msg": "not found"}),
+                                           (200, {"code": 1001, "msg": "not found"})])
+async def test_douban_failed_detail_does_not_poison_recovery(monkeypatch, asynchronous, status, payload) -> None:
+    """
+    失败响应不进入长期详情缓存，远端恢复后同一请求立即得到合法身份。
+    """
+    api = object.__new__(DoubanApi)
+    responses = [SimpleNamespace(status_code=status, json=lambda: copy.deepcopy(payload)),
+                 SimpleNamespace(status_code=200, json=lambda: {"id": "123456", "title": "Review", "type": "movie"})]
+    request = Mock(side_effect=responses)
+    api._request = SimpleNamespace(get_res=request)
+    async def get_res(**kwargs):
+        """
+        在异步入口复用失败后恢复的同一响应序列。
+        """
+        return request(**kwargs)
+    monkeypatch.setattr("app.modules.douban.apiv2.AsyncRequestUtils", lambda **_kwargs: SimpleNamespace(get_res=get_res))
+    api.clear_cache()
+    try:
+        operation = api.async_movie_detail if asynchronous else api.movie_detail
+        first = await operation("123456") if asynchronous else operation("123456")
+        second = await operation("123456") if asynchronous else operation("123456")
+        assert first is None
+        assert second["id"] == "123456"
+        assert request.call_count == 2
+    finally:
+        api.clear_cache()
+
+
+@pytest.mark.asyncio
+async def test_unified_music_search_can_return_second_page(monkeypatch) -> None:
+    """
+    音乐来源请求足够的候选，统一接口再对完整集合分页。
+    """
+    async def search(**kwargs):
+        """
+        按真实请求上限返回候选，以核对接口的第二页切片。
+        """
+        return [SimpleNamespace(to_dict=lambda i=i: {"title": f"Song {i}", "year": "2024"})
+                for i in range(min(kwargs["limit"], 6))]
+    monkeypatch.setattr(media_endpoint, "MediaChain", lambda: SimpleNamespace(async_search_music=search))
+    result = await media_endpoint.search(title="Review", type="music", page=2, count=2, media_source=())
+    assert [item["title"] for item in result] == ["Song 2", "Song 3"]
+
+
+def test_imdb_episode_nfo_has_episode_identity() -> None:
+    """
+    两个单集各自使用 IMDb 原生身份，不能写入同一剧集主 ID。
+    """
+    module = object.__new__(ImdbModule)
+    module.scraper = MediaScraperHelper()
+    episodes = [ImdbEpisode(id=f"tt100000{i}", title=f"Episode {i}", season="1", episode_number=i) for i in (1, 2)]
+    show = module._to_media_info(ImdbTitle(id="tt1000000", type="tvSeries", primary_title="Review"), episodes=episodes)
+    show.scrape_source = "imdb"
+    for item in episodes:
+        document = minidom.parseString(module.metadata_nfo(show, season=1, episode=item.episode_number))
+        identity = document.getElementsByTagName("uniqueid")[0]
+        assert identity.firstChild.data == item.id
+    assert show.media_id == "tt1000000"
+    missing = minidom.parseString(module.metadata_nfo(show, season=1, episode=99))
+    assert not missing.getElementsByTagName("uniqueid")
+
+
+@pytest.mark.parametrize("extension,failed_upload", [(".txt", False), (".mkv", True)])
+def test_manual_scrape_propagates_rejection_and_failed_upload(monkeypatch, tmp_path, extension, failed_upload) -> None:
+    """
+    真实刮削拒绝或 NFO 上传失败时，API 不再返回完成。
+    """
+    path = tmp_path / ("Review" + extension)
+    path.write_bytes(b"review-fixture")
+    chain = object.__new__(ScrapingChain)
+    config = SimpleNamespace(audio_extensions=(".flac",), video_extensions=(".mkv",))
+    monkeypatch.setattr(ScrapingChain, "runtime_config", PropertyMock(return_value=config))
+    info = MediaInfo(media_source=MediaSource.TMDB, media_id="1", type=MediaType.MOVIE, title="Review")
+    provider = Mock(return_value=SimpleNamespace(
+        recognize_media=lambda **_kwargs: info, obtain_images=lambda **_kwargs: info))
+    provider.is_audio_path.return_value = False
+    monkeypatch.setattr(media_endpoint, "MediaChain", provider)
+    monkeypatch.setattr(media_endpoint, "ScrapingChain", lambda: chain)
+    upload = Mock(return_value=None)
+    chain.storagechain = SimpleNamespace(
+        get_parent_item=lambda _item: FileItem(storage="local", type="dir", path=tmp_path.as_posix()),
+        get_file_item=lambda **_kwargs: None, upload_file=upload)
+    chain.scraping_policies = ScrapingConfig()
+    chain.metadata_nfo = lambda **kwargs: MediaScraperHelper.get_metadata_nfo(kwargs["mediainfo"])
+    chain.metadata_img = lambda **_kwargs: None
+    response = media_endpoint._scrape_impl(FileItem(storage="local", type="file", path=path.as_posix()),
+                                          storage="local", media_source=MediaSource.TMDB, media_id="1", type_name=MediaType.MOVIE)
+    assert not response.success
+    assert upload.called is failed_upload
+
+
+def test_directory_scrape_propagates_child_upload_failure(monkeypatch, tmp_path) -> None:
+    """
+    子文件的 NFO 上传失败传到目录结果，普通附属文件仍允许跳过。
+    """
+    video = tmp_path / "Review.mkv"
+    video.write_bytes(b"isolated-video")
+    note = tmp_path / "readme.txt"
+    note.write_text("isolated-note")
+    chain = object.__new__(ScrapingChain)
+    monkeypatch.setattr(ScrapingChain, "runtime_config", PropertyMock(return_value=SimpleNamespace(
+        audio_extensions=(".flac",), video_extensions=(".mkv",))))
+    directory = FileItem(storage="local", type="dir", path=tmp_path.as_posix())
+    children = [FileItem(storage="local", type="file", path=path.as_posix()) for path in (note, video)]
+    chain.storagechain = SimpleNamespace(
+        list_files=lambda **_kwargs: children, contains_bluray_subdirectories=lambda _files: False,
+        get_parent_item=lambda _item: directory, get_file_item=lambda **_kwargs: None,
+        upload_file=lambda **_kwargs: None)
+    chain.scraping_policies = ScrapingConfig()
+    chain.metadata_nfo = lambda **_kwargs: "<movie/>"
+    chain.metadata_img = lambda **_kwargs: None
+    result = chain.scrape_metadata(fileitem=directory, mediainfo=MediaInfo(type=MediaType.MOVIE, title="Review"))
+    assert result[0] is False
+
+
+@pytest.mark.parametrize("expression", ["4K &", "4K & !BLU)", "4K garbage"])
+def test_filter_rule_parser_requires_complete_input(monkeypatch, expression) -> None:
+    """
+    Python 回退解析器不接受仅合法前缀，错误尾部必须显式失败。
+    """
+    monkeypatch.setattr(rules, "_filter_rule_parser", None)
+    with pytest.raises(ParseBaseException):
+        rules.RuleParser().parse(expression)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("requested,expected", [("movie", "movie"), ("tv", "tv")])
+async def test_recommendation_filters_mixed_source_media_types(monkeypatch, requested, expected) -> None:
+    """
+    混合来源在裁剪数量之前应用请求的媒体类型过滤。
+    """
+    chain = SimpleNamespace(async_tmdb_trending=AsyncMock(return_value=[
+        {"title": "TV", "type": "电视剧", "media_id": "1"},
+        {"title": "Movie", "type": "电影", "media_id": "2"}]))
+    monkeypatch.setattr(recommend, "RecommendChain", lambda: chain)
+    response = await recommend.agent_recommendations(source="tmdb_trending", media_type=requested)
+    assert response.success
+    assert [item["type"] for item in response.data] == [expected]
--- a/tests/test_tmdbapi_sync_async_parity.py
+++ b/tests/test_tmdbapi_sync_async_parity.py
@@ -116,7 +116,7 @@ def test_search_lists_share_query_and_projection(
    year_kwargs: dict[str, str],
 ) -> None:
    """
-    电影和剧集列表双入口应共享参数、标题过滤和类型投影。
+    电影和剧集列表双入口应共享查询参数，保留上游跨语言命中并添加类型。
    """
    api = _api()
    sync_items = [
@@ -132,7 +132,10 @@ def test_search_lists_share_query_and_projection(
    sync_result = getattr(api, sync_name)("目标", "2024")
    async_result = asyncio.run(getattr(api, async_name)("目标", "2024"))

-    assert sync_result == async_result == [{"id": 1, title_field: "目标作品", "media_type": mtype}]
+    assert sync_result == async_result == [
+        {"id": 1, title_field: "目标作品", "media_type": mtype},
+        {"id": 2, title_field: "其他作品", "media_type": mtype},
+    ]
    sync_client.assert_called_once_with(**year_kwargs)
    async_client.assert_awaited_once_with(**year_kwargs)