import base64 from contextlib import contextmanager from contextvars import ContextVar from copy import deepcopy from dataclasses import dataclass, field from typing import Any, Iterator from fastapi import HTTPException, status from app.modules.feishu.constants import FEISHU_RECEIVE_ID_MISSING _IMAGE_PLACEHOLDER_PREFIX = "__feishu_reply_image__:" _reply_collector: ContextVar["FeishuReplyCollector | None"] = ContextVar( "feishu_reply_collector", default=None, ) @dataclass(slots=True) class FeishuReplyCollector: """Capture outbound Feishu operations before an inbound transaction commits.""" operations: list[dict[str, Any]] = field(default_factory=list) def capture_text( self, *, text: str, receive_id: str | None, default_receive_id: str | None, receive_id_type: str, actor: str, message_uuid: str | None, tenant_key: str | None, record_audit: bool, ) -> dict[str, Any]: target = _required_receive_id(receive_id, default_receive_id) self.operations.append( { "kind": "text", "text": text, "receive_id": target, "receive_id_type": str(receive_id_type), "actor": str(actor), "message_uuid": message_uuid, "tenant_key": tenant_key, "record_audit": bool(record_audit), } ) return {"code": 0, "queued": True} def capture_card( self, *, card: dict[str, Any], receive_id: str | None, default_receive_id: str | None, receive_id_type: str, actor: str, message_uuid: str | None, tenant_key: str | None, ) -> dict[str, Any]: target = _required_receive_id(receive_id, default_receive_id) self.operations.append( { "kind": "card", "card": deepcopy(card), "receive_id": target, "receive_id_type": str(receive_id_type), "actor": str(actor), "message_uuid": message_uuid, "tenant_key": tenant_key, "record_audit": True, } ) return {"code": 0, "queued": True} def capture_image( self, *, image: bytes, actor: str, tenant_key: str | None, ) -> dict[str, Any]: placeholder = f"{_IMAGE_PLACEHOLDER_PREFIX}{len(self.operations)}" self.operations.append( { "kind": "image", "placeholder": placeholder, "image_base64": base64.b64encode(image).decode("ascii"), "actor": str(actor), "tenant_key": tenant_key, } ) return { "code": 0, "queued": True, "data": {"image_key": placeholder}, } def as_payload( self, *, identity: tuple[str, str] | None, identity_fence_required: bool, ) -> dict[str, Any] | None: if not self.operations: return None return { "version": 1, "identity": ( {"tenant_key": identity[0], "open_id": identity[1]} if identity is not None else None ), "identity_fence_required": identity_fence_required, "operations": deepcopy(self.operations), } @contextmanager def bind_reply_outbox() -> Iterator[FeishuReplyCollector]: """Capture Feishu side effects for the current inbound command.""" collector = FeishuReplyCollector() token = _reply_collector.set(collector) try: yield collector finally: _reply_collector.reset(token) def current_reply_outbox() -> FeishuReplyCollector | None: return _reply_collector.get() def decode_image(operation: dict[str, Any]) -> bytes: encoded = operation.get("image_base64") if not isinstance(encoded, str) or not encoded: raise ValueError("Feishu reply image payload is unavailable") try: return base64.b64decode(encoded, validate=True) except (ValueError, TypeError) as exc: raise ValueError("Feishu reply image payload is invalid") from exc def resolved_message_operation(payload: dict[str, Any]) -> dict[str, Any]: operations = _operations(payload) messages = [ operation for operation in operations if operation.get("kind") in {"text", "card"} ] if len(messages) != 1: raise ValueError("Feishu reply outbox requires exactly one message") image_keys = { str(operation.get("placeholder")): str(operation.get("image_key")) for operation in operations if operation.get("kind") == "image" and operation.get("placeholder") and operation.get("image_key") } message = deepcopy(messages[0]) if message.get("kind") == "card": message["card"] = _replace_image_placeholders( message.get("card"), image_keys, ) return message def pending_image_indexes(payload: dict[str, Any]) -> list[int]: return [ index for index, operation in enumerate(_operations(payload)) if operation.get("kind") == "image" and not operation.get("image_key") ] def operations_copy(payload: dict[str, Any]) -> list[dict[str, Any]]: return deepcopy(_operations(payload)) def payload_identity(payload: Any) -> tuple[str, str] | None: if not isinstance(payload, dict): return None identity = payload.get("identity") if not isinstance(identity, dict): return None tenant_key = str(identity.get("tenant_key") or "").strip() open_id = str(identity.get("open_id") or "").strip() if not tenant_key or not open_id: return None return tenant_key, open_id def identity_fence_required(payload: Any) -> bool: return isinstance(payload, dict) and bool(payload.get("identity_fence_required")) def _operations(payload: dict[str, Any]) -> list[dict[str, Any]]: if payload.get("version") != 1: raise ValueError("Unsupported Feishu reply outbox payload version") operations = payload.get("operations") if not isinstance(operations, list) or not all( isinstance(operation, dict) for operation in operations ): raise ValueError("Feishu reply outbox operations are invalid") return operations def _replace_image_placeholders( value: Any, image_keys: dict[str, str], ) -> Any: if isinstance(value, dict): return { key: _replace_image_placeholders(item, image_keys) for key, item in value.items() } if isinstance(value, list): return [ _replace_image_placeholders(item, image_keys) for item in value ] if ( isinstance(value, str) and value.startswith(_IMAGE_PLACEHOLDER_PREFIX) ): image_key = image_keys.get(value) if not image_key: raise ValueError("Feishu reply image was not prepared") return image_key return value def _required_receive_id( receive_id: str | None, default_receive_id: str | None, ) -> str: target = str(receive_id or default_receive_id or "").strip() if not target: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=FEISHU_RECEIVE_ID_MISSING, ) return target