fix(classification): repair legacy slash category paths

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,

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

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 旧字段；仍被旧规则使用的字段会标记为“旧规则”，只能查看，
 改写规则时应优先选择新的统一字段。分类编辑弹窗和帮助弹窗支持透明主题，分类下拉选项会避免名称与路径末级重复显示。
 

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"