pikasTech-unidesk/scripts/src/ssh-remote-tools/skill-discover.py

#!/usr/bin/env python3
import argparse
import getpass
import json
import os
import platform
import socket
import sys
from datetime import datetime, timezone
from pathlib import Path


SKIP_PARTS = {"node_modules", ".git", ".state", "logs", "references", "__pycache__"}


def is_wsl():
    try:
        release = Path("/proc/sys/kernel/osrelease").read_text(errors="ignore").lower()
    except Exception:
        release = ""
    return "microsoft" in release or "wsl" in release or "WSL_INTEROP" in os.environ


def to_windows_path(path):
    text = str(path)
    if text.startswith("/mnt/") and len(text) >= 7 and text[5].isalpha() and text[6] == "/":
        drive = text[5].upper()
        rest = text[7:].replace("/", "\\")
        return drive + ":\\" + rest
    return None


def read_bounded(path, limit=16384):
    try:
        data = path.read_bytes()[:limit]
        return data.decode("utf-8", errors="replace")
    except Exception:
        return ""


def frontmatter_value(line):
    if ":" not in line:
        return None
    key, value = line.split(":", 1)
    return key.strip().lower(), value.strip().strip("\"'")


def parse_skill_metadata(skill_md):
    text = read_bounded(skill_md)
    name = skill_md.parent.name
    description = ""
    lines = text.splitlines()
    if lines and lines[0].strip() == "---":
        for line in lines[1:]:
            if line.strip() == "---":
                break
            item = frontmatter_value(line)
            if item is None:
                continue
            key, value = item
            if key == "name" and value:
                name = value
            if key == "description" and value:
                description = value
    if not description:
        for line in lines:
            stripped = line.strip()
            if stripped and not stripped.startswith("---") and not stripped.startswith("#"):
                description = stripped
                break
    return name, description


def iter_skill_files(root, max_depth):
    try:
        iterator = root.rglob("SKILL.md")
        for skill_md in iterator:
            try:
                rel = skill_md.relative_to(root)
            except ValueError:
                continue
            directory_parts = rel.parts[:-1]
            if len(directory_parts) == 0 or len(directory_parts) > max_depth:
                continue
            if any(part in SKIP_PARTS for part in directory_parts):
                continue
            yield skill_md
    except Exception as exc:
        raise RuntimeError(str(exc)) from exc


def unique_paths(paths):
    seen = set()
    output = []
    for raw in paths:
        path = Path(raw).expanduser()
        key = str(path)
        if key in seen:
            continue
        seen.add(key)
        output.append(path)
    return output


def default_wsl_roots():
    home = Path.home()
    roots = [home / ".agents" / "skills", home / ".codex" / "skills"]
    for raw in ("/root/.agents/skills", "/root/.codex/skills"):
        path = Path(raw)
        if str(path) not in {str(item) for item in roots}:
            roots.append(path)
    return roots


def default_windows_roots():
    if not is_wsl():
        return []
    users = Path("/mnt/c/Users")
    roots = []
    try:
        children = list(users.iterdir()) if users.exists() else []
    except Exception:
        children = []
    for child in children:
        try:
            if not child.is_dir():
                continue
        except Exception:
            continue
        lower = child.name.lower()
        if lower in {"all users", "default", "default user", "public"}:
            continue
        roots.append(child / ".agents" / "skills")
        roots.append(child / ".codex" / "skills")
    return roots


def scan_root(scope, root, max_depth):
    try:
        root_exists = root.exists()
        root_error = None
    except Exception as exc:
        root_exists = False
        root_error = str(exc)
    record = {
        "scope": scope,
        "path": str(root),
        "windowsPath": to_windows_path(root),
        "exists": root_exists,
        "skillCount": 0,
        "error": root_error,
    }
    skills = []
    if not record["exists"]:
        return record, skills
    try:
        for skill_md in iter_skill_files(root, max_depth):
            name, description = parse_skill_metadata(skill_md)
            skill = {
                "scope": scope,
                "name": name,
                "description": description,
                "path": str(skill_md.parent),
                "skillMd": str(skill_md),
                "windowsPath": to_windows_path(skill_md.parent),
                "root": str(root),
            }
            skills.append(skill)
    except Exception as exc:
        record["error"] = str(exc)
    record["skillCount"] = len(skills)
    return record, skills


def main():
    parser = argparse.ArgumentParser(description="discover WSL/Linux and Windows skill directories from a UniDesk ssh passthrough session")
    parser.add_argument("--scope", choices=["all", "wsl", "windows"], default="all", help="which skill roots to scan")
    parser.add_argument("--max-depth", type=int, default=4, help="maximum directory depth below each skill root")
    parser.add_argument("--limit", type=int, default=0, help="maximum skill rows to return; 0 means unlimited")
    parser.add_argument("--root", action="append", default=[], help="extra WSL/Linux skill root")
    parser.add_argument("--windows-root", action="append", default=[], help="extra Windows skill root, expressed as /mnt/<drive>/...")
    args = parser.parse_args()

    if args.max_depth <= 0:
        print(json.dumps({"ok": False, "error": "--max-depth must be positive"}, ensure_ascii=False))
        return 2
    if args.limit < 0:
        print(json.dumps({"ok": False, "error": "--limit must be >= 0"}, ensure_ascii=False))
        return 2

    roots = []
    if args.scope in ("all", "wsl"):
        roots.extend(("wsl", path) for path in default_wsl_roots())
        roots.extend(("wsl", Path(raw).expanduser()) for raw in args.root)
    if args.scope in ("all", "windows"):
        roots.extend(("windows", path) for path in default_windows_roots())
        roots.extend(("windows", Path(raw).expanduser()) for raw in args.windows_root)

    seen = set()
    unique = []
    for scope, path in roots:
        key = (scope, str(path))
        if key in seen:
            continue
        seen.add(key)
        unique.append((scope, path))

    root_records = []
    skills = []
    for scope, root in unique:
        record, found = scan_root(scope, root, args.max_depth)
        root_records.append(record)
        skills.extend(found)

    scope_order = {"wsl": 0, "windows": 1}
    skills.sort(key=lambda item: (scope_order.get(str(item["scope"]), 9), str(item["name"]).lower(), str(item["path"])))
    total_before_limit = len(skills)
    if args.limit > 0:
        skills = skills[:args.limit]

    counts = {"total": len(skills), "totalBeforeLimit": total_before_limit, "wsl": 0, "windows": 0}
    for skill in skills:
        scope = str(skill["scope"])
        if scope in counts:
            counts[scope] += 1

    payload = {
        "ok": True,
        "command": "unidesk ssh skills",
        "generatedAt": datetime.now(timezone.utc).isoformat(),
        "node": {
            "hostname": socket.gethostname(),
            "user": getpass.getuser(),
            "home": str(Path.home()),
            "platform": platform.platform(),
            "isWsl": is_wsl(),
            "python": sys.version.split()[0],
        },
        "counts": counts,
        "roots": root_records,
        "skills": skills,
    }
    print(json.dumps(payload, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())