128 lines
5.7 KiB
Python
128 lines
5.7 KiB
Python
from datetime import datetime
|
||
import logging
|
||
|
||
from colorama import Fore
|
||
import requests
|
||
from tqdm import tqdm
|
||
|
||
from config import ClockifyConfig, config
|
||
from classes import Activity, Tag, VetroProject
|
||
from exceptions import InvalidToken
|
||
from utils import time_to_hour
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class ClockifyAPI(ClockifyConfig):
|
||
def __init__(self, token: str, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
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
|
||
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
|
||
|
||
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 self.get_tags() if tag.id == tag_id][0]
|
||
try:
|
||
task_id, task_title = tag.title.split(" - ")
|
||
except ValueError:
|
||
logger.info(f'{activity["timeInterval"]["start"]}: {activity["description"]} - Неправильный формат тега {tag}')
|
||
continue
|
||
result.append(
|
||
Activity(
|
||
id=activity["id"],
|
||
tag_id=tag_id,
|
||
task_id=task_id,
|
||
task_title=task_title,
|
||
project_id=activity["projectId"],
|
||
vetro_project_id=activity["taskId"],
|
||
vetro_project_title=config.VETRO_PROJECTS.get(activity.get("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"]
|
||
)
|
||
)
|
||
logger.info(f"Сохранено задач: {len(result)}")
|
||
logger.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, params=kwargs, headers={"X-API-KEY": self.token}).json()
|
||
return [Tag(
|
||
id=tag["id"],
|
||
title=tag["name"]
|
||
) for tag in tags]
|
||
|
||
def get_vetro_projects(self) -> list[VetroProject]:
|
||
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]):
|
||
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"{activity} {Fore.GREEN} помечена как затреканная{Fore.RESET}")
|
||
else:
|
||
logger.debug(resp.json())
|
||
logger.error(f"{activity} {Fore.RED} не помечена как затреканная{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
|