nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-11-02 13:28:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(classification): retire migrated TMDB fields

Browse Source
This commit is contained in:
jxxghp
2026-09-03 15:17:03 +08:00
parent cda109492a
commit 3c817cdb9f
8 changed files with 219 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
app/application/classification/analysis.py
View File

@@ -10,7 +10,10 @@ from datetime import datetime, timezone
from enum import Enum
from typing import Literal, Optional, Protocol, cast
from app.application.classification.catalog import build_classification_field_catalog
from app.application.classification.catalog import (
build_classification_field_catalog,
build_retired_classification_field_catalog,
)
from app.application.classification.configuration import (
ClassificationPolicyConfigurationService,
ClassificationPolicyValidationError,
@@ -161,12 +164,14 @@ class ClassificationAnalysisService:
self._sample_provider = sample_provider
def fields(self) -> ClassificationFieldCatalog:
"""返回标准字段与当前动态来源扩展字段组成的隔离目录。"""
"""分开返回新规则可选字段和已有规则使用的退役字段。"""
extra_fields = self._configuration.extra_fields()
return ClassificationFieldCatalog(
fields=list(
build_classification_field_catalog(
self._configuration.extra_fields()
)
build_classification_field_catalog(extra_fields)
),
retired_fields=list(
build_retired_classification_field_catalog(extra_fields)
),
limits=ClassificationPolicyLimits(
max_category_depth=MAX_CATEGORY_DEPTH,

16
app/application/classification/catalog.py
View File

@@ -1,4 +1,4 @@
"""合并标准分类字段与当前注册媒体来源的能力目录。"""
"""合并并分层输出标准分类字段与来源扩展字段目录。"""
from collections.abc import Iterable
@@ -9,8 +9,20 @@ from app.schemas.category import ClassificationFieldDefinition
def build_classification_field_catalog(
extra_fields: Iterable[ClassificationFieldDefinition] = (),
) -> tuple[ClassificationFieldDefinition, ...]:
"""构造前端字段目录；动态来源未声明能力时保持键缺失。"""
"""构造可用于新增条件的字段目录；退役字段不会混入选择器。"""
return tuple(
definition.model_copy(deep=True)
for definition in merge_field_definitions(extra_fields)
if definition.selectable
)
def build_retired_classification_field_catalog(
extra_fields: Iterable[ClassificationFieldDefinition] = (),
) -> tuple[ClassificationFieldDefinition, ...]:
"""返回仅供已有规则解析和迁移提示使用的退役字段目录。"""
return tuple(
definition.model_copy(deep=True)
for definition in merge_field_definitions(extra_fields)
if not definition.selectable
)

106
app/application/classification/migration.py
View File

@@ -82,6 +82,10 @@ _TMDB_GENRE_KEYS: Final[dict[str, str]] = {
"10767": "talk",
"10770": "tv_movie",
}
_LEGACY_FIELD_PRESENTATION: Final[dict[str, tuple[str, str]]] = {
"genre_ids": ("风格（旧规则）", "media.genre_keys"),
"origin_country": ("原产国家/地区（旧规则）", "media.countries"),
}
@dataclass(frozen=True, slots=True)
@@ -177,18 +181,9 @@ class _MigrationContext:
aliases[value] = normalized
def build_field_definitions(self) -> tuple[ClassificationFieldDefinition, ...]:
"""按旧字段首次出现顺序构造动态 TMDB 字段声明。"""
"""按首次出现顺序构造仅供已有规则解析的退役字段声明。"""
return tuple(
ClassificationFieldDefinition(
id=field_id,
label=f"TMDB 旧字段 {field_id.removeprefix(_EXTENSION_PREFIX)}",
group="TMDB 兼容",
description="按旧 category.yaml 的大写字符串列表语义比较",
value_type="string_list",
operators=["contains_any", "contains_none", "exists", "not_exists"],
media_types=media_types,
source_support={_TMDB_SOURCE: "extension"},
)
_legacy_field_definition(field_id, media_types)
for field_id, media_types in self.field_media_types.items()
)
@@ -623,6 +618,37 @@ def _migrate_legacy_field(
return _legacy_list_condition(field_id, tokens, requires_exists)
def _legacy_field_definition(
field_id: str,
media_types: list[ClassificationMediaType],
) -> ClassificationFieldDefinition:
"""构造不会出现在新规则选择器中的旧 TMDB 字段说明。"""
field_name = field_id.removeprefix(_EXTENSION_PREFIX)
presentation = _LEGACY_FIELD_PRESENTATION.get(field_name)
label = presentation[0] if presentation else f"TMDB {field_name}"
replacement_field = presentation[1] if presentation else None
replacement_hint = (
f"；新规则请使用 {replacement_field}"
if replacement_field
else ""
)
return ClassificationFieldDefinition(
id=field_id,
label=label,
group="旧规则",
description=(
"仅用于保持已迁移 category.yaml 的原始比较语义"
f"{replacement_hint}"
),
value_type="string_list",
operators=["contains_any", "contains_none", "exists", "not_exists"],
media_types=media_types,
source_support={_TMDB_SOURCE: "extension"},
selectable=False,
replacement_field=replacement_field,
)
def _parse_legacy_tokens(value: str) -> tuple[tuple[_LegacyToken, ...], bool]:
"""逐项复现旧逗号、排除前缀和连字符范围展开算法。"""
raw_tokens = [item for item in value.split(",") if item]
@@ -703,42 +729,66 @@ def _migrate_genre_tokens(
media_type: ClassificationMediaType,
context: _MigrationContext,
) -> ClassificationConditionNode:
"""以原始 ID 视图保证等价，并为已知正值附加标准类型键分支。"""
"""已知正向 Genre ID 使用规范风格，其余条件保留原始视图。"""
positive_nodes: list[ClassificationConditionNode] = []
negative_nodes: list[ClassificationConditionNode] = []
field_id = context.register_extension_field("genre_ids", media_type)
extension_field_id: Optional[str] = None
for token in tokens:
if token.negative:
if extension_field_id is None:
extension_field_id = context.register_extension_field(
"genre_ids", media_type
)
context.register_aliases(extension_field_id, token.values)
negative_nodes.append(
ClassificationCondition(
field=extension_field_id,
operator="contains_none",
value=list(token.values),
)
)
continue
known_keys: list[str] = []
unknown_ids: list[str] = []
for value in token.values:
if genre_key := _TMDB_GENRE_KEYS.get(value.upper()):
_append_unique(known_keys, genre_key)
context.register_aliases(field_id, token.values)
extension_condition = ClassificationCondition(
field=field_id,
operator="contains_none" if token.negative else "contains_any",
value=list(token.values),
)
if token.negative:
negative_nodes.append(extension_condition)
continue
token_nodes: list[ClassificationConditionNode] = [extension_condition]
else:
_append_unique(unknown_ids, value)
token_nodes: list[ClassificationConditionNode] = []
if known_keys:
token_nodes.insert(
0,
token_nodes.append(
ClassificationCondition(
field="media.genre_keys",
operator="contains_any",
value=known_keys,
),
)
)
positive_nodes.append(_any_or_single(token_nodes))
if unknown_ids:
if extension_field_id is None:
extension_field_id = context.register_extension_field(
"genre_ids", media_type
)
context.register_aliases(extension_field_id, unknown_ids)
token_nodes.append(
ClassificationCondition(
field=extension_field_id,
operator="contains_any",
value=unknown_ids,
)
)
if token_nodes:
positive_nodes.append(_any_or_single(token_nodes))
nodes: list[ClassificationConditionNode] = []
if positive_nodes:
nodes.append(_any_or_single(positive_nodes))
nodes.extend(negative_nodes)
if not nodes and requires_exists:
return ClassificationCondition(field=field_id, operator="exists")
extension_field_id = context.register_extension_field(
"genre_ids", media_type
)
return ClassificationCondition(field=extension_field_id, operator="exists")
return _all_or_single(nodes)

11
app/domain/classification/fields.py
View File

@@ -94,6 +94,7 @@ class _FieldSpec(TypedDict):
id: str
label: str
group: NotRequired[str]
description: NotRequired[str]
value_type: str
media_types: tuple[str, ...]
options: NotRequired[tuple[str, ...]]
@@ -127,19 +128,22 @@ _STANDARD_FIELD_SPECS: Final[tuple[_FieldSpec, ...]] = (
},
{
"id": "media.countries",
"label": "国家或地区",
"label": "原产国家/地区",
"description": "跨数据源统一的原产国家或地区代码",
"value_type": "string_list",
"media_types": ALL_MEDIA_TYPES,
},
{
"id": "media.genre_keys",
"label": "规范类型",
"label": "风格",
"description": "跨数据源统一的 MoviePilot 规范风格",
"value_type": "string_list",
"media_types": ALL_MEDIA_TYPES,
},
{
"id": "media.genre_names",
"label": "来源类型",
"label": "来源风格",
"description": "媒体数据源返回的原始风格名称",
"value_type": "string_list",
"media_types": ALL_MEDIA_TYPES,
},
@@ -330,6 +334,7 @@ def _build_field_definition(values: _FieldSpec) -> ClassificationFieldDefinition
id=values["id"],
label=values["label"],
group=values.get("group", "通用"),
description=values.get("description"),
value_type=values["value_type"],
operators=[
cast(ClassificationOperator, operator) for operator in operators_for_value_type(values["value_type"])

14
app/schemas/category.py
View File

@@ -432,6 +432,14 @@ class ClassificationFieldDefinition(_ClassificationModel):
default_factory=dict,
description="各数据源对字段的支持等级",
)
selectable: bool = Field(
default=True,
description="是否允许在新规则的字段选择器中使用",
)
replacement_field: Optional[str] = Field(
default=None,
description="退役字段建议替换为的标准字段 ID",
)
@field_validator("options", mode="before") # type: ignore[misc]
@classmethod
@@ -499,7 +507,11 @@ class ClassificationFieldCatalog(_ClassificationModel):
fields: list[ClassificationFieldDefinition] = Field(
default_factory=list,
description="标准字段与当前已登记来源扩展字段",
description="允许在新规则中选择的标准字段与来源扩展字段",
)
retired_fields: list[ClassificationFieldDefinition] = Field(
default_factory=list,
description="仅用于解析已有规则、不可新增的退役字段",
)
limits: ClassificationPolicyLimits = Field(description="服务端策略结构限制")

29
tests/test_media_classification_api.py
View File

@@ -34,6 +34,7 @@ from app.schemas.category import (
ClassificationCategory,
ClassificationCondition,
ClassificationFacts,
ClassificationFieldDefinition,
ClassificationMusicFacts,
ClassificationPolicy,
ClassificationPolicyPublishRequest,
@@ -209,6 +210,34 @@ def test_field_catalog_exposes_typed_options_and_server_limits() -> None:
assert catalog.limits.max_condition_depth == 3
assert catalog.limits.max_conditions_per_rule == 30
assert catalog.limits.max_rules == 1000
assert catalog.retired_fields == []
def test_field_catalog_exposes_retired_fields_outside_new_rule_options() -> None:
"""字段 API 应保留旧规则解析信息，但不得把它混入新条件选项。"""
retired = ClassificationFieldDefinition(
id="extensions.themoviedb.genre_ids",
label="风格（旧规则）",
group="旧规则",
value_type="string_list",
operators=["contains_any"],
media_types=["电影"],
source_support={"themoviedb": "extension"},
selectable=False,
replacement_field="media.genre_keys",
)
configuration = ClassificationPolicyConfigurationService(
_MemoryPolicyStore(),
extra_fields=[retired],
clock=lambda: NOW,
)
configuration.initialize()
catalog = ClassificationAnalysisService(configuration).fields()
assert retired.id not in {field.id for field in catalog.fields}
assert [field.id for field in catalog.retired_fields] == [retired.id]
assert catalog.retired_fields[0].replacement_field == "media.genre_keys"
def test_preview_returns_condition_path_and_structured_missing_fact_warning() -> None:

47
tests/test_media_classification_catalog.py
View File

@@ -1,12 +1,16 @@
"""媒体分类字段与来源能力目录测试。"""
from app.application.classification.catalog import build_classification_field_catalog
from app.application.classification.catalog import (
build_classification_field_catalog,
build_retired_classification_field_catalog,
)
from app.domain.classification.fields import get_standard_classification_fields
from app.domain.classification.sources import (
BUILTIN_CLASSIFICATION_SOURCES,
STANDARD_CLASSIFICATION_FIELD_IDS,
builtin_source_field_support,
)
from app.schemas.category import ClassificationFieldDefinition
from app.schemas.types import MediaSource
@@ -53,6 +57,47 @@ def test_catalog_preserves_verified_builtin_capability_levels() -> None:
assert fields["music.secondary_types"].source_support["theaudiodb"] == "unavailable"
def test_standard_catalog_uses_user_facing_country_and_genre_names() -> None:
"""通用标准字段应直接使用用户理解的国家和风格名称。"""
fields = {
definition.id: definition
for definition in build_classification_field_catalog()
}
assert fields["media.countries"].label == "原产国家/地区"
assert fields["media.genre_keys"].label == "风格"
assert fields["media.genre_names"].label == "来源风格"
def test_catalog_separates_retired_fields_from_new_rule_options() -> None:
"""迁移专用字段只能解析已有规则，插件正常扩展字段仍可新增。"""
retired = ClassificationFieldDefinition(
id="extensions.themoviedb.origin_country",
label="原产国家/地区（旧规则）",
value_type="string_list",
media_types=["电视剧"],
selectable=False,
replacement_field="media.countries",
)
plugin = ClassificationFieldDefinition(
id="extensions.example.region",
label="地区",
value_type="string",
media_types=["电影"],
)
selectable_ids = {
field.id
for field in build_classification_field_catalog((retired, plugin))
}
retired_fields = build_retired_classification_field_catalog((retired, plugin))
assert "extensions.example.region" in selectable_ids
assert retired.id not in selectable_ids
assert [field.id for field in retired_fields] == [retired.id]
assert retired_fields[0].replacement_field == "media.countries"
def test_discover_only_builtin_sources_are_explicitly_unavailable() -> None:
"""仅提供发现入口的内置来源不能被误认为可形成分类事实。"""
discover_sources = {

23
tests/test_media_classification_legacy.py
View File

@@ -169,8 +169,8 @@ def _leaf_fields(node: ClassificationConditionNode) -> list[str]:
return [field for child in children for field in _leaf_fields(child)]
def test_default_style_config_preserves_order_and_uses_tmdb_extensions() -> None:
"""默认式配置应保持分类顺序，并避免把国家语言字段误升格为标准事实。"""
def test_default_style_config_preserves_order_and_uses_safe_standard_fields() -> None:
"""默认式配置应保持顺序，已知风格使用标准字段，其余字段保持等价。"""
result = migrate_legacy_category_config(_legacy_config())
assert result.valid
@@ -188,6 +188,8 @@ def test_default_style_config_preserves_order_and_uses_tmdb_extensions() -> None
"美国电影",
]
all_fields = [field for rule in result.policy.rules for field in _leaf_fields(rule.when)]
assert "media.genre_keys" in all_fields
assert "extensions.themoviedb.genre_ids" not in all_fields
assert "media.language" not in all_fields
assert "media.countries" not in all_fields
assert "extensions.themoviedb.original_language" in all_fields
@@ -195,6 +197,15 @@ def test_default_style_config_preserves_order_and_uses_tmdb_extensions() -> None
assert "extensions.themoviedb.origin_country" in all_fields
assert legacy_extension_fields_from_policy(result.policy) == result.extra_fields
assert ClassificationPolicyValidator.validate(result.policy, result.extra_fields).valid
origin_country = next(
field
for field in result.extra_fields
if field.id == "extensions.themoviedb.origin_country"
)
assert origin_country.label == "原产国家/地区（旧规则）"
assert origin_country.group == "旧规则"
assert origin_country.selectable is False
assert origin_country.replacement_field == "media.countries"
def test_first_empty_rule_becomes_source_fallback_and_later_entries_are_disabled() -> None:
@@ -246,6 +257,14 @@ def test_mixed_genre_ids_keep_positive_or_and_negative_and_semantics() -> None:
assert result.valid
extension_ids = {field.id for field in legacy_extension_fields_from_policy(result.policy)}
assert "extensions.themoviedb.genre_ids" in extension_ids
genre_field = next(
field
for field in result.extra_fields
if field.id == "extensions.themoviedb.genre_ids"
)
assert genre_field.label == "风格（旧规则）"
assert genre_field.selectable is False
assert genre_field.replacement_field == "media.genre_keys"
assert ClassificationPolicyValidator.validate(result.policy, result.extra_fields).valid
assert _category_name(result, _tmdb_facts(result, {"genre_ids": [16]}, "电影")) == "混合类型"
assert _category_name(result, _tmdb_facts(result, {"genre_ids": [999]}, "电影")) == "混合类型"