# ============================================================================= # 企微IT智能服务台 — 业务路由推荐 API # ============================================================================= # 说明：提供业务联系人查询和坐席手动发名片功能 # - GET /h5/routing/contact — H5查询联系人（按业务类别） # - GET /routing/contacts — 坐席端联系人列表（P1） # - POST /conversations/{id}/send-contact-card — 坐席手动发名片（P1） # ============================================================================= import logging from typing import Optional from fastapi import APIRouter, Depends, Query from pydantic import BaseModel from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.models.business_contact import BusinessContact from app.models.conversation import Conversation from app.models.message import Message from app.services.routing_service import send_contact_card from app.services.ws_manager import manager as ws_manager from app.utils.response import success_response logger = logging.getLogger(__name__) router = APIRouter() # ============================================================================= # 数据库依赖 # ============================================================================= async def get_db() -> AsyncSession: """获取异步 DB session。 使用 app.database 的 session_factory 创建独立 session。 """ from app.database import _get_session_factory factory = _get_session_factory() async with factory() as session: yield session # ============================================================================= # Schema 定义 # ============================================================================= class ContactResponse(BaseModel): """联系人信息响应。""" id: int name: str gender: str department: str position: str responsibility: str extension: str = "" service_area: str = "" wecom_userid: str avatar_url: str = "" business_category: str class ContactListResponse(BaseModel): """联系人列表响应。""" items: list[ContactResponse] class SendContactCardRequest(BaseModel): """坐席手动发名片请求。""" contact_id: int reason: Optional[str] = None class SendContactCardResponse(BaseModel): """坐席手动发名片响应。""" message_id: str contact_name: str business_category: str # ============================================================================= # API 端点 # ============================================================================= @router.get("/h5/routing/contact") async def get_h5_routing_contact( business_category: str = Query(..., description="业务类别（行政/人力资源/财务/法务/行政-物业）"), db: AsyncSession = Depends(get_db), ): """H5 端查询业务联系人。 按 business_category + is_active=true 查询，取第一条有效联系人。 H5 前端收到 WS 推送的 contact_card 消息后，也可通过此接口补充查询。 Args: business_category: 业务类别 db: 异步 DB session Returns: 统一响应格式，data 为联系人信息 """ result = await db.execute( select(BusinessContact) .where( BusinessContact.business_category == business_category, BusinessContact.is_active == True, # noqa: E712 ) .order_by(BusinessContact.id) .limit(1) ) contact = result.scalar_one_or_none() if not contact: return success_response(data=None, message=f"未找到{business_category}类别的联系人") return success_response(data=ContactResponse( id=contact.id, name=contact.name, gender=contact.gender, department=contact.department, position=contact.position, responsibility=contact.responsibility, extension=contact.extension or "", service_area=contact.service_area or "", wecom_userid=contact.wecom_userid, avatar_url=contact.avatar_url or "", business_category=contact.business_category, )) @router.get("/routing/contacts") async def get_routing_contacts( business_category: Optional[str] = Query(None, description="业务类别筛选"), keyword: Optional[str] = Query(None, description="姓名/部门关键词搜索"), db: AsyncSession = Depends(get_db), ): """坐席端获取联系人列表（P1）。 支持按业务类别筛选和关键词搜索，用于坐席手动发名片时的联系人选择面板。 Args: business_category: 业务类别筛选（可选） keyword: 姓名/部门关键词搜索（可选） db: 异步 DB session Returns: 统一响应格式，data 为 { items: [ContactResponse, ...] } """ query = select(BusinessContact).where( BusinessContact.is_active == True # noqa: E712 ) if business_category: query = query.where(BusinessContact.business_category == business_category) if keyword: # 在姓名或部门中搜索 search_pattern = f"%{keyword}%" query = query.where( BusinessContact.name.ilike(search_pattern) | BusinessContact.department.ilike(search_pattern) ) query = query.order_by(BusinessContact.business_category, BusinessContact.id) result = await db.execute(query) contacts = result.scalars().all() items = [ ContactResponse( id=c.id, name=c.name, gender=c.gender, department=c.department, position=c.position, responsibility=c.responsibility, extension=c.extension or "", service_area=c.service_area or "", wecom_userid=c.wecom_userid, avatar_url=c.avatar_url or "", business_category=c.business_category, ) for c in contacts ] return success_response(data={"items": [item.model_dump() for item in items]}) @router.post("/conversations/{conversation_id}/send-contact-card") async def send_contact_card_endpoint( conversation_id: str, request: SendContactCardRequest, db: AsyncSession = Depends(get_db), ): """坐席手动发送名片卡片（P1）。 坐席在会话中手动选择联系人，后端创建 contact_card 消息并 WS 推送双通道。 Args: conversation_id: 会话ID request: 包含 contact_id 和可选的 reason db: 异步 DB session Returns: 统一响应格式，data 为发送结果 """ # 1. 查询会话 conversation = await db.get(Conversation, conversation_id) if not conversation: return success_response( data=None, message=f"会话不存在: {conversation_id}", ) # 2. 查询联系人 contact = await db.get(BusinessContact, request.contact_id) if not contact: return success_response( data=None, message=f"联系人不存在: {request.contact_id}", ) if not contact.is_active: return success_response( data=None, message=f"联系人已停用: {contact.name}", ) # 3. 构建路由说明文本 reason = request.reason or f"为您推荐{contact.business_category}服务联系人：{contact.name}" employee_id = conversation.employee_id or "" # 4. 发送名片三段式消息（复用 routing_service 的发送逻辑） await send_contact_card( db=db, conversation=conversation, employee_id=employee_id, contact=contact, reason=reason, business_category=contact.business_category, routing_confidence=1.0, # 坐席手动发送，置信度设为 1.0 ) return success_response(data=SendContactCardResponse( message_id="sent", contact_name=contact.name, business_category=contact.business_category, ))