(self, instance: Dict, with_valid=True)
| 1306 | tool_id = serializers.CharField(required=True, label=_("tool id")) |
| 1307 | |
| 1308 | def add(self, instance: Dict, with_valid=True): |
| 1309 | if with_valid: |
| 1310 | self.is_valid(raise_exception=True) |
| 1311 | AddInternalToolRequest(data=instance).is_valid(raise_exception=True) |
| 1312 | |
| 1313 | versions = instance.get("versions", []) |
| 1314 | download_url = instance.get("download_url") |
| 1315 | if not download_url.startswith("https://apps-assets.fit2cloud.com/"): |
| 1316 | raise AppApiException(500, _("Illegal download url")) |
| 1317 | # 查找匹配的版本名称 |
| 1318 | version_name = next( |
| 1319 | (version.get("name") for version in versions if version.get("downloadUrl") == download_url), |
| 1320 | ) |
| 1321 | res = requests.get(download_url, timeout=5) |
| 1322 | tool_data = RestrictedUnpickler(io.BytesIO(res.content)).load().tool |
| 1323 | tool_id = uuid.uuid7() |
| 1324 | # 如果是SKILL类型的工具，保存文件内容到file表，并将code替换为file_id |
| 1325 | if tool_data.get("tool_type") == ToolType.SKILL: |
| 1326 | skill_file_id = uuid.uuid7() |
| 1327 | skill_file = File( |
| 1328 | id=skill_file_id, |
| 1329 | file_name=f"{tool_data.get('name')}.zip", |
| 1330 | source_type=FileSourceType.TOOL, |
| 1331 | source_id=tool_id, |
| 1332 | meta={}, |
| 1333 | ) |
| 1334 | skill_file.save(base64.b64decode(tool_data.get("code"))) |
| 1335 | tool_data["code"] = skill_file_id |
| 1336 | tool = Tool( |
| 1337 | id=tool_id, |
| 1338 | name=instance.get("name"), |
| 1339 | desc=tool_data.get("desc"), |
| 1340 | code=tool_data.get("code"), |
| 1341 | user_id=self.data.get("user_id"), |
| 1342 | icon=instance.get("icon", ""), |
| 1343 | workspace_id=self.data.get("workspace_id"), |
| 1344 | input_field_list=tool_data.get("input_field_list", []), |
| 1345 | init_field_list=tool_data.get("init_field_list", []), |
| 1346 | scope=ToolScope.WORKSPACE, |
| 1347 | tool_type=tool_data.get("tool_type", ToolType.CUSTOM), |
| 1348 | folder_id=instance.get("folder_id", self.data.get("workspace_id")), |
| 1349 | template_id=self.data.get("tool_id"), |
| 1350 | label=instance.get("label"), |
| 1351 | version=version_name, |
| 1352 | is_active=False, |
| 1353 | ) |
| 1354 | tool.save() |
| 1355 | |
| 1356 | # 自动授权给创建者 |
| 1357 | UserResourcePermissionSerializer( |
| 1358 | data={ |
| 1359 | "workspace_id": self.data.get("workspace_id"), |
| 1360 | "user_id": self.data.get("user_id"), |
| 1361 | "auth_target_type": AuthTargetType.TOOL.value, |
| 1362 | } |
| 1363 | ).auth_resource(str(tool_id)) |
| 1364 | try: |
| 1365 | requests.get(instance.get("download_callback_url"), timeout=5) |
no test coverage detected