feat: 迁移企业微信模块（crypto + client + handler）

2026-05-09 14:09:52 +08:00
parent 76b734b0a9
commit 52bb1f53cc
5 changed files with 437 additions and 0 deletions
@@ -0,0 +1,76 @@
+import time
+import httpx
+from typing import Optional
+
+from app.config import settings
+
+
+class WeChatClient:
+    def __init__(self):
+        self._access_token: Optional[str] = None
+        self._token_expires_at: float = 0
+
+    async def _get_access_token(self) -> Optional[str]:
+        now = time.time()
+        if self._access_token and now < self._token_expires_at:
+            return self._access_token
+
+        url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
+        params = {
+            "corpid": settings.wechat_corp_id,
+            "corpsecret": settings.wechat_secret,
+        }
+        async with httpx.AsyncClient(timeout=30) as client:
+            response = await client.get(url, params=params)
+            data = response.json()
+            if data.get("errcode") == 0:
+                self._access_token = data["access_token"]
+                self._token_expires_at = now + data.get("expires_in", 7200) - 300
+                return self._access_token
+        return None
+
+    async def send_text(self, content: str, to_user: str = "@all") -> bool:
+        return await self._send_message("text", {"content": content}, to_user)
+
+    async def send_markdown(self, content: str, to_user: str = "@all") -> bool:
+        return await self._send_message("markdown", {"content": content}, to_user)
+
+    async def send_textcard(
+        self,
+        title: str,
+        description: str,
+        url: str,
+        to_user: str = "@all",
+        btn_txt: str = "查看详情",
+    ) -> bool:
+        return await self._send_message(
+            "textcard",
+            {
+                "title": title,
+                "description": description,
+                "url": url,
+                "btntxt": btn_txt,
+            },
+            to_user,
+        )
+
+    async def _send_message(
+        self, msgtype: str, msg_data: dict, to_user: str = "@all"
+    ) -> bool:
+        token = await self._get_access_token()
+        if not token:
+            return False
+
+        url = "https://qyapi.weixin.qq.com/cgi-bin/message/send"
+        params = {"access_token": token}
+        body = {
+            "touser": to_user,
+            "msgtype": msgtype,
+            "agentid": int(settings.wechat_agent_id),
+            msgtype: msg_data,
+        }
+
+        async with httpx.AsyncClient(timeout=30) as client:
+            response = await client.post(url, params=params, json=body)
+            data = response.json()
+            return data.get("errcode") == 0