mildoc/mildoc_wxkf/wecom_api.py

"""
企业微信API调用模块
支持客服消息的读取和发送功能

作者：开发工程师
日期：2025年01月
"""

import time
import logging
import requests
from typing import Dict, Optional
from config import Config

logger = logging.getLogger(__name__)

class WeComAPI:
    """企业微信API调用类"""
    
    def __init__(self):
        self.corp_id = Config.CORP_ID
        self.app_secret = Config.APP_SECRET
        self.base_url = "https://qyapi.weixin.qq.com/cgi-bin"
        self._kf_access_token = None
        self._kf_token_expires_at = 0
    
    
    def get_kf_access_token(self) -> Optional[str]:
        """获取客服专用access_token"""
        # 检查token是否过期
        if self._kf_access_token and time.time() < self._kf_token_expires_at:
            return self._kf_access_token
        
        # 配置客服密钥
        secret = self.app_secret
        if not secret:
            logger.error("缺少APP_SECRET配置，无法获取客服access_token")
            return None
        
        url = f"{self.base_url}/gettoken"
        params = {
            'corpid': self.corp_id,
            'corpsecret': secret
        }
        
        try:
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()
            data = response.json()
            
            if data.get('errcode') == 0:
                self._kf_access_token = data.get('access_token')
                expires_in = data.get('expires_in', 7200)
                self._kf_token_expires_at = time.time() + expires_in - 300  # 提前5分钟过期
                logger.info(f"获取客服access_token成功，有效期: {expires_in}秒")
                return self._kf_access_token
            else:
                logger.error(f"获取客服access_token失败: {data.get('errmsg')}")
                return None
                
        except Exception as e:
            logger.error(f"获取客服access_token异常: {e}")
            return None
    
    def sync_kf_messages(self, token: str, open_kfid: str = "", cursor: str = "", limit: int = 1000) -> Optional[Dict]:
        """
        读取客服消息
        
        Args:
            token: 回调事件返回的token字段
            open_kfid: 指定拉取某个客服账号的消息
            cursor: 上一次调用时返回的next_cursor
            limit: 期望请求的数据量，默认1000
            
        Returns:
            消息列表数据
        """
        access_token = self.get_kf_access_token()
        if not access_token:
            return None
        
        url = f"{self.base_url}/kf/sync_msg"
        params = {'access_token': access_token}
        
        data = {
            'token': token,
            'limit': limit,
            'voice_format': 0  # 0-Amr 1-Silk
        }
        
        if cursor:
            data['cursor'] = cursor
        if open_kfid:
            data['open_kfid'] = open_kfid
        
        try:
            response = requests.post(url, params=params, json=data, timeout=10)
            response.raise_for_status()
            result = response.json()
            
            if result.get('errcode') == 0:
                logger.info(f"读取客服消息成功，获取到 {len(result.get('msg_list', []))} 条消息")
                return result
            else:
                logger.error(f"读取客服消息失败: {result.get('errmsg')}")
                return None
                
        except Exception as e:
            logger.error(f"读取客服消息异常: {e}")
            return None
    
    def send_kf_message(self, touser: str, open_kfid: str, msgtype: str, content: Dict, msgid: str = None) -> Optional[Dict]:
        """
        发送客服消息
        
        Args:
            touser: 接收消息的客户UserID
            open_kfid: 发送消息的客服帐号ID
            msgtype: 消息类型 (text, image, voice, video, file, link, miniprogram, msgmenu, location)
            content: 消息内容
            msgid: 指定消息ID
            
        Returns:
            发送结果
        """
        access_token = self.get_kf_access_token()
        if not access_token:
            return None
        
        url = f"{self.base_url}/kf/send_msg"
        params = {'access_token': access_token}
        
        data = {
            'touser': touser,
            'open_kfid': open_kfid,
            'msgtype': msgtype
        }
        
        # 添加消息内容
        data[msgtype] = content
        
        if msgid:
            data['msgid'] = msgid
        
        try:
            response = requests.post(url, params=params, json=data, timeout=10)
            response.raise_for_status()
            result = response.json()
            
            if result.get('errcode') == 0:
                logger.info(f"发送客服消息成功，msgid: {result.get('msgid')}")
                return result
            else:
                logger.error(f"发送客服消息失败: {result.get('errmsg')}")
                return None
                
        except Exception as e:
            logger.error(f"发送客服消息异常: {e}")
            return None
    
    def send_kf_text_message(self, touser: str, open_kfid: str, content: str, msgid: str = None) -> Optional[Dict]:
        """发送文本消息"""
        text_content = {'content': content}
        return self.send_kf_message(touser, open_kfid, 'text', text_content, msgid)
    
    def send_kf_image_message(self, touser: str, open_kfid: str, media_id: str, msgid: str = None) -> Optional[Dict]:
        """发送图片消息"""
        image_content = {'media_id': media_id}
        return self.send_kf_message(touser, open_kfid, 'image', image_content, msgid)
    
    def send_kf_link_message(self, touser: str, open_kfid: str, title: str, desc: str, url: str, thumb_media_id: str, msgid: str = None) -> Optional[Dict]:
        """发送链接消息"""
        link_content = {
            'title': title,
            'desc': desc,
            'url': url,
            'thumb_media_id': thumb_media_id
        }
        return self.send_kf_message(touser, open_kfid, 'link', link_content, msgid)
        
    def send_event_response_message(self, code: str, content: str) -> Optional[Dict]:
        """
        发送事件响应消息（如欢迎语、结束语）
        
        Args:
            code: 事件响应码
            msgtype: 消息类型 (text, msgmenu)
            content: 消息内容
            
        Returns:
            发送结果
        """
        access_token = self.get_kf_access_token()
        if not access_token:
            return None
        
        url = f"{self.base_url}/kf/send_msg_on_event"
        params = {'access_token': access_token}
        
        data = {
            'code': code,
            'msgtype': 'text',
            'text': {'content': content}
        }
        
        try:
            response = requests.post(url, params=params, json=data, timeout=10)
            response.raise_for_status()
            result = response.json()
            
            if result.get('errcode') == 0:
                logger.info(f"发送事件响应消息成功，code: {code}")
                return result
            else:
                logger.error(f"发送事件响应消息失败: {result.get('errmsg')}")
                return None
                
        except Exception as e:
            logger.error(f"发送事件响应消息异常: {e}")
            return None
    
    def get_service_session_state(self, external_userid: str, open_kfid: str) -> Optional[Dict]:
        """
        获取会话状态
        
        Args:
            open_kfid: 客服帐号ID
            external_userid: 用户的external_userid
            
        Returns:
            会话状态数据，包含以下字段:
            - service_state: 会话状态 (0: 未接待, 1: 由智能助手接待, 2: 接待池等待中, 3: 人工接待, 4: 用户已确认接待结束)
            - service_userid: 接待客服的userid (当 service_state 为3时返回)
            - service_session_id: 会话ID
        """
        access_token = self.get_kf_access_token()
        if not access_token:
            return None
        
        url = f"{self.base_url}/kf/service_state/get"
        params = {'access_token': access_token}
        
        data = {
            'open_kfid': open_kfid,
            'external_userid': external_userid
        }
        
        try:
            response = requests.post(url, params=params, json=data, timeout=10)
            response.raise_for_status()
            result = response.json()
            
            if result.get('errcode') == 0:
                logger.info(f"获取会话状态成功，状态: {result.get('service_state')}")
                return result
            else:
                logger.error(f"获取会话状态失败: {result.get('errmsg')}")
                return None
                
        except Exception as e:
            logger.error(f"获取会话状态异常: {e}")
            return None
    
    def update_service_session_state(self, external_userid: str, open_kfid: str, service_state: int, service_userid: str = None) -> Optional[Dict]:
        """
        变更会话状态
        
        Args:
            open_kfid: 客服帐号ID
            external_userid: 用户的external_userid
            service_state: 要变更的会话状态 (0: 结束接待, 1: 开启由智能助手接待, 2: 进入接待池由人工接待)
            service_userid: 接待客服的userid (当service_state为3时必填)
            
        Returns:
            变更结果
        """
        access_token = self.get_kf_access_token()
        if not access_token:
            return None
        
        url = f"{self.base_url}/kf/service_state/trans"
        params = {'access_token': access_token}
        
        data = {
            'open_kfid': open_kfid,
            'external_userid': external_userid,
            'service_state': service_state
        }
        
        # 当状态为3（开启由人工接待）时，必须指定接待人
        if service_state == 3:
            if not service_userid:
                logger.error("变更会话状态失败: service_state为3时必须指定service_userid")
                return None
            data['service_userid'] = service_userid
        
        try:

            logger.info(f"变更会话状态请求: {data}")
            logger.info(f"变更会话状态请求: {url}")
            logger.info(f"变更会话状态请求: {params}")

            response = requests.post(url, params=params, json=data, timeout=10)
            response.raise_for_status()
            result = response.json()
            
            if result.get('errcode') == 0:
                logger.info(f"变更会话状态成功，状态: {service_state}")
                return result
            else:
                logger.error(f"变更会话状态失败: {result.get('errmsg')}")
                return None
                
        except Exception as e:
            logger.error(f"变更会话状态异常: {e}")
            return None
    


# 全局API实例
wecom_api = WeComAPI()