MCP Atlassian

"""Module for Jira development information operations (PRs, commits, branches)."""

import logging
from typing import Any

from .client import JiraClient

logger = logging.getLogger("mcp-jira")


class DevelopmentMixin(JiraClient):
    """Mixin for Jira development information operations."""

    def get_issue_development_info(
        self,
        issue_key: str,
        application_type: str | None = None,
        data_type: str | None = None,
    ) -> dict[str, Any]:
        """
        Get development information (PRs, commits, branches) for a Jira issue.

        This uses the /rest/dev-status/1.0/issue/detail endpoint to retrieve
        development panel information linked to an issue.

        Args:
            issue_key: The issue key (e.g., PROJECT-123)
            application_type: Filter by application type
                (e.g., 'stash', 'github', 'bitbucket').
                If None, tries common application types.
            data_type: Filter by data type
                (e.g., 'pullrequest', 'branch', 'repository').
                If None, returns all data types.

        Returns:
            Dictionary containing development information with structure:
            {
                "detail": [...],  # List of development details by application
                "pullRequests": [...],  # Extracted list of all PRs
                "branches": [...],  # Extracted list of all branches
                "commits": [...],  # Extracted list of all commits
            }

        Raises:
            ValueError: If the issue is not found or issue ID cannot be retrieved
            Exception: If there is an error retrieving development info
        """
        try:
            # First, get the issue to obtain its numeric ID
            issue = self.jira.get_issue(issue_key, fields="id")
            if not isinstance(issue, dict):
                msg = f"Unexpected return value type from jira.get_issue: {type(issue)}"
                logger.error(msg)
                raise TypeError(msg)

            issue_id = issue.get("id")
            if not issue_id:
                msg = f"Could not get issue ID for {issue_key}"
                raise ValueError(msg)

            # If application_type is specified, use it directly
            if application_type:
                return self._fetch_dev_info_for_app_type(
                    issue_key, issue_id, application_type, data_type
                )

            # Otherwise, try common application types and merge results
            # Common types: stash (Bitbucket Server), bitbucket, github, gitlab
            app_types = ["stash", "bitbucket", "github", "gitlab"]
            # Data types to try for each app type
            data_types = ["pullrequest", "branch", "repository"]
            merged_result: dict[str, Any] = {
                "issue_key": issue_key,
                "detail": [],
                "pullRequests": [],
                "branches": [],
                "commits": [],
                "repositories": [],
            }

            for app_type in app_types:
                for dt in data_types:
                    try:
                        result = self._fetch_dev_info_for_app_type(
                            issue_key, issue_id, app_type, dt
                        )
                        if "error" in result:
                            # Plugin unavailable or access denied — capture the first
                            # error and stop; all subsequent calls will fail the same way
                            if not merged_result.get("error"):
                                merged_result["error"] = result["error"]
                            break
                        # Merge results
                        merged_result["detail"].extend(result.get("detail", []))
                        merged_result["pullRequests"].extend(
                            result.get("pullRequests", [])
                        )
                        merged_result["branches"].extend(result.get("branches", []))
                        merged_result["commits"].extend(result.get("commits", []))
                        for repo in result.get("repositories", []):
                            if repo not in merged_result["repositories"]:
                                merged_result["repositories"].append(repo)
                    except Exception as e:
                        # Log but continue trying other combinations
                        logger.debug(
                            f"No dev info for {issue_key} "
                            f"from {app_type}/{dt}: {str(e)}"
                        )
                if merged_result.get("error"):
                    break

            return merged_result

        except Exception as e:
            error_msg = str(e)
            logger.error(
                f"Error retrieving development info for {issue_key}: {error_msg}"
            )
            msg = f"Error retrieving development info for {issue_key}: {error_msg}"
            raise Exception(msg) from e

    def _fetch_dev_info_for_app_type(
        self,
        issue_key: str,
        issue_id: str,
        application_type: str,
        data_type: str | None = None,
    ) -> dict[str, Any]:
        """
        Fetch development info for a specific application type.

        Args:
            issue_key: The issue key
            issue_id: The numeric issue ID
            application_type: The application type (stash, github, etc.)
            data_type: Optional data type filter

        Returns:
            Parsed development information
        """
        # TODO: Consider caching results to reduce API calls (up to 12 per issue)
        params: dict[str, str] = {
            "issueId": str(issue_id),
            "applicationType": application_type,
        }
        if data_type:
            params["dataType"] = data_type

        # Use _session.get() directly: the dev-status endpoint is a plugin-specific
        # path (/rest/dev-status/1.0/...) not covered by the standard Jira client
        # wrappers. No higher-level wrapper method exists for this non-standard endpoint.
        url = f"{self.config.url}/rest/dev-status/1.0/issue/detail"
        http_response = self.jira._session.get(
            url, params=params, verify=self.config.ssl_verify
        )

        if http_response.status_code == 404:
            logger.debug(
                f"Dev-status plugin returned 404 for {issue_key}/{application_type}"
                f"/{data_type} — plugin may not be installed"
            )
            return {
                "issue_key": issue_key,
                "error": (
                    "Development info is not available — the Jira dev-status plugin"
                    " may not be installed on this instance."
                ),
                "detail": [],
                "pullRequests": [],
                "branches": [],
                "commits": [],
                "repositories": [],
            }

        if http_response.status_code == 403:
            logger.debug(
                f"Dev-status plugin returned 403 for {issue_key}/{application_type}"
                f"/{data_type} — permissions issue"
            )
            return {
                "issue_key": issue_key,
                "error": (
                    "Access denied to development info — check Jira permissions"
                    " or dev-status plugin configuration."
                ),
                "detail": [],
                "pullRequests": [],
                "branches": [],
                "commits": [],
                "repositories": [],
            }

        http_response.raise_for_status()
        response = http_response.json()

        if not isinstance(response, dict):
            msg = f"Unexpected response type from dev-status API: {type(response)}"
            logger.error(msg)
            raise TypeError(msg)

        # Extract and organize the development information
        return self._parse_development_info(response, issue_key)

    def _parse_development_info(
        self, response: dict[str, Any], issue_key: str
    ) -> dict[str, Any]:
        """
        Parse the development info response into a structured format.

        Args:
            response: Raw response from dev-status API
            issue_key: The issue key for reference

        Returns:
            Structured development information
        """
        result: dict[str, Any] = {
            "issue_key": issue_key,
            "detail": [],
            "pullRequests": [],
            "branches": [],
            "commits": [],
            "repositories": [],
        }

        # Get the detail array from response
        details = response.get("detail", [])
        if not details:
            return result

        for detail in details:
            if not isinstance(detail, dict):
                continue

            # Store the raw detail
            result["detail"].append(detail)

            # Get instance info for context
            instance = detail.get("_instance", {})
            instance_name = instance.get("name", "Unknown")
            instance_url = instance.get("baseUrl", "")

            # Extract PRs directly from detail (not nested under repositories)
            for pr in detail.get("pullRequests", []):
                if isinstance(pr, dict):
                    source = pr.get("source", {})
                    destination = pr.get("destination", {})
                    source_repo = source.get("repository", {})

                    result["pullRequests"].append(
                        {
                            "id": pr.get("id", ""),
                            "name": pr.get("name", ""),
                            "status": pr.get("status", ""),
                            "url": pr.get("url", ""),
                            "source": source.get("branch", ""),
                            "destination": destination.get("branch", ""),
                            "author": pr.get("author", {}).get("name", ""),
                            "reviewers": [
                                r.get("name", "")
                                for r in pr.get("reviewers", [])
                                if isinstance(r, dict)
                            ],
                            "lastUpdate": pr.get("lastUpdate", ""),
                            "repository": source_repo.get("name", ""),
                            "repositoryUrl": source_repo.get("url", ""),
                            "instance": instance_name,
                        }
                    )

            # Extract branches directly from detail
            for branch in detail.get("branches", []):
                if isinstance(branch, dict):
                    result["branches"].append(
                        {
                            "name": branch.get("name", ""),
                            "url": branch.get("url", ""),
                            "createPullRequestUrl": branch.get(
                                "createPullRequestUrl", ""
                            ),
                            "instance": instance_name,
                        }
                    )

            # Also check repositories array if present
            repositories = detail.get("repositories", [])
            for repo in repositories:
                if not isinstance(repo, dict):
                    continue

                repo_name = repo.get("name", "Unknown")
                repo_url = repo.get("url", "")
                avatar_url = repo.get("avatar", "")

                # Extract commits from repositories
                for commit in repo.get("commits", []):
                    if isinstance(commit, dict):
                        result["commits"].append(
                            {
                                "id": commit.get("id", ""),
                                "displayId": commit.get("displayId", ""),
                                "message": commit.get("message", ""),
                                "author": commit.get("author", {}).get("name", ""),
                                "authorTimestamp": commit.get("authorTimestamp", ""),
                                "url": commit.get("url", ""),
                                "repository": repo_name,
                                "repositoryUrl": repo_url,
                            }
                        )

                # Extract PRs from repositories (fallback)
                for pr in repo.get("pullRequests", []):
                    if isinstance(pr, dict):
                        result["pullRequests"].append(
                            {
                                "id": pr.get("id", ""),
                                "name": pr.get("name", ""),
                                "status": pr.get("status", ""),
                                "url": pr.get("url", ""),
                                "source": pr.get("source", {}).get("branch", ""),
                                "destination": pr.get("destination", {}).get(
                                    "branch", ""
                                ),
                                "author": pr.get("author", {}).get("name", ""),
                                "reviewers": [
                                    r.get("name", "")
                                    for r in pr.get("reviewers", [])
                                    if isinstance(r, dict)
                                ],
                                "lastUpdate": pr.get("lastUpdate", ""),
                                "repository": repo_name,
                                "repositoryUrl": repo_url,
                            }
                        )

                # Extract branches from repositories (fallback)
                for branch in repo.get("branches", []):
                    if isinstance(branch, dict):
                        result["branches"].append(
                            {
                                "name": branch.get("name", ""),
                                "url": branch.get("url", ""),
                                "createPullRequestUrl": branch.get(
                                    "createPullRequestUrl", ""
                                ),
                                "repository": repo_name,
                                "repositoryUrl": repo_url,
                            }
                        )

                # Track unique repositories
                if repo_name and repo_name != "Unknown":
                    repo_info = {
                        "name": repo_name,
                        "url": repo_url,
                        "avatar": avatar_url,
                    }
                    if repo_info not in result["repositories"]:
                        result["repositories"].append(repo_info)

        return result

    def get_issues_development_info(
        self,
        issue_keys: list[str],
        application_type: str | None = None,
        data_type: str | None = None,
    ) -> list[dict[str, Any]]:
        """
        Get development information for multiple Jira issues.

        Args:
            issue_keys: List of issue keys (e.g., ['PROJECT-123', 'PROJECT-456'])
            application_type: Filter by application type
            data_type: Filter by data type

        Returns:
            List of development information dictionaries, one per issue
        """
        results = []
        for issue_key in issue_keys:
            try:
                info = self.get_issue_development_info(
                    issue_key=issue_key,
                    application_type=application_type,
                    data_type=data_type,
                )
                results.append(info)
            except Exception as e:
                logger.warning(
                    f"Failed to get development info for {issue_key}: {str(e)}"
                )
                results.append(
                    {
                        "issue_key": issue_key,
                        "error": str(e),
                        "pullRequests": [],
                        "branches": [],
                        "commits": [],
                    }
                )
        return results