v6ole
18f415b363
- 新增 WECHAT_API_BASE_URL 配置项，支持企微 API 代理转发 - _send_message 增加 API 错误日志（errcode + errmsg） - _get_access_token 增加失败日志 - handle_text 不再为空，返回菜单引导提示 - 防重入触发时通知用户等待时间 - menu.py API 调用统一使用配置化的 base URL Co-Authored-By: Claude <noreply@anthropic.com>
89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
import logging
|
|
import time
|
|
|
|
import httpx
|
|
|
|
from app.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class WeChatClient:
|
|
def __init__(self):
|
|
self._access_token: str | None = None
|
|
self._token_expires_at: float = 0
|
|
|
|
async def _get_access_token(self) -> str | None:
|
|
now = time.time()
|
|
if self._access_token and now < self._token_expires_at:
|
|
return self._access_token
|
|
|
|
url = f"{settings.wechat_api_base_url}/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
|
|
logger.error(f"获取 access_token 失败: {data}")
|
|
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:
|
|
logger.error("无法获取 access_token，跳过消息发送")
|
|
return False
|
|
|
|
url = f"{settings.wechat_api_base_url}/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()
|
|
errcode = data.get("errcode")
|
|
if errcode == 0:
|
|
return True
|
|
logger.error(
|
|
f"企业微信消息发送失败: errcode={errcode} errmsg={data.get('errmsg')} "
|
|
f"msgtype={msgtype} touser={to_user}"
|
|
)
|
|
return False
|