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

import httpx
import logging
import uuid

logger = logging.getLogger(__name__)

_WECOM_ACCESS_TOKEN: dict = {"token": None, "expires_at": 0}


def _get_access_token(corp_id: str, app_secret: str) -> str | 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
            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():
    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:
    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:
    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:
    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"]