123
This commit is contained in:
125
api.py
125
api.py
@@ -1,49 +1,136 @@
|
||||
from datetime import datetime, timezone
|
||||
import logging
|
||||
|
||||
import requests
|
||||
from tqdm import tqdm
|
||||
|
||||
from config import ClockifyConfig, config
|
||||
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
|
||||
|
||||
|
||||
class ClockifyAPI(ClockifyConfig):
|
||||
def __init__(self, token):
|
||||
self._check_token(token)
|
||||
self.token = token
|
||||
WORKSPACE_ID = config.CLOCKIFY_WORKSPACE_ID
|
||||
USER_ID = config.CLOCKIFY_USER_ID
|
||||
|
||||
def get_activities(self, **kwargs):
|
||||
|
||||
"""
|
||||
{
|
||||
"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)
|
||||
"""
|
||||
activities = requests.get(self.base_url + self.time_entries.format(**kwargs), params=kwargs, headers={"X-API-KEY": self.token}).json()
|
||||
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 in activity["tagIds"]:
|
||||
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)
|
||||
|
||||
print(f"Получено задач: {len(activities)}")
|
||||
print(f"Отработано часов: {time_worked_out_total}")
|
||||
return activities
|
||||
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):
|
||||
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.format(**kwargs), params=kwargs, headers={"X-API-KEY": self.token}).json()
|
||||
return {tag["id"]: tag["name"] for tag in tags}
|
||||
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):
|
||||
resp = requests.get(self.base_url + self.user_url, headers={"X-API-KEY": token})
|
||||
print(resp)
|
||||
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:
|
||||
raise InvalidToken
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user