Skills Updater

#!/usr/bin/env python3
"""
Skill Update Checker - Scans installed skills and checks for available updates.

Usage:
    python check_updates.py [--skill <name>] [--json]

Examples:
    python check_updates.py                    # Check all installed skills
    python check_updates.py --skill skill-creator  # Check specific skill
    python check_updates.py --json             # Output as JSON
"""

import json
import sys
import argparse
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
import urllib.request
import urllib.error


class UpdateStatus(Enum):
    UP_TO_DATE = "up_to_date"
    UPDATE_AVAILABLE = "update_available"
    UNKNOWN_VERSION = "unknown_version"
    ERROR = "error"


@dataclass
class SkillInfo:
    name: str
    marketplace: str
    local_version: str
    remote_version: Optional[str]
    status: UpdateStatus
    install_path: str
    git_commit_sha: Optional[str] = None
    remote_commit_sha: Optional[str] = None
    error_message: Optional[str] = None


def get_plugins_dir() -> Path:
    """Get the Claude Code plugins directory."""
    return Path.home() / ".claude" / "plugins"


def get_npx_skills_dir() -> Optional[Path]:
    """Get the npx skills directory if it exists."""
    skills_dir = Path.home() / ".skills"
    return skills_dir if skills_dir.exists() else None


def load_installed_plugins() -> Dict:
    """Load the installed_plugins.json file."""
    plugins_file = get_plugins_dir() / "installed_plugins.json"
    if not plugins_file.exists():
        return {"version": 2, "plugins": {}}

    with open(plugins_file) as f:
        return json.load(f)


def load_known_marketplaces() -> Dict:
    """Load the known_marketplaces.json file."""
    marketplaces_file = get_plugins_dir() / "known_marketplaces.json"
    if not marketplaces_file.exists():
        return {}

    with open(marketplaces_file) as f:
        return json.load(f)


def parse_plugin_key(key: str) -> Tuple[str, str]:
    """Parse plugin key into (skill_name, marketplace)."""
    parts = key.rsplit("@", 1)
    if len(parts) == 2:
        return parts[0], parts[1]
    return key, "unknown"


def get_github_repo_from_marketplace(marketplace_name: str, marketplaces: Dict) -> Optional[str]:
    """Get the GitHub repo from marketplace info."""
    marketplace_info = marketplaces.get(marketplace_name, {})
    source = marketplace_info.get("source", {})

    if source.get("source") == "github":
        return source.get("repo")
    elif source.get("source") == "git":
        url = source.get("url", "")
        # Parse git URL to get owner/repo
        if "github.com" in url:
            # Handle formats: https://github.com/owner/repo.git or git@github.com:owner/repo.git
            url = url.replace(".git", "")
            if "github.com/" in url:
                return url.split("github.com/")[-1]
            elif "github.com:" in url:
                return url.split("github.com:")[-1]

    return None


def fetch_remote_marketplace_json(repo: str) -> Optional[Dict]:
    """Fetch marketplace.json from GitHub repo."""
    url = f"https://raw.githubusercontent.com/{repo}/main/.claude-plugin/marketplace.json"

    try:
        req = urllib.request.Request(url, headers={"User-Agent": "skills-updater/1.0"})
        with urllib.request.urlopen(req, timeout=10) as response:
            return json.loads(response.read().decode())
    except urllib.error.HTTPError as e:
        if e.code == 404:
            # Try HEAD branch instead of main
            url_head = url.replace("/main/", "/HEAD/")
            try:
                req = urllib.request.Request(url_head, headers={"User-Agent": "skills-updater/1.0"})
                with urllib.request.urlopen(req, timeout=10) as response:
                    return json.loads(response.read().decode())
            except:
                pass
        return None
    except Exception:
        return None


def fetch_remote_commit_sha(repo: str) -> Optional[str]:
    """Fetch the latest commit SHA from GitHub."""
    url = f"https://api.github.com/repos/{repo}/commits/main"

    try:
        req = urllib.request.Request(url, headers={
            "User-Agent": "skills-updater/1.0",
            "Accept": "application/vnd.github.v3+json"
        })
        with urllib.request.urlopen(req, timeout=10) as response:
            data = json.loads(response.read().decode())
            return data.get("sha")
    except:
        # Try HEAD branch
        url_head = url.replace("/main", "/HEAD")
        try:
            req = urllib.request.Request(url_head, headers={
                "User-Agent": "skills-updater/1.0",
                "Accept": "application/vnd.github.v3+json"
            })
            with urllib.request.urlopen(req, timeout=10) as response:
                data = json.loads(response.read().decode())
                return data.get("sha")
        except:
            return None


def get_skill_version_from_marketplace_json(marketplace_json: Dict, skill_name: str) -> Optional[str]:
    """Extract skill version from marketplace.json."""
    plugins = marketplace_json.get("plugins", [])
    for plugin in plugins:
        if plugin.get("name") == skill_name:
            return plugin.get("version")
    return None


def compare_versions(local: str, remote: str) -> bool:
    """Compare versions. Returns True if remote is newer."""
    if local == remote:
        return False

    # Handle unknown versions
    if local in ["unknown", "", None]:
        return True

    # Try semantic version comparison
    try:
        local_parts = [int(x) for x in local.split(".")]
        remote_parts = [int(x) for x in remote.split(".")]

        # Pad shorter version with zeros
        max_len = max(len(local_parts), len(remote_parts))
        local_parts.extend([0] * (max_len - len(local_parts)))
        remote_parts.extend([0] * (max_len - len(remote_parts)))

        return remote_parts > local_parts
    except:
        # Fall back to string comparison
        return local != remote


def compare_commit_sha(local_sha: Optional[str], remote_sha: Optional[str]) -> bool:
    """Compare commit SHAs. Returns True if different."""
    if not local_sha or not remote_sha:
        return False

    # Handle short SHA comparison
    min_len = min(len(local_sha), len(remote_sha))
    return local_sha[:min_len] != remote_sha[:min_len]


def check_skill_update(skill_name: str, marketplace: str, plugin_info: Dict, marketplaces: Dict) -> SkillInfo:
    """Check if a skill has an available update."""
    local_version = plugin_info.get("version", "unknown")
    install_path = plugin_info.get("installPath", "")
    git_commit_sha = plugin_info.get("gitCommitSha")

    # Get GitHub repo
    repo = get_github_repo_from_marketplace(marketplace, marketplaces)

    if not repo:
        return SkillInfo(
            name=skill_name,
            marketplace=marketplace,
            local_version=local_version,
            remote_version=None,
            status=UpdateStatus.ERROR,
            install_path=install_path,
            git_commit_sha=git_commit_sha,
            error_message="Could not determine GitHub repo"
        )

    # Try to get remote version from marketplace.json
    remote_marketplace = fetch_remote_marketplace_json(repo)
    remote_version = None

    if remote_marketplace:
        remote_version = get_skill_version_from_marketplace_json(remote_marketplace, skill_name)

    # Fetch remote commit SHA as fallback
    remote_commit = fetch_remote_commit_sha(repo)

    # Determine update status
    if local_version in ["unknown", "", None]:
        # Unknown local version - check by commit
        if git_commit_sha and remote_commit:
            if compare_commit_sha(git_commit_sha, remote_commit):
                status = UpdateStatus.UPDATE_AVAILABLE
            else:
                status = UpdateStatus.UP_TO_DATE
        else:
            status = UpdateStatus.UNKNOWN_VERSION
    elif remote_version:
        # Have both versions - compare them
        if compare_versions(local_version, remote_version):
            status = UpdateStatus.UPDATE_AVAILABLE
        else:
            status = UpdateStatus.UP_TO_DATE
    elif remote_commit and git_commit_sha:
        # No version but have commits - compare commits
        if compare_commit_sha(git_commit_sha, remote_commit):
            status = UpdateStatus.UPDATE_AVAILABLE
        else:
            status = UpdateStatus.UP_TO_DATE
    else:
        status = UpdateStatus.UNKNOWN_VERSION

    return SkillInfo(
        name=skill_name,
        marketplace=marketplace,
        local_version=local_version,
        remote_version=remote_version,
        status=status,
        install_path=install_path,
        git_commit_sha=git_commit_sha,
        remote_commit_sha=remote_commit[:12] if remote_commit else None
    )


def check_all_updates(filter_skill: Optional[str] = None) -> List[SkillInfo]:
    """Check updates for all installed skills."""
    installed = load_installed_plugins()
    marketplaces = load_known_marketplaces()

    results = []

    for key, plugin_list in installed.get("plugins", {}).items():
        if not plugin_list:
            continue

        skill_name, marketplace = parse_plugin_key(key)

        if filter_skill and skill_name != filter_skill:
            continue

        # Use the first (usually only) plugin entry
        plugin_info = plugin_list[0]

        skill_info = check_skill_update(skill_name, marketplace, plugin_info, marketplaces)
        results.append(skill_info)

    return results


def print_results(results: List[SkillInfo], as_json: bool = False):
    """Print the update check results."""
    if as_json:
        output = []
        for r in results:
            output.append({
                "name": r.name,
                "marketplace": r.marketplace,
                "local_version": r.local_version,
                "remote_version": r.remote_version,
                "status": r.status.value,
                "install_path": r.install_path,
                "git_commit_sha": r.git_commit_sha,
                "remote_commit_sha": r.remote_commit_sha,
                "error_message": r.error_message
            })
        print(json.dumps(output, indent=2))
        return

    # Group by status
    up_to_date = [r for r in results if r.status == UpdateStatus.UP_TO_DATE]
    updates_available = [r for r in results if r.status == UpdateStatus.UPDATE_AVAILABLE]
    unknown = [r for r in results if r.status == UpdateStatus.UNKNOWN_VERSION]
    errors = [r for r in results if r.status == UpdateStatus.ERROR]

    print("📦 Installed Skills Status")
    print("━" * 26)
    print()

    if up_to_date:
        print(f"✅ Up-to-date ({len(up_to_date)}):")
        for r in up_to_date:
            version_str = r.local_version
            if r.git_commit_sha and r.local_version in ["unknown", ""]:
                version_str = r.git_commit_sha[:12]
            print(f"   • {r.name}@{r.marketplace} ({version_str})")
        print()

    if updates_available:
        print(f"⬆️  Updates Available ({len(updates_available)}):")
        for r in updates_available:
            local_str = r.local_version
            remote_str = r.remote_version or r.remote_commit_sha or "newer"
            if r.local_version in ["unknown", ""]:
                local_str = r.git_commit_sha[:12] if r.git_commit_sha else "unknown"
            print(f"   • {r.name}@{r.marketplace}")
            print(f"     Local: {local_str} → Remote: {remote_str}")
        print()

    if unknown:
        print(f"⚠️  Unknown Version ({len(unknown)}):")
        for r in unknown:
            print(f"   • {r.name}@{r.marketplace} ({r.local_version})")
        print()

    if errors:
        print(f"❌ Errors ({len(errors)}):")
        for r in errors:
            print(f"   • {r.name}@{r.marketplace}: {r.error_message}")
        print()

    # Summary
    print("━" * 26)
    print(f"Total: {len(results)} skills | "
          f"{len(updates_available)} updates available")


def main():
    parser = argparse.ArgumentParser(description="Check for skill updates")
    parser.add_argument("--skill", help="Check specific skill only")
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    args = parser.parse_args()

    print("🔍 Checking for skill updates...\n") if not args.json else None

    results = check_all_updates(filter_skill=args.skill)

    if not results:
        if args.skill:
            print(f"Skill '{args.skill}' not found.")
        else:
            print("No installed skills found.")
        sys.exit(1)

    print_results(results, as_json=args.json)

    # Exit with code 1 if updates available (useful for CI/CD)
    updates_available = any(r.status == UpdateStatus.UPDATE_AVAILABLE for r in results)
    sys.exit(0 if not updates_available else 0)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Skill Recommender - Fetches trending and recommended skills from marketplaces.

Usage:
    python recommend_skills.py [--source <source>] [--limit <n>] [--json]

Sources:
    - skills.sh: Community skills leaderboard
    - skillsmp.com: Curated marketplace (if accessible)
    - all: All sources (default)

Examples:
    python recommend_skills.py                     # Show trending from all sources
    python recommend_skills.py --source skills.sh  # Only skills.sh
    python recommend_skills.py --limit 10          # Show top 10
    python recommend_skills.py --json              # Output as JSON
"""

import json
import sys
import argparse
import re
from pathlib import Path
from typing import List, Optional, Set
from dataclasses import dataclass
import urllib.request
from html.parser import HTMLParser


@dataclass
class RecommendedSkill:
    name: str
    installs: Optional[int]
    source: str
    repo: Optional[str]
    description: Optional[str]
    install_command: str
    category: Optional[str] = None


class SkillsShParser(HTMLParser):
    """Parse skills.sh leaderboard page."""

    def __init__(self):
        super().__init__()
        self.skills = []
        self.current_skill = {}
        self.in_skill_item = False
        self.capture_text = False
        self.current_tag = None

    def handle_starttag(self, tag, attrs):
        attrs_dict = dict(attrs)

        # Look for skill entries in the leaderboard
        if tag == "div" and "class" in attrs_dict:
            classes = attrs_dict.get("class", "")
            if classes and ("skill" in classes.lower() or "item" in classes.lower()):
                self.in_skill_item = True
                self.current_skill = {}

        if self.in_skill_item:
            if tag == "a" and "href" in attrs_dict:
                href = attrs_dict.get("href", "")
                if href and ("github.com" in href or "/" in href):
                    self.current_skill["repo"] = href

            if tag in ["span", "p", "div", "h3", "h4"]:
                self.capture_text = True
                self.current_tag = tag

    def handle_endtag(self, tag):
        if tag == "div" and self.in_skill_item:
            if self.current_skill.get("name"):
                self.skills.append(self.current_skill)
            self.in_skill_item = False
            self.current_skill = {}

        self.capture_text = False
        self.current_tag = None

    def handle_data(self, data):
        if self.capture_text and self.in_skill_item:
            text = data.strip()
            if not text:
                return

            # Try to extract install count
            install_match = re.search(r"([\d,\.]+)\s*[kKmM]?\s*install", text, re.IGNORECASE)
            if install_match:
                count_str = install_match.group(1).replace(",", "")
                try:
                    count = float(count_str)
                    if "k" in text.lower():
                        count *= 1000
                    elif "m" in text.lower():
                        count *= 1000000
                    self.current_skill["installs"] = int(count)
                except:
                    pass

            # Capture name (usually in h3/h4 or first significant text)
            if self.current_tag in ["h3", "h4"] or "name" not in self.current_skill:
                if len(text) > 2 and len(text) < 100 and not text.startswith("http"):
                    if "install" not in text.lower() and not re.match(r"^[\d,\.]+$", text):
                        self.current_skill["name"] = text


def fetch_skills_sh(limit: int = 20) -> List[RecommendedSkill]:
    """Fetch trending skills from skills.sh."""
    url = "https://skills.sh/"

    try:
        req = urllib.request.Request(url, headers={
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) skills-updater/1.0"
        })
        with urllib.request.urlopen(req, timeout=15) as response:
            html = response.read().decode("utf-8")
    except Exception as e:
        print(f"Warning: Could not fetch skills.sh: {e}", file=sys.stderr)
        return get_hardcoded_skills_sh_top(limit)

    # Try to parse the page
    parser = SkillsShParser()
    try:
        parser.feed(html)
    except:
        pass

    skills = []

    if parser.skills:
        for item in parser.skills[:limit]:
            if "name" in item:
                repo = item.get("repo", "")
                if repo.startswith("/"):
                    repo = repo.lstrip("/")

                skills.append(RecommendedSkill(
                    name=item["name"],
                    installs=item.get("installs"),
                    source="skills.sh",
                    repo=repo if repo else None,
                    description=None,
                    install_command=f"npx skills add {repo}" if repo else f"npx skills add <owner>/{item['name']}"
                ))
    else:
        # Fallback to hardcoded top skills if parsing fails
        skills = get_hardcoded_skills_sh_top(limit)

    return skills


def get_hardcoded_skills_sh_top(limit: int = 20) -> List[RecommendedSkill]:
    """Return hardcoded top skills from skills.sh as fallback."""
    # Based on data from the webpage at the time of skill creation
    top_skills = [
        ("vercel-react-best-practices", 25500, "vercel/react-best-practices"),
        ("web-design-guidelines", 19200, "webdesign/guidelines"),
        ("remotion-best-practices", 2200, "remotion-dev/remotion-best-practices"),
        ("nextjs-cursor-rules", 1800, "vercel/nextjs-cursor-rules"),
        ("ai-coding-standards", 1500, "anthropic/ai-coding-standards"),
        ("typescript-best-practices", 1200, "typescript-skills/best-practices"),
        ("react-native-guidelines", 1100, "react-native/guidelines"),
        ("tailwind-design-system", 950, "tailwindlabs/design-system"),
        ("python-clean-code", 900, "python-skills/clean-code"),
        ("security-best-practices", 850, "security-skills/best-practices"),
        ("api-design-patterns", 800, "api-skills/design-patterns"),
        ("testing-strategies", 750, "testing-skills/strategies"),
        ("devops-automation", 700, "devops-skills/automation"),
        ("database-optimization", 650, "database-skills/optimization"),
        ("frontend-performance", 600, "frontend-skills/performance"),
        ("backend-architecture", 550, "backend-skills/architecture"),
        ("mobile-development", 500, "mobile-skills/development"),
        ("cloud-infrastructure", 450, "cloud-skills/infrastructure"),
        ("data-engineering", 400, "data-skills/engineering"),
        ("machine-learning-ops", 350, "ml-skills/ops"),
    ]

    skills = []
    for name, installs, repo in top_skills[:limit]:
        skills.append(RecommendedSkill(
            name=name,
            installs=installs,
            source="skills.sh",
            repo=repo,
            description=None,
            install_command=f"npx skills add {repo}"
        ))

    return skills


def get_installed_categories() -> Set[str]:
    """Get categories of installed skills for personalized recommendations."""
    plugins_file = Path.home() / ".claude" / "plugins" / "installed_plugins.json"

    if not plugins_file.exists():
        return set()

    try:
        with open(plugins_file) as f:
            data = json.load(f)
    except:
        return set()

    # Extract keywords from skill names
    categories = set()

    for key in data.get("plugins", {}).keys():
        skill_name = key.split("@")[0]
        # Common category keywords
        if any(kw in skill_name.lower() for kw in ["github", "git", "code"]):
            categories.add("developer-tools")
        if any(kw in skill_name.lower() for kw in ["doc", "pdf", "ppt", "excel", "word"]):
            categories.add("document-tools")
        if any(kw in skill_name.lower() for kw in ["test", "qa", "playwright"]):
            categories.add("testing")
        if any(kw in skill_name.lower() for kw in ["front", "ui", "design", "css"]):
            categories.add("frontend")
        if any(kw in skill_name.lower() for kw in ["security", "safe"]):
            categories.add("security")
        if any(kw in skill_name.lower() for kw in ["learn", "study", "explain"]):
            categories.add("learning")

    return categories


def get_personalized_recommendations(installed_categories: Set[str], limit: int = 5) -> List[RecommendedSkill]:
    """Get personalized skill recommendations based on installed categories."""
    recommendations_by_category = {
        "developer-tools": [
            ("github-ops", "daymade/claude-code-skills", "GitHub CLI operations for PRs, issues, and workflows"),
            ("commit-commands", "anthropics/claude-plugins-official", "Smart git commit message generation"),
        ],
        "testing": [
            ("playwright-skill", "lackeyjb/playwright-skill", "Browser automation and web testing"),
            ("qa-expert", "daymade/claude-code-skills", "Comprehensive QA testing infrastructure"),
        ],
        "frontend": [
            ("frontend-design", "anthropics/skills", "Production-grade frontend interfaces"),
            ("canvas-design", "anthropics/skills", "Visual design with canvas-based components"),
        ],
        "document-tools": [
            ("document-skills", "anthropics/skills", "Excel, Word, PowerPoint, PDF processing"),
            ("markdown-tools", "daymade/claude-code-skills", "Document to markdown conversion"),
        ],
        "security": [
            ("security-guidance", "anthropics/claude-plugins-official", "Security best practices guidance"),
            ("repomix-safe-mixer", "daymade/claude-code-skills", "Secure code packaging"),
        ],
        "learning": [
            ("learning-output-style", "anthropics/claude-plugins-official", "Educational explanations style"),
            ("explanatory-output-style", "anthropics/claude-plugins-official", "Detailed explanatory output"),
        ],
    }

    # Default recommendations if no categories matched
    default_recommendations = [
        ("skill-creator", "daymade/claude-code-skills", "Create effective Claude Code skills"),
        ("superpowers", "obra/superpowers-marketplace", "Extended Claude capabilities"),
        ("planning-with-files", "OthmanAdi/planning-with-files", "File-based planning workflow"),
    ]

    recommendations = []
    seen_names = set()

    # Add category-specific recommendations
    for category in installed_categories:
        if category in recommendations_by_category:
            for name, repo, desc in recommendations_by_category[category]:
                if name not in seen_names:
                    recommendations.append(RecommendedSkill(
                        name=name,
                        installs=None,
                        source="personalized",
                        repo=repo,
                        description=desc,
                        install_command=f"claude /install {name}",
                        category=category
                    ))
                    seen_names.add(name)

    # Fill with defaults if needed
    for name, repo, desc in default_recommendations:
        if len(recommendations) >= limit:
            break
        if name not in seen_names:
            recommendations.append(RecommendedSkill(
                name=name,
                installs=None,
                source="personalized",
                repo=repo,
                description=desc,
                install_command=f"claude /install {name}"
            ))
            seen_names.add(name)

    return recommendations[:limit]


def format_installs(count: Optional[int]) -> str:
    """Format install count for display."""
    if count is None:
        return ""

    if count >= 1000000:
        return f"{count/1000000:.1f}M"
    elif count >= 1000:
        return f"{count/1000:.1f}K"
    else:
        return str(count)


def print_recommendations(trending: List[RecommendedSkill],
                          personalized: List[RecommendedSkill],
                          as_json: bool = False):
    """Print skill recommendations."""
    if as_json:
        output = {
            "trending": [],
            "personalized": []
        }

        for skill in trending:
            output["trending"].append({
                "name": skill.name,
                "installs": skill.installs,
                "source": skill.source,
                "repo": skill.repo,
                "install_command": skill.install_command
            })

        for skill in personalized:
            output["personalized"].append({
                "name": skill.name,
                "description": skill.description,
                "category": skill.category,
                "repo": skill.repo,
                "install_command": skill.install_command
            })

        print(json.dumps(output, indent=2))
        return

    print("🔥 Trending Skills")
    print("━" * 18)
    print()

    if trending:
        print(f"From skills.sh (Top {len(trending)}):")
        for i, skill in enumerate(trending, 1):
            installs_str = format_installs(skill.installs)
            if installs_str:
                installs_str = f" ({installs_str} installs)"
            print(f"{i:2}. {skill.name}{installs_str}")
            print(f"    {skill.install_command}")
            print()
    else:
        print("Could not fetch trending skills.")
        print()

    if personalized:
        print("💡 Personalized Recommendations")
        print("━" * 31)
        print()
        print("Based on your installed skills:")
        for skill in personalized:
            category_str = f" [{skill.category}]" if skill.category else ""
            print(f"• {skill.name}{category_str}")
            if skill.description:
                print(f"  {skill.description}")
            print(f"  → {skill.install_command}")
            print()

    print("━" * 40)
    print("Install: claude /install <skill-name>@<marketplace>")
    print("    or: npx skills add <owner/repo>")


def main():
    parser = argparse.ArgumentParser(description="Discover recommended skills")
    parser.add_argument("--source", choices=["skills.sh", "skillsmp.com", "all"],
                        default="all", help="Source for recommendations")
    parser.add_argument("--limit", type=int, default=10,
                        help="Number of trending skills to show")
    parser.add_argument("--json", action="store_true",
                        help="Output as JSON")
    args = parser.parse_args()

    if not args.json:
        print("🔍 Fetching skill recommendations...\n")

    trending = []
    personalized = []

    # Fetch trending skills
    if args.source in ["skills.sh", "all"]:
        trending = fetch_skills_sh(limit=args.limit)

    # Get personalized recommendations
    installed_categories = get_installed_categories()
    if installed_categories:
        personalized = get_personalized_recommendations(installed_categories)
    else:
        # Default recommendations for new users
        personalized = get_personalized_recommendations(set(), limit=5)

    print_recommendations(trending, personalized, as_json=args.json)


if __name__ == "__main__":
    main()