company-ai-platform/app/modules/feishu/services/reply_outbox.py

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