Tools

Общий пакет для готовых функций. Модули в этом пакете:

analysis: Позволяет находить все jpeg-файлы в директории.
download: Позволяет скачать и разархивировать датасет.
parser: Парсер сайта госкаталог.
notebook: Позволяет обновлять идентификаторы ячеек в ноутбуке.
logger_config: Позволяет настроить логгер для сервера и клиента.

display_images

display_images(image_paths, subtitle='', images_per_row=5)

Показывает изображения в графическом окне

PARAMETER	DESCRIPTION
`images_per_row`	Сколько изображений на одной строчке TYPE: `int` DEFAULT: `5`
`image_paths`	Список путей к изображениям TYPE: `list[str]`
`subtitle`	Подзаголовок TYPE: `str` DEFAULT: `''`

RETURNS	DESCRIPTION
`None`	None

Source code in Tools/analysis.py

def display_images(image_paths: list[str], subtitle: str = "", images_per_row: int = 5) -> None:
    """
    Показывает изображения в графическом окне
    :param images_per_row: Сколько изображений на одной строчке
    :param image_paths: Список путей к изображениям
    :param subtitle: Подзаголовок
    :return: None
    """
    image_count = len(image_paths)

    for start in range(0, image_count, images_per_row):
        end = min(start + images_per_row, image_count)
        fig, axs = plt.subplots(1, end - start, figsize=(15, 5))
        fig.suptitle(subtitle)

        axs = axs if isinstance(axs, Iterable) else [axs]
        for i, ax in enumerate(axs):
            img_path = image_paths[start + i]
            img_name = get_image_json_info(img_path.split(".")[0] + ".json").get("name", "")
            img_name = (img_name[:27] + "...") if img_name and len(img_name) > 30 else img_name

            ax.set_title(img_name)
            ax.imshow(Image.open(img_path))
            ax.axis("off")
        plt.show()

find_image_files

find_image_files(directory)

Получение списка путей к jpeg-файлам в нужной директории

PARAMETER	DESCRIPTION
`directory`	Путь к папке поиска TYPE: `Path`

RETURNS	DESCRIPTION
`list[str]`	Список путей к jpeg-файлам

Source code in Tools/analysis.py

def find_image_files(directory: Path) -> list[str]:
    """
    Получение списка путей к jpeg-файлам в нужной директории
    :param directory: Путь к папке поиска
    :return: Список путей к jpeg-файлам
    """
    jpeg_files = []
    for root, _, files in os.walk(directory):
        for file in files:
            if file.endswith(("jpg", "JPG", "jpeg", "tif")):
                full_path = os.path.join(root, file)
                jpeg_files.append(full_path)
    return jpeg_files

find_needed_jpeg_files

find_needed_jpeg_files(directory, target_names)

Поиск нужных jpeg-файлов в нужной директории по их id имени

PARAMETER	DESCRIPTION
`directory`	Путь к папке поиска TYPE: `Path`
`target_names`	Список имен для поиска TYPE: `list[str \| int]`

RETURNS	DESCRIPTION
`list[str]`	Список путей к jpeg-файлам

Source code in Tools/analysis.py

def find_needed_jpeg_files(directory: Path, target_names: list[str | int]) -> list[str]:
    """
    Поиск нужных jpeg-файлов в нужной директории по их id имени
    :param directory: Путь к папке поиска
    :param target_names: Список имен для поиска
    :return: Список путей к jpeg-файлам
    """
    if not isinstance(target_names[0], str):
        target_names = [str(name) for name in target_names]
    jpeg_files = find_image_files(directory)
    needed_jpeg_files = []
    for jpeg_file in jpeg_files:
        for name in target_names:
            if name in jpeg_file:
                needed_jpeg_files.append(jpeg_file)

    return needed_jpeg_files

download_zip

download_zip(url, path)

Скачивает zip-файл по ссылке и возвращает путь до него

PARAMETER	DESCRIPTION
`url`	ссылка на zip-файл с Yandex.Disk TYPE: `str`
`path`	место, куда нужно сохранить zip-файл TYPE: `str \| Path`

RETURNS	DESCRIPTION
`Path`	путь до загруженного разархивированного zip-файла

Source code in Tools/download.py

def download_zip(url: str, path: str | Path) -> Path:
    """Скачивает zip-файл по ссылке и возвращает путь до него
    :param url: ссылка на zip-файл с Yandex.Disk
    :param path: место, куда нужно сохранить zip-файл
    :return: путь до загруженного разархивированного zip-файла
    """
    response = requests.get(url, timeout=20)
    path = Path(path)
    return extract_zip(response.content, path)

extract_zip

extract_zip(zip_file, destination_directory)

Извлекает содержимое zip-файла в папку destination_directory

PARAMETER	DESCRIPTION
`zip_file`	файл с zip-архивом TYPE: `bytes`
`destination_directory`	место разархивирования TYPE: `Path`

RETURNS	DESCRIPTION
`Path`	место разархивирования

Source code in Tools/download.py

def extract_zip(zip_file: bytes, destination_directory: Path) -> Path:
    """Извлекает содержимое zip-файла в папку destination_directory
    :param zip_file: файл с zip-архивом
    :param destination_directory: место разархивирования
    :return: место разархивирования
    """
    destination_directory.mkdir(parents=True, exist_ok=True)

    with zipfile.ZipFile(io.BytesIO(zip_file)) as file:
        file.extractall(destination_directory)
    return destination_directory

get_ya_disk_url

get_ya_disk_url(public_key)

Возвращает ссылку на скачивание файла с https://disk.yandex.ru по публичной общей ссылке

PARAMETER	DESCRIPTION
`public_key`	публичная ссылка на файл TYPE: `str`

RETURNS	DESCRIPTION
`str`	ссылка на скачивание

Source code in Tools/download.py

def get_ya_disk_url(public_key: str) -> str:
    """Возвращает ссылку на скачивание файла с https://disk.yandex.ru по публичной общей ссылке
    :param public_key: публичная ссылка на файл
    :return: ссылка на скачивание
    """
    base_url = "https://cloud-api.yandex.net/v1/disk/public/resources/download?"

    final_url = base_url + urlencode({"public_key": public_key})
    response = requests.get(final_url, timeout=20)
    download_url = response.json()["href"]
    return download_url

configure_client_logging

configure_client_logging(log_folder=LOG_FOLDER)

Создание логгера для клиента

Source code in Tools/logger_config.py

def configure_client_logging(log_folder=LOG_FOLDER):
    """Создание логгера для клиента"""
    setup_logger(os.path.join(log_folder, "client.log"))

configure_server_logging

configure_server_logging(log_folder=LOG_FOLDER)

Создание логгера для сервера

Source code in Tools/logger_config.py

def configure_server_logging(log_folder=LOG_FOLDER):
    """Создание логгера для сервера"""
    setup_logger(os.path.join(log_folder, "server.log"))
    # Удаляем все существующие хэндлеры
    for name in logging.root.manager.loggerDict.keys():
        logging.getLogger(name).handlers = []
        logging.getLogger(name).propagate = True

    logging.basicConfig(
        handlers=[InterceptHandler()], level=logging.INFO  # Добавляем наш перехватчик в корневой логгер
    )
    sys.stderr = InterceptHandler()  # не все библиотеки используют логгеры. Sqlalchemy.exc использует sys.stderr

set_cell_id

set_cell_id(notebook_path)

Обновляет идентификаторы ячеек в файле ноутбука от 1 до N

PARAMETER	DESCRIPTION
`notebook_path`	путь к файлу ipynb TYPE: `str \| Path`

RETURNS	DESCRIPTION
`int`	1 - файл обновлен

Source code in Tools/notebook.py

def set_cell_id(notebook_path: str | Path) -> int:
    """
    Обновляет идентификаторы ячеек в файле ноутбука от 1 до N
    :param notebook_path: путь к файлу ipynb
    :return: 1 - файл обновлен
    """
    if isinstance(notebook_path, str):
        notebook_path = Path(notebook_path)

    with open(notebook_path, encoding="utf-8") as f_in:
        doc = json.load(f_in)
    cnt = 1

    for cell in doc["cells"]:
        if "execution_count" in cell:
            cell["execution_count"] = cnt

            for o in cell.get("outputs", []):
                if "execution_count" in o:
                    o["execution_count"] = cnt

            cnt += 1

    with open(notebook_path, "w", encoding="utf-8") as f_out:
        json.dump(doc, f_out, indent=1, ensure_ascii=False)
        f_out.write("\n")

    print(f"Notebook {notebook_path} updated")
    return 1

goskatalog_parser

goskatalog_parser(classes, path)

Скачивает картины с сайта госкаталог и возвращает словарь с результатами парсинга

PARAMETER	DESCRIPTION
`classes`	список скачиваемых категорий TYPE: `tuple`
`path`	место куда будут сохраняться картины TYPE: `Path`

RETURNS	DESCRIPTION
`(Path, list)`	(путь к папке с картинами, словарь с пропущенными картинами)

Source code in Tools/parser.py

def goskatalog_parser(classes: tuple, path: Path) -> (Path, list):
    """Скачивает картины с сайта госкаталог и возвращает словарь с результатами парсинга
    :param classes: список скачиваемых категорий
    :param path: место куда будут сохраняться картины
    :return: (путь к папке с картинами, словарь с пропущенными картинами)
    """
    missing_art = []

    session = requests.Session()
    session.headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:131.0) Gecko/20100101 Firefox/131.0",
        "Accept-Language": "en-GB,ru;q=0.7,en;q=0.3",
    }
    post_count_params = {
        "statusIds": 6,
        "publicationLimit": "false",
        "cacheEnabled": "true",
        "calcCountsType": 1,
        "limit": 0,
        "offset": 0,
    }
    post_url = "https://goskatalog.ru/muzfo-rest/rest/exhibits/ext"

    path /= "art_dataset"

    for class_name in classes:
        path_class = path / class_name
        path_class.mkdir(parents=True, exist_ok=True)
        json_params = [
            {
                "fieldName": "name",
                "fieldType": "String",
                "operator": "CONTAINS",
                "fromValue": "null",
                "toValue": "null",
                "value": class_name,
            },
            {
                "fieldName": "typologyId",
                "fieldType": "Number",
                "operator": "EQUALS",
                "fromValue": "null",
                "toValue": "null",
                "value": 1,
            },
        ]

        count_response = session.post(post_url, params=post_count_params, json=json_params)
        art_count = count_response.json().get("statistics")[0].get("count")

        for i in tqdm(range(0, art_count, 100), desc=f"Downloading {class_name}"):
            post_params = {
                "statusIds": 6,
                "publicationLimit": "false",
                "cacheEnabled": "true",
                "calcCountsType": 0,
                "dirFields": "desc",
                "limit": 100,
                "offset": i,
                "sortFields": "id",
            }
            response = session.post(post_url, params=post_params, json=json_params)
            for art in response.json().get("objects"):
                try:
                    download_art(art, path_class, session)
                except (requests.exceptions.HTTPError, IndexError, AttributeError):
                    missing_art.append({art.get("id"), art.get("name")})

    return path, missing_art

zip_files

zip_files(path)

Создает zip-файл из папки с картинками

PARAMETER	DESCRIPTION
`path`	путь к папке с картинками TYPE: `Path`

RETURNS	DESCRIPTION
`str`	путь к zip-файлу

Source code in Tools/parser.py

def zip_files(path: Path) -> str:
    """Создает zip-файл из папки с картинками
    :param path: путь к папке с картинками
    :return: путь к zip-файлу
    """
    return shutil.make_archive(path.name, "zip", path)

Tools.analysis

find_image_files

find_image_files(directory)

Получение списка путей к jpeg-файлам в нужной директории

PARAMETER	DESCRIPTION
`directory`	Путь к папке поиска TYPE: `Path`

RETURNS	DESCRIPTION
`list[str]`	Список путей к jpeg-файлам

Source code in Tools/analysis.py

def find_image_files(directory: Path) -> list[str]:
    """
    Получение списка путей к jpeg-файлам в нужной директории
    :param directory: Путь к папке поиска
    :return: Список путей к jpeg-файлам
    """
    jpeg_files = []
    for root, _, files in os.walk(directory):
        for file in files:
            if file.endswith(("jpg", "JPG", "jpeg", "tif")):
                full_path = os.path.join(root, file)
                jpeg_files.append(full_path)
    return jpeg_files

find_needed_jpeg_files

find_needed_jpeg_files(directory, target_names)

Поиск нужных jpeg-файлов в нужной директории по их id имени

PARAMETER	DESCRIPTION
`directory`	Путь к папке поиска TYPE: `Path`
`target_names`	Список имен для поиска TYPE: `list[str \| int]`

RETURNS	DESCRIPTION
`list[str]`	Список путей к jpeg-файлам

Source code in Tools/analysis.py

def find_needed_jpeg_files(directory: Path, target_names: list[str | int]) -> list[str]:
    """
    Поиск нужных jpeg-файлов в нужной директории по их id имени
    :param directory: Путь к папке поиска
    :param target_names: Список имен для поиска
    :return: Список путей к jpeg-файлам
    """
    if not isinstance(target_names[0], str):
        target_names = [str(name) for name in target_names]
    jpeg_files = find_image_files(directory)
    needed_jpeg_files = []
    for jpeg_file in jpeg_files:
        for name in target_names:
            if name in jpeg_file:
                needed_jpeg_files.append(jpeg_file)

    return needed_jpeg_files

display_images

display_images(image_paths, subtitle='', images_per_row=5)

Показывает изображения в графическом окне

PARAMETER	DESCRIPTION
`images_per_row`	Сколько изображений на одной строчке TYPE: `int` DEFAULT: `5`
`image_paths`	Список путей к изображениям TYPE: `list[str]`
`subtitle`	Подзаголовок TYPE: `str` DEFAULT: `''`

RETURNS	DESCRIPTION
`None`	None

Source code in Tools/analysis.py

def display_images(image_paths: list[str], subtitle: str = "", images_per_row: int = 5) -> None:
    """
    Показывает изображения в графическом окне
    :param images_per_row: Сколько изображений на одной строчке
    :param image_paths: Список путей к изображениям
    :param subtitle: Подзаголовок
    :return: None
    """
    image_count = len(image_paths)

    for start in range(0, image_count, images_per_row):
        end = min(start + images_per_row, image_count)
        fig, axs = plt.subplots(1, end - start, figsize=(15, 5))
        fig.suptitle(subtitle)

        axs = axs if isinstance(axs, Iterable) else [axs]
        for i, ax in enumerate(axs):
            img_path = image_paths[start + i]
            img_name = get_image_json_info(img_path.split(".")[0] + ".json").get("name", "")
            img_name = (img_name[:27] + "...") if img_name and len(img_name) > 30 else img_name

            ax.set_title(img_name)
            ax.imshow(Image.open(img_path))
            ax.axis("off")
        plt.show()

get_image_json_info

get_image_json_info(path)

Извлечение информации об изображении из JSON-файла

PARAMETER	DESCRIPTION
`path`	Путь к изображению TYPE: `str`

RETURNS	DESCRIPTION
`dict`	Словарь с информацией о JSON-файле

Source code in Tools/analysis.py

def get_image_json_info(path: str) -> dict:
    """
    Извлечение информации об изображении из JSON-файла
    :param path: Путь к изображению
    :return: Словарь с информацией о JSON-файле
    """
    try:
        with open(path, encoding="ascii") as file:
            return json.load(file)
    except (UnicodeDecodeError, FileNotFoundError):
        return {}

Tools.parser

goskatalog_parser

goskatalog_parser(classes, path)

Скачивает картины с сайта госкаталог и возвращает словарь с результатами парсинга

PARAMETER	DESCRIPTION
`classes`	список скачиваемых категорий TYPE: `tuple`
`path`	место куда будут сохраняться картины TYPE: `Path`

RETURNS	DESCRIPTION
`(Path, list)`	(путь к папке с картинами, словарь с пропущенными картинами)

Source code in Tools/parser.py

def goskatalog_parser(classes: tuple, path: Path) -> (Path, list):
    """Скачивает картины с сайта госкаталог и возвращает словарь с результатами парсинга
    :param classes: список скачиваемых категорий
    :param path: место куда будут сохраняться картины
    :return: (путь к папке с картинами, словарь с пропущенными картинами)
    """
    missing_art = []

    session = requests.Session()
    session.headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:131.0) Gecko/20100101 Firefox/131.0",
        "Accept-Language": "en-GB,ru;q=0.7,en;q=0.3",
    }
    post_count_params = {
        "statusIds": 6,
        "publicationLimit": "false",
        "cacheEnabled": "true",
        "calcCountsType": 1,
        "limit": 0,
        "offset": 0,
    }
    post_url = "https://goskatalog.ru/muzfo-rest/rest/exhibits/ext"

    path /= "art_dataset"

    for class_name in classes:
        path_class = path / class_name
        path_class.mkdir(parents=True, exist_ok=True)
        json_params = [
            {
                "fieldName": "name",
                "fieldType": "String",
                "operator": "CONTAINS",
                "fromValue": "null",
                "toValue": "null",
                "value": class_name,
            },
            {
                "fieldName": "typologyId",
                "fieldType": "Number",
                "operator": "EQUALS",
                "fromValue": "null",
                "toValue": "null",
                "value": 1,
            },
        ]

        count_response = session.post(post_url, params=post_count_params, json=json_params)
        art_count = count_response.json().get("statistics")[0].get("count")

        for i in tqdm(range(0, art_count, 100), desc=f"Downloading {class_name}"):
            post_params = {
                "statusIds": 6,
                "publicationLimit": "false",
                "cacheEnabled": "true",
                "calcCountsType": 0,
                "dirFields": "desc",
                "limit": 100,
                "offset": i,
                "sortFields": "id",
            }
            response = session.post(post_url, params=post_params, json=json_params)
            for art in response.json().get("objects"):
                try:
                    download_art(art, path_class, session)
                except (requests.exceptions.HTTPError, IndexError, AttributeError):
                    missing_art.append({art.get("id"), art.get("name")})

    return path, missing_art

download_art

download_art(art, path, session)

Скачивает одну картину с сайта и сохраняет ее в папку вместе с JSON-файлом

PARAMETER	DESCRIPTION
`art`	словарь с данными картины TYPE: `dict`
`path`	путь к папке TYPE: `Path`
`session`	сессия для запросов TYPE: `Session`

Source code in Tools/parser.py

def download_art(art: dict, path: Path, session: requests.Session) -> None:
    """Скачивает одну картину с сайта и сохраняет ее в папку вместе с JSON-файлом
    :param art: словарь с данными картины
    :param path: путь к папке
    :param session: сессия для запросов
    """
    sleep(1)

    data_url = f"https://goskatalog.ru/muzfo-rest/rest/exhibits/{art.get('id')}"
    data_json = session.get(data_url).json()
    try:
        image_name = data_json.get("images")[0].get("fileName")
    except IndexError as e:
        image_name = art.get("mainImage").get("code")
        if not image_name:
            raise IndexError from e

    image_url = (
        f"https://goskatalog.ru/muzfo-imaginator/rest/images/original/{art.get('mainImage')['code']}"
        f"?originalName={image_name}"
    )
    image = session.get(image_url)
    image.raise_for_status()

    image_id = str(data_json.get("regNumber"))
    image_name = image_id + "." + image_name.split(".")[-1]  # jpeg, png, jpg

    path_image = path / image_name
    path_json = (path / image_id).with_suffix(".json")

    path_image.write_bytes(image.content)
    with open(path_json, "w", encoding="utf-8") as file:
        json.dump(data_json, file)

zip_files

zip_files(path)

Создает zip-файл из папки с картинками

PARAMETER	DESCRIPTION
`path`	путь к папке с картинками TYPE: `Path`

RETURNS	DESCRIPTION
`str`	путь к zip-файлу

Source code in Tools/parser.py

def zip_files(path: Path) -> str:
    """Создает zip-файл из папки с картинками
    :param path: путь к папке с картинками
    :return: путь к zip-файлу
    """
    return shutil.make_archive(path.name, "zip", path)

Tools.download

get_ya_disk_url

get_ya_disk_url(public_key)

Возвращает ссылку на скачивание файла с https://disk.yandex.ru по публичной общей ссылке

PARAMETER	DESCRIPTION
`public_key`	публичная ссылка на файл TYPE: `str`

RETURNS	DESCRIPTION
`str`	ссылка на скачивание

Source code in Tools/download.py

def get_ya_disk_url(public_key: str) -> str:
    """Возвращает ссылку на скачивание файла с https://disk.yandex.ru по публичной общей ссылке
    :param public_key: публичная ссылка на файл
    :return: ссылка на скачивание
    """
    base_url = "https://cloud-api.yandex.net/v1/disk/public/resources/download?"

    final_url = base_url + urlencode({"public_key": public_key})
    response = requests.get(final_url, timeout=20)
    download_url = response.json()["href"]
    return download_url

download_zip

download_zip(url, path)

Скачивает zip-файл по ссылке и возвращает путь до него

PARAMETER	DESCRIPTION
`url`	ссылка на zip-файл с Yandex.Disk TYPE: `str`
`path`	место, куда нужно сохранить zip-файл TYPE: `str \| Path`

RETURNS	DESCRIPTION
`Path`	путь до загруженного разархивированного zip-файла

Source code in Tools/download.py

def download_zip(url: str, path: str | Path) -> Path:
    """Скачивает zip-файл по ссылке и возвращает путь до него
    :param url: ссылка на zip-файл с Yandex.Disk
    :param path: место, куда нужно сохранить zip-файл
    :return: путь до загруженного разархивированного zip-файла
    """
    response = requests.get(url, timeout=20)
    path = Path(path)
    return extract_zip(response.content, path)

extract_zip

extract_zip(zip_file, destination_directory)

Извлекает содержимое zip-файла в папку destination_directory

PARAMETER	DESCRIPTION
`zip_file`	файл с zip-архивом TYPE: `bytes`
`destination_directory`	место разархивирования TYPE: `Path`

RETURNS	DESCRIPTION
`Path`	место разархивирования

Source code in Tools/download.py

def extract_zip(zip_file: bytes, destination_directory: Path) -> Path:
    """Извлекает содержимое zip-файла в папку destination_directory
    :param zip_file: файл с zip-архивом
    :param destination_directory: место разархивирования
    :return: место разархивирования
    """
    destination_directory.mkdir(parents=True, exist_ok=True)

    with zipfile.ZipFile(io.BytesIO(zip_file)) as file:
        file.extractall(destination_directory)
    return destination_directory

Tools.notebook

set_cell_id

set_cell_id(notebook_path)

Обновляет идентификаторы ячеек в файле ноутбука от 1 до N

PARAMETER	DESCRIPTION
`notebook_path`	путь к файлу ipynb TYPE: `str \| Path`

RETURNS	DESCRIPTION
`int`	1 - файл обновлен

Source code in Tools/notebook.py

def set_cell_id(notebook_path: str | Path) -> int:
    """
    Обновляет идентификаторы ячеек в файле ноутбука от 1 до N
    :param notebook_path: путь к файлу ipynb
    :return: 1 - файл обновлен
    """
    if isinstance(notebook_path, str):
        notebook_path = Path(notebook_path)

    with open(notebook_path, encoding="utf-8") as f_in:
        doc = json.load(f_in)
    cnt = 1

    for cell in doc["cells"]:
        if "execution_count" in cell:
            cell["execution_count"] = cnt

            for o in cell.get("outputs", []):
                if "execution_count" in o:
                    o["execution_count"] = cnt

            cnt += 1

    with open(notebook_path, "w", encoding="utf-8") as f_out:
        json.dump(doc, f_out, indent=1, ensure_ascii=False)
        f_out.write("\n")

    print(f"Notebook {notebook_path} updated")
    return 1

Tools.logger_config

LOG_FOLDER `module-attribute`

LOG_FOLDER = 'logs'

InterceptHandler

Bases: Handler

Интеграция loguru с uvicorn. Default handler from examples in loguru documentation. See https://loguru.readthedocs.io/en/stable/overview.html#entirely-compatible-with-standard-logging https://pawamoy.github.io/posts/unify-logging-for-a-gunicorn-uvicorn-app/

emit

emit(record)

Source code in Tools/logger_config.py

def emit(self, record: logging.LogRecord):
    # Get corresponding Loguru level if it exists
    try:
        level = logger.level(record.levelname).name
    except ValueError:
        level = record.levelno

    # Find caller from where originated the logged message
    frame, depth = logging.currentframe(), 2
    while frame.f_code.co_filename == logging.__file__:
        frame = frame.f_back
        depth += 1

    logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())

write

write(message)

Интеграция loguru со стандартным выходом ошибок

Source code in Tools/logger_config.py

def write(self, message):
    """
    Интеграция loguru со стандартным выходом ошибок
    """
    if message.strip():
        logger.info(message.strip())

setup_logger

setup_logger(log_file=None)

Настройка loguru для записи логов.

PARAMETER	DESCRIPTION
`log_file`	Путь к файлу для записи логов. Если None, логирование будет только в stdout. DEFAULT: `None`

Source code in Tools/logger_config.py

def setup_logger(log_file=None):
    """
    Настройка loguru для записи логов.

    :param log_file: Путь к файлу для записи логов. Если None, логирование будет только в stdout.
    """
    logger.remove()

    logger.add(
        sys.stdout,
        colorize=True,
        format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green>|<level>{level}</level>| {message}",
    )

    if log_file:
        os.makedirs(os.path.dirname(log_file), exist_ok=True)
        logger.add(
            log_file,
            colorize=True,
            format="{time} | {level} | {message}",
            rotation="10 MB",
            retention="10 days",
            compression="zip",
        )

configure_client_logging

configure_client_logging(log_folder=LOG_FOLDER)

Создание логгера для клиента

Source code in Tools/logger_config.py

def configure_client_logging(log_folder=LOG_FOLDER):
    """Создание логгера для клиента"""
    setup_logger(os.path.join(log_folder, "client.log"))

configure_server_logging

configure_server_logging(log_folder=LOG_FOLDER)

Создание логгера для сервера

Source code in Tools/logger_config.py

def configure_server_logging(log_folder=LOG_FOLDER):
    """Создание логгера для сервера"""
    setup_logger(os.path.join(log_folder, "server.log"))
    # Удаляем все существующие хэндлеры
    for name in logging.root.manager.loggerDict.keys():
        logging.getLogger(name).handlers = []
        logging.getLogger(name).propagate = True

    logging.basicConfig(
        handlers=[InterceptHandler()], level=logging.INFO  # Добавляем наш перехватчик в корневой логгер
    )
    sys.stderr = InterceptHandler()  # не все библиотеки используют логгеры. Sqlalchemy.exc использует sys.stderr

Tools

Tools

display_images

find_image_files

find_needed_jpeg_files

download_zip

extract_zip

get_ya_disk_url

configure_client_logging

configure_server_logging

set_cell_id

goskatalog_parser

zip_files

Tools.analysis

find_image_files

find_needed_jpeg_files

display_images

get_image_json_info

Tools.parser

goskatalog_parser

download_art

zip_files

Tools.download

get_ya_disk_url

download_zip

extract_zip

Tools.notebook

set_cell_id

Tools.logger_config

LOG_FOLDER module-attribute

InterceptHandler

emit

write

setup_logger

configure_client_logging

configure_server_logging

LOG_FOLDER `module-attribute`