redmine_tracker/api/redmine.py

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,
    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, resp.json()["time_entry"]["id"])
                    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"""
        try:
            resp = requests.get(
                self.base_url + self.issue.format(**{"id": issue_id}),
                headers={"X-Redmine-Api-Key": self.token},
                timeout=REQUEST_TIMEOUT,
            )
        except requests.RequestException as error:
            logger.error(
                "Не удалось загрузить задачу #%s из Redmine: %s",
                issue_id,
                error,
            )
            return None
        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, issue_ids: set[int]) -> list[Task]:
        """Получает с сервера Redmine только задачи с указанными номерами."""
        return [
            task
            for issue_id in sorted(issue_ids)
            if (task := self.get_issue(issue_id))
        ]

    def _check_token(self):
        user = self.get_user()
        logger.debug(user)