From da7ae0a370c788ebd575f68fda8f3e8833e4621b Mon Sep 17 00:00:00 2001 From: Eduard Date: Thu, 15 May 2025 13:44:09 +0300 Subject: [PATCH] 123 --- api.py | 125 +++++++++++++++++++++++++++++++++++++++++++++-------- classes.py | 27 ++++++++++++ config.py | 25 ++++++++--- enums.py | 9 ++++ main.py | 24 +++++++++- utils.py | 2 +- 6 files changed, 185 insertions(+), 27 deletions(-) create mode 100644 classes.py create mode 100644 enums.py diff --git a/api.py b/api.py index e08e9a1..0d766d3 100644 --- a/api.py +++ b/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 \ No newline at end of file + 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 diff --git a/classes.py b/classes.py new file mode 100644 index 0000000..8c81e1b --- /dev/null +++ b/classes.py @@ -0,0 +1,27 @@ +from dataclasses import dataclass +from datetime import datetime + + +@dataclass +class Activity: + task_id: int + task_title: str + description: str + time_spent: float + date: datetime = datetime.now() + + @property + def is_tracked(self): + return self.description.startswith("*") + + def __str__(self): + return f'{self.date}: {self.description.strip("*")} для задачи "{self.task_title}"' + (" - ЗАТРЕКАНА" if self.is_tracked else "") + + +@dataclass +class Tag: + id: str + title: str + + def __str__(self): + return self.title diff --git a/config.py b/config.py index 24314d7..6fc9984 100644 --- a/config.py +++ b/config.py @@ -11,14 +11,25 @@ class Config(BaseSettings): -class RedmineConfig(BaseSettings): - BASE_API_URL: str = "https://redmine.sbps.ru" +class RedmineConfig: + @property + def base_url(self): + return "https://redmine.sbps.ru" - TIME_ENTRIES: str = "/time_entries.json" + @property + def time_entries_url(self): + return "/time_entries.json" + + @property + def my_account_url(self): + return "/my/account.json" + + @property + def datetime_format(self): + return "%Y-%m-%d" -class ClockifyConfig(BaseSettings): - +class ClockifyConfig: @property def base_url(self): return "https://api.clockify.me/api/v1" @@ -34,6 +45,10 @@ class ClockifyConfig(BaseSettings): @property def user_url(self): return "/user" + + @property + def datetime_format(self): + return "%Y-%m-%dT%H:%M:%SZ" config = Config() diff --git a/enums.py b/enums.py new file mode 100644 index 0000000..8adaf65 --- /dev/null +++ b/enums.py @@ -0,0 +1,9 @@ +from enum import Enum + + +class ActivityType(Enum): + DEVELOPMENT = 9 + BUSINESS_ANALYZE = 8 + CODE_REVIEW = 57 + CONSULTATIONS = 58 + SUPPORT = 60 diff --git a/main.py b/main.py index fb60ee3..6b3e653 100644 --- a/main.py +++ b/main.py @@ -1,11 +1,31 @@ +import logging +import sys -from api import ClockifyAPI +from api import ClockifyAPI, RedmineAPI from config import config +from utils import today_start, today_end def main(): - ClockifyAPI(config.CLOCKIFY_TOKEN) + clock = ClockifyAPI(config.CLOCKIFY_TOKEN) + for activity in clock.get_activities( + start=today_start(), + end=today_end() + ): + print(activity) + # red = RedmineAPI(config.REDMINE_TOKEN) if __name__ == "__main__": + logging.basicConfig( + level=logging.INFO, + format=('%(asctime)s, ' + '%(levelname)s, ' + '%(funcName)s, ' + '%(message)s' + ), + encoding='UTF-8', + handlers=[logging.FileHandler(__file__ + '.log'), + logging.StreamHandler(sys.stdout)] + ) main() diff --git a/utils.py b/utils.py index eeb6549..0942b96 100644 --- a/utils.py +++ b/utils.py @@ -1,6 +1,6 @@ from datetime import datetime, timedelta -import config +from config import config def time_to_hour(time: timedelta) -> float: return time.total_seconds() / 3600