fix: 无损合并旧分类枚举并隔离迁移校验失败

app/application/classification/migration.py

@@ -121,7 +121,7 @@ class LegacyClassificationMigrationResult:
 
 @dataclass(frozen=True, slots=True)
 class _LegacyToken:
-    """保留一个旧逗号项展开后的值集合及其排除语义。"""
+    """保留旧字段展开后的同向值集合及其排除语义。"""
 
     negative: bool
     values: tuple[str, ...]
@@ -530,26 +530,20 @@ def _legacy_field_definition(
 
 
 def _parse_legacy_tokens(value: str) -> tuple[tuple[_LegacyToken, ...], bool]:
-    """逐项复现旧逗号、排除前缀和连字符范围展开算法。"""
+    """沿用旧范围展开语义，并合并同向枚举，避免值数量膨胀为叶子数量。"""
     raw_tokens = [item for item in value.split(",") if item]
-    parsed: list[_LegacyToken] = []
+    values_by_sign: dict[bool, list[str]] = {}
     requires_exists = not raw_tokens
     for raw_token in raw_tokens:
         expanded = _expand_legacy_token(raw_token)
         if not expanded:
             requires_exists = True
             continue
-        grouped: list[_LegacyToken] = []
         for expanded_value in expanded:
             negative = expanded_value.startswith("!")
             plain_value = expanded_value[1:] if negative else expanded_value
-            if grouped and grouped[-1].negative == negative:
-                previous = grouped[-1]
-                grouped[-1] = _LegacyToken(negative, (*previous.values, plain_value))
-            else:
-                grouped.append(_LegacyToken(negative, (plain_value,)))
-        parsed.extend(grouped)
-    return tuple(parsed), requires_exists
+            values_by_sign.setdefault(negative, []).append(plain_value)
+    return tuple(_LegacyToken(negative, tuple(values)) for negative, values in values_by_sign.items()), requires_exists
 
 
 def _expand_legacy_token(value: str) -> tuple[str, ...]:

app/application/classification/projection.py

@@ -219,11 +219,11 @@ def _project_rule(
             continue
         if not values:
             continue
-        token = _render_legacy_token(values)
-        rendered = f"!{token}" if negative else token
         field_tokens = tokens_by_field.setdefault(field_name, [])
-        if rendered not in field_tokens:
-            field_tokens.append(rendered)
+        for value in values:
+            rendered = f"!{value}" if negative else value
+            if rendered not in field_tokens:
+                field_tokens.append(rendered)
     return (
         {field_name: ",".join(tokens) for field_name, tokens in tokens_by_field.items()},
         diagnostics,
@@ -300,17 +300,6 @@ def _original_alias_value(value: str, aliases: Mapping[str, str]) -> str:
     return value
 
 
-def _render_legacy_token(values: Sequence[str]) -> str:
-    """把一个迁移时保留边界的值集合恢复为单个旧逗号项。"""
-    if len(values) == 1:
-        return values[0]
-    if values and all(value.isdigit() for value in values):
-        numbers = [int(value) for value in values]
-        if numbers == list(range(numbers[0], numbers[-1] + 1)):
-            return f"{numbers[0]}-{numbers[-1]}"
-    return "-".join(values)
-
-
 def _projection_warning(
     code: str,
     message: str,

app/startup/composition/classification.py

@@ -13,6 +13,7 @@ from pydantic import ValidationError
 
 from app.application.classification.configuration import (
     ClassificationPolicyConfigurationService,
+    ClassificationPolicyValidationError,
 )
 from app.application.classification.contract import (
     ClassificationPolicyStateCorruptError,
@@ -196,7 +197,18 @@ async def compose_classification(
         )
 
     service.register_extra_fields(migration.extra_fields)
-    await service.async_initialize(migration.policy)
+    try:
+        await service.async_initialize(migration.policy)
+    except ClassificationPolicyValidationError as error:
+        logger.error("旧分类策略未通过发布校验，继续保留 legacy 只读兼容行为")
+        return finish(
+            ClassificationRuntime(
+                service,
+                legacy_config=legacy_config,
+                diagnostics=(*_migration_issues(migration.issues), *error.result.issues),
+            ),
+            migrated=False,
+        )
     _log_migration_issues(migration)
     logger.info("已将 category.yaml 无损迁移为 MediaClassificationPolicy revision 1")
     return finish(

docs/architecture/media-classification-design.md

@@ -934,12 +934,17 @@ API 常规读取只返回 `active`，历史接口按需读取 `history`。选择
 8. 首个空规则分类转换为该媒体类型的全局 `fallbacks`，其后的 legacy 项在旧实现中本就不可达，迁移时保持禁用并向管理员报告。
 9. 新策略只按媒体类型配置通用兜底；数据源只能作为规则的筛选条件，不能单独决定默认分类。
 10. 保存新策略 revision 1，并保留原 YAML 文件只读备份，不再继续写入。
+11. 同一字段的正向枚举合并为 `contains_any`，排除枚举合并为 `contains_none`；国家、语言等值数量
+    不应展开为叶子数量。已知和未知 Genre ID 仍分别使用标准字段与受控扩展字段，保持原有 OR 语义。
+12. 迁移后的策略仍需通过完整发布校验；超出真实条件复杂度或引用约束时不写入新策略，保留旧分类
+    只读兼容行为和结构化错误诊断，不得让发布校验异常中断宿主启动。
 
 ### 13.2 行为兼容
 
 - 迁移测试必须证明同一批 TMDB fixture 在旧分类器和新分类器下得到完全相同的目录分类。
 - 旧配置中的 `!值` 转换为明确的排除操作符。
 - 旧逗号字符串转换为数组，后续 API 不再使用逗号编码多值。
+- 旧只读 API 将集合逐值投影为逗号项，每个排除值独立添加 `!`；范围允许展开为等价枚举，不改写原 YAML。
 - 迁移后 TMDB 模块不再实例化 `CategoryHelper`，也不再写 `MediaInfo.category`。
 
 ### 13.3 字段迁移

tests/test_media_classification_legacy.py

@@ -273,6 +273,37 @@ def test_mixed_genre_ids_keep_positive_or_and_negative_and_semantics() -> None:
     assert _category_name(result, _tmdb_facts(result, {}, "电影")) == "兜底"
 
 
+@pytest.mark.parametrize("negative", [False, True])
+@pytest.mark.parametrize("field_name", ["origin_country", "genre_ids"])
+def test_large_value_sets_remain_compact_and_preserve_legacy_semantics(
+    field_name: str,
+    negative: bool,
+) -> None:
+    """大枚举只产生集合条件，旧 API 投影后每个正向或排除值仍保持原语义。"""
+    values = [str(index) for index in range(1, 61)]
+    prefix = "!" if negative else ""
+    config = {
+        "movie": {
+            "目标": {field_name: ",".join(f"{prefix}{value}" for value in values)},
+            "兜底": None,
+        },
+        "tv": {},
+    }
+    result = migrate_legacy_category_config(config)
+    projection = project_policy_to_legacy_category_projection(result.policy)
+    remigrated = migrate_legacy_category_config(projection.config)
+
+    assert result.valid
+    assert ClassificationPolicyValidator.validate(result.policy, result.extra_fields).valid
+    assert len(_leaf_fields(result.policy.rules[0].when)) <= 2
+    assert projection.exact
+    for actual in [*([value] for value in values), ["999"], ["1", "999"], [], None]:
+        tmdb_info = {field_name: actual, "id": 1}
+        expected = _legacy_tmdb_category(config["movie"], tmdb_info)
+        assert _category_name(result, _tmdb_facts(result, tmdb_info, "电影")) == expected
+        assert _category_name(remigrated, _tmdb_facts(remigrated, tmdb_info, "电影")) == expected
+
+
 def test_safe_unknown_field_is_declared_but_unsafe_field_blocks_publish() -> None:
     """合法未知 TMDB 一级字段可迁移，非法字段段必须产生阻断错误。"""
     safe = migrate_legacy_category_config(
@@ -387,7 +418,15 @@ def test_release_year_supports_values_ranges_and_non_numeric_hyphen_endpoints()
     assert _category_name(result, _tmdb_facts(result, {"release_date": "2023-01-01"}, "电影")) == "兜底"
     projection = project_policy_to_legacy_category_projection(result.policy)
     assert projection.exact
-    assert projection.config.movie == CategoryConfig.model_validate(config).movie
+    assert projection.config.movie == CategoryConfig.model_validate(
+        {
+            "movie": {
+                "近年": {"release_year": "2020,2021,2022,2024"},
+                "字母年": {"release_year": "ABCD,EFGH"},
+                "兜底": None,
+            }
+        }
+    ).movie
 
 
 def test_stable_ids_are_repeatable_ascii_and_media_type_scoped() -> None:
@@ -485,5 +524,9 @@ def test_migrated_policy_round_trips_to_category_config() -> None:
     projected = project_policy_to_legacy_category_projection(migrated.policy)
 
     assert projected.exact
-    assert projected.config == CategoryConfig.model_validate(config)
+    expected = CategoryConfig.model_validate(config)
+    assert expected.movie is not None
+    assert expected.movie["组合"] is not None
+    expected.movie["组合"].release_year = "2020,2021,2022,2024"
+    assert projected.config == expected
     assert project_policy_to_legacy_category_config(migrated.policy) == projected.config

tests/test_media_classification_startup.py

@@ -17,6 +17,7 @@ from app.application.classification.configuration import (
 from app.application.classification.contract import ClassificationPolicyConflictError
 from app.application.classification.runtime import ClassificationRuntime
 from app.schemas.category import (
+    ClassificationConditionGroup,
     ClassificationPolicyState,
 )
 from app.schemas.types import SystemConfigKey
@@ -152,6 +153,83 @@ async def test_absent_policy_migrates_yaml_once_without_rewriting_file(
     assert legacy_path.read_text(encoding="utf-8") == content
 
 
+@pytest.mark.asyncio
+async def test_large_legacy_enumerations_migrate_and_reload_without_losing_values(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """复现 56/31 个旧枚举值的升级场景，验证首次发布、重启和旧接口均保留全部值。"""
+    languages = ",".join(f"language{index}" for index in range(26))
+    countries = ",".join(f"country{index}" for index in range(30))
+    content = (
+        f"movie:\n  欧美电影:\n    original_language: '{languages}'\n"
+        f"    production_countries: '{countries}'\n  未分类:\n"
+        f"tv:\n  欧美漫:\n    genre_ids: '16'\n    origin_country: '{countries}'\n  未分类:\n"
+    )
+    legacy_path = tmp_path / "category.yaml"
+    legacy_path.write_text(content, encoding="utf-8")
+    store = _MemoryPolicyStore()
+    monkeypatch.setattr(classification_composition, "SystemConfigClassificationPolicyStore", lambda *_args: store)
+    system_config = _SystemConfig({})
+
+    composition = await classification_composition.compose_classification(
+        executor=cast(Any, _InlineExecutor()),
+        settings=cast(Any, SimpleNamespace(CONFIG_PATH=tmp_path)),
+        system_config=cast(Any, system_config),
+    )
+
+    assert composition.migrated is True
+    policy = composition.runtime.require_policy()
+    assert policy.revision == 1
+    assert len(policy.rules) == 2
+    for rule in policy.rules:
+        assert isinstance(rule.when, ClassificationConditionGroup)
+        assert rule.when.all is not None and len(rule.when.all) == 2
+    assert store.state is not None
+    system_config.publish_many({SystemConfigKey.MediaClassificationPolicy: store.state.model_dump(mode="json")})
+    reloaded = await classification_composition.compose_classification(
+        executor=cast(Any, _InlineExecutor()),
+        settings=cast(Any, SimpleNamespace(CONFIG_PATH=tmp_path)),
+        system_config=cast(Any, system_config),
+    )
+
+    assert reloaded.migrated is False
+    assert reloaded.runtime.require_policy() == policy
+    projected = reloaded.runtime.legacy_config().model_dump(exclude_none=True)
+    assert projected["movie"]["欧美电影"]["original_language"] == languages
+    assert projected["movie"]["欧美电影"]["production_countries"] == countries
+    assert projected["tv"]["欧美漫"]["origin_country"] == countries
+    assert store.write_count == 1
+    assert legacy_path.read_text(encoding="utf-8") == content
+
+
+@pytest.mark.asyncio
+async def test_publish_validation_failure_preserves_legacy_runtime_without_writes(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """真正超出叶子上限的不同字段保留错误诊断和旧配置，不得中断启动或写入新策略。"""
+    content = "movie:\n  复杂规则:\n" + "".join(f"    field{index}: 'X'\n" for index in range(31))
+    legacy_path = tmp_path / "category.yaml"
+    legacy_path.write_text(content, encoding="utf-8")
+    store = _MemoryPolicyStore()
+    monkeypatch.setattr(classification_composition, "SystemConfigClassificationPolicyStore", lambda *_args: store)
+
+    composition = await classification_composition.compose_classification(
+        executor=cast(Any, _InlineExecutor()),
+        settings=cast(Any, SimpleNamespace(CONFIG_PATH=tmp_path)),
+        system_config=cast(Any, _SystemConfig({})),
+    )
+
+    assert composition.migrated is False
+    assert composition.runtime.active_policy() is None
+    assert any(issue.code == "max_conditions_exceeded" for issue in composition.runtime.diagnostics())
+    assert "复杂规则" in (composition.runtime.legacy_config().movie or {})
+    assert store.write_count == 0
+    assert store.state is None
+    assert legacy_path.read_text(encoding="utf-8") == content
+
+
 @pytest.mark.asyncio  # type: ignore[misc]
 async def test_invalid_legacy_config_keeps_runtime_uninitialized_and_writes_nothing(
     monkeypatch: pytest.MonkeyPatch,