tjdify/api/controllers/console/intention.py

import logging

from flask import request
from flask_restful import Resource, marshal, marshal_with, reqparse
from werkzeug.exceptions import Forbidden, NotFound

from controllers.console import api
from controllers.console.wraps import account_initialization_required, setup_required
from fields.intention_fields import (
    intention_corpus_detail_fields,
    intention_corpus_similarity_question_fields,
    intention_detail_fields,
    intention_keyword_detail_fields,
    intention_keyword_fields,
    intention_page_fields,
    intention_type_detail_fields,
    intention_type_page_fields,
)
from libs.login import login_required
from services.intention_service import (
    IntentionCorpusService,
    IntentionCorpusSimilarityQuestionService,
    IntentionKeywordService,
    IntentionService,
    IntentionTypeService,
)


class IntentionListApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    def get(self):
        page = request.args.get("page", default=1, type=int)
        limit = request.args.get("limit", default=20, type=int)
        type_id = request.args.get("type_id", default=None, type=str)
        name_search = request.args.get("name_search", default=None, type=str)
        intentions, total = IntentionService.get_intentions(page, limit, type_id, name_search)
        data = marshal(intentions, intention_page_fields)
        response = {"data": data, "has_more": len(intentions) == limit, "limit": limit, "total": total, "page": page}
        return response, 200

    @setup_required
    @login_required
    @account_initialization_required
    def post(self):
        parser = reqparse.RequestParser()
        parser.add_argument(
            "name",
            nullable=False,
            required=True,
            help="type is required. Name must be between 1 to 40 characters.",
        )
        parser.add_argument(
            "type_id",
            nullable=False,
            required=True,
            help="type is required.",
        )
        args = parser.parse_args()
        intention = IntentionService.save_intention(args)
        response = marshal(intention, intention_detail_fields)
        return response, 200


class IntentionApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    def get(self, intention_id):
        intention = IntentionService.get_intention(intention_id)
        return marshal(intention, intention_detail_fields)

    @setup_required
    @login_required
    @account_initialization_required
    def patch(self, intention_id):
        parser = reqparse.RequestParser()
        parser.add_argument(
            "name",
            nullable=False,
            required=True,
            help="type is required. Name must be between 1 to 40 characters.",
        )
        parser.add_argument(
            "type_id",
            nullable=False,
            required=True,
            help="type is required.",
        )
        args = parser.parse_args()
        intention = IntentionService.update_intention(intention_id, args)
        response = marshal(intention, intention_detail_fields)
        return response, 200

    @setup_required
    @login_required
    @account_initialization_required
    def delete(self, intention_id):
        IntentionService.delete_intention(intention_id)
        return 200


class IntentionTypeListApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    def get(self):
        page = request.args.get("page", default=1, type=int)
        limit = request.args.get("limit", default=20, type=int)
        search = request.args.get("search", default=None, type=str)
        intention_types, total = IntentionTypeService.get_intention_types(page, limit, search)
        data = marshal(intention_types, intention_type_page_fields)
        response = {
            "data": data,
            "has_more": len(intention_types) == limit,
            "limit": limit,
            "total": total,
            "page": page,
        }
        return response, 200

    @setup_required
    @login_required
    @account_initialization_required
    def post(self):
        parser = reqparse.RequestParser()
        parser.add_argument(
            "name",
            nullable=False,
            required=True,
            help="type is required. Name must be between 1 to 40 characters.",
        )
        args = parser.parse_args()
        intention_type = IntentionTypeService.save_intention_type(args)
        response = marshal(intention_type, intention_type_detail_fields)
        return response, 200


class IntentionTypeApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    def get(self, intention_type_id):
        intention_type = IntentionTypeService.get_intention_type(intention_type_id)
        return marshal(intention_type, intention_type_detail_fields)

    @setup_required
    @login_required
    @account_initialization_required
    def patch(self, intention_type_id):
        parser = reqparse.RequestParser()
        parser.add_argument(
            "name",
            nullable=False,
            required=True,
            help="type is required. Name must be between 1 to 40 characters.",
        )
        args = parser.parse_args()
        intention_type = IntentionTypeService.update_intention_type(intention_type_id, args)
        return marshal(intention_type, intention_type_detail_fields), 200

    @setup_required
    @login_required
    @account_initialization_required
    def delete(self, intention_type_id):
        IntentionTypeService.delete_intention_type(intention_type_id)
        return 200


class IntentionKeywordListApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    @marshal_with(intention_keyword_fields)
    def get(self, intention_id):
        search = request.args.get("search", default=None, type=str)
        intention = IntentionService.get_intention(intention_id)
        if not intention:
            raise NotFound("Intention not found")
        intention_keywords = IntentionKeywordService.get_intention_keywords(intention_id, search)
        return intention_keywords, 200

    @setup_required
    @login_required
    @account_initialization_required
    @marshal_with(intention_keyword_detail_fields)
    def post(self, intention_id):
        parser = reqparse.RequestParser()
        parser.add_argument(
            "name",
            nullable=False,
            required=True,
            help="type is required. Name must be between 1 to 40 characters.",
        )
        args = parser.parse_args()
        intention = IntentionService.get_intention(intention_id)
        if not intention:
            raise NotFound("Intention not found")
        intention_keyword = IntentionKeywordService.save_intention_keyword(intention_id, args)
        return intention_keyword, 200

    @setup_required
    @login_required
    @account_initialization_required
    def delete(self, intention_id):
        intention = IntentionService.get_intention(intention_id)
        if not intention:
            raise NotFound("Intention not found")

        IntentionKeywordService.delete_intention_keywords_by_intention_id(intention_id)
        return 200


class IntentionKeywordApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    def get(self, intention_keyword_id):
        intention_keyword = IntentionKeywordService.get_intention_keyword(intention_keyword_id)
        if not intention_keyword:
            return {}, 200
        return marshal(intention_keyword, intention_keyword_detail_fields), 200

    @setup_required
    @login_required
    @account_initialization_required
    @marshal_with(intention_keyword_detail_fields)
    def patch(self, intention_keyword_id):
        parser = reqparse.RequestParser()
        parser.add_argument(
            "name",
            nullable=False,
            required=True,
            help="type is required. Name must be between 1 to 40 characters.",
        )
        parser.add_argument(
            "intention_id",
            nullable=False,
            required=True,
            help="type is required.",
        )
        args = parser.parse_args()
        intention_keyword = IntentionKeywordService.update_intention_keyword(intention_keyword_id, args)
        return intention_keyword, 200

    @setup_required
    @login_required
    @account_initialization_required
    def delete(self, intention_keyword_id):
        IntentionKeywordService.delete_intention_keyword(intention_keyword_id)
        return 200


class IntentionKeywordBatchApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    def post(self):
        parser = reqparse.RequestParser()
        parser.add_argument(
            "method",
            nullable=False,
            required=True,
            help="method is required.",
            choices=["create", "update", "delete"],
            type=str,
            location="json",
        )
        parser.add_argument(
            "delete_data",
            nullable=False,
            required=True,
            help="delete_data is required.",
            type=list,
            location="json",
        )
        args = parser.parse_args()
        logging.info(args)

        method = args["method"]
        if method == "delete":
            intention_keyword_ids = args["delete_data"]
            IntentionKeywordService.delete_intention_keywords(intention_keyword_ids)
            return 200
        else:
            raise NotFound(f"method with name {method} not found")


class IntentionCorpusListApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    def get(self):
        page = request.args.get("page", default=1, type=int)
        limit = request.args.get("limit", default=20, type=int)
        question_search = request.args.get("question_search", default=None, type=str)
        intention_id = request.args.get("intention_id", default=None, type=str)
        intention_corpus, total = IntentionCorpusService.get_page_intention_corpus(
            page, limit, question_search, intention_id
        )
        data = marshal(intention_corpus, intention_corpus_detail_fields)
        response = {
            "data": data,
            "has_more": len(intention_corpus) == limit,
            "limit": limit,
            "total": total,
            "page": page,
        }
        return response, 200

    @setup_required
    @login_required
    @account_initialization_required
    def post(self):
        parser = reqparse.RequestParser()
        parser.add_argument(
            "question",
            nullable=False,
            required=True,
            help="type is required. Question must be between 1 to 40 characters.",
        )
        parser.add_argument(
            "question_config",
            nullable=True,
            required=False,
            location="json",
        )
        parser.add_argument(
            "intention_id",
            nullable=False,
            required=True,
            help="type is required.",
        )
        args = parser.parse_args()
        intention_corpus = IntentionCorpusService.save_intention_corpus(args)
        return marshal(intention_corpus, intention_corpus_detail_fields), 200


class IntentionCorpusApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    def get(self, corpus_id):
        intention_corpus = IntentionCorpusService.get_intention_corpus(corpus_id)
        if not intention_corpus:
            raise NotFound(f"IntentionCorpus with id {corpus_id} not found")
        return marshal(intention_corpus, intention_corpus_detail_fields), 200

    @setup_required
    @login_required
    @account_initialization_required
    def patch(self, corpus_id):
        parser = reqparse.RequestParser()
        parser.add_argument(
            "question",
            nullable=True,
            required=False,
            type=str,
            location="json",
        )
        parser.add_argument(
            "question_config",
            nullable=True,
            required=False,
            location="json",
        )
        parser.add_argument(
            "intention_id",
            nullable=True,
            required=False,
            type=str,
            location="json",
        )
        args = parser.parse_args()
        intention_corpus = IntentionCorpusService.update_intention_corpus(corpus_id, args)
        return marshal(intention_corpus, intention_corpus_detail_fields), 200

    @setup_required
    @login_required
    @account_initialization_required
    def delete(self, corpus_id):
        intention_corpus = IntentionCorpusService.get_intention_corpus(corpus_id)
        if not intention_corpus:
            raise NotFound(f"未发现Id未{corpus_id}的训练语料")

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

        IntentionCorpusService.delete_intention_corpus(intention_corpus)
        return 200


class IntentionCorpusSimilarityQuestionApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    def get(self, corpus_id):
        search = request.args.get("search", default=None, type=str)
        similarity_questions = (
            IntentionCorpusSimilarityQuestionService.get_similarity_questions_by_corpus_id_like_question(
                corpus_id, search
            )
        )
        return marshal(similarity_questions, intention_corpus_similarity_question_fields), 200

    @setup_required
    @login_required
    @account_initialization_required
    def post(self, corpus_id):
        parser = reqparse.RequestParser()
        parser.add_argument(
            "question",
            nullable=False,
            required=True,
            help="type is required. Question must be between 1 to 40 characters.",
            location="json",
        )
        parser.add_argument(
            "question_config",
            nullable=True,
            required=False,
            location="json",
        )
        args = parser.parse_args()
        intention_corpus_similarity_question = IntentionCorpusSimilarityQuestionService.save_similarity_question(
            corpus_id, args
        )
        return marshal(intention_corpus_similarity_question, intention_corpus_similarity_question_fields), 200

    @setup_required
    @login_required
    @account_initialization_required
    def delete(self, corpus_id):
        IntentionCorpusSimilarityQuestionService.delete_similarity_question_by_corpus_id(corpus_id)
        return 200


class IntentionCorpusSimilarityQuestionUpdateAndDeleteApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    def patch(self, similarity_question_id):
        parser = reqparse.RequestParser()
        parser.add_argument(
            "question",
            nullable=True,
            required=False,
            help="type is required. Question must be between 1 to 40 characters.",
            location="json",
        )
        parser.add_argument(
            "question_config",
            nullable=True,
            required=False,
            location="json",
        )
        parser.add_argument(
            "corpus_id",
            nullable=True,
            required=False,
            location="json",
        )
        args = parser.parse_args()

        similarity_question = IntentionCorpusSimilarityQuestionService.update_similarity_question(
            similarity_question_id, args
        )
        return marshal(similarity_question, intention_corpus_similarity_question_fields), 200

    @setup_required
    @login_required
    @account_initialization_required
    def delete(self, similarity_question_id):
        IntentionCorpusSimilarityQuestionService.delete_similarity_question_by_id(similarity_question_id)
        return 200


class IntentionCorpusSimilarityQuestionBatchApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    def post(self):
        parser = reqparse.RequestParser()
        parser.add_argument(
            "method",
            nullable=False,
            required=True,
            help="method is required.",
            choices=["create", "update", "delete"],
            type=str,
            location="json",
        )
        parser.add_argument(
            "data",
            nullable=False,
            required=True,
            help="data is required.",
            type=list,
            location="json",
        )
        args = parser.parse_args()
        logging.info(args)

        method = args["method"]
        if method == "delete":
            similarity_question_ids = args["data"]
            IntentionCorpusSimilarityQuestionService.delete_similarity_questions_by_ids(similarity_question_ids)
            return 200
        else:
            raise NotFound(f"method with name {method} not found")


api.add_resource(IntentionListApi, "/intentions")
api.add_resource(IntentionApi, "/intentions/<uuid:intention_id>")
api.add_resource(IntentionTypeListApi, "/intentions/types")
api.add_resource(IntentionTypeApi, "/intentions/types/<uuid:intention_type_id>")
api.add_resource(IntentionKeywordListApi, "/intentions/<uuid:intention_id>/keywords")
api.add_resource(IntentionKeywordApi, "/intentions/keywords/<uuid:intention_keyword_id>")
api.add_resource(IntentionKeywordBatchApi, "/intentions/keywords/batch")

api.add_resource(IntentionCorpusListApi, "/intentions/corpus")
api.add_resource(IntentionCorpusApi, "/intentions/corpus/<uuid:corpus_id>")
api.add_resource(IntentionCorpusSimilarityQuestionApi, "/intentions/corpus/<uuid:corpus_id>/similarity_questions")

api.add_resource(
    IntentionCorpusSimilarityQuestionUpdateAndDeleteApi,
    "/intentions/similarity_questions/<uuid:similarity_question_id>",
)
api.add_resource(IntentionCorpusSimilarityQuestionBatchApi, "/intentions/similarity_questions/batch")