.

api/clockify.py

@@ -0,0 +1,113 @@
+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
+from exceptions import InvalidToken
+from utils import time_to_hour
+
+
+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()
+        logging.info(f"Получено задач: {len(activities)}")
+        time_worked_out_total = 0
+
+        result = []
+        for activity in tqdm(activities, desc="Парсинг задач"):
+            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() 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"],
+                        tag_id=tag_id,
+                        task_id=task_id,
+                        task_title=task_title,
+                        project_id=activity["projectId"],
+                        description=activity["description"],
+                        time_spent=time_to_hour((end if end else datetime.now()) - start),
+                        date_start=start,
+                        date_end=end,
+                        is_tracked=activity["billable"]
+                    )
+                )
+        logging.info(f"Сохранено задач: {len(result)}")
+        logging.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 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],
+                        "projectId": activity.project_id
+                    }, headers={
+                        "X-API-KEY": self.token
+                    }
+                )
+                if resp.ok:
+                    logging.debug(resp.json())
+                    logging.info(f"{activity} {Fore.GREEN} помечена как затреканная{Fore.RESET}")
+                else:
+                    logging.debug(resp.json())
+                    logging.error(f"{activity} {Fore.RED} не помечена как затреканная{Fore.RESET}")
+            except Exception as e:
+                logging.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:
+            logging.error(resp.json())
+            raise InvalidToken
+        resp = resp.json()
+        logging.debug({"id": resp["id"], "email": resp["email"], "name": resp["name"]})
+        return token