""" Issue fetching from Gitea API. """ import json import subprocess from dataclasses import dataclass from datetime import datetime from typing import List, Optional, Dict, Any from .config import get_config from .exceptions import IssueError @dataclass class Issue: """Represents a Gitea issue.""" number: int title: str body: str state: str created_at: datetime updated_at: datetime html_url: str assignee: Optional[str] = None labels: List[str] = None def __post_init__(self): if self.labels is None: self.labels = [] class IssueFetcher: """Fetches issues from Gitea API.""" def __init__(self, config=None): self.config = config or get_config() def fetch_issue(self, issue_number: int) -> Issue: """Fetch a specific issue by number.""" try: result = subprocess.run( ['curl', '-s', f"{self.config.issues_api_url}/{issue_number}"], capture_output=True, text=True, check=True ) if result.returncode != 0: raise IssueError(f"Failed to fetch issue #{issue_number}: {result.stderr}") issue_data = json.loads(result.stdout) if 'message' in issue_data: raise IssueError(f"Issue #{issue_number} not found: {issue_data['message']}") return self._parse_issue(issue_data) except subprocess.CalledProcessError as e: raise IssueError(f"Failed to fetch issue #{issue_number}: {e}") except json.JSONDecodeError as e: raise IssueError(f"Failed to parse issue data: {e}") def fetch_issues(self, state: str = "all") -> List[Issue]: """Fetch all issues with optional state filter.""" try: url = self.config.issues_api_url if state != "all": url += f"?state={state}" result = subprocess.run( ['curl', '-s', url], capture_output=True, text=True, check=True ) if result.returncode != 0: raise IssueError(f"Failed to fetch issues: {result.stderr}") issues_data = json.loads(result.stdout) if isinstance(issues_data, dict) and 'message' in issues_data: raise IssueError(f"Failed to fetch issues: {issues_data['message']}") if not isinstance(issues_data, list): raise IssueError("Invalid response format: expected list of issues") return [self._parse_issue(issue_data) for issue_data in issues_data] except subprocess.CalledProcessError as e: raise IssueError(f"Failed to fetch issues: {e}") except json.JSONDecodeError as e: raise IssueError(f"Failed to parse issues data: {e}") def fetch_open_issues(self) -> List[Issue]: """Fetch only open issues.""" return self.fetch_issues(state="open") def _parse_issue(self, issue_data: Dict[str, Any]) -> Issue: """Parse issue data from API response.""" try: labels = [label['name'] for label in issue_data.get('labels', [])] assignee = None if issue_data.get('assignee'): assignee = issue_data['assignee'].get('login') return Issue( number=issue_data['number'], title=issue_data['title'], body=issue_data.get('body', ''), state=issue_data['state'], created_at=datetime.fromisoformat(issue_data['created_at'].replace('Z', '+00:00')), updated_at=datetime.fromisoformat(issue_data['updated_at'].replace('Z', '+00:00')), html_url=issue_data['html_url'], assignee=assignee, labels=labels ) except (KeyError, ValueError) as e: raise IssueError(f"Failed to parse issue data: {e}") def get_issue_data_dict(self, issue_number: int) -> Dict[str, Any]: """Get issue data as dictionary for workspace creation.""" issue = self.fetch_issue(issue_number) return { 'number': issue.number, 'title': issue.title, 'body': issue.body, 'state': issue.state, 'created_at': issue.created_at.isoformat(), 'updated_at': issue.updated_at.isoformat(), 'html_url': issue.html_url, 'assignee': {'login': issue.assignee} if issue.assignee else None, 'labels': [{'name': label} for label in issue.labels] }