From 71364ea579bd85971a9faf8fec23c613e4c581a0 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Thu, 21 May 2026 10:39:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(alipay):=20=E6=96=B0=E5=A2=9E=E6=94=AF?= =?UTF-8?q?=E4=BB=98=E5=AE=9D=E6=B8=A0=E9=81=93=E6=8F=92=E4=BB=B6=E5=AE=8C?= =?UTF-8?q?=E6=95=B4=E5=AE=9E=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实现了支付宝生活号渠道的完整功能，包括消息接收回调、发送、安全验证、去重、配对绑定、流式回复支持等完整能力。 --- .../channel/extensions/alipay/__init__.py | 440 ++++++++++++++++++ .../yuxi/channel/extensions/alipay/config.py | 169 +++++++ .../channel/extensions/alipay/constants.py | 36 ++ .../yuxi/channel/extensions/alipay/crypto.py | 82 ++++ .../yuxi/channel/extensions/alipay/dedup.py | 28 ++ .../yuxi/channel/extensions/alipay/errors.py | 65 +++ .../yuxi/channel/extensions/alipay/format.py | 34 ++ .../yuxi/channel/extensions/alipay/gateway.py | 124 +++++ .../yuxi/channel/extensions/alipay/media.py | 42 ++ .../yuxi/channel/extensions/alipay/monitor.py | 97 ++++ .../channel/extensions/alipay/outbound.py | 314 +++++++++++++ .../yuxi/channel/extensions/alipay/pairing.py | 32 ++ .../channel/extensions/alipay/plugin.json | 29 ++ .../channel/extensions/alipay/security.py | 42 ++ .../yuxi/channel/extensions/alipay/status.py | 76 +++ .../channel/extensions/alipay/streaming.py | 33 ++ .../yuxi/channel/extensions/alipay/types.py | 67 +++ .../yuxi/channel/extensions/alipay/webhook.py | 258 ++++++++++ 18 files changed, 1968 insertions(+) create mode 100644 backend/package/yuxi/channel/extensions/alipay/__init__.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/config.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/constants.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/crypto.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/dedup.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/errors.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/format.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/gateway.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/media.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/monitor.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/outbound.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/pairing.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/plugin.json create mode 100644 backend/package/yuxi/channel/extensions/alipay/security.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/status.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/streaming.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/types.py create mode 100644 backend/package/yuxi/channel/extensions/alipay/webhook.py diff --git a/backend/package/yuxi/channel/extensions/alipay/__init__.py b/backend/package/yuxi/channel/extensions/alipay/__init__.py new file mode 100644 index 00000000..df947f3b --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/__init__.py @@ -0,0 +1,440 @@ +import logging + +from yuxi.channel.capabilities import ChannelCapabilities +from yuxi.channel.context import ChannelContext +from yuxi.channel.extensions.alipay.config import AlipayConfig +from yuxi.channel.extensions.alipay.dedup import AlipayMessageDeduplicator +from yuxi.channel.extensions.alipay.errors import AlipayError, classify_alipay_error +from yuxi.channel.extensions.alipay.format import remove_markdown +from yuxi.channel.extensions.alipay.gateway import AlipayGateway +from yuxi.channel.extensions.alipay.media import AlipayMedia +from yuxi.channel.extensions.alipay.monitor import convert_alipay_event +from yuxi.channel.extensions.alipay.outbound import AlipayOutbound +from yuxi.channel.extensions.alipay.pairing import AlipayPairing +from yuxi.channel.extensions.alipay.security import AlipaySecurity +from yuxi.channel.extensions.alipay.status import AlipayStatus +from yuxi.channel.extensions.alipay.streaming import AlipayStreaming +from yuxi.channel.extensions.alipay import webhook as alipay_webhook +from yuxi.channel.extensions.base import BaseChannelPlugin +from yuxi.channel.message.models import PeerKind +from yuxi.channel.protocols import SessionResolution +from yuxi.channel.plugins.registry import ChannelPluginRegistry + +_logger = logging.getLogger(__name__) + + +class AlipayPlugin(BaseChannelPlugin): + id = "alipay" + name = "支付宝" + order = 50 + label = "支付宝 (生活号+)" + aliases = ["alipay", "支付宝", "生活号"] + resolve_reply_to_mode = "off" + + def __init__(self): + super().__init__() + self._config_adapter = AlipayConfig() + self._gateway: AlipayGateway | None = None + self._outbound: AlipayOutbound | None = None + self._status: AlipayStatus | None = None + self._security = AlipaySecurity() + self._pairing = AlipayPairing() + self._deduplicator = AlipayMessageDeduplicator() + self._streaming = AlipayStreaming() + self._media: AlipayMedia | None = None + self._remove_markdown: bool = True + + @property + def capabilities(self) -> ChannelCapabilities: + return ChannelCapabilities( + chat_types=["direct"], + message_types=["text", "image"], + reactions=False, + typing_indicator=False, + threads=False, + edit=False, + unsend=True, + reply=False, + media=True, + effects=False, + native_commands=False, + polls=False, + group_management=False, + streaming=True, + streaming_mode="block", + block_streaming=True, + block_streaming_chunk_min_chars=50, + block_streaming_chunk_max_chars=500, + block_streaming_chunk_break_preference="double_newline_second_paragraph", + block_streaming_coalesce_idle_ms=1200, + tts=None, + ) + + def list_account_ids(self, config: dict | None = None) -> list[str]: + return self._config_adapter.list_account_ids(config) + + async def resolve_account(self, account_id: str = "default") -> dict: + account = self._config_adapter.resolve_account(account_id) + return { + "account_id": account.account_id, + "app_id": account.app_id, + "app_private_key": account.app_private_key, + "alipay_public_key": account.alipay_public_key, + "aes_key": account.aes_key, + "mode": account.mode.value, + "gateway_url": account.gateway_url, + "name": account.name, + "enabled": account.enabled, + "dm_policy": account.dm_policy.value, + "allow_from": account.allow_from, + "remove_markdown": account.remove_markdown, + "subscribe_msg": account.subscribe_msg, + } + + def is_configured(self, account: dict | None = None) -> bool: + return self._config_adapter.is_configured(account) + + def is_enabled(self, account: dict | None = None, config: dict | None = None) -> bool: + return self._config_adapter.is_enabled(account, config) + + def disabled_reason(self, account: dict | None = None, config: dict | None = None) -> str: + return self._config_adapter.disabled_reason(account, config) + + def unconfigured_reason(self, account: dict | None = None, config: dict | None = None) -> str: + return self._config_adapter.unconfigured_reason(account, config) + + def describe_account(self, account: dict | None = None, config: dict | None = None) -> dict: + return self._config_adapter.describe_account(account, config) + + def resolve_allow_from(self, config: dict | None = None, account_id: str | None = None) -> list[str] | None: + return self._config_adapter.resolve_allow_from(config or {}, account_id) + + def format_allow_from(self, config: dict, account_id: str | None, allow_from: list) -> list[str]: + return self._config_adapter.format_allow_from(config, account_id, allow_from) + + def has_configured_state(self, config: dict | None = None) -> bool: + return self._config_adapter.has_configured_state(config) + + def has_persisted_auth_state(self, config: dict | None = None) -> bool: + return self._config_adapter.has_persisted_auth_state(config) + + def default_account_id(self, config: dict | None = None) -> str: + return self._config_adapter.default_account_id(config) + + def config_schema(self) -> dict: + return self._config_adapter.config_schema() + + def collect_warnings( + self, config: dict | None = None, account_id: str | None = None, account: dict | None = None + ) -> list[str]: + alipay_account = self._config_adapter.resolve_account(account_id or "default") + return self._security.collect_warnings(config, account_id, alipay_account) + + async def start(self, ctx) -> object: + self._gateway = AlipayGateway() + self._outbound = AlipayOutbound(self._gateway) + self._status = AlipayStatus(self._gateway) + self._media = AlipayMedia(self._gateway) + + account = self._config_adapter.resolve_account() + self._remove_markdown = account.remove_markdown + alipay_webhook.init_webhook( + self._config_adapter, + self._deduplicator, + self._security, + self._pairing, + self._outbound, + ) + return {"running": True, "account_id": account.account_id, "mode": "webhook"} + + async def stop(self, ctx) -> None: + if self._gateway: + await self._gateway.close() + self._gateway = None + self._outbound = None + self._status = None + self._media = None + + async def on_config_changed(self, prev_cfg: dict, next_cfg: dict, account_id: str) -> None: + if prev_cfg != next_cfg: + _logger.info("Alipay config changed, reloading...") + ctx = ChannelContext(channel_type="alipay", account_id=account_id, config=next_cfg) + await self.stop(ctx) + await self.start(ctx) + + async def send_text( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + if not content or not self._outbound: + return + + account = self._config_adapter.resolve_account(account_id or "default") + if self._remove_markdown: + content = remove_markdown(content) + + await self._outbound.send_text( + target_id=target_id, + content=content, + account=account, + reply_to_id=reply_to_id, + thread_id=thread_id, + ) + + async def send_media( + self, + target_id: str, + media_url: str, + media_type: str, + reply_to_id: str | None = None, + thread_id: str | None = None, + ) -> None: + if not media_url or not self._outbound: + return + + account = self._config_adapter.resolve_account() + if self._remove_markdown: + media_url = remove_markdown(media_url) + await self._outbound.send_media( + target_id=target_id, + media_url=media_url, + media_type=media_type, + account=account, + reply_to_id=reply_to_id, + thread_id=thread_id, + ) + + async def probe(self, account: dict | None = None) -> bool: + if self._status: + alipay_account = self._config_adapter.resolve_account() + result = await self._status.probe(alipay_account) + return result.ok + return False + + def build_summary(self, snapshot: object | None = None) -> dict: + if self._status: + account = self._config_adapter.resolve_account() + return self._status.build_summary(account) + return {} + + async def check_ready(self, account_id: str | None = None) -> bool: + if self._status: + account = self._config_adapter.resolve_account(account_id or "default") + return await self._status.check_ready(account) + return False + + def resolve_dm_policy(self) -> dict: + account = self._config_adapter.resolve_account() + return self._security.resolve_dm_policy(account) + + async def check_allowlist(self, peer_id: str, channel_type: str) -> bool: + account = self._config_adapter.resolve_account() + return self._security.check_allowlist(peer_id, channel_type, account) + + async def generate_code(self, peer_id: str) -> str: + code = self._pairing.generate_code(peer_id) + return code or "" + + async def verify_code(self, peer_id: str, code: str) -> bool: + return self._pairing.verify_code(peer_id, code) + + async def send_template( + self, + target_id: str, + template_id: str, + context: dict, + url: str | None = None, + account_id: str | None = None, + ) -> None: + if not self._outbound: + return + account = self._config_adapter.resolve_account(account_id or "default") + await self._outbound.send_template( + account=account, + to_user_id=target_id, + template_id=template_id, + context=context, + url=url, + ) + + async def recall_message(self, msg_id: str, account_id: str | None = None) -> None: + if not self._outbound: + return + account = self._config_adapter.resolve_account(account_id or "default") + await self._outbound.recall_message(account=account, msg_id=msg_id) + + async def query_followers(self, next_token: str = "", account_id: str | None = None) -> dict: + if not self._outbound: + return {} + account = self._config_adapter.resolve_account(account_id or "default") + result = await self._outbound.query_followers(account=account, next_token=next_token) + return result.result or {} + + @property + def streaming_mode(self) -> str: + return self._streaming.streaming_mode + + @property + def preview_stream_throttle_ms(self) -> int: + return self._streaming.preview_stream_throttle_ms + + @property + def preview_min_initial_chars(self) -> int: + return self._streaming.preview_min_initial_chars + + @property + def block_streaming_enabled(self) -> bool: + return self._streaming.block_streaming_enabled + + @property + def block_streaming_break(self) -> str: + return self._streaming.block_streaming_break + + @property + def block_streaming_chunk_min_chars(self) -> int: + return self._streaming.block_streaming_chunk_min_chars + + @property + def block_streaming_chunk_max_chars(self) -> int: + return self._streaming.block_streaming_chunk_max_chars + + @property + def block_streaming_chunk_break_preference(self) -> str: + return self._streaming.block_streaming_chunk_break_preference + + @property + def block_streaming_coalesce_defaults(self) -> dict: + return self._streaming.block_streaming_coalesce_defaults + + def create_draft_stream_session(self, target_id: str) -> dict: + return self._streaming.create_draft_stream_session(target_id) + + def create_block_chunker(self) -> dict: + return self._streaming.create_block_chunker() + + def parse_explicit_target(self, content: str) -> str | None: + return None + + def resolve_session(self, msg): + if hasattr(msg, "sender") and hasattr(msg.sender, "kind"): + if msg.sender.kind == PeerKind.DIRECT: + return SessionResolution(kind="direct", conversation_id=msg.sender.id) + gid = msg.group.id if hasattr(msg, "group") and msg.group and msg.group.id else "unknown" + return SessionResolution(kind="group", conversation_id=gid) + + # ── ErrorHandlingProtocol ──────────────────────────── + + def classify_error(self, error: BaseException): + if isinstance(error, AlipayError): + code = error.code + sub_code = error.sub_code + error_code, _ = classify_alipay_error(code, sub_code) + from yuxi.channel.errors import ErrorSeverity, ClassifiedError + + severity_map = { + "sign_failed": ErrorSeverity.FATAL, + "verify_failed": ErrorSeverity.FATAL, + "auth_failed": ErrorSeverity.FORBIDDEN, + "rate_limited": ErrorSeverity.RATE_LIMITED, + "network_error": ErrorSeverity.NETWORK, + } + return ClassifiedError( + severity=severity_map.get(error_code, ErrorSeverity.RETRYABLE), + error_message=str(error), + ) + return super().classify_error(error) + + def is_retryable(self, error: BaseException) -> bool: + if isinstance(error, AlipayError): + return error.retryable + return super().is_retryable(error) + + # ── InboundHandlerProtocol ─────────────────────────── + + async def handle_raw_event(self, event: dict, account: dict) -> object | None: + from yuxi.channel.extensions.alipay.types import AlipayAccount as _AlipayAccount, AlipayInboundEvent + + alipay_account = _AlipayAccount( + account_id=account.get("account_id", "default"), + app_id=account.get("app_id", ""), + ) + inbound = AlipayInboundEvent( + msg_type=event.get("msg_type", event.get("MsgType", "")), + event_type=event.get("event_type", event.get("EventType")), + from_user_id=event.get("from_user_id", event.get("FromAlipayUserId", "")), + from_user_name=event.get("from_user_name", event.get("FromAlipayUserName")), + create_time=int(event.get("create_time", event.get("CreateTime", 0))), + msg_id=event.get("msg_id", event.get("MsgId", "")), + app_id=event.get("app_id", event.get("AppId", "")), + biz_content=event.get("biz_content", {}), + raw=event, + ) + return convert_alipay_event(inbound, alipay_account) + + def parse_to_unified(self, raw_event: dict, account_id: str) -> object | None: + from yuxi.channel.extensions.alipay.types import AlipayAccount as _AlipayAccount, AlipayInboundEvent + + account = _AlipayAccount(account_id=account_id) + inbound = AlipayInboundEvent( + msg_type=raw_event.get("msg_type", raw_event.get("MsgType", "")), + event_type=raw_event.get("event_type", raw_event.get("EventType")), + from_user_id=raw_event.get("from_user_id", raw_event.get("FromAlipayUserId", "")), + from_user_name=raw_event.get("from_user_name", raw_event.get("FromAlipayUserName")), + create_time=int(raw_event.get("create_time", raw_event.get("CreateTime", 0))), + msg_id=raw_event.get("msg_id", raw_event.get("MsgId", "")), + app_id=raw_event.get("app_id", raw_event.get("AppId", "")), + biz_content=raw_event.get("biz_content", {}), + raw=raw_event, + ) + return convert_alipay_event(inbound, account) + + # ── DedupeProtocol ─────────────────────────────────── + + def is_duplicate(self, key: str) -> bool: + return self._deduplicator.is_duplicate(key) + + def mark_seen(self, key: str) -> None: + pass + + @property + def ttl_seconds(self) -> int: + from yuxi.channel.extensions.alipay.constants import ALIPAY_DEDUPE_TTL_SECONDS + + return ALIPAY_DEDUPE_TTL_SECONDS + + @property + def max_entries(self) -> int: + from yuxi.channel.extensions.alipay.constants import ALIPAY_DEDUPE_MAX_SIZE + + return ALIPAY_DEDUPE_MAX_SIZE + + # ── FormatProtocol ─────────────────────────────────── + + def sanitize_text(self, text: str, payload: object | None = None) -> str: + if self._remove_markdown: + return remove_markdown(text) + return text + + # ── AgentPromptProtocol ────────────────────────────── + + def build_system_prompt(self, context) -> str | None: + return ( + "当前渠道为支付宝生活号+，仅支持文本和图片消息。\n" + "支付宝消息格式为纯文本（不支持 Markdown），回复内容会自动去除 Markdown 符号。\n" + "每条文本消息长度不超过 2048 字符，每天向每个用户最多发送 100 条消息。\n" + ) + + def build_context_note(self, context) -> str: + return "请用简洁的纯文本回复用户，避免使用 Markdown 格式。" + + @property + def channel_format_instructions(self) -> str: + return "支付宝生活号+ 纯文本渠道：不使用 Markdown 格式、不使用代码块、简洁直接回复。文本上限 2048 字符。" + + +alipay_plugin = ChannelPluginRegistry.register(AlipayPlugin()) diff --git a/backend/package/yuxi/channel/extensions/alipay/config.py b/backend/package/yuxi/channel/extensions/alipay/config.py new file mode 100644 index 00000000..63ecb2fe --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/config.py @@ -0,0 +1,169 @@ +import os + +from yuxi.channel.extensions.alipay.types import AlipayAccount, AlipayDmPolicy, AlipayMode + + +class AlipayConfig: + ENV_MAP = { + "app_id": "ALIPAY_APP_ID", + "app_private_key": "ALIPAY_PRIVATE_KEY", + "alipay_public_key": "ALIPAY_PUBLIC_KEY", + "aes_key": "ALIPAY_AES_KEY", + "mode": "ALIPAY_MODE", + "dm_policy": "ALIPAY_DM_POLICY", + "subscribe_msg": "ALIPAY_SUBSCRIBE_MSG", + "remove_markdown": "ALIPAY_REMOVE_MD", + } + + def list_account_ids(self, config: dict | None = None) -> list[str]: + if self._resolve_app_id(): + return ["default"] + return [] + + def resolve_account(self, account_id: str = "default") -> AlipayAccount: + app_id = self._resolve_app_id() + return AlipayAccount( + account_id=account_id, + app_id=app_id or "", + app_private_key=self._env_or_config("app_private_key") or "", + alipay_public_key=self._env_or_config("alipay_public_key") or "", + aes_key=self._env_or_config("aes_key"), + mode=AlipayMode(self._env_or_config("mode") or "production"), + gateway_url=( + "https://openapi.alipaydev.com/gateway.do" + if self._env_or_config("mode") == "sandbox" + else "https://openapi.alipay.com/gateway.do" + ), + name=self._env_or_config("name") or "", + enabled=self._env_or_config("enabled") != "false", + dm_policy=AlipayDmPolicy(self._env_or_config("dm_policy") or "open"), + allow_from=[], + remove_markdown=self._env_or_config("remove_markdown") != "false", + subscribe_msg=self._env_or_config("subscribe_msg") or "", + ) + + def is_configured(self, account: dict | None = None) -> bool: + return bool( + self._resolve_app_id() + and self._env_or_config("app_private_key") + and self._env_or_config("alipay_public_key") + ) + + def _resolve_app_id(self) -> str | None: + return self._env_or_config("app_id") + + @staticmethod + def _env_or_config(key: str) -> str | None: + env_key = AlipayConfig.ENV_MAP.get(key, "") + if env_key: + val = os.getenv(env_key) + if val: + return val + return None + + def config_schema(self) -> dict: + return { + "$schema": "https://json-schema.org/draft-07/schema#", + "type": "object", + "title": "支付宝渠道配置", + "properties": { + "app_id": { + "type": "string", + "title": "AppID", + "description": "支付宝开放平台应用 AppID", + }, + "app_private_key": { + "type": "string", + "title": "应用私钥", + "x-ui-password": True, + "description": "RSA2 应用私钥 (PEM 格式)", + }, + "alipay_public_key": { + "type": "string", + "title": "支付宝公钥", + "description": "支付宝公钥 (PEM 格式)", + }, + "aes_key": { + "type": "string", + "title": "AES 解密密钥", + "x-ui-password": True, + "description": "AES 密钥（可选，16 字符）", + }, + "mode": { + "type": "string", + "enum": ["production", "sandbox"], + "default": "production", + "title": "运行模式", + }, + "dm_policy": { + "type": "string", + "enum": ["open", "pairing", "allowlist", "disabled"], + "default": "open", + "title": "DM 策略", + }, + "subscribe_msg": { + "type": "string", + "title": "关注欢迎语", + "description": "用户关注生活号后自动回复的文本", + }, + "remove_markdown": { + "type": "boolean", + "default": True, + "title": "移除 Markdown 格式", + "description": "自动移除 AI 回复中的 Markdown 符号", + }, + }, + "required": ["app_id", "app_private_key", "alipay_public_key"], + } + + def is_enabled(self, account: dict | None = None, config: dict | None = None) -> bool: + if account is None: + return self._env_or_config("enabled") != "false" + return account.get("enabled", True) + + def disabled_reason(self, account: dict | None = None, config: dict | None = None) -> str: + if not account: + return "未配置账户" + if not account.get("enabled", True): + return "账户已被禁用" + return "" + + def unconfigured_reason(self, account: dict | None = None, config: dict | None = None) -> str: + if not account: + return "未配置账户" + missing = [] + if not account.get("app_id"): + missing.append("AppId") + if not account.get("app_private_key"): + missing.append("应用私钥") + if not account.get("alipay_public_key"): + missing.append("支付宝公钥") + return f"缺少: {', '.join(missing)}" if missing else "" + + def describe_account(self, account: dict | None = None, config: dict | None = None) -> dict: + if not account: + return {"account_id": ""} + return { + "account_id": account.get("account_id", "default"), + "app_id": account.get("app_id", ""), + "name": account.get("name", ""), + "mode": account.get("mode", "production"), + "dm_policy": account.get("dm_policy", "open"), + "enabled": account.get("enabled", True), + } + + def resolve_allow_from(self, config: dict, account_id: str | None = None) -> list[str] | None: + account = self.resolve_account(account_id or "default") + return account.allow_from or None + + def format_allow_from(self, config: dict, account_id: str | None, allow_from: list) -> list[str]: + return [str(item) for item in allow_from] + + def has_configured_state(self, config: dict | None = None) -> bool: + return self.is_configured() + + def has_persisted_auth_state(self, config: dict | None = None) -> bool: + return bool(self._resolve_app_id() and self._env_or_config("app_private_key")) + + def default_account_id(self, config: dict | None = None) -> str: + return "default" diff --git a/backend/package/yuxi/channel/extensions/alipay/constants.py b/backend/package/yuxi/channel/extensions/alipay/constants.py new file mode 100644 index 00000000..8304cd30 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/constants.py @@ -0,0 +1,36 @@ +ALIPAY_TEXT_LIMIT = 2048 +ALIPAY_IMAGE_TEXT_ARTICLE_LIMIT = 6 +ALIPAY_DAILY_MSG_LIMIT_PER_USER = 100 +ALIPAY_MSG_RATE_LIMIT_PER_SECOND = 1 +ALIPAY_INTERACTION_WINDOW_HOURS = 48 +ALIPAY_WEBHOOK_RESPONSE_TIMEOUT_SECONDS = 5 +ALIPAY_RETRY_MAX_INTERVALS = [15, 30, 60, 120, 180] + +ALIPAY_GATEWAY_PRODUCTION = "https://openapi.alipay.com/gateway.do" +ALIPAY_GATEWAY_SANDBOX = "https://openapi.alipaydev.com/gateway.do" +ALIPAY_OAUTH_AUTHORIZE_PRODUCTION = "https://openauth.alipay.com/oauth2/publicAppAuthorize.htm" +ALIPAY_OAUTH_AUTHORIZE_SANDBOX = "https://openapi.alipaydev.com/oauth2/publicAppAuthorize.htm" + +ALIPAY_SIGN_TYPE = "RSA2" +ALIPAY_FORMAT = "JSON" +ALIPAY_CHARSET = "utf-8" +ALIPAY_VERSION = "1.0" + +ALIPAY_SUCCESS_CODE = "10000" +ALIPAY_APP_ID_PREFIX = "2021" + +ALIPAY_APP_ID_PATTERN = r"^\d{16,32}$" +ALIPAY_USER_ID_PREFIX = "2088" + +ALIPAY_DEDUPE_TTL_SECONDS = 300 +ALIPAY_DEDUPE_MAX_SIZE = 10000 + +ALIPAY_RETRY_MAX_ATTEMPTS = 3 +ALIPAY_RETRY_BASE_DELAY = 1.0 +ALIPAY_RETRY_MAX_DELAY = 30.0 + +ALIPAY_PAIRING_CODE_TTL_SECONDS = 300 +ALIPAY_PAIRING_CODE_LENGTH = 6 + +ALIPAY_BLOCK_STREAMING_CHUNK_MIN_CHARS = 50 +ALIPAY_BLOCK_STREAMING_CHUNK_MAX_CHARS = 500 diff --git a/backend/package/yuxi/channel/extensions/alipay/crypto.py b/backend/package/yuxi/channel/extensions/alipay/crypto.py new file mode 100644 index 00000000..fe5ad5b3 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/crypto.py @@ -0,0 +1,82 @@ +import base64 +from urllib.parse import quote + +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import padding +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + + +class AlipayAESCrypto: + def __init__(self, aes_key: str): + raw = base64.b64decode(aes_key + "=" * (4 - len(aes_key) % 4)) + self._key = raw[:16] + + def decrypt(self, encrypted: str | bytes) -> str: + from cryptography.hazmat.primitives import padding as sym_padding + + data = base64.b64decode(encrypted) + iv = self._key + cipher = Cipher(algorithms.AES(self._key), modes.CBC(iv)) + decryptor = cipher.decryptor() + padded = decryptor.update(data) + decryptor.finalize() + unpadder = sym_padding.PKCS7(128).unpadder() + plaintext = unpadder.update(padded) + unpadder.finalize() + return plaintext.decode("utf-8") + + def encrypt(self, plaintext: str) -> str: + from cryptography.hazmat.primitives import padding as sym_padding + + iv = self._key + padder = sym_padding.PKCS7(128).padder() + padded = padder.update(plaintext.encode("utf-8")) + padder.finalize() + cipher = Cipher(algorithms.AES(self._key), modes.CBC(iv)) + encryptor = cipher.encryptor() + encrypted = encryptor.update(padded) + encryptor.finalize() + return base64.b64encode(encrypted).decode("utf-8") + + +class AlipayCrypto: + def __init__(self, app_private_key_pem: str, alipay_public_key_pem: str): + self._private_key = serialization.load_pem_private_key( + app_private_key_pem.encode("utf-8"), + password=None, + backend=default_backend(), + ) + self._public_key = serialization.load_pem_public_key( + alipay_public_key_pem.encode("utf-8"), + backend=default_backend(), + ) + + def sign(self, params: dict) -> str: + content = self._build_sign_string(params) + signature = self._private_key.sign( + content.encode("utf-8"), + padding.PKCS1v15(), + hashes.SHA256(), + ) + return base64.b64encode(signature).decode("utf-8") + + def verify(self, params: dict, signature: str) -> bool: + content = self._build_sign_string(params) + try: + self._public_key.verify( + base64.b64decode(signature), + content.encode("utf-8"), + padding.PKCS1v15(), + hashes.SHA256(), + ) + return True + except Exception: + return False + + @staticmethod + def _build_sign_string(params: dict) -> str: + filtered = {k: v for k, v in params.items() if k not in ("sign", "sign_type") and v is not None} + sorted_items = sorted(filtered.items(), key=lambda x: x[0]) + parts = [] + for k, v in sorted_items: + encoded_key = quote(str(k), safe="~") + encoded_value = quote(str(v), safe="~") + parts.append(f"{encoded_key}={encoded_value}") + return "&".join(parts) diff --git a/backend/package/yuxi/channel/extensions/alipay/dedup.py b/backend/package/yuxi/channel/extensions/alipay/dedup.py new file mode 100644 index 00000000..4bbf1278 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/dedup.py @@ -0,0 +1,28 @@ +import time + +from yuxi.channel.extensions.alipay.constants import ( + ALIPAY_DEDUPE_MAX_SIZE, + ALIPAY_DEDUPE_TTL_SECONDS, +) + + +class AlipayMessageDeduplicator: + def __init__(self): + self._cache: dict[str, float] = {} + + def is_duplicate(self, msg_id: str) -> bool: + if not msg_id: + return False + now = time.time() + if msg_id in self._cache: + if now - self._cache[msg_id] < ALIPAY_DEDUPE_TTL_SECONDS: + return True + self._cache[msg_id] = now + self._evict_expired(now) + return False + + def _evict_expired(self, now: float): + if len(self._cache) > ALIPAY_DEDUPE_MAX_SIZE: + expired = [k for k, v in self._cache.items() if now - v > ALIPAY_DEDUPE_TTL_SECONDS] + for k in expired: + del self._cache[k] diff --git a/backend/package/yuxi/channel/extensions/alipay/errors.py b/backend/package/yuxi/channel/extensions/alipay/errors.py new file mode 100644 index 00000000..342ebc7f --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/errors.py @@ -0,0 +1,65 @@ +from enum import StrEnum + + +class AlipayErrorCode(StrEnum): + SIGN_FAILED = "sign_failed" + VERIFY_FAILED = "verify_failed" + AUTH_FAILED = "auth_failed" + RATE_LIMITED = "rate_limited" + OUT_OF_WINDOW = "out_of_window" + DAILY_LIMIT_EXCEEDED = "daily_limit_exceeded" + USER_NOT_FOLLOWED = "user_not_followed" + MSG_TOO_LONG = "msg_too_long" + NETWORK_ERROR = "network_error" + BUSINESS_ERROR = "business_error" + UNKNOWN = "unknown" + + +class AlipayError(Exception): + def __init__( + self, + code: AlipayErrorCode, + message: str = "", + sub_code: str = "", + sub_msg: str = "", + retryable: bool = False, + retry_after: float | None = None, + ): + self.code = code + self.message = message + self.sub_code = sub_code + self.sub_msg = sub_msg + self.retryable = retryable + self.retry_after = retry_after + super().__init__(f"[{code}] {message}" + (f" | {sub_code}: {sub_msg}" if sub_code else "")) + + +_ERROR_CLASSIFICATION: dict[str, tuple[AlipayErrorCode, bool]] = { + "40001": (AlipayErrorCode.SIGN_FAILED, False), + "40002": (AlipayErrorCode.VERIFY_FAILED, False), + "40004": (AlipayErrorCode.BUSINESS_ERROR, False), + "40006": (AlipayErrorCode.AUTH_FAILED, False), + "20000": (AlipayErrorCode.NETWORK_ERROR, True), + "20001": (AlipayErrorCode.AUTH_FAILED, False), +} + +_SUB_CODE_CLASSIFICATION: dict[str, tuple[AlipayErrorCode, bool]] = { + "isv.invalid-signature": (AlipayErrorCode.SIGN_FAILED, False), + "isv.invalid-app-id": (AlipayErrorCode.VERIFY_FAILED, False), + "isv.message-count-per-day-isv-error": (AlipayErrorCode.DAILY_LIMIT_EXCEEDED, False), + "isv.invalid-auth-token": (AlipayErrorCode.AUTH_FAILED, False), + "isv.business-fail": (AlipayErrorCode.USER_NOT_FOLLOWED, False), +} + + +def classify_alipay_error(code: str, sub_code: str = "") -> tuple[AlipayErrorCode, bool]: + if sub_code and sub_code in _SUB_CODE_CLASSIFICATION: + return _SUB_CODE_CLASSIFICATION[sub_code] + if code in _ERROR_CLASSIFICATION: + return _ERROR_CLASSIFICATION[code] + return (AlipayErrorCode.UNKNOWN, False) + + +def is_retryable_error(code: str, sub_code: str = "") -> bool: + _, retryable = classify_alipay_error(code, sub_code) + return retryable diff --git a/backend/package/yuxi/channel/extensions/alipay/format.py b/backend/package/yuxi/channel/extensions/alipay/format.py new file mode 100644 index 00000000..ab7616d8 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/format.py @@ -0,0 +1,34 @@ +import re + + +def remove_markdown(text: str) -> str: + text = re.sub(r"\*\*(.+?)\*\*", r"\1", text) + text = re.sub(r"\*(.+?)\*", r"\1", text) + text = re.sub(r"__(.+?)__", r"\1", text) + text = re.sub(r"_(.+?)_", r"\1", text) + text = re.sub(r"~~(.+?)~~", r"\1", text) + text = re.sub(r"```[\s\S]*?```", "", text) + text = re.sub(r"`{1,3}[^`]*`{1,3}", "", text) + text = re.sub(r"\[([^\]]*)\]\([^)]*\)", r"\1", text) + text = re.sub(r"!\[.*?\]\(.*?\)", "[图片]", text) + text = re.sub(r"^#+ (.*?)$", r"\1", text, flags=re.MULTILINE) + text = re.sub(r"^[*-] (.*?)$", r"· \1", text, flags=re.MULTILINE) + text = re.sub(r"^> (.*?)$", r" \1", text, flags=re.MULTILINE) + text = re.sub(r"\n{3,}", "\n\n", text) + return text.strip() + + +def split_utf8_safe(text: str, max_len: int) -> list[str]: + if len(text) <= max_len: + return [text] + chunks = [] + while text: + if len(text) <= max_len: + chunks.append(text) + break + split_at = text.rfind("\n", 0, max_len) + if split_at == -1 or split_at < max_len // 2: + split_at = max_len + chunks.append(text[:split_at]) + text = text[split_at:].lstrip() + return chunks diff --git a/backend/package/yuxi/channel/extensions/alipay/gateway.py b/backend/package/yuxi/channel/extensions/alipay/gateway.py new file mode 100644 index 00000000..d32c3541 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/gateway.py @@ -0,0 +1,124 @@ +import asyncio +import json +import logging +from datetime import datetime + +import httpx + +from yuxi.channel.extensions.alipay.constants import ( + ALIPAY_CHARSET, + ALIPAY_FORMAT, + ALIPAY_RETRY_BASE_DELAY, + ALIPAY_RETRY_MAX_ATTEMPTS, + ALIPAY_SIGN_TYPE, + ALIPAY_SUCCESS_CODE, + ALIPAY_VERSION, +) +from yuxi.channel.extensions.alipay.crypto import AlipayCrypto +from yuxi.channel.extensions.alipay.errors import ( + AlipayError, + AlipayErrorCode, + classify_alipay_error, +) +from yuxi.channel.extensions.alipay.types import AlipayAccount + +logger = logging.getLogger(__name__) + + +class AlipayGateway: + def __init__(self, crypto: AlipayCrypto | None = None): + self._crypto = crypto + self._clients: dict[str, httpx.AsyncClient] = {} + self._lock = asyncio.Lock() + + async def _get_client(self, account: AlipayAccount) -> httpx.AsyncClient: + if account.account_id not in self._clients: + self._clients[account.account_id] = httpx.AsyncClient( + base_url=account.gateway_url, + timeout=httpx.Timeout(10.0), + limits=httpx.Limits(max_connections=20, max_keepalive_connections=10), + ) + return self._clients[account.account_id] + + async def close(self): + for client in self._clients.values(): + await client.aclose() + self._clients.clear() + + async def request( + self, + account: AlipayAccount, + method: str, + biz_content: dict | None = None, + ) -> dict: + crypto = self._crypto or AlipayCrypto(account.app_private_key, account.alipay_public_key) + params = { + "app_id": account.app_id, + "method": method, + "format": ALIPAY_FORMAT, + "charset": ALIPAY_CHARSET, + "sign_type": ALIPAY_SIGN_TYPE, + "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + "version": ALIPAY_VERSION, + } + if biz_content: + params["biz_content"] = json.dumps(biz_content, ensure_ascii=False) + + params["sign"] = crypto.sign(params) + client = await self._get_client(account) + + last_error = None + for attempt in range(ALIPAY_RETRY_MAX_ATTEMPTS): + try: + resp = await client.post("", data=params) + resp.raise_for_status() + data = resp.json() + + response_key = next((k for k in data if k.endswith("_response")), None) + if not response_key: + raise AlipayError( + code=AlipayErrorCode.UNKNOWN, + message="无法识别的响应格式", + ) + + body = data[response_key] + code = body.get("code", "") + + if code != ALIPAY_SUCCESS_CODE: + sub_code = body.get("sub_code", "") + error_code, retryable = classify_alipay_error(code, sub_code) + err = AlipayError( + code=error_code, + message=body.get("msg", ""), + sub_code=sub_code, + sub_msg=body.get("sub_msg", ""), + retryable=retryable, + ) + if retryable and attempt < ALIPAY_RETRY_MAX_ATTEMPTS - 1: + last_error = err + delay = ALIPAY_RETRY_BASE_DELAY * (2**attempt) + await asyncio.sleep(delay) + continue + raise err + + return body + + except (httpx.ConnectError, httpx.TimeoutException, httpx.NetworkError) as e: + if attempt < ALIPAY_RETRY_MAX_ATTEMPTS - 1: + last_error = AlipayError( + code=AlipayErrorCode.NETWORK_ERROR, + message=str(e), + retryable=True, + ) + delay = ALIPAY_RETRY_BASE_DELAY * (2**attempt) + await asyncio.sleep(delay) + continue + raise AlipayError( + code=AlipayErrorCode.NETWORK_ERROR, + message=str(e), + ) + + raise last_error or AlipayError( + code=AlipayErrorCode.UNKNOWN, + message="请求失败，已达最大重试次数", + ) diff --git a/backend/package/yuxi/channel/extensions/alipay/media.py b/backend/package/yuxi/channel/extensions/alipay/media.py new file mode 100644 index 00000000..005dde74 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/media.py @@ -0,0 +1,42 @@ +import json +import logging +from datetime import datetime + +import httpx + +from yuxi.channel.extensions.alipay.types import AlipayAccount + +logger = logging.getLogger(__name__) + +ALIPAY_FILE_GATEWAY_PRODUCTION = "https://openfile.alipay.com/chat/multimedia.do" +ALIPAY_FILE_GATEWAY_SANDBOX = "https://openfile.alipaydev.com/chat/multimedia.do" + + +class AlipayMedia: + def __init__(self, gateway=None): + self._gateway = gateway + + async def download(self, account: AlipayAccount, media_id: str) -> bytes | None: + gateway_url = ( + ALIPAY_FILE_GATEWAY_SANDBOX if "sandbox" in account.gateway_url else ALIPAY_FILE_GATEWAY_PRODUCTION + ) + try: + async with httpx.AsyncClient(timeout=httpx.Timeout(15.0)) as client: + resp = await client.get( + gateway_url, + params={ + "app_id": account.app_id, + "method": "alipay.mobile.public.multimedia.download", + "format": "JSON", + "charset": "utf-8", + "sign_type": "RSA2", + "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + "version": "1.0", + "biz_content": json.dumps({"media_id": media_id}), + }, + ) + resp.raise_for_status() + return resp.content + except Exception: + logger.exception("Failed to download media %s", media_id) + return None diff --git a/backend/package/yuxi/channel/extensions/alipay/monitor.py b/backend/package/yuxi/channel/extensions/alipay/monitor.py new file mode 100644 index 00000000..eda0778b --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/monitor.py @@ -0,0 +1,97 @@ +from datetime import datetime, UTC + +from yuxi.channel.extensions.alipay.types import AlipayAccount, AlipayInboundEvent +from yuxi.channel.message.models import MessageType, PeerInfo, PeerKind, UnifiedMessage + + +def convert_alipay_event(event: AlipayInboundEvent, account: AlipayAccount) -> UnifiedMessage | None: + msg_type = _map_msg_type(event) + + if event.msg_type == "event": + if event.event_type in ("follow", "enter"): + return _build_event_message(event, account, msg_type) + return None + + return UnifiedMessage( + msg_id=event.msg_id, + channel_type="alipay", + account_id=account.account_id, + content=_extract_content(event), + sender=PeerInfo( + kind=PeerKind.DIRECT, + id=event.from_user_id, + display_name=event.from_user_name or event.from_user_id, + ), + message_type=msg_type, + media_urls=_extract_media_urls(event), + timestamp=datetime.fromtimestamp(event.create_time / 1000, tz=UTC), + raw_payload=event.raw, + metadata={ + "FromAlipayUserId": event.from_user_id, + "AppId": event.app_id, + "MsgType": event.msg_type, + "EventType": event.event_type or "", + }, + body_for_agent=_extract_content(event), + ) + + +def _map_msg_type(event: AlipayInboundEvent) -> MessageType: + if event.msg_type == "text": + return MessageType.TEXT + if event.msg_type == "image": + return MessageType.IMAGE + return MessageType.EVENT + + +def _extract_content(event: AlipayInboundEvent) -> str: + if event.msg_type == "text": + return (event.biz_content or {}).get("content", "") + if event.msg_type == "image": + return "[图片消息]" + if event.msg_type == "event": + event_names = { + "follow": "关注了生活号", + "unfollow": "取消关注", + "enter": "进入了生活号", + "click": f"点击了菜单: {(event.biz_content or {}).get('action_name', '')}", + } + return event_names.get(event.event_type or "", f"事件: {event.event_type}") + return "" + + +def _extract_media_urls(event: AlipayInboundEvent) -> list[str]: + if event.msg_type == "image": + biz = event.biz_content or {} + urls = [] + if biz.get("pic_url"): + urls.append(biz["pic_url"]) + if biz.get("media_id"): + urls.append(f"alipay://media/{biz['media_id']}") + return urls + return [] + + +def _build_event_message(event: AlipayInboundEvent, account: AlipayAccount, msg_type: MessageType) -> UnifiedMessage: + return UnifiedMessage( + msg_id=event.msg_id, + channel_type="alipay", + account_id=account.account_id, + content=_extract_content(event), + sender=PeerInfo( + kind=PeerKind.DIRECT, + id=event.from_user_id, + display_name=event.from_user_name or event.from_user_id, + ), + message_type=msg_type, + media_urls=[], + timestamp=datetime.fromtimestamp(event.create_time / 1000, tz=UTC), + raw_payload=event.raw, + metadata={ + "FromAlipayUserId": event.from_user_id, + "AppId": event.app_id, + "MsgType": event.msg_type, + "EventType": event.event_type or "", + }, + body_for_agent=_extract_content(event), + ) diff --git a/backend/package/yuxi/channel/extensions/alipay/outbound.py b/backend/package/yuxi/channel/extensions/alipay/outbound.py new file mode 100644 index 00000000..8816dfea --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/outbound.py @@ -0,0 +1,314 @@ +import logging +import time +from datetime import date + +from yuxi.channel.extensions.alipay.constants import ( + ALIPAY_DAILY_MSG_LIMIT_PER_USER, + ALIPAY_IMAGE_TEXT_ARTICLE_LIMIT, + ALIPAY_INTERACTION_WINDOW_HOURS, + ALIPAY_TEXT_LIMIT, +) +from yuxi.channel.extensions.alipay.errors import AlipayError, AlipayErrorCode +from yuxi.channel.extensions.alipay.format import split_utf8_safe +from yuxi.channel.extensions.alipay.gateway import AlipayGateway +from yuxi.channel.extensions.alipay.types import AlipayAccount, AlipayOutboundResult + +logger = logging.getLogger(__name__) + + +class AlipayOutbound: + delivery_mode = "direct" + chunker_mode = "length" + text_chunk_limit = ALIPAY_TEXT_LIMIT + poll_max_options = None + supports_poll_duration_seconds = False + supports_anonymous_polls = False + extract_markdown_images = True + presentation_capabilities = None + delivery_capabilities = None + + def __init__(self, gateway: AlipayGateway): + self._gateway = gateway + self._daily_counters: dict[str, int] = {} + self._last_interaction: dict[str, float] = {} + + def _get_counter_key(self, account_id: str, user_id: str) -> str: + today = date.today().isoformat() + return f"{account_id}:{user_id}:{today}" + + def _check_daily_limit(self, user_id: str, account_id: str) -> bool: + key = self._get_counter_key(account_id, user_id) + count = self._daily_counters.get(key, 0) + return count < ALIPAY_DAILY_MSG_LIMIT_PER_USER + + def _increment_daily_counter(self, user_id: str, account_id: str): + key = self._get_counter_key(account_id, user_id) + self._daily_counters[key] = self._daily_counters.get(key, 0) + 1 + today = date.today().isoformat() + stale = [k for k in self._daily_counters if not k.endswith(f":{today}")] + for k in stale: + del self._daily_counters[k] + + def record_interaction(self, user_id: str, account_id: str): + key = f"{account_id}:{user_id}" + self._last_interaction[key] = time.time() + + def _check_window(self, user_id: str, account_id: str) -> bool: + key = f"{account_id}:{user_id}" + last = self._last_interaction.get(key, 0) + return (time.time() - last) < ALIPAY_INTERACTION_WINDOW_HOURS * 3600 + + async def send_text( + self, + target_id: str, + content: str, + account: AlipayAccount, + reply_to_id: str | None = None, + thread_id: str | None = None, + ) -> AlipayOutboundResult: + if not account.is_configured(): + raise AlipayError(code=AlipayErrorCode.UNKNOWN, message="缺少账户配置") + + if not self._check_window(target_id, account.account_id): + return AlipayOutboundResult( + msg_id=None, + success=False, + error="超过 48 小时交互窗口，无法发送消息", + ) + + if not self._check_daily_limit(target_id, account.account_id): + return AlipayOutboundResult( + msg_id=None, + success=False, + error="单用户每日消息数已达上限（100条）", + ) + + if len(content) > ALIPAY_TEXT_LIMIT: + content = content[:ALIPAY_TEXT_LIMIT] + + try: + result = await self._gateway.request( + account=account, + method="alipay.open.public.message.custom.send", + biz_content={ + "to_user_id": target_id, + "msg_type": "text", + "text": {"content": content}, + }, + ) + self._increment_daily_counter(target_id, account.account_id) + return AlipayOutboundResult( + msg_id=result.get("msg_id"), + success=True, + result=result, + ) + except AlipayError as e: + return AlipayOutboundResult( + msg_id=None, + success=False, + error=str(e), + ) + + async def send_template( + self, + account: AlipayAccount, + to_user_id: str, + template_id: str, + context: dict, + url: str | None = None, + ) -> AlipayOutboundResult: + if not account.is_configured(): + raise AlipayError(code=AlipayErrorCode.UNKNOWN, message="缺少账户配置") + + biz_content: dict = { + "to_user_id": to_user_id, + "template": { + "template_id": template_id, + "context": context, + }, + } + if url: + biz_content["url"] = url + + try: + result = await self._gateway.request( + account=account, + method="alipay.open.public.message.single.send", + biz_content=biz_content, + ) + return AlipayOutboundResult( + msg_id=result.get("msg_id"), + success=True, + result=result, + ) + except AlipayError as e: + return AlipayOutboundResult( + msg_id=None, + success=False, + error=str(e), + ) + + async def recall_message( + self, + account: AlipayAccount, + msg_id: str, + ) -> AlipayOutboundResult: + if not account.is_configured(): + raise AlipayError(code=AlipayErrorCode.UNKNOWN, message="缺少账户配置") + + try: + result = await self._gateway.request( + account=account, + method="alipay.open.public.life.msg.recall", + biz_content={"msg_id": msg_id}, + ) + return AlipayOutboundResult( + msg_id=msg_id, + success=True, + result=result, + ) + except AlipayError as e: + return AlipayOutboundResult( + msg_id=None, + success=False, + error=str(e), + ) + + async def query_followers( + self, + account: AlipayAccount, + next_token: str = "", + ) -> AlipayOutboundResult: + if not account.is_configured(): + raise AlipayError(code=AlipayErrorCode.UNKNOWN, message="缺少账户配置") + + biz_content: dict = {} + if next_token: + biz_content["next_token"] = next_token + + try: + result = await self._gateway.request( + account=account, + method="alipay.open.public.follow.batchquery", + biz_content=biz_content, + ) + return AlipayOutboundResult( + msg_id=None, + success=True, + result=result, + ) + except AlipayError as e: + return AlipayOutboundResult( + msg_id=None, + success=False, + error=str(e), + ) + + def sanitize_text(self, text: str, payload: dict | None = None) -> str: + return text + + def should_skip_plain_text_sanitization(self, payload: dict | None = None) -> bool: + return False + + def normalize_payload(self, payload: dict, config: dict, account_id: str | None = None) -> dict: + return payload + + def resolve_effective_text_chunk_limit( + self, config: dict, account_id: str | None = None, fallback_limit: int | None = None + ) -> int: + return fallback_limit or ALIPAY_TEXT_LIMIT + + def chunker(self, text: str, limit: int, ctx=None) -> list[str]: + return split_utf8_safe(text, limit) + + async def send_payload(self, ctx) -> None: + return None + + async def send_poll(self, ctx) -> None: + return None + + async def send_media( + self, + target_id: str, + media_url: str, + media_type: str, + account: AlipayAccount, + reply_to_id: str | None = None, + thread_id: str | None = None, + ) -> AlipayOutboundResult: + if not account.is_configured(): + raise AlipayError(code=AlipayErrorCode.UNKNOWN, message="缺少账户配置") + + if not self._check_window(target_id, account.account_id): + return AlipayOutboundResult( + msg_id=None, + success=False, + error="超过 48 小时交互窗口，无法发送消息", + ) + + try: + result = await self._gateway.request( + account=account, + method="alipay.open.public.message.custom.send", + biz_content={ + "to_user_id": target_id, + "msg_type": "image-text", + "articles": [ + { + "title": "", + "desc": "", + "image_url": media_url, + "url": media_url, + } + ], + }, + ) + return AlipayOutboundResult( + msg_id=result.get("msg_id"), + success=True, + result=result, + ) + except AlipayError as e: + return AlipayOutboundResult( + msg_id=None, + success=False, + error=str(e), + ) + + async def send_image_text( + self, + account: AlipayAccount, + to_user_id: str, + articles: list[dict], + ) -> AlipayOutboundResult: + if not self._check_window(to_user_id, account.account_id): + return AlipayOutboundResult( + msg_id=None, + success=False, + error="超过 48 小时交互窗口，无法发送消息", + ) + + if len(articles) > ALIPAY_IMAGE_TEXT_ARTICLE_LIMIT: + articles = articles[:ALIPAY_IMAGE_TEXT_ARTICLE_LIMIT] + + try: + result = await self._gateway.request( + account=account, + method="alipay.open.public.message.custom.send", + biz_content={ + "to_user_id": to_user_id, + "msg_type": "image-text", + "articles": articles, + }, + ) + return AlipayOutboundResult( + msg_id=result.get("msg_id"), + success=True, + result=result, + ) + except AlipayError as e: + return AlipayOutboundResult( + msg_id=None, + success=False, + error=str(e), + ) diff --git a/backend/package/yuxi/channel/extensions/alipay/pairing.py b/backend/package/yuxi/channel/extensions/alipay/pairing.py new file mode 100644 index 00000000..220aafd7 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/pairing.py @@ -0,0 +1,32 @@ +import secrets +import time + +from yuxi.channel.extensions.alipay.constants import ( + ALIPAY_PAIRING_CODE_LENGTH, + ALIPAY_PAIRING_CODE_TTL_SECONDS, +) + + +class AlipayPairing: + def __init__(self): + self._codes: dict[str, tuple[str, float]] = {} + + def generate_code(self, peer_id: str) -> str: + import random + + code = "".join(str(random.randint(0, 9)) for _ in range(ALIPAY_PAIRING_CODE_LENGTH)) + self._codes[peer_id] = (code, time.time()) + return code + + def verify_code(self, peer_id: str, code: str) -> bool: + entry = self._codes.get(peer_id) + if not entry: + return False + stored_code, created_at = entry + if time.time() - created_at > ALIPAY_PAIRING_CODE_TTL_SECONDS: + del self._codes[peer_id] + return False + if not secrets.compare_digest(stored_code, code): + return False + del self._codes[peer_id] + return True diff --git a/backend/package/yuxi/channel/extensions/alipay/plugin.json b/backend/package/yuxi/channel/extensions/alipay/plugin.json new file mode 100644 index 00000000..0f0411cb --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/plugin.json @@ -0,0 +1,29 @@ +{ + "id": "alipay", + "name": "支付宝", + "version": "0.1.0", + "description": "支付宝生活号+渠道插件，支持 RSA2 签名认证、被动回调消息接收、客服消息发送。覆盖 P0+P1 适配器（meta/capabilities/config/gateway/outbound/status/streaming/security/pairing/agentPrompt）", + "author": "ForcePilot Team", + "order": 50, + "dependencies": ["httpx>=0.27.0", "cryptography>=42.0.0"], + "capabilities": { + "chat_types": ["direct"], + "message_types": ["text", "image"], + "streaming": true, + "streaming_mode": "block", + "block_streaming": true, + "reactions": false, + "typing_indicator": false, + "threads": false, + "edit": false, + "unsend": true, + "reply": false, + "media": true, + "effects": false, + "native_commands": false, + "polls": false, + "group_management": false + }, + "enabled": false, + "python_requires": ">=3.12" +} \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/alipay/security.py b/backend/package/yuxi/channel/extensions/alipay/security.py new file mode 100644 index 00000000..5878dc88 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/security.py @@ -0,0 +1,42 @@ +from yuxi.channel.extensions.alipay.types import AlipayAccount, AlipayDmPolicy + + +class AlipaySecurity: + def __init__(self): + self._allow_from: list[str] = [] + + def resolve_dm_policy(self, account: AlipayAccount | None = None) -> dict: + if not account: + return {"mode": "disabled", "allow_from": []} + return { + "mode": account.dm_policy.value, + "allow_from": account.allow_from or [], + } + + def check_allowlist(self, peer_id: str, channel_type: str = "direct", account: AlipayAccount | None = None) -> bool: + if not account: + return False + if account.dm_policy == AlipayDmPolicy.OPEN: + return True + if account.dm_policy == AlipayDmPolicy.DISABLED: + return False + if account.dm_policy == AlipayDmPolicy.ALLOWLIST: + return peer_id in (account.allow_from or []) + if account.dm_policy == AlipayDmPolicy.PAIRING: + return peer_id in self._allow_from + return False + + def add_to_allowlist(self, peer_id: str) -> None: + if peer_id not in self._allow_from: + self._allow_from.append(peer_id) + + def collect_warnings( + self, config: dict | None = None, account_id: str | None = None, account: AlipayAccount | None = None + ) -> list[str]: + warnings = [] + if account: + if account.dm_policy == AlipayDmPolicy.OPEN: + warnings.append("DM 策略为 'open'，所有关注者均可发送消息") + if account.dm_policy == AlipayDmPolicy.OPEN and not account.allow_from: + warnings.append("DM 策略为 'open' 且无白名单限制") + return warnings diff --git a/backend/package/yuxi/channel/extensions/alipay/status.py b/backend/package/yuxi/channel/extensions/alipay/status.py new file mode 100644 index 00000000..163e49f9 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/status.py @@ -0,0 +1,76 @@ +import logging +import time + +from yuxi.channel.extensions.alipay.gateway import AlipayGateway +from yuxi.channel.extensions.alipay.types import AlipayAccount, AlipayMode, AlipayProbeResult + +logger = logging.getLogger(__name__) + + +class AlipayStatus: + def __init__(self, gateway: AlipayGateway): + self._gateway = gateway + + async def probe(self, account: AlipayAccount) -> AlipayProbeResult: + start = time.monotonic() + try: + result = await self._gateway.request( + account=account, + method="alipay.open.public.life.account.get", + ) + latency_ms = (time.monotonic() - start) * 1000 + return AlipayProbeResult( + ok=True, + app_id=account.app_id, + name=account.name or result.get("name", ""), + mode=account.mode, + latency_ms=latency_ms, + ) + except Exception as e: + latency_ms = (time.monotonic() - start) * 1000 + return AlipayProbeResult( + ok=False, + app_id=account.app_id, + name=account.name, + mode=account.mode, + latency_ms=latency_ms, + error=str(e), + ) + + def build_summary(self, account: AlipayAccount | None = None) -> dict: + if not account: + return {"status": "unconfigured"} + configured = account.is_configured() + return { + "status": "configured" if configured else "unconfigured", + "app_id": account.app_id, + "name": account.name or account.app_id, + "mode": account.mode.value, + "dm_policy": account.dm_policy.value, + "enabled": account.enabled, + } + + async def check_ready(self, account: AlipayAccount | None = None) -> bool: + if not account: + return False + if not account.is_configured(): + return False + if not account.enabled: + return False + result = await self.probe(account) + return result.ok + + def collect_status_issues(self, account: AlipayAccount | None = None) -> list[str]: + issues = [] + if not account: + issues.append("未配置支付宝账户") + return issues + if not account.app_id: + issues.append("未设置 AppId") + if not account.app_private_key: + issues.append("未设置应用私钥") + if not account.alipay_public_key: + issues.append("未设置支付宝公钥") + if account.mode == AlipayMode.SANDBOX: + issues.append("当前使用沙箱环境") + return issues diff --git a/backend/package/yuxi/channel/extensions/alipay/streaming.py b/backend/package/yuxi/channel/extensions/alipay/streaming.py new file mode 100644 index 00000000..98e36380 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/streaming.py @@ -0,0 +1,33 @@ +from yuxi.channel.extensions.alipay.constants import ( + ALIPAY_BLOCK_STREAMING_CHUNK_MAX_CHARS, + ALIPAY_BLOCK_STREAMING_CHUNK_MIN_CHARS, +) + + +class AlipayStreaming: + streaming_mode = "block" + preview_stream_throttle_ms = 160 + preview_min_initial_chars = 18 + + block_streaming_enabled = True + block_streaming_break = "double_newline" + block_streaming_chunk_min_chars = ALIPAY_BLOCK_STREAMING_CHUNK_MIN_CHARS + block_streaming_chunk_max_chars = ALIPAY_BLOCK_STREAMING_CHUNK_MAX_CHARS + block_streaming_chunk_break_preference = "double_newline_second_paragraph" + block_streaming_coalesce_defaults = { + "mintime": 1.2, + "maxtime": 3.5, + } + + ENABLED = True + STRATEGY = "block" + MIN_CHARS = ALIPAY_BLOCK_STREAMING_CHUNK_MIN_CHARS + MAX_INTERVAL_MS = 1000 + MAX_BLOCKS = 20 + FINISH_MARK = "" + + def create_draft_stream_session(self, target_id: str) -> dict: + return {"target_id": target_id, "mode": "block"} + + def create_block_chunker(self) -> dict: + return {"mode": "length", "min_chars": 50, "max_chars": 500} diff --git a/backend/package/yuxi/channel/extensions/alipay/types.py b/backend/package/yuxi/channel/extensions/alipay/types.py new file mode 100644 index 00000000..41ee5ca5 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/types.py @@ -0,0 +1,67 @@ +from dataclasses import dataclass, field +from enum import StrEnum + + +class AlipayMode(StrEnum): + PRODUCTION = "production" + SANDBOX = "sandbox" + + +class AlipayDmPolicy(StrEnum): + OPEN = "open" + PAIRING = "pairing" + ALLOWLIST = "allowlist" + DISABLED = "disabled" + + +@dataclass +class AlipayAccount: + account_id: str + app_id: str = "" + app_private_key: str = "" + alipay_public_key: str = "" + aes_key: str | None = None + mode: AlipayMode = AlipayMode.PRODUCTION + gateway_url: str = "https://openapi.alipay.com/gateway.do" + name: str = "" + enabled: bool = True + dm_policy: AlipayDmPolicy = AlipayDmPolicy.OPEN + allow_from: list[str] = field(default_factory=list) + remove_markdown: bool = True + subscribe_msg: str = "" + streaming_enabled: bool = True + block_streaming_chunk_max_chars: int = 500 + + def is_configured(self) -> bool: + return bool(self.app_id and self.app_private_key and self.alipay_public_key) + + +@dataclass +class AlipayInboundEvent: + msg_type: str + event_type: str | None + from_user_id: str + from_user_name: str | None + create_time: int + msg_id: str + app_id: str + biz_content: dict + raw: dict + + +@dataclass +class AlipayOutboundResult: + msg_id: str | None + success: bool + error: str | None = None + result: dict | None = None + + +@dataclass +class AlipayProbeResult: + ok: bool + app_id: str + name: str | None + mode: AlipayMode + latency_ms: float + error: str | None = None diff --git a/backend/package/yuxi/channel/extensions/alipay/webhook.py b/backend/package/yuxi/channel/extensions/alipay/webhook.py new file mode 100644 index 00000000..481b2788 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/alipay/webhook.py @@ -0,0 +1,258 @@ +import asyncio +import json +import logging + +from fastapi import APIRouter, Request, Response + +from yuxi.channel.extensions.alipay.config import AlipayConfig +from yuxi.channel.extensions.alipay.crypto import AlipayAESCrypto, AlipayCrypto +from yuxi.channel.extensions.alipay.dedup import AlipayMessageDeduplicator +from yuxi.channel.extensions.alipay.format import remove_markdown +from yuxi.channel.extensions.alipay.monitor import convert_alipay_event +from yuxi.channel.extensions.alipay.outbound import AlipayOutbound +from yuxi.channel.extensions.alipay.pairing import AlipayPairing +from yuxi.channel.extensions.alipay.security import AlipaySecurity +from yuxi.channel.extensions.alipay.types import AlipayInboundEvent +from yuxi.channel.runtime.manager import gateway +from yuxi.channel.message.models import MessageType, PeerInfo, PeerKind, UnifiedMessage + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/channel/alipay", tags=["alipay"]) + +_webhook_state: dict | None = None + + +def init_webhook( + config: AlipayConfig, + deduplicator: AlipayMessageDeduplicator, + security: AlipaySecurity, + pairing: AlipayPairing, + outbound: AlipayOutbound | None = None, +) -> None: + global _webhook_state + _webhook_state = { + "config": config, + "deduplicator": deduplicator, + "security": security, + "pairing": pairing, + "outbound": outbound, + } + + +def _get_state() -> dict: + if _webhook_state is None: + raise RuntimeError("Webhook not initialized") + return _webhook_state + + +def _build_crypto() -> AlipayCrypto | None: + state = _get_state() + account = state["config"].resolve_account() + if not account.is_configured(): + return None + return AlipayCrypto( + app_private_key_pem=account.app_private_key, + alipay_public_key_pem=account.alipay_public_key, + ) + + +@router.get("/callback") +async def alipay_url_verify(request: Request): + params = dict(request.query_params) + sign = params.pop("sign", None) + params.pop("sign_type", None) + + crypto = _build_crypto() + if crypto is None: + return Response(content="fail", status_code=503) + + if sign and crypto.verify(params, sign): + echostr = params.get("echostr", "") + return Response(content=echostr) + return Response(content="fail", status_code=400) + + +@router.post("/callback") +async def alipay_message_callback(request: Request): + try: + data = await request.json() + except (json.JSONDecodeError, UnicodeDecodeError): + return Response(content="fail", status_code=400) + + sign = data.pop("sign", None) + data.pop("sign_type", None) + + state = _get_state() + account = state["config"].resolve_account() + if not account.is_configured(): + return Response(content="fail", status_code=500) + + crypto = _build_crypto() + if crypto is None: + return Response(content="fail", status_code=500) + + if sign and not crypto.verify(data, sign): + logger.warning("支付宝回调签名验证失败") + return Response(content="fail", status_code=400) + + encrypt_type = data.get("encrypt_type", "") + if encrypt_type == "aes" and account.aes_key: + aes = AlipayAESCrypto(account.aes_key) + encrypted = data.get("biz_content", "") + if isinstance(encrypted, str) and encrypted: + try: + decrypted = aes.decrypt(encrypted) + data = json.loads(decrypted) + except Exception: + logger.exception("AES decrypt failed") + return Response(content="fail", status_code=400) + + msg_type = data.get("msg_type", data.get("MsgType", "")) + from_user_id = data.get("from_user_id", data.get("FromAlipayUserId", "")) + msg_id = data.get("msg_id", data.get("MsgId", "")) + + biz_content = data.get("biz_content", {}) + if isinstance(biz_content, str): + try: + biz_content = json.loads(biz_content) + except json.JSONDecodeError: + biz_content = {} + + if not msg_id or state["deduplicator"].is_duplicate(msg_id): + return Response(content="success") + + event = AlipayInboundEvent( + msg_type=msg_type, + event_type=data.get("event_type", data.get("EventType")), + from_user_id=from_user_id, + from_user_name=data.get("from_user_name", data.get("FromAlipayUserName")), + create_time=int(data.get("create_time", data.get("CreateTime", 0))), + msg_id=msg_id, + app_id=data.get("app_id", data.get("AppId", "")), + biz_content=biz_content, + raw=data, + ) + + security = state["security"] + if not security.check_allowlist(from_user_id, "direct", account): + if account.dm_policy.value == "pairing": + content = biz_content.get("content", "") if isinstance(biz_content, dict) else "" + if content.strip().startswith("配对 "): + code_input = content.strip()[3:].strip() + if state["pairing"].verify_code(from_user_id, code_input): + return await _handle_pairing_success(from_user_id, security) + else: + await _send_alipay_text(from_user_id, "配对码无效或已过期，请重新发送消息获取配对码。") + else: + code = state["pairing"].generate_code(from_user_id) + if code: + await _send_alipay_text( + from_user_id, + f"首次对话需要验证身份，请输入以下配对码：\n\n配对 {code}\n\n（配对码有效期 5 分钟）", + ) + else: + await _send_alipay_text(from_user_id, "配对请求过于频繁，请稍后再试。") + return Response(content="success") + return Response(content="success") + + outbound = state.get("outbound") + if outbound: + outbound.record_interaction(from_user_id, account.account_id) + + if event.msg_type == "event": + return await _handle_event(event, account) + + content_str = (event.biz_content or {}).get("content", "") + if not content_str: + return Response(content="success") + + if account.remove_markdown: + content_str = remove_markdown(content_str) + + return await _dispatch_to_agent(event, content_str) + + +async def _handle_event(event: AlipayInboundEvent, account) -> Response: + if event.event_type in ("follow", "enter"): + unified = convert_alipay_event(event, account) + if unified: + await _dispatch_unified(unified) + welcome = account.subscribe_msg or "" + if welcome: + await _send_alipay_text(event.from_user_id, welcome) + logger.info("Sent welcome message to %s", event.from_user_id) + return Response(content="success") + + +async def _dispatch_to_agent(event: AlipayInboundEvent, content_str: str) -> Response: + state = _get_state() + account = state["config"].resolve_account() + + unified = UnifiedMessage( + msg_id=event.msg_id, + channel_type="alipay", + account_id=account.account_id, + content=content_str, + message_type=MessageType.TEXT if event.msg_type == "text" else MessageType.IMAGE, + sender=PeerInfo( + id=event.from_user_id, + kind=PeerKind.DIRECT, + display_name=event.from_user_name or event.from_user_id, + ), + raw_payload=event.raw, + body_for_agent=content_str, + metadata={ + "FromAlipayUserId": event.from_user_id, + "AppId": event.app_id, + "MsgType": event.msg_type, + "EventType": event.event_type or "", + }, + ) + + await _dispatch_unified(unified) + return Response(content="success") + + +async def _dispatch_unified(msg: UnifiedMessage) -> None: + processor = gateway._processor + if processor is None: + logger.warning("Message processor not available, cannot dispatch Alipay message") + return + + asyncio.create_task( + _run_processor(processor, msg), + name=f"alipay-dispatch-{msg.sender.id}", + ) + + +async def _run_processor(processor, msg: UnifiedMessage) -> None: + try: + await asyncio.wait_for(processor.process(msg), timeout=120.0) + except TimeoutError: + logger.error("Agent response timeout for alipay user %s", msg.sender.id) + except Exception: + logger.exception("Failed to process Alipay message for user %s", msg.sender.id) + + +async def _handle_pairing_success(from_user_id: str, security: AlipaySecurity) -> Response: + security.add_to_allowlist(from_user_id) + await _send_alipay_text(from_user_id, "配对成功！现在可以开始对话了。") + return Response(content="success") + + +async def _send_alipay_text(to_user: str, content: str) -> None: + state = _get_state() + account = state["config"].resolve_account() + if not account.is_configured(): + logger.warning("Alipay account not configured, cannot send text") + return + + outbound: AlipayOutbound | None = state.get("outbound") + if outbound is None: + return + + try: + await outbound.send_text(to_user, content, account=account) + except Exception: + logger.exception("Failed to send Alipay text to %s", to_user)