Spaces:
Paused
Paused
| """ | |
| GitHub Repository Assessor - Real GitHub assessment (no mocks) | |
| Assesses GitHub repos for architecture, dependencies, build process, test coverage, security | |
| """ | |
| import asyncio | |
| import aiohttp | |
| import json | |
| import time | |
| from typing import Dict, Any, List, Optional | |
| from dataclasses import dataclass | |
| from .core import BaseCrawler, Source, SubjectType, EvidenceType | |
| class GitHubAssessment: | |
| """Assessment result for a GitHub repository""" | |
| repo: str | |
| stars: int | |
| forks: int | |
| open_issues: int | |
| watchers: int | |
| has_readme: bool | |
| has_license: bool | |
| has_tests: bool | |
| has_ci: bool | |
| has_workflow: bool | |
| languages: Dict[str, int] | |
| topics: List[str] | |
| archived: bool | |
| size_kb: int | |
| updated_at: str | |
| created_at: str | |
| verified: bool | |
| timestamp: str | |
| class GitHubAssessor(BaseCrawler): | |
| """Real GitHub repository assessor - no mocks""" | |
| def __init__(self, token: Optional[str] = None): | |
| super().__init__("github_assessor") | |
| self.token = token | |
| self.base_url = "https://api.github.com" | |
| self.headers = { | |
| "Accept": "application/vnd.github.v3+json", | |
| "User-Agent": "LayerCrawler/1.0" | |
| } | |
| if token: | |
| self.headers["Authorization"] = f"token {token}" | |
| async def get_repo_info(self, owner: str, repo: str) -> Optional[Dict[str, Any]]: | |
| """Get repository information from GitHub API""" | |
| try: | |
| async with aiohttp.ClientSession() as session: | |
| async with session.get( | |
| f"{self.base_url}/repos/{owner}/{repo}", | |
| headers=self.headers, | |
| timeout=aiohttp.ClientTimeout(total=30) | |
| ) as resp: | |
| if resp.status == 200: | |
| return await resp.json() | |
| elif resp.status == 404: | |
| return None | |
| else: | |
| return None | |
| except Exception: | |
| return None | |
| async def get_repo_contents(self, owner: str, repo: str) -> List[Dict[str, Any]]: | |
| """Get repository contents""" | |
| try: | |
| async with aiohttp.ClientSession() as session: | |
| async with session.get( | |
| f"{self.base_url}/repos/{owner}/{repo}/contents/", | |
| headers=self.headers, | |
| timeout=aiohttp.ClientTimeout(total=30) | |
| ) as resp: | |
| if resp.status == 200: | |
| return await resp.json() | |
| return [] | |
| except Exception: | |
| return [] | |
| async def assess_repo(self, repo_full_name: str) -> GitHubAssessment: | |
| """Assess a GitHub repository""" | |
| start_time = time.time() | |
| try: | |
| owner, repo = repo_full_name.split("/", 1) | |
| except: | |
| return GitHubAssessment( | |
| repo=repo_full_name, | |
| stars=0, | |
| forks=0, | |
| open_issues=0, | |
| watchers=0, | |
| has_readme=False, | |
| has_license=False, | |
| has_tests=False, | |
| has_ci=False, | |
| has_workflow=False, | |
| languages={}, | |
| topics=[], | |
| archived=False, | |
| size_kb=0, | |
| updated_at="", | |
| created_at="", | |
| verified=False, | |
| timestamp=time.strftime("%Y-%m-%d %H:%M:%S") | |
| ) | |
| repo_info = await self.get_repo_info(owner, repo) | |
| if not repo_info: | |
| return GitHubAssessment( | |
| repo=repo_full_name, | |
| stars=0, | |
| forks=0, | |
| open_issues=0, | |
| watchers=0, | |
| has_readme=False, | |
| has_license=False, | |
| has_tests=False, | |
| has_ci=False, | |
| has_workflow=False, | |
| languages={}, | |
| topics=[], | |
| archived=False, | |
| size_kb=0, | |
| updated_at="", | |
| created_at="", | |
| verified=False, | |
| timestamp=time.strftime("%Y-%m-%d %H:%M:%S") | |
| ) | |
| # Get contents to check for specific files | |
| contents = await self.get_repo_contents(owner, repo) | |
| has_readme = any( | |
| c.get("name", "").lower().startswith("readme") | |
| for c in contents if isinstance(c, dict) | |
| ) | |
| has_license = any( | |
| c.get("name", "").lower().startswith("license") | |
| for c in contents if isinstance(c, dict) | |
| ) | |
| # Check for test files | |
| has_tests = any( | |
| "test" in c.get("name", "").lower() | |
| for c in contents if isinstance(c, dict) | |
| ) | |
| # Check for CI/workflows | |
| has_ci = any( | |
| c.get("name", "").lower() in [".github", "ci", ".circleci", ".travis.yml"] | |
| for c in contents if isinstance(c, dict) | |
| ) | |
| has_workflow = has_ci # Simplified | |
| # Get languages | |
| languages = repo_info.get("language", {}) | |
| if isinstance(languages, str): | |
| languages = {languages: 100} | |
| return GitHubAssessment( | |
| repo=repo_full_name, | |
| stars=repo_info.get("stargazers_count", 0), | |
| forks=repo_info.get("forks_count", 0), | |
| open_issues=repo_info.get("open_issues_count", 0), | |
| watchers=repo_info.get("subscribers_count", 0), | |
| has_readme=has_readme, | |
| has_license=has_license, | |
| has_tests=has_tests, | |
| has_ci=has_ci, | |
| has_workflow=has_workflow, | |
| languages=languages, | |
| topics=repo_info.get("topics", []), | |
| archived=repo_info.get("archived", False), | |
| size_kb=repo_info.get("size", 0), | |
| updated_at=repo_info.get("updated_at", ""), | |
| created_at=repo_info.get("created_at", ""), | |
| verified=True, | |
| timestamp=time.strftime("%Y-%m-%d %H:%M:%S") | |
| ) | |
| async def crawl(self, source: Source) -> List: | |
| """Crawl GitHub repo and generate signals""" | |
| signals = [] | |
| repo = source.location | |
| assessment = await self.assess_repo(repo) | |
| # Generate signals based on assessment | |
| self.add_signal( | |
| SubjectType.CODE, | |
| EvidenceType.BUILD_VERIFIED, | |
| assessment.has_readme and assessment.has_license, | |
| source.location | |
| ) | |
| self.add_signal( | |
| SubjectType.CODE, | |
| EvidenceType.TESTS_VERIFIED, | |
| assessment.has_tests, | |
| source.location | |
| ) | |
| self.add_signal( | |
| SubjectType.LICENSE, | |
| EvidenceType.BUILD_VERIFIED, | |
| assessment.has_license, | |
| source.location | |
| ) | |
| self.add_signal( | |
| SubjectType.TEST_BUILD, | |
| EvidenceType.BUILD_VERIFIED, | |
| assessment.has_ci, | |
| source.location | |
| ) | |
| signals.extend(self.signals) | |
| return signals | |
| async def batch_assess(self, repos: List[str]) -> List[GitHubAssessment]: | |
| """Assess multiple repos in parallel""" | |
| tasks = [self.assess_repo(repo) for repo in repos] | |
| return await asyncio.gather(*tasks) | |
| # Example usage | |
| if __name__ == "__main__": | |
| async def main(): | |
| import os | |
| token = os.environ.get("GITHUB_TOKEN") | |
| assessor = GitHubAssessor(token=token) | |
| # Test some popular repos | |
| repos = [ | |
| "facebook/react", | |
| "tensorflow/tensorflow", | |
| "openai/openai-python", | |
| ] | |
| print("Assessing GitHub repositories...") | |
| results = await assessor.batch_assess(repos) | |
| for result in results: | |
| print(f"\n{result.repo}:") | |
| print(f" Stars: {result.stars}") | |
| print(f" Forks: {result.forks}") | |
| print(f" Open Issues: {result.open_issues}") | |
| print(f" Has README: {result.has_readme}") | |
| print(f" Has License: {result.has_license}") | |
| print(f" Has Tests: {result.has_tests}") | |
| print(f" Has CI: {result.has_ci}") | |
| print(f" Languages: {result.languages}") | |
| print(f" Archived: {result.archived}") | |
| asyncio.run(main()) | |