markitect-main/tddai/issue_writer.py

"""
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."""
        return self.update_issue(issue_number, {'labels': labels})

    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}")