Method edit

apps/tools/serializers/tool.py:641–700

(self, instance, with_valid=True)

        @transaction.atomic
        def edit(self, instance, with_valid=True):
            if with_valid:
                self.is_valid(raise_exception=True)
                ToolEditRequest(data=instance).is_valid(raise_exception=True)
                # 校验代码是否包括禁止的关键字
                if instance.get("tool_type") == ToolType.MCP:
                    ToolExecutor().validate_mcp_transport(instance.get("code", ""))

            if not QuerySet(Tool).filter(id=self.data.get("id")).exists():
                raise serializers.ValidationError(_("Tool not found"))

            edit_field_list = [
                "name",
                "desc",
                "code",
                "icon",
                "input_field_list",
                "init_field_list",
                "init_params",
                "is_active",
                "folder_id",
            ]
            edit_dict = {
                field: instance.get(field)
                for field in edit_field_list
                if (field in instance and instance.get(field) is not None)
            }

            tool = QuerySet(Tool).filter(id=self.data.get("id")).first()
            if "init_params" in edit_dict:
                if edit_dict["init_field_list"] is not None:
                    rm_key = []
                    for key in edit_dict["init_params"]:
                        if key not in [field["field"] for field in edit_dict["init_field_list"]]:
                            rm_key.append(key)
                    for key in rm_key:
                        edit_dict["init_params"].pop(key)
                if tool.init_params:
                    old_init_params = json.loads(rsa_long_decrypt(tool.init_params))
                    for key in edit_dict["init_params"]:
                        if key in old_init_params and edit_dict["init_params"][key] == encryption(old_init_params[key]):
                            edit_dict["init_params"][key] = old_init_params[key]
                edit_dict["init_params"] = rsa_long_encrypt(json.dumps(edit_dict["init_params"]))

            edit_dict["update_time"] = timezone.now()
            QuerySet(Tool).filter(id=self.data.get("id")).update(**edit_dict)
            if "is_active" in instance:
                QuerySet(TriggerTask).filter(source_type="TOOL", source_id=self.data.get("id")).update(
                    is_active=instance.get("is_active")
                )

            # 如果是SKILL类型的工具，修改file表中对应的记录
            if instance.get("tool_type") == ToolType.SKILL:
                old_file_id = tool.code
                file_id = instance.get("code")
                if old_file_id != file_id:
                    QuerySet(File).filter(id=old_file_id).delete()
                    QuerySet(File).filter(id=file_id).update(source_id=tool.id, source_type=FileSourceType.TOOL)