import logging from colorama import Fore import requests from tqdm import tqdm from config import RedmineConfig, config from classes import Activity, ActivityType from enums import ActivityTypes from exceptions import InvalidToken logger = logging.getLogger(__name__) WORKSPACE_ID = config.WORKSPACE_ID USER_ID = config.USER_ID class RedmineAPI(RedmineConfig): """API клиент для взаимодействия с Redmine. Позволяет заносить трудочасы и получать виды деятельности. """ def __init__(self, token): super().__init__() self.token = token self._check_token() def track_activities(self, activities: list[Activity]) -> list[Activity]: """Заносит трудочасы на сервер Redmine""" is_beginner = config.is_beginner if is_beginner: logging.info(f"{Fore.YELLOW}Делим трудочасы пополам{Fore.RESET}") tracked = [] for activity in tqdm(activities, desc="Занесение трудочасов"): if not(not activity.is_tracked and activity.date_end): continue try: resp = requests.post( self.base_url + self.time_entries_url, headers={"X-Redmine-Api-Key": self.token}, json={ "time_entry": { "issue_id": activity.task_id, "hours": activity.time_spent / (2 if is_beginner else 1), "activity_id": ActivityTypes.DEVELOPMENT, "comments": activity.description, "spent_on": activity.date_start.strftime(self.datetime_format) } } ) if resp.ok: activity.is_tracked = True tracked.append(activity) logger.debug(activity) else: logger.error(Fore.RED + f"{activity} Не была затрекана" + Fore.RED) logger.error(f"{resp.json()['errors']} {resp.status_code}") except Exception as e: logger.error(e) continue if len(activities) == len(tracked): logger.info(Fore.GREEN + "Все задачи были успешно занесены в Redmine" + Fore.RESET) else: logger.info(f"{Fore.YELLOW}Не все задачи были занесены в Redmine. Занесено {len(tracked)}{Fore.RESET}") return tracked def get_time_entry_activities(self) -> list[ActivityTypes]: """Получает виды Деятельности с сервера Redmine""" resp = requests.get(self.base_url + self.time_entry_activities, headers={"X-Redmine-API-KEY": self.token}) logging.debug(resp.json()) if resp.ok: return [ ActivityType(id=act.get("id"), title=act.get("name")) for act in resp.json().get("time_entry_activities") if act.get("active") ] def get_user(self): """Проверяет токен и возвращает данные юзера""" resp = requests.get(self.base_url + self.my_account_url, headers={"X-Redmine-Api-Key": self.token}) if not resp.ok: logger.error("Invalid Token for RedmineAPI") raise InvalidToken json = resp.json()["user"] return { "id": json["id"], "first_name": json["firstname"], "last_name": json["lastname"], "email": json["mail"], "username": json["login"] } def get_issue(self, issue_id: int) -> dict: """Получает задачу по её id с сервера Redmine""" resp = requests.get( self.base_url + self.issues.format(**{"id": issue_id}), headers={"X-Redmine-Api-Key": self.token}, timeout=None ) if not resp.ok: logger.error("Ошибка RedmineAPI") return {} issue = resp.json()["issue"] return { "id": issue["id"], "subject": issue["subject"], "description": issue["description"], "status": issue["status"]["name"], "priority": issue["priority"]["name"], "author": issue["author"]["name"], "tracker": issue["tracker"]["name"], "project": issue["project"]["name"] } def _check_token(self): user = self.get_user() logging.debug(user)