tjdify/api/services/intention_service.py

import logging
import uuid
from datetime import datetime
from typing import Optional

from flask_login import current_user
from werkzeug.exceptions import Forbidden, NotFound

from extensions.ext_database import db
from models.intention import (
    Intention,
    IntentionCorpus,
    IntentionCorpusSimilarityQuestion,
    IntentionKeyword,
    IntentionType,
)
from services.errors.intention import (
    IntentionCorpusQuestionDuplicateError,
    IntentionKeywordNameDuplicateError,
    IntentionNameDuplicateError,
    IntentionTypeNameDuplicateError,
)


class IntentionTypeService:
    @staticmethod
    def get_intention_types(page, per_page, search=None):
        query = IntentionType.query.order_by(IntentionType.created_at.desc())
        if search:
            query = query.filter(IntentionType.name.like(f"%{search}%"))
        intention_types = query.paginate(page=page, per_page=per_page, error_out=False)
        return intention_types.items, intention_types.total

    @staticmethod
    def save_intention_type(args: dict) -> IntentionType:
        name = args["name"]
        intention_type = IntentionTypeService.get_intention_type_by_name(name)
        if intention_type:
            raise IntentionTypeNameDuplicateError(f"IntentionType with name {name} already exists.")
        intention_type = IntentionType(
            id=str(uuid.uuid4()),
            name=name,
            created_by=current_user.id,
            created_at=datetime.now(),
        )
        db.session.add(intention_type)
        db.session.commit()
        return intention_type

    @staticmethod
    def update_intention_type(id: str, args: dict) -> IntentionType:
        intention_type = IntentionTypeService.get_intention_type(id)
        if not intention_type:
            raise NotFound("IntentionType not found")

        name = args["name"]
        intention_type_new = IntentionType.query.filter(
            IntentionType.id != intention_type.id,
            IntentionType.name == name,
        ).first()
        if intention_type_new:
            raise IntentionTypeNameDuplicateError(f"IntentionType with name {name} already exists.")
        intention_type.name = name

        intention_type.updated_by = current_user.id
        intention_type.updated_at = datetime.now()
        db.session.add(intention_type)
        db.session.commit()
        return intention_type

    @staticmethod
    def delete_intention_type(id: str):
        intention_type = IntentionTypeService.get_intention_type(id)
        if not intention_type:
            raise NotFound("IntentionType not found")
        db.session.delete(intention_type)
        db.session.commit()
        return intention_type

    @staticmethod
    def get_intention_type(id: str) -> Optional[IntentionType]:
        intention_type: Optional[IntentionType] = IntentionType.query.filter_by(id=id).first()
        return intention_type

    @staticmethod
    def get_intention_type_by_name(name: str) -> Optional[IntentionType]:
        intention_type: Optional[IntentionType] = IntentionType.query.filter_by(name=name).first()
        return intention_type


class IntentionService:
    @staticmethod
    def get_intentions(page, per_page, type_id=None, name_search=None):
        query = Intention.query.order_by(Intention.created_at.desc())

        if type_id:
            query = query.filter(Intention.type_id == type_id)

        if name_search:
            query = query.filter(Intention.name.ilike(f"%{name_search}%"))

        intentions = query.paginate(page=page, per_page=per_page, error_out=False)
        return intentions.items, intentions.total

    @staticmethod
    def save_intention(args: dict) -> Intention:
        name = args["name"]
        intention = IntentionService.get_intention_by_name(name)
        if intention:
            raise IntentionNameDuplicateError(f"Intention with name {name} already exists.")

        type_id = args["type_id"]
        intention_type = IntentionTypeService.get_intention_type(type_id)
        if not intention_type:
            raise NotFound("IntentionType not found")

        intention = Intention(
            id=str(uuid.uuid4()),
            name=name,
            type_id=type_id,
            created_by=current_user.id,
            created_at=datetime.now(),
        )
        db.session.add(intention)
        db.session.commit()
        return intention

    @staticmethod
    def update_intention(intention_id: str, args: dict) -> Intention:
        intention: Optional[Intention] = IntentionService.get_intention(intention_id)
        if not intention:
            raise NotFound("Intention not found")

        name = args["name"]
        intention_new = Intention.query.filter(Intention.id != intention.id, Intention.name == name).first()
        if intention_new:
            raise IntentionNameDuplicateError(f"Intention with name {name} already exists.")

        intention.name = name

        type_id = args["type_id"]
        intention_type = IntentionTypeService.get_intention_type(type_id)
        if not intention_type:
            raise NotFound("IntentionType not found")
        intention.type_id = type_id
        intention.updated_by = current_user.id
        intention.updated_at = datetime.now()
        db.session.add(intention)
        db.session.commit()
        return intention

    @staticmethod
    def delete_intention(intention_id: str):
        intention: Optional[Intention] = IntentionService.get_intention(intention_id)
        if not intention:
            raise NotFound("Intention not found")

        # 1.若存在关键词，则无法删除
        if intention.keywords_count > 0:
            raise Forbidden(
                f"You are not allowed to delete intention, because {intention.keywords_count} keywords were found."
            )

        # 2.若关联训练预料，则无法删除
        if intention.corpus_count > 0:
            raise Forbidden(
                f"You are not allowed to delete intention, because {intention.corpus_count} corpus were found."
            )

        db.session.delete(intention)
        db.session.commit()
        return intention

    @staticmethod
    def get_intention(intention_id: str) -> Optional[Intention]:
        intention: Optional[Intention] = Intention.query.filter(Intention.id == intention_id).first()
        return intention

    @staticmethod
    def get_intention_by_name(name: str) -> Optional[Intention]:
        intention: Optional[Intention] = Intention.query.filter(Intention.name == name).first()
        return intention


class IntentionKeywordService:
    @staticmethod
    def get_intention_keywords(intention_id: str, search=None):
        query = IntentionKeyword.query.filter_by(intention_id=intention_id)
        if search:
            query = query.filter(IntentionKeyword.name.ilike(f"%{search}%"))
        intention_keywords = query.all()
        return intention_keywords

    @staticmethod
    def save_intention_keyword(intention_id: str, args: dict) -> IntentionKeyword:
        name = args["name"]
        intention_keyword = IntentionKeywordService.get_intention_keyword_by_name(intention_id, name)
        if intention_keyword:
            raise IntentionKeywordNameDuplicateError(f"IntentionKeyword with name {name} already exists.")
        intention_keyword = IntentionKeyword(
            id=str(uuid.uuid4()),
            intention_id=intention_id,
            name=name,
            created_by=current_user.id,
            created_at=datetime.now(),
        )
        db.session.add(intention_keyword)
        db.session.commit()
        return intention_keyword

    @staticmethod
    def update_intention_keyword(intention_keyword_id: str, args: dict) -> IntentionKeyword:
        intention_keyword = IntentionKeywordService.get_intention_keyword(intention_keyword_id)
        if not intention_keyword:
            raise NotFound("IntentionKeyword not found")

        name = args["name"]
        intention_keyword_new = IntentionKeyword.query.filter(
            IntentionKeyword.id != intention_keyword.id, IntentionKeyword.name == name
        ).first()
        if intention_keyword_new:
            raise IntentionKeywordNameDuplicateError(f"IntentionKeyword with name {name} already exists.")
        intention_keyword.name = name

        intention_id = args["intention_id"]
        intention = IntentionService.get_intention(intention_id)
        if not intention:
            raise NotFound("Intention not found")

        intention_keyword.intention_id = intention_id
        intention_keyword.updated_by = current_user.id
        intention_keyword.updated_at = datetime.now()
        db.session.add(intention_keyword)
        db.session.commit()
        return intention_keyword

    @staticmethod
    def delete_intention_keyword(intention_keyword_id: str):
        intention_keyword = IntentionKeywordService.get_intention_keyword(intention_keyword_id)
        if not intention_keyword:
            raise NotFound("IntentionKeyword not found")
        db.session.delete(intention_keyword)
        db.session.commit()

    @staticmethod
    def delete_intention_keywords(intention_keyword_ids: list[str]):
        intention_keywords = IntentionKeyword.query.filter(IntentionKeyword.id.in_(intention_keyword_ids)).all()
        for intention_keyword in intention_keywords:
            db.session.delete(intention_keyword)
        db.session.commit()

    @staticmethod
    def delete_intention_keywords_by_intention_id(intention_id: str):
        intention_keywords = IntentionKeyword.query.filter(IntentionKeyword.intention_id == intention_id).all()
        for intention_keyword in intention_keywords:
            db.session.delete(intention_keyword)
        db.session.commit()

    @staticmethod
    def get_intention_keyword(intention_keyword_id: str) -> Optional[IntentionKeyword]:
        intention_keyword: Optional[IntentionKeyword] = IntentionKeyword.query.filter_by(
            id=intention_keyword_id
        ).first()
        return intention_keyword

    @staticmethod
    def get_intention_keyword_by_name(intention_id, name: str) -> Optional[IntentionKeyword]:
        intention_keyword: Optional[IntentionKeyword] = IntentionKeyword.query.filter(
            IntentionKeyword.intention_id == intention_id,
            IntentionKeyword.name == name,
        ).first()
        return intention_keyword


class IntentionCorpusService:
    @staticmethod
    def get_intention_corpus(corpus_id: str) -> Optional[IntentionCorpus]:
        intention_corpus: Optional[IntentionCorpus] = IntentionCorpus.query.filter(
            IntentionCorpus.id == corpus_id
        ).first()
        return intention_corpus

    @staticmethod
    def get_page_intention_corpus(page, per_page, search=None, intention_id=None):
        query = IntentionCorpus.query.order_by(IntentionCorpus.created_at.desc())
        if search:
            query = query.filter(IntentionCorpus.question.like(f"%{search}%"))

        if intention_id:
            query = query.filter(IntentionCorpus.intention_id == intention_id)

        intention_corpus = query.paginate(page=page, per_page=per_page, error_out=False)
        return intention_corpus.items, intention_corpus.total

    @staticmethod
    def get_intention_corpus_by_question(question: str) -> Optional[IntentionCorpus]:
        intention_corpus: Optional[IntentionCorpus] = IntentionCorpus.query.filter(
            IntentionCorpus.question == question
        ).first()
        return intention_corpus

    @staticmethod
    def save_intention_corpus(args: dict):
        question = args["question"]
        intention_corpus = IntentionCorpusService.get_intention_corpus_by_question(question)
        if intention_corpus:
            raise IntentionCorpusQuestionDuplicateError(f"IntentionCorpus with question {question} already exists.")

        intention_id = args["intention_id"]
        intention = IntentionService.get_intention(intention_id)
        if not intention:
            raise NotFound(f"Intention with id {intention_id} not found")

        intention_corpus = IntentionCorpus(
            id=str(uuid.uuid4()),
            question=question,
            intention_id=intention_id,
            created_by=current_user.id,
            created_at=datetime.now(),
        )

        if "question_config" in args:
            intention_corpus.question_config = args["question_config"]

        db.session.add(intention_corpus)
        db.session.commit()
        return intention_corpus

    @staticmethod
    def update_intention_corpus(corpus_id: str, args: dict):
        intention_corpus = IntentionCorpusService.get_intention_corpus(corpus_id)
        if not intention_corpus:
            raise NotFound(f"IntentionCorpus with id {corpus_id} not found")

        if "question" in args:
            question = args["question"]
            intention_corpus_new = IntentionCorpus.query.filter(
                IntentionCorpus.id != corpus_id, IntentionCorpus.question == question
            ).first()
            if intention_corpus_new:
                raise IntentionCorpusQuestionDuplicateError(f"IntentionCorpus with question {question} already exists.")
            intention_corpus.question = question

        if "question_config" in args:
            intention_corpus.question_config = args["question_config"]

        if "intention_id" in args:
            intention_id = args["intention_id"]
            intention = IntentionService.get_intention(intention_id)
            if not intention:
                raise NotFound(f"Intention with id {intention_id} not found")
            intention_corpus.intention_id = intention.id

        intention_corpus.updated_at = datetime.now()
        intention_corpus.updated_by = current_user.id
        db.session.add(intention_corpus)
        db.session.commit()
        return intention_corpus

    @staticmethod
    def delete_intention_corpus_by_id(corpus_id: str):
        intention_corpus = IntentionCorpusService.get_intention_corpus(corpus_id)
        if not intention_corpus:
            raise NotFound(f"IntentionCorpus with id {corpus_id} not found")
        IntentionCorpusService.delete_intention_corpus(intention_corpus)

    @staticmethod
    def delete_intention_corpus(intention_corpus: IntentionCorpus):
        similarity_questions = intention_corpus.similarity_questions
        if similarity_questions:
            raise Forbidden(f"存在与其关联的相似问题，无法删除Id为{intention_corpus.id}训练语料")

        db.session.delete(intention_corpus)
        db.session.commit()


class IntentionCorpusSimilarityQuestionService:
    @staticmethod
    def save_similarity_question(corpus_id: str, args: dict):
        intention_corpus = IntentionCorpusService.get_intention_corpus(corpus_id)
        if not intention_corpus:
            raise NotFound(f"IntentionCorpus with id {corpus_id} not found")

        question = args["question"]
        intention_corpus_similarity_question = (
            IntentionCorpusSimilarityQuestionService.get_similarity_question_by_question(question)
        )
        if intention_corpus_similarity_question:
            raise IntentionCorpusQuestionDuplicateError(f"IntentionCorpus with question {question} already exists.")

        intention_corpus_similarity_question = IntentionCorpusSimilarityQuestion(
            id=str(uuid.uuid4()),
            question=question,
            corpus_id=corpus_id,
            created_by=current_user.id,
            created_at=datetime.now(),
        )

        if "question_config" in args:
            intention_corpus_similarity_question.question_config = args["question_config"]
        db.session.add(intention_corpus_similarity_question)
        db.session.commit()
        return intention_corpus_similarity_question

    @staticmethod
    def update_similarity_question(similarity_question_id: str, args: dict):
        similarity_question = IntentionCorpusSimilarityQuestionService.get_similarity_question(similarity_question_id)
        if not similarity_question:
            raise NotFound(f"IntentionCorpus with id {similarity_question_id} not found")

        if "corpus_id" in args:
            corpus_id = args["corpus_id"]
            intention_corpus = IntentionCorpusService.get_intention_corpus(corpus_id)
            if not intention_corpus:
                raise NotFound(f"IntentionCorpus with id {corpus_id} not found")
            similarity_question.corpus_id = corpus_id

        if "question" in args:
            similarity_question.question = args["question"]

        if "question_config" in args:
            similarity_question.question_config = args["question_config"]

        db.session.add(similarity_question)
        db.session.commit()
        return similarity_question

    @staticmethod
    def get_similarity_question(similarity_question_id: str) -> Optional[IntentionCorpusSimilarityQuestion]:
        similarity_question: Optional[IntentionCorpus] = IntentionCorpusSimilarityQuestion.query.filter_by(
            id=similarity_question_id
        ).first()
        return similarity_question

    @staticmethod
    def get_similarity_question_by_question(question: str) -> Optional[IntentionCorpusSimilarityQuestion]:
        similarity_question: Optional[IntentionCorpusSimilarityQuestion] = (
            IntentionCorpusSimilarityQuestion.query.filter_by(question=question).first()
        )
        return similarity_question

    @staticmethod
    def get_similarity_questions_by_corpus_id_like_question(corpus_id, search=None):
        query = IntentionCorpusSimilarityQuestion.query.filter(
            IntentionCorpusSimilarityQuestion.corpus_id == corpus_id
        ).order_by(IntentionCorpusSimilarityQuestion.created_at.desc())
        if search:
            query = query.filter(IntentionCorpusSimilarityQuestion.question.ilike(f"%{search}%"))

        similarity_questions = query.all()
        return similarity_questions

    @staticmethod
    def delete_similarity_question_by_corpus_id(corpus_id: str):
        intention_corpus = IntentionCorpusService.get_intention_corpus(corpus_id)
        if not intention_corpus:
            raise NotFound(f"IntentionCorpus with id {corpus_id} not found")
        logging.info(intention_corpus.similarity_questions)
        IntentionCorpusSimilarityQuestionService.delete_similarity_questions(intention_corpus.similarity_questions)

    @staticmethod
    def delete_similarity_question_by_id(similarity_question_id: str):
        similarity_question = IntentionCorpusSimilarityQuestionService.get_similarity_question(similarity_question_id)
        if not similarity_question:
            raise NotFound(f"IntentionCorpus with id {similarity_question_id} not found")

        IntentionCorpusSimilarityQuestionService.delete_similarity_question(similarity_question)

    @staticmethod
    def delete_similarity_question(similarity_question: IntentionCorpusSimilarityQuestion):
        db.session.delete(similarity_question)
        db.session.commit()

    @staticmethod
    def delete_similarity_questions_by_ids(similarity_question_ids: list[str]):
        similarity_questions = IntentionCorpusSimilarityQuestion.query.filter(
            IntentionCorpusSimilarityQuestion.id.in_(similarity_question_ids)
        ).all()
        IntentionCorpusSimilarityQuestionService.delete_similarity_questions(similarity_questions)

    @staticmethod
    def delete_similarity_questions(similarity_questions: list[IntentionCorpusSimilarityQuestion]):
        if not similarity_questions:
            return

        for similarity_question in similarity_questions:
            db.session.delete(similarity_question)
        db.session.commit()