company-ai-platform/app/modules/feishu/service.py

from secrets import compare_digest
from typing import Any

from fastapi import HTTPException, status
from sqlalchemy.orm import Session

from app.core.constants import ActorValue
from app.core.config import get_settings
from app.modules.audit.constants import AuditAction, AuditSource
from app.modules.audit.schemas import AuditLogCreate
from app.modules.audit.service import AuditService
from app.modules.feishu.client import FeishuClient
from app.modules.feishu.constants import (
    FEISHU_EMPTY_CARD_TEXT,
    FEISHU_INVALID_TOKEN,
    FEISHU_VERIFICATION_TOKEN_REQUIRED,
    FeishuApprovalAction,
    FeishuApprovalValueKey,
    FeishuCardKey,
    FeishuPayloadKey,
    FeishuReceiveIdType,
)


class FeishuService:
    """Send Feishu messages and record audit entries for outbound actions."""

    def __init__(self, db: Session):
        self.db = db
        self.audit = AuditService(db)
        self.client = FeishuClient()

    def verify_event(self, payload: dict[str, Any]) -> None:
        settings = get_settings()
        expected = settings.feishu_verification_token
        header = payload.get(FeishuPayloadKey.HEADER) or {}
        token = payload.get(FeishuPayloadKey.TOKEN) or header.get(FeishuPayloadKey.TOKEN)
        if not expected:
            raise HTTPException(
                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                detail=FEISHU_VERIFICATION_TOKEN_REQUIRED,
            )
        if not token or not compare_digest(str(token), expected):
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail=FEISHU_INVALID_TOKEN,
            )

    def send_text(
        self,
        text: str,
        receive_id: str | None = None,
        receive_id_type: str = FeishuReceiveIdType.CHAT_ID,
        actor: str = ActorValue.SYSTEM,
    ) -> dict[str, Any]:
        result = self.client.send_text(text, receive_id, receive_id_type)
        self.audit.log(
            AuditLogCreate(
                actor=actor,
                source=AuditSource.FEISHU,
                action=AuditAction.FEISHU_SEND_TEXT,
                request_payload={
                    FeishuPayloadKey.RECEIVE_ID: receive_id,
                    FeishuPayloadKey.RECEIVE_ID_TYPE: receive_id_type,
                    FeishuPayloadKey.TEXT: text,
                },
                response_payload=result,
            )
        )
        return result

    def send_card(
        self,
        card: dict[str, Any],
        receive_id: str | None = None,
        receive_id_type: str = FeishuReceiveIdType.CHAT_ID,
        actor: str = ActorValue.SYSTEM,
    ) -> dict[str, Any]:
        result = self.client.send_card(card, receive_id, receive_id_type)
        self.audit.log(
            AuditLogCreate(
                actor=actor,
                source=AuditSource.FEISHU,
                action=AuditAction.FEISHU_SEND_CARD,
                request_payload={
                    FeishuPayloadKey.RECEIVE_ID: receive_id,
                    FeishuPayloadKey.RECEIVE_ID_TYPE: receive_id_type,
                    FeishuPayloadKey.CARD: card,
                },
                response_payload=result,
            )
        )
        return result

    @staticmethod
    def build_basic_card(title: str, lines: list[str]) -> dict[str, Any]:
        return {
            FeishuPayloadKey.CONFIG: {FeishuPayloadKey.WIDE_SCREEN_MODE: True},
            FeishuPayloadKey.HEADER: {
                FeishuPayloadKey.TITLE: {
                    FeishuPayloadKey.TAG: FeishuPayloadKey.PLAIN_TEXT,
                    FeishuPayloadKey.CONTENT: title,
                }
            },
            FeishuPayloadKey.ELEMENTS: [
                {
                    FeishuPayloadKey.TAG: FeishuPayloadKey.DIV,
                    FeishuPayloadKey.TEXT: {
                        FeishuPayloadKey.TAG: FeishuPayloadKey.LARK_MARKDOWN,
                        FeishuPayloadKey.CONTENT: "\n".join(lines) or FEISHU_EMPTY_CARD_TEXT,
                    },
                }
            ],
        }

    @staticmethod
    def build_approval_card(
        title: str,
        lines: list[str],
        ticket_id: str,
    ) -> dict[str, Any]:
        card = FeishuService.build_basic_card(title, lines)
        card[FeishuPayloadKey.ELEMENTS].append(
            {
                FeishuPayloadKey.TAG: FeishuApprovalValueKey.ACTION,
                FeishuCardKey.ACTIONS: [
                    {
                        FeishuPayloadKey.TAG: "button",
                        FeishuPayloadKey.TEXT: {
                            FeishuPayloadKey.TAG: FeishuPayloadKey.PLAIN_TEXT,
                            FeishuPayloadKey.CONTENT: "批准",
                        },
                        FeishuCardKey.BUTTON_TYPE: "primary",
                        FeishuCardKey.VALUE: {
                            FeishuApprovalValueKey.TICKET_ID: ticket_id,
                            FeishuApprovalValueKey.DECISION: FeishuApprovalAction.APPROVE,
                        },
                    },
                    {
                        FeishuPayloadKey.TAG: "button",
                        FeishuPayloadKey.TEXT: {
                            FeishuPayloadKey.TAG: FeishuPayloadKey.PLAIN_TEXT,
                            FeishuPayloadKey.CONTENT: "拒绝",
                        },
                        FeishuCardKey.BUTTON_TYPE: "danger",
                        FeishuCardKey.VALUE: {
                            FeishuApprovalValueKey.TICKET_ID: ticket_id,
                            FeishuApprovalValueKey.DECISION: FeishuApprovalAction.REJECT,
                        },
                    },
                ],
            }
        )
        return card