From 6bd13a7a8e591227c106caad2bdaba3194e058e8 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Thu, 9 Jul 2026 04:18:53 +0800 Subject: [PATCH] =?UTF-8?q?feat(instagram):=20=E5=AE=8C=E6=88=90Instagram?= =?UTF-8?q?=E6=8F=92=E4=BB=B6=E5=A4=9A=E7=BB=B4=E5=BA=A6=E5=8A=9F=E8=83=BD?= =?UTF-8?q?=E4=BC=98=E5=8C=96=E4=B8=8E=E8=A7=84=E8=8C=83=E8=B0=83=E6=95=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 本次提交对Instagram插件进行了全面改进： 1. 调整OAuth凭据配置项，将app_id/app_secret移至必填字段 2. 重构异常链处理逻辑，移除手动设置__cause__代码 3. 新增OAuth基础URL常量并统一API端点调用 4. 实现可配置的token刷新提前天数阈值，替代硬编码7天 5. 修复白名单适配器异常捕获逻辑，精准处理Error类型 6. 优化入站适配器对location消息、referral事件的处理 7. 重构消息缓存键，新增账户前缀支持按账户清理缓存 8. 完善OAuth回调流程，新增expires_in校验与过期时间计算 9. 修复消息窗口检查逻辑，调整重试退避策略 10. 新增凭据轮换相关辅助方法，优化验证与刷新流程 11. 移除废弃的飞书DTO文件 --- .../channels/plugins/feishu/dtos/__init__.py | 59 -------- .../channels/plugins/instagram/_constants.py | 6 + .../instagram/adapters/doctor_adapter.py | 9 +- .../instagram/adapters/inbound_adapter.py | 40 ++++-- .../instagram/adapters/lifecycle_adapter.py | 130 +++++++++++++----- .../instagram/adapters/message_ops_adapter.py | 34 +++-- .../instagram/adapters/outbound_adapter.py | 71 ++++++++-- .../instagram/adapters/probeable_adapter.py | 9 +- .../instagram/adapters/status_adapter.py | 21 ++- .../instagram/adapters/whitelist_adapter.py | 5 - .../instagram/adapters/wizard_adapter.py | 26 ++++ .../plugins/instagram/error_translator.py | 4 +- .../plugins/instagram/instagram_client.py | 85 +++++++++--- .../channels/plugins/instagram/manifest.json | 4 +- 14 files changed, 334 insertions(+), 169 deletions(-) delete mode 100644 backend/package/yuxi/channels/plugins/feishu/dtos/__init__.py diff --git a/backend/package/yuxi/channels/plugins/feishu/dtos/__init__.py b/backend/package/yuxi/channels/plugins/feishu/dtos/__init__.py deleted file mode 100644 index cb1b42e0..00000000 --- a/backend/package/yuxi/channels/plugins/feishu/dtos/__init__.py +++ /dev/null @@ -1,59 +0,0 @@ -"""飞书插件内部 DTO。 - -仅插件内部使用的中间解析结构，不导出到契约层。所有 DTO 为 ``frozen`` -dataclass，用于飞书事件 payload 与消息内容的中间解析。 -""" - -from __future__ import annotations - -from dataclasses import dataclass -from typing import Any - - -@dataclass(frozen=True) -class FeishuEventHeader: - """飞书事件头。 - - 字段： - event_id: 事件 ID。 - event_type: 事件类型。 - create_time: 事件创建时间。 - token: 校验 token。 - app_id: 应用 ID。 - tenant_key: 租户标识。 - """ - - event_id: str - event_type: str - create_time: str - token: str - app_id: str - tenant_key: str - - -@dataclass(frozen=True) -class FeishuEventPayload: - """飞书事件 payload 解析中间结构。 - - 字段： - schema: 协议版本（如 ``2.0``）。 - header: 事件头。 - event: 事件业务数据。 - """ - - schema: str - header: FeishuEventHeader - event: dict[str, Any] - - -@dataclass(frozen=True) -class FeishuMessageContent: - """飞书消息内容解析中间结构。 - - 字段： - msg_type: 消息类型（如 ``text`` / ``post`` / ``image``）。 - content: 消息内容字典。 - """ - - msg_type: str - content: dict[str, Any] diff --git a/backend/package/yuxi/channels/plugins/instagram/_constants.py b/backend/package/yuxi/channels/plugins/instagram/_constants.py index d1733ed0..2b938e8b 100644 --- a/backend/package/yuxi/channels/plugins/instagram/_constants.py +++ b/backend/package/yuxi/channels/plugins/instagram/_constants.py @@ -71,6 +71,11 @@ WHITELIST_CACHE_KEY_PREFIX = "instagram:whitelist:" # token_refresh_lead_days.default=7） DEFAULT_TOKEN_REFRESH_LEAD_DAYS = 7 +# OAuth 授权端点基础 URL（Instagram Login 路径，与 wizard_adapter 共享） +# Step 1 短期 token 交换必须使用 api.instagram.com，graph.instagram.com 仅用于 +# Step 2 长期 token 交换（Meta 官方约束） +OAUTH_API_BASE = "https://api.instagram.com" + # Long-lived Token 有效期（秒），Instagram User Access Token 60 天 LONG_LIVED_TOKEN_TTL_SECONDS = 60 * 24 * 60 * 60 @@ -144,6 +149,7 @@ __all__ = [ "TOKEN_CACHE_KEY_PREFIX", "WHITELIST_CACHE_KEY_PREFIX", "DEFAULT_TOKEN_REFRESH_LEAD_DAYS", + "OAUTH_API_BASE", "LONG_LIVED_TOKEN_TTL_SECONDS", "MAX_MESSAGE_LENGTH", "MAX_QUICK_REPLY_OPTIONS", diff --git a/backend/package/yuxi/channels/plugins/instagram/adapters/doctor_adapter.py b/backend/package/yuxi/channels/plugins/instagram/adapters/doctor_adapter.py index 8f963fce..939064ec 100644 --- a/backend/package/yuxi/channels/plugins/instagram/adapters/doctor_adapter.py +++ b/backend/package/yuxi/channels/plugins/instagram/adapters/doctor_adapter.py @@ -214,7 +214,12 @@ class InstagramDoctorAdapter: # ------------------------------------------------------------------ async def _check_token_validity(self, account_id: str) -> DiagnosticResult: - """检查 Token 有效性。""" + """检查 Token 有效性。 + + 过期阈值通过 ``InstagramClient.get_token_refresh_lead_days`` 读取 + 账户级 ``token_refresh_lead_days``，未配置时回退默认值。 + """ + lead_days = await self._client.get_token_refresh_lead_days(account_id) try: resp = await self._client.get_debug_token(account_id) except Error as exc: @@ -263,7 +268,7 @@ class InstagramDoctorAdapter: if expires_at and isinstance(expires_at, (int, float)): now = datetime.now(UTC).timestamp() remaining_days = (expires_at - now) / 86400 - if remaining_days < 7: + if remaining_days < lead_days: return DiagnosticResult( check_id=_CHECK_TOKEN_VALIDITY, passed=False, diff --git a/backend/package/yuxi/channels/plugins/instagram/adapters/inbound_adapter.py b/backend/package/yuxi/channels/plugins/instagram/adapters/inbound_adapter.py index ca7e32dd..b1f097b8 100644 --- a/backend/package/yuxi/channels/plugins/instagram/adapters/inbound_adapter.py +++ b/backend/package/yuxi/channels/plugins/instagram/adapters/inbound_adapter.py @@ -33,6 +33,7 @@ from yuxi.channels.contract.ports.driven.logger_port import LoggerPort from .._constants import ( CHANNEL_TARGET, MESSAGING_WINDOW_SECONDS, + MSG_ACCOUNT_CACHE_KEY_PREFIX, WINDOW_CACHE_KEY_PREFIX, ) from ..instagram_client import InstagramClient @@ -133,6 +134,16 @@ class InstagramInboundAdapter: "message_type": message_type, } + # location 消息补充坐标到 metadata（_parse_message 仅标记 message_type， + # 此处从原始 attachments 提取 coordinates 供下游使用） + if message_type == "location": + for attach in message.get("attachments", []): + if isinstance(attach, dict) and attach.get("type") == "location": + coords = attach.get("payload", {}).get("coordinates") + if isinstance(coords, dict): + metadata["coordinates"] = coords + break + # 引用回复信息 reply_to = message.get("reply_to") if isinstance(reply_to, dict): @@ -155,6 +166,22 @@ class InstagramInboundAdapter: ttl_seconds=MESSAGING_WINDOW_SECONDS, ) + # 写入 channel_msg_id → {account_id, recipient_igsid} 反向映射， + # 供 MessageOpsAdapter（如 Reactions）解析入站消息的账户与接收方上下文。 + # 双键策略：legacy 键（无前缀）用于 sendReaction 反查；带 account_id + # 前缀键用于 beforeAccountDelete 按 pattern 清理。 + # 仅当 mid 存在时写入（部分事件如纯 referral 无 mid）。 + if mid: + legacy_value = {"account_id": account_id, "recipient_igsid": sender_igsid} + await self._cache.set( + f"{MSG_ACCOUNT_CACHE_KEY_PREFIX}{mid}", + legacy_value, + ) + await self._cache.set( + f"{MSG_ACCOUNT_CACHE_KEY_PREFIX}{account_id}:{mid}", + legacy_value, + ) + return MessageContent( text=text, format=MessageFormat.TEXT, @@ -379,17 +406,10 @@ def _parse_message( message_type = "story_reply" continue - # location 附件构造为 file 类型（契约 Attachment 仅支持 image/file/audio/video） + # location 附件无可下载 URL，跳过构造 Attachment（避免 downloadAttachment + # 尝试 HTTP GET 失败）。仅标记 message_type，坐标由调用方从原始 + # attachments 提取到 metadata.coordinates。 if attach_type == "location": - lat = attach_payload.get("coordinates", {}).get("lat", 0) - long = attach_payload.get("coordinates", {}).get("long", 0) - # location 无可下载 URL，使用占位 URL 编码坐标信息 - url = f"instagram://location?lat={lat}&long={long}" - attachment = Attachment( - type="file", - url=url, - ) - attachments.append(attachment) message_type = "location" continue diff --git a/backend/package/yuxi/channels/plugins/instagram/adapters/lifecycle_adapter.py b/backend/package/yuxi/channels/plugins/instagram/adapters/lifecycle_adapter.py index 7c262d47..1de97bd8 100644 --- a/backend/package/yuxi/channels/plugins/instagram/adapters/lifecycle_adapter.py +++ b/backend/package/yuxi/channels/plugins/instagram/adapters/lifecycle_adapter.py @@ -30,6 +30,7 @@ from yuxi.channels.contract.dtos.channel import ( ChannelAccount, RotateCredentialsResult, ) +from yuxi.channels.contract.dtos.config import ConfigScope from yuxi.channels.contract.errors import ( DependencyError, Error, @@ -40,11 +41,7 @@ from yuxi.channels.contract.ports.driven.cache_port import CachePort from yuxi.channels.contract.ports.driven.config_port import ConfigPort from yuxi.channels.contract.ports.driven.logger_port import LoggerPort -from .._constants import ( - DEFAULT_API_VERSION, - DEFAULT_TOKEN_REFRESH_LEAD_DAYS, - LONG_LIVED_TOKEN_TTL_SECONDS, -) +from .._constants import DEFAULT_API_VERSION from ..instagram_client import InstagramClient # 必填配置字段 @@ -57,9 +54,10 @@ _FIELD_API_VERSION = "api_version" _FIELD_WEBHOOK_VERIFY_TOKEN = "webhook_verify_token" # CachePort 清理键模式 -_CACHE_KEY_PREFIX_TOKEN = "instagram:token:" _CACHE_KEY_PREFIX_WHITELIST = "instagram:whitelist:" _CACHE_KEY_PREFIX_WINDOW = "instagram:window:" +_CACHE_KEY_PREFIX_MSG = "instagram:msg:" +_CACHE_KEY_PREFIX_REACTION = "instagram:reaction:" # IGID 纯数字校验 _IGID_PATTERN = re.compile(r"^\d+$") @@ -138,9 +136,14 @@ class InstagramLifecycleAdapter: # 补全默认值 if _FIELD_API_VERSION not in normalized or not normalized[_FIELD_API_VERSION]: normalized[_FIELD_API_VERSION] = DEFAULT_API_VERSION - # 计算并补全 token_expires_at（若缺失则按 60 天计算） + # token_expires_at 必须由调用方（OAuth 流程）传入，缺失则视为配置错误。 + # OAuth 响应包含 expires_in（秒），调用方应基于此计算绝对过期时间。 + # 不在此处硬编码补全，避免掩盖调用方未传 expires_in 的设计缺陷。 if _FIELD_TOKEN_EXPIRES_AT not in normalized or not normalized[_FIELD_TOKEN_EXPIRES_AT]: - normalized[_FIELD_TOKEN_EXPIRES_AT] = time.time() + LONG_LIVED_TOKEN_TTL_SECONDS + raise ValidationError( + field=_FIELD_TOKEN_EXPIRES_AT, + message="token_expires_at_required_oauth_must_provide_expires_in", + ) return normalized async def afterAccountConfigWritten(self, account: ChannelAccount) -> None: @@ -160,7 +163,7 @@ class InstagramLifecycleAdapter: ) return - lead_days = await self._get_token_refresh_lead_days() + lead_days = await self._client.get_token_refresh_lead_days(account.account_id) now = time.time() remaining_days = (expires_at - now) / 86400 if remaining_days < lead_days: @@ -183,22 +186,32 @@ class InstagramLifecycleAdapter: """删除前回调：清空 CachePort 中该账户的所有缓存。 清理键模式： - - ``instagram:token:{account_id}`` - ``instagram:whitelist:{account_id}:*`` - ``instagram:window:{account_id}:*`` + - ``instagram:msg:{account_id}:*``（出站消息反向映射） + - ``instagram:reaction:{account_id}:*``（reaction emoji 缓存） + + 注：``instagram:msg:{mid}`` legacy 键（无 account_id 前缀）无法按账户 + 清理，依赖 TTL 自然过期。新数据写入时已同时写入带前缀键，此处可 + 按账户清理带前缀键。 CachePort 异常 **不抛**（仅日志告警，避免阻塞删除流程）。 幂等：重复调用安全。 """ try: - await self._cache.delete(f"{_CACHE_KEY_PREFIX_TOKEN}{account.account_id}") await self._cache.invalidate( f"{_CACHE_KEY_PREFIX_WHITELIST}{account.account_id}:*", ) await self._cache.invalidate( f"{_CACHE_KEY_PREFIX_WINDOW}{account.account_id}:*", ) + await self._cache.invalidate( + f"{_CACHE_KEY_PREFIX_MSG}{account.account_id}:*", + ) + await self._cache.invalidate( + f"{_CACHE_KEY_PREFIX_REACTION}{account.account_id}:*", + ) await self._logger.info( "Instagram account delete: cache cleared", account_id=account.account_id, @@ -258,17 +271,27 @@ class InstagramLifecycleAdapter: self, account: ChannelAccount, ) -> RotateCredentialsResult: - """轮换账户凭据：调用 ``refreshToken`` 获取新 Token。 + """轮换账户凭据：刷新 Token 并持久化到 ConfigPort。 - @failure 刷新失败抛 ``DependencyError``；Token 无效抛 ``AuthError``。 + 直接调用 ``client.refresh_token``（保留错误传播），成功后调 + ``_persist_refreshed_token`` 写回 ConfigPort，确保轮换后后续 + 请求使用新 token。 + + @failure 刷新失败抛 ``DependencyError``。 """ try: - await self._client.refresh_token(account.account_id) + resp = await self._client.refresh_token(account.account_id) except Error: raise except Exception as exc: raise DependencyError(dep="instagram_api", cause=exc) from exc + if not await self._persist_refreshed_token(account.account_id, resp): + raise DependencyError( + dep="instagram_api", + cause=ValueError("token refresh response invalid or persist failed"), + ) + return RotateCredentialsResult( account_id=account.account_id, rotated_at=datetime.now(UTC), @@ -278,15 +301,17 @@ class InstagramLifecycleAdapter: async def validateCredentials(self, config: dict[str, Any]) -> bool: """验证凭据有效性。 - 调用 ``InstagramClient.get_debug_token`` 验证 ``access_token`` - 有效性。有效返回 ``True``，无效返回 ``False``。API 异常 **不抛**， - 返回 ``False``。 + 直接使用 ``config["access_token"]`` 调用 + ``InstagramClient.get_debug_token_for_token`` 验证，适用于凭据 + 轮换场景（新 token 尚未落库）。 + + @return 有效返回 ``True``，无效或 API 异常返回 ``False``。 """ - ig_id = config.get(_FIELD_IG_ID) - if not ig_id: + access_token = config.get(_FIELD_ACCESS_TOKEN) + if not access_token: return False try: - resp = await self._client.get_debug_token(ig_id) + resp = await self._client.get_debug_token_for_token(access_token) except Exception: return False @@ -308,7 +333,11 @@ class InstagramLifecycleAdapter: # ------------------------------------------------------------------ async def _refresh_token(self, account_id: str) -> None: - """刷新 Token 并记录日志。""" + """刷新 Token 并更新 ConfigPort（容错：异常仅日志告警不抛）。 + + 供 ``afterAccountConfigWritten`` 调用——刷新失败不应阻塞配置写入 + 后的回调流程。 + """ try: resp = await self._client.refresh_token(account_id) except Error as exc: @@ -326,26 +355,53 @@ class InstagramLifecycleAdapter: ) return - expires_in = resp.get("expires_in", 0) + await self._persist_refreshed_token(account_id, resp) + + async def _persist_refreshed_token(self, account_id: str, resp: dict) -> bool: + """将刷新响应中的新 Token 持久化到 ConfigPort。 + + @return 成功返回 ``True``，响应无效或 ConfigPort 写入失败返回 + ``False``（仅日志告警，不抛异常）。 + """ + new_token = resp.get("access_token") + expires_in = int(resp.get("expires_in", 0)) + if not new_token or not expires_in: + await self._logger.warn( + "Instagram token refresh response missing access_token or expires_in", + account_id=account_id, + ) + return False + + # 更新 ConfigPort 中的 access_token 与 token_expires_at， + # 确保后续请求使用新 token，且刷新调度基于新的过期时间 + new_expires_at = time.time() + expires_in + try: + await self._config.set( + "access_token", + new_token, + scope=ConfigScope.ACCOUNT, + target=account_id, + ) + await self._config.set( + "token_expires_at", + new_expires_at, + scope=ConfigScope.ACCOUNT, + target=account_id, + ) + except Exception as exc: + await self._logger.warn( + "Instagram token refresh: failed to update ConfigPort", + account_id=account_id, + error=str(exc), + ) + return False + await self._logger.info( - "Instagram token refreshed", + "Instagram token refreshed and persisted", account_id=account_id, expires_in=expires_in, ) - - async def _get_token_refresh_lead_days(self) -> int: - """从 ConfigPort 读取 ``token_refresh_lead_days``。 - - ConfigPort 未命中时回退到 ``DEFAULT_TOKEN_REFRESH_LEAD_DAYS``。 - """ - try: - cv = await self._config.get("token_refresh_lead_days") - value = cv.value if hasattr(cv, "value") else None - if isinstance(value, int) and value > 0: - return value - except Exception: - pass - return DEFAULT_TOKEN_REFRESH_LEAD_DAYS + return True __all__ = ["InstagramLifecycleAdapter"] diff --git a/backend/package/yuxi/channels/plugins/instagram/adapters/message_ops_adapter.py b/backend/package/yuxi/channels/plugins/instagram/adapters/message_ops_adapter.py index 48e7125c..c5ae4e93 100644 --- a/backend/package/yuxi/channels/plugins/instagram/adapters/message_ops_adapter.py +++ b/backend/package/yuxi/channels/plugins/instagram/adapters/message_ops_adapter.py @@ -183,7 +183,7 @@ class InstagramMessageOpsAdapter: # 缓存 reaction emoji 供 deleteReaction 反查 await self._cache.set( - _reaction_cache_key(channel_msg_id), + _reaction_cache_key(account_id, channel_msg_id), emoji, ) return True @@ -191,7 +191,8 @@ class InstagramMessageOpsAdapter: async def deleteReaction(self, channel_msg_id: str, emoji: str) -> bool: """删除表情反应。 - Instagram Send API 通过发送空 emoji 移除反应。 + Instagram Send API 通过发送空 emoji 移除反应。``emoji`` 参数保留 + 仅为契约兼容，实际不使用（Instagram 不支持按 emoji 选择性删除）。 """ account_id, recipient_igsid = await self._resolve_msg_context(channel_msg_id) try: @@ -206,7 +207,7 @@ class InstagramMessageOpsAdapter: except Exception as exc: raise DependencyError(dep=INSTAGRAM_API_DEP, cause=exc) from exc - await self._cache.delete(_reaction_cache_key(channel_msg_id)) + await self._cache.delete(_reaction_cache_key(account_id, channel_msg_id)) return True async def pinMessage(self, channel_msg_id: str) -> bool: @@ -276,16 +277,22 @@ class InstagramMessageOpsAdapter: async def _resolve_msg_context(self, channel_msg_id: str) -> tuple[str, str]: """通过 CachePort 反向映射解析 account_id 与 recipient_igsid。 - 出站适配器在发送消息时写入 ``instagram:msg:{channel_msg_id}`` → - ``{"account_id": ..., "recipient_igsid": ...}`` 复合值，供 Reactions - 等后续操作反查账户与接收方上下文。 + 出站适配器与入站适配器在消息处理时写入双键： + - legacy 键 ``instagram:msg:{channel_msg_id}``（无 account_id + 前缀，供本方法反查） + - 带前缀键 ``instagram:msg:{account_id}:{channel_msg_id}`` + （供 beforeAccountDelete 按 pattern 清理） + + 本方法仅查询 legacy 键（因契约 ``sendReaction`` 仅接收 + ``channel_msg_id``，无法预知 account_id）。多账户下 channel_msg_id + 冲突时返回首个匹配（Instagram message_id 全局唯一，实际不冲突）。 @return ``(account_id, recipient_igsid)`` 元组。 @failure 无缓存记录抛 ``NotFoundError``。 """ from yuxi.channels.contract.errors import NotFoundError - opt = await self._cache.get(_msg_account_cache_key(channel_msg_id)) + opt = await self._cache.get(_msg_account_cache_key_legacy(channel_msg_id)) if opt.is_nothing(): raise NotFoundError( resource="channel_msg_id", @@ -304,11 +311,18 @@ class InstagramMessageOpsAdapter: return account_id, recipient_igsid -def _reaction_cache_key(channel_msg_id: str) -> str: - return f"{REACTION_CACHE_KEY_PREFIX}{channel_msg_id}" +def _reaction_cache_key(account_id: str, channel_msg_id: str) -> str: + """构造 reaction 缓存键（含 account_id 前缀，便于按账户清理）。""" + return f"{REACTION_CACHE_KEY_PREFIX}{account_id}:{channel_msg_id}" -def _msg_account_cache_key(channel_msg_id: str) -> str: +def _msg_account_cache_key_legacy(channel_msg_id: str) -> str: + """构造旧格式反向映射缓存键（无 account_id 前缀）。 + + 保留用于兼容查询：由于 ``_resolve_msg_context`` 仅接收 channel_msg_id， + 无法预知 account_id，需通过旧格式键回退查询。新数据写入使用 + outbound_adapter 中的带 account_id 前缀键。 + """ return f"{MSG_ACCOUNT_CACHE_KEY_PREFIX}{channel_msg_id}" diff --git a/backend/package/yuxi/channels/plugins/instagram/adapters/outbound_adapter.py b/backend/package/yuxi/channels/plugins/instagram/adapters/outbound_adapter.py index 1be6fcb2..8eba3aa8 100644 --- a/backend/package/yuxi/channels/plugins/instagram/adapters/outbound_adapter.py +++ b/backend/package/yuxi/channels/plugins/instagram/adapters/outbound_adapter.py @@ -48,8 +48,20 @@ _MARKDOWN_PATTERNS = ( ) -def _msg_account_cache_key(channel_msg_id: str) -> str: - """构造 channel_msg_id → account_id 反向映射缓存键。""" +def _msg_account_cache_key(account_id: str, channel_msg_id: str) -> str: + """构造 channel_msg_id → account_id 反向映射缓存键。 + + 缓存键含 account_id 前缀，确保 beforeAccountDelete 可按账户清理。 + """ + return f"{MSG_ACCOUNT_CACHE_KEY_PREFIX}{account_id}:{channel_msg_id}" + + +def _msg_account_cache_key_legacy(channel_msg_id: str) -> str: + """构造 legacy 反向映射缓存键（无 account_id 前缀）。 + + 供 MessageOpsAdapter._resolve_msg_context 反查（仅接收 channel_msg_id， + 无法预知 account_id）。与 ``_msg_account_cache_key`` 双写保证兼容。 + """ return f"{MSG_ACCOUNT_CACHE_KEY_PREFIX}{channel_msg_id}" @@ -58,6 +70,11 @@ def _window_cache_key(account_id: str, peer_id: str) -> str: return f"{WINDOW_CACHE_KEY_PREFIX}{account_id}:{peer_id}" +# 窗口关闭后的合理 backoff（秒）。窗口不会自然重开（需用户再次交互）， +# 此值供 outbox 重试机制参考，避免立即重试必然再次失败 +_WINDOW_CLOSED_BACKOFF_SECONDS = 3600 + + class InstagramOutboundAdapter: """Instagram 出站适配器。 @@ -174,9 +191,15 @@ class InstagramOutboundAdapter: cause=ValueError("send_message_missing_message_id"), ) # 写入 channel_msg_id → {account_id, recipient_igsid} 反向映射， - # 供 MessageOpsAdapter 解析账户与接收方上下文 + # 供 MessageOpsAdapter 解析账户与接收方上下文。 + # 双键策略：legacy 键（无前缀）用于 sendReaction 反查；带 account_id + # 前缀键用于 beforeAccountDelete 按 pattern 清理。 await self._cache.set( - _msg_account_cache_key(message_id), + _msg_account_cache_key_legacy(message_id), + {"account_id": account_id, "recipient_igsid": peer_id}, + ) + await self._cache.set( + _msg_account_cache_key(account_id, message_id), {"account_id": account_id, "recipient_igsid": peer_id}, ) return message_id @@ -234,10 +257,16 @@ class InstagramOutboundAdapter: ) if msg_id: await self._cache.set( - _msg_account_cache_key(msg_id), + _msg_account_cache_key_legacy(msg_id), {"account_id": account_id, "recipient_igsid": peer_id}, ) - except Error as exc: + await self._cache.set( + _msg_account_cache_key(account_id, msg_id), + {"account_id": account_id, "recipient_igsid": peer_id}, + ) + except Error: + raise + except Exception as exc: raise DependencyError( dep=INSTAGRAM_API_DEP, cause=exc, @@ -270,10 +299,16 @@ class InstagramOutboundAdapter: ) if msg_id: await self._cache.set( - _msg_account_cache_key(msg_id), + _msg_account_cache_key_legacy(msg_id), {"account_id": account_id, "recipient_igsid": peer_id}, ) - except Error as exc: + await self._cache.set( + _msg_account_cache_key(account_id, msg_id), + {"account_id": account_id, "recipient_igsid": peer_id}, + ) + except Error: + raise + except Exception as exc: raise DependencyError( dep=INSTAGRAM_API_DEP, cause=exc, @@ -333,6 +368,16 @@ class InstagramOutboundAdapter: dep=INSTAGRAM_API_DEP, cause=ValueError("send_thread_message_missing_message_id"), ) + # 写入双键反向映射（与 sendMessage 一致），供 MessageOpsAdapter + # 解析账户与接收方上下文（线程回复消息也支持 reaction） + await self._cache.set( + _msg_account_cache_key_legacy(message_id), + {"account_id": account_id, "recipient_igsid": peer_id}, + ) + await self._cache.set( + _msg_account_cache_key(account_id, message_id), + {"account_id": account_id, "recipient_igsid": peer_id}, + ) return message_id async def batchSendMessages( @@ -367,6 +412,9 @@ class InstagramOutboundAdapter: 从 CachePort 读取该 peer_id 的最后用户交互时间戳，若距当前时间 > 24h，抛 ``RateLimitError`` 由 outbox 重试机制在窗口重开后自动 投递。 + + 窗口不会自然重开（需用户再次交互触发），retry_after 设为合理 backoff + （1 小时），避免 outbox 立即重试必然再次失败。 """ cache_key = self._window_cache_key(account_id, peer_id) opt = await self._cache.get(cache_key) @@ -379,12 +427,11 @@ class InstagramOutboundAdapter: elapsed = time.time() - float(last_interaction_ts) if elapsed > MESSAGING_WINDOW_SECONDS: - retry_after = MESSAGING_WINDOW_SECONDS - int(elapsed) - if retry_after < 0: - retry_after = 0 + # 窗口已关闭，retry_after 设为合理 backoff（窗口不会自然重开， + # 需用户再次交互；outbox 应按指数退避，此处仅给最小建议值） raise RateLimitError( resource=INSTAGRAM_API_DEP, - retry_after_ms=retry_after * 1000, + retry_after_ms=_WINDOW_CLOSED_BACKOFF_SECONDS * 1000, ) @staticmethod diff --git a/backend/package/yuxi/channels/plugins/instagram/adapters/probeable_adapter.py b/backend/package/yuxi/channels/plugins/instagram/adapters/probeable_adapter.py index 4cbeec5b..0112ac9f 100644 --- a/backend/package/yuxi/channels/plugins/instagram/adapters/probeable_adapter.py +++ b/backend/package/yuxi/channels/plugins/instagram/adapters/probeable_adapter.py @@ -148,7 +148,12 @@ class InstagramProbeableAdapter: self, account_id: str, ) -> tuple[bool, str | None]: - """深度探测 Token 有效性。""" + """深度探测 Token 有效性。 + + 过期阈值通过 ``InstagramClient.get_token_refresh_lead_days`` 读取 + 账户级 ``token_refresh_lead_days``，未配置时回退默认值。 + """ + lead_days = await self._client.get_token_refresh_lead_days(account_id) try: resp = await self._client.get_debug_token(account_id) except Error as exc: @@ -170,7 +175,7 @@ class InstagramProbeableAdapter: remaining = expires_at - now if remaining < 0: return False, "token_expired" - if remaining < 7 * 86400: + if remaining < lead_days * 86400: return False, f"token_expiring_soon: {int(remaining / 86400)} days" return True, "token_valid" diff --git a/backend/package/yuxi/channels/plugins/instagram/adapters/status_adapter.py b/backend/package/yuxi/channels/plugins/instagram/adapters/status_adapter.py index 20897eb5..8a3e8f6f 100644 --- a/backend/package/yuxi/channels/plugins/instagram/adapters/status_adapter.py +++ b/backend/package/yuxi/channels/plugins/instagram/adapters/status_adapter.py @@ -15,17 +15,6 @@ from yuxi.channels.contract.dtos.common import RawEvent from yuxi.channels.contract.dtos.status import EventType, StatusPayload from yuxi.channels.contract.errors import ValidationError -# Webhook 事件字段名 → EventType 映射 -_FIELD_EVENT_MAP = ( - ("message", EventType.MESSAGE), - ("reaction", EventType.UNKNOWN), # REACTION 不在 EventType 枚举中，使用 UNKNOWN - ("postback", EventType.UNKNOWN), # POSTBACK 不在 EventType 枚举中，使用 UNKNOWN - ("referral", EventType.UNKNOWN), # REFERRAL 不在 EventType 枚举中，使用 UNKNOWN - ("delivery", EventType.DELIVERED), - ("read", EventType.READ), - ("deleted_message", EventType.RECALLED), # MESSAGE_DELETE 映射为 RECALLED -) - # 反应事件专用 EventType（存入 StatusPayload.event_type） _REACTION_EVENT_TYPE = EventType.UNKNOWN _POSTBACK_EVENT_TYPE = EventType.MESSAGE # Postback 触发对话，按 MESSAGE 处理 @@ -123,8 +112,16 @@ class InstagramStatusAdapter: # 提取关联消息 ID（不同事件类型字段不同） ref_msg_id = _extract_ref_msg_id(payload, event_type) + + # referral 事件（ig.me 链接入口）无 message.mid，使用 sender.id 作为 + # 关联 ID 兜底，避免 referral 事件被 ValidationError 误拒 + if not ref_msg_id and "referral" in payload: + sender = payload.get("sender", {}) + if isinstance(sender, dict): + ref_msg_id = sender.get("id") + if not ref_msg_id: - # 状态事件必须携带关联消息 ID + # 状态事件必须携带关联消息 ID（referral 已兜底，仍无则视为异常） raise ValidationError( field="ref_channel_msg_id", message="ref_msg_id_required", diff --git a/backend/package/yuxi/channels/plugins/instagram/adapters/whitelist_adapter.py b/backend/package/yuxi/channels/plugins/instagram/adapters/whitelist_adapter.py index b5afb12f..16405c2b 100644 --- a/backend/package/yuxi/channels/plugins/instagram/adapters/whitelist_adapter.py +++ b/backend/package/yuxi/channels/plugins/instagram/adapters/whitelist_adapter.py @@ -21,7 +21,6 @@ from yuxi.channels.contract.dtos.whitelist import ( ) from yuxi.channels.contract.errors import ( DependencyError, - Error, ValidationError, ) from yuxi.channels.contract.ports.driven.cache_port import CachePort @@ -233,8 +232,6 @@ class InstagramWhitelistAdapter: continue try: entries.append(whitelistEntryFromDict(item)) - except Error: - continue except Exception: continue return entries @@ -250,8 +247,6 @@ class InstagramWhitelistAdapter: raw = [whitelistEntryToDict(e) for e in entries] try: await self._cache.set(cache_key, raw) - except Error as exc: - raise DependencyError(dep="cache", cause=exc) from exc except Exception as exc: raise DependencyError(dep="cache", cause=exc) from exc diff --git a/backend/package/yuxi/channels/plugins/instagram/adapters/wizard_adapter.py b/backend/package/yuxi/channels/plugins/instagram/adapters/wizard_adapter.py index 6858c3ac..27806830 100644 --- a/backend/package/yuxi/channels/plugins/instagram/adapters/wizard_adapter.py +++ b/backend/package/yuxi/channels/plugins/instagram/adapters/wizard_adapter.py @@ -11,6 +11,7 @@ from __future__ import annotations +import time from typing import Any from urllib.parse import urlencode, urlparse @@ -260,12 +261,23 @@ class InstagramWizardAdapter: message="OAuth callback did not return access_token or user_id", ) + # 基于 OAuth 响应的 expires_in 计算绝对过期时间戳（秒）， + # 供 applyAccountConfig 校验（M-3）与 token 刷新调度使用 + expires_in = data.get("expires_in", 0) + if not expires_in: + raise ValidationError( + field="expires_in", + message="OAuth callback did not return expires_in", + ) + token_expires_at = time.time() + int(expires_in) + return WizardConfigPatch( step_id=_STEP_OAUTH_CALLBACK, values={ "access_token": access_token, "ig_id": str(user_id), "oauth_state": state, + "token_expires_at": token_expires_at, }, ) @@ -386,6 +398,7 @@ class InstagramWizardAdapter: access_token = data.get("access_token") user_id = data.get("user_id") + expires_in = data.get("expires_in", 0) if not access_token or not user_id: return WizardStepResult( step_id=_STEP_OAUTH_CALLBACK, @@ -394,6 +407,18 @@ class InstagramWizardAdapter: config_patch=None, ) + # 基于 OAuth 响应的 expires_in 计算绝对过期时间戳（秒）， + # 供 applyAccountConfig 校验（M-3：缺失 token_expires_at 抛 + # ValidationError）与 token 刷新调度使用 + token_expires_at = time.time() + int(expires_in) if expires_in else 0 + if not token_expires_at: + return WizardStepResult( + step_id=_STEP_OAUTH_CALLBACK, + valid=False, + errors=("token_exchange_failed: missing expires_in",), + config_patch=None, + ) + return WizardStepResult( step_id=_STEP_OAUTH_CALLBACK, valid=True, @@ -403,6 +428,7 @@ class InstagramWizardAdapter: values={ "access_token": access_token, "ig_id": str(user_id), + "token_expires_at": token_expires_at, }, ), ) diff --git a/backend/package/yuxi/channels/plugins/instagram/error_translator.py b/backend/package/yuxi/channels/plugins/instagram/error_translator.py index 89f132a5..e30ecea2 100644 --- a/backend/package/yuxi/channels/plugins/instagram/error_translator.py +++ b/backend/package/yuxi/channels/plugins/instagram/error_translator.py @@ -200,8 +200,8 @@ def _classify_http_status_error( error_subcode=error_subcode, ) if translated is not None: - # 保留精确的 Error 子类，将原始异常挂在 cause 链上 - translated.__cause__ = exc + # 原始异常通过 raise ... from exc 链由调用方挂载（参见 + # instagram_client._execute_http），此处无需手动设置 __cause__。 return translated if status_code == 401: diff --git a/backend/package/yuxi/channels/plugins/instagram/instagram_client.py b/backend/package/yuxi/channels/plugins/instagram/instagram_client.py index 10d7ea13..f6d34cf7 100644 --- a/backend/package/yuxi/channels/plugins/instagram/instagram_client.py +++ b/backend/package/yuxi/channels/plugins/instagram/instagram_client.py @@ -31,10 +31,11 @@ from ._constants import ( CHANNEL_TARGET, DEFAULT_API_VERSION, INSTAGRAM_API_DEP, + OAUTH_API_BASE, ) # Instagram Graph API 端点路径常量 -_OAUTH_AUTHORIZE_PATH = "https://api.instagram.com/oauth/authorize" +# Step 1 短期 token 交换端点（必须使用 api.instagram.com 域名） _OAUTH_TOKEN_PATH = "/oauth/access_token" _IG_TOKEN_EXCHANGE_PATH = "/access_token" # graph.instagram.com 路径 _IG_REFRESH_TOKEN_PATH = "/refresh_access_token" @@ -147,6 +148,39 @@ class InstagramClient: return DEFAULT_API_VERSION return value + async def _get_api_version_for_token_check(self) -> str: + """获取凭据轮换场景使用的 Graph API 版本。 + + ``api_version`` 在 config_schema 中声明为 account 作用域，凭据轮换 + 场景（新 token 尚未落库，无 account_id 上下文）无法从账户级读取。 + /debug_token 端点兼容各 API 版本，直接使用默认版本即可。 + """ + return DEFAULT_API_VERSION + + async def get_token_refresh_lead_days(self, account_id: str) -> int: + """读取账户级 ``token_refresh_lead_days``，未配置回退默认值。 + + 供 doctor / probeable 适配器共享，避免每个适配器独立注入 ConfigPort。 + """ + from ._constants import DEFAULT_TOKEN_REFRESH_LEAD_DAYS + + try: + cv = await self._config.get( + "token_refresh_lead_days", + scope=ConfigScope.ACCOUNT, + target=account_id, + ) + value = cv.value if hasattr(cv, "value") else None + if isinstance(value, int) and value > 0: + return value + except Exception as exc: + await self._logger.warn( + "Failed to read token_refresh_lead_days, fallback to default", + account_id=account_id, + error=str(exc), + ) + return DEFAULT_TOKEN_REFRESH_LEAD_DAYS + async def _get_access_token(self, account_id: str) -> str: """读取账户 access_token（直接从 ConfigPort 读取，Instagram User Access Token 是 OAuth 颁发的长期 token，不通过 CachePort 缓存）。 @@ -315,7 +349,9 @@ class InstagramClient: """用 OAuth 授权码交换 Long-lived Token。 两步流程（Instagram Login 路径）： - 1. POST ``graph.instagram.com/oauth/access_token`` 短期 token 交换 + 1. POST ``api.instagram.com/oauth/access_token`` 短期 token 交换 + （Step1 必须使用 api.instagram.com 域名，graph.instagram.com + 不支持此端点） 2. GET ``graph.instagram.com/access_token?grant_type=ig_exchange_token`` 短期 token 换长期 token（60 天） @@ -323,9 +359,9 @@ class InstagramClient: @return Meta 响应字典，含 ``access_token`` / ``user_id`` / ``expires_in``。 """ - ig_api_base_url = await self._get_channel_config("ig_api_base_url") - # Step 1: 短期 token 交换 - step1_url = f"{ig_api_base_url}{_OAUTH_TOKEN_PATH}" + timeout = await self._get_request_timeout_seconds() + # Step 1: 短期 token 交换（必须使用 api.instagram.com） + step1_url = f"{OAUTH_API_BASE}{_OAUTH_TOKEN_PATH}" step1_body = { "client_id": app_id, "client_secret": app_secret, @@ -334,9 +370,7 @@ class InstagramClient: "code": code, } try: - async with httpx.AsyncClient( - timeout=await self._get_request_timeout_seconds(), - ) as client: + async with httpx.AsyncClient(timeout=timeout) as client: resp1 = await client.post(step1_url, data=step1_body) resp1.raise_for_status() except httpx.HTTPError as exc: @@ -346,7 +380,7 @@ class InstagramClient: ) raise error_translator.wrap_http_exception( exc, - timeout_ms=int((await self._get_request_timeout_seconds()) * 1000), + timeout_ms=int(timeout * 1000), ) from exc data1 = self._parse_json_response(resp1) @@ -357,7 +391,8 @@ class InstagramClient: cause=ValueError("oauth step1 missing access_token"), ) - # Step 2: 短期 token 换长期 token + # Step 2: 短期 token 换长期 token（graph.instagram.com） + ig_api_base_url = await self._get_channel_config("ig_api_base_url") step2_url = f"{ig_api_base_url}{_IG_TOKEN_EXCHANGE_PATH}" step2_params = { "grant_type": "ig_exchange_token", @@ -365,9 +400,7 @@ class InstagramClient: "access_token": short_token, } try: - async with httpx.AsyncClient( - timeout=await self._get_request_timeout_seconds(), - ) as client: + async with httpx.AsyncClient(timeout=timeout) as client: resp2 = await client.get(step2_url, params=step2_params) resp2.raise_for_status() except httpx.HTTPError as exc: @@ -377,7 +410,7 @@ class InstagramClient: ) raise error_translator.wrap_http_exception( exc, - timeout_ms=int((await self._get_request_timeout_seconds()) * 1000), + timeout_ms=int(timeout * 1000), ) from exc data2 = self._parse_json_response(resp2) @@ -419,6 +452,28 @@ class InstagramClient: params={"input_token": token}, ) + async def get_debug_token_for_token(self, access_token: str) -> dict: + """调试指定 access_token（用于凭据轮换场景验证新 token）。 + + 与 ``get_debug_token`` 不同，本方法不依赖 ConfigPort 读取 token， + 直接用传入的 ``access_token`` 调用 debug_token API。用于凭据轮换 + 新 token 尚未落库时的有效性验证。 + + @param access_token: 待验证的 access_token。 + @return Meta 响应字典，含 ``data`` 字段。 + """ + api_base_url = await self._get_channel_config("api_base_url") + api_version = await self._get_api_version_for_token_check() + url = f"{api_base_url}/{api_version}{_DEBUG_TOKEN_PATH}" + headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"} + resp = await self._execute_http( + "GET", + url, + headers, + params={"input_token": access_token}, + ) + return self._parse_json_response(resp) + # ------------------------------------------------------------------ # 账户信息 # ------------------------------------------------------------------ @@ -499,8 +554,6 @@ class InstagramClient: """ ig_id = await self._get_account_value(account_id, "ig_id") payload: dict[str, Any] = {"url": media_url} - if media_type == "file": - payload["is_reusable"] = False body = { "recipient": {"id": recipient_igsid}, "message": { diff --git a/backend/package/yuxi/channels/plugins/instagram/manifest.json b/backend/package/yuxi/channels/plugins/instagram/manifest.json index 2bfeb7a9..2e3f2339 100644 --- a/backend/package/yuxi/channels/plugins/instagram/manifest.json +++ b/backend/package/yuxi/channels/plugins/instagram/manifest.json @@ -19,8 +19,8 @@ "credential_strategy": { "type": "dynamic", "acquire_method": "oauth", - "required_fields": ["access_token", "ig_id", "token_expires_at"], - "optional_fields": ["app_id", "app_secret", "page_id"], + "required_fields": ["access_token", "ig_id", "token_expires_at", "app_id", "app_secret"], + "optional_fields": [], "supports_rotation": true, "supports_revocation": true },