#!/usr/bin/env python3 """Find, vet and rank agent skills for one task. Part of the "Znajdź skille" skill by Dawid Gac, adapted from vercel-labs/skills find-skills (MIT, see LICENSE.md). Standard library only, Python 3.9+. What it does that a plain install-count search does not: 1. searches two pools: the curated list next to this file (curated.json) and the skills.sh registry, with several queries at once; 2. treats skills with the same name under different owners as one, keeps the copy with the best source (official publisher, then repo stars), and says so; 3. opens each candidate's GitHub repo and skill folder: licence, stars, last update, archived or deprecated, every file in the folder; 4. reads every script in the folder for risky patterns (pipe-to-shell installs, credential reads, hidden instructions, destructive commands); a script or binary it could not read makes the skill "not checkable", never "clean"; 5. scores fit, adoption, source trust, upkeep and licence out of 100. Usage: python3 find_skills.py --category video --query "video editing" --query captions python3 find_skills.py --query "opisy produktów shopify" --lang pl --top 5 --json python3 find_skills.py --scan ~/Documents --scan-only # read-only: names only, shows the plan It never installs anything. Network: skills.sh and GitHub only. GitHub allows 60 anonymous requests an hour; one search needs about 15-30, a scan more. Set GITHUB_TOKEN (a read-only token) to lift the limit. """ import argparse import itertools import json import math import os import re import sys import urllib.error import urllib.parse import urllib.request from datetime import datetime, timezone from pathlib import Path, PurePosixPath REGISTRY = "https://skills.sh/api/search" GITHUB = "https://api.github.com" RAW = "https://raw.githubusercontent.com" HERE = Path(__file__).resolve().parent UA = "znajdz-skille/1.1 (+https://dawidgac.com/pl/skills)" STATE = {"lang": "pl", "rate_limited": False} SAFE_SOURCE = re.compile(r"^[A-Za-z0-9._-]+(/[A-Za-z0-9._-]+)?$") SAFE_SKILL = re.compile(r"^[A-Za-z0-9._-]+$") # Publishers whose repos are run by the company behind the product. OFFICIAL = { "anthropics", "vercel-labs", "vercel", "openai", "microsoft", "google", "google-gemini", "shopify", "heygen-com", "remotion-dev", "github", "cloudflare", "stripe", "supabase", "firecrawl", "figma", "notion", "huggingface", } OSI = {"MIT", "Apache-2.0", "BSD-2-Clause", "BSD-3-Clause", "ISC", "MPL-2.0", "0BSD", "Unlicense", "CC0-1.0", "GPL-3.0", "LGPL-3.0", "AGPL-3.0", "GPL-2.0"} NET = r"(curl|wget|iwr|Invoke-WebRequest|Invoke-RestMethod|irm)\b" HIGH = [ (NET + r"[^\n|]{0,200}\|\s*(sudo\s+)?(ba|z|k)?sh\b|" + NET + r"[^\n|]{0,200}\|\s*(python3?|node|perl|ruby|iex)\b", "pobiera i od razu uruchamia kod z sieci (curl | sh)", "pipes a download straight into an interpreter (curl | sh)"), (r"(ba|z)?sh\s+<\(\s*" + NET + r"|(ba|z)?sh\s+-c\s+[\"']?\$\(\s*" + NET + r"|\b(python3?|node|perl|ruby)\s+-[ce]\s+[\"']?\$\(\s*" + NET + r"|\biex\b[^\n]{0,40}" + NET + "|" + NET + r"[^\n]{0,120}\|\s*iex\b", "uruchamia kod pobrany z sieci", "runs code downloaded from the web"), (NET + r"[^\n]{0,160}(-o|--output|-O)\s*\S+[^\n]{0,40}(&&|;)\s*(sudo\s+)?(ba|z)?sh\s", "pobiera skrypt i od razu go uruchamia", "downloads a script and runs it at once"), (r"base64\s+(-d|--decode|-D)[^\n]{0,80}\|\s*(ba|z)?sh|frombase64string[^\n]{0,80}(iex|invoke-expression)", "dekoduje ukryty kod i go uruchamia", "decodes hidden code and runs it"), (r"\brm\s+-(rf|fr|r\s+-f|f\s+-r)\s+(--no-preserve-root\s+)?(/|~/?|\$HOME/?|\$\{HOME\}/?)(\*|\s|$|[\"'])", "kasuje katalog domowy albo cały dysk", "deletes the home directory or root"), (r"(~|\$HOME|\$\{HOME\}|home\(\)|homedir\(\))\s*[/+,]?\s*[\"']?/?\.(ssh|aws|gnupg|kube|docker/config)\b|\bid_(rsa|ed25519|ecdsa)\b|\.netrc\b|\.config/gh/hosts\.yml|Login Data|Keychains?/", "czyta klucze SSH, dane logowania do chmury albo hasła z przeglądarki", "reads SSH keys, cloud credentials or browser passwords"), (r"(printenv|\benv\b|\bset\b)\s*\|\s*(curl|wget|nc)\b|(curl|wget)\b[^\n]{0,80}\$\((printenv|env)\)|(os\.environ(?![.\[(])|process\.env(?![.\[])|dict\(os\.environ\))[^\n]{0,120}(curl|wget|fetch\(|requests\.(post|put)|urlopen|http)|(requests\.(post|put)|urlopen|fetch\()[^\n]{0,120}(dict\(os\.environ\)|os\.environ\b(?![.\[(])|process\.env\b(?![.\[]))", "wysyła wszystkie zmienne środowiskowe (klucze) do sieci", "sends all environment variables (keys) over the network"), (r"do not (tell|inform) the user|hide (this|it) from the user|without the user (knowing|noticing)", "ukryta instrukcja dla agenta (omija Ciebie)", "hidden instruction to the agent that bypasses you"), (r"\b((ba|z|k)?sh|python3?|node|perl|ruby|source)\s+[\"']?[\w./-]+\.(png|jpe?g|gif|webp|ico|woff2?|ttf|otf|mp3|wav|mp4|mov|pdf)\b", "uruchamia plik udający obraz albo dokument", "runs a file disguised as an image or document"), ] INJECT = re.compile(r"(ignore|disregard) (all |any )?(previous|prior|above|earlier) (instructions|rules)", re.I) MEDIUM = [ (r"without (telling|asking) the user", "robi coś bez pytania Cię o zgodę", "acts without asking you"), (r"--dangerously-skip-permissions|bypassPermissions", "wyłącza pytania o zgodę", "turns off permission prompts"), (r"(^|[;&|`(]\s*|\$\(\s*)sudo\s+[a-z-]", "używa sudo", "uses sudo"), (r"crontab\s+-|>>?\s*~?/?\.(bashrc|zshrc|profile)\b|launchctl\s+load|systemctl\s+enable", "zmienia autostart albo profil powłoki", "changes autostart or the shell profile"), (r"\beval\s*\(|\bexec\s*\(", "uruchamia kod z tekstu (eval/exec)", "runs code from a string (eval/exec)"), (r"npx\s+(-y|--yes)\s|pip3? install\s+(?!-r\b)(?!-e\b)[A-Za-z][\w.-]*(\s|$)", "instaluje paczki bez przypiętej wersji", "installs unpinned packages"), ] TEXT_EXT = {".md", ".txt", ".json", ".yaml", ".yml", ".toml", ".cfg", ".ini", ".html", ".xml", ".csv", ".css", ".svg"} EXEC_EXT = {".sh", ".bash", ".zsh", ".fish", ".py", ".js", ".mjs", ".cjs", ".ts", ".tsx", ".jsx", ".ps1", ".psm1", ".bat", ".cmd", ".rb", ".pl", ".php", ".lua", ".go", ".rs", ".java", ".kt", ".swift", ".applescript"} INERT_EXT = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".ico", ".woff", ".woff2", ".ttf", ".otf", ".mp3", ".wav", ".mp4", ".mov", ".pdf"} # media cannot run BINARY_EXT = {".exe", ".bin", ".so", ".dylib", ".dll", ".jar", ".wasm", ".pyc", ".app", ".msi", ".pkg", ".deb", ".dmg", ".node"} SCAN_CAP, RUN_CAP, SIZE_CAP = 150, 300, 2_000_000 # raw.githubusercontent reads do not count against the API limit T = { "pl": { "title": "Wyniki dla", "top": "Najlepsze dopasowania", "rejected": "Odrzucone albo niesprawdzalne", "none": "Nie znalazłem skilla, który przeszedł sprawdzenie.", "cols": "| # | Skill | Wynik | Dlaczego | Licencja | Ostrzeżenia | Instalacja |", "curated": "używamy go sami", "copies": "ta sama nazwa u innych autorów", "official": "oficjalny wydawca", "stale": "brak aktualizacji od ponad roku", "nolic": "brak licencji: możesz używać, nie możesz rozpowszechniać", "folderlic": "licencja tylko w folderze skilla: przeczytaj warunki, może zabraniać rozpowszechniania", "unreadable": "nie da się sprawdzić: źródło poza GitHubem albo SKILL.md nie ma już w repo (wpis w katalogu nieaktualny)", "unsafe_id": "nie da się sprawdzić: nazwa albo źródło z niedozwolonymi znakami", "unscanned": "nie da się sprawdzić: nie przeczytałem plików, które mogą się uruchomić", "unread_docs": "nie przeczytałem {n} plików (dokumentacja albo multimedia ponad limit)", "ratelimited": "nie sprawdzono: wyczerpany limit GitHuba", "rate_note": "Uwaga: GitHub odciął zapytania (60 na godzinę bez tokena). Ustaw GITHUB_TOKEN z tokenem tylko do odczytu i uruchom jeszcze raz.", "deprecated": "wydawca oznaczył go jako przestarzały (folder deprecated)", "archived": "repo zarchiwizowane", "inflated": "liczba instalacji niewiarygodna: dużo instalacji, prawie zero gwiazdek repo", "quoted_inject": "cytuje polecenie w stylu „ignore previous instructions”: sprawdź, czy to przykład, czy rozkaz", "high": "WYSOKIE RYZYKO", "own": "skill autora, strona z plikami", "skipped": "Pomijam skille, które już masz: {}", "read": "Zanim zainstalujesz: otwórz SKILL.md i pliki skryptów i przeczytaj, co robią.", }, "en": { "title": "Results for", "top": "Best matches", "rejected": "Rejected or not checkable", "none": "No skill passed the checks.", "cols": "| # | Skill | Score | Why | Licence | Warnings | Install |", "curated": "we use it ourselves", "copies": "same name under other owners", "official": "official publisher", "stale": "no update for over a year", "nolic": "no licence: you may use it, not redistribute it", "folderlic": "licence only inside the skill folder: read its terms, it may forbid redistribution", "unreadable": "not checkable: source outside GitHub, or SKILL.md is no longer in the repo (stale registry entry)", "unsafe_id": "not checkable: name or source contains characters that are not allowed", "unscanned": "not checkable: files that can run were not read", "unread_docs": "{n} files not read (documentation or media over the limit)", "ratelimited": "not checked: GitHub rate limit reached", "rate_note": "Note: GitHub stopped answering (60 requests an hour without a token). Set GITHUB_TOKEN to a read-only token and run again.", "deprecated": "the publisher marked it deprecated (deprecated folder)", "archived": "repo archived", "inflated": "install count not credible: many installs, almost no repo stars", "quoted_inject": "quotes an \"ignore previous instructions\" line: check whether it is an example or an order", "high": "HIGH RISK", "own": "the author's own skill, page with files", "skipped": "Skipping skills you already have: {}", "read": "Before you install: open SKILL.md and any scripts and read what they do.", }, } def get(url, raw=False): """GET with error handling: returns (body, error). A GitHub 403/429 flips STATE['rate_limited'].""" headers = {"User-Agent": UA, "Accept": "*/*" if raw else "application/json"} tok = os.environ.get("GITHUB_TOKEN") if tok and url.startswith(GITHUB): headers["Authorization"] = f"Bearer {tok}" try: with urllib.request.urlopen(urllib.request.Request(url, headers=headers), timeout=20) as r: body = r.read() return (body if raw else json.loads(body)), None except urllib.error.HTTPError as e: if url.startswith(GITHUB) and e.code in (403, 429): STATE["rate_limited"] = True return None, "rate limit" return None, f"HTTP {e.code}" except Exception as e: # network error or a non-JSON page: "not checkable", never a crash return None, str(e) _REPOS, _RAW, LINKS = {}, {}, {} # LINKS: repo -> symlink paths (the installer copies what they point to) def repo(owner, name): """(meta, paths, branch) for one repo, fetched once per run (2 GitHub calls).""" key = f"{owner}/{name}".lower() if key not in _REPOS: meta, err = get(f"{GITHUB}/repos/{owner}/{name}") if err or not isinstance(meta, dict): return None, err or "no repo" # not cached: a rate-limited miss may be retried later branch = meta.get("default_branch") or "main" tree, err = get(f"{GITHUB}/repos/{owner}/{name}/git/trees/{urllib.parse.quote(branch)}?recursive=1") if err or not isinstance(tree, dict): return None, err or "no tree" paths = [t["path"] for t in tree.get("tree", []) if t.get("type") == "blob" and isinstance(t.get("path"), str)] LINKS[key] = {t["path"] for t in tree.get("tree", []) if t.get("mode") == "120000"} _REPOS[key] = (meta, paths, branch) return _REPOS[key], None def raw_file(owner, name, branch, path): key = (owner, name, branch, path) if key not in _RAW: body, err = get(f"{RAW}/{owner}/{name}/{urllib.parse.quote(branch)}/{urllib.parse.quote(path)}", raw=True) _RAW[key] = None if err or body is None or len(body) > SIZE_CAP else body.decode("utf-8", "replace") return _RAW[key] def words(text): return {w for w in re.findall(r"[a-ząćęłńóśźż0-9]{3,}", text.lower())} def search_registry(queries, limit): found = {} for q in queries: data, err = get(f"{REGISTRY}?{urllib.parse.urlencode({'q': q, 'limit': limit})}") if err or not isinstance(data, dict): print(f"registry search failed for {q!r}: {err}", file=sys.stderr) continue for s in data.get("skills") or []: if not isinstance(s, dict) or not s.get("id"): continue skill = s.get("skillId") or s.get("name") or "" found.setdefault(s["id"], {"id": s["id"], "source": s.get("source") or "", "skill": skill, "installs": s.get("installs") or 0, "pool": "registry", "hits": 0}) found[s["id"]]["hits"] += 1 return list(found.values()) def load_curated(category): path = HERE / "curated.json" if not path.exists(): return [] data = json.loads(path.read_text(encoding="utf-8")) return [{**it, "pool": "curated", "category": cat, "hits": 0, "installs": it.get("installs", 0)} for cat, items in data.get("categories", {}).items() if not category or cat == category for it in items] def group_by_name(cands): """Skills with the same name under several owners: one group each. The copy that represents the group is chosen after vetting (official publisher, then stars), never by install count alone.""" groups = {} for c in cands: groups.setdefault(c["skill"].lower(), []).append(c) for items in groups.values(): items.sort(key=lambda c: (c["pool"] == "curated", (c["source"].split("/")[0]).lower() in OFFICIAL, c["installs"]), reverse=True) for i in items: if i["pool"] == "curated" and not i["installs"]: # a curated entry keeps its registry twin's count i["installs"] = max([j["installs"] for j in items if j["id"].lower() == i["id"].lower()] + [0]) return list(groups.values()) def frontmatter_desc(text): m = re.search(r"^description:[ \t]*(.*)$", text, re.M) if not m: return text[:200] first = m.group(1).strip() if first in (">", "|", ">-", "|-", ">+", "|+"): # YAML block scalar: the indented lines below rest = text[m.end():].split("\n")[1:] first = " ".join(l.strip() for l in itertools.takewhile(lambda l: l.startswith((" ", "\t")), rest)) return first.strip().strip("\"'") def vet(c): """Read the repo and the skill folder. Sets c['state']: ok | unreadable | unsafe_id | unscanned | ratelimited.""" own_licence = c.get("licence") c.update(stars=0, licence=None, pushed=None, archived=False, files=[], desc="", high=[], medium=[], unscanned=[], deprecated=False, folder_licence=False, state="unreadable") if c.get("page"): # the author's own skill: a dawidgac.com page, files reviewed at publication c.update(state="ok", licence=own_licence or "free", desc=c.get("why_en", "")) return c if not SAFE_SOURCE.match(c["source"]) or not SAFE_SKILL.match(c["skill"]): c["state"] = "unsafe_id" return c parts = c["source"].split("/") if len(parts) != 2 or "." in parts[0]: return c # not a GitHub owner/repo (a mirror or another host) owner, name = parts got, err = repo(owner, name) if not got: c["state"] = "ratelimited" if err == "rate limit" else "unreadable" return c meta, paths, branch = got lic = (meta.get("license") or {}).get("spdx_id") c.update(stars=meta.get("stargazers_count") or 0, archived=bool(meta.get("archived")), pushed=meta.get("pushed_at"), licence=None if lic in (None, "NOASSERTION", "NONE") else lic, owner=owner) plain_first = lambda p: (p.startswith(".") or "/." in p, len(p)) # prefer skills/x over mirrors like .openclaw/skills/x all_md = [p for p in paths if PurePosixPath(p).name == "SKILL.md"] skill_md = sorted((p for p in all_md if PurePosixPath(p).parent.name.lower() == c["skill"].lower()), key=plain_first) if not skill_md and len(all_md) == 1: skill_md = all_md if not skill_md: # the registry uses the frontmatter name, which can differ from the folder name for p in sorted(all_md, key=plain_first)[:25]: body = raw_file(owner, name, branch, p) if body and re.search(rf"^name:\s*['\"]?{re.escape(c['skill'])}['\"]?\s*$", body, re.M | re.I): skill_md = [p] break if not skill_md: return c folder = str(PurePosixPath(skill_md[0]).parent) prefix = "" if folder == "." else folder + "/" # A skill at the repo root owns the whole repo: everything in it is read like a skill folder. in_skill = [p for p in paths if p.startswith(prefix) and p != skill_md[0]] c["files"] = [skill_md[0]] + in_skill c["url"] = f"https://github.com/{owner}/{name}/tree/{branch}/{prefix}" c["deprecated"] = "deprecated" in PurePosixPath(prefix).parts or "deprecated/" in prefix.lower() if not c["licence"] and any(PurePosixPath(p).name.upper().startswith(("LICENSE", "LICENCE")) for p in c["files"]): c["licence"], c["folder_licence"] = ("LICENSE w folderze skilla" if STATE["lang"] == "pl" else "LICENSE in the skill folder"), True ext = lambda p: PurePosixPath(p).suffix.lower() # Unknown or missing extensions (helper.command, a bare "install") count as runnable: read, or not checkable. texts = [p for p in in_skill if ext(p) in TEXT_EXT] runnable = [p for p in in_skill if p not in texts and ext(p) not in INERT_EXT] media = [p for p in in_skill if ext(p) in INERT_EXT] # read too: a "logo.png" can hold a script skill_body = (raw_file(owner, name, branch, skill_md[0]) or "").lower() named = lambda p: PurePosixPath(p).name.lower() in skill_body # docs SKILL.md sends the agent to texts.sort(key=lambda p: not named(p)) order = [skill_md[0]] + runnable + media + texts # SKILL.md, what can run, media, the docs it names, other docs links = LINKS.get(f"{owner}/{name}".lower(), set()) c["unscanned"] += [p for p in c["files"] if p in links] unread_docs = 0 for i, p in enumerate(order): if p in links: continue must = p in runnable or named(p) # always read (up to RUN_CAP); media too, other docs only up to SCAN_CAP if ext(p) in BINARY_EXT or i >= (RUN_CAP if must or p in media else SCAN_CAP): if must: c["unscanned"].append(p) else: unread_docs += 1 continue body = raw_file(owner, name, branch, p) if body is None: if must or p == skill_md[0]: c["unscanned"].append(p) else: unread_docs += 1 # a media file over SIZE_CAP or a doc that failed to download continue if p == skill_md[0]: c["desc"] = re.sub(r"[|`\r\n<>]+", " ", frontmatter_desc(body))[:200] for pat, pl, en in HIGH: if re.search(pat, body, re.I | re.M): c["high"].append((p, pl, en)) for line in body.splitlines(): m = INJECT.search(line) if m: quoted = re.search(r"[\"'`“„«][^\"'`”»]{0,40}" + re.escape(m.group(0)), line, re.I) if quoted: c["medium"].append((p, T["pl"]["quoted_inject"], T["en"]["quoted_inject"])) else: c["high"].append((p, "ukryta instrukcja dla agenta (omija Ciebie)", "hidden instruction to the agent that bypasses you")) break for pat, pl, en in MEDIUM: if re.search(pat, body, re.I | re.M): c["medium"].append((p, pl, en)) if unread_docs: c["medium"].append((f"{unread_docs} files", T["pl"]["unread_docs"].format(n=unread_docs), T["en"]["unread_docs"].format(n=unread_docs))) if skill_md[0] in c["unscanned"]: c["state"] = "unreadable" elif c["unscanned"]: c["state"] = "unscanned" else: c["state"] = "ok" return c def score(c, qwords): hay = words(c["skill"].replace("-", " ") + " " + c.get("desc", "") + " " + c.get("why_pl", "") + " " + c.get("why_en", "")) fit = min(35, 35 * len(qwords & hay) / max(1, min(len(qwords), 4))) adoption = min(20, 5 * math.log10(c["installs"] + 1)) # Installs can be farmed; stars are harder to fake. A huge count on a near-empty repo earns nothing. c["inflated"] = not c.get("page") and c["installs"] >= 10_000 and c["stars"] < 50 if c["inflated"]: adoption = 0 owner = (c.get("owner") or c["source"].split("/")[0]).lower() trust = 15 if owner in OFFICIAL or c.get("page") else min(12, 3 * math.log10(c["stars"] + 1)) upkeep = 10 if c.get("page") else 0 if c.get("pushed"): try: age = (datetime.now(timezone.utc) - datetime.fromisoformat(c["pushed"].replace("Z", "+00:00"))).days upkeep = 10 if age <= 90 else 5 if age <= 365 else 0 except ValueError: pass if c["archived"]: upkeep = -10 lic = 10 if c.get("page") or c["licence"] in OSI else 5 if c["licence"] else 0 curated = 10 if c["pool"] == "curated" else 0 risk = -10 * min(2, len({m[1] for m in c["medium"]})) c["score"] = round(fit + adoption + trust + upkeep + lic + curated + risk) c["parts"] = dict(fit=round(fit), adoption=round(adoption), trust=round(trust), upkeep=upkeep, licence=lic, curated=curated, risk=risk) return c SKIP_DIRS = {"node_modules", ".git", ".venv", "venv", "__pycache__", ".cache", ".next", "dist", "build", "Library", ".Trash", "target", ".npm", ".cargo", ".rustup", "site-packages"} SECRET = re.compile(r"^\.env|\.pem$|\.key$|^id_|secret|credential|token|\.p12$|\.keychain", re.I) VIDEO_EXT = {".mp4", ".mov", ".mkv", ".prproj", ".drp", ".fcpxml", ".aep", ".braw", ".r3d"} DOC_EXT = {".docx", ".xlsx", ".pptx", ".pdf", ".csv"} DESIGN_EXT = {".fig", ".psd", ".sketch", ".ai", ".svg"} def installed_skills(): """Names of the folders in the usual skill directories (names only).""" names = set() for base in (Path.home() / ".claude/skills", Path.cwd() / ".claude/skills", Path.home() / ".agents/skills"): if base.is_dir(): names |= {p.name.lower() for p in base.iterdir() if p.is_dir() and not p.name.startswith((".", "_"))} return names def scan(root, max_files=30000): """Read-only look at what the user works on. Reads file and folder NAMES, plus the dependency list of package.json; never opens anything else, never follows a symlink into a file, and skips anything named like a key or password file.""" root = Path(root).expanduser().resolve() sig = dict(projects=0, no_claude_md=0, node=0, frontend=0, python=0, shopify=0, video=0, docs=0, design=0, markdown=0, fitness=0) seen = 0 for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS and not (d.startswith(".") and d != ".claude")] if len(Path(dirpath).relative_to(root).parts) > 4: dirnames[:] = [] names = {f for f in filenames if not SECRET.search(f)} if {"package.json", "pyproject.toml", "requirements.txt", "go.mod", "Cargo.toml"} & names or ".git" in os.listdir(dirpath): sig["projects"] += 1 if not {"CLAUDE.md", "AGENTS.md"} & names: sig["no_claude_md"] += 1 pkg_path = Path(dirpath) / "package.json" if "package.json" in names and not pkg_path.is_symlink(): sig["node"] += 1 try: pkg = json.loads(pkg_path.read_text(encoding="utf-8")) deps = {**(pkg.get("dependencies") or {}), **(pkg.get("devDependencies") or {})} except Exception: deps = {} sig["frontend"] += any(d in deps for d in ("react", "next", "vue", "svelte", "astro", "tailwindcss")) sig["shopify"] += any("shopify" in d for d in deps) sig["python"] += bool({"pyproject.toml", "requirements.txt"} & names) sig["shopify"] += "theme.liquid" in names or "settings_schema.json" in names for f in names: ext = Path(f).suffix.lower() sig["video"] += ext in VIDEO_EXT sig["docs"] += ext in DOC_EXT sig["design"] += ext in DESIGN_EXT sig["markdown"] += ext == ".md" sig["fitness"] += bool(re.search(r"trening|workout|fitness|dieta|diet|makro|macro|silownia|gym", f, re.I)) seen += len(filenames) if seen > max_files: break # minimal: name-only cap; a huge folder gets a partial picture, flagged as truncated sig["truncated"] = seen > max_files return sig def plan_from_scan(sig, lang): """Signals -> (category, queries, reason) the search runs on. Only these terms leave the machine.""" pl = lang == "pl" plan = [] if sig["projects"] >= 5: plan.append(("coding", ["code minimalism", "code review"], f"masz {sig['projects']} projektów z kodem: mniej kodu i recenzja przed scaleniem to mniej długu" if pl else f"you have {sig['projects']} code projects: less code and review before merge mean less debt")) if sig["no_claude_md"] >= 3: plan.append(("coding", ["claude md project instructions"], f"{sig['no_claude_md']} projektów bez CLAUDE.md ani AGENTS.md: agent zaczyna każdy od zera" if pl else f"{sig['no_claude_md']} projects have no CLAUDE.md or AGENTS.md: the agent starts each from zero")) if sig["frontend"]: plan.append(("design", ["frontend design"], f"{sig['frontend']} projektów frontendowych (React/Next/Vue...)" if pl else f"{sig['frontend']} frontend projects (React/Next/Vue...)")) if sig["shopify"]: plan.append(("ecommerce", ["shopify"], "pliki albo zależności Shopify w projektach" if pl else "Shopify files or dependencies in your projects")) if sig["video"] >= 5: plan.append(("video", ["video editing", "captions"], f"{sig['video']} plików wideo albo projektów montażu" if pl else f"{sig['video']} video files or edit projects")) if sig["docs"] >= 10: plan.append(("documents", ["excel spreadsheet", "pdf"], f"{sig['docs']} dokumentów i arkuszy" if pl else f"{sig['docs']} documents and spreadsheets")) if sig["design"] >= 5: plan.append(("design", ["design system"], f"{sig['design']} plików graficznych" if pl else f"{sig['design']} design files")) if sig["markdown"] >= 50 and sig["projects"] < 3: plan.append(("marketing", ["writing"], f"{sig['markdown']} notatek i tekstów .md" if pl else f"{sig['markdown']} .md notes and texts")) if sig["fitness"] >= 3: plan.append(("fitness", ["fitness workout plan"], f"{sig['fitness']} plików o treningu albo diecie" if pl else f"{sig['fitness']} training or diet files")) return plan def install_line(c): return c["page"] if c.get("page") else f"npx skills add {c['source']} --skill {c['skill']}" def passes(c): return c["state"] == "ok" and not c["high"] and not c["deprecated"] def render(cands, queries, lang, top, skipped=()): """Markdown for the user. Returns (text, names shown as picks).""" t = T[lang] ok = sorted((c for c in cands if passes(c)), key=lambda c: c["score"], reverse=True)[:top] bad = [c for c in cands if not passes(c)] out = [f"## {t['title']}: {', '.join(queries) or '-'}", ""] if STATE["rate_limited"]: out += [t["rate_note"], ""] if skipped: out += [t["skipped"].format(", ".join(sorted(skipped))), ""] out += [f"### {t['top']}", ""] if not ok: out.append(t["none"]) else: out += [t["cols"], "|---|---|---|---|---|---|---|"] for i, c in enumerate(ok, 1): why = [t["curated"]] if c["pool"] == "curated" else [] if (c.get("owner") or "").lower() in OFFICIAL: why.append(t["official"]) if c["installs"]: why.append(f"{c['installs']:,} inst.".replace(",", " ")) if c["stars"]: why.append(f"{c['stars']:,}★".replace(",", " ")) if c.get("page"): why.append(t["own"]) warn = [m[1 if lang == "pl" else 2] for m in c["medium"]] if c.get("inflated"): warn.append(t["inflated"]) if c.get("copies"): warn.append(f"{t['copies']}: {c['copies']}") if c["archived"]: warn.append(t["archived"]) elif c["parts"]["upkeep"] == 0 and not c.get("page"): warn.append(t["stale"]) if not c["licence"]: warn.append(t["nolic"]) elif c.get("folder_licence"): warn.append(t["folderlic"]) name = f"[{c['skill']}]({c.get('url') or c.get('page') or 'https://skills.sh/' + c['id']})" out.append(f"| {i} | {name} | {c['score']} | {'; '.join(why)} | {c['licence'] or '-'} | {'; '.join(dict.fromkeys(warn)) or '-'} | `{install_line(c)}` |") out += ["", t["read"]] if bad: out += ["", f"### {t['rejected']}", ""] for c in bad[:10]: why = [f"{t['high']}: {m[1 if lang == 'pl' else 2]} ({m[0]})" for m in c["high"][:3]] if c["state"] == "unscanned": why.append(f"{t['unscanned']} ({', '.join(c['unscanned'][:3])})") elif c["state"] != "ok": why.append(t[c["state"]]) if c["deprecated"]: why.append(t["deprecated"]) ident = f'{c["source"]}/{c["skill"]}' if SAFE_SOURCE.match(c["source"]) and SAFE_SKILL.match(c["skill"]) else "(?)" out.append(f"- {ident} ({c['installs']} inst.): {'; '.join(why)}") return "\n".join(out), {c["skill"].lower() for c in ok} def find(queries, category, a, exclude=frozenset()): """Search both pools, vet, score. Returns (vetted candidates, installed names that were skipped).""" qwords = set().union(*(words(q) for q in queries)) if queries else set() if category: qwords |= words(category) curated = load_curated(category) if not category: # no category picked: only curated items that match the words compete curated = [c for c in curated if qwords & words(c["skill"].replace("-", " ") + " " + c["why_pl"] + " " + c["why_en"])] groups = group_by_name(curated + (search_registry(queries, a.limit) if queries else [])) skipped = {g[0]["skill"].lower() for g in groups if g[0]["skill"].lower() in exclude} groups = [g for g in groups if g[0]["skill"].lower() not in exclude] for g in groups: # curated entries the queries did not surface: one lookup for their install count c = g[0] if c["pool"] == "curated" and not c.get("page") and not c["installs"]: twin = [r for r in search_registry([c["skill"]], 10) if r["id"].lower() == c["id"].lower()] c["installs"] = twin[0]["installs"] if twin else 0 cur = [g for g in groups if g[0]["pool"] == "curated"] reg = sorted((g for g in groups if g[0]["pool"] != "curated"), key=lambda g: (sum(c["hits"] for c in g), max(c["installs"] for c in g)), reverse=True) # Curated never eats the whole vet budget: the registry always gets at least half of it. picked = cur[: a.vet // 2] + reg[: a.vet - min(len(cur), a.vet // 2)] vetted = [] for g in picked: tried = [vet(dict(c)) for c in g[:3]] # up to 3 owners of the same name; the best source represents them best = max(tried, key=lambda c: (passes(c), c["pool"] == "curated", (c.get("owner") or "").lower() in OFFICIAL, c["stars"])) best["copies"] = len({c["source"].lower() for c in g}) - 1 best["hits"] = sum(c["hits"] for c in g) if best["pool"] == "registry": best["installs"] = max(c["installs"] for c in tried if c["id"] == best["id"]) vetted.append(score(best, qwords)) return vetted, skipped def main(): if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") # Windows consoles default to cp1250/cp1252 ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--query", "-q", action="append", default=[], help="search phrase; repeat for synonyms and both languages") ap.add_argument("--category", help="curated category: ecommerce, video, design, marketing, coding, documents, fitness, cinematography") ap.add_argument("--top", type=int, default=5) ap.add_argument("--limit", type=int, default=15, help="registry results per query") ap.add_argument("--vet", type=int, default=10, help="how many candidates to open and scan (per scan reason: half)") ap.add_argument("--lang", choices=["pl", "en"], default="pl") ap.add_argument("--json", action="store_true", help="print machine-readable results instead of the table") ap.add_argument("--scan", metavar="DIR", help="read-only scan of DIR (names only), then search for what it suggests") ap.add_argument("--scan-only", action="store_true", help="with --scan: print the signals and the search plan, search nothing") a = ap.parse_args() if not a.query and not a.category and not a.scan: ap.error("give at least one --query, a --category or --scan DIR") STATE["lang"] = a.lang have = installed_skills() if a.scan: sig = scan(a.scan) plan = plan_from_scan(sig, a.lang) print(json.dumps({"signals": sig, "plan": [{"category": c, "queries": q, "reason": r} for c, q, r in plan], "installed": sorted(have)}, ensure_ascii=False, indent=1)) if a.scan_only or not plan: return if not os.environ.get("GITHUB_TOKEN"): print("tip: a scan checks many repos; without GITHUB_TOKEN GitHub may stop answering after 60 requests", file=sys.stderr) a.vet = max(4, a.vet // 2) shown = set() # a skill recommended for one reason is not repeated under the next for cat, queries, reason in plan: vetted, skipped = find(queries, cat, a, exclude=have | shown) text, picks = render(vetted, queries, a.lang, min(a.top, 3), skipped - shown) print(f"\n> {reason}\n\n{text}") shown |= picks return vetted, skipped = find(a.query, a.category, a, exclude=have) if a.json: print(json.dumps({"rate_limited": STATE["rate_limited"], "skipped_installed": sorted(skipped), "candidates": vetted}, ensure_ascii=False, indent=1, default=str)) else: print(render(vetted, a.query or [a.category], a.lang, a.top, skipped)[0]) if __name__ == "__main__": main()