import logging import time from typing import Dict from wecom_api import wecom_api from config import Config from cursor_manager import cursor_manager from rag_service import get_rag_service logger = logging.getLogger(__name__) WELCOME_MESSAGE = '''🎉 欢迎使用微信客服！ 我是您的专属客服助手，很高兴为您服务！ 🔹 有任何问题随时咨询 🔹 输入"帮助"查看功能菜单 🔹 我们提供7×24小时服务 请问有什么可以帮助您的吗？''' class KfMessageHandler: """微信客服消息处理器""" def __init__(self): self.processed_messages = set() # 内存中的消息去重缓存 def process_kf_event(self, token: str, open_kfid: str) -> bool: """ 处理客服事件，拉取并处理消息（优化版本） Args: token: 回调事件中的token open_kfid: 客服账号ID Returns: 处理是否成功 """ try: # 从持久化存储获取该客服账号的最后cursor cursor = cursor_manager.get_cursor(open_kfid) # 确定拉取限制：如果没有cursor（首次或丢失），只拉取最近1条消息作为保底 limit = 1 if not cursor else 100 logger.info(f"开始拉取客服消息 - OpenKfId: {open_kfid}, Cursor: {'有' if cursor else '无'}, Limit: {limit}") # 拉取消息 result = wecom_api.sync_kf_messages(token, open_kfid, cursor, limit) if not result: logger.error("拉取客服消息失败") return False msg_list = result.get('msg_list', []) next_cursor = result.get('next_cursor', '') has_more = result.get('has_more', 0) logger.info(f"拉取到 {len(msg_list)} 条客服消息，next_cursor: {'有' if next_cursor else '无'}, has_more: {has_more}") # 处理每条消息（带去重） processed_count = 0 new_messages = 0 for msg in msg_list: msgid = msg.get('msgid', '') # 检查消息是否已处理（双重检查：内存缓存 + 数据库） if msgid in self.processed_messages or cursor_manager.is_message_processed(msgid): logger.debug(f"消息已处理，跳过: {msgid}") continue # 处理新消息 if self.process_single_kf_message(msg): processed_count += 1 new_messages += 1 # 添加到内存缓存 self.processed_messages.add(msgid) # 限制内存缓存大小 if len(self.processed_messages) > 1000: # 清理一半的缓存 self.processed_messages = set(list(self.processed_messages)[500:]) # 保存cursor到持久化存储 if next_cursor: cursor_manager.save_cursor(open_kfid, next_cursor, new_messages) logger.info(f"已保存新cursor - OpenKfId: {open_kfid}, 新消息数: {new_messages}") # 如果还有更多消息，继续拉取（但要避免无限循环） if has_more == 1 and new_messages > 0: logger.info("还有更多消息，继续拉取...") return self.process_kf_event(token, open_kfid) logger.info(f"客服消息处理完成 - 总拉取: {len(msg_list)}, 新处理: {processed_count}") return True except Exception as e: logger.error(f"处理客服事件异常: {e}") return False def process_single_kf_message(self, msg: Dict) -> bool: """ 处理单条客服消息（优化版本） Args: msg: 消息数据 Returns: 处理是否成功 """ try: msgid = msg.get('msgid', '') open_kfid = msg.get('open_kfid', '') external_userid = msg.get('external_userid', '') send_time = msg.get('send_time', 0) origin = msg.get('origin', 0) # 3-微信客户发送 4-系统推送事件 5-接待人员发送 servicer_userid = msg.get('servicer_userid', '') msgtype = msg.get('msgtype', '') logger.info(f"处理客服消息 - msgid: {msgid}, 类型: {msgtype}, 来源: {origin}, 客户: {external_userid}") # 检查消息时效性（仅对客户发送的消息进行时效检查） current_time = int(time.time()) message_age_seconds = current_time - send_time message_age_minutes = message_age_seconds / 60 # 标记消息为已处理 reply_sent = False # 只处理微信客户发送的消息 if origin == 3 and external_userid: # 检查消息是否在10分钟内发送 if message_age_minutes <= 10: reply_sent = self.handle_customer_message(msg) logger.info(f"消息已处理，消息年龄：{message_age_minutes:.1f}分钟") else: logger.info(f"消息超过10分钟时效限制，不予回复 - msgid: {msgid}, " f"消息时间: {send_time}, 当前时间: {current_time}, " f"消息年龄: {message_age_minutes:.1f}分钟") # 虽然不回复，但仍然标记为已处理，避免重复处理 reply_sent = True elif origin == 4: # 处理系统事件 self.handle_system_event(msg) elif origin == 5: # 接待人员发送的消息，记录日志 logger.info(f"接待人员 {servicer_userid} 发送消息: {msgtype}") # 标记消息为已处理 cursor_manager.mark_message_processed( msgid, open_kfid, external_userid, msgtype, origin, reply_sent ) return True except Exception as e: logger.error(f"处理单条客服消息异常: {e}") return False def handle_customer_message(self, msg: Dict) -> bool: """ 处理客户发送的消息 Returns: 是否发送了回复 """ try: msgtype = msg.get('msgtype', '') external_userid = msg.get('external_userid', '') open_kfid = msg.get('open_kfid', '') reply_sent = False service_state = self.get_service_session_state(external_userid, open_kfid) if (service_state != 1 and service_state != 0): # 如果不是智能助手接待状态，则不处理 logger.info(f"非智能助手接待状态，不处理消息 - open_kfid: {open_kfid}, 客户: {external_userid}") return False if msgtype == 'text': # 处理文本消息，使用智能回复生成回复内容 text_data = msg.get('text', {}) content = text_data.get('content', '') logger.info(f"收到客户文本消息: {content}") if '转人工' in content: # 转人工 reply_content = "🤖 收到您的转人工请求，我会尽快转接人工服务。" reply_sent = self.send_kf_reply(external_userid, open_kfid, reply_content) self.update_service_session_state_to_service_pool(external_userid, open_kfid) else: # 生成智能回复 reply_content = self.get_smart_reply(content) # 发送回复 if reply_content: reply_sent = self.send_kf_reply(external_userid, open_kfid, reply_content) elif msgtype in ['image', 'voice', 'video', 'file']: # 处理多媒体消息，暂不处理，仅作简单回复收到消息 replies = { 'image': '''📷 收到您发送的图片，请简单描述一下图片内容，我能更好地为您服务！''', 'voice': '''🎤 收到您的语音消息，感谢您的留言！''', 'video': '''🎬 收到您发送的视频，感谢分享！''', 'file': '''📎 收到您发送的文件，我会尽快查看处理。''' } reply_content = replies.get(msgtype, '收到您的消息，感谢分享！我会尽快为您处理。') if reply_content: reply_sent = self.send_kf_reply(external_userid, open_kfid, reply_content) elif msgtype == 'location': # 处理位置消息，暂不处理，仅作简单回复收到消息 location = msg.get('location', {}) name = location.get('name', '') address = location.get('address', '') reply_content = f"📍 收到您分享的位置：{name}\n地址：{address}\n\n感谢分享！如需导航或周边服务，请告诉我具体需求。" reply_sent = self.send_kf_reply(external_userid, open_kfid, reply_content) elif msgtype == 'link': # 处理链接消息，暂不处理，仅作简单回复收到消息 link = msg.get('link', {}) title = link.get('title', '') reply_content = f"🔗 收到您分享的链接：{title}\n\n感谢分享！我会查看相关内容。" reply_sent = self.send_kf_reply(external_userid, open_kfid, reply_content) elif msgtype == 'business_card': # 处理名片消息，暂不处理，仅作简单回复收到消息 reply_content = "👤 收到您的名片，感谢分享联系方式！\n\n如有业务合作需求，我们会及时与您联系。" reply_sent = self.send_kf_reply(external_userid, open_kfid, reply_content) elif msgtype == 'miniprogram': # 处理小程序消息，暂不处理，仅作简单回复收到消息 mini = msg.get('miniprogram', {}) title = mini.get('title', '') reply_content = f"📱 收到您分享的小程序：{title}\n\n感谢分享！我会查看相关功能。" reply_sent = self.send_kf_reply(external_userid, open_kfid, reply_content) elif msgtype == 'channels_shop_product': # 处理视频号商品消息，暂不处理，仅作简单回复收到消息 product = msg.get('channels_shop_product', {}) title = product.get('title', '') price = product.get('sales_price', '') reply_content = f"🛍️ 收到您关注的商品：{title}\n价格：{price}分\n\n如需了解更多商品信息或购买咨询，请告诉我！" reply_sent = self.send_kf_reply(external_userid, open_kfid, reply_content) elif msgtype == 'channels_shop_order': # 处理视频号订单消息，暂不处理，仅作简单回复收到消息 order = msg.get('channels_shop_order', {}) order_id = order.get('order_id', '') state = order.get('state', '') reply_content = f"📦 收到您的订单信息：{order_id}\n状态：{state}\n\n如需查询订单详情或有其他问题，请随时联系我！" reply_sent = self.send_kf_reply(external_userid, open_kfid, reply_content) return reply_sent except Exception as e: logger.error(f"处理客户消息异常: {e}") return False def handle_system_event(self, msg: Dict) -> None: """处理系统事件""" try: event_data = msg.get('event', {}) event_type = event_data.get('event_type', '') logger.info(f"处理系统事件: {event_type}") if event_type == 'enter_session': # 用户进入会话事件 self.handle_enter_session_event(event_data) elif event_type == 'msg_send_fail': # 消息发送失败事件 self.handle_send_fail_event(event_data) elif event_type == 'servicer_status_change': # 接待人员状态变更事件 self.handle_servicer_status_change_event(event_data) elif event_type == 'session_status_change': # 会话状态变更事件 self.handle_session_change_event(event_data) elif event_type == 'user_recall_msg': # 用户撤回消息事件 self.handle_user_recall_event(event_data) elif event_type == 'servicer_recall_msg': # 接待人员撤回消息事件 self.handle_servicer_recall_event(event_data) except Exception as e: logger.error(f"处理系统事件异常: {e}") def handle_enter_session_event(self, event_data: Dict) -> None: """处理用户进入会话事件""" try: external_userid = event_data.get('external_userid', '') open_kfid = event_data.get('open_kfid', '') scene = event_data.get('scene', '') welcome_code = event_data.get('welcome_code', '') logger.info(f"用户 {external_userid} 进入会话 - 场景: {scene} - 欢迎码: {welcome_code} - open_kfid: {open_kfid}") # 发送欢迎消息 welcome_msg = WELCOME_MESSAGE # 使用事件响应消息接口发送欢迎语 wecom_api.send_event_response_message(welcome_code, welcome_msg) except Exception as e: logger.error(f"处理进入会话事件异常: {e}") def handle_send_fail_event(self, event_data: Dict) -> None: """处理消息发送失败事件，暂不处理，仅做日志记录""" try: external_userid = event_data.get('external_userid', '') fail_msgid = event_data.get('fail_msgid', '') fail_type = event_data.get('fail_type', 0) fail_reasons = { 0: "未知原因", 1: "客服账号已删除", 2: "应用已关闭", 4: "会话已过期，超过48小时", 5: "会话已关闭", 6: "超过5条限制", 7: "未绑定视频号", 8: "主体未验证", 9: "未绑定视频号且主体未验证", 10: "用户拒收" } reason = fail_reasons.get(fail_type, "未知原因") logger.warning(f"消息发送失败 - 用户: {external_userid}, 消息ID: {fail_msgid}, 原因: {reason}") except Exception as e: logger.error(f"处理发送失败事件异常: {e}") def handle_servicer_status_change_event(self, event_data: Dict) -> None: """处理接待人员状态变更事件，暂不处理，仅做日志记录""" try: servicer_userid = event_data.get('servicer_userid', '') status = event_data.get('status', 0) open_kfid = event_data.get('open_kfid', '') logger.info(f"接待人员 {servicer_userid} 状态变更: {status}, open_kfid: {open_kfid}") except Exception as e: logger.error(f"处理接待人员状态变更事件异常: {e}") def handle_session_change_event(self, event_data: Dict) -> None: """处理会话状态变更，回复欢迎语或者结束语""" try: external_userid = event_data.get('external_userid', '') change_type = event_data.get('change_type', 0) msg_code = event_data.get('msg_code', '') change_types = { 1: "从接待池接入会话", 2: "转接会话", 3: "结束会话", 4: "重新接入已结束/已转接会话" } change_text = change_types.get(change_type, "未知变更") logger.info(f"会话状态变更 - 用户: {external_userid}, 变更: {change_text}") # 如果有消息码，可以发送相应的回复语或结束语 if msg_code: if change_type == 1: # 接入会话 response_msg = "您好！我是您的专属客服，很高兴为您服务！有什么可以帮助您的吗？" elif change_type == 3: # 结束会话 response_msg = "感谢您的咨询！如有其他问题，欢迎随时联系我们。祝您生活愉快！" else: response_msg = None if response_msg: wecom_api.send_event_response_message(msg_code, response_msg) except Exception as e: logger.error(f"处理会话状态变更事件异常: {e}") def handle_user_recall_event(self, event_data: Dict) -> None: """处理用户撤回消息事件，暂不处理，仅做日志记录""" try: external_userid = event_data.get('external_userid', '') recall_msgid = event_data.get('recall_msgid', '') logger.info(f"用户 {external_userid} 撤回消息: {recall_msgid}") except Exception as e: logger.error(f"处理用户撤回消息事件异常: {e}") def handle_servicer_recall_event(self, event_data: Dict) -> None: """处理接待人员撤回消息事件，暂不处理，仅做日志记录""" try: external_userid = event_data.get('external_userid', '') servicer_userid = event_data.get('servicer_userid', '') recall_msgid = event_data.get('recall_msgid', '') logger.info(f"接待人员 {servicer_userid} 撤回消息: {recall_msgid}") except Exception as e: logger.error(f"处理接待人员撤回消息事件异常: {e}") def get_smart_reply(self, content: str) -> str: """智能回复，使用RAG服务获取回复内容，并记录token消耗情况和参考文档""" logger.info(f"智能客服回复开始") try: # 调用智能客服接口，获取智能回复内容和token消耗情况 response = get_rag_service().query_service(content) if response.success: # 记录token使用情况 if response.token_usage: logger.info(f"💰 本次查询Token使用: 输入{response.token_usage.prompt_tokens}, " f"输出{response.token_usage.completion_tokens}, " f"总计{response.token_usage.total_tokens}") # 记录参考文档 if response.source_documents: logger.info(f"📚 参考文档: {[doc.doc_name for doc in response.source_documents]}") return response.content else: logger.error(f"RAG查询失败: {response.error_message}") return "抱歉，我暂时无法理解您的问题，请稍后再试或联系人工客服。" except Exception as e: logger.error(f"智能回复处理异常: {e}") return "抱歉，我暂时无法理解您的问题，请稍后再试或联系人工客服。" def send_kf_reply(self, external_userid: str, open_kfid: str, content: str) -> bool: """发送客服回复""" try: # 检查回复内容长度 if len(content) > Config.KF_MAX_REPLY_LENGTH: content = content[:Config.KF_MAX_REPLY_LENGTH-10] + "...(内容过长已截断)" result = wecom_api.send_kf_text_message(external_userid, open_kfid, content) if result: logger.info(f"发送客服回复成功 - 用户: {external_userid}") return True else: logger.error(f"发送客服回复失败 - 用户: {external_userid}") return False except Exception as e: logger.error(f"发送客服回复异常: {e}") return False def get_service_session_state(self, external_userid: str, open_kfid: str) -> int: """获取服务会话状态""" try: result = wecom_api.get_service_session_state(external_userid, open_kfid) if result: logger.info(f"获取服务会话状态成功 - 用户: {external_userid}") logger.info(f"服务会话状态: {result}") return result.get('service_state', -1) else: logger.error(f"获取服务会话状态失败 - 用户: {external_userid}") return -1 except Exception as e: logger.error(f"获取服务会话状态异常: {e}") return -1 def update_service_session_state_to_service_pool(self, external_userid: str, open_kfid: str) -> bool: """更新服务会话状态""" try: result = wecom_api.update_service_session_state(external_userid, open_kfid, 2) # 2: 转接会话进入接待池，后续由人工接待 if result: logger.info(f"更新服务会话状态成功 - 用户: {external_userid}") return True else: logger.error(f"更新服务会话状态失败 - 用户: {external_userid}") return False except Exception as e: logger.error(f"更新服务会话状态异常: {e}") return False # 全局客服消息处理器实例 kf_message_handler = KfMessageHandler()