fix(classification): restore legacy alias fields after rule deletion

This commit is contained in:
jxxghp
2026-09-12 07:54:58 +08:00
parent b6ba20371e
commit 75c25651ff
3 changed files with 81 additions and 1 deletions

View File

@@ -209,7 +209,7 @@ def migrate_legacy_category_config(
def legacy_extension_fields_from_policy(
policy: ClassificationPolicy,
) -> tuple[ClassificationFieldDefinition, ...]:
"""按策略条件顺序重建可直接注册的 TMDB 旧比较扩展字段声明。"""
"""按策略条件和别名重建可直接注册的 TMDB 旧比较扩展字段声明。"""
context = _MigrationContext()
for rule in policy.rules:
for field_id in _condition_field_ids(rule.when):
@@ -220,6 +220,15 @@ def legacy_extension_fields_from_policy(
continue
for media_type in rule.media_types:
context.register_extension_field(field_name, media_type)
# 别名与规则分开保存;删除最后一条规则后仍需恢复字段,避免重启后校验失败。
for field_id in policy.field_aliases:
if field_id in context.field_media_types or not field_id.startswith(_EXTENSION_PREFIX):
continue
field_name = field_id.removeprefix(_EXTENSION_PREFIX)
if not _SAFE_FIELD_SEGMENT.fullmatch(field_name):
continue
for media_type in _MEDIA_TYPES.values():
context.register_extension_field(field_name, media_type)
return context.build_field_definitions()

View File

@@ -208,6 +208,24 @@ def test_default_style_config_preserves_order_and_uses_safe_standard_fields() ->
assert origin_country.replacement_field == "media.countries"
def test_deleted_legacy_rules_keep_alias_only_fields_registered() -> None:
"""删除最后一条旧规则后,仍需登记孤立别名字段以通过策略校验。"""
migrated = migrate_legacy_category_config(
{
"movie": {"中文电影": {"original_language": "zh"}},
"tv": {},
}
)
policy = migrated.policy.model_copy(deep=True, update={"rules": []})
fields = legacy_extension_fields_from_policy(policy)
assert [field.id for field in fields] == [
"extensions.themoviedb.original_language"
]
assert ClassificationPolicyValidator.validate(policy, fields).valid
def test_first_empty_rule_becomes_global_fallback_and_later_entries_are_disabled() -> None:
"""首个全空项应成为媒体类型全局兜底,后续分类和规则保留但永远禁用。"""
result = migrate_legacy_category_config(

View File

@@ -10,11 +10,14 @@ from typing import Any, TypeVar, cast
import pytest
from app.api.endpoints import media as media_endpoint
from app.application.classification.analysis import ClassificationAnalysisService
from app.application.classification.configuration import (
ClassificationPolicyConfigurationService,
build_default_classification_policy,
with_default_music_classification,
)
from app.application.classification.contract import ClassificationPolicyConflictError
from app.application.classification.legacy import migrate_legacy_category_config
from app.application.classification.runtime import ClassificationRuntime
from app.schemas.category import (
ClassificationConditionGroup,
@@ -142,6 +145,56 @@ async def test_existing_policy_never_reads_legacy_yaml(
assert store.write_count == 0
@pytest.mark.asyncio # type: ignore[misc]
async def test_deleted_legacy_rules_keep_validation_and_impact_analysis_working(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
"""重启加载只剩旧别名的策略后,校验和影响分析仍应可用。"""
migrated = migrate_legacy_category_config(
{
"movie": {"中文电影": {"original_language": "zh"}},
"tv": {},
}
)
policy = with_default_music_classification(migrated.policy).model_copy(
deep=True,
update={"rules": []},
)
state = ClassificationPolicyState(active=policy)
store = _MemoryPolicyStore(state)
system_config = _SystemConfig(
{
SystemConfigKey.MediaClassificationPolicy.value: state.model_dump(
mode="json"
)
}
)
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, system_config),
)
analysis = ClassificationAnalysisService(composition.runtime.service)
assert analysis.validate(policy).valid
impact = await analysis.impact(
policy,
expected_revision=1,
sample_limit=10,
example_limit=2,
)
assert impact.baseline_revision == 1
assert impact.candidate_revision == 2
assert impact.sample_count == 0
@pytest.mark.asyncio # type: ignore[misc]
async def test_untouched_legacy_default_policy_gains_music_rules_once(
monkeypatch: pytest.MonkeyPatch,