MessageGateway/app/core/jira_client.py

"""
Клиент для работы с Jira API.

Автор: Сергей Антропов
Сайт: https://devops.org.ru
"""
import logging
from typing import Optional, Dict, Any
from jira import JIRA
from jira.exceptions import JIRAError

logger = logging.getLogger(__name__)


class JiraClient:
    """Клиент для работы с Jira API."""
    
    def __init__(
        self,
        url: str,
        email: str,
        api_token: str
    ):
        """
        Инициализация клиента Jira.
        
        Args:
            url: URL Jira сервера.
            email: Email пользователя Jira.
            api_token: API токен Jira.
        """
        self.url = url.rstrip('/')
        self.email = email
        self.api_token = api_token
        self._client: Optional[JIRA] = None
    
    def get_client(self) -> JIRA:
        """
        Получить экземпляр клиента Jira (создается при первом обращении).
        
        Returns:
            Экземпляр JIRA клиента.
        """
        if self._client is None:
            try:
                self._client = JIRA(
                    server=self.url,
                    basic_auth=(self.email, self.api_token)
                )
                logger.info(f"Jira клиент подключен к {self.url}")
            except JIRAError as e:
                logger.error(f"Ошибка подключения к Jira: {e}")
                raise
        return self._client
    
    def create_issue(
        self,
        project: str,
        summary: str,
        description: str,
        issue_type: str = "Bug",
        assignee: Optional[str] = None,
        priority: Optional[str] = None,
        labels: Optional[list] = None,
        components: Optional[list] = None
    ) -> Optional[str]:
        """
        Создать тикет в Jira.
        
        Args:
            project: Ключ проекта Jira.
            summary: Заголовок тикета.
            description: Описание тикета.
            issue_type: Тип задачи.
            assignee: Email исполнителя (опционально).
            priority: Приоритет задачи (опционально).
            labels: Список меток (опционально).
            components: Список компонентов (опционально).
        
        Returns:
            Ключ созданного тикета (например, "MON-123") или None в случае ошибки.
        """
        try:
            client = self.get_client()
            
            # Формируем словарь для создания тикета
            issue_dict = {
                'project': {'key': project},
                'summary': summary,
                'description': description,
                'issuetype': {'name': issue_type}
            }
            
            # Добавляем приоритет, если указан
            if priority:
                issue_dict['priority'] = {'name': priority}
            
            # Добавляем метки, если указаны
            if labels:
                issue_dict['labels'] = labels
            
            # Добавляем компоненты, если указаны
            if components:
                issue_dict['components'] = [{'name': comp} for comp in components]
            
            # Создаем тикет
            issue = client.create_issue(fields=issue_dict)
            
            # Назначаем исполнителя, если указан
            if assignee:
                try:
                    # Пытаемся найти пользователя по email или username
                    users = client.search_users(query=assignee)
                    if users:
                        # Назначаем первого найденного пользователя
                        user_account_id = users[0].accountId
                        client.assign_issue(issue, user_account_id)
                        logger.info(f"Исполнитель {assignee} (accountId: {user_account_id}) назначен на тикет {issue.key}")
                    else:
                        logger.warning(f"Пользователь {assignee} не найден в Jira, тикет создан без исполнителя")
                except JIRAError as e:
                    logger.error(f"Ошибка назначения исполнителя {assignee}: {e}")
                except Exception as e:
                    logger.error(f"Неожиданная ошибка при назначении исполнителя: {e}")
            
            logger.info(f"Тикет {issue.key} создан в Jira")
            return issue.key
            
        except JIRAError as e:
            logger.error(f"Ошибка создания тикета в Jira: {e}")
            return None
        except Exception as e:
            logger.error(f"Неожиданная ошибка при создании тикета: {e}")
            return None
    
    def get_issue_url(self, issue_key: str) -> str:
        """
        Получить URL тикета в Jira.
        
        Args:
            issue_key: Ключ тикета (например, "MON-123").
        
        Returns:
            URL тикета в Jira.
        """
        return f"{self.url}/browse/{issue_key}"
    
    def update_issue(
        self,
        issue_key: str,
        summary: Optional[str] = None,
        description: Optional[str] = None,
        assignee: Optional[str] = None,
        priority: Optional[str] = None,
        labels: Optional[list] = None
    ) -> bool:
        """
        Обновить тикет в Jira.
        
        Args:
            issue_key: Ключ тикета.
            summary: Новый заголовок (опционально).
            description: Новое описание (опционально).
            assignee: Новый исполнитель (опционально).
            priority: Новый приоритет (опционально).
            labels: Новые метки (опционально).
        
        Returns:
            True если тикет обновлен успешно, False в противном случае.
        """
        try:
            client = self.get_client()
            issue = client.issue(issue_key)
            
            update_dict = {}
            
            if summary:
                update_dict['summary'] = [{'set': summary}]
            if description:
                update_dict['description'] = [{'set': description}]
            if priority:
                update_dict['priority'] = [{'set': {'name': priority}}]
            if labels:
                update_dict['labels'] = [{'set': labels}]
            
            if update_dict:
                issue.update(fields=update_dict)
            
            if assignee:
                try:
                    users = client.search_users(query=assignee)
                    if users:
                        user_account_id = users[0].accountId
                        client.assign_issue(issue, user_account_id)
                        logger.info(f"Исполнитель {assignee} (accountId: {user_account_id}) назначен на тикет {issue_key}")
                    else:
                        logger.warning(f"Пользователь {assignee} не найден в Jira")
                except JIRAError as e:
                    logger.error(f"Ошибка назначения исполнителя {assignee}: {e}")
                except Exception as e:
                    logger.error(f"Неожиданная ошибка при назначении исполнителя: {e}")
            
            logger.info(f"Тикет {issue_key} обновлен в Jira")
            return True
            
        except JIRAError as e:
            logger.error(f"Ошибка обновления тикета {issue_key}: {e}")
            return False
        except Exception as e:
            logger.error(f"Неожиданная ошибка при обновлении тикета: {e}")
            return False
    
    def add_comment(self, issue_key: str, comment: str) -> bool:
        """
        Добавить комментарий к тикету.
        
        Args:
            issue_key: Ключ тикета.
            comment: Текст комментария.
        
        Returns:
            True если комментарий добавлен успешно, False в противном случае.
        """
        try:
            client = self.get_client()
            issue = client.issue(issue_key)
            issue.add_comment(comment)
            logger.info(f"Комментарий добавлен к тикету {issue_key}")
            return True
        except JIRAError as e:
            logger.error(f"Ошибка добавления комментария к тикету {issue_key}: {e}")
            return False
        except Exception as e:
            logger.error(f"Неожиданная ошибка при добавлении комментария: {e}")
            return False