| 1 | from __future__ import annotations |
| 2 | |
| 3 | from helpers.api import ApiHandler, Input, Output, Request |
| 4 | from werkzeug.datastructures import FileStorage |
| 5 | |
| 6 | from plugins._plugin_installer.helpers.install import ( |
| 7 | get_plugin_hub_index, |
| 8 | install_from_git, |
| 9 | install_uploaded_zip, |
| 10 | update_from_git, |
| 11 | ) |
| 12 | |
| 13 | class PluginInstall(ApiHandler): |
| 14 | """Plugin installation API. Handles ZIP upload, Git clone, and index fetch.""" |
| 15 | |
| 16 | async def process(self, input: Input, request: Request) -> Output: |
| 17 | action = input.get("action", "") or request.form.get("action", "") |
| 18 | |
| 19 | try: |
| 20 | if action == "install_zip": |
| 21 | return self._install_zip(request) |
| 22 | elif action == "install_git": |
| 23 | return self._install_git(input) |
| 24 | elif action == "update_plugin": |
| 25 | return self._update_git(input) |
| 26 | elif action == "fetch_index": |
| 27 | return self._fetch_index(input) |
| 28 | else: |
| 29 | return {"success": False, "error": f"Unknown action: {action}"} |
| 30 | except ValueError as e: |
| 31 | return {"success": False, "error": str(e)} |
| 32 | except Exception as e: |
| 33 | return {"success": False, "error": f"Installation failed: {e}"} |
| 34 | |
| 35 | def _install_zip(self, request: Request) -> dict: |
| 36 | if "plugin_file" not in request.files: |
| 37 | return {"success": False, "error": "No file provided"} |
| 38 | |
| 39 | plugin_file: FileStorage = request.files["plugin_file"] |
| 40 | if not plugin_file.filename: |
| 41 | return {"success": False, "error": "No file selected"} |
| 42 | |
| 43 | return install_uploaded_zip(plugin_file) |
| 44 | |
| 45 | def _install_git(self, input: dict) -> dict: |
| 46 | git_url = (input.get("git_url", "") or "").strip() |
| 47 | git_token = (input.get("git_token", "") or "").strip() or None |
| 48 | plugin_name = input.get("plugin_name", "") |
| 49 | thumbnail_url = (input.get("thumbnail_url") or "").strip() |
| 50 | if not git_url: |
| 51 | return {"success": False, "error": "Git URL is required"} |
| 52 | |
| 53 | return install_from_git(url=git_url, token=git_token, plugin_name=plugin_name, thumbnail_url=thumbnail_url) |
| 54 | |
| 55 | def _update_git(self, input: dict) -> dict: |
| 56 | return update_from_git(input.get("plugin_name", "")) |
| 57 | |
| 58 | def _fetch_index(self, input: dict) -> dict: |
| 59 | return {"success": True, **get_plugin_hub_index()} |