From 5202c04dee5dc409092c219c43239a03978a47a6 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 29 Aug 2026 19:54:08 +0800 Subject: [PATCH] refactor(application): unify sync async decisions --- app/application/image.py | 141 +++++++++++++++++---------- app/application/messaging/ingress.py | 95 ++++++++++++------ app/application/plugin/catalog.py | 79 +++++++++++---- app/application/security/url.py | 139 +++++++++++++++----------- app/application/server/share.py | 140 ++++++++++++++++---------- 5 files changed, 380 insertions(+), 214 deletions(-) diff --git a/app/application/image.py b/app/application/image.py index a50b4a7fb..d4eb5f773 100644 --- a/app/application/image.py +++ b/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 diff --git a/app/application/messaging/ingress.py b/app/application/messaging/ingress.py index 2a3d8998c..772e56ad6 100644 --- a/app/application/messaging/ingress.py +++ b/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( diff --git a/app/application/plugin/catalog.py b/app/application/plugin/catalog.py index 4fefcfe97..c8a9d5656 100644 --- a/app/application/plugin/catalog.py +++ b/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, diff --git a/app/application/security/url.py b/app/application/security/url.py index c06cdc1ee..53203bcc6 100644 --- a/app/application/security/url.py +++ b/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( diff --git a/app/application/server/share.py b/app/application/server/share.py index 19f840e9d..1a04abecd 100644 --- a/app/application/server/share.py +++ b/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,