Spaces:
Configuration error
Configuration error
| """GitHub Service module. | |
| Interfaces with the GitHub API to fetch repository structures, file contents, | |
| issue lists, and pull request information, or clones repositories locally. | |
| """ | |
| import os | |
| import re | |
| import shutil | |
| import stat | |
| import urllib.parse | |
| import logging | |
| from typing import Dict, List, Any, Optional | |
| import requests | |
| from services.storage_paths import get_cloned_repos_dir | |
| from utils.subprocess_runner import ( | |
| run_safe_command, | |
| SafeSubprocessError, | |
| SHORT_GIT_TIMEOUT, | |
| INSPECTION_TIMEOUT, | |
| CLONE_TIMEOUT, | |
| ) | |
| logger = logging.getLogger(__name__) | |
| # Client-safe failure messages. These must never disclose whether a repository | |
| # exists, whether credentials are configured, or whether credentials are valid, | |
| # and must never carry git stderr, git commands, or exception text. | |
| ACCESS_DENIED_MESSAGE = "Repository not found or access denied." | |
| NETWORK_FAILURE_MESSAGE = ( | |
| "Network failure: unable to reach the repository host. Please try again." | |
| ) | |
| GIT_FAILURE_MESSAGE = "Connection failure: unable to complete the repository operation." | |
| # Internal diagnostic categories. Every category is logged distinctly for | |
| # operators; several intentionally collapse to one public message. | |
| _CATEGORY_INVALID_CREDENTIALS = "invalid_credentials" | |
| _CATEGORY_PERMISSION_DENIED = "permission_denied" | |
| _CATEGORY_NOT_FOUND = "not_found" | |
| _CATEGORY_NETWORK = "network" | |
| _CATEGORY_GIT_FAILURE = "git_failure" | |
| _CATEGORY_UNEXPECTED = "unexpected" | |
| _NON_DISCLOSING_CATEGORIES = frozenset( | |
| { | |
| _CATEGORY_INVALID_CREDENTIALS, | |
| _CATEGORY_PERMISSION_DENIED, | |
| _CATEGORY_NOT_FOUND, | |
| } | |
| ) | |
| class InvalidGitHubRepoURLError(ValueError): | |
| """Raised when the provided repo URL/identifier is not a supported GitHub format.""" | |
| class RepositoryNotFoundError(RuntimeError): | |
| """Raised when the target repository cannot be accessed (404-like git failure).""" | |
| class GitOperationError(RuntimeError): | |
| """Raised when a git operation fails for network or execution reasons.""" | |
| class BranchNotFoundError(ValueError): | |
| """Raised when the requested branch/ref does not exist in the repository.""" | |
| def classify_git_failure( | |
| stderr: str, returncode: int = 0, timed_out: bool = False | |
| ) -> str: | |
| """Classify a git failure into an internal diagnostic category. | |
| Args: | |
| stderr: Redacted git stderr output. | |
| returncode: Process exit code. | |
| timed_out: Whether the command exceeded its timeout. | |
| Returns: | |
| One of the internal ``_CATEGORY_*`` values. | |
| """ | |
| if timed_out: | |
| return _CATEGORY_NETWORK | |
| text = (stderr or "").lower() | |
| # Local execution problems must not be mistaken for repository access | |
| # results, because their stderr also contains "not found". | |
| if any( | |
| kw in text | |
| for kw in ( | |
| "executable not found", | |
| "no such file or directory", | |
| "cannot run program", | |
| ) | |
| ): | |
| return _CATEGORY_GIT_FAILURE | |
| if any( | |
| kw in text | |
| for kw in ( | |
| "could not resolve host", | |
| "temporary failure", | |
| "network is unreachable", | |
| "connection refused", | |
| "connection timed out", | |
| "timed out", | |
| "operation timed out", | |
| "failed to connect", | |
| ) | |
| ): | |
| return _CATEGORY_NETWORK | |
| if any( | |
| kw in text | |
| for kw in ( | |
| "401", | |
| "invalid username or password", | |
| "bad credentials", | |
| "authentication failed", | |
| "could not read username", | |
| "terminal prompts disabled", | |
| ) | |
| ): | |
| return _CATEGORY_INVALID_CREDENTIALS | |
| if any( | |
| kw in text | |
| for kw in ( | |
| "403", | |
| "forbidden", | |
| "permission denied", | |
| "write access", | |
| "authorization", | |
| ) | |
| ): | |
| return _CATEGORY_PERMISSION_DENIED | |
| if any( | |
| kw in text | |
| for kw in ("404", "not found", "repository not found", "fatal: repository") | |
| ): | |
| return _CATEGORY_NOT_FOUND | |
| if returncode != 0: | |
| return _CATEGORY_GIT_FAILURE | |
| return _CATEGORY_UNEXPECTED | |
| class GitHubConfig: | |
| token: Optional[str] = None | |
| def load_token(cls) -> Optional[str]: | |
| if cls.token is None: | |
| from core.config import settings | |
| cls.token = settings.github_token | |
| return cls.token | |
| class GitHubService: | |
| """Wrapper class containing helpers to query GitHub repositories or clone them locally.""" | |
| _APPROVED_HOST = "github.com" | |
| _MAX_SOURCE_FILE_SIZE_BYTES = 2 * 1024 * 1024 | |
| _REPOSITORY_PART = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") | |
| def _raise_safe_git_error( | |
| self, | |
| category: str, | |
| *, | |
| repo_fullName: str, | |
| cmd: Optional[List[str]] = None, | |
| stderr: str = "", | |
| returncode: int = 0, | |
| exc: Optional[BaseException] = None, | |
| ) -> None: | |
| """Log full git diagnostics for operators, then raise a client-safe error. | |
| Diagnostics (category, git command, repository, exit code, stderr and the | |
| underlying exception) are written to the server log only. The raised | |
| exception message is generic and never reveals repository existence or | |
| credential state. | |
| Raises: | |
| RepositoryNotFoundError: For credential, permission and not-found cases. | |
| GitOperationError: For network, git-execution and unexpected cases. | |
| """ | |
| logger.error( | |
| "GitHub operation failed | category=%s repository=%s exit_code=%s " | |
| "credentials_configured=%s command=%s stderr=%s", | |
| category, | |
| repo_fullName, | |
| returncode, | |
| bool(self.token), | |
| " ".join(cmd) if cmd else "n/a", | |
| (stderr or "").strip() or "n/a", | |
| exc_info=exc if exc is not None else False, | |
| ) | |
| if category in _NON_DISCLOSING_CATEGORIES: | |
| raise RepositoryNotFoundError(ACCESS_DENIED_MESSAGE) | |
| if category == _CATEGORY_NETWORK: | |
| raise GitOperationError(NETWORK_FAILURE_MESSAGE) | |
| raise GitOperationError(GIT_FAILURE_MESSAGE) | |
| def _run_git( | |
| self, | |
| cmd: List[str], | |
| *, | |
| timeout: float, | |
| repo_fullName: str, | |
| env: Optional[Dict[str, str]] = None, | |
| secrets: Optional[List[str]] = None, | |
| cwd: Optional[str] = None, | |
| ) -> Any: | |
| """Run a git command, converting execution failures into client-safe errors.""" | |
| try: | |
| return run_safe_command( | |
| cmd, timeout=timeout, env=env, secrets=secrets or [], cwd=cwd | |
| ) | |
| except SafeSubprocessError as sub_exc: | |
| self._raise_safe_git_error( | |
| classify_git_failure( | |
| sub_exc.stderr, sub_exc.returncode, sub_exc.timed_out | |
| ), | |
| repo_fullName=repo_fullName, | |
| cmd=cmd, | |
| stderr=sub_exc.stderr, | |
| returncode=sub_exc.returncode, | |
| exc=sub_exc, | |
| ) | |
| def __init__(self, token: Optional[str] = None) -> None: | |
| """Initializes the GitHub client with an optional authentication token. | |
| Args: | |
| token: GitHub Personal Access Token (PAT). | |
| """ | |
| self.token = token or GitHubConfig.load_token() | |
| self.session = requests.Session() | |
| if self.token: | |
| self.session.headers.update({"Authorization": f"Bearer {self.token}"}) | |
| self.session.headers.update({"Accept": "application/vnd.github+json"}) | |
| # Verify request headers | |
| logger.info( | |
| "GitHub Authorization header present: %s", | |
| "Authorization" in self.session.headers, | |
| ) | |
| def parse_repo_url(self, repo_url: str) -> Dict[str, str]: | |
| """Parse a supported GitHub URL or ``owner/repo`` identifier safely.""" | |
| raw_url = repo_url.strip() | |
| if not raw_url: | |
| raise ValueError("Invalid GitHub repository URL.") | |
| owner: Optional[str] = None | |
| repo: Optional[str] = None | |
| parsed = urllib.parse.urlparse(raw_url) | |
| if parsed.scheme or parsed.netloc: | |
| if ( | |
| parsed.scheme not in {"http", "https"} | |
| or parsed.hostname != self._APPROVED_HOST | |
| or parsed.port is not None | |
| or parsed.username is not None | |
| or parsed.password is not None | |
| or parsed.params | |
| or parsed.query | |
| or parsed.fragment | |
| ): | |
| raise ValueError("Invalid GitHub repository URL.") | |
| path_parts = [part for part in parsed.path.split("/") if part] | |
| else: | |
| path_parts = [part for part in raw_url.split("/") if part] | |
| if len(path_parts) == 3 and path_parts[0] == self._APPROVED_HOST: | |
| path_parts = path_parts[1:] | |
| if len(path_parts) == 2: | |
| owner, repo = path_parts | |
| if repo.endswith(".git"): | |
| repo = repo[:-4] | |
| if ( | |
| not owner | |
| or not repo | |
| or not self._REPOSITORY_PART.fullmatch(owner) | |
| or not self._REPOSITORY_PART.fullmatch(repo) | |
| ): | |
| raise ValueError("Invalid GitHub repository URL.") | |
| return {"owner": owner, "repo": repo} | |
| def get_local_repo_path(self, repo_fullName: str) -> str: | |
| """Returns the local path where a repository is or should be cloned. | |
| Args: | |
| repo_fullName: GitHub owner/repo identifier. | |
| Returns: | |
| The local directory path. | |
| """ | |
| # Store clones OUTSIDE the project tree so uvicorn --reload (WatchFiles) | |
| # does not treat clone activity as source-code changes. Configurable via | |
| # CLONED_REPOS_PATH; defaults to ~/.repo_intelligence/cloned_repos. | |
| base_dir = str(get_cloned_repos_dir()) | |
| safe_name = repo_fullName.replace("/", "_").replace("\\", "_") | |
| return os.path.abspath(os.path.join(base_dir, safe_name)) | |
| def clone_repository(self, repo_url: str, branch: Optional[str] = None) -> str: | |
| """Clones a GitHub repository to a local directory using Git CLI with fallback reliability. | |
| Args: | |
| repo_url: GitHub repository URL or owner/repo identifier. | |
| branch: Optional branch/ref name to clone. If provided, existence is validated. | |
| Returns: | |
| The local path to the cloned repository. | |
| """ | |
| try: | |
| parsed = self.parse_repo_url(repo_url) | |
| except ValueError as e: | |
| raise InvalidGitHubRepoURLError(str(e)) | |
| repo_fullName = f"{parsed['owner']}/{parsed['repo']}" | |
| dest_dir = self.get_local_repo_path(repo_fullName) | |
| # Always use the canonical approved host. Credentials are never attached | |
| # to a caller-provided remote URL. | |
| public_url = f"https://{self._APPROVED_HOST}/{repo_fullName}.git" | |
| # Prepare authorization environment & secret redaction list | |
| extra_env: Dict[str, str] = {} | |
| secrets_list: List[str] = [] | |
| if self.token: | |
| import base64 | |
| auth_str = base64.b64encode( | |
| f"x-access-token:{self.token}".encode("utf-8") | |
| ).decode("utf-8") | |
| extra_env = { | |
| "GIT_CONFIG_KEY_0": "http.extraHeader", | |
| "GIT_CONFIG_VALUE_0": f"Authorization: Basic {auth_str}", | |
| } | |
| secrets_list = [self.token, auth_str] | |
| # 2. Check if repository is publicly accessible (anonymous check) | |
| is_public = False | |
| try: | |
| cmd_check = ["git", "ls-remote", public_url, "HEAD"] | |
| res = run_safe_command( | |
| cmd_check, timeout=SHORT_GIT_TIMEOUT, secrets=secrets_list | |
| ) | |
| if res.returncode == 0: | |
| is_public = True | |
| except Exception: | |
| pass | |
| # 3. Determine actual URL to use | |
| clone_url = public_url | |
| if is_public: | |
| logger.info("Cloning public repository anonymously: %s", repo_fullName) | |
| elif self.token: | |
| logger.info("Cloning private repository using PAT: %s", repo_fullName) | |
| else: | |
| logger.info( | |
| "Cloning repository anonymously (no token available): %s", repo_fullName | |
| ) | |
| # 4. Perform ls-remote connection diagnostics | |
| cmd_check = ["git", "ls-remote", clone_url, "HEAD"] | |
| res = self._run_git( | |
| cmd_check, | |
| timeout=SHORT_GIT_TIMEOUT, | |
| repo_fullName=repo_fullName, | |
| env=extra_env, | |
| secrets=secrets_list, | |
| ) | |
| if res.returncode != 0: | |
| self._raise_safe_git_error( | |
| classify_git_failure(res.stderr, res.returncode), | |
| repo_fullName=repo_fullName, | |
| cmd=cmd_check, | |
| stderr=res.stderr, | |
| returncode=res.returncode, | |
| ) | |
| # 5. Resolve Branch name (and auto-discover if requested branch is default 'main' but not present) | |
| actual_branch = branch | |
| if branch: | |
| # Check if requested branch exists | |
| cmd_check = ["git", "ls-remote", "--heads", clone_url, branch] | |
| res_check = self._run_git( | |
| cmd_check, | |
| timeout=INSPECTION_TIMEOUT, | |
| repo_fullName=repo_fullName, | |
| env=extra_env, | |
| secrets=secrets_list, | |
| ) | |
| branch_exists = res_check.returncode == 0 and bool(res_check.stdout.strip()) | |
| if not branch_exists: | |
| if branch == "main": | |
| # Try to auto-discover default branch | |
| try: | |
| cmd_sym = ["git", "ls-remote", "--symref", clone_url, "HEAD"] | |
| res_sym = self._run_git( | |
| cmd_sym, | |
| timeout=INSPECTION_TIMEOUT, | |
| repo_fullName=repo_fullName, | |
| env=extra_env, | |
| secrets=secrets_list, | |
| ) | |
| discovered = None | |
| if res_sym.returncode == 0: | |
| for line in res_sym.stdout.splitlines(): | |
| if line.startswith("ref:"): | |
| parts = line.split() | |
| if len(parts) >= 2 and parts[1].startswith( | |
| "refs/heads/" | |
| ): | |
| discovered = parts[1].replace("refs/heads/", "") | |
| break | |
| if discovered: | |
| actual_branch = discovered | |
| logger.info( | |
| f"Branch 'main' not found. Auto-discovered default branch: '{actual_branch}'" | |
| ) | |
| else: | |
| raise BranchNotFoundError( | |
| "Branch 'main' not found, and failed to auto-discover default branch." | |
| ) | |
| except Exception as e: | |
| if isinstance( | |
| e, | |
| ( | |
| BranchNotFoundError, | |
| RepositoryNotFoundError, | |
| GitOperationError, | |
| ), | |
| ): | |
| raise e | |
| raise BranchNotFoundError( | |
| f"Branch 'main' not found for repository {repo_fullName}." | |
| ) | |
| else: | |
| raise BranchNotFoundError( | |
| f"Branch '{branch}' does not exist for repository {repo_fullName}." | |
| ) | |
| # 6. Check if target directory already exists with a valid git repository | |
| if os.path.exists(dest_dir) and os.path.exists(os.path.join(dest_dir, ".git")): | |
| target_branch = actual_branch or "main" | |
| logger.info( | |
| f"Existing repository clone found at {dest_dir}. Updating via git fetch (branch={target_branch})..." | |
| ) | |
| try: | |
| cmd_fetch = [ | |
| "git", | |
| "-c", | |
| "http.lowSpeedLimit=1000", | |
| "-c", | |
| "http.lowSpeedTime=20", | |
| "fetch", | |
| "--depth", | |
| "1", | |
| clone_url, | |
| target_branch, | |
| ] | |
| res_fetch = self._run_git( | |
| cmd_fetch, | |
| timeout=CLONE_TIMEOUT, | |
| repo_fullName=repo_fullName, | |
| env=extra_env, | |
| secrets=secrets_list, | |
| cwd=dest_dir, | |
| ) | |
| if res_fetch.returncode == 0: | |
| cmd_reset = ["git", "reset", "--hard", "FETCH_HEAD"] | |
| res_reset = self._run_git( | |
| cmd_reset, | |
| timeout=SHORT_GIT_TIMEOUT, | |
| repo_fullName=repo_fullName, | |
| env=extra_env, | |
| secrets=secrets_list, | |
| cwd=dest_dir, | |
| ) | |
| if res_reset.returncode == 0: | |
| logger.info(f"Successfully updated repository at {dest_dir}") | |
| return dest_dir | |
| except Exception as exc: | |
| logger.warning( | |
| f"Failed to update existing clone at {dest_dir}: {exc}. Cleaning up for full clone..." | |
| ) | |
| # 7. Clear target directory if it exists and needs clean re-clone | |
| if os.path.exists(dest_dir): | |
| import stat | |
| def _remove_readonly(func, path, exc_info): | |
| os.chmod(path, stat.S_IWRITE) | |
| func(path) | |
| try: | |
| shutil.rmtree(dest_dir, onerror=_remove_readonly) | |
| except Exception as e: | |
| logger.warning( | |
| f"Failed to completely remove existing directory {dest_dir}: {e}." | |
| ) | |
| os.makedirs(os.path.dirname(dest_dir), exist_ok=True) | |
| # 8. Perform Clone with HTTP low-speed protection to abort stalled transfers early | |
| logger.info( | |
| f"Cloning repository {repo_fullName} to {dest_dir} (branch={actual_branch})..." | |
| ) | |
| cmd = [ | |
| "git", | |
| "-c", | |
| "http.lowSpeedLimit=1000", | |
| "-c", | |
| "http.lowSpeedTime=20", | |
| "clone", | |
| "--depth", | |
| "1", | |
| "--single-branch", | |
| ] | |
| if actual_branch: | |
| cmd.extend(["--branch", actual_branch]) | |
| cmd.extend([clone_url, dest_dir]) | |
| result = self._run_git( | |
| cmd, | |
| timeout=CLONE_TIMEOUT, | |
| repo_fullName=repo_fullName, | |
| env=extra_env, | |
| secrets=secrets_list, | |
| ) | |
| if result.returncode != 0: | |
| self._raise_safe_git_error( | |
| classify_git_failure(result.stderr, result.returncode), | |
| repo_fullName=repo_fullName, | |
| cmd=cmd, | |
| stderr=result.stderr, | |
| returncode=result.returncode, | |
| ) | |
| return dest_dir | |
| def _safe_source_file(self, file_path: str, repository_path: str) -> bool: | |
| """Return whether a file can be read without escaping the checkout.""" | |
| try: | |
| if os.path.islink(file_path): | |
| return False | |
| file_stat = os.lstat(file_path) | |
| if not stat.S_ISREG(file_stat.st_mode): | |
| return False | |
| if file_stat.st_size > self._MAX_SOURCE_FILE_SIZE_BYTES: | |
| return False | |
| resolved_repository = os.path.realpath(repository_path) | |
| resolved_file = os.path.realpath(file_path) | |
| return ( | |
| os.path.commonpath([resolved_repository, resolved_file]) | |
| == resolved_repository | |
| ) | |
| except (OSError, ValueError): | |
| return False | |
| _BINARY_EXTENSIONS = frozenset( | |
| { | |
| ".png", | |
| ".jpg", | |
| ".jpeg", | |
| ".gif", | |
| ".ico", | |
| ".pdf", | |
| ".zip", | |
| ".tar", | |
| ".gz", | |
| ".mp3", | |
| ".mp4", | |
| ".woff", | |
| ".woff2", | |
| ".ttf", | |
| ".eot", | |
| ".svg", | |
| ".pyc", | |
| ".db", | |
| ".sqlite", | |
| ".exe", | |
| ".bin", | |
| ".dll", | |
| ".so", | |
| ".dylib", | |
| ".pkl", | |
| ".h5", | |
| } | |
| ) | |
| _MAX_TOTAL_SOURCE_BYTES = 150 * 1024 * 1024 # 150 MB total extracted text limit | |
| _MAX_TOTAL_SOURCE_FILES = 50_000 | |
| def iter_source_files( | |
| self, | |
| local_path: str, | |
| max_total_bytes: Optional[int] = None, | |
| max_files: Optional[int] = None, | |
| ): | |
| """Yields safe source files one by one to prevent loading the entire repository into RAM. | |
| Yields: | |
| Dict containing {"path": rel_path, "content": file_content} | |
| """ | |
| ignored_names = { | |
| "node_modules", | |
| ".git", | |
| "dist", | |
| "build", | |
| ".next", | |
| "venv", | |
| ".venv", | |
| "__pycache__", | |
| ".tox", | |
| "coverage", | |
| "data", | |
| } | |
| effective_max_bytes = max_total_bytes or self._MAX_TOTAL_SOURCE_BYTES | |
| effective_max_files = max_files or self._MAX_TOTAL_SOURCE_FILES | |
| files_processed = 0 | |
| bytes_processed = 0 | |
| for root, dirs, files in os.walk(local_path): | |
| dirs[:] = [ | |
| d | |
| for d in dirs | |
| if d not in ignored_names and not os.path.islink(os.path.join(root, d)) | |
| ] | |
| for file in files: | |
| file_path = os.path.join(root, file) | |
| rel_path = os.path.relpath(file_path, local_path) | |
| parts = rel_path.split(os.sep) | |
| if any(part in ignored_names for part in parts): | |
| continue | |
| ext = os.path.splitext(file)[1].lower() | |
| if ext in self._BINARY_EXTENSIONS: | |
| continue | |
| if not self._safe_source_file(file_path, local_path): | |
| logger.debug( | |
| "Skipping unsafe or oversized source file: %s", rel_path | |
| ) | |
| continue | |
| files_processed += 1 | |
| if files_processed > effective_max_files: | |
| raise ValueError( | |
| f"Repository exceeds maximum allowed file count ({effective_max_files} files)." | |
| ) | |
| try: | |
| with open(file_path, "r", encoding="utf-8", errors="ignore") as f: | |
| content = f.read() | |
| content_bytes = len(content.encode("utf-8", errors="replace")) | |
| bytes_processed += content_bytes | |
| if bytes_processed > effective_max_bytes: | |
| raise ValueError( | |
| f"Repository exceeds maximum allowed source size ({effective_max_bytes // (1024 * 1024)} MB)." | |
| ) | |
| yield { | |
| "path": rel_path.replace(os.sep, "/"), | |
| "content": content, | |
| } | |
| except OSError as exc: | |
| logger.debug( | |
| "Skipping file %s due to read error: %s", rel_path, exc | |
| ) | |
| def extract_source_files(self, local_path: str) -> List[Dict[str, Any]]: | |
| """Walks the cloned repository and extracts safe, bounded text files.""" | |
| return list(self.iter_source_files(local_path)) | |
| def fetch_repository_files( | |
| self, repo_fullName: str, branch: str = "main" | |
| ) -> List[Dict[str, Any]]: | |
| """Queries the local repository clone or fallback API to get all file metadata recursively. | |
| Args: | |
| repo_fullName: GitHub owner/repo identifier (e.g., "google/guava"). | |
| branch: Target branch name. | |
| Returns: | |
| A list of dictionary records containing file paths, types, sizes, and URLs. | |
| """ | |
| dest_dir = self.get_local_repo_path(repo_fullName) | |
| # If not cloned, clone it | |
| if not os.path.exists(dest_dir): | |
| repo_url = f"https://github.com/{repo_fullName}.git" | |
| try: | |
| self.clone_repository(repo_url) | |
| except Exception as e: | |
| logger.error(f"Clone failed inside fetch_repository_files: {e}") | |
| raise | |
| files_meta = [] | |
| ignored_names = { | |
| "node_modules", | |
| ".git", | |
| "dist", | |
| "build", | |
| ".next", | |
| "venv", | |
| ".venv", | |
| "__pycache__", | |
| ".tox", | |
| "coverage", | |
| "data", | |
| } | |
| for root, dirs, files in os.walk(dest_dir): | |
| dirs[:] = [ | |
| d | |
| for d in dirs | |
| if d not in ignored_names and not os.path.islink(os.path.join(root, d)) | |
| ] | |
| for file in files: | |
| file_path = os.path.join(root, file) | |
| rel_path = os.path.relpath(file_path, dest_dir).replace(os.sep, "/") | |
| if any(part in ignored_names for part in rel_path.split("/")): | |
| continue | |
| if not self._safe_source_file(file_path, dest_dir): | |
| continue | |
| size = os.path.getsize(file_path) | |
| files_meta.append( | |
| { | |
| "path": rel_path, | |
| "type": "blob", | |
| "size": size, | |
| "url": f"https://github.com/{repo_fullName}/blob/{branch}/{rel_path}", | |
| } | |
| ) | |
| return files_meta | |
| def fetch_file_content( | |
| self, repo_fullName: str, file_path: str, ref: str = "main" | |
| ) -> str: | |
| """Downloads/reads the raw content of a specific file from a GitHub repository clone. | |
| Args: | |
| repo_fullName: GitHub owner/repo identifier. | |
| file_path: Relative path to the file. | |
| ref: Git commit or branch ref. | |
| Returns: | |
| The raw text content of the file. | |
| """ | |
| dest_dir = self.get_local_repo_path(repo_fullName) | |
| local_file = os.path.join(dest_dir, file_path.replace("/", os.sep)) | |
| if os.path.exists(local_file): | |
| try: | |
| with open(local_file, "r", encoding="utf-8", errors="ignore") as f: | |
| return f.read() | |
| except Exception as e: | |
| raise IOError(f"Error reading file {file_path} from local storage: {e}") | |
| # If not found locally, try to fetch via GitHub API | |
| if not self.token: | |
| raise RuntimeError("GitHub credentials are not configured.") | |
| url = f"https://api.github.com/repos/{repo_fullName}/contents/{file_path}?ref={ref}" | |
| try: | |
| resp = self.session.get(url) | |
| resp.raise_for_status() | |
| data = resp.json() | |
| if "content" in data and data.get("encoding") == "base64": | |
| import base64 | |
| return base64.b64decode(data["content"]).decode( | |
| "utf-8", errors="ignore" | |
| ) | |
| elif "download_url" in data: | |
| raw_resp = requests.get(data["download_url"]) | |
| raw_resp.raise_for_status() | |
| return raw_resp.text | |
| except Exception as e: | |
| logger.error(f"Failed to fetch remote file content for {file_path}: {e}") | |
| raise FileNotFoundError( | |
| f"File {file_path} not found locally or remotely for repository {repo_fullName}." | |
| ) | |
| def fetch_issues( | |
| self, repo_fullName: str, state: str = "open" | |
| ) -> List[Dict[str, Any]]: | |
| """Queries GitHub Issues API to fetch issues for mapping analysis. | |
| Args: | |
| repo_fullName: GitHub owner/repo identifier. | |
| state: Status of issues to retrieve ("open", "closed", "all"). | |
| Returns: | |
| A list of dictionary records containing issue numbers, titles, bodies, and URLs. | |
| """ | |
| if not self.token: | |
| raise RuntimeError("GitHub credentials are not configured.") | |
| url = f"https://api.github.com/repos/{repo_fullName}/issues" | |
| params = {"state": state, "per_page": 100} | |
| try: | |
| resp = self.session.get(url, params=params) | |
| resp.raise_for_status() | |
| issues = resp.json() | |
| result = [] | |
| for issue in issues: | |
| # GitHub issues endpoint also returns pull requests, filter them out | |
| if "pull_request" in issue: | |
| continue | |
| result.append( | |
| { | |
| "number": issue.get("number"), | |
| "title": issue.get("title"), | |
| "body": issue.get("body", ""), | |
| "url": issue.get("html_url"), | |
| "state": issue.get("state"), | |
| } | |
| ) | |
| return result | |
| except Exception as e: | |
| logger.error(f"Failed to fetch issues for {repo_fullName}: {e}") | |
| # Fallback to empty list or raise depending on preferences | |
| return [] | |
| def fetch_pull_request_metadata( | |
| self, owner: str, repo: str, pr_number: int | |
| ) -> Dict[str, Any]: | |
| """Fetch PR metadata from the GitHub API. | |
| Args: | |
| owner: Owner of the repository. | |
| repo: Name of the repository. | |
| pr_number: Pull request number. | |
| Returns: | |
| Dict containing title, state, html_url, additions, deletions, etc. | |
| """ | |
| if not self.token: | |
| raise RuntimeError("GitHub credentials are not configured.") | |
| url = f"https://api.github.com/repos/{owner}/{repo}/pulls/{pr_number}" | |
| try: | |
| resp = self.session.get(url) | |
| resp.raise_for_status() | |
| data = resp.json() | |
| return { | |
| "title": data.get("title", ""), | |
| "state": data.get("state", "open"), | |
| "html_url": data.get("html_url", ""), | |
| "additions": data.get("additions", 0), | |
| "deletions": data.get("deletions", 0), | |
| "head_sha": data.get("head", {}).get("sha", ""), | |
| } | |
| except Exception as e: | |
| logger.error( | |
| f"Failed to fetch PR metadata for {owner}/{repo}/pulls/{pr_number}: {e}" | |
| ) | |
| raise RuntimeError(f"Failed to fetch PR metadata: {e}") | |
| def fetch_pull_request_files( | |
| self, owner: str, repo: str, pr_number: int | |
| ) -> List[Dict[str, Any]]: | |
| """Fetch files changed in a PR from the GitHub API (handles pagination). | |
| Args: | |
| owner: Owner of the repository. | |
| repo: Name of the repository. | |
| pr_number: Pull request number. | |
| Returns: | |
| List of dict records for each changed file. | |
| """ | |
| if not self.token: | |
| raise RuntimeError("GitHub credentials are not configured.") | |
| url = f"https://api.github.com/repos/{owner}/{repo}/pulls/{pr_number}/files" | |
| result = [] | |
| page = 1 | |
| per_page = 100 | |
| while True: | |
| try: | |
| resp = self.session.get( | |
| url, params={"page": page, "per_page": per_page} | |
| ) | |
| resp.raise_for_status() | |
| files = resp.json() | |
| if not files: | |
| break | |
| for f in files: | |
| result.append( | |
| { | |
| "filename": f.get("filename", ""), | |
| "status": f.get("status", ""), | |
| "additions": f.get("additions", 0), | |
| "deletions": f.get("deletions", 0), | |
| "changes": f.get("changes", 0), | |
| } | |
| ) | |
| if len(files) < per_page: | |
| break | |
| page += 1 | |
| except Exception as e: | |
| logger.error( | |
| f"Failed to fetch PR files for {owner}/{repo}/pulls/{pr_number} page {page}: {e}" | |
| ) | |
| raise RuntimeError(f"Failed to fetch PR files: {e}") | |
| return result | |
| def get_rate_limit_info(self) -> Dict[str, Any]: | |
| """Fetch rate limit information from the GitHub API. | |
| Returns: | |
| Dict containing remaining rate limit, reset time, etc. | |
| """ | |
| if not self.token: | |
| raise RuntimeError("GitHub credentials are not configured.") | |
| url = "https://api.github.com/rate_limit" | |
| try: | |
| resp = self.session.get(url) | |
| resp.raise_for_status() | |
| data = resp.json() | |
| rate = data.get("resources", {}).get("core", {}) | |
| return { | |
| "limit": rate.get("limit", 0), | |
| "remaining": rate.get("remaining", 0), | |
| "reset": rate.get("reset", 0), | |
| } | |
| except Exception as e: | |
| logger.error(f"Failed to fetch rate limit info: {e}") | |
| return { | |
| "limit": 0, | |
| "remaining": 0, | |
| "reset": 0, | |
| } | |