Веб-мониторинг МФУ и уровня тонера через SNMP на Python + Flask

from flask import Flask, jsonify, render_template   # Импортируем Flask и функции для API и шаблонов
from pysnmp.hlapi import *                          # Импортируем всё для SNMP-опроса
import csv                                          # Для работы с CSV-файлом
from concurrent.futures import ThreadPoolExecutor   # Для параллельного опроса принтеров

def is_ip(address): # Проверка, что строка — это IP-адрес
    parts = address.split('.')
    return len(parts) == 4 and all(p.isdigit() and 0 <= int(p) <= 255 for p in parts)

def load_printers(filename='printers.csv'): # Загрузка списка принтеров из CSV-файла
    printers = []
    with open(filename, 'r', encoding='utf-8-sig') as f:
        reader = csv.DictReader(f)
        for row in reader:
            # Приводим все ключи к стандартному виду без кавычек и пробелов
            clean_row = {k.strip().replace('"', ''): v for k, v in row.items()}
            # Для отладки: Если раскомментировать, то при запуске сервера в консоли увидим словари с ключами и значениями из каждой строки файла.
            # print(clean_row)
            name = clean_row.get('Name') or clean_row.get('Имя')  # если вдруг русское имя
            port = clean_row.get('PortName') or clean_row.get('Порт')  # если вдруг русское имя
            if port and is_ip(port):
                printers.append({'name': name, 'ip': port})
    return printers

PRINTERS = load_printers()

app = Flask(__name__) #Создаём Flask-приложение

def monitor_mfu_from_csv(file_path):
    try:
        with open(file_path, newline='', encoding='utf-8') as csvfile:
            reader = csv.DictReader(csvfile)
            for row in reader:
                mfu_name = row.get('МФУ')
                status = row.get('Статус')
                loaded = row.get('Загружено')
                print(f"МФУ: {mfu_name}, Статус: {status}, Загружено: {loaded}")
    except FileNotFoundError:
        print(f"Файл не найден: {file_path}")
    except Exception as e:
        print(f"Ошибка при чтении файла: {e}")

# Пример вызова функции
# monitor_mfu_from_csv('mfu_status.csv')]

def get_printer_status(printer): # Основная функция опроса принтера по SNMP
    ip = printer['ip']
    name = printer['name']
    toner_current_oid = '1.3.6.1.2.1.43.11.1.1.9.1.1'  # уровень тонера
    toner_max_oid = '1.3.6.1.2.1.43.11.1.1.8.1.1'      # максимальное значение тонера
    toner_type_oid = '1.3.6.1.2.1.43.11.1.1.6.1.1'     # тип тонера
    serial_oid = '1.3.6.1.2.1.43.5.1.1.17.1'           # серийный номер
    try: # SNMP-запрос сразу по всем нужным OID
        iterator = getCmd(
            SnmpEngine(),
            CommunityData('public', mpModel=0),
            UdpTransportTarget((ip, 161), timeout=1, retries=1),
            ContextData(),
            ObjectType(ObjectIdentity(toner_current_oid)),
            ObjectType(ObjectIdentity(toner_max_oid)),
            ObjectType(ObjectIdentity(toner_type_oid)),
            ObjectType(ObjectIdentity(serial_oid))
        )
        errorIndication, errorStatus, errorIndex, varBinds = next(iterator)
        if errorIndication or errorStatus:
            return {
                'ip': ip,
                'name': name,
                'status': 'Error',
                'toner_level': None,
                'toner_type': None,
                'serial': None
            }
        else:
            current = int(varBinds[0][1])                   # Текущее значение тонера
            maximum = int(varBinds[1][1])                   # Максимальное значение тонера
            toner_type = str(varBinds[2][1])                # Тип тонера (модель/цвет/код)
            serial = str(varBinds[3][1])                    # Серийный номер принтера
            if maximum > 0 and current >= 0:
                percent = int(current / maximum * 100)      # Считаем процент тонера
            else:
                percent = None
            return {
                'ip': ip,
                'name': name,
                'status': 'OK',
                'toner_level': percent,
                'toner_type': toner_type,
                'serial': serial
            }
    except Exception as e:                      # Если возникла ошибка при SNMP-запросе — возвращаем статус Error
        print(f"Ошибка при запросе {ip}: {e}")
        return {
            'ip': ip,
            'name': name,
            'status': 'Error',
            'toner_level': None,
            'toner_type': None,
            'serial': None
        }

@app.route('/')
def index():
    return render_template('index.html')       # Отдаём HTML-страницу с таблицей         

@app.route('/api/printers')
def api_printers():                            # Параллельно опрашиваем все принтеры (ускоряет работу)
    with ThreadPoolExecutor(max_workers=10) as executor:
        results = list(executor.map(get_printer_status, PRINTERS))
        results.sort(key=lambda x: (x['status'] == 'Error',     # Сортировка: сначала без ошибок, потом по возрастанию тонера (ошибки внизу)
                                    x['toner_level'] is None or x['toner_level'] > 20, 
                                    x['toner_level'] if x['toner_level'] is not None else 101))
        return jsonify(results)                # Отдаём данные в формате JSON для таблицы

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)    # Запускаем сервер на нужном IP и порту         

<!DOCTYPE html>
<html>
<head>
    <title>Статус принтеров</title>
</head>
<body>
<h1>Статус принтеров в реальном времени</h1>
<table border="1" id="printers-table">
    <tr>
        <th>Имя</th>
        <th>IP</th>
        <th>Статус</th>
        <th>Уровень чернил</th>
		<th>Тип тонера</th>
		<th>Серийный</th>
    </tr>
</table>

<script>
function fetchData() {
    fetch('/api/printers')
        .then(response => response.json())
        .then(data => {
            const table = document.getElementById('printers-table');
            table.innerHTML = `
                <tr>
                    <th>Имя</th>
                    <th>IP</th>
                    <th>Статус</th>
                    <th>Уровень чернил</th>
					<th>Тип тонера</th>
					<th>Серийный</th>
                </tr>`;
            data.forEach(printer => {
                const row = table.insertRow();
                row.insertCell().innerText = printer.name || '—';
                row.insertCell().innerText = printer.ip;
                row.insertCell().innerText = printer.status;
                row.insertCell().innerText = (printer.toner_level !== null) ? (printer.toner_level + ' %') : 'N/A';
				row.insertCell().innerText = printer.toner_type || '—';
				row.insertCell().innerText = printer.serial || '—';
            });
        });
}
setInterval(fetchData, 30000);
fetchData();
</script>
</body>
</html>