bars-icloud-drive/src/config_parser.py

import os

from src import (
    DEFAULT_DRIVE_DESTINATION,
    DEFAULT_DRIVE_DESTINATION_EXPORT,
    DEFAULT_RETRY_LOGIN_INTERVAL_SEC,
    DEFAULT_ROOT_DESTINATION,
    DEFAULT_SYNC_INTERVAL_SEC,
    LOGGER,
)

def config_path_to_string(config_path):
    return " > ".join(config_path)

def traverse_config_path(config, config_path: list[str]) -> bool:
    if len(config_path) == 0:
        return True
    if not (config and config_path[0] in config):
        return False
    return traverse_config_path(config[config_path[0]], config_path=config_path[1:])

def get_config_value(config, config_path):
    if len(config_path) == 1:
        return config[config_path[0]]
    return get_config_value(config=config[config_path[0]], config_path=config_path[1:])

def get_username(config):
    username = None
    config_path = ["app", "credentials", "username"]
    if not traverse_config_path(config=config, config_path=config_path):
        LOGGER.error(
            f"username отсутствует в {config_path_to_string(config_path)}. Пожалуйста, установите имя пользователя."
        )
    else:
        username = get_config_value(config=config, config_path=config_path)
        username = username.strip()
        if len(username) == 0:
            username = None
            LOGGER.error(f"username пустое в {config_path_to_string(config_path)}.")
    return username

def get_retry_login_interval(config):
    retry_login_interval = DEFAULT_RETRY_LOGIN_INTERVAL_SEC
    config_path = ["app", "credentials", "retry_login_interval"]
    if not traverse_config_path(config=config, config_path=config_path):
        LOGGER.warning(
            f"retry_login_interval не найден в {config_path_to_string(config_path=config_path)}."
            + f" Использование по умолчанию {retry_login_interval} секунд ..."
        )
    else:
        retry_login_interval = get_config_value(config=config, config_path=config_path)
        LOGGER.info(f"Повторная попытка входа каждые {retry_login_interval} секунд.")
    return retry_login_interval

def get_drive_sync_interval(config):
    sync_interval = DEFAULT_SYNC_INTERVAL_SEC
    config_path = ["drive", "sync_interval"]
    if not traverse_config_path(config=config, config_path=config_path):
        LOGGER.warning(
            f"sync_interval не найден в {config_path_to_string(config_path=config_path)}."
            + f" Использование sync_interval по умолчанию: {sync_interval} секунд ..."
        )
    else:
        sync_interval = get_config_value(config=config, config_path=config_path)
    return sync_interval

def prepare_root_destination(config):
    LOGGER.debug("Проверка root ...")
    root_destination = DEFAULT_ROOT_DESTINATION
    config_path = ["app", "root"]
    if not traverse_config_path(config=config, config_path=config_path):
        LOGGER.warning(
            f"Предупреждение: root отсутствует в {config_path_to_string(config_path)}."
            + f" Использование root по умолчанию: {root_destination}",
        )
    else:
        root_destination = get_config_value(config=config, config_path=config_path)
    root_destination_path = os.path.abspath(root_destination)
    os.makedirs(root_destination_path, exist_ok=True)
    return root_destination_path

def prepare_drive_destination(config):
    LOGGER.debug("Проверка пути сохранения фалов...")
    config_path = ["drive", "destination"]
    drive_destination = DEFAULT_DRIVE_DESTINATION
    if not traverse_config_path(config=config, config_path=config_path):
        LOGGER.warning(
            f"Внимание: путь сохранения фалов отсутствует в {config_path_to_string(config_path)}."
            + f" Использование путь сохранения фалов по умолчанию: {drive_destination}."
        )
    else:
        drive_destination = get_config_value(config=config, config_path=config_path)
    drive_destination_path = os.path.abspath(
        os.path.join(prepare_root_destination(config=config), drive_destination)
    )
    os.makedirs(drive_destination_path, exist_ok=True)
    return drive_destination_path

def prepare_drive_destination_export(config):
    LOGGER.debug("Проверка пути конвертации фалов ...")
    config_path = ["drive", "destination_export"]
    drive_destination_export = DEFAULT_DRIVE_DESTINATION_EXPORT
    if not traverse_config_path(config=config, config_path=config_path):
        LOGGER.warning(
            f"Внимание: путь конвертации фалов отсутствует в {config_path_to_string(config_path)}."
            + f" Использование путь конвертации фалов по умолчанию: {drive_destination_export}."
        )
    else:
        drive_destination_export = get_config_value(config=config, config_path=config_path)
    drive_destination_export_path = os.path.abspath(
        os.path.join(prepare_root_destination(config=config), drive_destination_export)
    )
    os.makedirs(drive_destination_export_path, exist_ok=True)
    return drive_destination_export_path


def get_drive_remove_obsolete(config):
    drive_remove_obsolete = False
    config_path = ["drive", "remove_obsolete"]
    if not traverse_config_path(config=config, config_path=config_path):
        LOGGER.warning(
            f"Предупреждение: remove_obsolete не найден в {config_path_to_string(config_path)}."
            + " Используется параметр не удалять устаревшие файлы и папки."
        )
    else:
        drive_remove_obsolete = get_config_value(config=config, config_path=config_path)  
    return drive_remove_obsolete

def get_region(config):
    region = "global"
    config_path = ["app", "region"]
    if not traverse_config_path(config=config, config_path=config_path):
        LOGGER.warning(
            f"{config_path_to_string(config_path=config_path)} не найдено. Использует значение по умолчанию - global ..."
        )
    else:
        region = get_config_value(config=config, config_path=config_path)
        if region not in ["global", "china"]:
            LOGGER.error(
                f"{config_path_to_string(config_path=config_path)} недействительно. \
                            Допустимые значения - global или china. Использование значения по умолчанию - global ..."
            )
            region = "global"

    return region