fix(classification): upgrade missing music defaults

This commit is contained in:
yanjunpu
2026-09-08 13:41:26 +08:00
parent 9985e22e82
commit 64b5685cb3
3 changed files with 80 additions and 16 deletions
+16 -12
View File
@@ -94,11 +94,7 @@ def _build_uncategorized_classification_policy() -> ClassificationPolicy:
def with_default_music_classification(policy: ClassificationPolicy) -> ClassificationPolicy:
"""为尚未配置音乐分类的策略追加安全、结构化的常用专辑分类。"""
music_categories = [
item for item in policy.categories if item.media_type == "音乐"
]
music_rules = [item for item in policy.rules if "音乐" in item.media_types]
if music_rules or any(item.id != "music.uncategorized" for item in music_categories):
if not needs_default_music_classification(policy):
return cast(ClassificationPolicy, policy.model_copy(deep=True))
categories = [
@@ -174,15 +170,23 @@ def with_default_music_classification(policy: ClassificationPolicy) -> Classific
)
def is_untouched_legacy_default_policy(state: ClassificationPolicyState) -> bool:
"""判断状态是否为旧版本自动创建且从未编辑的 revision 1 默认策略"""
if state.active.revision != 1 or state.history:
def needs_default_music_classification(policy: ClassificationPolicy) -> bool:
"""判断音乐侧是否为旧版原始兜底,未包含任何用户分类"""
music_rules = [item for item in policy.rules if "音乐" in item.media_types]
music_categories = [
item for item in policy.categories if item.media_type == "音乐"
]
if music_rules or len(music_categories) != 1:
return False
normalized = state.active.model_copy(
deep=True,
update={"revision": 0, "updated_at": None},
category = music_categories[0]
return bool(
category.id == "music.uncategorized"
and category.name == "未分类"
and category.path == ["未分类"]
and category.enabled
and not category.labels
and policy.fallbacks.get("音乐") == category.id
)
return bool(normalized == _build_uncategorized_classification_policy())
def build_default_classification_policy() -> ClassificationPolicy:
+4 -4
View File
@@ -14,7 +14,7 @@ from pydantic import ValidationError
from app.application.classification.configuration import (
ClassificationPolicyConfigurationService,
ClassificationPolicyValidationError,
is_untouched_legacy_default_policy,
needs_default_music_classification,
with_default_music_classification,
)
from app.application.classification.contract import (
@@ -159,8 +159,8 @@ async def compose_classification(
ClassificationRuntime(service, diagnostics=(issue,)),
migrated=False,
)
if stored_state is not None and is_untouched_legacy_default_policy(
stored_state
if stored_state is not None and needs_default_music_classification(
stored_state.active
):
try:
await service.async_publish(
@@ -177,7 +177,7 @@ async def compose_classification(
migrated=False,
)
else:
logger.info("已为未编辑的默认分类策略补充常用音乐分类 revision 2")
logger.info("已为仅有旧版兜底的分类策略补充常用音乐分类")
return finish(
ClassificationRuntime(service),
migrated=True,
@@ -210,6 +210,66 @@ async def test_untouched_legacy_default_policy_gains_music_rules_once(
assert store.write_count == 1
@pytest.mark.asyncio # type: ignore[misc]
async def test_custom_movie_policy_with_legacy_music_fallback_gains_music_rules(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
"""其它媒体已自定义时,仍应只升级完全未编辑的旧版音乐兜底。"""
state = _published_legacy_default_state()
categories = [
category.model_copy(
deep=True,
update={"name": "我的电影", "path": ["我的电影"]},
)
if category.id == "movie.uncategorized"
else category.model_copy(deep=True)
for category in state.active.categories
]
state = ClassificationPolicyState(
active=state.active.model_copy(
deep=True,
update={"categories": categories},
)
)
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),
)
policy = composition.runtime.require_policy()
assert composition.migrated is True
assert policy.revision == 2
assert next(
category for category in policy.categories if category.id == "movie.uncategorized"
).path == ["我的电影"]
assert [
category.id for category in policy.categories if category.media_type == "音乐"
] == [
"music.uncategorized",
"music.album",
"music.compilation",
"music.ep",
"music.single",
]
assert store.write_count == 1
@pytest.mark.asyncio # type: ignore[misc]
async def test_edited_legacy_default_policy_is_not_automatically_changed(
monkeypatch: pytest.MonkeyPatch,