from datetime import datetime, timezone import logging 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 """ { "time_entry": { "issue_id": 28882, "hours": 0.55, "activity_id": 9, "comments": "Дейлик" } } """ 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"] or str(datetime.now(timezone.utc).strftime(config.DATETIME_FORMAT)), config.DATETIME_FORMAT) time_worked_out_total += time_to_hour(end - 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( task_id=task_id, task_title=task_title, description=activity["description"], time_spent=time_to_hour(end - start), date=end ) ) 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 _check_token(self, token: str) -> str: resp = requests.get(self.base_url + self.config.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]): if len(activities) == 0: logging.info("Нет задач для занесения трудочасов") return tracked = 0 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}, data={ "time_entry": { "issue_id": activity.task_id, "hours": activity.time_spent, "activity_id": ActivityType.DEVELOPMENT, "comments": activity.description, "spent_on": activity.date.strftime(self.datetime_format) } } ) if resp.ok: activity.is_tracked = True tracked += 1 logging.debug(activity) else: logging.error(activity, "Не была затрекана") logging.error(resp.json(), resp) except Exception as e: logging.error(e) continue if len(activities) == tracked: logging.info(f"Все задачи были успешно занесены в Redmine: {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