Method edit

apps/knowledge/serializers/paragraph.py:222–288

(self, instance: Dict)

        @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 instance.get("content") is not None:
                _paragraph.chunks = text_to_chunk(instance.get("content", ""))

            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.uuid7(),
                            content=p.get("content"),
                            paragraph_id=self.data.get("paragraph_id"),
                            knowledge_id=self.data.get("knowledge_id"),
                            document_id=self.data.get("document_id"),
                        )
                        for p in create_problem_list
                    ]
                ) if len(create_problem_list) else None