paragraph_serializers.py

# coding=utf-8
"""
    @project: maxkb
    @Author：虎
    @file： paragraph_serializers.py
    @date：2023/10/16 15:51
    @desc:
"""
import uuid
from typing import Dict

from celery_once import AlreadyQueued
from django.db import transaction
from django.db.models import QuerySet, Count
from drf_yasg import openapi
from rest_framework import serializers

from common.db.search import page_search
from common.event import ListenerManagement
from common.exception.app_exception import AppApiException
from common.mixins.api_mixin import ApiMixin
from common.util.common import post
from common.util.field_message import ErrMessage
from dataset.models import Paragraph, Problem, Document, ProblemParagraphMapping, DataSet, TaskType, State
from dataset.serializers.common_serializers import update_document_char_length, BatchSerializer, ProblemParagraphObject, \
    ProblemParagraphManage, get_embedding_model_id_by_dataset_id
from dataset.serializers.problem_serializers import ProblemInstanceSerializer, ProblemSerializer, ProblemSerializers
from embedding.models import SourceType
from embedding.task.embedding import embedding_by_problem as embedding_by_problem_task, embedding_by_problem, \
    delete_embedding_by_source, enable_embedding_by_paragraph, disable_embedding_by_paragraph, embedding_by_paragraph, \
    delete_embedding_by_paragraph, delete_embedding_by_paragraph_ids, update_embedding_document_id
from dataset.task import generate_related_by_paragraph_id_list
from django.utils.translation import gettext_lazy as _


class ParagraphSerializer(serializers.ModelSerializer):
    class Meta:
        model = Paragraph
        fields = ['id', 'content', 'is_active', 'document_id', 'title',
                  'create_time', 'update_time']


class ParagraphInstanceSerializer(ApiMixin, serializers.Serializer):
    """
    段落实例对象
    """
    content = serializers.CharField(required=True, error_messages=ErrMessage.char(_('content')),
                                    max_length=102400,
                                    min_length=1,
                                    allow_null=True, allow_blank=True)

    title = serializers.CharField(required=False, max_length=256, error_messages=ErrMessage.char(_('section title')),
                                  allow_null=True, allow_blank=True)

    problem_list = ProblemInstanceSerializer(required=False, many=True)

    is_active = serializers.BooleanField(required=False, error_messages=ErrMessage.char(_('Is active')))

    @staticmethod
    def get_request_body_api():
        return openapi.Schema(
            type=openapi.TYPE_OBJECT,
            required=['content'],
            properties={
                'content': openapi.Schema(type=openapi.TYPE_STRING, max_length=4096, title=_('section content'),
                                          description=_('section content')),

                'title': openapi.Schema(type=openapi.TYPE_STRING, max_length=256, title=_('section title'),
                                        description=_('section title')),

                'is_active': openapi.Schema(type=openapi.TYPE_BOOLEAN, title=_('Is active'), description=_('Is active')),

                'problem_list': openapi.Schema(type=openapi.TYPE_ARRAY, title=_('problem list'),
                                               description=_('problem list'),
                                               items=ProblemInstanceSerializer.get_request_body_api())
            }
        )


class EditParagraphSerializers(serializers.Serializer):
    title = serializers.CharField(required=False, max_length=256, error_messages=ErrMessage.char(
        _('section title')), allow_null=True, allow_blank=True)
    content = serializers.CharField(required=False, max_length=102400, allow_null=True, allow_blank=True,
                                    error_messages=ErrMessage.char(
                                        _('section title')))
    problem_list = ProblemInstanceSerializer(required=False, many=True)


class ParagraphSerializers(ApiMixin, serializers.Serializer):
    title = serializers.CharField(required=False, max_length=256, error_messages=ErrMessage.char(
        _('section title')), allow_null=True, allow_blank=True)
    content = serializers.CharField(required=True, max_length=102400, error_messages=ErrMessage.char(
        _('section title')))

    class Problem(ApiMixin, serializers.Serializer):
        dataset_id = serializers.UUIDField(required=True, error_messages=ErrMessage.uuid(_('dataset id')))

        document_id = serializers.UUIDField(required=True, error_messages=ErrMessage.uuid(_('document id')))

        paragraph_id = serializers.UUIDField(required=True, error_messages=ErrMessage.uuid(_('paragraph id')))

        def is_valid(self, *, raise_exception=False):
            super().is_valid(raise_exception=True)
            if not QuerySet(Paragraph).filter(id=self.data.get('paragraph_id')).exists():
                raise AppApiException(500, _('Paragraph id does not exist'))

        def list(self, with_valid=False):
            """
            获取问题列表
            :param with_valid: 是否校验
            :return: 问题列表
            """
            if with_valid:
                self.is_valid(raise_exception=True)
            problem_paragraph_mapping = QuerySet(ProblemParagraphMapping).filter(dataset_id=self.data.get("dataset_id"),
                                                                                 paragraph_id=self.data.get(
                                                                                     'paragraph_id'))
            return [ProblemSerializer(row).data for row in
                    QuerySet(Problem).filter(id__in=[row.problem_id for row in problem_paragraph_mapping])]

        @transaction.atomic
        def save(self, instance: Dict, with_valid=True, with_embedding=True, embedding_by_problem=None):
            if with_valid:
                self.is_valid()
                ProblemInstanceSerializer(data=instance).is_valid(raise_exception=True)
            problem = QuerySet(Problem).filter(dataset_id=self.data.get('dataset_id'),
                                               content=instance.get('content')).first()
            if problem is None:
                problem = Problem(id=uuid.uuid1(), dataset_id=self.data.get('dataset_id'),
                                  content=instance.get('content'))
                problem.save()
            if QuerySet(ProblemParagraphMapping).filter(dataset_id=self.data.get('dataset_id'), problem_id=problem.id,
                                                        paragraph_id=self.data.get('paragraph_id')).exists():
                raise AppApiException(500, _('Already associated, please do not associate again'))
            problem_paragraph_mapping = ProblemParagraphMapping(id=uuid.uuid1(),
                                                                problem_id=problem.id,
                                                                document_id=self.data.get('document_id'),
                                                                paragraph_id=self.data.get('paragraph_id'),
                                                                dataset_id=self.data.get('dataset_id'))
            problem_paragraph_mapping.save()
            model_id = get_embedding_model_id_by_dataset_id(self.data.get('dataset_id'))
            if with_embedding:
                embedding_by_problem_task({'text': problem.content,
                                           'is_active': True,
                                           'source_type': SourceType.PROBLEM,
                                           'source_id': problem_paragraph_mapping.id,
                                           'document_id': self.data.get('document_id'),
                                           'paragraph_id': self.data.get('paragraph_id'),
                                           'dataset_id': self.data.get('dataset_id'),
                                           }, model_id)

            return ProblemSerializers.Operate(
                data={'dataset_id': self.data.get('dataset_id'),
                      'problem_id': problem.id}).one(with_valid=True)

        @staticmethod
        def get_request_params_api():
            return [openapi.Parameter(name='dataset_id',
                                      in_=openapi.IN_PATH,
                                      type=openapi.TYPE_STRING,
                                      required=True,
                                      description=_('dataset id')),
                    openapi.Parameter(name='document_id',
                                      in_=openapi.IN_PATH,
                                      type=openapi.TYPE_STRING,
                                      required=True,
                                      description=_('document id')),
                    openapi.Parameter(name='paragraph_id',
                                      in_=openapi.IN_PATH,
                                      type=openapi.TYPE_STRING,
                                      required=True,
                                      description=_('paragraph id'))]

        @staticmethod
        def get_request_body_api():
            return openapi.Schema(type=openapi.TYPE_OBJECT,
                                  required=["content"],
                                  properties={
                                      'content': openapi.Schema(
                                          type=openapi.TYPE_STRING, title=_('content'),)
                                  })

        @staticmethod
        def get_response_body_api():
            return openapi.Schema(
                type=openapi.TYPE_OBJECT,
                required=['id', 'content', 'hit_num', 'dataset_id', 'create_time', 'update_time'],
                properties={
                    'id': openapi.Schema(type=openapi.TYPE_STRING, title="id",
                                         description="id", default="xx"),
                    'content': openapi.Schema(type=openapi.TYPE_STRING, title=_('question content'),
                                              description=_('question content'), default=_('question content')),
                    'hit_num': openapi.Schema(type=openapi.TYPE_INTEGER, title=_('hit num'), description=_('hit num'),
                                              default=1),
                    'dataset_id': openapi.Schema(type=openapi.TYPE_STRING, title=_('dataset id'),
                                                 description=_('dataset id'), default='xxx'),
                    'update_time': openapi.Schema(type=openapi.TYPE_STRING, title=_('update time'),
                                                  description=_('update time'),
                                                  default="1970-01-01 00:00:00"),
                    'create_time': openapi.Schema(type=openapi.TYPE_STRING, title=_('create time'),
                                                  description=_('create time'),
                                                  default="1970-01-01 00:00:00"
                                                  )
                }
            )

    class Association(ApiMixin, serializers.Serializer):
        dataset_id = serializers.UUIDField(required=True, error_messages=ErrMessage.uuid(_('dataset id')))

        problem_id = serializers.UUIDField(required=True, error_messages=ErrMessage.uuid(_('problem id')))

        document_id = serializers.UUIDField(required=True, error_messages=ErrMessage.uuid(_('document id')))

        paragraph_id = serializers.UUIDField(required=True, error_messages=ErrMessage.uuid(_('paragraph id')))

        def is_valid(self, *, raise_exception=True):
            super().is_valid(raise_exception=True)
            dataset_id = self.data.get('dataset_id')
            paragraph_id = self.data.get('paragraph_id')
            problem_id = self.data.get("problem_id")
            if not QuerySet(Paragraph).filter(dataset_id=dataset_id, id=paragraph_id).exists():
                raise AppApiException(500, _('Paragraph does not exist'))
            if not QuerySet(Problem).filter(dataset_id=dataset_id, id=problem_id).exists():
                raise AppApiException(500, _('Problem does not exist'))

        def association(self, with_valid=True, with_embedding=True):
            if with_valid:
                self.is_valid(raise_exception=True)
            problem = QuerySet(Problem).filter(id=self.data.get("problem_id")).first()
            problem_paragraph_mapping = ProblemParagraphMapping(id=uuid.uuid1(),
                                                                document_id=self.data.get('document_id'),
                                                                paragraph_id=self.data.get('paragraph_id'),
                                                                dataset_id=self.data.get('dataset_id'),
                                                                problem_id=problem.id)
            problem_paragraph_mapping.save()
            if with_embedding:
                model_id = get_embedding_model_id_by_dataset_id(self.data.get('dataset_id'))
                embedding_by_problem({'text': problem.content,
                                      'is_active': True,
                                      'source_type': SourceType.PROBLEM,
                                      'source_id': problem_paragraph_mapping.id,
                                      'document_id': self.data.get('document_id'),
                                      'paragraph_id': self.data.get('paragraph_id'),
                                      'dataset_id': self.data.get('dataset_id'),
                                      }, model_id)

        def un_association(self, with_valid=True):
            if with_valid:
                self.is_valid(raise_exception=True)
            problem_paragraph_mapping = QuerySet(ProblemParagraphMapping).filter(
                paragraph_id=self.data.get('paragraph_id'),
                dataset_id=self.data.get('dataset_id'),
                problem_id=self.data.get(
                    'problem_id')).first()
            problem_paragraph_mapping_id = problem_paragraph_mapping.id
            problem_paragraph_mapping.delete()
            delete_embedding_by_source(problem_paragraph_mapping_id)
            return True

        @staticmethod
        def get_request_params_api():
            return [openapi.Parameter(name='dataset_id',
                                      in_=openapi.IN_PATH,
                                      type=openapi.TYPE_STRING,
                                      required=True,
                                      description=_('dataset id')),
                    openapi.Parameter(name='document_id',
                                      in_=openapi.IN_PATH,
                                      type=openapi.TYPE_STRING,
                                      required=True,
                                      description=_('document id'))
                , openapi.Parameter(name='paragraph_id',
                                    in_=openapi.IN_PATH,
                                    type=openapi.TYPE_STRING,
                                    required=True,
                                    description=_('paragraph id')),
                    openapi.Parameter(name='problem_id',
                                      in_=openapi.IN_PATH,
                                      type=openapi.TYPE_STRING,
                                      required=True,
                                      description=_('problem id'))
                    ]

    class Batch(serializers.Serializer):
        dataset_id = serializers.UUIDField(required=True, error_messages=ErrMessage.uuid(_('dataset id')))
        document_id = serializers.UUIDField(required=True, error_messages=ErrMessage.uuid(_('document id')))

        @transaction.atomic
        def batch_delete(self, instance: Dict, with_valid=True):
            if with_valid:
                BatchSerializer(data=instance).is_valid(model=Paragraph, raise_exception=True)
                self.is_valid(raise_exception=True)
            paragraph_id_list = instance.get("id_list")
            QuerySet(Paragraph).filter(id__in=paragraph_id_list).delete()
            delete_problems_and_mappings(paragraph_id_list)
            update_document_char_length(self.data.get('document_id'))
            # 删除向量库
            delete_embedding_by_paragraph_ids(paragraph_id_list)
            return True

    class Migrate(ApiMixin, serializers.Serializer):
        dataset_id = serializers.UUIDField(required=True, error_messages=ErrMessage.uuid(_('dataset id')))
        document_id = serializers.UUIDField(required=True, error_messages=ErrMessage.uuid(_('document id')))
        target_dataset_id = serializers.UUIDField(required=True, error_messages=ErrMessage.uuid(_('target dataset id')))
        target_document_id = serializers.UUIDField(required=True, error_messages=ErrMessage.uuid(_('target document id')))
        paragraph_id_list = serializers.ListField(required=True, error_messages=ErrMessage.char(_('paragraph id list')),
                                                  child=serializers.UUIDField(required=True,
                                                                              error_messages=ErrMessage.uuid(_('paragraph id'))))

        def is_valid(self, *, raise_exception=False):
            super().is_valid(raise_exception=True)
            document_list = QuerySet(Document).filter(
                id__in=[self.data.get('document_id'), self.data.get('target_document_id')])
            document_id = self.data.get('document_id')
            target_document_id = self.data.get('target_document_id')
            if document_id == target_document_id:
                raise AppApiException(5000, _('The document to be migrated is consistent with the target document'))
            if len([document for document in document_list if str(document.id) == self.data.get('document_id')]) < 1:
                raise AppApiException(5000, _('The document id does not exist [{document_id}]').format(
                    document_id=self.data.get('document_id')))
            if len([document for document in document_list if
                    str(document.id) == self.data.get('target_document_id')]) < 1:
                raise AppApiException(5000, _('The target document id does not exist [{document_id}]').format(
                    document_id=self.data.get('target_document_id')))

        @transaction.atomic
        def migrate(self, with_valid=True):
            if with_valid:
                self.is_valid(raise_exception=True)
            dataset_id = self.data.get('dataset_id')
            target_dataset_id = self.data.get('target_dataset_id')
            document_id = self.data.get('document_id')
            target_document_id = self.data.get('target_document_id')
            paragraph_id_list = self.data.get('paragraph_id_list')
            paragraph_list = QuerySet(Paragraph).filter(dataset_id=dataset_id, document_id=document_id,
                                                        id__in=paragraph_id_list)
            problem_paragraph_mapping_list = QuerySet(ProblemParagraphMapping).filter(paragraph__in=paragraph_list)
            # 同数据集迁移
            if target_dataset_id == dataset_id:
                if len(problem_paragraph_mapping_list):
                    problem_paragraph_mapping_list = [
                        self.update_problem_paragraph_mapping(target_document_id,
                                                              problem_paragraph_mapping) for problem_paragraph_mapping
                        in
                        problem_paragraph_mapping_list]
                    # 修改mapping
                    QuerySet(ProblemParagraphMapping).bulk_update(problem_paragraph_mapping_list,
                                                                  ['document_id'])
                update_embedding_document_id([paragraph.id for paragraph in paragraph_list],
                                             target_document_id, target_dataset_id, None)
                # 修改段落信息
                paragraph_list.update(document_id=target_document_id)
            # 不同数据集迁移
            else:
                problem_list = QuerySet(Problem).filter(
                    id__in=[problem_paragraph_mapping.problem_id for problem_paragraph_mapping in
                            problem_paragraph_mapping_list])
                # 目标数据集问题
                target_problem_list = list(
                    QuerySet(Problem).filter(content__in=[problem.content for problem in problem_list],
                                             dataset_id=target_dataset_id))

                target_handle_problem_list = [
                    self.get_target_dataset_problem(target_dataset_id, target_document_id, problem_paragraph_mapping,
                                                    problem_list, target_problem_list) for
                    problem_paragraph_mapping
                    in
                    problem_paragraph_mapping_list]

                create_problem_list = [problem for problem, is_create in target_handle_problem_list if
                                       is_create is not None and is_create]
                # 插入问题
                QuerySet(Problem).bulk_create(create_problem_list)
                # 修改mapping
                QuerySet(ProblemParagraphMapping).bulk_update(problem_paragraph_mapping_list,
                                                              ['problem_id', 'dataset_id', 'document_id'])
                target_dataset = QuerySet(DataSet).filter(id=target_dataset_id).first()
                dataset = QuerySet(DataSet).filter(id=dataset_id).first()
                embedding_model_id = None
                if target_dataset.embedding_mode_id != dataset.embedding_mode_id:
                    embedding_model_id = str(target_dataset.embedding_mode_id)
                pid_list = [paragraph.id for paragraph in paragraph_list]
                # 修改段落信息
                paragraph_list.update(dataset_id=target_dataset_id, document_id=target_document_id)
                # 修改向量段落信息
                update_embedding_document_id(pid_list, target_document_id, target_dataset_id, embedding_model_id)

            update_document_char_length(document_id)
            update_document_char_length(target_document_id)

        @staticmethod
        def update_problem_paragraph_mapping(target_document_id: str, problem_paragraph_mapping):
            problem_paragraph_mapping.document_id = target_document_id
            return problem_paragraph_mapping

        @staticmethod
        def get_target_dataset_problem(target_dataset_id: str,
                                       target_document_id: str,
                                       problem_paragraph_mapping,
                                       source_problem_list,
                                       target_problem_list):
            source_problem_list = [source_problem for source_problem in source_problem_list if
                                   source_problem.id == problem_paragraph_mapping.problem_id]
            problem_paragraph_mapping.dataset_id = target_dataset_id
            problem_paragraph_mapping.document_id = target_document_id
            if len(source_problem_list) > 0:
                problem_content = source_problem_list[-1].content
                problem_list = [problem for problem in target_problem_list if problem.content == problem_content]
                if len(problem_list) > 0:
                    problem = problem_list[-1]
                    problem_paragraph_mapping.problem_id = problem.id
                    return problem, False
                else:
                    problem = Problem(id=uuid.uuid1(), dataset_id=target_dataset_id, content=problem_content)
                    target_problem_list.append(problem)
                    problem_paragraph_mapping.problem_id = problem.id
                    return problem, True
            return None

        @staticmethod
        def get_request_params_api():
            return [openapi.Parameter(name='dataset_id',
                                      in_=openapi.IN_PATH,
                                      type=openapi.TYPE_STRING,
                                      required=True,
                                      description=_('document id')),
                    openapi.Parameter(name='document_id',
                                      in_=openapi.IN_PATH,
                                      type=openapi.TYPE_STRING,
                                      required=True,
                                      description=_('document id')),
                    openapi.Parameter(name='target_dataset_id',
                                      in_=openapi.IN_PATH,
                                      type=openapi.TYPE_STRING,
                                      required=True,
                                      description=_('target dataset id')),
                    openapi.Parameter(name='target_document_id',
                                      in_=openapi.IN_PATH,
                                      type=openapi.TYPE_STRING,
                                      required=True,
                                      description=_('target document id')),
                    ]

        @staticmethod
        def get_request_body_api():
            return openapi.Schema(
                type=openapi.TYPE_ARRAY,
                items=openapi.Schema(type=openapi.TYPE_STRING),
                title=_('paragraph id list'),
                description=_('paragraph id list')
            )

    class Operate(ApiMixin, serializers.Serializer):
        # 段落id
        paragraph_id = serializers.UUIDField(required=True, error_messages=ErrMessage.char(
            _('paragraph id')))
        # 知识库id
        dataset_id = serializers.UUIDField(required=True, error_messages=ErrMessage.char(
            _('dataset id')))
        # 文档id
        document_id = serializers.UUIDField(required=True, error_messages=ErrMessage.char(
            _('document id')))

        def is_valid(self, *, raise_exception=True):
            super().is_valid(raise_exception=True)
            if not QuerySet(Paragraph).filter(id=self.data.get('paragraph_id')).exists():
                raise AppApiException(500, _('Paragraph id does not exist'))

        @staticmethod
        def post_embedding(paragraph, instance, dataset_id):
            if 'is_active' in instance and instance.get('is_active') is not None:
                (enable_embedding_by_paragraph if instance.get(
                    'is_active') else disable_embedding_by_paragraph)(paragraph.get('id'))

            else:
                model_id = get_embedding_model_id_by_dataset_id(dataset_id)
                embedding_by_paragraph(paragraph.get('id'), model_id)
            return paragraph

        @post(post_embedding)
        @transaction.atomic
        def edit(self, instance: Dict):
            self.is_valid()
            EditParagraphSerializers(data=instance).is_valid(raise_exception=True)
            _paragraph = QuerySet(Paragraph).get(id=self.data.get("paragraph_id"))
            update_keys = ['title', 'content', 'is_active']
            for update_key in update_keys:
                if update_key in instance and instance.get(update_key) is not None:
                    _paragraph.__setattr__(update_key, instance.get(update_key))

            if 'problem_list' in instance:
                update_problem_list = list(
                    filter(lambda row: 'id' in row and row.get('id') is not None, instance.get('problem_list')))

                create_problem_list = list(filter(lambda row: row.get('id') is None, instance.get('problem_list')))

                # 问题集合
                problem_list = QuerySet(Problem).filter(paragraph_id=self.data.get("paragraph_id"))

                # 校验前端 携带过来的id
                for update_problem in update_problem_list:
                    if not set([str(row.id) for row in problem_list]).__contains__(update_problem.get('id')):
                        raise AppApiException(500, _('Problem id does not exist'))
                # 对比需要删除的问题
                delete_problem_list = list(filter(
                    lambda row: not [str(update_row.get('id')) for update_row in update_problem_list].__contains__(
                        str(row.id)), problem_list)) if len(update_problem_list) > 0 else []
                # 删除问题
                QuerySet(Problem).filter(id__in=[row.id for row in delete_problem_list]).delete() if len(
                    delete_problem_list) > 0 else None
                # 插入新的问题
                QuerySet(Problem).bulk_create(
                    [Problem(id=uuid.uuid1(), content=p.get('content'), paragraph_id=self.data.get('paragraph_id'),
                             dataset_id=self.data.get('dataset_id'), document_id=self.data.get('document_id')) for
                     p in create_problem_list]) if len(create_problem_list) else None

                # 修改问题集合
                QuerySet(Problem).bulk_update(
                    [Problem(id=row.get('id'), content=row.get('content')) for row in update_problem_list],
                    ['content']) if len(
                    update_problem_list) > 0 else None

            _paragraph.save()
            update_document_char_length(self.data.get('document_id'))
            return self.one(), instance, self.data.get('dataset_id')

        def get_problem_list(self):
            ProblemParagraphMapping(ProblemParagraphMapping)
            problem_paragraph_mapping = QuerySet(ProblemParagraphMapping).filter(
                paragraph_id=self.data.get("paragraph_id"))
            if len(problem_paragraph_mapping) > 0:
                return [ProblemSerializer(problem).data for problem in
                        QuerySet(Problem).filter(id__in=[ppm.problem_id for ppm in problem_paragraph_mapping])]
            return []

        def one(self, with_valid=False):
            if with_valid:
                self.is_valid(raise_exception=True)
            return {**ParagraphSerializer(QuerySet(model=Paragraph).get(id=self.data.get('paragraph_id'))).data,
                    'problem_list': self.get_problem_list()}

        def delete(self, with_valid=False):
            if with_valid:
                self.is_valid(raise_exception=True)
            paragraph_id = self.data.get('paragraph_id')
            Paragraph.objects.filter(id=paragraph_id).delete()
            delete_problems_and_mappings([paragraph_id])

            update_document_char_length(self.data.get('document_id'))
            delete_embedding_by_paragraph(paragraph_id)

        @staticmethod
        def get_request_body_api():
            return ParagraphInstanceSerializer.get_request_body_api()

        @staticmethod
        def get_response_body_api():
            return ParagraphInstanceSerializer.get_request_body_api()

        @staticmethod
        def get_request_params_api():
            return [openapi.Parameter(type=openapi.TYPE_STRING, in_=openapi.IN_PATH, name='paragraph_id',
                                      description=_('paragraph id'))]

    class Create(ApiMixin, serializers.Serializer):
        dataset_id = serializers.UUIDField(required=True, error_messages=ErrMessage.char(
            _('dataset id')))

        document_id = serializers.UUIDField(required=True, error_messages=ErrMessage.char(
            _('document id')))

        def is_valid(self, *, raise_exception=False):
            super().is_valid(raise_exception=True)
            if not QuerySet(Document).filter(id=self.data.get('document_id'),
                                             dataset_id=self.data.get('dataset_id')).exists():
                raise AppApiException(500, _('The document id is incorrect'))

        def save(self, instance: Dict, with_valid=True, with_embedding=True):
            if with_valid:
                ParagraphSerializers(data=instance).is_valid(raise_exception=True)
                self.is_valid()
            dataset_id = self.data.get("dataset_id")
            document_id = self.data.get('document_id')
            paragraph_problem_model = self.get_paragraph_problem_model(dataset_id, document_id, instance)
            paragraph = paragraph_problem_model.get('paragraph')
            problem_paragraph_object_list = paragraph_problem_model.get('problem_paragraph_object_list')
            problem_model_list, problem_paragraph_mapping_list = (ProblemParagraphManage(problem_paragraph_object_list,
                                                                                         dataset_id).
                                                                  to_problem_model_list())
            # 插入段落
            paragraph_problem_model.get('paragraph').save()
            # 插入問題
            QuerySet(Problem).bulk_create(problem_model_list) if len(problem_model_list) > 0 else None
            # 插入问题关联关系
            QuerySet(ProblemParagraphMapping).bulk_create(problem_paragraph_mapping_list) if len(
                problem_paragraph_mapping_list) > 0 else None
            # 修改长度
            update_document_char_length(document_id)
            if with_embedding:
                model_id = get_embedding_model_id_by_dataset_id(dataset_id)
                embedding_by_paragraph(str(paragraph.id), model_id)
            return ParagraphSerializers.Operate(
                data={'paragraph_id': str(paragraph.id), 'dataset_id': dataset_id, 'document_id': document_id}).one(
                with_valid=True)

        @staticmethod
        def get_paragraph_problem_model(dataset_id: str, document_id: str, instance: Dict):
            paragraph = Paragraph(id=uuid.uuid1(),
                                  document_id=document_id,
                                  content=instance.get("content"),
                                  dataset_id=dataset_id,
                                  title=instance.get("title") if 'title' in instance else '')
            problem_paragraph_object_list = [
                ProblemParagraphObject(dataset_id, document_id, paragraph.id, problem.get('content')) for problem in
                (instance.get('problem_list') if 'problem_list' in instance else [])]

            return {'paragraph': paragraph,
                    'problem_paragraph_object_list': problem_paragraph_object_list}

        @staticmethod
        def or_get(exists_problem_list, content, dataset_id):
            exists = [row for row in exists_problem_list if row.content == content]
            if len(exists) > 0:
                return exists[0]
            else:
                return Problem(id=uuid.uuid1(), content=content, dataset_id=dataset_id)

        @staticmethod
        def get_request_body_api():
            return ParagraphInstanceSerializer.get_request_body_api()

        @staticmethod
        def get_request_params_api():
            return [openapi.Parameter(name='dataset_id',
                                      in_=openapi.IN_PATH,
                                      type=openapi.TYPE_STRING,
                                      required=True,
                                      description=_('dataset id')),
                    openapi.Parameter(name='document_id', in_=openapi.IN_PATH,
                                      type=openapi.TYPE_STRING,
                                      required=True,
                                      description=_('document id'))
                    ]

    class Query(ApiMixin, serializers.Serializer):
        dataset_id = serializers.UUIDField(required=True, error_messages=ErrMessage.char(
            _('dataset id')))

        document_id = serializers.UUIDField(required=True, error_messages=ErrMessage.char(
            _('document id')))

        title = serializers.CharField(required=False, error_messages=ErrMessage.char(
            _('section title')))

        content = serializers.CharField(required=False)

        def get_query_set(self):
            query_set = QuerySet(model=Paragraph)
            query_set = query_set.filter(
                **{'dataset_id': self.data.get('dataset_id'), 'document_id': self.data.get("document_id")})
            if 'title' in self.data:
                query_set = query_set.filter(
                    **{'title__icontains': self.data.get('title')})
            if 'content' in self.data:
                query_set = query_set.filter(**{'content__icontains': self.data.get('content')})
            query_set.order_by('-create_time', 'id')
            return query_set

        def list(self):
            return list(map(lambda row: ParagraphSerializer(row).data, self.get_query_set()))

        def page(self, current_page, page_size):
            query_set = self.get_query_set()
            return page_search(current_page, page_size, query_set, lambda row: ParagraphSerializer(row).data)

        @staticmethod
        def get_request_params_api():
            return [openapi.Parameter(name='document_id',
                                      in_=openapi.IN_PATH,
                                      type=openapi.TYPE_STRING,
                                      required=True,
                                      description=_('document id')),
                    openapi.Parameter(name='title',
                                      in_=openapi.IN_QUERY,
                                      type=openapi.TYPE_STRING,
                                      required=False,
                                      description=_('title')),
                    openapi.Parameter(name='content',
                                      in_=openapi.IN_QUERY,
                                      type=openapi.TYPE_STRING,
                                      required=False,
                                      description=_('content'))
                    ]

        @staticmethod
        def get_response_body_api():
            return openapi.Schema(
                type=openapi.TYPE_OBJECT,
                required=['id', 'content', 'hit_num', 'star_num', 'trample_num', 'is_active', 'dataset_id',
                          'document_id', 'title',
                          'create_time', 'update_time'],
                properties={
                    'id': openapi.Schema(type=openapi.TYPE_STRING, title="id",
                                         description="id", default="xx"),
                    'content': openapi.Schema(type=openapi.TYPE_STRING, title=_('content'),
                                              description=_('content'), default=_('content')),
                    'title': openapi.Schema(type=openapi.TYPE_STRING, title=_('title'),
                                            description=_('title'), default="xxx"),
                    'hit_num': openapi.Schema(type=openapi.TYPE_INTEGER, title=_('hit num'), description=_('hit num'),
                                              default=1),
                    'star_num': openapi.Schema(type=openapi.TYPE_INTEGER, title=_('Number of likes'),
                                               description=_('Number of likes'), default=1),
                    'trample_num': openapi.Schema(type=openapi.TYPE_INTEGER, title=_('Number of dislikes'),
                                                  description=_('Number of dislikes'), default=1),
                    'dataset_id': openapi.Schema(type=openapi.TYPE_STRING, title=_('dataset id'),
                                                 description=_('dataset id'), default='xxx'),
                    'document_id': openapi.Schema(type=openapi.TYPE_STRING, title=_('document id'),
                                                  description=_('document id'), default='xxx'),
                    'is_active': openapi.Schema(type=openapi.TYPE_BOOLEAN, title=_('Is active'),
                                                description=_('Is active'), default=True),
                    'update_time': openapi.Schema(type=openapi.TYPE_STRING, title=_('update time'),
                                                  description=_('update time'),
                                                  default="1970-01-01 00:00:00"),
                    'create_time': openapi.Schema(type=openapi.TYPE_STRING, title=_('create time'),
                                                  description=_('create time'),
                                                  default="1970-01-01 00:00:00"
                                                  )
                }
            )

    class BatchGenerateRelated(ApiMixin, serializers.Serializer):
        dataset_id = serializers.UUIDField(required=True, error_messages=ErrMessage.uuid(_('dataset id')))
        document_id = serializers.UUIDField(required=True, error_messages=ErrMessage.uuid(_('document id')))

        def batch_generate_related(self, instance: Dict, with_valid=True):
            if with_valid:
                self.is_valid(raise_exception=True)
            paragraph_id_list = instance.get("paragraph_id_list")
            model_id = instance.get("model_id")
            prompt = instance.get("prompt")
            document_id = self.data.get('document_id')
            ListenerManagement.update_status(QuerySet(Document).filter(id=document_id),
                                             TaskType.GENERATE_PROBLEM,
                                             State.PENDING)
            ListenerManagement.update_status(QuerySet(Paragraph).filter(id__in=paragraph_id_list),
                                             TaskType.GENERATE_PROBLEM,
                                             State.PENDING)
            ListenerManagement.get_aggregation_document_status(document_id)()
            try:
                generate_related_by_paragraph_id_list.delay(document_id, paragraph_id_list, model_id,
                                                            prompt)
            except AlreadyQueued as e:
                raise AppApiException(500, _('The task is being executed, please do not send it again.'))


def delete_problems_and_mappings(paragraph_ids):
    problem_paragraph_mappings = ProblemParagraphMapping.objects.filter(paragraph_id__in=paragraph_ids)
    problem_ids = set(problem_paragraph_mappings.values_list('problem_id', flat=True))

    if problem_ids:
        problem_paragraph_mappings.delete()
        remaining_problem_counts = ProblemParagraphMapping.objects.filter(problem_id__in=problem_ids).values(
            'problem_id').annotate(count=Count('problem_id'))
        remaining_problem_ids = {pc['problem_id'] for pc in remaining_problem_counts}
        problem_ids_to_delete = problem_ids - remaining_problem_ids
        Problem.objects.filter(id__in=problem_ids_to_delete).delete()
    else:
        problem_paragraph_mappings.delete()