227 lines
9.9 KiB
Python
227 lines
9.9 KiB
Python
from datetime import datetime
|
||
import logging
|
||
from typing import Union
|
||
|
||
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()
|
||
if not activities:
|
||
return []
|
||
logger.info(f"Получено задач: {len(activities)}")
|
||
time_worked_out_total = 0
|
||
|
||
tags = self.get_tags(is_updating=is_updating)
|
||
if not tags:
|
||
logger.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:
|
||
tag = self.get_tags(True, tag_id=tag_id, **{"page-size": self.tags_count})
|
||
if isinstance(tag, Tag):
|
||
db.insert_tags([tag])
|
||
tags.append(tag)
|
||
from config import red
|
||
try:
|
||
task_id = parse_task_id(tag.title)
|
||
except Exception as e:
|
||
logger.error(f"При обновлении тега произошла ошибка: {e}\nПопробуйте обновить флагом -i [--init-db]")
|
||
continue
|
||
task = red.get_issue(task_id)
|
||
if task:
|
||
db.insert_tasks([task])
|
||
else:
|
||
logger.error(f"{Fore.RED} Тег не найден: {tag_id}{Fore.RESET}")
|
||
continue
|
||
else:
|
||
tag = tag[0]
|
||
task_id = parse_task_id(tag.title)
|
||
|
||
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, tag_id="", **kwargs) -> Union[list[Tag], Tag, None]:
|
||
"""
|
||
Получает с сервера Clockify теги
|
||
name: str - optional
|
||
archived: bool - optional
|
||
page-size: int - optional
|
||
page: int - optional
|
||
"""
|
||
if is_updating:
|
||
page = 1
|
||
page_size = int(kwargs.get("page-size", self.tags_count))
|
||
result = []
|
||
while True:
|
||
params = {
|
||
"sort-column": "name",
|
||
"sort-order": "descending",
|
||
**kwargs,
|
||
"page": page
|
||
}
|
||
tags = requests.get(
|
||
self.base_url + self.tags_url,
|
||
params=params,
|
||
headers={"X-API-KEY": self.token}
|
||
).json()
|
||
|
||
for tag in tags:
|
||
clockify_tag = Tag(
|
||
id=tag["id"],
|
||
title=tag["name"]
|
||
)
|
||
if tag_id and tag_id == clockify_tag.id:
|
||
return clockify_tag
|
||
if not tag_id:
|
||
result.append(clockify_tag)
|
||
|
||
if len(tags) < page_size:
|
||
break
|
||
page += 1
|
||
|
||
if tag_id:
|
||
return None
|
||
|
||
logger.debug(f"{Fore.YELLOW}Получено тегов: {len(result)}.{Fore.RESET}")
|
||
return result
|
||
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:
|
||
if not config.is_debug:
|
||
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 config.is_debug or resp.ok:
|
||
logger.info(f"{Fore.GREEN}{activity}{Fore.RESET}".ljust(60) + "✅")
|
||
else:
|
||
logger.debug(resp.json())
|
||
logger.error(f"{Fore.RED}{activity}{Fore.RESET}".ljust(60) + "❌")
|
||
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
|