import logging from colorama import Fore import requests from classes.classes import Task from config import RedmineConfig, config from classes import Activity, ActivityType from exceptions import InvalidToken from utils import format_hours from .constants import ( ACTIVITY_TYPE_BY_VETRO_PROJECT, DEFAULT_ACTIVITY_TYPE, REDMINE_ISSUES_LIMIT, REQUEST_TIMEOUT, ) logger = logging.getLogger(__name__) class RedmineAPI(RedmineConfig): """API клиент для взаимодействия с Redmine. Позволяет заносить трудочасы и получать виды деятельности. """ def __init__(self, token): super().__init__() self.token = token self._check_token() def track_activities(self, activities: list[Activity], on_success=None) -> list[Activity]: """Заносит трудочасы на сервер Redmine""" is_beginner = config.is_beginner if is_beginner: logger.info(f"{Fore.YELLOW}Делим трудочасы пополам{Fore.RESET}") tracked = [] failed_activities = [] activities_to_track = [ activity for activity in activities if not activity.is_tracked and activity.date_end ] for activity in activities_to_track: try: if not config.is_debug: 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": round(activity.time_spent / (2 if is_beginner else 1), 3), "activity_id": self.get_activity_type_id(activity), "comments": activity.description, "spent_on": activity.date_start.strftime(self.datetime_format) } }, timeout=REQUEST_TIMEOUT, ) if config.is_debug or resp.ok: activity.is_tracked = True if on_success: on_success(activity) tracked.append(activity) logger.info( f"{Fore.GREEN}{self.format_activity_result(activity, is_beginner)}" f"{Fore.RESET}".ljust(100) + "✅" ) else: try: errors = resp.json()["errors"] except Exception: errors = resp.text logger.debug(errors) logger.error(f"{Fore.RED}{activity}: с ошибкой {Fore.RESET}".ljust(100) + "❌") logger.error(f"{errors} {resp.status_code}") failed_activities.append(activity) except Exception as e: logger.error(e) failed_activities.append(activity) continue if len(activities_to_track) == len(tracked): logger.info(Fore.GREEN + "Все задачи были успешно занесены в Redmine" + Fore.RESET) else: logger.info(f"{Fore.YELLOW}Не все задачи были занесены в Redmine. Занесено {len(tracked)}{Fore.RESET}") logger.info(f"{Fore.YELLOW}Не занесены задачи:{Fore.RESET}") for activity in failed_activities: logger.info(f"{Fore.RED}{activity}{Fore.RESET}".ljust(60) + "❌") return tracked def format_activity_result(self, activity: Activity, is_beginner: bool = False) -> str: """Форматирует строку результата переноса активности.""" hours = round(activity.time_spent / (2 if is_beginner else 1), 3) return f"#{activity.task_id} | {format_hours(hours)} | {activity.description}" def get_activity_type_id(self, activity: Activity) -> int: """Возвращает тип деятельности Redmine для активности Clockify.""" return ACTIVITY_TYPE_BY_VETRO_PROJECT.get( activity.vetro_project_id, DEFAULT_ACTIVITY_TYPE, ) def get_time_entry_activities(self) -> list[ActivityType]: """Получает виды Деятельности с сервера Redmine""" resp = requests.get( self.base_url + self.time_entry_activities, headers={"X-Redmine-API-KEY": self.token}, timeout=REQUEST_TIMEOUT, ) if not resp.ok: logger.error(f"Ошибка RedmineAPI: {resp.text} {resp.status_code}") return [] data = resp.json() logger.debug(data) return [ ActivityType(id=act.get("id"), title=act.get("name")) for act in data.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}, timeout=REQUEST_TIMEOUT, ) if not resp.ok: logger.error(f"Invalid Token for RedmineAPI: {resp.text}") 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) -> Task | None: """Получает задачу по её id с сервера Redmine""" resp = requests.get( self.base_url + self.issue.format(**{"id": issue_id}), headers={"X-Redmine-Api-Key": self.token}, timeout=REQUEST_TIMEOUT, ) if not resp.ok: logger.error(f"Ошибка RedmineAPI: {resp.text} {resp.status_code}") return None issue = resp.json()["issue"] return Task(**{ "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 get_issues(self) -> list[Task]: """Получает задачи с сервера Redmine""" issues = [] offset = 0 total_count = None while total_count is None or offset < total_count: resp = requests.get( self.base_url + self.issues, headers={"X-Redmine-Api-Key": self.token}, params={"limit": REDMINE_ISSUES_LIMIT, "offset": offset, "status_id": "*"}, timeout=REQUEST_TIMEOUT, ) if not resp.ok: logger.error(f"Ошибка RedmineAPI: {resp.text} {resp.status_code}") return [] data = resp.json() if not data: logger.error("Ошибка RedmineAPI: пустой ответ") return [] issues.extend(data["issues"]) total_count = data.get("total_count", len(issues)) offset += data.get("limit", REDMINE_ISSUES_LIMIT) return [Task(**{ "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"] }) for issue in issues] def _check_token(self): user = self.get_user() logger.debug(user)