tools-api/app/call_center/api/intent/evaluate.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import uuid
from typing import Annotated, Optional

from app.call_center.schema.intent_records import CreateIntentRecordsParam, GetIntentRecordsListDetails, \
    UpdateIntentRecordsParam, GetIntentRecordsByIdParam, intent_status_map
from app.call_center.service.intent_records_service import intent_records_service
from common.pagination import paging_data, DependsPagination
from common.response.response_schema import ResponseModel, response_base
from common.security.jwt_call_center import DependsJwtAuth
from common.security.permission import RequestPermission
from database.db_mysql import CurrentSession
from fastapi import APIRouter, Depends, Query, Request

router = APIRouter()

@router.get(
    '/result',
    summary='话单意向度结果查询',
    dependencies=[
        DependsJwtAuth,
    ],
)
async def get_intent_record(
        request: Request,
        external_id: Annotated[Optional[str], Query(description='外部ID')] = None,
        internal_id: Annotated[Optional[str], Query(description='内部ID')] = None
) -> ResponseModel:
    obj = GetIntentRecordsByIdParam(
        external_id=external_id,
        internal_id=internal_id,
        org_id=request.user.id
    )
    intent_record_select = await intent_records_service.get_select_by_id(obj=obj)
    data = None
    if intent_record_select:
        intent = None
        if intent_record_select.llm_intent:
            intent = intent_status_map.get(intent_record_select.llm_intent, "其他")
        data = {
            "status": intent_record_select.status,
            "internal_id": intent_record_select.id,
            "external_id": intent_record_select.external_id,
            "score": intent_record_select.llm_intent,
            "intent": intent
        }
    return response_base.success(data=data)

@router.get(
    '',
    summary='（模糊条件）分页获取所有record',
    dependencies=[
        DependsJwtAuth,
        DependsPagination,
    ],
)
async def get_pagination_intent_record(db: CurrentSession) -> ResponseModel:
    intent_record_select = await intent_records_service.get_select()
    page_data = await paging_data(db, intent_record_select, GetIntentRecordsListDetails)
    return response_base.success(data=page_data)


@router.post(
    '',
    summary='话单意向度判断',
    dependencies=[
        DependsJwtAuth
    ],
)
async def create_intent_record(obj: CreateIntentRecordsParam, request: Request) -> ResponseModel:
    id_str = str(uuid.uuid4())
    obj.id = id_str
    obj.org_id = request.user.id
    await intent_records_service.create(obj=obj)
    return response_base.success()


@router.post(
    '/update',
    summary='话单意向度更新',
    dependencies=[
        DependsJwtAuth
    ],
)
async def update_intent_record(obj: UpdateIntentRecordsParam, request: Request) -> ResponseModel:
    obj.org_id = request.user.id
    count = await intent_records_service.update_manual_intent(obj=obj)
    if count > 0:
        return response_base.success()
    return response_base.fail()


@router.delete(
    '',
    summary='（批量）删除record',
    dependencies=[
        Depends(RequestPermission('call:record:del')),
    ],
)
async def delete_intent_record(pk: Annotated[list[int], Query(...)]) -> ResponseModel:
    count = await intent_records_service.delete(pk=pk)
    if count > 0:
        return response_base.success()
    return response_base.fail()