diff --git a/app/application/classification/migration.py b/app/application/classification/migration.py index 1ddba42fe..88daef859 100644 --- a/app/application/classification/migration.py +++ b/app/application/classification/migration.py @@ -304,7 +304,7 @@ def _migrate_media_categories( id=category_id, media_type=media_type, name=name, - path=[name], + path=_legacy_category_path(name), enabled=not unreachable, ) ) @@ -430,14 +430,12 @@ def _diagnose_category_name( path: Sequence[LegacyDiagnosticPathPart], context: _MigrationContext, ) -> None: - """在仍保留分类的同时标记无法安全作为目录段的名称。""" + """在仍保留分类的同时标记无法安全投影为目录路径的名称。""" invalid = ( not isinstance(raw_name, str) or not name or name != name.strip() - or name in {".", ".."} - or name.endswith((".", " ")) - or any(character in _ILLEGAL_PATH_CHARACTERS or ord(character) < 32 for character in name) + or any(_legacy_path_segment_is_invalid(segment) for segment in _legacy_category_path(name)) ) if invalid: context.add_diagnostic( @@ -448,6 +446,23 @@ def _diagnose_category_name( ) +def _legacy_category_path(name: str) -> list[str]: + """把旧分类名中的斜杠还原为目录层级，同时保留原始显示名称。""" + return name.split("/") + + +def _legacy_path_segment_is_invalid(segment: str) -> bool: + """判断旧分类名拆出的目录段是否违反跨平台路径安全约束。""" + illegal_characters = _ILLEGAL_PATH_CHARACTERS - frozenset({"/"}) + return ( + not segment + or segment in {".", ".."} + or segment != segment.strip() + or segment.endswith((".", " ")) + or any(character in illegal_characters or ord(character) < 32 for character in segment) + ) + + def _migrate_legacy_field( *, raw_field: object, diff --git a/database/versions/e7f3a9c1d5b2_3_0_29.py b/database/versions/e7f3a9c1d5b2_3_0_29.py new file mode 100644 index 000000000..fba799155 --- /dev/null +++ b/database/versions/e7f3a9c1d5b2_3_0_29.py @@ -0,0 +1,167 @@ +"""3.0.29 修复旧分类名称中的目录层级路径。""" + +from collections.abc import Mapping + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.sql.selectable import TableClause + +revision = "e7f3a9c1d5b2" +down_revision = "d8f2b6a4c1e7" +branch_labels = None +depends_on = None + +_TABLE = "systemconfig" +_POLICY_KEY = "MediaClassificationPolicy" +_LEGACY_CATEGORY_PREFIXES = ("legacy.movie.", "legacy.tv.") +_MAX_CATEGORY_DEPTH = 4 +_MAX_CATEGORY_SEGMENT_LENGTH = 64 +_MAX_CATEGORY_PATH_LENGTH = 240 +_ILLEGAL_PATH_CHARACTERS = frozenset('<>:"/\\|?*') +_WINDOWS_RESERVED_NAMES = { + "CON", + "PRN", + "AUX", + "NUL", + *(f"COM{index}" for index in range(1, 10)), + *(f"LPT{index}" for index in range(1, 10)), +} + + +def _table_exists() -> bool: + """检查系统配置表是否存在。""" + return _TABLE in set(sa.inspect(op.get_bind()).get_table_names()) + + +def _system_config_relation() -> TableClause: + """构造只包含本次 JSON 数据修复所需字段的系统配置关系。""" + return sa.table( + _TABLE, + sa.column("id", sa.Integer()), + sa.column("key", sa.String()), + sa.column("value", sa.JSON()), + ) + + +def _path_segment_is_invalid(segment: str) -> bool: + """判断分类路径段是否包含目录穿越或跨平台非法文件名。""" + if not segment or segment != segment.strip(): + return True + if segment in {".", ".."} or segment.endswith((".", " ")): + return True + if any(character in _ILLEGAL_PATH_CHARACTERS for character in segment): + return True + if any(ord(character) < 32 for character in segment): + return True + return segment.split(".", 1)[0].upper() in _WINDOWS_RESERVED_NAMES + + +def _safe_legacy_path(name: str) -> list[str] | None: + """把可安全还原的旧分类名转换成目录路径段。""" + segments = name.split("/") + if ( + not segments + or len(segments) > _MAX_CATEGORY_DEPTH + or any(len(segment) > _MAX_CATEGORY_SEGMENT_LENGTH or _path_segment_is_invalid(segment) for segment in segments) + or len("/".join(segments)) > _MAX_CATEGORY_PATH_LENGTH + ): + return None + return segments + + +def _repair_category(category: object) -> tuple[object, bool]: + """仅修复首版 legacy 影视分类的单段斜杠路径。""" + if not isinstance(category, Mapping): + return category, False + category_id = category.get("id") + name = category.get("name") + path = category.get("path") + if not ( + isinstance(category_id, str) + and category_id.startswith(_LEGACY_CATEGORY_PREFIXES) + and isinstance(name, str) + and "/" in name + and isinstance(path, list) + and path == [name] + ): + return category, False + repaired_path = _safe_legacy_path(name) + if repaired_path is None: + return category, False + repaired = dict(category) + repaired["path"] = repaired_path + return repaired, True + + +def _repair_policy_snapshot(snapshot: object) -> tuple[object, bool]: + """修复一个活动或历史策略快照中的 legacy 分类路径。""" + if not isinstance(snapshot, Mapping): + return snapshot, False + categories = snapshot.get("categories") + if not isinstance(categories, list): + return snapshot, False + repaired_categories: list[object] = [] + changed = False + for category in categories: + repaired, category_changed = _repair_category(category) + repaired_categories.append(repaired) + changed = changed or category_changed + if not changed: + return snapshot, False + repaired_snapshot = dict(snapshot) + repaired_snapshot["categories"] = repaired_categories + return repaired_snapshot, True + + +def _repair_policy_state(value: object) -> tuple[object, bool]: + """修复策略状态中的活动版本和有限历史版本。""" + if not isinstance(value, Mapping): + return value, False + repaired_state = dict(value) + changed = False + + active, active_changed = _repair_policy_snapshot(value.get("active")) + if active_changed: + repaired_state["active"] = active + changed = True + + history = value.get("history") + if isinstance(history, list): + repaired_history: list[object] = [] + history_changed = False + for snapshot in history: + repaired, snapshot_changed = _repair_policy_snapshot(snapshot) + repaired_history.append(repaired) + history_changed = history_changed or snapshot_changed + if history_changed: + repaired_state["history"] = repaired_history + changed = True + + return (repaired_state, True) if changed else (value, False) + + +def _repair_stored_policy() -> None: + """回写数据库中已迁移策略的安全路径段，保持 revision 和目录字符串不变。""" + relation = _system_config_relation() + connection = op.get_bind() + rows = ( + connection.execute(sa.select(relation.c.id, relation.c.value).where(relation.c.key == _POLICY_KEY)) + .mappings() + .all() + ) + for row in rows: + repaired, changed = _repair_policy_state(row["value"]) + if not changed: + continue + connection.execute(relation.update().where(relation.c.id == row["id"]).values(value=repaired)) + + +def upgrade() -> None: + """修复已完成迁移的分类策略，避免旧斜杠名称触发路径安全错误。""" + if _table_exists(): + _repair_stored_policy() + + +def downgrade() -> None: + """数据修复不可逆，降级不恢复为原有的不安全路径表示。""" + pass diff --git a/docs/v2-to-v3-overview.md b/docs/v2-to-v3-overview.md index ee2327201..50559639a 100644 --- a/docs/v2-to-v3-overview.md +++ b/docs/v2-to-v3-overview.md @@ -120,6 +120,8 @@ V2 自动分类主要由 TMDB 详情和 `category.yaml` 驱动，只覆盖电影 升级时，如果尚未存在 V3 分类策略，系统会读取现有 `category.yaml` 并自动迁移；旧 TMDB 规则的顺序、排除条件、年份范围和兜底语义会保留。迁移完成后不再继续写入 YAML。旧 `GET /api/v1/media/category` 和 `GET /api/v1/media/category/config` 暂时保留为只读投影，旧 `POST /api/v1/media/category/config` 已移除，所有新写入都通过带 revision 校验的策略接口完成。 +旧分类名称中用于表示目录层级的 `/` 会在迁移时转换为多个安全路径段，分类名称和目录显示结果保持不变；已经完成迁移的策略会在数据库升级时自动修复。 + 迁移完成后，前端默认不再显示没有被现有规则引用的 TMDB 旧字段；仍被旧规则使用的字段会标记为“旧规则”，只能查看， 改写规则时应优先选择新的统一字段。分类编辑弹窗和帮助弹窗支持透明主题，分类下拉选项会避免名称与路径末级重复显示。 diff --git a/tests/test_media_classification_legacy_path_migration.py b/tests/test_media_classification_legacy_path_migration.py new file mode 100644 index 000000000..3f81c749d --- /dev/null +++ b/tests/test_media_classification_legacy_path_migration.py @@ -0,0 +1,166 @@ +"""旧分类斜杠路径迁移和已持久化策略修复测试。""" + +import importlib +from collections.abc import Mapping +from typing import Any, Protocol, cast + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy.engine import Connection + +from app.application.classification.legacy import migrate_legacy_category_config +from app.domain.classification.validation import ClassificationPolicyValidator + +_MIGRATION = "database.versions.e7f3a9c1d5b2_3_0_29" + + +class _MigrationModule(Protocol): + """声明本测试调用的 Alembic 迁移模块接口。""" + + revision: str + down_revision: str + op: Any + + def upgrade(self) -> None: + """执行升级迁移。""" + + +def _bind_migration( + monkeypatch: pytest.MonkeyPatch, + connection: Connection, +) -> _MigrationModule: + """把策略路径迁移绑定到隔离 SQLite 连接。""" + migration = cast(_MigrationModule, importlib.import_module(_MIGRATION)) + monkeypatch.setattr( + migration, + "op", + Operations(MigrationContext.configure(connection)), + ) + return migration + + +def _policy_state_payload() -> dict[str, object]: + """构造包含活动和历史 legacy 斜杠路径的最小策略状态。""" + legacy_category = { + "id": "legacy.movie.0123456789abcdef", + "media_type": "电影", + "name": "电影/日韩电影", + "path": ["电影/日韩电影"], + "enabled": True, + } + unchanged_category = { + "id": "movie.custom", + "media_type": "电影", + "name": "用户分类/保留原样", + "path": ["用户分类/保留原样"], + "enabled": True, + } + snapshot = { + "schema_version": 2, + "revision": 2, + "categories": [legacy_category, unchanged_category], + "rules": [], + "fallbacks": {}, + "source_fallbacks": {}, + "field_aliases": {}, + } + history = { + **snapshot, + "revision": 1, + "categories": [legacy_category], + } + return {"active": snapshot, "history": [history]} + + +def _create_system_config_table(connection: Connection) -> None: + """创建迁移所需的最小系统配置表。""" + metadata = sa.MetaData() + sa.Table( + "systemconfig", + metadata, + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("key", sa.String()), + sa.Column("value", sa.JSON()), + ) + metadata.create_all(connection) + + +def _policy_value(connection: Connection) -> Mapping[str, object]: + """读取系统配置表中的分类策略 JSON。""" + table = sa.table( + "systemconfig", + sa.column("key", sa.String()), + sa.column("value", sa.JSON()), + ) + value = connection.execute(sa.select(table.c.value).where(table.c.key == "MediaClassificationPolicy")).scalar_one() + return cast(Mapping[str, object], value) + + +def test_legacy_migration_splits_slash_names_into_safe_path_segments() -> None: + """新迁移应保留原分类名和稳定身份，仅把斜杠恢复为目录层级。""" + result = migrate_legacy_category_config( + { + "movie": { + "电影/日韩电影": {"genre_ids": "16"}, + "兜底": None, + }, + "tv": {}, + } + ) + + category = next(item for item in result.policy.categories if item.name == "电影/日韩电影") + assert category.path == ["电影", "日韩电影"] + assert result.valid + assert not result.issues + assert ClassificationPolicyValidator.validate( + result.policy, + result.extra_fields, + ).valid + + +def test_persisted_policy_migration_repairs_active_and_history_without_touching_other_categories( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """数据库迁移应修复活动和历史版本，并保留非 legacy 分类原值。""" + engine = sa.create_engine("sqlite://") + with engine.begin() as connection: + _create_system_config_table(connection) + table = sa.table( + "systemconfig", + sa.column("key", sa.String()), + sa.column("value", sa.JSON()), + ) + original = _policy_state_payload() + connection.execute( + table.insert().values( + key="MediaClassificationPolicy", + value=original, + ) + ) + migration = _bind_migration(monkeypatch, connection) + + migration.upgrade() + repaired = _policy_value(connection) + active = cast(Mapping[str, object], repaired["active"]) + history = cast(list[Mapping[str, object]], repaired["history"]) + active_categories = cast(list[Mapping[str, object]], active["categories"]) + history_categories = cast(list[Mapping[str, object]], history[0]["categories"]) + + assert active_categories[0]["path"] == ["电影", "日韩电影"] + assert history_categories[0]["path"] == ["电影", "日韩电影"] + assert active_categories[1]["path"] == ["用户分类/保留原样"] + assert active["revision"] == 2 + assert history[0]["revision"] == 1 + + migration.upgrade() + assert _policy_value(connection) == repaired + + +def test_persisted_policy_migration_revision_chain() -> None: + """策略路径修复迁移应接在 3.0.28 检查点迁移之后。""" + migration = cast(_MigrationModule, importlib.import_module(_MIGRATION)) + + assert migration.revision == "e7f3a9c1d5b2" + assert migration.down_revision == "d8f2b6a4c1e7"