#!/usr/bin/env python3 import argparse import json import os import re from pathlib import Path from typing import Any from urllib.parse import quote, urlparse from urllib.request import Request, urlopen API_ROOT = "https://api.github.com" API_VERSION = "2022-11-28" USER_AGENT = "yao-meta-skill/1.0" DEFAULT_TOP_N = 3 DEFAULT_PER_PAGE = 8 MIN_STARS = 50 ALLOWED_API_HOSTS = {"api.github.com"} STOPWORDS = { "the", "and", "for", "with", "into", "from", "this", "that", "your", "will", "make", "turn", "build", "create", "reusable", "skill", "package", "agent", "workflow", "prompt", "notes", "docs", "internal", "team", "real", "output", "outputs", "input", "inputs", } PATTERN_RULES = [ { "category": "method", "keywords": ["workflow", "process", "pipeline", "orchestrat", "runbook", "playbook"], "borrow": "Borrow the way it turns a messy workflow into a repeatable operating path.", "avoid": "Do not import process overhead that only exists for that project's scale.", }, { "category": "execution", "keywords": ["cli", "command", "shell", "script", "automation", "toolchain"], "borrow": "Borrow the clear execution entrypoints and command structure.", "avoid": "Do not copy repo-specific commands or environment assumptions verbatim.", }, { "category": "evaluation", "keywords": ["test", "benchmark", "eval", "assert", "validation", "quality"], "borrow": "Borrow explicit validation and quality gates that make iteration safer.", "avoid": "Do not clone heavyweight evaluation scaffolding if a lighter gate is enough here.", }, { "category": "structure", "keywords": ["template", "example", "guide", "docs", "reference", "starter"], "borrow": "Borrow the way it separates explanation, examples, and reusable structure.", "avoid": "Do not mirror documentation bulk that adds context cost without improving reliability.", }, { "category": "portability", "keywords": ["adapter", "portable", "plugin", "integration", "export", "compatibility"], "borrow": "Borrow how it keeps core semantics stable across environments or integrations.", "avoid": "Do not inherit platform lock-in or vendor-specific branching unless truly required.", }, { "category": "governance", "keywords": ["govern", "policy", "security", "audit", "compliance", "review"], "borrow": "Borrow the explicit review, safety, or operational trust boundaries.", "avoid": "Do not import compliance-heavy process where the new skill does not need it.", }, ] def github_headers() -> dict[str, str]: headers = { "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": API_VERSION, "User-Agent": USER_AGENT, } token = os.getenv("GITHUB_TOKEN") or os.getenv("GH_TOKEN") if token: headers["Authorization"] = f"Bearer {token}" return headers def ensure_allowed_api_url(url: str) -> str: parsed = urlparse(url) if parsed.scheme != "https" or parsed.hostname not in ALLOWED_API_HOSTS: raise ValueError(f"Refusing outbound request to non-GitHub API host: {url}") return url def api_get(url: str) -> dict[str, Any]: request = Request(ensure_allowed_api_url(url), headers=github_headers()) with urlopen(request, timeout=15) as response: # nosec B310 - GitHub API fetch data = response.read().decode("utf-8") return json.loads(data) def load_fixture_bundle(fixture_dir: Path | None) -> dict[str, Any] | None: if not fixture_dir: return None bundle_path = fixture_dir / "bundle.json" if not bundle_path.exists(): raise FileNotFoundError(f"Missing fixture bundle: {bundle_path}") return json.loads(bundle_path.read_text(encoding="utf-8")) def build_query(seed_text: str) -> str: words = re.findall(r"[A-Za-z][A-Za-z0-9_-]{2,}", seed_text.lower()) keywords = [] seen = set() for word in words: if word in STOPWORDS or word in seen: continue seen.add(word) keywords.append(word) if len(keywords) == 4: break if keywords: return " ".join(keywords) compact = re.sub(r"\s+", " ", seed_text.strip()) return compact[:64] or "agent workflow automation" def search_repositories(query: str, top_n: int, fixture_bundle: dict[str, Any] | None = None) -> list[dict[str, Any]]: if fixture_bundle is not None: items = fixture_bundle.get("search_items", []) return items[:top_n] github_query = f"{query} fork:false archived:false stars:>={MIN_STARS}" url = ( f"{API_ROOT}/search/repositories?q={quote(github_query)}" f"&sort=stars&order=desc&per_page={DEFAULT_PER_PAGE}" ) payload = api_get(url) items = payload.get("items", []) filtered = [item for item in items if not item.get("fork") and not item.get("archived")] return filtered[:top_n] def fetch_readme(full_name: str, fixture_bundle: dict[str, Any] | None = None) -> str: if fixture_bundle is not None: return fixture_bundle.get("readmes", {}).get(full_name, "") url = f"{API_ROOT}/repos/{full_name}/readme" request = Request( ensure_allowed_api_url(url), headers={**github_headers(), "Accept": "application/vnd.github.raw+json"}, ) with urlopen(request, timeout=15) as response: # nosec B310 - GitHub API fetch return response.read().decode("utf-8", errors="replace") def detect_patterns(text: str) -> list[dict[str, str]]: lowered = text.lower() hits = [] for rule in PATTERN_RULES: matched = [token for token in rule["keywords"] if token in lowered] if matched: hits.append( { "category": rule["category"], "matched": ", ".join(matched[:4]), "borrow": rule["borrow"], "avoid": rule["avoid"], } ) return hits[:4] def repo_summary(repo: dict[str, Any], readme_text: str) -> dict[str, Any]: combined = " ".join( filter( None, [ repo.get("name", ""), repo.get("description", "") or "", " ".join(repo.get("topics") or []), readme_text[:4000], ], ) ) patterns = detect_patterns(combined) if not patterns: patterns = [ { "category": "general", "matched": "high-level fit", "borrow": "Borrow only the parts that make the new skill clearer, safer, or easier to reuse.", "avoid": "Do not import source-specific scope, branding, or excess weight.", } ] borrow = [item["borrow"] for item in patterns[:3]] avoid = [item["avoid"] for item in patterns[:3]] return { "full_name": repo.get("full_name"), "html_url": repo.get("html_url"), "description": repo.get("description"), "stars": repo.get("stargazers_count", 0), "topics": repo.get("topics") or [], "primary_language": repo.get("language"), "patterns": patterns, "borrow": borrow, "avoid": avoid, "readme_excerpt": readme_text[:1200].strip(), } def cross_repo_insights(repos: list[dict[str, Any]]) -> dict[str, list[str]]: borrow = [] avoid = [] seen_borrow = set() seen_avoid = set() for repo in repos: for item in repo.get("borrow", []): if item not in seen_borrow: seen_borrow.add(item) borrow.append(item) for item in repo.get("avoid", []): if item not in seen_avoid: seen_avoid.add(item) avoid.append(item) borrow.append("Ask the user which of these patterns feels most worth borrowing before freezing the package shape.") return {"borrow": borrow[:6], "avoid": avoid[:6]} def repo_to_external_reference(repo: dict[str, Any]) -> dict[str, str]: primary = repo.get("patterns", [{}])[0] return { "name": repo.get("full_name") or "Unnamed repo", "category": primary.get("category", "general"), "borrow": repo.get("borrow", ["Borrow the strongest reusable pattern."])[0], "avoid": repo.get("avoid", ["Do not copy source-specific language or excess scope."])[0], "source": "external", } def render_markdown(payload: dict[str, Any]) -> str: lines = [ "# GitHub Benchmark Scan", "", f"- Query: `{payload['query']}`", f"- Source: `{payload['source']}`", f"- Top repositories: `{len(payload['repositories'])}`", "", "## Suggested Next Step", "", "Review the three benchmark objects below, then decide whether any of their patterns are worth borrowing into the new skill.", "", ] if not payload["repositories"]: lines.extend( [ "No benchmark repositories were collected.", "", "Possible reasons:", "- the query was too weak or too local", "- the network or API rate limit blocked the scan", "- this skill idea may need a manual benchmark query", ] ) return "\n".join(lines).strip() + "\n" lines.extend(["## Top 3 Benchmark Repositories", ""]) for repo in payload["repositories"]: lines.extend( [ f"### {repo['full_name']}", f"- URL: {repo['html_url']}", f"- Stars: `{repo['stars']}`", f"- Description: {repo.get('description') or 'No public description provided.'}", f"- Topics: {', '.join(repo.get('topics') or []) or 'None'}", "", "#### Patterns worth studying", "", ] ) for pattern in repo["patterns"]: lines.append( f"- **{pattern['category']}** (`{pattern['matched']}`): {pattern['borrow']}" ) lines.extend(["", "#### Borrow", ""]) for item in repo["borrow"]: lines.append(f"- {item}") lines.extend(["", "#### Avoid", ""]) for item in repo["avoid"]: lines.append(f"- {item}") lines.extend(["", "#### README glimpse", "", repo["readme_excerpt"] or "No README excerpt collected.", ""]) lines.extend(["## Cross-Repo Borrow Recommendations", ""]) for item in payload["cross_repo"]["borrow"]: lines.append(f"- {item}") lines.extend(["", "## Cross-Repo Avoid Recommendations", ""]) for item in payload["cross_repo"]["avoid"]: lines.append(f"- {item}") lines.extend( [ "", "## Borrow Prompt", "", payload["borrow_prompt"], "", ] ) return "\n".join(lines).strip() + "\n" def run_github_benchmark_scan( skill_dir: Path, query: str, top_n: int = DEFAULT_TOP_N, fixture_dir: Path | None = None, output_md: Path | None = None, output_json: Path | None = None, ) -> dict[str, Any]: skill_dir = skill_dir.resolve() reports_dir = skill_dir / "reports" reports_dir.mkdir(parents=True, exist_ok=True) output_md = output_md or reports_dir / "github-benchmark-scan.md" output_json = output_json or reports_dir / "github-benchmark-scan.json" fixture_bundle = load_fixture_bundle(fixture_dir) warnings = [] try: repos = search_repositories(query, top_n=top_n, fixture_bundle=fixture_bundle) repo_summaries = [] for repo in repos: readme_text = fetch_readme(repo["full_name"], fixture_bundle=fixture_bundle) repo_summaries.append(repo_summary(repo, readme_text)) cross_repo = cross_repo_insights(repo_summaries) payload = { "ok": True, "query": query, "source": "fixture" if fixture_bundle is not None else "github-api", "network_boundary": "github-api-only", "repositories": repo_summaries, "cross_repo": cross_repo, "external_references": [repo_to_external_reference(repo) for repo in repo_summaries], "borrow_prompt": "I found three public GitHub projects worth studying. Do you want to borrow any of these patterns for method, structure, execution, or portability?", "warnings": warnings, } except Exception as exc: # pragma: no cover - network failures are environment-dependent payload = { "ok": False, "query": query, "source": "fixture" if fixture_bundle is not None else "github-api", "network_boundary": "github-api-only", "repositories": [], "cross_repo": { "borrow": [], "avoid": [], }, "external_references": [], "borrow_prompt": "No benchmark suggestions were collected yet. You can retry with a stronger query or add manual references.", "warnings": [f"GitHub benchmark scan failed: {exc}"], } output_md.write_text(render_markdown(payload), encoding="utf-8") output_json.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") payload["artifacts"] = { "markdown": str(output_md), "json": str(output_json), } payload["skill_dir"] = str(skill_dir) return payload def main() -> None: parser = argparse.ArgumentParser(description="Search GitHub for top benchmark repositories and extract reusable patterns.") parser.add_argument("skill_dir", nargs="?", default=".") parser.add_argument("--query", required=True) parser.add_argument("--top-n", type=int, default=DEFAULT_TOP_N) parser.add_argument("--fixture-dir") parser.add_argument("--output-md") parser.add_argument("--output-json") args = parser.parse_args() result = run_github_benchmark_scan( Path(args.skill_dir), query=args.query, top_n=args.top_n, fixture_dir=Path(args.fixture_dir).resolve() if args.fixture_dir else None, output_md=Path(args.output_md).resolve() if args.output_md else None, output_json=Path(args.output_json).resolve() if args.output_json else None, ) print(json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()