""" Issue writing to Gitea API. """ import json import os import subprocess from subprocess import PIPE from typing import Dict, Any, Optional from .config import get_config from .exceptions import IssueError class IssueWriter: """Writes issue updates to Gitea API.""" def __init__(self, config=None, auth_token=None): self.config = config or get_config() self.auth_token = auth_token or os.getenv('GITEA_API_TOKEN') def update_issue(self, issue_number: int, update_data: Dict[str, Any]) -> Dict[str, Any]: """Update an issue via PATCH operation.""" if not self.auth_token: raise IssueError("Authentication token required for issue updates") url = f"{self.config.issues_api_url}/{issue_number}" try: # Prepare curl command with authentication curl_cmd = [ 'curl', '-s', '-X', 'PATCH', '-H', 'Content-Type: application/json', '-H', f'Authorization: token {self.auth_token}', '-d', json.dumps(update_data), url ] result = subprocess.run( curl_cmd, stdout=PIPE, stderr=PIPE, universal_newlines=True, check=True ) if result.returncode != 0: raise IssueError(f"Failed to update issue #{issue_number}: {result.stderr}") response_data = json.loads(result.stdout) if 'message' in response_data and 'number' not in response_data: raise IssueError(f"Failed to update issue #{issue_number}: {response_data['message']}") return response_data except subprocess.CalledProcessError as e: raise IssueError(f"Failed to update issue #{issue_number}: {e}") except json.JSONDecodeError as e: raise IssueError(f"Failed to parse response data: {e}") def update_issue_title(self, issue_number: int, new_title: str) -> Dict[str, Any]: """Update only the title of an issue.""" return self.update_issue(issue_number, {'title': new_title}) def update_issue_body(self, issue_number: int, new_body: str) -> Dict[str, Any]: """Update only the body of an issue.""" return self.update_issue(issue_number, {'body': new_body}) def update_issue_state(self, issue_number: int, new_state: str) -> Dict[str, Any]: """Update only the state of an issue (open/closed).""" if new_state not in ['open', 'closed']: raise IssueError(f"Invalid state '{new_state}'. Must be 'open' or 'closed'") return self.update_issue(issue_number, {'state': new_state}) def close_issue(self, issue_number: int) -> Dict[str, Any]: """Close an issue.""" return self.update_issue_state(issue_number, 'closed') def reopen_issue(self, issue_number: int) -> Dict[str, Any]: """Reopen a closed issue.""" return self.update_issue_state(issue_number, 'open') def assign_to_milestone(self, issue_number: int, milestone_id: int) -> Dict[str, Any]: """Assign issue to a milestone (project).""" return self.update_issue(issue_number, {'milestone': milestone_id}) def remove_from_milestone(self, issue_number: int) -> Dict[str, Any]: """Remove issue from its current milestone.""" return self.update_issue(issue_number, {'milestone': None}) def update_labels(self, issue_number: int, labels: list) -> Dict[str, Any]: """Update issue labels completely using dedicated labels endpoint.""" if not self.auth_token: raise IssueError("Authentication token required for label updates") # Use the dedicated labels endpoint which works more reliably url = f"{self.config.issues_api_url}/{issue_number}/labels" try: # Use PUT to replace all labels curl_cmd = [ 'curl', '-s', '-X', 'PUT', '-H', 'Content-Type: application/json', '-H', f'Authorization: token {self.auth_token}', '-d', json.dumps({'labels': labels}), url ] result = subprocess.run( curl_cmd, stdout=PIPE, stderr=PIPE, universal_newlines=True, check=True ) if result.returncode != 0: raise IssueError(f"Failed to update labels for issue #{issue_number}: {result.stderr}") # Parse the response - labels endpoint returns array of labels if result.stdout.strip(): response_data = json.loads(result.stdout) # Convert labels response back to issue format for consistency return { 'number': issue_number, 'labels': response_data if isinstance(response_data, list) else [] } else: return {'number': issue_number, 'labels': []} except subprocess.CalledProcessError as e: raise IssueError(f"Failed to update labels for issue #{issue_number}: {e}") except json.JSONDecodeError as e: raise IssueError(f"Failed to parse labels response: {e}") def add_labels(self, issue_number: int, new_labels: list) -> Dict[str, Any]: """Add labels to issue (preserving existing labels).""" # First get current labels url = f"{self.config.issues_api_url}/{issue_number}" curl_cmd = [ 'curl', '-s', '-X', 'GET', '-H', f'Authorization: token {self.auth_token}', url ] try: result = subprocess.run(curl_cmd, stdout=PIPE, stderr=PIPE, universal_newlines=True, check=True) issue_data = json.loads(result.stdout) current_labels = [label['name'] for label in issue_data.get('labels', [])] # Add new labels (avoid duplicates) updated_labels = list(set(current_labels + new_labels)) return self.update_labels(issue_number, updated_labels) except (subprocess.CalledProcessError, json.JSONDecodeError) as e: raise IssueError(f"Failed to add labels to issue #{issue_number}: {e}") def remove_labels(self, issue_number: int, labels_to_remove: list) -> Dict[str, Any]: """Remove specific labels from issue.""" # First get current labels url = f"{self.config.issues_api_url}/{issue_number}" curl_cmd = [ 'curl', '-s', '-X', 'GET', '-H', f'Authorization: token {self.auth_token}', url ] try: result = subprocess.run(curl_cmd, stdout=PIPE, stderr=PIPE, universal_newlines=True, check=True) issue_data = json.loads(result.stdout) current_labels = [label['name'] for label in issue_data.get('labels', [])] # Remove specified labels updated_labels = [label for label in current_labels if label not in labels_to_remove] return self.update_labels(issue_number, updated_labels) except (subprocess.CalledProcessError, json.JSONDecodeError) as e: raise IssueError(f"Failed to remove labels from issue #{issue_number}: {e}")