(self, with_valid=True)
| 1377 | versions = serializers.ListField(required=True, label=_("versions"), child=serializers.DictField()) |
| 1378 | |
| 1379 | def update_tool(self, with_valid=True): |
| 1380 | if with_valid: |
| 1381 | self.is_valid(raise_exception=True) |
| 1382 | tool = QuerySet(Tool).filter(id=self.data.get("tool_id")).first() |
| 1383 | if tool is None: |
| 1384 | raise AppApiException(500, _("Tool does not exist")) |
| 1385 | # 查找匹配的版本名称 |
| 1386 | version_name = next( |
| 1387 | ( |
| 1388 | version.get("name") |
| 1389 | for version in self.data.get("versions") |
| 1390 | if version.get("downloadUrl") == self.data.get("download_url") |
| 1391 | ), |
| 1392 | ) |
| 1393 | res = requests.get(self.data.get("download_url"), timeout=5) |
| 1394 | tool_data = RestrictedUnpickler(io.BytesIO(res.content)).load().tool |
| 1395 | # 如果是SKILL类型的工具，保存文件内容到file表，并将code替换为file_id |
| 1396 | if tool_data.get("tool_type") == ToolType.SKILL: |
| 1397 | skill_file_id = uuid.uuid7() |
| 1398 | skill_file = File( |
| 1399 | id=skill_file_id, |
| 1400 | file_name=f"{tool_data.get('name')}.zip", |
| 1401 | source_type=FileSourceType.TOOL, |
| 1402 | source_id=tool.id, |
| 1403 | meta={}, |
| 1404 | ) |
| 1405 | skill_file.save(base64.b64decode(tool_data.get("code"))) |
| 1406 | tool_data["code"] = skill_file_id |
| 1407 | tool.desc = tool_data.get("desc") |
| 1408 | tool.code = tool_data.get("code") |
| 1409 | tool.input_field_list = tool_data.get("input_field_list", []) |
| 1410 | tool.init_field_list = tool_data.get("init_field_list", []) |
| 1411 | tool.icon = self.data.get("icon", tool.icon) |
| 1412 | tool.version = version_name |
| 1413 | # tool.is_active = False |
| 1414 | tool.save() |
| 1415 | try: |
| 1416 | requests.get(self.data.get("download_callback_url"), timeout=5) |
| 1417 | except Exception as e: |
| 1418 | maxkb_logger.error(f"callback appstore tool download error: {e}") |
| 1419 | return ToolModelSerializer(tool).data |
| 1420 | |
| 1421 | class ToolRecord(serializers.Serializer): |
| 1422 | workspace_id = serializers.CharField(required=False, allow_null=True, label=_("workspace id")) |
no test coverage detected