hg-agents/backend/agentscope_integration/tools/wecom_tools.py

"""企业微信工具模块。

提供企业微信 API 的封装，支持发送消息、查询用户信息、群消息发送等功能。
包含 access_token 的自动获取和缓存机制。
"""
import httpx
import logging

logger = logging.getLogger(__name__)  # 当前模块的日志记录器

_WECOM_ACCESS_TOKEN: dict = {"token": None, "expires_at": 0}  # 企业微信 access_token 缓存


def _get_access_token(corp_id: str, app_secret: str) -> str | None:
    """获取或刷新企业微信 access_token。

    优先使用缓存中的 token，如果已过期或不存在则重新请求。
    Token 过期前 5 分钟会自动刷新。

    Args:
        corp_id: 企业微信 CorpID。
        app_secret: 企业微信应用 Secret。

    Returns:
        str | None: 有效的 access_token，获取失败返回 None。
    """
    if not corp_id or not app_secret:
        logger.warning("WECOM_CORP_ID 或 WECOM_APP_SECRET 未配置，无法发送企微通知")
        return None

    import time
    now = time.time()
    if _WECOM_ACCESS_TOKEN["token"] and _WECOM_ACCESS_TOKEN["expires_at"] > now + 60:
        return _WECOM_ACCESS_TOKEN["token"]

    try:
        url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corp_id}&corpsecret={app_secret}"
        resp = httpx.get(url, timeout=10)
        data = resp.json()
        if data.get("errcode") == 0:
            _WECOM_ACCESS_TOKEN["token"] = data["access_token"]
            _WECOM_ACCESS_TOKEN["expires_at"] = now + data.get("expires_in", 7200) - 300  # 提前 5 分钟过期
            return _WECOM_ACCESS_TOKEN["token"]
        else:
            logger.error(f"获取企微 token 失败: {data}")
            return None
    except Exception as e:
        logger.error(f"请求企微 token 异常: {e}")
        return None


def _get_config():
    """从全局配置中获取企业微信 CorpID 和 AppSecret。

    Returns:
        tuple: (corp_id, app_secret) 元组。
    """
    from config import settings
    return settings.WECOM_CORP_ID, settings.WECOM_APP_SECRET


def send_notification(to_user: str, message: str, msg_type: str = "text") -> str:
    """向指定企业微信用户发送通知消息。

    支持文本和文本卡片两种消息类型。

    Args:
        to_user: 接收消息的企业微信用户 ID。
        message: 消息内容。
        msg_type: 消息类型，支持 text/textcard。

    Returns:
        str: 发送结果描述信息。
    """
    corp_id, app_secret = _get_config()
    token = _get_access_token(corp_id, app_secret)
    if not token:
        return "企业微信通知发送失败: 未配置 WECOM_CORP_ID/WECOM_APP_SECRET 或获取 access_token 失败"

    try:
        url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"

        if msg_type == "textcard":
            body = {
                "touser": to_user,
                "msgtype": "textcard",
                "agentid": 0,
                "textcard": {
                    "title": "企业AI平台通知",
                    "description": message,
                    "url": "",
                },
            }
        else:
            body = {
                "touser": to_user,
                "msgtype": "text",
                "agentid": 0,
                "text": {"content": message},
            }

        resp = httpx.post(url, json=body, timeout=10)
        data = resp.json()
        if data.get("errcode") == 0:
            return f"企业微信通知已成功发送至 {to_user}"
        else:
            logger.error(f"企微消息发送失败: {data}")
            return f"企业微信通知发送失败: {data.get('errmsg', '未知错误')}"
    except Exception as e:
        logger.error(f"企微消息发送异常: {e}")
        return f"企业微信通知发送失败: {e}"


def query_wecom_user(user_id: str) -> str:
    """查询企业微信用户的详细信息。

    Args:
        user_id: 企业微信用户 ID。

    Returns:
        str: 用户信息描述或错误信息。
    """
    corp_id, app_secret = _get_config()
    token = _get_access_token(corp_id, app_secret)
    if not token:
        return "企业微信用户查询失败: 未配置或 access_token 获取失败"

    try:
        url = f"https://qyapi.weixin.qq.com/cgi-bin/user/get?access_token={token}&userid={user_id}"
        resp = httpx.get(url, timeout=10)
        data = resp.json()
        if data.get("errcode") == 0:
            user = data
            return f"用户 {user.get('name', user_id)} - 部门: {user.get('department', [])} - 职位: {user.get('position', '未知')}"
        else:
            return f"企业微信用户查询失败: {data.get('errmsg', '未知错误')}"
    except Exception as e:
        return f"企业微信用户查询失败: {e}"


def send_wecom_group_message(message: str, group_id: str | None = None, msg_type: str = "text") -> str:
    """向企业微信群发送消息。

    支持文本和 Markdown 两种消息格式。

    Args:
        message: 消息内容。
        group_id: 企业微信群聊 ID。
        msg_type: 消息类型，支持 text/markdown。

    Returns:
        str: 发送结果描述信息。
    """
    corp_id, app_secret = _get_config()
    token = _get_access_token(corp_id, app_secret)
    if not token:
        return "企业微信群消息发送失败: 未配置或 access_token 获取失败"

    try:
        url = f"https://qyapi.weixin.qq.com/cgi-bin/appchat/send?access_token={token}"

        body = {
            "chatid": group_id,
            "msgtype": msg_type,
        }
        if msg_type == "text":
            body["text"] = {"content": message}
        elif msg_type == "markdown":
            body["markdown"] = {"content": message}

        resp = httpx.post(url, json=body, timeout=10)
        data = resp.json()
        if data.get("errcode") == 0:
            return f"企业微信群消息已成功发送至群 {group_id}"
        else:
            return f"企业微信群消息发送失败: {data.get('errmsg', '未知错误')}"
    except Exception as e:
        return f"企业微信群消息发送失败: {e}"


__all__ = ["send_notification", "query_wecom_user", "send_wecom_group_message"]