""" 企业微信通知工具 - 用于与物业人员沟通 注意：当前为MVP版本，使用模拟通知。 实际部署时需要接入企业微信API。 """ import json from typing import Optional from langchain.tools import tool from storage.database.supabase_client import get_supabase_client import logging from datetime import datetime logger = logging.getLogger(__name__) @tool def notify_staff_new_task( staff_id: int, task_id: int, task_title: str, location: Optional[str] = None, contact_phone: Optional[str] = None, priority: str = "normal" ) -> str: """ 通过企业微信通知物业人员有新任务。 MVP版本：记录通知日志，实际部署时接入企业微信API。 参数: staff_id: 物业人员ID task_id: 任务ID task_title: 任务标题 location: 服务地点（可选） contact_phone: 联系电话（可选） priority: 优先级 - urgent/high/normal/low 返回: 通知发送结果JSON字符串 """ try: client = get_supabase_client() # 获取物业人员信息 staff_result = client.table('staff').select('*').eq('id', staff_id).execute() if not staff_result.data: return json.dumps({ "success": False, "error": f"未找到物业人员ID: {staff_id}" }, ensure_ascii=False) staff = staff_result.data[0] # type: ignore staff_name = staff.get('name', '工作人员') # type: ignore wecom_id = staff.get('wecom_id', '') # type: ignore # 构建通知内容 priority_text = { 'urgent': '【紧急】', 'high': '【高优】', 'normal': '', 'low': '【低优】' }.get(priority, '') notification_content = f""" {priority_text}新任务通知 任务编号：{task_id} 任务内容：{task_title} 服务地点：{location or '未指定'} 联系电话：{contact_phone or '未提供'} 请尽快处理并在完成后反馈结果。 """ # MVP版本：记录通知日志 notification_log = { 'staff_id': staff_id, 'staff_name': staff_name, 'wecom_id': wecom_id, 'task_id': task_id, 'content': notification_content, 'priority': priority, 'sent_at': datetime.now().isoformat(), 'status': 'sent' } logger.info(f"企业微信通知已发送: {json.dumps(notification_log, ensure_ascii=False)}") # TODO: 实际部署时，调用企业微信API发送消息 # 示例： # wecom_api.send_message( # to_user=wecom_id, # content=notification_content # ) return json.dumps({ "success": True, "notification": { "staff_id": staff_id, "staff_name": staff_name, "task_id": task_id, "sent_at": datetime.now().isoformat(), "status": "sent", "message": f"已通过企业微信通知 {staff_name}（工号：{staff.get('employee_id', 'N/A')})" # type: ignore } }, ensure_ascii=False) except Exception as e: error_msg = f"发送企业微信通知时发生错误: {str(e)}" logger.error(error_msg, exc_info=True) return json.dumps({ "success": False, "error": error_msg }, ensure_ascii=False) @tool def process_staff_feedback( task_id: int, staff_id: int, action: str, note: Optional[str] = None ) -> str: """ 处理物业人员反馈（模拟双向沟通）。 当物业人员通过企业微信回复时，AI接收反馈并更新任务状态。 参数: task_id: 任务ID staff_id: 物业人员ID action: 反馈动作 - accept(接受)/complete(完成)/reject(拒绝)/report(报告问题) note: 备注说明（可选） 返回: 处理结果JSON字符串 """ try: client = get_supabase_client() # 获取任务信息 task_result = client.table('tasks').select('*, users(name, phone, room_number)').eq('id', task_id).execute() if not task_result.data: return json.dumps({ "success": False, "error": f"任务 {task_id} 不存在" }, ensure_ascii=False) task = task_result.data[0] # 获取物业人员信息 staff_result = client.table('staff').select('*').eq('id', staff_id).execute() staff_name = staff_result.data[0].get('name', '工作人员') if staff_result.data else '工作人员' # type: ignore # 根据动作更新任务 if action == 'accept': # 接受任务 client.table('tasks').update({ 'status': 'in_progress', 'assigned_to': staff_id }).eq('id', task_id).execute() user_message = f"{staff_name}已接受您的任务，正在处理中..." elif action == 'complete': # 完成任务 client.table('tasks').update({ 'status': 'completed', 'completion_note': note, 'completed_at': datetime.now().isoformat() }).eq('id', task_id).execute() # 更新人员任务数 current_tasks = int(staff_result.data[0].get('current_tasks', 0) or 0) # type: ignore client.table('staff').update({ 'current_tasks': max(0, current_tasks - 1) }).eq('id', staff_id).execute() user_message = f"{staff_name}已完成您的任务！" if note: user_message += f" 备注：{note}" elif action == 'reject': # 拒绝任务 client.table('tasks').update({ 'status': 'pending', 'assigned_to': None }).eq('id', task_id).execute() user_message = f"{staff_name}暂时无法处理您的任务，正在重新分配..." elif action == 'report': # 报告问题 client.table('tasks').update({ 'completion_note': note }).eq('id', task_id).execute() user_message = f"{staff_name}反馈：{note}" else: return json.dumps({ "success": False, "error": f"未知的反馈动作: {action}" }, ensure_ascii=False) logger.info(f"物业人员反馈处理完成: task_id={task_id}, action={action}, staff={staff_name}") return json.dumps({ "success": True, "task_id": task_id, "action": action, "staff_name": staff_name, "user_notification": user_message, "message": "反馈已处理，已通知业主" }, ensure_ascii=False) except Exception as e: error_msg = f"处理物业人员反馈时发生错误: {str(e)}" logger.error(error_msg, exc_info=True) return json.dumps({ "success": False, "error": error_msg }, ensure_ascii=False) @tool def simulate_staff_response( task_id: int ) -> str: """ 模拟物业人员响应（用于测试和演示）。 在MVP版本中，用于演示双向沟通流程。 实际部署时，物业人员通过企业微信直接回复。 参数: task_id: 任务ID 返回: 模拟响应结果JSON字符串 """ try: client = get_supabase_client() # 获取任务信息 task_result = client.table('tasks').select('*, staff(name)').eq('id', task_id).execute() if not task_result.data: return json.dumps({ "success": False, "error": f"任务 {task_id} 不存在" }, ensure_ascii=False) task = task_result.data[0] # 模拟物业人员响应 # 在实际场景中，物业人员会在企业微信中回复，AI接收消息后处理 import random actions = ['accept', 'complete'] simulated_action = random.choice(actions) if simulated_action == 'accept': simulated_note = "收到，我马上处理" else: simulated_notes = [ "问题已解决", "已完成维修", "服务已完成，请验收" ] simulated_note = random.choice(simulated_notes) logger.info(f"模拟物业人员响应: task_id={task_id}, action={simulated_action}") return json.dumps({ "success": True, "simulation": True, "task_id": task_id, "simulated_action": simulated_action, "simulated_note": simulated_note, "message": "这是模拟响应，实际部署时物业人员会通过企业微信回复", "hint": f"您可以调用 process_staff_feedback(task_id={task_id}, staff_id={task.get('assigned_to', 1)}, action='{simulated_action}', note='{simulated_note}') 来模拟处理反馈" # type: ignore }, ensure_ascii=False) except Exception as e: error_msg = f"模拟物业人员响应时发生错误: {str(e)}" logger.error(error_msg, exc_info=True) return json.dumps({ "success": False, "error": error_msg }, ensure_ascii=False)