from argparse import Namespace from datetime import datetime, timedelta import logging import os import re from colorama import Fore from config import config logger = logging.getLogger(__name__) def time_to_hour(time: timedelta) -> float: return time.total_seconds() / 3600 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) 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, minute=59, second=59) def str_to_date(date: str) -> datetime | None: if date: return datetime.strptime(date + "T00:00:00Z", config.DATETIME_FORMAT) def format_date(date: datetime) -> str | None: if date: return date.strftime(config.DATETIME_FORMAT) def parse_task_id(tag: str) -> int: if match := re.search(r"^(?P\d{5})", tag): return int(match.group("task_id")) raise ValueError(f'Тег "{tag}" должен начинаться с номера задачи из 5 цифр') def db_update(clock, db, red): """Обновляет локальную БД данными из Clockify и Redmine.""" logger.info(Fore.YELLOW + "Обновление базы данных..." + Fore.RESET) try: # Обновление Тегов logger.info(f"{Fore.YELLOW}Обновление Тегов...\n{Fore.RESET}") clock_tags = clock.get_tags(is_updating=True, **{"page-size": clock.tags_count}) if clock_tags: db_tags = db.get_tags() new_tags = [tag for tag in clock_tags if tag not in db_tags] if new_tags: try: db.insert_tags(new_tags) logger.info(f"{Fore.GREEN}Обновлено тегов: {len(new_tags)}{Fore.RESET}") except Exception as e: logger.error(f"{Fore.RED}Ошибка при сохранении тегов!{Fore.RESET}") else: logger.info(f"{Fore.YELLOW}Обновление тегов не требуется.\n{Fore.RESET}") # Обновление Задач logger.info(f"{Fore.YELLOW}Обновление Задач...\n{Fore.RESET}") red_tasks = red.get_issues() if red_tasks: db_tasks = db.get_tasks() clock_task_ids = {parse_task_id(tag.title) for tag in [*db_tags, *new_tags]} clock_tasks = [task for task in red_tasks if task.id in clock_task_ids] new_tasks = [task for task in clock_tasks if task not in db_tasks] if new_tasks: try: db.insert_tasks(new_tasks) logger.info(f"{Fore.GREEN}Обновлено задач: {len(new_tasks)}{Fore.RESET}") except Exception as e: logger.error(f"{Fore.RED}Ошибка при сохранении задач!{Fore.RESET}") else: logger.info(f"{Fore.YELLOW}Обновление задач не требуется\n{Fore.RESET}") # Обновление Проектов Ветро logger.info(f"{Fore.YELLOW}Обновление Проектов Ветро{Fore.RESET}") clock_projects = clock.get_vetro_projects() if clock_projects: db_projects = db.get_vetro_projects() new_projects = [project for project in clock_projects if project not in db_projects] if new_projects: try: db.insert_vetro_projects(new_projects) logger.info(f"{Fore.GREEN}Обновлено Проектов Ветро: {len(new_projects)}{Fore.RESET}") except Exception as e: logger.error(f"{Fore.RED}Ошибка при сохранении Проектов Ветро!{Fore.RESET}") else: logger.info(f"{Fore.YELLOW}Обновление Проектов Ветро не требуется\n{Fore.RESET}") # Обновление Типов деятельности logger.info(f"{Fore.YELLOW}Обновление Типов деятельности{Fore.RESET}") red_types = red.get_time_entry_activities() if red_types: db_types = db.get_activity_types() new_types = [type_ for type_ in red_types if type_ not in db_types] if new_types: try: db.insert_activity_types(new_types) logger.info(f"{Fore.GREEN}Обновлено Типов деятельности: {len(new_types)}{Fore.RESET}") except Exception as e: logger.error(f"{Fore.RED}Ошибка при сохранении Типов деятельности!{Fore.RESET}") else: logger.info(f"{Fore.YELLOW}Обновление Типов деятельности не требуется\n{Fore.RESET}") except Exception as e: logger.error(Fore.RED + f"Ошибка при обновлении базы данных: {e}" + Fore.RESET) else: logger.info(Fore.GREEN + "База данных теперь актуальна." + Fore.RESET) def get_start_end_dates(args: Namespace): 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: logger.info("Парсим задачи за сегодняшний день") start = today_start() end = today_end() else: if start: from_ = f"{start.day} {config.MONTHS.get(start.month)} {start.year} г." if end: to_ = f"{end.day} {config.MONTHS.get(end.month)} {end.year} г." if start == end: logger.info(f"{Fore.YELLOW}Парсим задачи за {from_}{Fore.RESET}") else: logger.info(Fore.YELLOW + "Парсим задачи" + (f" с {from_} " if start else " ") + (f"до {to_}" if end else "") + Fore.RESET) return start, end def check_envs(clock): """Проверяет наличие обязательных переменных окружения Clockify.""" if not config.WORKSPACE_ID or not config.USER_ID: if not os.path.exists(config.ENV_TXT): logger.info(f"{Fore.YELLOW}Необходимые переменные не были обнаружены. Попытка получения...{Fore.RESET}") clock.getenvs() else: logger.info(f'{Fore.YELLOW}Проверьте файл "{config.ENV_TXT}". Если необходимых переменных в нём нет, то просто удалите его.{Fore.RESET}') return False return True