libmax.py - master - MaxIs/maxlib

libmax.py
972 строки44 KB
MaxIs
Добавлен класс WorkCMD для работы с командной строкой
29 июн 2025, 20:02
29 июн 2025, 20:02f39a046
import locale
import os
import secrets
import string
import subprocess
import sys
import time
from statistics import median_low
from string import punctuation

from dotenv import load_dotenv
from sqlalchemy import String, Integer, Float, DateTime, create_engine, text
from pathlib import Path
from cryptography.fernet import Fernet
import numpy as np
import pandas as pd
import datetime
from chardet import UniversalDetector
from win32com.client import Dispatch


def get_diff_day(date_values, current, holidays, work: bool = True):
    """Функция расчёта разницы дней в зависимости от того нужно нам считать по календарным дням или по рабочим дням
    :param date_values = array datetime64[D] массив дней к которым нам нужно посчитать разницу
    :param current = array datetime64[D] or datetime64[D] value значение дня или массив значений к которым мы считаем
    разницу
    :param holidays = array datetime64[D] праздники, (включая выходные дни недели) если мы считаем разницу в рабочих
    днях
    :param work = bool как считать по рабочим дням или по календарным"""
    date_values = np.array(date_values, dtype='datetime64[D]')
    if type(current) == datetime.datetime:
        current = np.datetime64(current, 'D')
    if type(current) == np.ndarray:
        current = np.array(current, dtype='datetime64[D]')
    if work:
        go_list = np.busday_count(begindates=date_values, enddates=current, holidays=holidays)
    else:
        go_list = np.busday_count(begindates=date_values, enddates=current, weekmask=[1, 1, 1, 1, 1, 1, 1])

    return go_list


def read_config(path):
    """
    Получаем словарь настройки в пути текстовый файл вида
    key1=value1
    key2=value2
    :param path: Путь до файла с настройками
    :return: словарь с настройками
    """
    conf_dict = {}
    with open(path, 'r') as f:
        lines = f.read().split('\n')
        for line in lines:
            conf_dict[line.split('=')[0]] = line.split('=')[-1]
    return conf_dict


def df_type(df):
    """
    Cоздание словаря типов данных для выделения типов при переносе на sql сервер, расчёт максимальной длины строки
    :param df: принимает таблицу для анализа
    :return: словарь с названием столбца и типом данных
    """
    df_types = {}
    for x, y in enumerate(df.columns):
        if df[y].dtypes == object:
            df[y] = df[y].astype(str)
            df_types[y] = String(length=df[y].str.len().max())
        elif (df[y].dtypes == np.int64 or df[y].dtypes == np.int32 or df[y].dtypes == np.int16 or
              df[y].dtypes == np.int8):
            df_types[y] = Integer()
        elif df[y].dtypes == float:
            df_types[y] = Float()
        elif df[y].dtypes == np.dtype('datetime64[ns]') or df[y].dtypes == np.dtype('<M8[us]'):
            df_types[y] = DateTime()
    return df_types


def decrypt(path, path_dir):
    """Расшифровка данных в файле с помощью ключа crypto.key"""
    # Загружаем ключ 'crypto.key' из текущего каталога
    key = open(path_dir + r'\crypto.key', 'rb').read()
    # Расшифруем файл и записываем его
    f = Fernet(key)
    with open(path, 'rb') as file:
        # читать зашифрованные данные
        encrypted_data = file.read()
    # расшифровать данные
    decrypted_data = f.decrypt(encrypted_data)
    with open(path, 'wb') as file:
        file.write(decrypted_data)
    with open(path, 'r') as file:
        # читать зашифрованные данные
        decrypted_data1 = file.read()
    with open(path, 'rb') as file:
        # прочитать все данные файла
        file_data = file.read()
    encrypted_data = f.encrypt(file_data)
    with open(path, 'wb') as file:
        file.write(encrypted_data)
    return decrypted_data1


def execute_oracle_sql(sql_str: str, server: str, port: str, service_name: str, user: str, pswd: str,
                       path_client: str, is_table: bool = True):
    """
    Выполнение курсора в бд Oracle и получение результата
    :param sql_str: строка запроса
    :param server: название сервера
    :param port: порт по которому происходит подключение
    :param service_name: сервисное имя для подключения
    :param user: логин
    :param pswd: пароль
    :param path_client: путь до расположения клиентских библиотек oracle
    :param is_table: bool нужно ли получить ответ в формате pd.DataFrame
    :return:
    """
    try:
        import cx_Oracle
        cx_Oracle.init_oracle_client(path_client)
    except Exception as _ex:
        print('[INFO] Error connection Oracle DB', _ex)

    try:
        dsn_tns = cx_Oracle.makedsn(server, port, service_name=service_name)
        with cx_Oracle.connect(user=user, password=pswd, dsn=dsn_tns) as conn:
            cur = conn.cursor()
            cur.execute(sql_str)
            res = cur.fetchall()
            if is_table:
                cols = [col[0] for col in cur.description]
                res = pd.DataFrame(data=res, columns=cols)
    except Exception as _ex:
        print('[INFO] Error Oracle DB', _ex)
    return res


def connect(**kwargs):
    """
    Осуществляем подключение к серверу
    :param server: имя сервера
    :param database: база данных
    :param username: имя пользователя
    :param password: пароль
    :param driver: драйвер используемый для подключения по умолчанию 'driver=SQL+Server'
    :param fast_executemany: использовать ли пакетную загрузку, по умолчанию да
    :param use_setinputsizes: использовать параметры ширины таблицы (нужно отключить в версии 2.0.0 или выше) = False
    :param trust_con: использовать ли доверенное подключение от windows, если мы заходим под своей учёткой
    :param condrive: используемый драйвер и структура БД по умолчанию mssql+pyodbc
    :return: engine экземпляр подключения create_engine
    """
    server = kwargs['server']
    database = kwargs['database']
    username = kwargs['username']
    password = kwargs['password']
    driver = kwargs['driver']  # : str = 'driver=SQL+Server',
    fast_executemany = kwargs['fast_executemany']  # : bool = True,
    use_setinputsizes = kwargs['use_setinputsizes']  # : bool = False,
    trust_con = kwargs['trust_con']  # : str = 'trusted_connection=no',
    condrive = kwargs['condrive']  # : str = 'mssql+pyodbc'
    if trust_con == 'trusted_connection=no':
        engine = create_engine(f'{condrive}://{username}:{password}@{server}/{database}?{driver}',
                               fast_executemany=fast_executemany, use_setinputsizes=use_setinputsizes)
    else:
        engine = create_engine(f'{condrive}://{server}/{database}?{driver}&{trust_con}',
                               fast_executemany=fast_executemany, use_setinputsizes=use_setinputsizes)
    return engine


def exec_mssql_db(engine, sql):
    """Использование курсора для извлечения данных, передача соединения из предыдущей функции"""
    with engine.connect() as con:
        data = con.execute(text(sql))
        sql_data = pd.DataFrame(data.fetchall())
        sql_data.columns = data.keys()
        con.commit()
        con.close()
    return sql_data


def optimal_chunksize(
        df1,
        table: str,
        server: str,
        database: str,
        username: str,
        password: str,
        optimal_chunk='chunk',
        if_exists='replace',
        driver='driver=SQL+Server',
        dtypes: bool | dict = True,
        schema='dbo',
        start=100,
        stop=5000,
        step=100):
    """
     Функция ищет оптимальное количество строк в пакете при котором предполагается минимальное время передачи
     информации на сервер и определяет словарь с типами данных. Она предоставляет значение оптимального пакета,
     дополнительно предоставляет словарь с типами данных. Вы можете изменить list(optimal.keys())[0] на optimal
     и получите словарь, где ключ это количество строк в пакете, а значение это среднее время затраченное
     на экспорт 1 строки на сервер.
     Функция принимает:
     df1 - таблица которую мы будем загружать на сервер
     dtypes - словарь с конкретными типами данных в формате {Название столбца : тип данных} например
     {столбец1: Integer(),
     столбец2: Float(),
     столбец3: DateTime(),
     столбец4: String(length=65)} при строковых значениях указываем максимальную длину столбца, это необходимо для
     использования функции пакетной отправки, по умолчанию стоит True что означает вычислить в процессе необходимые типы
     table - название таблицы которую мы записываем на сервере
     schema - название схемы на сервере по умолчанию 'dbo'
     server - название сервера
     database - название базы данных
     username - имя для подключения к серверу
     password - пароль для подключения к серверу
     driver - драйвер для подключения к серверу по умолчанию подключение к mssql 'driver=SQL+Server'
     if_exists - что нужно сделать с таблицей если она существует на сервере при поиске пакета,
     по умолчанию нужно перезаписать таблицу.
     optimal_chunk - параметр определяющий в каком виде мы хотим получить результат, 2 возможных значения:
         'chunk' - мы получаем число - лучшее соотношение количества строк в пакете из вашего периода
         'dict' - получаем отсортированный словарь с ключами равными количеству строк в пакете,
         с значениями равными среднем временем загрузки 1 строки на сервер.
     start - значение количества строк с которого начнётся поиск оптимального пакета, по умолчанию 100 строк
     stop - значение которым окончится поиск строк, по умолчанию 5000 строк
     step - шаг увеличения количества строк с которым будет осуществляться поиск оптимального пакета, по умолчанию 100
     шаг должен делить диапазон проверки без остатка.
     Зависимости
     Pandas==1.4.3, numpy==1.20.3, sqlalchemy==1.4.22 или выше, само собой драйвер pyodbc для подключения к БД и
     учётные данные для подключения
    """
    start_time = time.time()
    if dtypes is True:
        df_types = df_type(df1)
    else:
        df_types = dtypes

    optimal = {}
    engine = connect(server=server, database=database, username=username, password=password, driver=driver,
                     connection=r'mssql+pyodbc', fast_executemany=True)
    num = 0

    for x in range(start, stop, step):
        df = df1[:x].copy()
        start_time_pyodbc = time.time()
        df.to_sql(table, con=engine, if_exists=if_exists, index=False, schema=schema,
                  dtype=df_types)
        full_time_pyodbc = round(time.time() - start_time_pyodbc, 5)
        optimal[x] = full_time_pyodbc / x
        percent = float((int((stop - start) / step) - num + 1) / int(stop / step))
        print(f'Окончена проверка {num} пакетов, осталось проверить {int(100 * percent)}% пакетов')
        num += 1
    engine.execute(f'''DROP TABLE IF EXISTS {database}.{schema}.{table}''')
    engine.dispose()
    sorted_tuples = sorted(optimal.items(), key=lambda item: item[1])
    optimal = {k: v for k, v in sorted_tuples}
    full_time = time.time() - start_time
    economy_time = ((median_low(list(optimal.values())) - min(list(optimal.values()))) * len(df1)) - full_time
    print(f"""
    Проверка окончена, затрачено времени {round(full_time, 2)} сек., \n
    съэкономлено времени {round(economy_time, 2)} сек.""")
    if dtypes is True:
        if optimal_chunk == 'chunk':
            return list(optimal.keys())[0], df_types
        elif optimal_chunk == 'dict':
            return optimal, df_types
    if optimal_chunk == 'chunk':
        return list(optimal.keys())[0]
    elif optimal_chunk == 'dict':
        return optimal


def generate_pswd():
    """Генерируем пароль"""
    pstr = string.ascii_letters + string.digits + '!#&?.@<>'
    return ''.join(secrets.choice(pstr) for _ in range(20))


def str_sum(df_sum, key, col_sum):
    """
    Эта функция группирует значения col_sum по не идентичным ключам key и выводит готовую таблицу,
    в которой будет заменённый столбец col_sum, со всеми возможными значениями в группе через символ | .
    Она принимает
    1. df_sum исходная таблица - формат dataframe
    2. key ключ - может быть как строкой (название столбца), так и списком (названий столбцов)
    3. col_sum название агрегируемого столбца - пока поддерживается только формат строки, т.е. 1 столбец
    """
    df = df_sum.copy()
    df_sum = df_sum.copy()
    for col in df_sum.columns:
        df_sum[col] = df_sum[col].astype(str).str.strip(punctuation + ' ')
        try:
            df_sum[col] = pd.to_numeric(df_sum[col], errors='raise')
        except:
            continue
    if type(key) is list:
        df[col_sum] = df[col_sum] + r'|'
        for col in key:
            df[col] = df[col].astype(str) + r'|'
            df_sum[col] = df_sum[col].astype(str) + r'|'
        key_sum = '|'.join(key)
        df[key_sum] = df[key].sum(axis=1)
        df = df.groupby(by=[key_sum], as_index=False).agg({col_sum: 'sum'})
        df = df[[key_sum, col_sum]].copy()
        df_sum[key_sum] = df_sum[key].sum(axis=1)
        if df[key_sum].dtypes != df_sum[key_sum].dtypes:
            if df_sum[key_sum].dtypes == float:
                df[key_sum] = df[key_sum].astype(float)
            elif df_sum[key_sum].dtypes == int:
                df[key_sum] = df[key_sum].astype(int)
            elif df_sum[key_sum].dtypes == object:
                df[key_sum] = df[key_sum].astype(object)
        df_sum = df_sum.merge(df, how='left', left_on=[key_sum], right_on=[key_sum])
        col_sum_x, col_sum_y = sorted([x for x in df_sum.columns if col_sum in str(x)])
        for col in [key_sum, col_sum_x, col_sum_y]:
            df_sum[col] = df_sum[col].str.rstrip('|')
        del df_sum[col_sum_x]
        df_sum.rename(columns={col_sum_y: col_sum}, inplace=True)
        for col in key:
            del df_sum[col]
        df_sum[key] = df_sum[key_sum].str.split(r'|', expand=True)
        del df_sum[key_sum]
        for col in key:
            df_sum[col] = pd.to_numeric(df_sum[col])
        return df_sum
    elif type(key) is str:
        df[col_sum] = df[col_sum].astype(str) + r'|'
        df = df.groupby(by=[key], as_index=False).agg({col_sum: 'sum'})
        df[col_sum] = df[col_sum].str.rstrip('|')
        if df[key].dtypes != df_sum[key].dtypes:
            if df_sum[key].dtypes == float:
                df[key] = df[key].astype(float)
            elif df_sum[key].dtypes == int:
                df[key] = df[key].astype(int)
            elif df_sum[key].dtypes == object:
                df[key] = df[key].astype(object)
        df_sum = df_sum.merge(df, how='left', left_on=[key], right_on=[key])
        col_sum_x, col_sum_y = sorted([x for x in df_sum.columns if col_sum in str(x)])
        for col in [col_sum_x, col_sum_y]:
            df_sum[col] = df_sum[col].str.strip('|na')
        del df_sum[col_sum_x]
        df_sum.rename(columns={col_sum_y: col_sum}, inplace=True)
        return df_sum


class SyncDir:
    """Класс синхронизации двух папок ищет разницу в файлах.
    1. Добавляет файлы, которые есть в path1, но нет в path2
    2. Удаляет файлы, которые есть в path2, но нет в path1
    3. Чистит пустые подпапки в path2"""

    def __init__(self, path1: str, path2: str):
        self.path1 = Path(path1)
        self.path2 = Path(path2)
        self.relative_files1 = None
        self.relative_files2 = None
        self.move_files = []
        self.delete_files = []
        self.empty_dirs = set()

    @staticmethod
    def copy_files_bytes(src, dst):
        """Побитное копирование файлов с созданием необходимых каталогов"""
        if not dst.parent.is_dir():
            dst.parent.mkdir(parents=True, exist_ok=True)
        if src.is_file():
            if not dst.is_file():
                try:
                    dst.write_bytes(src.read_bytes())
                    return True
                except Exception as e:
                    print(e)
                    return False
            else:
                return False

    @staticmethod
    def get_all_paths(path: str | Path, is_absolute: bool = False) -> set[str]:
        """
        Получаем все пути до файлов в папке и подпапках
        :param path: str | Path путь до папки
        :param is_absolute: флаг устанавливающий нужно получить абсолютный или относительный путь
        :return: set путей до файлов внутри
        """
        if not isinstance(path, Path):
            path = Path(path)
        files_paths = set()
        for root, dirs, files in os.walk(os.path.abspath(path)):
            for file in files:
                if is_absolute:
                    files_paths.add(os.path.join(root, file))
                else:
                    fsplit = f"{path.name}\\"
                    tmp = root.split(fsplit)
                    if len(tmp) > 1:
                        tmp = tmp[-1]
                    else:
                        tmp = ''
                    files_paths.add(os.path.join(tmp, file))
        return files_paths

    @staticmethod
    def get_empty_dirs(path: str | Path) -> set[Path]:
        """Получаем пути до пустых подпапок в заданной папке"""
        empty_dirs = set()
        if not isinstance(path, Path):
            path = Path(path)
        for dirpath, dirnames, filenames in os.walk(path):
            for dirname in dirnames:
                dr = Path(os.path.join(dirpath, dirname))
                if not any(dr.iterdir()):
                    empty_dirs.add(dr)

        return empty_dirs

    def set_files(self):
        """Устанавливаем относительные пути до файлов"""
        self.relative_files1 = self.get_all_paths(self.path1, is_absolute=False)
        self.relative_files2 = self.get_all_paths(self.path2, is_absolute=False)

    def set_sync_files(self):
        """Создаёт множество файлов которые нужно перенести в path2 и удалить из path2"""
        self.move_files = []
        movef = self.relative_files1 - self.relative_files2
        deletef = self.relative_files2 - self.relative_files1
        for f in movef:
            self.move_files.append([Path(os.path.join(self.path1, f)), Path(os.path.join(self.path2, f))])
        for f in deletef:
            self.delete_files.append(Path(os.path.join(self.path2, f)))

    def sync(self):
        self.set_files()
        self.set_sync_files()

        if len(self.move_files) > 0:
            count = 0
            for fme, fms in self.move_files:
                _ = self.copy_files_bytes(src=fme, dst=fms)
                if _:
                    count += 1
            print('=' * 60)
            print(f"Файлы перенесены из {self.path1} в {self.path2}")
            print(f"Общее количество файлов подлежащих синхронизации = {len(self.move_files)}")
            print(f"Общее количество перенесённых файлов = {count}")
            print(f"Общее количество не перенесённых файлов = {len(self.move_files) - count}")
            print('=' * 60)
        if len(self.delete_files) > 0:
            for fd in self.delete_files:
                fd.unlink(missing_ok=True)
            print('=' * 60)
            print(f"Файлы удалены в {self.path2}")
            print(f"Общее количество файлов подлежащих удалению = {len(self.delete_files)}")
            print('=' * 60)

        # после синхронизации файлов удаляем пустые подпапки
        self.empty_dirs = self.get_empty_dirs(self.path2)

        if len(self.empty_dirs) > 0:
            for ed in self.empty_dirs:
                ed.rmdir()


class SqlQuery:
    """Класс запросов SQL"""

    schema = 'dbo'
    method = 'replace'

    def __init__(self, sql: str):
        self.sql = sql
        self.table_pbi = None

    @classmethod
    def get_schema(cls):
        return cls.schema

    @classmethod
    def get_method(cls):
        return cls.method

    def get(self):
        return self.sql

    def set_table(self, value):
        if isinstance(value, str):
            self.table_pbi = value
        else:
            raise TypeError('Тип таблицы не строка!')

    def get_table(self):
        if isinstance(self.table_pbi, str):
            return self.table_pbi
        else:
            raise TypeError('Установите название таблицы!')


def get_db_table(sql_txt):
    """Получаем таблицу из БД
    example 'mssql+pyodbc://{server}/{database}?driver=SQL+Server&trusted_connection=yes'
    """
    if os.path.basename(sys.argv[0]).__contains__('ipykernel_launcher'):
        load_dotenv('path/to/.env', override=True, encoding='cp1251')
    else:
        load_dotenv()

    # engine
    engine_db = create_engine(os.getenv('DB_CONNECT'))
    df_source = pd.read_sql(sql_txt, engine_db.connect())
    engine_db.dispose()
    engine_db.close()
    return df_source


def uniq_to_tuple_lst(df, col: str):
    """
    Изменяем формат для поиска в БД oracle, чтобы обойти ограничение в 2500
    :param df: таблица
    :param col: название столбца
    :return: формат выхода((x0, 0), (x1, 0))
    """
    return tuple([(int(x), 0) for x in df[df[col].notna()][col].unique()])


def set_calendar(df):
    """Таблица календарь по min и max дате из таблицы"""
    date_cols = [x for x in df.columns if str(df[x].dtypes).__contains__('datetime')]
    rasch_date_min = df[date_cols[0]].min()
    rasch_date_max = df[date_cols[0]].max()
    for col in date_cols:
        tmp_date_min = df[col].min()
        tmp_date_max = df[col].max()
        if tmp_date_min < rasch_date_min:
            rasch_date_min = tmp_date_min
        if tmp_date_max > rasch_date_max:
            rasch_date_max = tmp_date_max
    df_cal = pd.DataFrame({'date': pd.date_range(rasch_date_min, rasch_date_max)})
    return df_cal, date_cols


def create_shortcut(file_name: str, target: str, work_dir: str, arguments: str = ''):
    """
    Создаём ярлык нужного нам файла
    :param file_name:
    :param target:
    :param work_dir:
    :param arguments:
    :return:
    """
    shell = Dispatch('WScript.Shell')
    shortcut = shell.CreateShortCut(file_name)
    shortcut.TargetPath = target
    shortcut.Arguments = arguments
    shortcut.WorkingDirectory = work_dir
    shortcut.save()


def enable_download_in_headless_chrome(br, download_dir):
    # добавляет в опции путь к сохранению файлов по умолчанию
    br.command_executor._commands["send_command"] = ("POST", '/session/$sessionId/chromium/send_command')

    params = {'cmd': 'Page.setDownloadBehavior', 'params': {'behavior': 'allow', 'downloadPath': download_dir}}
    command_result = br.execute("send_command", params)
    for key in command_result:
        print("result:" + key + ":" + str(command_result[key]))
    return br


def run_browser(path_for_downloads, userAgent: str, webdriver_manager: str):
    # создаёт экземпляр драйвера
    chrome_options = Options()
    #     chrome_options.add_argument('--headless')
    chrome_options.add_argument('user-agent={0}'.format(userAgent))
    chrome_options.add_argument("--window-size=1920x1080")
    prefs = {'download.default_directory': path_for_downloads, "download.prompt_for_download": False}
    chrome_options.add_experimental_option("prefs", prefs)
    driver = webdriver.Chrome(service=Service(service=Service(ChromeDriverManager(path=webdriver_manager).install()),
                                              options=chrome_options))
    driver = enable_download_in_headless_chrome(driver, path_for_downloads)
    # запуск браузера
    return driver


class FileEncoding:
    """Получаем возможную кодировку файла и язык"""

    def __init__(self, path: Path):
        if not isinstance(path, Path):
            try:
                path = Path(path)
            except TypeError:
                raise TypeError(f"path not convertible to Path: {path}")

        if path.is_file():
            self.path = path
        else:
            raise FileNotFoundError('File not found')
        self.encoding = 'utf-8'  # кодировка файла по умолчанию utf-8
        self.language = ''  # язык файла
        self.confidence = None  # насколько правильно угадана кодировка
        self.set_ec_vars()

    def set_ec_vars(self):
        detector = UniversalDetector()
        with open(self.path, 'rb') as fh:
            for line in fh:
                detector.feed(line)
                if detector.done:
                    break
            detector.close()
        self.encoding = detector.result['encoding']
        self.language = detector.result['language']
        self.confidence = detector.result['confidence']


class XMLMap:
    """
    Класс для создания карты ключей к каким элементам можно обращаться в xml и какие элементы в нём существуют.
    Смысл найти возможность для обращения к конкретному элементу в xml по ключу,
    не дописан!
    """

    def __init__(self, xml):
        self.xml = xml

    @staticmethod
    def check_key(d: dict, key: str):
        """Проверка существования вложенного ключа в словаре, в целях безопасности добавлена проверка входных данных
        d - это исходный словарь (type = dict) со всеми вложенными ключами
        key - это вычисленный элемент карты который предположительно является ключом вида [key1][key2][key3] или
        [key1][key2] или [key1]"""
        if isinstance(d, dict):
            dd = True
        else:
            dd = False
        if isinstance(key, str) and key[0] == '[' and key[-1] == ']':
            iskey = True
        else:
            iskey = False

        if dd and iskey:
            check = f"{d}{key}"
            try:
                eval(check)
                return True
            except (TypeError, KeyError):
                return False
        else:
            if not dd and iskey:
                raise TypeError(f"d не является словарём {type(d)}")
            elif not dd and not iskey:
                raise ValueError(f"Вы подали что то не то, type(d) = {type(d)}, type(key) = {type(key)}, "
                                 f"key[0] == {key[0]}, key[-1] = {key[-1]}, ключ должен быть вида "
                                 f"'[key1]' или '[key1][key2]'")
            elif dd and not iskey:
                raise ValueError(f"Не правильный key. Проверьте type(key) = {type(key)}, key[0] == {key[0]}, "
                                 f"key[-1] = {key[-1]}, ключ должен быть вида '[key1]' или '[key1][key2]'")
            else:
                raise ValueError(f"Не понятно что подано type(key) = {type(key)}, key[0] == {key[0]}, "
                                 f"key[-1] = {key[-1]}")

    @staticmethod
    def recursive_ident_dict(attrs: dict,
                             attrs1: dict,
                             indent: list,
                             _add: bool = False,
                             _ladd: bool = False):
        """Рекурсивный обход полученного словаря
        attrs - исследуемый словарь, который при рекурсии погружается на вложенные словари
        attrs1 - наш исследуемый словарь без изменений для проверки существования ключей
        indent - список глубины с ключами вида ["[key1]", "[key1][key2]", "[key1][key2][key3]", "[key4]"]
        _add - признак добавлен ли ключ значение которого не является словарём
        _ladd - признак добавления родительского ключа

        Например из словаря:

        a = {
            '1': {
                '2': {
                    '3':
                      {
                          '4': {5},
                          '6': {
                              '7': {
                                  '8': {9}
                              }
                          },
                       },
                    '10': {
                        '11': '12',
                        '13': 14,
                        '15': 16.
                    }

                    },
                '28': 29
            },
            '17': {
                '18':{
                    '19': [20, '21', 22.],
                    '23': (24, '25')
                }
            },
            '26': '27'
        }

        Получается карта вида

        ["['1']",
         "['1']['2']",
         "['1']['2']['3']",
         "['1']['2']['3']['4']",
         "['1']['2']['3']['6']",
         "['1']['2']['3']['6']['7']",
         "['1']['2']['3']['6']['7']['8']",
         "['1']['2']['10']",
         "['1']['2']['10']['11']",
         "['1']['2']['10']['13']",
         "['1']['2']['10']['15']",
         "['1']['28']",
         "['17']",
         "['17']['18']",
         "['17']['18']['19']",
         "['17']['18']['23']",
         "['26']"]

        Ещё пример с одинаковыми вложенными ключами
        b = {
            '1': {
                1: {
                    tuple([1, 2]): {
                        '1': {
                            '1': {
                                '1': {1}
                            }
                        }
                    }
                }
            }
        }

        Выход:
        ["['1']",
         "['1'][1]",
         "['1'][1][(1, 2)]",
         "['1'][1][(1, 2)]['1']",
         "['1'][1][(1, 2)]['1']['1']",
         "['1'][1][(1, 2)]['1']['1']['1']"]
        """
        if getattr(attrs, 'items', False):
            for k, v in attrs.items():
                # изменяем представление ключа
                key = f"['{k}']" if isinstance(k, str) else f"[{k}]"
                # проверяем является ли значение словарём
                if getattr(v, 'items', False) is False:
                    if len(indent) > 0:
                        last = indent[-1] + key
                    else:
                        last = key

                    # если по ключу можно получить значение добавляем его как отдельный элемент карты
                    if XMLMap.check_key(attrs1, last):
                        indent.append(last)
                    else:
                        # Если значение получить не удаётся в случаи как несколько вложенных ключей,
                        # тогда перебираем все ключи из карты, важно карта должна быть перевёрнута, т.е. сначала
                        # будем искать совпадение во вложенных ключах, в противном случае поднимаемся на уровень
                        # выше
                        append = False
                        indent1 = sorted(indent, reverse=True)
                        for i in indent1:
                            last1 = i + key
                            if XMLMap.check_key(attrs1, last1):
                                append = True
                                indent.append(last1)
                                break
                        if not append:
                            # если ключ не удалось найти во вложенных словарях тогда он находится на одном уровне с
                            # начальным, поэтому добавляем его в карту
                            indent.append(key)
                    _add = True
                    # если значение ключа не словарь дальнейшая проверка не требуется
                    continue

                if _add is True:
                    # Если предыдущий проход по ключам дал значение не являющиеся словарём, соответственно мы
                    # находимся внутри словаря с несколькими ключами, поэтому принимаем всю часть из карты
                    # предидущего значения без учёта последнего ключа.
                    last = '['.join(indent[-1].split('[')[:-1])
                    last += key
                    _ladd = True
                    _add = False
                else:
                    if _ladd:
                        last = ''.join(indent[-1])
                        last += key
                    else:
                        last = key

                # проверка существования сформированных вложенных ключей
                if len(indent) > 0:
                    # изначальная проверка существования вложенного идентичного ключа
                    checkkey = ''.join(indent[-1]) + last
                    if XMLMap.check_key(attrs1, checkkey):
                        indent.append(checkkey)
                    # если вложенного такого ключа не существует тогда проверка на существования самого ключа
                    elif XMLMap.check_key(attrs1, last):
                        indent.append(last)
                    # если не существует сам ключ, обходим уровни выше
                    else:
                        indent1 = sorted(indent, reverse=True)
                        for i in indent1:
                            last1 = i + last
                            if XMLMap.check_key(attrs1, last1):
                                indent.append(last1)
                                break
                else:
                    # соответственно если начальной карты ещё нет принимаем первый ключ
                    indent.append(last)
                XMLMap.recursive_ident_dict(v, attrs1, indent, _add=_add, _ladd=_ladd)

        return indent

    @staticmethod
    def get_id(ch):
        idx = ch.attrib.get('id', None)
        if idx is not None:
            key = f"{ch.tag}_{idx}"
        else:
            key = ch.tag
        return key


class DfTypesClick:
    """
    Класс составления типов столбцов для хранения таблиц в БД clickhouse.
    Смысл в более экономичном выборе типа данных при создании таблиц в clickhouse
    """

    def __init__(self):
        self.df_types = None
        self.click_types = {
            'int': {
                'Int8': [-128, 127],
                'Int16': [-32768, 32767],
                'Int32': [-2147483648, 2147483647],
                'Int64': [-9223372036854775808, 9223372036854775807],
                # 'Int128': [-170141183460469231731687303715884105728,
                #            170141183460469231731687303715884105727],
                # 'Int256': [-57896044618658097711785492504343953926634992332820282019728792003956564819968,
                #            57896044618658097711785492504343953926634992332820282019728792003956564819967],
            },
            'uint': {
                'UInt8': [0, 255],
                'UInt16': [0, 65535],
                'UInt32': [0, 4294967295],
                'UInt64': [0, 18446744073709551615],
                # 'UInt128': [0, 340282366920938463463374607431768211455],
                # 'UInt256': [0, 115792089237316195423570985008687907853269984665640564039457584007913129639935]
            },
            'float': {
                'float32': 'FLOAT',
                'float64': 'DOUBLE',
                'decimal': ['Decimal32(S)', 'Decimal64(S)']
            },
            'bool': 'Bool',
            'string': ['String', 'FixedString'],
            'datetime64': ['DateTime', 'DateTime64']
        }

    def set_df_types(self, df):
        """Создание минимально возможных типов данных для таблицы"""
        df = df.copy()
        df = df.convert_dtypes()
        for col in df.columns:
            t = df[col].dtype
            if str(t).lower().startswith('float'):
                df[col] = pd.to_numeric(df[col], downcast='float')
            if str(t).lower().startswith('int'):
                df[col] = pd.to_numeric(df[col], downcast='integer')

        df_types = {k: '' for k in list(df.columns)}
        for col in df.columns:
            t = str(df[col].dtype).lower()
            if t.startswith(('string', 'object')):
                mins = df[col].str.len().min()
                maxs = df[col].str.len().max()
                means = df[col].str.len().mean()
                if mins == maxs == means:
                    df_types[col] = f"{self.click_types['string'][1]}({mins})"
                else:
                    df_types[col] = f"{self.click_types['string'][0]}({mins})"
            elif t.startswith(('bool')):
                df_types[col] = self.click_types[t]
            elif t.startswith(('float')):
                t = str(t).lower()
                df_types[col] = f"{self.click_types['float'][t]}"
            elif t.startswith(('datetime')):
                ms = df[col].dt.microsecond.sum()
                if ms > 0:
                    df_types[col] = self.click_types['datetime64'][1]
                else:
                    df_types[col] = self.click_types['datetime64'][0]
            elif t.startswith(('int')):
                mini = df[col].min()
                maxi = df[col].max()
                if mini < 0:
                    for k, v in self.click_types['int'].items():
                        if v[0] <= mini and v[1] >= maxi:
                            df_types[col] = k
                            break
                else:
                    for k, v in self.click_types['uint'].items():
                        if v[0] <= mini and v[1] >= maxi:
                            df_types[col] = k
                            break

        self.df_types = df_types


class WorkCMD:
    """
    Класс для использования командной строки из python.
    """
    COMMAND_LEN = 8191

    @staticmethod
    def run_cmd(c):
        """
        Исполнение команды и вывод обратной информации
        :param c: str команда
        :return: tuple(dt str - дата и время до секунды,
                       c str - команда,
                       code int - код 0 выполнился без ошибок
                                  код 1 с ошибками
                                  код 206 не выполнился из-за ограничения командной строки
                       stdout str - вывод выполнения команды в cmd,
                       error str - ошибка если она есть иначе '')
        """
        if len(c) > WorkCMD.COMMAND_LEN:
            code1 = 206
            stdout1 = f"Длина команды превышает максимально допустимый предел в {WorkCMD.COMMAND_LEN} символов"
            error1 = "[WinError 206]"
            dt1 = datetime.datetime.strftime(datetime.datetime.now(), '%d.%m.%Y %H:%M:%S')
        else:
            res = subprocess.run(c, shell=True, capture_output=True)
            dt1 = datetime.datetime.strftime(datetime.datetime.now(), '%d.%m.%Y %H:%M:%S')
            stdout1 = res.stdout.decode(locale.getpreferredencoding())
            error1 = res.stderr.decode(locale.getpreferredencoding())
            code1 = res.returncode
        return dt1, c, code1, stdout1, error1

    @staticmethod
    def pack_command(args: list):
        """
        Упаковываем команды в строку
        :param args: list список аргументов
        :return: str строка подготовленной команды
        """
        args = [str(x) for x in args]
        pack = ''
        for idx, arg in enumerate(args):
            if "/" in arg or "\\" in arg:
                # если аргумент является путём, то засовываем его в кавычки
                if not arg.startswith('\"'):
                    arg = '"' + arg
                if not arg.endswith('\"'):
                    arg = arg + '"'
            if idx + 1 == len(args):
                pack += arg
            else:
                pack += f"{arg} "
        return pack