redmine_tracker/api.py

from datetime import datetime
import logging

from colorama import Fore
import requests
from tqdm import tqdm

from config import ClockifyConfig, RedmineConfig, config
from classes import Activity, Tag
from enums import ActivityType
from exceptions import InvalidToken
from utils import time_to_hour


WORKSPACE_ID = config.CLOCKIFY_WORKSPACE_ID
USER_ID = config.CLOCKIFY_USER_ID


class ClockifyAPI(ClockifyConfig):
    def __init__(self, token: str):
        self.token = self._check_token(token)

    def get_activities(self, **kwargs) -> list[Activity]:
        """
        start: datetime - Начальная дата (обязательно)
        end: datetime - Конечная дата (обязательно)
        page: int - Номер страницы
        page-size: int - Размер страницы (по умолчанию 50)
        """
        time_entries_url = self.time_entries_url.format(workspace_id=WORKSPACE_ID, user_id=USER_ID)
        activities = requests.get(self.base_url + time_entries_url, params=kwargs, headers={"X-API-KEY": self.token}).json()
        logging.info(f"Получено задач: {len(activities)}")
        time_worked_out_total = 0

        result = []
        for activity in tqdm(activities, desc="Парсинг задач"):
            for tag_id in activity["tagIds"]:
                start = datetime.strptime(activity["timeInterval"]["start"], config.DATETIME_FORMAT)
                end = datetime.strptime(activity["timeInterval"]["end"], config.DATETIME_FORMAT) if activity["timeInterval"]["end"] else None
                time_worked_out_total += time_to_hour((end if end else datetime.now()) - start)
                tag = [tag for tag in self.get_tags(workspace_id=WORKSPACE_ID) if tag.id == tag_id][0]
                task_id, task_title = tag.title.split(" - ")
                result.append(
                    Activity(
                        id=activity["id"],
                        tag_id=tag_id,
                        task_id=task_id,
                        task_title=task_title,
                        project_id=activity["projectId"],
                        description=activity["description"],
                        time_spent=time_to_hour((end if end else datetime.now()) - start),
                        date_start=start,
                        date_end=end,
                        is_tracked=activity["billable"]
                    )
                )
        logging.info(f"Сохранено задач: {len(result)}")
        logging.info(f"Отработано часов: {time_worked_out_total}")
        return result
    

    def get_tags(self, **kwargs) -> list[Tag]:
        """
        workspace_id: str - required
        name: str - optional
        archived: bool - optional
        """
        tags = requests.get(self.base_url + self.tags_url.format(**kwargs), params=kwargs, headers={"X-API-KEY": self.token}).json()
        return [Tag(
            id=tag["id"],
            title=tag["name"]
        ) for tag in tags]
    
    def mark_tracked(self, activities: list[Activity]):
        for activity in activities:
            try:
                resp = requests.put(
                    self.base_url + self.update_time_entry.format(
                        workspace_id=WORKSPACE_ID,
                        id=activity.id
                    ), json={
                        "billable": True,
                        "start": activity.date_start.strftime(config.DATETIME_FORMAT),
                        "end": activity.date_end.strftime(config.DATETIME_FORMAT) if activity.date_end else None,
                        "description": activity.description,
                        "tagIds": [activity.tag_id],
                        "projectId": activity.project_id
                    }, headers={
                        "X-API-KEY": self.token
                    }
                )
                if resp.ok:
                    logging.debug(resp.json())
                    logging.info(f"{activity} {Fore.GREEN} помечена как затреканная{Fore.RESET}")
                else:
                    logging.debug(resp.json())
                    logging.error(f"{activity} {Fore.RED} не помечена как затреканная{Fore.RESET}")
            except Exception as e:
                logging.error(e)
                return
        

    
    def _check_token(self, token: str) -> str:
        resp = requests.get(self.base_url + self.user_url, headers={"X-API-KEY": token})
        if not resp.ok:
            logging.error(resp.json())
            raise InvalidToken
        resp = resp.json()
        logging.debug({"id": resp["id"], "email": resp["email"], "name": resp["name"]})
        return token
    

class RedmineAPI(RedmineConfig):
    def __init__(self, token):
        self.token = self._check_token(token) 

    def track_activities(self, activities: list[Activity]) -> list[Activity]:
        if len(activities) == 0:
            logging.info("Нет задач для занесения трудочасов")
            return
        tracked = []
        for activity in tqdm(activities, desc="Занесение трудочасов"):
            if not activity.is_tracked:
                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,
                                "activity_id": ActivityType.DEVELOPMENT,
                                "comments": activity.description,
                                "spent_on": activity.date_start.strftime(self.datetime_format)
                            }
                        }
                    )
                    if resp.ok:
                        activity.is_tracked = True
                        tracked.append(activity)
                        logging.debug(activity)
                    else:
                        logging.error(f"{activity} Не была затрекана")
                        logging.error(f"{resp.json()['errors']} {resp.status_code}")
                except Exception as e:
                    logging.error(e)
                    continue
        if len(activities) == len(tracked):
            logging.info("Все задачи были успешно занесены в Redmine")
        else:
            logging.info(f"Не все задачи были занесены в Redmine. Занесено {len(tracked)}")
        return tracked


    def _check_token(self, token):
        resp = requests.get(self.base_url + self.my_account_url, headers={"X-Redmine-Api-Key": token})
        if not resp.ok:
            logging.error("Invalid Token for RedmineAPI")
            raise InvalidToken
        resp = resp.json()['user']
        logging.debug({"login": resp["login"], "email": resp["mail"], "full_name": f"{resp["firstname"]} {resp["lastname"]}"})
        return token