Получение тегов с сервера только при обновлении БД. Исправил баг с предупреждением об устаревшей БД, когда не указан vetro_project

This commit is contained in:
2025-06-19 11:19:48 +03:00
parent b5a9232467
commit 77f35bc757
3 changed files with 42 additions and 13 deletions

View File

@@ -23,7 +23,7 @@ class ClockifyAPI(ClockifyConfig):
super().__init__(*args, **kwargs)
self.token = self._check_token(token)
def get_activities(self, **kwargs) -> list[Activity]:
def get_activities(self, is_updating=False, **kwargs) -> list[Activity]:
"""
Получает с сервера Clockify активности за определнный период
start: datetime - Начальная дата (обязательно)
@@ -36,7 +36,10 @@ class ClockifyAPI(ClockifyConfig):
logger.info(f"Получено задач: {len(activities)}")
time_worked_out_total = 0
tags = self.get_tags(**{"page-size": self.tags_count})
tags = self.get_tags(is_updating=is_updating)
if not tags:
logging.info(f"{Fore.YELLOW}В базе данных нет тегов! Обновите БД при помощи флага -i [--init-db]{Fore.RESET}")
return []
result = []
for activity in tqdm(activities, desc="Парсинг задач"):
@@ -48,8 +51,13 @@ class ClockifyAPI(ClockifyConfig):
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 tags if tag.id == tag_id]
if not tag:
logging.info(f"{Fore.YELLOW}В базе данных нет нужного тега {Fore.GREEN}{tag_id}{Fore.YELLOW}! Обновите БД при помощи флага -i [--init-db]{Fore.RESET}")
break
else:
tag = tag[0]
try:
task_id = parse_task_id(tag[0].title)
task_id = parse_task_id(tag.title)
except ValueError:
logger.info(f'{activity["timeInterval"]["start"]}: {activity["description"]} - Неправильный формат тега {tag}')
continue
@@ -76,18 +84,21 @@ class ClockifyAPI(ClockifyConfig):
return result
def get_tags(self, **kwargs) -> list[Tag]:
def get_tags(self, is_updating=False, **kwargs) -> list[Tag]:
"""
Получает с сервера Clockify теги
name: str - optional
archived: bool - optional
page-size: int - 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]
if is_updating:
tags = requests.get(self.base_url + self.tags_url, params={"sort-column": "name", "sort-order": "descending", **kwargs}, headers={"X-API-KEY": self.token}).json()
logging.debug(f"{Fore.YELLOW}Получено тегов: {len(tags)}.{Fore.RESET}")
return [Tag(
id=tag["id"],
title=tag["name"]
) for tag in tags]
return db.get_tags()
def get_vetro_projects(self) -> list[VetroProject]:
"""