refactor(application): unify sync async decisions

app/application/image.py

@@ -1,6 +1,7 @@
 import io
 import threading
 from collections.abc import Mapping
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Callable, List, Optional, Protocol
 
@@ -16,6 +17,15 @@ WallpaperProvider = Callable[[], Optional[str]]
 WallpaperListProvider = Callable[[int], List[str]]
 
 
+@dataclass(frozen=True, slots=True)
+class _ImageFetchRequest:
+    """冻结同步与异步图片获取共用的缓存键和传输参数。"""
+
+    url: str
+    cache_path: str
+    use_cache: bool
+
+
 class ImageResponsePort(Protocol):
     """图片应用服务消费的最小 HTTP 响应契约。"""
 
@@ -400,6 +410,43 @@ class ImageHelper(metaclass=Singleton):
             "accept_type": "image/avif,image/webp,image/apng,*/*",
         }
 
+    @staticmethod
+    def _fetch_request(
+        url: str,
+        proxy: Optional[bool],
+        use_cache: bool,
+        cookies: Optional[str | dict[str, str]],
+    ) -> Optional[_ImageFetchRequest]:
+        """统一空 URL 拒绝、缓存键和远端传输参数生成。"""
+        if not url:
+            return None
+        return _ImageFetchRequest(
+            url=url,
+            cache_path=ImageHelper._prepare_cache_path(url),
+            use_cache=use_cache,
+        )
+
+    def _cached_image(
+        self, content: Optional[bytes]
+    ) -> Optional[tuple[bytes, str]]:
+        """缓存内容只读取格式头，返回可直接复用的图片结果。"""
+        if not content:
+            return None
+        mime_type = self.get_image_mime_type(content, verify=False)
+        return (content, mime_type) if mime_type else None
+
+    def _fetched_image(
+        self,
+        request: _ImageFetchRequest,
+        response: Optional[ImageResponsePort],
+    ) -> Optional[tuple[bytes, str]]:
+        """统一远端状态码与图片内容校验，拒绝结果不得进入缓存。"""
+        if response is None or response.status_code != 200:
+            logger.warn(f"Failed to fetch image from URL: {request.url}")
+            return None
+        mime_type = self.get_image_mime_type(response.content)
+        return (response.content, mime_type) if mime_type else None
+
     def fetch_image(
         self,
         url: str,
@@ -429,35 +476,29 @@ class ImageHelper(metaclass=Singleton):
 
         网络响应在写入缓存前完整验证一次；缓存命中仅重新识别格式头。
         """
-        if not url:
+        request = ImageHelper._fetch_request(url, proxy, use_cache, cookies)
+        if request is None:
             return None
-
-        cache_path = self._prepare_cache_path(url)
-
-        # 检查缓存
-        if use_cache:
-            content = self.file_cache.get(cache_path, region="images")
-            if content:
-                mime_type = self.get_image_mime_type(content, verify=False)
-                if mime_type:
-                    return content, mime_type
-
-        # 请求远程图片
-        params = self._get_request_params(url, proxy, cookies)
+        if request.use_cache:
+            cached_result = self._cached_image(
+                self.file_cache.get(request.cache_path, region="images")
+            )
+            if cached_result is not None:
+                return cached_result
         transport, _ = _image_ports_snapshot()
-        response = transport.get(url, options=params)
-        if response is None or response.status_code != 200:
-            logger.warn(f"Failed to fetch image from URL: {url}")
+        result = self._fetched_image(
+            request,
+            transport.get(
+                request.url,
+                options=ImageHelper._get_request_params(
+                    request.url, proxy, cookies
+                ),
+            ),
+        )
+        if result is None:
             return None
-
-        content = response.content
-        mime_type = self.get_image_mime_type(content)
-        if not mime_type:
-            return None
-
-        # 保存缓存
-        self.file_cache.set(cache_path, content, region="images")
-        return content, mime_type
+        self.file_cache.set(request.cache_path, result[0], region="images")
+        return result
 
     async def async_fetch_image(
         self,
@@ -488,32 +529,28 @@ class ImageHelper(metaclass=Singleton):
 
         网络响应在写入缓存前完整验证一次；缓存命中仅重新识别格式头。
         """
-        if not url:
+        request = ImageHelper._fetch_request(url, proxy, use_cache, cookies)
+        if request is None:
             return None
-
-        cache_path = self._prepare_cache_path(url)
-
-        # 检查缓存
-        if use_cache:
-            content = await self.async_file_cache.get(cache_path, region="images")
-            if content:
-                mime_type = self.get_image_mime_type(content, verify=False)
-                if mime_type:
-                    return content, mime_type
-
-        # 请求远程图片
-        params = self._get_request_params(url, proxy, cookies)
+        if request.use_cache:
+            cached_result = self._cached_image(
+                await self.async_file_cache.get(request.cache_path, region="images")
+            )
+            if cached_result is not None:
+                return cached_result
         transport, _ = _image_ports_snapshot()
-        response = await transport.async_get(url, options=params)
-        if response is None or response.status_code != 200:
-            logger.warn(f"Failed to fetch image from URL: {url}")
+        result = self._fetched_image(
+            request,
+            await transport.async_get(
+                request.url,
+                options=ImageHelper._get_request_params(
+                    request.url, proxy, cookies
+                ),
+            ),
+        )
+        if result is None:
             return None
-
-        content = response.content
-        mime_type = self.get_image_mime_type(content)
-        if not mime_type:
-            return None
-
-        # 保存缓存
-        await self.async_file_cache.set(cache_path, content, region="images")
-        return content, mime_type
+        await self.async_file_cache.set(
+            request.cache_path, result[0], region="images"
+        )
+        return result

app/application/messaging/ingress.py

@@ -2,6 +2,7 @@
 
 import threading
 from collections.abc import Callable, Mapping
+from dataclasses import dataclass
 from typing import Any, Optional, Protocol
 from urllib.parse import urlencode
 
@@ -11,6 +12,16 @@ from app.runtime.settings import get_runtime_setting
 BackgroundSubmitter = Callable[..., object]
 
 
+@dataclass(frozen=True, slots=True)
+class _MessageIngressRequest:
+    """冻结同步与异步本地消息回环共用的请求参数。"""
+
+    url: str
+    payload: Mapping[str, Any]
+    source: str
+    timeout: float
+
+
 class MessageIngressPort(Protocol):
     """宿主本地消息入口所需的最小同步、异步传输端口。"""
 
@@ -77,6 +88,48 @@ def build_message_ingress_url(source: str | None) -> str:
     )
 
 
+def _message_ingress_request(
+    payload: Mapping[str, Any],
+    source: str | None,
+    timeout: float,
+) -> _MessageIngressRequest:
+    """统一构造 URL、payload 副本和日志使用的来源标识。"""
+    return _MessageIngressRequest(
+        url=build_message_ingress_url(source),
+        payload=dict(payload),
+        source=source or "-",
+        timeout=timeout,
+    )
+
+
+def _message_ingress_confirmed(
+    request: _MessageIngressRequest,
+    status_code: Optional[int],
+) -> bool:
+    """统一判断本地入口是否确认接收，并输出稳定失败原因。"""
+    if status_code is None:
+        logger.error(f"转发渠道消息到本地入口失败：source={request.source} - 无响应")
+        return False
+    if status_code >= 400:
+        logger.error(
+            "转发渠道消息到本地入口失败："
+            f"source={request.source} - HTTP {status_code}"
+        )
+        return False
+    return True
+
+
+def _message_ingress_failed(
+    source: str | None,
+    error: Exception,
+) -> bool:
+    """统一记录传输或组合根异常，并按关闭策略返回失败。"""
+    logger.error(
+        f"转发渠道消息到本地入口失败：source={source or '-'} - {error}"
+    )
+    return False
+
+
 def forward_message_to_host(
     payload: Mapping[str, Any],
     source: str | None,
@@ -85,24 +138,15 @@ def forward_message_to_host(
 ) -> bool:
     """同步转发渠道 payload，统一判断本地入口是否确认接收。"""
     try:
+        request = _message_ingress_request(payload, source, timeout)
         status_code = _message_ingress_snapshot().post(
-            build_message_ingress_url(source),
-            dict(payload),
-            timeout=timeout,
+            request.url,
+            request.payload,
+            timeout=request.timeout,
         )
-        if status_code is None:
-            logger.error(f"转发渠道消息到本地入口失败：source={source or '-'} - 无响应")
-            return False
-        if status_code >= 400:
-            logger.error(
-                "转发渠道消息到本地入口失败："
-                f"source={source or '-'} - HTTP {status_code}"
-            )
-            return False
-        return True
+        return _message_ingress_confirmed(request, status_code)
     except Exception as error:
-        logger.error(f"转发渠道消息到本地入口失败：source={source or '-'} - {error}")
-        return False
+        return _message_ingress_failed(source, error)
 
 
 async def async_forward_message_to_host(
@@ -113,24 +157,15 @@ async def async_forward_message_to_host(
 ) -> bool:
     """异步转发渠道 payload，供自有事件循环的消息 SDK 复用同一确认语义。"""
     try:
+        request = _message_ingress_request(payload, source, timeout)
         status_code = await _message_ingress_snapshot().async_post(
-            build_message_ingress_url(source),
-            dict(payload),
-            timeout=timeout,
+            request.url,
+            request.payload,
+            timeout=request.timeout,
         )
-        if status_code is None:
-            logger.error(f"转发渠道消息到本地入口失败：source={source or '-'} - 无响应")
-            return False
-        if status_code >= 400:
-            logger.error(
-                "转发渠道消息到本地入口失败："
-                f"source={source or '-'} - HTTP {status_code}"
-            )
-            return False
-        return True
+        return _message_ingress_confirmed(request, status_code)
     except Exception as error:
-        logger.error(f"转发渠道消息到本地入口失败：source={source or '-'} - {error}")
-        return False
+        return _message_ingress_failed(source, error)
 
 
 def submit_message_to_host(

app/application/plugin/catalog.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import asyncio
 import concurrent.futures
 from collections.abc import Awaitable, Callable, Mapping, Sequence
+from dataclasses import dataclass
 from typing import Any, Optional, cast
 
 from app.application.plugin.identity import (
@@ -30,6 +31,15 @@ PluginMapper = Callable[[str, dict, str, list[str], int, Optional[str]], Any]
 ProgressCallback = Callable[..., Any]
 
 
+@dataclass(frozen=True, slots=True)
+class _CatalogLoadRequest:
+    """冻结同步与异步插件市场加载共用的仓库和代际参数。"""
+
+    market: str
+    package_version: Optional[str]
+    force: bool
+
+
 def apply_declared_metadata_fallback(
     plugins: Sequence[Plugin],
     identities: Mapping[str, PluginIdentity],
@@ -104,6 +114,37 @@ class PluginCatalogService:
         self._warning = warning
         self._error = error
 
+    @staticmethod
+    def _load_request(
+            market: str,
+            package_version: Optional[str],
+            force: bool,
+    ) -> Optional[_CatalogLoadRequest]:
+        """统一拒绝空市场，并冻结一次目录加载的完整输入。"""
+        if not market:
+            return None
+        return _CatalogLoadRequest(
+            market=market,
+            package_version=package_version,
+            force=force,
+        )
+
+    def _loaded_plugins(
+            self,
+            request: _CatalogLoadRequest,
+            online_plugins: Optional[dict[str, dict[str, Any]]],
+    ) -> list[Any]:
+        """统一加载失败告警与在线目录到插件 DTO 的投影。"""
+        if online_plugins is None:
+            self._warning(
+                f"获取{request.package_version if request.package_version else ''}插件库失败："
+                f"{request.market}，请检查 GitHub 网络连接"
+            )
+            return []
+        return self._map_plugins(
+            online_plugins, request.market, request.package_version
+        )
+
     def load(
             self,
             market: str,
@@ -111,16 +152,17 @@ class PluginCatalogService:
             force: bool = False,
     ) -> list[Any]:
         """同步读取并映射指定市场和插件代际。"""
-        if not market:
+        request = PluginCatalogService._load_request(
+            market, package_version, force
+        )
+        if request is None:
             return []
-        online_plugins = self._market_loader(market, package_version, force)
-        if online_plugins is None:
-            self._warning(
-                f"获取{package_version if package_version else ''}插件库失败："
-                f"{market}，请检查 GitHub 网络连接"
-            )
-            return []
-        return self._map_plugins(online_plugins, market, package_version)
+        return self._loaded_plugins(
+            request,
+            self._market_loader(
+                request.market, request.package_version, request.force
+            ),
+        )
 
     async def async_load(
             self,
@@ -129,20 +171,17 @@ class PluginCatalogService:
             force: bool = False,
     ) -> list[Any]:
         """异步读取并映射指定市场和插件代际。"""
-        if not market:
+        request = PluginCatalogService._load_request(
+            market, package_version, force
+        )
+        if request is None:
             return []
         online_plugins = await self._async_market_loader(
-            market,
-            package_version,
-            force,
+            request.market,
+            request.package_version,
+            request.force,
         )
-        if online_plugins is None:
-            self._warning(
-                f"获取{package_version if package_version else ''}插件库失败："
-                f"{market}，请检查 GitHub 网络连接"
-            )
-            return []
-        return self._map_plugins(online_plugins, market, package_version)
+        return self._loaded_plugins(request, online_plugins)
 
     def collect(
             self,

app/application/security/url.py

@@ -8,7 +8,7 @@ from dataclasses import dataclass, field
 from enum import Enum
 from hashlib import sha256
 from pathlib import Path
-from typing import Dict, Iterable, List, Optional, Protocol, Set, Union
+from typing import Dict, Iterable, List, Optional, Protocol, Set, Union, cast
 from urllib.parse import parse_qsl, quote, urlencode, urlparse, urlunparse
 
 from anyio import Path as AsyncPath
@@ -176,6 +176,26 @@ class SecurityUtils:
     _SIGNED_URL_PURPOSE = "image-proxy"
     _SUBTITLE_DOWNLOAD_PURPOSE_PREFIX = "subtitle-download"
 
+    @staticmethod
+    def _resolved_path_is_safe(
+        base_path: Union[Path, AsyncPath],
+        user_path: Union[Path, AsyncPath],
+        user_suffix: str,
+        allowed_suffixes: Optional[Union[Set[str], List[str]]],
+    ) -> bool:
+        """统一解析后路径 containment 与后缀白名单判定。"""
+        if base_path != user_path and base_path not in user_path.parents:
+            return False
+        if allowed_suffixes is None:
+            return True
+        return user_suffix.lower() in set(allowed_suffixes)
+
+    @staticmethod
+    def _path_validation_failed(error: Exception) -> bool:
+        """统一记录路径解析异常并按默认拒绝策略返回。"""
+        logger.debug(f"Error occurred while validating paths: {error}")
+        return False
+
     @staticmethod
     def is_safe_path(base_path: Path, user_path: Path,
                      allowed_suffixes: Optional[Union[Set[str], List[str]]] = None) -> bool:
@@ -193,19 +213,14 @@ class SecurityUtils:
             base_path_resolved = base_path.resolve()
             user_path_resolved = user_path.resolve()
 
-            # 检查用户路径是否在基准目录或基准目录的子目录内
-            if base_path_resolved != user_path_resolved and base_path_resolved not in user_path_resolved.parents:
-                return False
-
-            if allowed_suffixes is not None:
-                allowed_suffixes = set(allowed_suffixes)
-                if user_path.suffix.lower() not in allowed_suffixes:
-                    return False
-
-            return True
+            return SecurityUtils._resolved_path_is_safe(
+                base_path_resolved,
+                user_path_resolved,
+                user_path.suffix,
+                allowed_suffixes,
+            )
         except Exception as e:
-            logger.debug(f"Error occurred while validating paths: {e}")
-            return False
+            return SecurityUtils._path_validation_failed(e)
 
     @staticmethod
     async def async_is_safe_path(base_path: AsyncPath, user_path: AsyncPath,
@@ -224,19 +239,14 @@ class SecurityUtils:
             base_path_resolved = await base_path.resolve()
             user_path_resolved = await user_path.resolve()
 
-            # 检查用户路径是否在基准目录或基准目录的子目录内
-            if base_path_resolved != user_path_resolved and base_path_resolved not in user_path_resolved.parents:
-                return False
-
-            if allowed_suffixes is not None:
-                allowed_suffixes = set(allowed_suffixes)
-                if user_path.suffix.lower() not in allowed_suffixes:
-                    return False
-
-            return True
+            return SecurityUtils._resolved_path_is_safe(
+                base_path_resolved,
+                user_path_resolved,
+                user_path.suffix,
+                allowed_suffixes,
+            )
         except Exception as e:
-            logger.debug(f"Error occurred while validating paths: {e}")
-            return False
+            return SecurityUtils._path_validation_failed(e)
 
     @staticmethod
     def _literal_ip(hostname: str) -> Optional[ipaddress._BaseAddress]:
@@ -721,6 +731,37 @@ class SecurityUtils:
         )
         return diagnosis.allowed
 
+    @staticmethod
+    def _url_safety_prelude(
+        url: str,
+        allowed_domains: Union[Set[str], List[str]],
+        strict: bool,
+        block_private: bool,
+    ) -> tuple[Optional[str], Optional["UrlSafetyDiagnosis"]]:
+        """统一协议/域名白名单与无需 DNS 场景的终态诊断。"""
+        hostname = SecurityUtils._check_url_allowlist(url, allowed_domains, strict)
+        if hostname is None:
+            return None, UrlSafetyDiagnosis(
+                allowed=False,
+                reason=UrlSafetyReason.DOMAIN_NOT_ALLOWED,
+            )
+        if not block_private:
+            return hostname, UrlSafetyDiagnosis(
+                allowed=True,
+                reason=UrlSafetyReason.ALLOWED,
+                host=hostname,
+            )
+        return hostname, None
+
+    @staticmethod
+    def _url_safety_failed(error: Exception) -> "UrlSafetyDiagnosis":
+        """统一记录 URL 校验异常并返回默认拒绝诊断。"""
+        logger.debug(f"Error occurred while validating URL: {error}")
+        return UrlSafetyDiagnosis(
+            allowed=False,
+            reason=UrlSafetyReason.DOMAIN_NOT_ALLOWED,
+        )
+
     @staticmethod
     def evaluate_url_safety(
         url: str,
@@ -738,28 +779,18 @@ class SecurityUtils:
         归类为 `DOMAIN_NOT_ALLOWED`，避免任何解析路径漏过 SSRF 校验。
         """
         try:
-            hostname = SecurityUtils._check_url_allowlist(url, allowed_domains, strict)
-            if hostname is None:
-                return UrlSafetyDiagnosis(
-                    allowed=False,
-                    reason=UrlSafetyReason.DOMAIN_NOT_ALLOWED,
-                )
-            if not block_private:
-                return UrlSafetyDiagnosis(
-                    allowed=True,
-                    reason=UrlSafetyReason.ALLOWED,
-                    host=hostname,
-                )
+            hostname, diagnosis = SecurityUtils._url_safety_prelude(
+                url, allowed_domains, strict, block_private
+            )
+            if diagnosis is not None:
+                return diagnosis
+            hostname = cast(str, hostname)
             addresses = SecurityUtils._hostname_addresses(hostname)
             return SecurityUtils._diagnose_resolved_addresses(
                 url, hostname, addresses, allowed_private_ranges
             )
         except Exception as e:  # noqa: BLE001 - 默认拒绝，避免漏过 SSRF 校验
-            logger.debug(f"Error occurred while validating URL: {e}")
-            return UrlSafetyDiagnosis(
-                allowed=False,
-                reason=UrlSafetyReason.DOMAIN_NOT_ALLOWED,
-            )
+            return SecurityUtils._url_safety_failed(e)
 
     @staticmethod
     async def evaluate_url_safety_async(
@@ -776,28 +807,18 @@ class SecurityUtils:
         的事件循环；校验顺序、字段含义、异常归类均与同步版本相同。
         """
         try:
-            hostname = SecurityUtils._check_url_allowlist(url, allowed_domains, strict)
-            if hostname is None:
-                return UrlSafetyDiagnosis(
-                    allowed=False,
-                    reason=UrlSafetyReason.DOMAIN_NOT_ALLOWED,
-                )
-            if not block_private:
-                return UrlSafetyDiagnosis(
-                    allowed=True,
-                    reason=UrlSafetyReason.ALLOWED,
-                    host=hostname,
-                )
+            hostname, diagnosis = SecurityUtils._url_safety_prelude(
+                url, allowed_domains, strict, block_private
+            )
+            if diagnosis is not None:
+                return diagnosis
+            hostname = cast(str, hostname)
             addresses = await SecurityUtils._hostname_addresses_async(hostname)
             return SecurityUtils._diagnose_resolved_addresses(
                 url, hostname, addresses, allowed_private_ranges
             )
         except Exception as e:  # noqa: BLE001 - 默认拒绝，避免漏过 SSRF 校验
-            logger.debug(f"Error occurred while validating URL: {e}")
-            return UrlSafetyDiagnosis(
-                allowed=False,
-                reason=UrlSafetyReason.DOMAIN_NOT_ALLOWED,
-            )
+            return SecurityUtils._url_safety_failed(e)
 
     @staticmethod
     async def is_safe_image_url_async(

app/application/server/share.py

@@ -4,13 +4,22 @@ from __future__ import annotations
 
 import json
 from collections.abc import Awaitable, Callable
-from dataclasses import asdict
+from dataclasses import asdict, dataclass
 from typing import Any, Optional
 
 from app.application.workflow import WorkflowSnapshot
 from app.schemas.media import resolve_media_identity
 
 
+@dataclass(frozen=True, slots=True)
+class _ShareMetadata:
+    """冻结订阅与工作流分享共用的用户填写元数据。"""
+
+    title: str
+    comment: str
+    user: str
+
+
 class ServerSharingService:
     """协调本地订阅、工作流读取与中心服务分享传输。"""
 
@@ -92,6 +101,53 @@ class ServerSharingService:
             return False, "请分享有动作和流程的工作流"
         return True, ""
 
+    @staticmethod
+    def _sharing_disabled(
+        enabled: bool,
+        kind: str,
+    ) -> Optional[tuple[bool, str]]:
+        """统一共享开关拒绝结果，同时保持不同实体的用户提示。"""
+        if enabled:
+            return None
+        return False, f"当前没有开启{kind}数据共享功能"
+
+    def _share_fields(self, metadata: _ShareMetadata) -> dict[str, str]:
+        """统一生成分享标题、备注、用户和当前宿主身份字段。"""
+        return {
+            "share_title": metadata.title,
+            "share_comment": metadata.comment,
+            "share_user": metadata.user,
+            "share_uid": self._user_uuid_provider(),
+        }
+
+    def _subscribe_share_payload(
+        self,
+        subscribe: Any,
+        metadata: _ShareMetadata,
+    ) -> tuple[Optional[dict[str, Any]], str]:
+        """统一订阅存在性、宿主字段投影和媒体身份完整性校验。"""
+        if not subscribe:
+            return None, "订阅不存在"
+        payload = self.build_subscribe_payload({
+            **self._share_fields(metadata),
+            **subscribe.to_dict(),
+        })
+        return (payload, "") if payload else (None, "订阅媒体身份不完整")
+
+    def _workflow_share_payload(
+        self,
+        workflow: Optional[WorkflowSnapshot],
+        metadata: _ShareMetadata,
+    ) -> tuple[Optional[dict[str, Any]], str]:
+        """统一工作流完整性校验和中心服务载荷投影。"""
+        valid, message = self.validate_workflow(workflow)
+        if not valid or workflow is None:
+            return None, message
+        return {
+            **self._share_fields(metadata),
+            **self.prepare_workflow(workflow),
+        }, ""
+
     def share_subscribe(
             self,
             *,
@@ -102,20 +158,15 @@ class ServerSharingService:
             share_user: str,
     ) -> tuple[bool, str]:
         """同步读取并分享指定订阅。"""
-        if not enabled:
-            return False, "当前没有开启订阅数据共享功能"
+        disabled = self._sharing_disabled(enabled, "订阅")
+        if disabled is not None:
+            return disabled
         subscribe = self._subscribe_provider(subscribe_id)
-        if not subscribe:
-            return False, "订阅不存在"
-        payload = self.build_subscribe_payload({
-            "share_title": share_title,
-            "share_comment": share_comment,
-            "share_user": share_user,
-            "share_uid": self._user_uuid_provider(),
-            **subscribe.to_dict(),
-        })
-        if not payload:
-            return False, "订阅媒体身份不完整"
+        payload, message = self._subscribe_share_payload(
+            subscribe, _ShareMetadata(share_title, share_comment, share_user)
+        )
+        if payload is None:
+            return False, message
         return self._response_handler(
             self._subscribe_sender(payload),
             self._subscribe_cache_clearer,
@@ -131,20 +182,15 @@ class ServerSharingService:
             share_user: str,
     ) -> tuple[bool, str]:
         """异步读取并分享指定订阅。"""
-        if not enabled:
-            return False, "当前没有开启订阅数据共享功能"
+        disabled = self._sharing_disabled(enabled, "订阅")
+        if disabled is not None:
+            return disabled
         subscribe = await self._async_subscribe_provider(subscribe_id)
-        if not subscribe:
-            return False, "订阅不存在"
-        payload = self.build_subscribe_payload({
-            "share_title": share_title,
-            "share_comment": share_comment,
-            "share_user": share_user,
-            "share_uid": self._user_uuid_provider(),
-            **subscribe.to_dict(),
-        })
-        if not payload:
-            return False, "订阅媒体身份不完整"
+        payload, message = self._subscribe_share_payload(
+            subscribe, _ShareMetadata(share_title, share_comment, share_user)
+        )
+        if payload is None:
+            return False, message
         return self._response_handler(
             await self._async_subscribe_sender(payload),
             self._subscribe_cache_clearer,
@@ -160,21 +206,15 @@ class ServerSharingService:
             share_user: str,
     ) -> tuple[bool, str]:
         """同步读取并分享指定工作流。"""
-        if not enabled:
-            return False, "当前没有开启工作流数据共享功能"
+        disabled = self._sharing_disabled(enabled, "工作流")
+        if disabled is not None:
+            return disabled
         workflow = self._workflow_provider(workflow_id)
-        valid, message = self.validate_workflow(workflow)
-        if not valid:
+        payload, message = self._workflow_share_payload(
+            workflow, _ShareMetadata(share_title, share_comment, share_user)
+        )
+        if payload is None:
             return False, message
-        if workflow is None:
-            return False, "工作流不存在"
-        payload = {
-            "share_title": share_title,
-            "share_comment": share_comment,
-            "share_user": share_user,
-            "share_uid": self._user_uuid_provider(),
-            **self.prepare_workflow(workflow),
-        }
         return self._response_handler(
             self._workflow_sender(payload),
             self._workflow_cache_clearer,
@@ -190,21 +230,15 @@ class ServerSharingService:
             share_user: str,
     ) -> tuple[bool, str]:
         """异步读取并分享指定工作流。"""
-        if not enabled:
-            return False, "当前没有开启工作流数据共享功能"
+        disabled = self._sharing_disabled(enabled, "工作流")
+        if disabled is not None:
+            return disabled
         workflow = await self._async_workflow_provider(workflow_id)
-        valid, message = self.validate_workflow(workflow)
-        if not valid:
+        payload, message = self._workflow_share_payload(
+            workflow, _ShareMetadata(share_title, share_comment, share_user)
+        )
+        if payload is None:
             return False, message
-        if workflow is None:
-            return False, "工作流不存在"
-        payload = {
-            "share_title": share_title,
-            "share_comment": share_comment,
-            "share_user": share_user,
-            "share_uid": self._user_uuid_provider(),
-            **self.prepare_workflow(workflow),
-        }
         return self._response_handler(
             await self._async_workflow_sender(payload),
             self._workflow_cache_clearer,