WPSBot/utils/message.py

"""WPS消息构造和发送工具"""
import httpx
import logging
from typing import Dict, Any, Optional
from config import WEBHOOK_URL

logger = logging.getLogger(__name__)


class MessageSender:
    """消息发送器"""
    
    def __init__(self, webhook_url: str = WEBHOOK_URL):
        """初始化消息发送器
        
        Args:
            webhook_url: Webhook URL
        """
        self.webhook_url = webhook_url
        self.client: Optional[httpx.AsyncClient] = None
    
    async def _get_client(self) -> httpx.AsyncClient:
        """获取HTTP客户端（懒加载）"""
        if self.client is None:
            self.client = httpx.AsyncClient(timeout=10.0)
        return self.client
    
    async def send_message(self, message: Dict[str, Any]) -> bool:
        """发送消息到WPS
        
        Args:
            message: 消息字典
            
        Returns:
            是否发送成功
        """
        try:
            client = await self._get_client()
            response = await client.post(self.webhook_url, json=message)
            
            if response.status_code == 200:
                logger.info(f"消息发送成功: {message.get('msgtype')}")
                return True
            else:
                logger.error(f"消息发送失败: status={response.status_code}, body={response.text}")
                return False
                
        except Exception as e:
            logger.error(f"发送消息异常: {e}", exc_info=True)
            return False
    
    async def send_text(self, content: str, at_user_id: Optional[int] = None) -> bool:
        """发送文本消息
        
        Args:
            content: 文本内容
            at_user_id: @用户ID（可选）
            
        Returns:
            是否发送成功
        """
        # 如果需要@人
        if at_user_id:
            content = f'<at user_id="{at_user_id}"></at> {content}'
        
        message = {
            "msgtype": "text",
            "text": {
                "content": content
            }
        }
        return await self.send_message(message)
    
    async def send_markdown(self, text: str) -> bool:
        """发送Markdown消息
        
        Args:
            text: Markdown文本
            
        Returns:
            是否发送成功
        """
        message = {
            "msgtype": "markdown",
            "markdown": {
                "text": text
            }
        }
        return await self.send_message(message)
    
    async def send_link(self, title: str, text: str, 
                       message_url: str = "", btn_title: str = "查看详情") -> bool:
        """发送链接消息
        
        Args:
            title: 标题
            text: 文本内容
            message_url: 跳转URL
            btn_title: 按钮文字
            
        Returns:
            是否发送成功
        """
        message = {
            "msgtype": "link",
            "link": {
                "title": title,
                "text": text,
                "messageUrl": message_url,
                "btnTitle": btn_title
            }
        }
        return await self.send_message(message)
    
    async def close(self):
        """关闭HTTP客户端"""
        if self.client:
            await self.client.aclose()
            self.client = None


# 全局消息发送器实例
_sender_instance: Optional[MessageSender] = None


def get_message_sender() -> MessageSender:
    """获取全局消息发送器实例（单例模式）"""
    global _sender_instance
    if _sender_instance is None:
        _sender_instance = MessageSender()
    return _sender_instance