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

import json
import time
from typing import Any

import httpx
from fastapi import HTTPException, status

from app.core.constants import BEARER_TOKEN_TEMPLATE, HttpHeader
from app.core.config import get_settings
from app.modules.feishu.constants import (
    FEISHU_AUTH_MISSING,
    FEISHU_DEFAULT_TOKEN_EXPIRE_SECONDS,
    FEISHU_MESSAGE_PATH,
    FEISHU_IMAGE_PATH,
    FEISHU_RECEIVE_ID_MISSING,
    FEISHU_SUCCESS_CODE,
    FEISHU_TENANT_TOKEN_PATH,
    FEISHU_TOKEN_EXPIRE_SAFETY_SECONDS,
    FeishuMessageType,
    FeishuPayloadKey,
    FeishuReceiveIdType,
)


class FeishuClient:
    """Small Feishu Open Platform client for tenant token and message APIs."""

    def __init__(self) -> None:
        self.settings = get_settings()
        self._tenant_access_token: str | None = None
        self._token_expires_at: float = 0

    def _is_configured(self) -> bool:
        return bool(self.settings.feishu_app_id and self.settings.feishu_app_secret)

    def _get_tenant_access_token(self) -> str:
        if not self._is_configured():
            raise HTTPException(
                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                detail=FEISHU_AUTH_MISSING,
            )
        if self._tenant_access_token and time.time() < self._token_expires_at:
            return self._tenant_access_token

        url = f"{self.settings.feishu_base_url}{FEISHU_TENANT_TOKEN_PATH}"
        payload = {
            FeishuPayloadKey.APP_ID: self.settings.feishu_app_id,
            FeishuPayloadKey.APP_SECRET: self.settings.feishu_app_secret,
        }
        with httpx.Client(timeout=20) as client:
            response = client.post(url, json=payload)
            response.raise_for_status()
            data = response.json()
        if data.get(FeishuPayloadKey.CODE) != FEISHU_SUCCESS_CODE:
            raise HTTPException(
                status_code=status.HTTP_502_BAD_GATEWAY,
                detail={FeishuPayloadKey.FEISHU_ERROR: data},
            )
        self._tenant_access_token = data[FeishuPayloadKey.TENANT_ACCESS_TOKEN]
        expire_seconds = int(
            data.get(FeishuPayloadKey.EXPIRE, FEISHU_DEFAULT_TOKEN_EXPIRE_SECONDS)
        )
        self._token_expires_at = time.time() + expire_seconds - FEISHU_TOKEN_EXPIRE_SAFETY_SECONDS
        return self._tenant_access_token

    def send_message(
        self,
        receive_id: str,
        receive_id_type: str,
        msg_type: str,
        content: dict[str, Any],
    ) -> dict[str, Any]:
        token = self._get_tenant_access_token()
        url = f"{self.settings.feishu_base_url}{FEISHU_MESSAGE_PATH}"
        headers = {HttpHeader.AUTHORIZATION: BEARER_TOKEN_TEMPLATE.format(token=token)}
        params = {FeishuPayloadKey.RECEIVE_ID_TYPE: receive_id_type}
        payload = {
            FeishuPayloadKey.RECEIVE_ID: receive_id,
            FeishuPayloadKey.MESSAGE_TYPE: msg_type,
            FeishuPayloadKey.CONTENT: json.dumps(content, ensure_ascii=False),
        }
        with httpx.Client(timeout=20) as client:
            response = client.post(url, headers=headers, params=params, json=payload)
            response.raise_for_status()
            data = response.json()
        return data

    def send_text(
        self,
        text: str,
        receive_id: str | None = None,
        receive_id_type: str = FeishuReceiveIdType.CHAT_ID,
    ) -> dict:
        chat_id = receive_id or self.settings.feishu_default_chat_id
        if not chat_id:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=FEISHU_RECEIVE_ID_MISSING,
            )
        return self.send_message(
            chat_id,
            receive_id_type,
            FeishuMessageType.TEXT,
            {FeishuPayloadKey.TEXT: text},
        )

    def upload_image(
        self,
        image: bytes,
        filename: str = "lifecycle-report.png",
    ) -> dict[str, Any]:
        token = self._get_tenant_access_token()
        url = f"{self.settings.feishu_base_url}{FEISHU_IMAGE_PATH}"
        headers = {HttpHeader.AUTHORIZATION: BEARER_TOKEN_TEMPLATE.format(token=token)}
        with httpx.Client(timeout=30) as client:
            response = client.post(
                url,
                headers=headers,
                data={FeishuPayloadKey.IMAGE_TYPE: FeishuPayloadKey.MESSAGE},
                files={
                    FeishuPayloadKey.IMAGE: (filename, image, "image/png"),
                },
            )
            response.raise_for_status()
            data = response.json()
        if data.get(FeishuPayloadKey.CODE) != FEISHU_SUCCESS_CODE:
            raise HTTPException(
                status_code=status.HTTP_502_BAD_GATEWAY,
                detail={FeishuPayloadKey.FEISHU_ERROR: data},
            )
        return data

    def send_card(
        self,
        card: dict[str, Any],
        receive_id: str | None = None,
        receive_id_type: str = FeishuReceiveIdType.CHAT_ID,
    ) -> dict:
        chat_id = receive_id or self.settings.feishu_default_chat_id
        if not chat_id:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=FEISHU_RECEIVE_ID_MISSING,
            )
        return self.send_message(chat_id, receive_id_type, FeishuMessageType.INTERACTIVE, card)