| 1 | from __future__ import annotations |
| 2 | |
| 3 | import shutil |
| 4 | import time |
| 5 | import uuid |
| 6 | from pathlib import Path |
| 7 | from typing import Any |
| 8 | |
| 9 | from helpers import files, skills |
| 10 | from helpers.api import ApiHandler, Request, Response |
| 11 | from helpers.skills_import import extract_skills_zip |
| 12 | from werkzeug.datastructures import FileStorage |
| 13 | from werkzeug.utils import secure_filename |
| 14 | |
| 15 | |
| 16 | class SkillsScan(ApiHandler): |
| 17 | """ |
| 18 | Prepare skill scan targets for the Settings > Skills scanner. |
| 19 | """ |
| 20 | |
| 21 | async def process(self, input: dict[str, Any], request: Request) -> dict[str, Any] | Response: |
| 22 | if "skills_file" in request.files: |
| 23 | return self._prepare_uploaded_archive(request.files["skills_file"]) |
| 24 | |
| 25 | action = str(input.get("action") or "targets").strip().lower() |
| 26 | if action == "targets": |
| 27 | return self._list_installed_targets() |
| 28 | |
| 29 | return {"success": False, "error": "Invalid action"} |
| 30 | |
| 31 | def _list_installed_targets(self) -> dict[str, Any]: |
| 32 | targets: list[dict[str, Any]] = [] |
| 33 | seen: set[str] = set() |
| 34 | total_skills = 0 |
| 35 | |
| 36 | for raw_root in skills.get_skill_roots(): |
| 37 | root = Path(raw_root) |
| 38 | if not root.is_dir(): |
| 39 | continue |
| 40 | |
| 41 | skill_files = skills.discover_skill_md_files(root) |
| 42 | if not skill_files: |
| 43 | continue |
| 44 | |
| 45 | key = str(root.resolve()) |
| 46 | if key in seen: |
| 47 | continue |
| 48 | seen.add(key) |
| 49 | |
| 50 | skill_count = len(skill_files) |
| 51 | total_skills += skill_count |
| 52 | targets.append( |
| 53 | { |
| 54 | "path": str(root), |
| 55 | "display_path": files.normalize_a0_path(str(root)), |
| 56 | "skill_count": skill_count, |
| 57 | } |
| 58 | ) |
| 59 | |
| 60 | targets.sort(key=lambda item: item["path"]) |
| 61 | return { |
| 62 | "success": True, |
| 63 | "target_type": "installed", |
| 64 | "target_label": "Installed Agent Zero skills", |
| 65 | "targets": targets, |
| 66 | "paths": [item["path"] for item in targets], |
| 67 | "skill_count": total_skills, |
| 68 | } |
| 69 | |
| 70 | def _prepare_uploaded_archive(self, skills_file: FileStorage) -> dict[str, Any]: |
| 71 | if not skills_file.filename: |
| 72 | return {"success": False, "error": "No file selected"} |
| 73 | |
| 74 | base = secure_filename(skills_file.filename) # type: ignore[arg-type] |
| 75 | if not base.lower().endswith(".zip"): |
| 76 | return {"success": False, "error": "Skill scan uploads must be .zip files"} |
| 77 | |
| 78 | tmp_dir = Path(files.get_abs_path("tmp", "uploads")) |
| 79 | tmp_dir.mkdir(parents=True, exist_ok=True) |
| 80 | unique = uuid.uuid4().hex[:8] |
| 81 | stamp = time.strftime("%Y%m%d_%H%M%S") |
| 82 | tmp_path = tmp_dir / f"skills_scan_{stamp}_{unique}_{base}" |
| 83 | skills_file.save(str(tmp_path)) |
| 84 | |
| 85 | cleanup_root: Path | None = None |
| 86 | try: |
| 87 | scan_root, cleanup_root = extract_skills_zip( |
| 88 | tmp_path, |
| 89 | tmp_subdir="skill_scans", |
| 90 | prefix=f"scan_{unique}", |
| 91 | ) |
| 92 | skill_files = skills.discover_skill_md_files(scan_root) |
| 93 | skill_entries = [ |
| 94 | { |
| 95 | "path": str(skill_md.parent), |
| 96 | "relative_path": str(skill_md.parent.relative_to(scan_root)), |
| 97 | } |
| 98 | for skill_md in skill_files |
| 99 | ] |
| 100 | warnings = [] |
| 101 | if not skill_entries: |
| 102 | warnings.append("No SKILL.md files were found in the uploaded archive.") |
| 103 | |
| 104 | return { |
| 105 | "success": True, |
| 106 | "target_type": "uploaded_archive", |
| 107 | "target_label": base, |
| 108 | "paths": [str(scan_root)], |
| 109 | "scan_path": str(scan_root), |
| 110 | "display_path": files.normalize_a0_path(str(scan_root)), |
| 111 | "cleanup_paths": [str(cleanup_root)], |
| 112 | "skill_count": len(skill_entries), |
| 113 | "skills": skill_entries, |
| 114 | "warnings": warnings, |
| 115 | } |
| 116 | except Exception as exc: |
| 117 | if cleanup_root: |
| 118 | shutil.rmtree(cleanup_root, ignore_errors=True) |
| 119 | return {"success": False, "error": f"Failed to prepare skill scan: {exc}"} |
| 120 | finally: |
| 121 | try: |
| 122 | tmp_path.unlink(missing_ok=True) # type: ignore[arg-type] |
| 123 | except Exception: |
| 124 | pass |