refactor(application): 收敛共享重复逻辑

app/application/classification/enrichment.py

@@ -16,6 +16,7 @@ from typing import Any, Protocol, TypeAlias, cast
 from pydantic import ValidationError
 
 from app.domain.classification.evaluator import read_fact
+from app.domain.classification.conditions import condition_field_ids
 from app.domain.classification.fields import (
     classification_fact_matches_definition,
     field_definition_map,
@@ -429,7 +430,7 @@ def _referenced_missing_standard_fields(
             continue
         if rule.sources and media_source not in rule.sources:
             continue
-        for field_id in _condition_field_ids(rule.when):
+        for field_id in condition_field_ids(rule.when):
             if field_id.startswith(("identity.", "extensions.")):
                 continue
             definition = definitions.get(field_id)
@@ -441,27 +442,6 @@ def _referenced_missing_standard_fields(
     return tuple(fields)
 
 
-def _condition_field_ids(node: ClassificationConditionNode) -> tuple[str, ...]:
-    """按策略声明顺序递归返回条件树叶子字段。"""
-    if isinstance(node, ClassificationCondition):
-        return (node.field,)
-    if not isinstance(node, ClassificationConditionGroup):
-        return ()
-    if node.all is not None:
-        children: Sequence[ClassificationConditionNode] = node.all
-    elif node.any is not None:
-        children = node.any
-    elif node.not_ is not None:
-        children = (node.not_,)
-    else:
-        children = ()
-    return tuple(
-        field_id
-        for child in children
-        for field_id in _condition_field_ids(child)
-    )
-
-
 def _provider_fields(
     provider: ClassificationEnrichmentProvider,
     missing_fields: Sequence[str],

app/application/classification/migration.py

@@ -22,6 +22,7 @@ from app.domain.classification.vocabulary import (
 from app.domain.classification.vocabulary import (
     TMDB_GENRE_KEYS as _TMDB_GENRE_KEYS,
 )
+from app.domain.classification.conditions import condition_field_ids
 from app.schemas.category import (
     CategoryConfig,
     CategoryRule,
@@ -222,7 +223,7 @@ def legacy_extension_fields_from_policy(
     """按策略条件和别名重建可直接注册的 TMDB 旧比较扩展字段声明。"""
     context = _MigrationContext()
     for rule in policy.rules:
-        for field_id in _condition_field_ids(rule.when):
+        for field_id in condition_field_ids(rule.when):
             if not field_id.startswith(_EXTENSION_PREFIX):
                 continue
             field_name = field_id.removeprefix(_EXTENSION_PREFIX)
@@ -392,7 +393,7 @@ def _migrate_media_categories(
             if any(
                 field_id.startswith(_EXTENSION_PREFIX)
                 for node in nodes
-                for field_id in _condition_field_ids(node)
+                for field_id in condition_field_ids(node)
             )
             else []
         )
@@ -943,21 +944,6 @@ def _error_count(diagnostics: Sequence[LegacyClassificationDiagnostic]) -> int:
     return sum(item.severity == "error" for item in diagnostics)
 
 
-def _condition_field_ids(node: ClassificationConditionNode) -> list[str]:
-    """按条件树顺序提取全部叶子字段 ID。"""
-    if isinstance(node, ClassificationCondition):
-        return [node.field]
-    if node.all is not None:
-        children = node.all
-    elif node.any is not None:
-        children = node.any
-    elif node.not_ is not None:
-        children = [node.not_]
-    else:
-        children = []
-    return [field_id for child in children for field_id in _condition_field_ids(child)]
-
-
 def _append_unique(values: list[str], value: str) -> None:
     """按首次出现顺序追加非重复字符串。"""
     if value not in values:

app/application/classification/projection.py

@@ -16,6 +16,7 @@ from app.application.classification.migration import (
     LegacyDiagnosticPathPart,
     LegacyMediaKey,
 )
+from app.domain.classification.conditions import condition_field_ids
 from app.domain.classification.vocabulary import TMDB_GENRE_KEYS
 from app.schemas.category import (
     CategoryConfig,
@@ -473,7 +474,7 @@ def _controlled_tmdb_fields(
     field_ids: list[str] = []
     if isinstance(policy_or_field_defs, ClassificationPolicy):
         for rule in policy_or_field_defs.rules:
-            field_ids.extend(_condition_field_ids(rule.when))
+            field_ids.extend(condition_field_ids(rule.when))
     else:
         for definition in policy_or_field_defs:
             if definition.source_support.get(_TMDB_SOURCE) == "extension":
@@ -489,21 +490,6 @@ def _controlled_tmdb_fields(
     return tuple(fields)
 
 
-def _condition_field_ids(node: ClassificationConditionNode) -> list[str]:
-    """按条件树顺序提取全部叶子字段 ID。"""
-    if isinstance(node, ClassificationCondition):
-        return [node.field]
-    if node.all is not None:
-        children = node.all
-    elif node.any is not None:
-        children = node.any
-    elif node.not_ is not None:
-        children = [node.not_]
-    else:
-        children = []
-    return [field_id for child in children for field_id in _condition_field_ids(child)]
-
-
 def _project_legacy_tmdb_field(
     field_name: str,
     tmdb_info: Mapping[str, object],

app/application/download/classification.py

@@ -13,6 +13,7 @@ from app.application.directory import (
     build_media_download_path,
     validate_download_save_path,
 )
+from app.application.download.validation import validate_torrent_hash
 from app.application.history import DownloadHistorySnapshot
 from app.domain.classification.validation import validate_classification_category_path
 from app.domain.context import MediaInfo, MusicInfo
@@ -161,14 +162,6 @@ class DownloadSourceClassificationService:
         self._update_torrent = update_torrent
         self._resolve_plan = resolve_plan
 
-    @staticmethod
-    def _validate_hash(hash_value: str) -> None:
-        """校验 BitTorrent v1 Hash，拒绝模糊任务定位。"""
-        if len(hash_value) != 40 or any(
-            character not in "0123456789abcdefABCDEF" for character in hash_value
-        ):
-            raise ValueError("hash 格式无效")
-
     def plan(
         self,
         *,
@@ -178,7 +171,7 @@ class DownloadSourceClassificationService:
         media_category: Optional[str] = None,
     ) -> dict[str, Any]:
         """生成分类计划，只有 execute 为真时才请求下载器移动。"""
-        self._validate_hash(hash_value)
+        validate_torrent_hash(hash_value)
         torrents = self._list_torrents(
             hashs=[hash_value],
             downloader=downloader,

app/application/download/organization.py

@@ -6,6 +6,7 @@ from typing import Any
 
 from app.application.configuration import get_configured_system_config
 from app.application.directory import DirectoryHelper, validate_download_save_path
+from app.application.download.validation import validate_torrent_hash
 from app.domain.classification.validation import validate_classification_category_path
 from app.domain.meta.metabase import MetaBase
 from app.domain.meta.metamusic import MetaMusic
@@ -286,8 +287,7 @@ def _rename_qb_root(chain: Any, downloader: str, hash_value: str, old_name: str,
 
 def organize_existing_source(hash_value: str, request: Any, chain: Any, media_chain: Any) -> dict[str, Any]:
     """生成可重放的预览计划，确认后仅通过下载器修改任务路径。"""
-    if not re.fullmatch(r"[0-9a-fA-F]{40}", hash_value):
-        raise ValueError("hash 格式无效")
+    validate_torrent_hash(hash_value)
     history = chain.download_history_repository.get_by_hash(hash_value)
     if history is None:
         raise ValueError("未找到该任务的下载历史")

app/application/download/tasks.py

@@ -3,6 +3,7 @@
 from typing import Any, Callable, List, Optional
 
 from app.application.directory import validate_download_save_path
+from app.application.download.validation import validate_torrent_hash
 from app.application.history import DownloadHistorySnapshot
 from app.schemas.transfer import DownloaderTorrent, DownloadTaskMedia
 from app.schemas.types import TorrentStatus
@@ -99,12 +100,6 @@ class DownloadTaskMutationService:
         self._set_downloading = set_downloading
         self._update_torrent = update_torrent
 
-    @staticmethod
-    def _validate_hash(hash_value: str) -> None:
-        """校验 BitTorrent v1 Hash，拒绝模糊任务定位。"""
-        if len(hash_value) != 40 or any(character not in "0123456789abcdefABCDEF" for character in hash_value):
-            raise ValueError("hash 格式无效")
-
     @staticmethod
     def _normalize_list(values: Optional[list[str]]) -> Optional[list[str]]:
         """移除字符串列表中的空项。"""
@@ -137,7 +132,7 @@ class DownloadTaskMutationService:
         seeding_time_limit: Optional[int] = None,
     ) -> dict[str, Any]:
         """校验并执行启停、标签、限速、Tracker 与保存位置修改。"""
-        self._validate_hash(hash_value)
+        validate_torrent_hash(hash_value)
         if action not in {None, "start", "stop"}:
             raise ValueError("action 只支持 start 或 stop")
         tags = self._normalize_list(tags)

app/application/download/validation.py

@@ -0,0 +1,13 @@
+"""下载应用服务共享的输入校验。"""
+
+from __future__ import annotations
+
+import re
+
+_TORRENT_HASH_PATTERN = re.compile(r"^[0-9a-fA-F]{40}$")
+
+
+def validate_torrent_hash(hash_value: str) -> None:
+    """校验 BitTorrent v1 Hash，拒绝模糊任务定位。"""
+    if not _TORRENT_HASH_PATTERN.fullmatch(hash_value):
+        raise ValueError("hash 格式无效")

app/application/server/payload.py

@@ -0,0 +1,24 @@
+"""中心服务订阅载荷的共享投影。"""
+
+from __future__ import annotations
+
+from collections.abc import Collection
+from typing import Any
+
+from app.schemas.media import resolve_media_identity
+
+
+def build_subscribe_payload(
+    item: object,
+    fields: Collection[str],
+) -> dict[str, Any] | None:
+    """过滤本地订阅字段并规范化中心服务使用的媒体身份。"""
+    if not isinstance(item, dict):
+        return None
+    media_source, media_id = resolve_media_identity(media=item)
+    if not media_source or not media_id:
+        return None
+    payload = {key: value for key, value in item.items() if key in fields}
+    payload["media_source"] = str(media_source)
+    payload["media_id"] = media_id
+    return payload

app/application/server/report.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 from collections.abc import Awaitable, Callable
 from typing import Any, Optional
 
-from app.schemas.media import resolve_media_identity
+from app.application.server.payload import build_subscribe_payload
 
 
 class ServerReportService:
@@ -78,19 +78,7 @@ class ServerReportService:
 
     def build_subscribe_payload(self, item: Optional[dict]) -> Optional[dict]:
         """构造中心服务订阅统计载荷并移除本地运行字段。"""
-        if not isinstance(item, dict):
-            return None
-        media_source, media_id = resolve_media_identity(media=item)
-        if not media_source or not media_id:
-            return None
-        payload = {
-            key: value
-            for key, value in item.items()
-            if key in self.SUBSCRIBE_FIELDS
-        }
-        payload["media_source"] = str(media_source)
-        payload["media_id"] = media_id
-        return payload
+        return build_subscribe_payload(item, self.SUBSCRIBE_FIELDS)
 
     def build_plugin_payload(
             self,

app/application/server/share.py

@@ -8,7 +8,7 @@ from dataclasses import asdict, dataclass
 from typing import Any, Optional
 
 from app.application.workflow import WorkflowSnapshot
-from app.schemas.media import resolve_media_identity
+from app.application.server.payload import build_subscribe_payload
 
 
 @dataclass(frozen=True, slots=True)
@@ -67,19 +67,7 @@ class ServerSharingService:
 
     def build_subscribe_payload(self, item: Optional[dict]) -> Optional[dict]:
         """构造订阅分享载荷并隔离本地字段和旧专用 ID。"""
-        if not isinstance(item, dict):
-            return None
-        media_source, media_id = resolve_media_identity(media=item)
-        if not media_source or not media_id:
-            return None
-        payload = {
-            key: value
-            for key, value in item.items()
-            if key in self.SUBSCRIBE_FIELDS
-        }
-        payload["media_source"] = str(media_source)
-        payload["media_id"] = media_id
-        return payload
+        return build_subscribe_payload(item, self.SUBSCRIBE_FIELDS)
 
     @staticmethod
     def prepare_workflow(workflow: WorkflowSnapshot) -> dict:

app/domain/classification/conditions.py

@@ -0,0 +1,30 @@
+"""分类条件树的共享纯函数。"""
+
+from __future__ import annotations
+
+from app.schemas.category import (
+    ClassificationCondition,
+    ClassificationConditionGroup,
+    ClassificationConditionNode,
+)
+
+
+def condition_field_ids(node: ClassificationConditionNode) -> tuple[str, ...]:
+    """按条件树声明顺序递归返回全部叶子字段 ID。"""
+    if isinstance(node, ClassificationCondition):
+        return (node.field,)
+    if not isinstance(node, ClassificationConditionGroup):
+        return ()
+    if node.all is not None:
+        children = node.all
+    elif node.any is not None:
+        children = node.any
+    elif node.not_ is not None:
+        children = (node.not_,)
+    else:
+        children = ()
+    return tuple(
+        field_id
+        for child in children
+        for field_id in condition_field_ids(child)
+    )

tests/test_application_shared_utilities.py

@@ -0,0 +1,63 @@
+"""Application 重复逻辑收敛后的共享纯函数回归测试。"""
+
+from types import SimpleNamespace
+
+import pytest
+
+from app.application.download.validation import validate_torrent_hash
+from app.application.server.payload import build_subscribe_payload
+from app.domain.classification.conditions import condition_field_ids
+from app.schemas.category import ClassificationCondition, ClassificationConditionGroup
+
+
+def test_condition_field_ids_preserves_nested_declaration_order() -> None:
+    """条件树公共遍历必须保留 all/any/not 的声明顺序。"""
+    node = ClassificationConditionGroup(
+        all=[
+            ClassificationCondition(field="media.title", operator="exists"),
+            ClassificationConditionGroup(
+                any=[
+                    ClassificationCondition(field="media.year", operator="exists"),
+                    ClassificationConditionGroup(
+                        not_=ClassificationCondition(
+                            field="media.adult",
+                            operator="exists",
+                        )
+                    ),
+                ]
+            ),
+        ]
+    )
+
+    assert condition_field_ids(node) == (
+        "media.title",
+        "media.year",
+        "media.adult",
+    )
+
+
+@pytest.mark.parametrize("value", ["", "a" * 39, "g" * 40, "a" * 41])
+def test_validate_torrent_hash_rejects_non_v1_hash(value: str) -> None:
+    """下载应用服务必须拒绝长度或字符集不符合 v1 Hash 的输入。"""
+    with pytest.raises(ValueError, match="hash 格式无效"):
+        validate_torrent_hash(value)
+
+
+def test_build_subscribe_payload_filters_fields_and_requires_identity() -> None:
+    """中心服务公共载荷投影只保留白名单字段并要求媒体身份完整。"""
+    payload = build_subscribe_payload(
+        {
+            "name": "Demo",
+            "media_source": "tmdb",
+            "media_id": "123",
+            "private": "hidden",
+        },
+        {"name", "media_source", "media_id"},
+    )
+
+    assert payload == {
+        "name": "Demo",
+        "media_source": "themoviedb",
+        "media_id": "123",
+    }
+    assert build_subscribe_payload(SimpleNamespace(), {"name"}) is None

tests/test_architecture_dependencies.py

@@ -392,6 +392,7 @@ def test_domain_classification_is_a_pure_direct_import_package() -> None:
     package = APP_ROOT / "domain" / "classification"
     assert {path.name for path in package.glob("*.py")} == {
         "__init__.py",
+        "conditions.py",
         "evaluator.py",
         "facts.py",
         "fields.py",