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
from .constants import 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": 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)
                            }
                        },
                        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}{activity}: успешно {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 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)