feat: add feishu api notification services

review_agent/notifications/feishu_token.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+
+from django.conf import settings
+from django.utils import timezone
+import httpx
+
+from review_agent.models import FeishuAccessTokenCache
+
+
+@dataclass(frozen=True)
+class FeishuTokenResult:
+    ok: bool
+    tenant_access_token: str = ""
+    error_code: str = ""
+    error_message: str = ""
+
+
+def app_id_hash(app_id: str) -> str:
+    return hashlib.sha256(app_id.encode("utf-8")).hexdigest()
+
+
+def get_tenant_access_token(*, force_refresh: bool = False) -> FeishuTokenResult:
+    app_id = getattr(settings, "FEISHU_APP_ID", "")
+    app_secret = getattr(settings, "FEISHU_APP_SECRET", "")
+    if not app_id or not app_secret:
+        return FeishuTokenResult(
+            ok=False,
+            error_code="config_missing",
+            error_message="未配置 FEISHU_APP_ID 或 FEISHU_APP_SECRET",
+        )
+
+    hashed_app_id = app_id_hash(app_id)
+    now = timezone.now()
+    cache = FeishuAccessTokenCache.objects.filter(app_id_hash=hashed_app_id).first()
+    if cache and not force_refresh and cache.is_valid(now=now):
+        return FeishuTokenResult(ok=True, tenant_access_token=cache.tenant_access_token)
+
+    try:
+        response = httpx.post(
+            getattr(settings, "FEISHU_TOKEN_API_URL"),
+            json={"app_id": app_id, "app_secret": app_secret},
+            timeout=10,
+        )
+        data = response.json()
+    except httpx.TimeoutException:
+        return _save_token_error(hashed_app_id, "timeout", "获取 tenant_access_token 超时")
+    except Exception as exc:
+        return _save_token_error(hashed_app_id, "request_error", str(exc))
+
+    if response.status_code >= 400:
+        return _save_token_error(hashed_app_id, str(response.status_code), response.text[:500])
+    if int(data.get("code") or 0) != 0:
+        return _save_token_error(hashed_app_id, str(data.get("code") or "api_error"), str(data.get("msg") or "token API 失败"))
+
+    token = str(data.get("tenant_access_token") or "")
+    expire_seconds = int(data.get("expire") or getattr(settings, "FEISHU_TENANT_TOKEN_CACHE_SECONDS", 6600))
+    if not token:
+        return _save_token_error(hashed_app_id, "token_missing", "飞书未返回 tenant_access_token")
+
+    FeishuAccessTokenCache.objects.update_or_create(
+        app_id_hash=hashed_app_id,
+        defaults={
+            "tenant_access_token": token,
+            "expires_at": now + timezone.timedelta(seconds=max(expire_seconds - 60, 60)),
+            "error_message": "",
+        },
+    )
+    return FeishuTokenResult(ok=True, tenant_access_token=token)
+
+
+def _save_token_error(app_id_hash_value: str, error_code: str, error_message: str) -> FeishuTokenResult:
+    FeishuAccessTokenCache.objects.update_or_create(
+        app_id_hash=app_id_hash_value,
+        defaults={
+            "tenant_access_token": "",
+            "expires_at": None,
+            "error_message": error_message[:1000],
+        },
+    )
+    return FeishuTokenResult(ok=False, error_code=error_code, error_message=error_message[:1000])