88 lines
2.9 KiB
Python
88 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import time
|
|
|
|
from django.conf import settings
|
|
import httpx
|
|
|
|
from .feishu_token import get_tenant_access_token
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FeishuMessageResult:
|
|
ok: bool
|
|
external_message_id: str = ""
|
|
error_code: str = ""
|
|
error_message: str = ""
|
|
request_duration_ms: int | None = None
|
|
refreshed_token: bool = False
|
|
|
|
|
|
def send_personal_message(
|
|
*,
|
|
tenant_access_token: str,
|
|
receive_id_type: str,
|
|
payload: dict,
|
|
retry_on_token_expired: bool = True,
|
|
) -> FeishuMessageResult:
|
|
start = time.monotonic()
|
|
try:
|
|
response = httpx.post(
|
|
getattr(settings, "FEISHU_MESSAGE_API_URL"),
|
|
params={"receive_id_type": receive_id_type},
|
|
json=payload,
|
|
headers={"Authorization": f"Bearer {tenant_access_token}"},
|
|
timeout=10,
|
|
)
|
|
duration_ms = int((time.monotonic() - start) * 1000)
|
|
data = response.json()
|
|
except httpx.TimeoutException:
|
|
return FeishuMessageResult(ok=False, error_code="timeout", error_message="发送飞书消息超时")
|
|
except Exception as exc:
|
|
return FeishuMessageResult(ok=False, error_code="request_error", error_message=str(exc))
|
|
|
|
if response.status_code >= 400:
|
|
return FeishuMessageResult(
|
|
ok=False,
|
|
error_code=str(response.status_code),
|
|
error_message=response.text[:500],
|
|
request_duration_ms=duration_ms,
|
|
)
|
|
|
|
code = int(data.get("code") or 0)
|
|
if code == 0:
|
|
message_id = str((data.get("data") or {}).get("message_id") or "")
|
|
return FeishuMessageResult(ok=True, external_message_id=message_id, request_duration_ms=duration_ms)
|
|
|
|
if retry_on_token_expired and code in {99991663, 99991664, 99991668, 99991669}:
|
|
token_result = get_tenant_access_token(force_refresh=True)
|
|
if token_result.ok:
|
|
retry_result = send_personal_message(
|
|
tenant_access_token=token_result.tenant_access_token,
|
|
receive_id_type=receive_id_type,
|
|
payload=payload,
|
|
retry_on_token_expired=False,
|
|
)
|
|
return FeishuMessageResult(
|
|
ok=retry_result.ok,
|
|
external_message_id=retry_result.external_message_id,
|
|
error_code=retry_result.error_code,
|
|
error_message=retry_result.error_message,
|
|
request_duration_ms=retry_result.request_duration_ms,
|
|
refreshed_token=True,
|
|
)
|
|
return FeishuMessageResult(
|
|
ok=False,
|
|
error_code=token_result.error_code,
|
|
error_message=token_result.error_message,
|
|
request_duration_ms=duration_ms,
|
|
)
|
|
|
|
return FeishuMessageResult(
|
|
ok=False,
|
|
error_code=str(code or "api_error"),
|
|
error_message=str(data.get("msg") or "飞书消息 API 失败"),
|
|
request_duration_ms=duration_ms,
|
|
)
|