From fe0ea5faaca3d3ac6a610e532a1c4c818ab8dca7 Mon Sep 17 00:00:00 2001 From: Eduard Date: Mon, 19 May 2025 14:23:07 +0300 Subject: [PATCH] . --- README.MD | 2 +- api/__init__.py | 2 + api.py => api/clockify.py | 86 ++++++++------------------------------- api/redmine.py | 65 +++++++++++++++++++++++++++++ config.py | 58 -------------------------- config/__init__.py | 32 +++++++++++++++ config/clockify.py | 28 +++++++++++++ config/redmine.py | 16 ++++++++ exceptions.py | 2 +- main.py | 32 --------------- track.py | 52 +++++++++++++++++++++++ utils.py | 18 ++++++-- 12 files changed, 229 insertions(+), 164 deletions(-) create mode 100644 api/__init__.py rename api.py => api/clockify.py (55%) create mode 100644 api/redmine.py delete mode 100644 config.py create mode 100644 config/__init__.py create mode 100644 config/clockify.py create mode 100644 config/redmine.py delete mode 100644 main.py create mode 100644 track.py diff --git a/README.MD b/README.MD index 41e5880..0454b92 100644 --- a/README.MD +++ b/README.MD @@ -59,7 +59,7 @@ CLOCKIFY_WORKSPACE_ID= ## Запуск. ``` -python main.py +python track.py ``` diff --git a/api/__init__.py b/api/__init__.py new file mode 100644 index 0000000..ced9c65 --- /dev/null +++ b/api/__init__.py @@ -0,0 +1,2 @@ +from .clockify import ClockifyAPI +from .redmine import RedmineAPI diff --git a/api.py b/api/clockify.py similarity index 55% rename from api.py rename to api/clockify.py index 56c3d52..5fd0175 100644 --- a/api.py +++ b/api/clockify.py @@ -5,19 +5,15 @@ from colorama import Fore import requests from tqdm import tqdm -from config import ClockifyConfig, RedmineConfig, config +from config import ClockifyConfig, config from classes import Activity, Tag -from enums import ActivityType from exceptions import InvalidToken from utils import time_to_hour -WORKSPACE_ID = config.CLOCKIFY_WORKSPACE_ID -USER_ID = config.CLOCKIFY_USER_ID - - class ClockifyAPI(ClockifyConfig): - def __init__(self, token: str): + def __init__(self, token: str, *args, **kwargs): + super().__init__(*args, **kwargs) self.token = self._check_token(token) def get_activities(self, **kwargs) -> list[Activity]: @@ -27,19 +23,26 @@ class ClockifyAPI(ClockifyConfig): page: int - Номер страницы page-size: int - Размер страницы (по умолчанию 50) """ - time_entries_url = self.time_entries_url.format(workspace_id=WORKSPACE_ID, user_id=USER_ID) + 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() logging.info(f"Получено задач: {len(activities)}") time_worked_out_total = 0 result = [] for activity in tqdm(activities, desc="Парсинг задач"): - for tag_id in activity["tagIds"]: + if not activity.get("tagIds"): + logging.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.now()) - start) - 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(" - ") + tag = [tag for tag in self.get_tags() if tag.id == tag_id][0] + try: + task_id, task_title = tag.title.split(" - ") + except ValueError: + logging.info(f"{activity["timeInterval"]["start"]}: {activity["description"]} - Неправильный формат тега {tag}") + continue result.append( Activity( id=activity["id"], @@ -65,7 +68,7 @@ class ClockifyAPI(ClockifyConfig): name: str - optional archived: bool - optional """ - tags = requests.get(self.base_url + self.tags_url.format(**kwargs), params=kwargs, headers={"X-API-KEY": self.token}).json() + 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"] @@ -75,9 +78,8 @@ class ClockifyAPI(ClockifyConfig): for activity in activities: try: resp = requests.put( - self.base_url + self.update_time_entry.format( - workspace_id=WORKSPACE_ID, - id=activity.id + self.base_url + self.update_time_entry( + activity_id=activity.id ), json={ "billable": True, "start": activity.date_start.strftime(config.DATETIME_FORMAT), @@ -108,56 +110,4 @@ class ClockifyAPI(ClockifyConfig): 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]) -> list[Activity]: - if len(activities) == 0: - logging.info("Нет задач для занесения трудочасов") - return - tracked = [] - 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}, - json={ - "time_entry": { - "issue_id": activity.task_id, - "hours": activity.time_spent, - "activity_id": ActivityType.DEVELOPMENT, - "comments": activity.description, - "spent_on": activity.date_start.strftime(self.datetime_format) - } - } - ) - if resp.ok: - activity.is_tracked = True - tracked.append(activity) - logging.debug(activity) - else: - logging.error(f"{activity} Не была затрекана") - logging.error(f"{resp.json()['errors']} {resp.status_code}") - except Exception as e: - logging.error(e) - continue - if len(activities) == len(tracked): - logging.info("Все задачи были успешно занесены в Redmine") - else: - logging.info(f"Не все задачи были занесены в Redmine. Занесено {len(tracked)}") - return 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 + return token \ No newline at end of file diff --git a/api/redmine.py b/api/redmine.py new file mode 100644 index 0000000..546414f --- /dev/null +++ b/api/redmine.py @@ -0,0 +1,65 @@ +import logging + +import requests +from tqdm import tqdm + +from config import RedmineConfig, config +from classes import Activity +from enums import ActivityType +from exceptions import InvalidToken + + +WORKSPACE_ID = config.CLOCKIFY_WORKSPACE_ID +USER_ID = config.CLOCKIFY_USER_ID + + +class RedmineAPI(RedmineConfig): + def __init__(self, token): + self.token = self._check_token(token) + + def track_activities(self, activities: list[Activity]) -> list[Activity]: + if len(activities) == 0: + logging.info("Нет задач для занесения трудочасов") + return + tracked = [] + 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}, + json={ + "time_entry": { + "issue_id": activity.task_id, + "hours": activity.time_spent, + "activity_id": ActivityType.DEVELOPMENT, + "comments": activity.description, + "spent_on": activity.date_start.strftime(self.datetime_format) + } + } + ) + if resp.ok: + activity.is_tracked = True + tracked.append(activity) + logging.debug(activity) + else: + logging.error(f"{activity} Не была затрекана") + logging.error(f"{resp.json()['errors']} {resp.status_code}") + except Exception as e: + logging.error(e) + continue + if len(activities) == len(tracked): + logging.info("Все задачи были успешно занесены в Redmine") + else: + logging.info(f"Не все задачи были занесены в Redmine. Занесено {len(tracked)}") + return 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 \ No newline at end of file diff --git a/config.py b/config.py deleted file mode 100644 index c3f1ef9..0000000 --- a/config.py +++ /dev/null @@ -1,58 +0,0 @@ -from pydantic_settings import BaseSettings - - -class Config(BaseSettings): - CLOCKIFY_TOKEN: str - REDMINE_TOKEN: str - CLOCKIFY_USER_ID: str - CLOCKIFY_WORKSPACE_ID: str - - DATETIME_FORMAT: str = "%Y-%m-%dT%H:%M:%SZ" - - - -class RedmineConfig: - @property - def base_url(self): - return "https://redmine.sbps.ru" - - @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: - @property - def base_url(self): - return "https://api.clockify.me/api/v1" - - @property - def tags_url(self): - return "/workspaces/{workspace_id}/tags" - - @property - def time_entries_url(self): - return "/workspaces/{workspace_id}/user/{user_id}/time-entries" - - @property - def user_url(self): - return "/user" - - @property - def datetime_format(self): - return "%Y-%m-%dT%H:%M:%SZ" - - @property - def update_time_entry(self): - return "/workspaces/{workspace_id}/time-entries/{id}" - - -config = Config() diff --git a/config/__init__.py b/config/__init__.py new file mode 100644 index 0000000..bc85cb5 --- /dev/null +++ b/config/__init__.py @@ -0,0 +1,32 @@ +import argparse +from pydantic_settings import BaseSettings + + +from .clockify import ClockifyConfig # noqa: F401 +from .redmine import RedmineConfig # noqa: F401 + + +class Config(BaseSettings): + CLOCKIFY_TOKEN: str + REDMINE_TOKEN: str + CLOCKIFY_USER_ID: str + CLOCKIFY_WORKSPACE_ID: str + + DATETIME_FORMAT: str = "%Y-%m-%dT%H:%M:%SZ" + + +def configure_argument_parser(): + parser = argparse.ArgumentParser(description='Трекер трудочасов с Clockify в Redmine') + parser.add_argument( + '-s', + '--start', + help='Дата начала парсинга активностей в формате гггг-мм-дд' + ) + parser.add_argument( + '-e', + '--end', + help='Дата окончания парсинга активностей в формате гггг-мм-дд' + ) + return parser + +config = Config() \ No newline at end of file diff --git a/config/clockify.py b/config/clockify.py new file mode 100644 index 0000000..78006d9 --- /dev/null +++ b/config/clockify.py @@ -0,0 +1,28 @@ +class ClockifyConfig: + def __init__(self, workspace_id: str, user_id: str): + self.workspace_id = workspace_id + self.user_id = user_id + + @property + def base_url(self): + return "https://api.clockify.me/api/v1" + + @property + def tags_url(self): + return f"/workspaces/{self.workspace_id}/tags" + + @property + def time_entries_url(self): + return f"/workspaces/{self.workspace_id}/user/{self.user_id}/time-entries" + + @property + def user_url(self): + return "/user" + + @property + def datetime_format(self): + return "%Y-%m-%dT%H:%M:%SZ" + + @property + def update_time_entry(self, activity_id): + return f"/workspaces/{self.workspace_id}/time-entries/{activity_id}" \ No newline at end of file diff --git a/config/redmine.py b/config/redmine.py new file mode 100644 index 0000000..d30cc77 --- /dev/null +++ b/config/redmine.py @@ -0,0 +1,16 @@ +class RedmineConfig: + @property + def base_url(self): + return "https://redmine.sbps.ru" + + @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" \ No newline at end of file diff --git a/exceptions.py b/exceptions.py index f61a875..3a3bdda 100644 --- a/exceptions.py +++ b/exceptions.py @@ -3,4 +3,4 @@ class InvalidToken(Exception): pass def __str__(self): - return "Токен недействителен" \ No newline at end of file + return "Токен недействителен" diff --git a/main.py b/main.py deleted file mode 100644 index 58fd4cf..0000000 --- a/main.py +++ /dev/null @@ -1,32 +0,0 @@ -import logging -import sys - -from api import ClockifyAPI, RedmineAPI -from config import config -from utils import today_start, today_end - - -def main(): - clock = ClockifyAPI(config.CLOCKIFY_TOKEN) - red = RedmineAPI(config.REDMINE_TOKEN) - activities = clock.get_activities( - start=today_start(), - end=today_end() - ) - tracked_activities = red.track_activities(activities) - clock.mark_tracked(tracked_activities) - - -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/track.py b/track.py new file mode 100644 index 0000000..308baeb --- /dev/null +++ b/track.py @@ -0,0 +1,52 @@ +import logging +import sys + +from api import ClockifyAPI, RedmineAPI +from config import config, configure_argument_parser +from utils import format_date, str_to_date, today_start, today_end + + +WORKSPACE_ID = config.CLOCKIFY_WORKSPACE_ID +USER_ID = config.CLOCKIFY_USER_ID + + +def main(): + arg_parser = configure_argument_parser() + args = arg_parser.parse_args() + start, end = None, None + if args.start: + start = str_to_date(args.start) + if args.end: + end = str_to_date(args.end) + if args.start is None and args.end is None: + start = today_start() + end = today_end() + + clock = ClockifyAPI( + token=config.CLOCKIFY_TOKEN, + workspace_id=WORKSPACE_ID, + user_id=USER_ID + ) + red = RedmineAPI(config.REDMINE_TOKEN) + activities = clock.get_activities( + start=format_date(start), + end=format_date(end) + ) + # tracked_activities = red.track_activities(activities) + # clock.mark_tracked(tracked_activities) + print(activities[-1]) + + +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 0942b96..04b37d2 100644 --- a/utils.py +++ b/utils.py @@ -1,20 +1,30 @@ from datetime import datetime, timedelta +from typing import Optional from config import config def time_to_hour(time: timedelta) -> float: return time.total_seconds() / 3600 -def today_start(): +def today_start() -> datetime: td = datetime.today() year = td.year month = td.month day = td.day - return datetime(year=year, month=month, day=day, hour=0).strftime(config.DATETIME_FORMAT) + return datetime(year=year, month=month, day=day, hour=0) -def today_end(): +def today_end() -> datetime: td = datetime.today() year = td.year month = td.month day = td.day - return datetime(year=year, month=month, day=day, hour=23).strftime(config.DATETIME_FORMAT) \ No newline at end of file + return datetime(year=year, month=month, day=day, hour=23) + + +def str_to_date(date: str) -> Optional[datetime]: + if date: + return datetime.strptime(date + "T00:00:00Z", config.DATETIME_FORMAT) + +def format_date(date: datetime) -> datetime: + if date: + return date.strftime(config.DATETIME_FORMAT)