"""GitHub API client for fetching issues and comments.""" import json import subprocess from datetime import datetime from typing import Dict, List, Optional, Any from dataclasses import dataclass @dataclass class Comment: """Represents a GitHub issue comment.""" author: str body: str created_at: datetime url: str @dataclass class Issue: """Represents a GitHub issue with its comments.""" number: int title: str body: str author: str state: str created_at: datetime updated_at: datetime url: str comments: List[Comment] class GitHubClient: """Client for interacting with GitHub API using GitHub CLI.""" def __init__(self): """Initialize the GitHub client.""" self._verify_gh_cli() def _verify_gh_cli(self) -> None: """Verify that GitHub CLI is installed and authenticated.""" try: result = subprocess.run( ["gh", "auth", "status"], capture_output=True, text=True, check=True ) except subprocess.CalledProcessError: raise RuntimeError( "GitHub CLI is not authenticated. Please run 'gh auth login' first." ) except FileNotFoundError: raise RuntimeError( "GitHub CLI is not installed. Please install it first." ) def _run_gh_command(self, command: List[str]) -> str: """Run a GitHub CLI command and return the output.""" try: result = subprocess.run( command, capture_output=True, text=True, check=True ) return result.stdout except subprocess.CalledProcessError as e: raise RuntimeError(f"GitHub CLI command failed: {e.stderr}") def get_issues( self, repo: str, before: Optional[datetime] = None, after: Optional[datetime] = None, limit: Optional[int] = None, state: str = "all" ) -> List[Issue]: """ Fetch issues from a GitHub repository. Args: repo: Repository in format "owner/repo" before: Only issues created before this date after: Only issues created after this date limit: Maximum number of issues to fetch state: Issue state (open, closed, all) Returns: List of Issue objects """ command = [ "gh", "issue", "list", "--repo", repo, "--state", state, "--json", "number,title,body,author,state,createdAt,updatedAt,url" ] if limit: command.extend(["--limit", str(limit)]) output = self._run_gh_command(command) issues_data = json.loads(output) issues = [] for issue_data in issues_data: created_at = datetime.fromisoformat( issue_data["createdAt"].replace("Z", "+00:00") ) updated_at = datetime.fromisoformat( issue_data["updatedAt"].replace("Z", "+00:00") ) # Apply date filters if before and created_at >= before: continue if after and created_at <= after: continue # Fetch comments for this issue comments = self._get_issue_comments(repo, issue_data["number"]) issue = Issue( number=issue_data["number"], title=issue_data["title"], body=issue_data["body"] or "", author=issue_data["author"]["login"], state=issue_data["state"], created_at=created_at, updated_at=updated_at, url=issue_data["url"], comments=comments ) issues.append(issue) return issues def _get_issue_comments(self, repo: str, issue_number: int) -> List[Comment]: """ Fetch comments for a specific issue. Args: repo: Repository in format "owner/repo" issue_number: Issue number Returns: List of Comment objects """ command = [ "gh", "issue", "view", str(issue_number), "--repo", repo, "--json", "comments" ] output = self._run_gh_command(command) issue_data = json.loads(output) comments = [] for comment_data in issue_data.get("comments", []): created_at = datetime.fromisoformat( comment_data["createdAt"].replace("Z", "+00:00") ) comment = Comment( author=comment_data["author"]["login"], body=comment_data["body"], created_at=created_at, url=comment_data["url"] ) comments.append(comment) return comments