Simon
257 lines
8.1 KiB
Python
257 lines
8.1 KiB
Python
# =============================================================================
|
||
# 企微IT智能服务台 — 快速回复模板 API
|
||
# =============================================================================
|
||
# 说明：坐席端的快速回复模板管理接口，包括：
|
||
# 1. GET /api/quick-replies — 获取模板列表（按分类）
|
||
# 2. POST /api/quick-replies — 创建模板
|
||
# 3. PUT /api/quick-replies/{id} — 更新模板
|
||
# 4. DELETE /api/quick-replies/{id} — 删除模板
|
||
# =============================================================================
|
||
|
||
import logging
|
||
from typing import Optional
|
||
from uuid import UUID
|
||
|
||
from fastapi import APIRouter, Depends, Header, Query
|
||
from sqlalchemy import or_, and_
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.database import get_db
|
||
from app.models.agent import Agent
|
||
from app.models.quick_reply_template import QuickReplyTemplate
|
||
from app.schemas.quick_reply import (
|
||
QuickReplyCreate,
|
||
QuickReplyResponse,
|
||
QuickReplyUpdate,
|
||
)
|
||
from app.utils.response import AppException, ERR_NOT_FOUND, ERR_UNAUTHORIZED, success_response
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 创建路由器
|
||
router = APIRouter()
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 可选坐席认证（有 token 则认证，无 token 则跳过）
|
||
# --------------------------------------------------------------------------
|
||
async def get_optional_agent(
|
||
authorization: Optional[str] = Header(None, alias="Authorization"),
|
||
db: AsyncSession = Depends(get_db),
|
||
) -> Optional[Agent]:
|
||
"""可选坐席认证依赖。
|
||
|
||
有 Authorization 头时尝试认证，无或认证失败时返回 None。
|
||
|
||
Args:
|
||
authorization: 请求头中的 Authorization 字段
|
||
db: 数据库会话
|
||
|
||
Returns:
|
||
Optional[Agent]: 认证成功返回坐席对象，否则返回 None
|
||
"""
|
||
if not authorization:
|
||
return None
|
||
|
||
token = authorization.replace("Bearer ", "") if authorization.startswith("Bearer ") else authorization
|
||
if not token:
|
||
return None
|
||
|
||
try:
|
||
import redis.asyncio as aioredis
|
||
from app.config import settings
|
||
|
||
redis_client = settings.create_redis_client()
|
||
try:
|
||
agent_user_id = await redis_client.get(f"agent:token:{token}")
|
||
if not agent_user_id:
|
||
return None
|
||
|
||
uid = agent_user_id.decode("utf-8") if isinstance(agent_user_id, bytes) else agent_user_id
|
||
stmt = select(Agent).where(Agent.user_id == uid)
|
||
result = await db.execute(stmt)
|
||
agent = result.scalars().first()
|
||
return agent
|
||
finally:
|
||
try:
|
||
await redis_client.close()
|
||
except Exception:
|
||
pass
|
||
except Exception as e:
|
||
logger.warning(f"可选坐席认证失败: {e}")
|
||
return None
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# GET /api/quick-replies — 获取模板列表
|
||
# --------------------------------------------------------------------------
|
||
@router.get("/quick-replies")
|
||
async def list_quick_replies(
|
||
category: Optional[str] = Query(None, description="按分类过滤: 账号/网络/软件/硬件/通用"),
|
||
db: AsyncSession = Depends(get_db),
|
||
agent: Optional[Agent] = Depends(get_optional_agent),
|
||
):
|
||
"""获取快速回复模板列表。
|
||
|
||
支持按分类过滤，按 sort_order 排序。
|
||
坐席端可见性规则：
|
||
- 有认证：返回 approved + 自己的 pending_review
|
||
- 无认证：只返回 approved
|
||
|
||
Args:
|
||
category: 按分类过滤（可选）
|
||
db: 数据库会话
|
||
agent: 当前坐席（可选认证）
|
||
|
||
Returns:
|
||
Dict: 统一响应格式，包含模板列表
|
||
"""
|
||
stmt = select(QuickReplyTemplate).order_by(
|
||
QuickReplyTemplate.category, QuickReplyTemplate.sort_order
|
||
)
|
||
|
||
if category:
|
||
stmt = stmt.where(QuickReplyTemplate.category == category)
|
||
|
||
# 状态筛选：坐席端可见性规则
|
||
if agent:
|
||
# 有认证：approved + 自己的 pending_review
|
||
stmt = stmt.where(
|
||
or_(
|
||
QuickReplyTemplate.status == "approved",
|
||
and_(
|
||
QuickReplyTemplate.status == "pending_review",
|
||
QuickReplyTemplate.submitted_by == agent.id,
|
||
),
|
||
)
|
||
)
|
||
else:
|
||
# 无认证：只返回 approved
|
||
stmt = stmt.where(QuickReplyTemplate.status == "approved")
|
||
|
||
result = await db.execute(stmt)
|
||
templates = list(result.scalars().all())
|
||
|
||
items = [QuickReplyResponse.model_validate(t).model_dump() for t in templates]
|
||
return success_response(data={"items": items})
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# POST /api/quick-replies — 创建模板
|
||
# --------------------------------------------------------------------------
|
||
@router.post("/quick-replies")
|
||
async def create_quick_reply(
|
||
body: QuickReplyCreate,
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""创建快速回复模板。
|
||
|
||
Args:
|
||
body: 创建请求体（包含 category、title、content、variables、sort_order）
|
||
db: 数据库会话
|
||
|
||
Returns:
|
||
Dict: 统一响应格式，包含创建的模板
|
||
"""
|
||
template = QuickReplyTemplate(
|
||
category=body.category,
|
||
title=body.title,
|
||
content=body.content,
|
||
variables=body.variables,
|
||
sort_order=body.sort_order,
|
||
)
|
||
db.add(template)
|
||
await db.flush()
|
||
|
||
logger.info(f"创建快速回复模板: category={body.category}, title={body.title}")
|
||
|
||
template_data = QuickReplyResponse.model_validate(template).model_dump()
|
||
return success_response(data=template_data)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# PUT /api/quick-replies/{id} — 更新模板
|
||
# --------------------------------------------------------------------------
|
||
@router.put("/quick-replies/{template_id}")
|
||
async def update_quick_reply(
|
||
template_id: UUID,
|
||
body: QuickReplyUpdate,
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""更新快速回复模板。
|
||
|
||
只更新传入的字段（部分更新）。
|
||
|
||
Args:
|
||
template_id: 模板ID
|
||
body: 更新请求体（所有字段可选）
|
||
db: 数据库会话
|
||
|
||
Returns:
|
||
Dict: 统一响应格式，包含更新后的模板
|
||
"""
|
||
# 查找模板
|
||
stmt = select(QuickReplyTemplate).where(QuickReplyTemplate.id == template_id)
|
||
result = await db.execute(stmt)
|
||
template = result.scalars().first()
|
||
|
||
if not template:
|
||
raise ERR_NOT_FOUND
|
||
|
||
# 只更新传入的字段
|
||
if body.category is not None:
|
||
template.category = body.category
|
||
if body.title is not None:
|
||
template.title = body.title
|
||
if body.content is not None:
|
||
template.content = body.content
|
||
if body.variables is not None:
|
||
template.variables = body.variables
|
||
if body.sort_order is not None:
|
||
template.sort_order = body.sort_order
|
||
|
||
db.add(template)
|
||
await db.flush()
|
||
|
||
logger.info(f"更新快速回复模板: id={template_id}")
|
||
|
||
template_data = QuickReplyResponse.model_validate(template).model_dump()
|
||
return success_response(data=template_data)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# DELETE /api/quick-replies/{id} — 删除模板
|
||
# --------------------------------------------------------------------------
|
||
@router.delete("/quick-replies/{template_id}")
|
||
async def delete_quick_reply(
|
||
template_id: UUID,
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""删除快速回复模板。
|
||
|
||
第一步使用物理删除。
|
||
|
||
Args:
|
||
template_id: 模板ID
|
||
db: 数据库会话
|
||
|
||
Returns:
|
||
Dict: 统一响应格式
|
||
"""
|
||
# 查找模板
|
||
stmt = select(QuickReplyTemplate).where(QuickReplyTemplate.id == template_id)
|
||
result = await db.execute(stmt)
|
||
template = result.scalars().first()
|
||
|
||
if not template:
|
||
raise ERR_NOT_FOUND
|
||
|
||
# 物理删除
|
||
await db.delete(template)
|
||
await db.flush()
|
||
|
||
logger.info(f"删除快速回复模板: id={template_id}")
|
||
|
||
return success_response(data=None, message="删除成功")
|