tools-api/app/call_center/crud/crud_intent_records.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Sequence

from model.intent_records import IntentRecords
from app.call_center.schema.intent_records import CreateIntentRecordsParam, UpdateIntentRecordsParam, \
    GetIntentRecordsByIdParam
from sqlalchemy import Select, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy_crud_plus import CRUDPlus


class CRUDIntentRecords(CRUDPlus[IntentRecords]):
    async def get(self, db: AsyncSession, obj_in: GetIntentRecordsByIdParam) -> IntentRecords | None:
        """
        获取call record

        :param db:
        :param obj_in:
        :return:
        """
        intent_records = None
        if obj_in.internal_id:
            intent_records = await self.select_model_by_column(db, id=obj_in.internal_id, org_id=obj_in.org_id)
        elif obj_in.external_id:
            intent_records = await self.select_model_by_column(db, external_id=obj_in.external_id, org_id=obj_in.org_id)
        return intent_records

    async def get_earliest_record(self, db: AsyncSession) -> IntentRecords | None:
        """
        获取最早的 llm_intent 不为空的记录

        :param db:
        :return:
        """
        stmt = select(self.model).where(IntentRecords.llm_intent == None).order_by(IntentRecords.created_at.asc())
        query = await db.execute(stmt)
        return query.scalars().first()

    async def get_list(self) -> Select:
        """
        获取call record列表

        :return:
        """
        return await self.select_order('created_at', 'desc')

    async def get_all(self, db: AsyncSession) -> Sequence[IntentRecords]:
        """
        获取所有call record

        :param db:
        :return:
        """
        return await self.select_models(db)

    async def create(self, db: AsyncSession, obj_in: CreateIntentRecordsParam) -> None:
        """
        创建call record

        :param db:
        :param obj_in:
        :return:
        """
        await self.create_model(db, obj_in)

    async def update(self, db: AsyncSession, pk: int, obj_in: UpdateIntentRecordsParam) -> int:
        """
        更新call record

        :param db:
        :param pk:
        :param obj_in:
        :return:
        """
        return await self.update_model(db, pk, obj_in)

    async def update_manual_intent(self, db: AsyncSession, obj_in: UpdateIntentRecordsParam) -> int:
        """
        更新call record

        :param db:
        :param pk:
        :param obj_in:
        :return:
        """
        if obj_in.internal_id:
            return await self.update_model_by_column(db, {'manual_intent': obj_in.manual_intent}, id=obj_in.internal_id)
        else:
            return await self.update_model_by_column(db, {'manual_intent': obj_in.manual_intent}, external_id=obj_in.external_id)

    async def update_llm_intent(self, db: AsyncSession, internal_id: str, llm_intent: int, request_data: dict, response_data: dict, status: int) -> int:
        """
        更新call record

        :param db:
        :param pk:
        :param internal_id:
        :param llm_intent:
        :param request_data:
        :param response_data:
        :param status:
        :return:
        """

        return await self.update_model_by_column(db, {'llm_intent': llm_intent, 'request_data': request_data, 'response_data': response_data, 'status': status}, id=internal_id)


    async def delete(self, db: AsyncSession, pk: list[int]) -> int:
        """
        删除call record

        :param db:
        :param pk:
        :return:
        """
        return  await self.delete_model_by_column(db, allow_multiple=True, id__in=pk)


intent_records_dao: CRUDIntentRecords = CRUDIntentRecords(IntentRecords)