wecom_it_smart_desk/backend/app/api/knowledge_base.py

# =============================================================================
# 企微IT智能服务台 — 知识库 API
# =============================================================================
# 说明：知识库FAQ管理接口，包括：
#   1. GET /api/knowledge - 获取知识库列表
#   2. POST /api/knowledge - 创建知识条目
#   3. PUT /api/knowledge/{id} - 更新知识条目
#   4. DELETE /api/knowledge/{id} - 删除知识条目
#   5. GET /api/knowledge/search - 搜索知识
# =============================================================================

import logging
from typing import Optional
from uuid import UUID

from fastapi import APIRouter, Depends, Query
from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db
from app.models.agent import Agent
from app.models.knowledge_base import KnowledgeBase
from app.schemas.knowledge_base import (
    KnowledgeBaseCreate,
    KnowledgeBaseResponse,
    KnowledgeBaseUpdate,
)
from app.utils.response import AppException, ERR_NOT_FOUND, success_response

logger = logging.getLogger(__name__)

# 创建路由器
router = APIRouter()


# --------------------------------------------------------------------------
# GET /api/knowledge — 获取知识库列表
# --------------------------------------------------------------------------
@router.get("/knowledge")
async def list_knowledge(
    category: Optional[str] = Query(None, description="按分类筛选"),
    keyword: Optional[str] = Query(None, description="关键词搜索"),
    db: AsyncSession = Depends(get_db),
):
    """获取知识库列表。

    支持按分类筛选和关键词搜索。

    Args:
        category: 按分类筛选（可选）
        keyword: 关键词搜索（可选，搜索标题和内容）
        db: 数据库会话

    Returns:
        Dict: 统一响应格式，包含知识库列表
    """
    stmt = select(KnowledgeBase).order_by(KnowledgeBase.view_count.desc())

    if category:
        stmt = stmt.where(KnowledgeBase.category == category)

    if keyword:
        # 关键词搜索：标题或内容包含关键字
        stmt = stmt.where(
            or_(
                KnowledgeBase.title.ilike(f"%{keyword}%"),
                KnowledgeBase.content.ilike(f"%{keyword}%"),
            )
        )

    result = await db.execute(stmt)
    items = list(result.scalars().all())

    data = [KnowledgeBaseResponse.model_validate(t).model_dump() for t in items]
    return success_response(data={"items": data})


# --------------------------------------------------------------------------
# POST /api/knowledge — 创建知识条目
# --------------------------------------------------------------------------
@router.post("/knowledge")
async def create_knowledge(
    body: KnowledgeBaseCreate,
    db: AsyncSession = Depends(get_db),
):
    """创建知识库条目。

    Args:
        body: 创建请求体
        db: 数据库会话

    Returns:
        Dict: 统一响应格式，包含创建的知识条目
    """
    knowledge = KnowledgeBase(
        category=body.category,
        title=body.title,
        content=body.content,
        tags=body.tags,
    )
    db.add(knowledge)
    await db.flush()

    logger.info(f"创建知识库条目: category={body.category}, title={body.title}")

    data = KnowledgeBaseResponse.model_validate(knowledge).model_dump()
    return success_response(data=data)


# --------------------------------------------------------------------------
# PUT /api/knowledge/{id} — 更新知识条目
# --------------------------------------------------------------------------
@router.put("/knowledge/{knowledge_id}")
async def update_knowledge(
    knowledge_id: UUID,
    body: KnowledgeBaseUpdate,
    db: AsyncSession = Depends(get_db),
):
    """更新知识库条目。

    Args:
        knowledge_id: 知识ID
        body: 更新请求体
        db: 数据库会话

    Returns:
        Dict: 统一响应格式，包含更新后的知识条目
    """
    stmt = select(KnowledgeBase).where(KnowledgeBase.id == knowledge_id)
    result = await db.execute(stmt)
    knowledge = result.scalars().first()

    if not knowledge:
        raise ERR_NOT_FOUND

    # 只更新传入的字段
    if body.category is not None:
        knowledge.category = body.category
    if body.title is not None:
        knowledge.title = body.title
    if body.content is not None:
        knowledge.content = body.content
    if body.tags is not None:
        knowledge.tags = body.tags

    db.add(knowledge)
    await db.flush()

    logger.info(f"更新知识库条目: id={knowledge_id}")

    data = KnowledgeBaseResponse.model_validate(knowledge).model_dump()
    return success_response(data=data)


# --------------------------------------------------------------------------
# DELETE /api/knowledge/{id} — 删除知识条目
# --------------------------------------------------------------------------
@router.delete("/knowledge/{knowledge_id}")
async def delete_knowledge(
    knowledge_id: UUID,
    db: AsyncSession = Depends(get_db),
):
    """删除知识库条目。

    Args:
        knowledge_id: 知识ID
        db: 数据库会话

    Returns:
        Dict: 统一响应格式
    """
    stmt = select(KnowledgeBase).where(KnowledgeBase.id == knowledge_id)
    result = await db.execute(stmt)
    knowledge = result.scalars().first()

    if not knowledge:
        raise ERR_NOT_FOUND

    await db.delete(knowledge)
    await db.flush()

    logger.info(f"删除知识库条目: id={knowledge_id}")

    return success_response(data=None, message="删除成功")


# --------------------------------------------------------------------------
# PUT /api/knowledge/{id}/view — 更新查看次数
# --------------------------------------------------------------------------
@router.put("/knowledge/{knowledge_id}/view")
async def view_knowledge(
    knowledge_id: UUID,
    db: AsyncSession = Depends(get_db),
):
    """记录知识库条目被查看。

    Args:
        knowledge_id: 知识ID
        db: 数据库会话

    Returns:
        Dict: 统一响应格式
    """
    stmt = select(KnowledgeBase).where(KnowledgeBase.id == knowledge_id)
    result = await db.execute(stmt)
    knowledge = result.scalars().first()

    if not knowledge:
        raise ERR_NOT_FOUND

    knowledge.view_count += 1
    db.add(knowledge)
    await db.flush()

    return success_response(data={"view_count": knowledge.view_count})


# --------------------------------------------------------------------------
# PUT /api/knowledge/{id}/use — 更新使用次数
# --------------------------------------------------------------------------
@router.put("/knowledge/{knowledge_id}/use")
async def use_knowledge(
    knowledge_id: UUID,
    db: AsyncSession = Depends(get_db),
):
    """记录知识库条目被使用（坐席引用）。

    Args:
        knowledge_id: 知识ID
        db: 数据库会话

    Returns:
        Dict: 统一响应格式
    """
    stmt = select(KnowledgeBase).where(KnowledgeBase.id == knowledge_id)
    result = await db.execute(stmt)
    knowledge = result.scalars().first()

    if not knowledge:
        raise ERR_NOT_FOUND

    knowledge.use_count += 1
    db.add(knowledge)
    await db.flush()

    return success_response(data={"use_count": knowledge.use_count})