from datetime import datetime import logging from colorama import Fore import requests from tqdm import tqdm from config import ClockifyConfig, config, db from classes import Activity, Tag, VetroProject from exceptions import InvalidToken from utils import parse_task_id, time_to_hour logger = logging.getLogger(__name__) class ClockifyAPI(ClockifyConfig): """API клиент для взаимодействия с Clockify. Позволяет получать активности, теги, проекты ветро и помечать активности затрканными. """ def __init__(self, token: str, *args, **kwargs): super().__init__(*args, **kwargs) self.token = self._check_token(token) def get_activities(self, is_updating=False, **kwargs) -> list[Activity]: """ Получает с сервера Clockify активности за определнный период start: datetime - Начальная дата (обязательно) end: datetime - Конечная дата (обязательно) page: int - Номер страницы page-size: int - Размер страницы (по умолчанию 50) """ time_entries_url = self.time_entries_url activities = requests.get(self.base_url + time_entries_url, params=kwargs, headers={"X-API-KEY": self.token}).json() logger.info(f"Получено задач: {len(activities)}") time_worked_out_total = 0 tags = self.get_tags(is_updating=is_updating) if not tags: logging.info(f"{Fore.YELLOW}В базе данных нет тегов! Обновите БД при помощи флага -i [--init-db]{Fore.RESET}") return [] result = [] for activity in tqdm(activities, desc="Парсинг задач"): if not activity.get("tagIds"): logger.info(f'{activity["timeInterval"]["start"]}: {activity["description"]} - У задачи нет тегов') continue for tag_id in activity.get("tagIds", {"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.utcnow()) - start) tag = [tag for tag in tags if tag.id == tag_id] if not tag: logging.info(f"{Fore.YELLOW}В базе данных нет нужного тега {Fore.GREEN}{tag_id}{Fore.YELLOW}! Обновите БД при помощи флага -i [--init-db]{Fore.RESET}") break else: tag = tag[0] try: task_id = parse_task_id(tag.title) except ValueError: logger.info(f'{activity["timeInterval"]["start"]}: {activity["description"]} - Неправильный формат тега {tag}') continue author = db.get_user_by("clockify_user_id", activity["userId"]) result.append( Activity( id=activity["id"], tag_id=tag_id, task_id=task_id, project_id=activity["projectId"], vetro_project_id=activity["taskId"], description=activity["description"], time_spent=time_to_hour((end if end else datetime.utcnow()) - start), date_start=start, date_end=end, is_tracked=activity["billable"], author_id=author["id"] ) ) logger.info(f"Сохранено задач: {len(result)}") logger.info(f"Отработано часов: {time_worked_out_total}") return result def get_tags(self, is_updating=False, **kwargs) -> list[Tag]: """ Получает с сервера Clockify теги name: str - optional archived: bool - optional page-size: int - optional """ if is_updating: tags = requests.get(self.base_url + self.tags_url, params={"sort-column": "name", "sort-order": "descending", **kwargs}, headers={"X-API-KEY": self.token}).json() logging.debug(f"{Fore.YELLOW}Получено тегов: {len(tags)}.{Fore.RESET}") return [Tag( id=tag["id"], title=tag["name"] ) for tag in tags] return db.get_tags() def get_vetro_projects(self) -> list[VetroProject]: """ Получает с сервера Clockify Проекты (ДО) Ветро """ projects = requests.get(self.base_url + self.projects_url, headers={"X-API-KEY": self.token}).json() for project in projects: if not project.get("archived"): vetro_projects = requests.get(self.base_url + self.tasks_url(project.get("id")), headers={"X-API-KEY": self.token}).json() return [VetroProject( id=vetro_project["id"], title=vetro_project["name"] ) for vetro_project in vetro_projects] def mark_tracked(self, activities: list[Activity]): """ Помечает активности затреканными (Billable) на сервере Clockify """ for activity in activities: try: resp = requests.put( self.base_url + self.update_time_entry( activity_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], "taskId": activity.vetro_project_id, "projectId": activity.project_id }, headers={ "X-API-KEY": self.token } ) if resp.ok: logger.debug(resp.json()) logger.info(f"{Fore.GREEN} {activity} помечена как затреканная{Fore.RESET}") else: logger.debug(resp.json()) logger.error(f"{Fore.RED} {activity} не помечена как затреканная{Fore.RESET}") except Exception as e: logger.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: logger.error(resp.json()) raise InvalidToken resp = resp.json() logger.debug({"id": resp["id"], "email": resp["email"], "name": resp["name"]}) return token def get_user(self): """Получает нужные данные пользователя с сервера Clockify""" resp = requests.get(self.base_url + self.user_url, headers={"X-API-KEY": self.token}) if not resp.ok: logger.error(resp.json()) raise requests.exceptions.RequestException(Fore.RED + "Ошибка запроса." + Fore.RESET) json = resp.json() return { "id": json['id'], "workspace_id": json['activeWorkspace'], "email": json["email"], "username": json["name"] } def getenvs(self): """Получает нужные USER_ID и WORKSPACE_ID с сервера Clockify""" user = self.get_user() with open(config.ENV_TXT, "w") as file: try: file.write(f"USER_ID={user['id']}\nWORKSPACE_ID={user['workspace_id']}") logger.info(f"{Fore.GREEN}Необходимые переменные были сохранены в файл {config.ENV_TXT}{Fore.RESET}") except Exception as e: logger.error(f"{Fore.RED}Не удалось сохранить необходимые переменные.{Fore.RESET}") raise e