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 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], 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": 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 config.is_debug or resp.ok:
                    activity.is_tracked = True
                    if on_success:
                        on_success(activity)
                    tracked.append(activity)
                    logger.info(f"{Fore.GREEN}{activity} успешно занесена в Redmine{Fore.RESET}".ljust(60) + "✅")
                else:
                    try:
                        errors = resp.json()["errors"]
                    except Exception:
                        errors = resp.text
                    logger.debug(errors)
                    logger.error(f"{Fore.RED}{activity} не была занесена в Redmine{Fore.RESET}".ljust(60) + "❌")
                    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 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})
        logger.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) -> Task:
        """Получает задачу по её id с сервера Redmine"""
        resp = requests.get(
            self.base_url + self.issue.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 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 = []
        for i in range(25):
            resp = requests.get(
                self.base_url + self.issues,
                headers={"X-Redmine-Api-Key": self.token},
                params={"limit": 100, "offset": 100 * i, "status_id": "*"},
                timeout=None
            )
            if not resp.ok or not resp.json():
                logger.error("Ошибка RedmineAPI")
                return {}
            issues.extend(resp.json()["issues"])
        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)