f8d479b69b
skills.sh repos that keep SKILL.md at the repo root (no skill directory, e.g. orzcls/win-disk-cleaner) are listed by `hermes skills search` but `inspect` and `install` failed with "Could not find '<identifier>' in any source": discovery only matched `<name>/SKILL.md` paths and the repo-root scan skipped non-directory entries by construction, so no identifier form could express "the skill directory IS the repo root". Install goes through the same resolver, so those hub-listed skills were uninstallable. - GitHubSource._find_repo_root_skill() returns the `owner/repo/` identifier (empty path segment = repo root) when the tree holds exactly one SKILL.md and it sits at the root, so a categorized multi-skill repo never resolves this way. - SKILL.md, support-file, bundle-name and source_url paths go through _skill_file_path(), which collapses an empty skill path to the repo root; the bundle is named after the repo (a root skill has no directory to be named after). - SkillsShSource._discover_identifier() consults the root fallback after the tree walk.
381 lines
18 KiB
Python
381 lines
18 KiB
Python
"""Skills Hub skills.sh adapter: catalog discovery via skills.sh, content via GitHub."""
|
|
|
|
import hashlib
|
|
import logging
|
|
import re
|
|
import time
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from tools.skills_hub_github import GitHubAuth, GitHubSource, _split_repo_id
|
|
from tools.skills_hub_models import (
|
|
SkillBundle, SkillMeta, SkillSource, _cache_metas, _cached_metas, _get_json, _get_text, _memo_json, hub,
|
|
)
|
|
|
|
logger = logging.getLogger("tools.skills_hub")
|
|
|
|
|
|
def _strip_html(value: str) -> str:
|
|
return re.sub(r'<[^>]+>', '', value)
|
|
|
|
|
|
class SkillsShSource(SkillSource):
|
|
"""Discover skills via skills.sh and fetch content from the underlying GitHub repo."""
|
|
|
|
BASE_URL = "https://skills.sh"
|
|
SEARCH_URL = f"{BASE_URL}/api/search"
|
|
# The sitemap is the real catalog source: the homepage only exposes a
|
|
# ~200-entry featured strip; sitemap.xml points at sitemap-skills-N.xml
|
|
# files (10k URLs each) covering the full ~20k+ catalog.
|
|
SITEMAP_INDEX_URL = "https://www.skills.sh/sitemap.xml"
|
|
# skills.sh serves per-skill sitemaps brotli-compressed and httpx's optional
|
|
# brotlicffi backend has a streaming-decode bug on them; asking for gzip
|
|
# only makes the server fall back to gzip/identity on every httpx install.
|
|
_SITEMAP_HEADERS = {"Accept-Encoding": "gzip"}
|
|
_SITEMAP_LOC_RE = re.compile(r"<loc>([^<]+)</loc>", re.IGNORECASE)
|
|
_SITEMAP_SKILL_RE = re.compile(
|
|
r"^https?://(?:www\.)?skills\.sh/(?P<owner>[^/]+)/(?P<repo>[^/]+)/(?P<skill>[^/]+)/?$", re.IGNORECASE,
|
|
)
|
|
_SKILL_LINK_RE = re.compile(r'href=["\']/(?P<id>(?!agents/|_next/|api/)[^"\'/]+/[^"\'/]+/[^"\'/]+)["\']')
|
|
_INSTALL_CMD_RE = re.compile(
|
|
r'npx\s+skills\s+add\s+(?P<repo>https?://github\.com/[^\s<]+|[^\s<]+)'
|
|
r'(?:\s+--skill\s+(?P<skill>[^\s<]+))?',
|
|
re.IGNORECASE,
|
|
)
|
|
_PAGE_H1_RE = re.compile(r'<h1[^>]*>(?P<title>.*?)</h1>', re.IGNORECASE | re.DOTALL)
|
|
_PROSE_H1_RE = re.compile(
|
|
r'<div[^>]*class=["\'][^"\']*prose[^"\']*["\'][^>]*>.*?<h1[^>]*>(?P<title>.*?)</h1>',
|
|
re.IGNORECASE | re.DOTALL,
|
|
)
|
|
_PROSE_P_RE = re.compile(
|
|
r'<div[^>]*class=["\'][^"\']*prose[^"\']*["\'][^>]*>.*?<p[^>]*>(?P<body>.*?)</p>',
|
|
re.IGNORECASE | re.DOTALL,
|
|
)
|
|
_WEEKLY_INSTALLS_RE = re.compile(r'Weekly Installs.*?children\\":\\"(?P<count>[0-9.,Kk]+)\\"', re.DOTALL)
|
|
_ID_PREFIX_ALIASES = ("skills-sh/", "skills.sh/", "skils-sh/", "skils.sh/")
|
|
# Standard skill sub-paths tried before any tree/discovery walk.
|
|
_STANDARD_BASE_PATHS = ("skills/", ".agents/skills/", ".claude/skills/")
|
|
|
|
_strip_html = staticmethod(_strip_html)
|
|
SOURCE_ID = "skills-sh"
|
|
|
|
def __init__(self, auth: GitHubAuth):
|
|
self.auth, self.github = auth, GitHubSource(auth=auth)
|
|
|
|
def trust_level_for(self, identifier: str) -> str:
|
|
return self.github.trust_level_for(self._normalize_identifier(identifier))
|
|
|
|
def _meta(self, canonical: str, *, name: str, description: str, path: str,
|
|
extra: Optional[Dict[str, Any]] = None) -> SkillMeta:
|
|
return SkillMeta(
|
|
name=name, description=description, source="skills.sh", identifier=self._wrap_identifier(canonical),
|
|
trust_level=self.github.trust_level_for(canonical), repo="/".join(canonical.split("/", 2)[:2]),
|
|
path=path, extra=extra if extra is not None else {},
|
|
)
|
|
|
|
def _urls_for(self, canonical: str, repo: str) -> Dict[str, str]:
|
|
return {"detail_url": f"{self.BASE_URL}/{canonical}", "repo_url": f"https://github.com/{repo}"}
|
|
|
|
def search(self, query: str, limit: int = 10) -> List[SkillMeta]:
|
|
if not query.strip():
|
|
# Empty query = bulk catalog dump (build_skills_index.py) — walk the sitemap.
|
|
return self._sitemap_catalog(limit)
|
|
cache_key = f"skills_sh_search_{hashlib.md5(f'{query}|{limit}'.encode()).hexdigest()}"
|
|
cached = _cached_metas(cache_key)
|
|
if cached is not None:
|
|
return cached[:limit]
|
|
data = _get_json(self.SEARCH_URL, params={"q": query, "limit": limit})
|
|
if data is None:
|
|
return []
|
|
items = data.get("skills", []) if isinstance(data, dict) else []
|
|
if not isinstance(items, list):
|
|
return []
|
|
results = [m for m in map(self._meta_from_search_item, items[:limit]) if m]
|
|
_cache_metas(cache_key, results)
|
|
return results
|
|
|
|
def fetch(self, identifier: str) -> Optional[SkillBundle]:
|
|
canonical = self._normalize_identifier(identifier)
|
|
detail = self._fetch_detail_page(canonical)
|
|
|
|
def _relabel(github_id: Optional[str]) -> Optional[SkillBundle]:
|
|
bundle = self.github.fetch(github_id) if github_id else None
|
|
if bundle:
|
|
bundle.source, bundle.identifier = "skills.sh", self._wrap_identifier(canonical)
|
|
bundle.metadata.update(self._detail_to_metadata(canonical, detail))
|
|
return bundle or None
|
|
|
|
for candidate in self._candidate_identifiers(canonical):
|
|
bundle = _relabel(candidate)
|
|
if bundle:
|
|
return bundle
|
|
return _relabel(self._discover_identifier(canonical, detail=detail))
|
|
|
|
def inspect(self, identifier: str) -> Optional[SkillMeta]:
|
|
canonical = self._normalize_identifier(identifier)
|
|
detail = self._fetch_detail_page(canonical)
|
|
meta = self._resolve_github_meta(canonical, detail=detail)
|
|
return self._finalize_inspect_meta(meta, canonical, detail) if meta else None
|
|
|
|
def _sitemap_catalog(self, limit: int) -> List[SkillMeta]:
|
|
"""Enumerate the full catalog via the sitemap (cached for the index TTL —
|
|
~2 MB of XML). Falls back to ``_featured_skills`` when unreachable/empty."""
|
|
cache_key = "skills_sh_sitemap_v1"
|
|
cached = _cached_metas(cache_key)
|
|
if cached is not None:
|
|
return cached[:limit] if limit > 0 else cached
|
|
|
|
# Every hop goes through the hub's guarded GET: the index is a root of trust
|
|
# that may redirect, and its <loc> entries are remote-party-controlled — a
|
|
# hostile index could point a sitemap at an internal address.
|
|
def _xml(url: str, timeout: int) -> Optional[str]:
|
|
resp = hub()._guarded_http_get(url, timeout=timeout, headers=self._SITEMAP_HEADERS)
|
|
return resp.text if resp is not None and resp.status_code == 200 else None
|
|
|
|
# Step 1: sitemap index -> per-skill sitemap URLs.
|
|
index_xml = _xml(self.SITEMAP_INDEX_URL, 20)
|
|
skill_sitemap_urls = [m.group(1).strip() for m in self._SITEMAP_LOC_RE.finditer(index_xml or "")
|
|
if "sitemap-skills" in m.group(1)]
|
|
if not skill_sitemap_urls:
|
|
return self._featured_skills(limit)
|
|
|
|
# Step 2: collect canonical "owner/repo/skill" IDs from each sitemap. A shard
|
|
# ``_xml`` returns None for is a hole, not an empty shard: retry it, and
|
|
# if it stays dark return the partial slice without publishing it to the cache.
|
|
seen, results, partial = set(), [], False
|
|
for sitemap_url in skill_sitemap_urls:
|
|
for attempt in range(1, self.CATALOG_PAGE_RETRIES + 1):
|
|
xml = _xml(sitemap_url, 30)
|
|
if xml is not None or attempt == self.CATALOG_PAGE_RETRIES:
|
|
break
|
|
time.sleep(min(2 ** attempt, 8))
|
|
if xml is None:
|
|
partial = True
|
|
continue
|
|
for loc_match in self._SITEMAP_LOC_RE.finditer(xml):
|
|
m = self._SITEMAP_SKILL_RE.match(loc_match.group(1).strip())
|
|
if not m:
|
|
continue
|
|
repo, skill = f"{m.group('owner')}/{m.group('repo')}", m.group("skill")
|
|
canonical = f"{repo}/{skill}"
|
|
if canonical not in seen:
|
|
seen.add(canonical)
|
|
results.append(self._meta(canonical, name=skill, description=f"Indexed by skills.sh from {repo}",
|
|
path=skill, extra=self._urls_for(canonical, repo)))
|
|
if not results:
|
|
return self._featured_skills(limit)
|
|
if not partial:
|
|
_cache_metas(cache_key, results)
|
|
return results[:limit] if limit > 0 else results
|
|
|
|
def _featured_skills(self, limit: int) -> List[SkillMeta]:
|
|
cache_key = "skills_sh_featured"
|
|
cached = _cached_metas(cache_key)
|
|
if cached is not None:
|
|
return cached[:limit]
|
|
html = _get_text(self.BASE_URL)
|
|
if html is None:
|
|
return []
|
|
seen, results = set(), []
|
|
for match in self._SKILL_LINK_RE.finditer(html):
|
|
canonical = match.group("id")
|
|
split = None if canonical in seen else _split_repo_id(canonical)
|
|
seen.add(canonical)
|
|
if split is None:
|
|
continue
|
|
repo, skill_path = split
|
|
results.append(self._meta(canonical, name=skill_path.split("/")[-1],
|
|
description=f"Featured on skills.sh from {repo}", path=skill_path))
|
|
if len(results) >= limit:
|
|
break
|
|
_cache_metas(cache_key, results)
|
|
return results
|
|
|
|
def _meta_from_search_item(self, item: dict) -> Optional[SkillMeta]:
|
|
if not isinstance(item, dict):
|
|
return None
|
|
canonical, repo, skill_path = item.get("id"), item.get("source"), item.get("skillId")
|
|
if not isinstance(canonical, str) or canonical.count("/") < 2:
|
|
if not (isinstance(repo, str) and isinstance(skill_path, str)):
|
|
return None
|
|
canonical = f"{repo}/{skill_path}"
|
|
split = _split_repo_id(canonical)
|
|
if split is None:
|
|
return None
|
|
repo, skill_path = split
|
|
installs = item.get("installs")
|
|
installs_label = f" · {int(installs):,} installs" if isinstance(installs, int) else ""
|
|
return self._meta(
|
|
canonical, name=str(item.get("name") or skill_path.split("/")[-1]),
|
|
description=f"Indexed by skills.sh from {repo}{installs_label}", path=skill_path,
|
|
extra={"installs": installs, **self._urls_for(canonical, repo)},
|
|
)
|
|
|
|
def _fetch_detail_page(self, identifier: str) -> Optional[dict]:
|
|
def compute():
|
|
html = _get_text(f"{self.BASE_URL}/{identifier}")
|
|
return None if html is None else self._parse_detail_page(identifier, html) or None
|
|
key = f"skills_sh_detail_{hashlib.md5(identifier.encode()).hexdigest()}"
|
|
return _memo_json(key, compute, valid=lambda c: isinstance(c, dict))
|
|
|
|
def _parse_detail_page(self, identifier: str, html: str) -> Optional[dict]:
|
|
split = _split_repo_id(identifier)
|
|
if split is None:
|
|
return None
|
|
repo, install_skill = split
|
|
install_command, install_match = None, self._INSTALL_CMD_RE.search(html)
|
|
if install_match:
|
|
install_command = install_match.group(0).strip()
|
|
install_skill = (install_match.group("skill") or install_skill).strip()
|
|
repo = self._extract_repo_slug((install_match.group("repo") or "").strip()) or repo
|
|
return {
|
|
"repo": repo, "install_skill": install_skill,
|
|
"page_title": self._extract_first_match(self._PAGE_H1_RE, html),
|
|
"body_title": self._extract_first_match(self._PROSE_H1_RE, html),
|
|
"body_summary": self._extract_first_match(self._PROSE_P_RE, html),
|
|
"weekly_installs": self._extract_weekly_installs(html),
|
|
"install_command": install_command,
|
|
**self._urls_for(identifier, repo),
|
|
"security_audits": self._extract_security_audits(html, identifier),
|
|
}
|
|
|
|
def _discover_identifier(self, identifier: str, detail: Optional[dict] = None) -> Optional[str]:
|
|
split = _split_repo_id(identifier)
|
|
if split is None:
|
|
return None
|
|
default_repo, skill_path = split
|
|
repo = detail.get("repo", default_repo) if isinstance(detail, dict) else default_repo
|
|
skill_token = skill_path.split("/")[-1]
|
|
tokens = [skill_token]
|
|
if isinstance(detail, dict):
|
|
tokens.extend(detail.get(k, "") for k in ("install_skill", "page_title", "body_title"))
|
|
|
|
def _match_in(base_path: str) -> Optional[str]:
|
|
try:
|
|
skills = self.github._list_skills_in_repo(repo, base_path)
|
|
except Exception:
|
|
return None
|
|
return next((m.identifier for m in skills if self._matches_skill_tokens(m, tokens)), None)
|
|
|
|
# One recursive tree lookup before brute-forcing every top-level dir
|
|
# (avoids request bursts on categorized repos like borghei/claude-skills).
|
|
found = (next((f for f in map(_match_in, self._STANDARD_BASE_PATHS) if f), None)
|
|
or self.github._find_skill_in_repo_tree(repo, skill_token)
|
|
or self.github._find_repo_root_skill(repo))
|
|
if found:
|
|
return found
|
|
|
|
# Fallback: scan repo root for directories that might contain skills.
|
|
try:
|
|
entries = _get_json(f"https://api.github.com/repos/{repo}/contents/",
|
|
headers=self.github.auth.get_headers(), timeout=15, follow_redirects=True)
|
|
for entry in entries if isinstance(entries, list) else []:
|
|
if entry.get("type") != "dir":
|
|
continue
|
|
dir_name = entry["name"]
|
|
if dir_name.startswith((".", "_")) or dir_name in {"skills", ".agents", ".claude"}:
|
|
continue
|
|
meta = self.github.inspect(f"{repo}/{dir_name}/{skill_token}")
|
|
if meta:
|
|
return meta.identifier
|
|
found = _match_in(dir_name + "/")
|
|
if found:
|
|
return found
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
def _resolve_github_meta(self, identifier: str, detail: Optional[dict] = None) -> Optional[SkillMeta]:
|
|
for candidate in self._candidate_identifiers(identifier):
|
|
meta = self.github.inspect(candidate)
|
|
if meta:
|
|
return meta
|
|
resolved = self._discover_identifier(identifier, detail=detail)
|
|
return self.github.inspect(resolved) if resolved else None
|
|
|
|
def _finalize_inspect_meta(self, meta: SkillMeta, canonical: str, detail: Optional[dict]) -> SkillMeta:
|
|
meta.source, meta.identifier = "skills.sh", self._wrap_identifier(canonical)
|
|
meta.trust_level = self.trust_level_for(canonical)
|
|
meta.extra = {**meta.extra, **self._detail_to_metadata(canonical, detail)}
|
|
if isinstance(detail, dict):
|
|
body_summary, weekly_installs = detail.get("body_summary"), detail.get("weekly_installs")
|
|
if body_summary:
|
|
meta.description = body_summary
|
|
elif meta.description and weekly_installs:
|
|
meta.description = f"{meta.description} · {weekly_installs} weekly installs on skills.sh"
|
|
return meta
|
|
|
|
@classmethod
|
|
def _matches_skill_tokens(cls, meta: SkillMeta, skill_tokens: List[str]) -> bool:
|
|
candidates = (cls._token_variants(meta.name) | cls._token_variants(meta.path)
|
|
| cls._token_variants(meta.identifier.split("/", 2)[-1] if meta.identifier else None))
|
|
return any(cls._token_variants(token) & candidates for token in skill_tokens)
|
|
|
|
@staticmethod
|
|
def _token_variants(value: Optional[str]) -> set[str]:
|
|
if not value:
|
|
return set()
|
|
plain = _strip_html(str(value)).strip().strip("/").lower()
|
|
if not plain:
|
|
return set()
|
|
base, sanitized = plain.split("/")[-1], re.sub(r'[^a-z0-9/_-]+', '-', plain).strip('-')
|
|
tail = base.lstrip('@')
|
|
variants = {
|
|
plain, plain.replace("_", "-"), plain.replace("/", "-"), base, base.replace("_", "-"),
|
|
sanitized, sanitized.replace("/", "-"), sanitized.split("/")[-1], tail, tail.replace("_", "-"),
|
|
}
|
|
return {v for v in variants if v}
|
|
|
|
@staticmethod
|
|
def _extract_repo_slug(repo_value: str) -> Optional[str]:
|
|
parts = repo_value.strip().removeprefix("https://github.com/").strip("/").split("/")
|
|
return f"{parts[0]}/{parts[1]}" if len(parts) >= 2 else None
|
|
|
|
@staticmethod
|
|
def _extract_first_match(pattern: re.Pattern, text: str) -> Optional[str]:
|
|
match = pattern.search(text)
|
|
value = next((group for group in match.groups() if group), None) if match else None
|
|
return (_strip_html(value).strip() or None) if value is not None else None
|
|
|
|
def _detail_to_metadata(self, canonical: str, detail: Optional[dict]) -> Dict[str, Any]:
|
|
parts = canonical.split("/", 2)
|
|
metadata = {"detail_url": f"{self.BASE_URL}/{canonical}"}
|
|
if len(parts) >= 2:
|
|
metadata["repo_url"] = f"https://github.com/{parts[0]}/{parts[1]}"
|
|
if isinstance(detail, dict):
|
|
for key in ("weekly_installs", "install_command", "repo_url", "detail_url", "security_audits"):
|
|
if detail.get(key):
|
|
metadata[key] = detail[key]
|
|
return metadata
|
|
|
|
@classmethod
|
|
def _extract_weekly_installs(cls, html: str) -> Optional[str]:
|
|
match = cls._WEEKLY_INSTALLS_RE.search(html)
|
|
return match.group("count") if match else None
|
|
|
|
@staticmethod
|
|
def _extract_security_audits(html: str, identifier: str) -> Dict[str, str]:
|
|
audits: Dict[str, str] = {}
|
|
for audit in ("agent-trust-hub", "socket", "snyk"):
|
|
idx = html.find(f"/security/{audit}")
|
|
match = re.search(r'(Pass|Warn|Fail)', html[idx:idx + 500], re.IGNORECASE) if idx != -1 else None
|
|
if match:
|
|
audits[audit] = match.group(1).title()
|
|
return audits
|
|
|
|
@classmethod
|
|
def _normalize_identifier(cls, identifier: str) -> str:
|
|
prefix = next((p for p in cls._ID_PREFIX_ALIASES if identifier.startswith(p)), "")
|
|
return identifier[len(prefix):]
|
|
|
|
@classmethod
|
|
def _candidate_identifiers(cls, identifier: str) -> List[str]:
|
|
split = _split_repo_id(identifier)
|
|
if split is None:
|
|
return [identifier]
|
|
repo, path = split[0], split[1].lstrip("/")
|
|
return list(dict.fromkeys([f"{repo}/{path}"] + [f"{repo}/{b}{path}" for b in cls._STANDARD_BASE_PATHS]))
|
|
|
|
@staticmethod
|
|
def _wrap_identifier(identifier: str) -> str:
|
|
return f"skills-sh/{identifier}"
|