""" Клиент для работы с 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