Как подружить непрерывно выполняющую функцию по мониторингу сети и Телеграм бота на aiogram?

Question

[Nickolay Tapokpy]

Здравствуйте, добрые люди. Спасите мою голову.
Имеем функцию по мониторингу интернет соединения:

import asyncio
import datetime
import logging
import socket


def send_ping_request(host="google.com", port=443, timeout=3):
    try:
        socket.setdefaulttimeout(timeout)
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.connect((host, port))
    except OSError as error:
        return False
    else:
        s.close()
        return True


def calculate_time(start, stop):
    time_difference = stop - start
    seconds = float(str(time_difference.total_seconds()))
    return str(datetime.timedelta(seconds=seconds)).split(".")[0]


logger = logging.getLogger("Testy")  # Создаем логгер
fileHandler = logging.FileHandler('logs.log')  # создаем хэндлер и формат для записи в файл
# fileFormat = logging.Formatter("[%(asctime)s]  %(message)s")
fileHandler.setLevel(logging.WARNING)  # устанавливаем уровень хэндлера и добавляем к логгеру
# fileHandler.setFormatter(fileFormat)
logger.addHandler(fileHandler)


async def mon_net_connection(ping_freq=2):  # поставить 20>
    await asyncio.sleep(1)
    motd = f"{str(datetime.datetime.now()).split('.')[0]} - Мониторинг сетевого подключения начат. \n" \
           f" Отправка запроса ping каждые {ping_freq} секунды "
    logger.warning(motd)
    print(motd)
    while True:
        if send_ping_request():
            await asyncio.sleep(ping_freq)
        else:
            down_time = datetime.datetime.now()
            fail_msg = f"{str(down_time).split('.')[0]} False - Сетевое соединение недоступно"
            print(fail_msg)
            logger.warning(fail_msg)
            await asyncio.sleep(1)

            while not send_ping_request():
                await asyncio.sleep(1)  # ждать секунду
            up_time = datetime.datetime.now()
            uptime_message = f"{str(up_time).split('.')[0]} - Сетевое подключение восстановлено"
            down_time = calculate_time(down_time, up_time)
            _m = f"Были вне сети: {down_time} "
            print(uptime_message)
            print(_m)
            logger.warning(uptime_message)
            logger.warning(_m)
            await asyncio.sleep(1)

asyncio.run(mon_net_connection())

И стандартный телеграм-бот на aiogram:

from aiogram import executor
from loader import dp

from utils.notify_admins import on_startup_notify
from utils.set_bot_commands import set_default_commands
import handlers


async def on_startup(dispatcher):
    await set_default_commands(dispatcher)  # Устанавливаем дефолтные команды
    await on_startup_notify(dispatcher)  # Уведомляет админов про запуск

if __name__ == '__main__':
    executor.start_polling(dp, on_startup=on_startup)

Как их запустить, что бы и мониторинг и бот с его хендлерами и т.д. заработали дружно и весело?

Answer 1

[Сергей Горностаев]

Во-первых, asyncio - это выполнение сопрограмм в бесконечном цикле событий. Если в одной из сопрограмм запустить другой бесконечный цикл, он её заблокирует и сломает асинхронность. Поэтому while True нужно убрать и проверки раз в секунду поручать циклу событий. Во-вторых, в асинхронной программе надо максимально избегать блокировок, поэтому блокирующийся send_ping_request надо переписать на асинхронных средствах работы с сокетами, предоставляемых asyncio.

Comments

[Nickolay Tapokpy]

Доброй, если с pingom разобрался:

async def ping_google():
    try:
        conn = await asyncio.open_connection('google.com', 443)
    except Exception as ex:
        print(ex)
        return False
    else:
        return True
ping = asyncio.run(ping_google())

Никак не могу найти как правильно уйти от while=True, везде он используется в подобных моментах.

[Сергей Горностаев]

Nickolay Tapokpy, можно и while оставить, раз он теперь не блокируется. Но я бы лучше сделал асинхронную функцию, которая после своего выполнения создаёт новую асинхронную таску для самой себя.

[Nickolay Tapokpy]

Сергей Горностаев, уже немного едет крыша.

import asyncio
import logging

from icecream import ic

ic.configureOutput(includeContext=True)

logger = logging.getLogger("name")  # Создаем логгер
fileHandler = logging.FileHandler('logs.log')  # создаем хэндлер и формат для записи в файл
fileFormat = logging.Formatter("[%(asctime)s]  %(message)s")
fileHandler.setLevel(logging.WARNING)  # устанавливаем уровень хэндлера и добавляем к логгеру
fileHandler.setFormatter(fileFormat)
logger.addHandler(fileHandler)


async def ping_google():
    try:
       conn = await asyncio.open_connection('google.com', 443)
        ic()
    except Exception as ex:
        logger.warning(ex)
        await asyncio.sleep(1)
        ic()

        return False
    else:
        await asyncio.sleep(1)
        ic()
        return True


async def start_monitoring():
    ic("Мониторинг сетевого подключения начат.")
    logger.warning(f"\n Мониторинг сетевого подключения начат.")


async def mon_net_connection(ping_freq=3):
    ping = asyncio.create_task(ping_google())

    if await ping:
        await asyncio.sleep(ping_freq)
        ic("True - Сетевое соединение установлено")
        logger.warning("True - Сетевое соединение установлено")

    else:
        ic("False - Сетевое соединение недоступно")
        logger.warning("False - Сетевое соединение недоступно")
        await asyncio.sleep(ping_freq)

    reply = asyncio.create_task(mon_net_connection())
    await reply



async def testing(sec=5):
    while True:
        await asyncio.sleep(sec)
        print(f'---- 5 секунд прошло')


async def main():
    tasks = [
        start_monitoring(),
        mon_net_connection(),
        testing()
    ]

    await asyncio.gather(*tasks)


asyncio.run(main())

Все работает когда запускается при наличии соединения, при отключении соединения тоже все работает, но при включении соединения функция зависает и получается await reply не запускает все заново.
Спасибо пожалуйста, ибо кукуха уже начинает сильно свистеть.
Большое спасибо.

[Nickolay Tapokpy]

Наверно разобрался

conn = await asyncio.open_connection('google.com', 443)

, тормозит при потере соединения секунд на 40 иногда. Можно ли это исправить?
И главный вопрос с которого все началось: Как запустить вместе с ботом?

from aiogram import executor
from loader import dp

from utils.notify_admins import on_startup_notify
from utils.set_bot_commands import set_default_commands
import handlers

async def on_startup(dispatcher):
    await set_default_commands(dispatcher)  # Устанавливаем дефолтные команды
    await on_startup_notify(dispatcher)  # Уведомляет админов про запуск


if __name__ == '__main__':
    executor.start_polling(dp, on_startup=on_startup)

как я только не пытался запустить, в итоге или бот работает или мониторинг.

[Сергей Горностаев]

Nickolay Tapokpy,

тормозит при потере соединения секунд на 40 иногда. Можно ли это исправить?

Попробуйте завернуть в asyncio.wait_for()

как я только не пытался запустить, в итоге или бот работает или мониторинг.

Давным давно я писал такой ответ. Сейчас нагуглился ещё такой вариант:

dp.loop.create_task(some_async_function())
executor.start_polling(dp, on_startup=on_startup)

[Nickolay Tapokpy]

Сергей Горностаев,

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.create_task(all_monitoring())
    executor.start_polling(dp, on_startup=on_startup)

Вот так только заработало. Огромное спасибо.

[Nickolay Tapokpy]

Кому интересно, рабочий код асинхронного пингования гугла/интернет соединения

import asyncio
import logging
from utils.logger import logger
# from loader import dp

from icecream import ic

async def ping_google():
    try:
        reader, writer = await asyncio.open_connection('google.com', 443)
        await asyncio.sleep(1)

    except Exception as ex:
        logger.info(ex)
        await asyncio.sleep(1)
        return False
    else:
        writer.close()  # может и надо
        await asyncio.sleep(1)
        return True


async def start_monitoring():
    ic("Мониторинг сетевого подключения начат.")
    logger.info(f"\n Мониторинг сетевого подключения начат.")
    await asyncio.sleep(1)


async def mon_net_connection(ping_freq=3):
    ping = asyncio.create_task(ping_google())
    # ping = asyncio.wait_for(ping_google(), timeout=1.0)

    if await ping:

        ic("True - Сетевое соединение установлено")
        logger.info("True - Сетевое соединение установлено")
        await asyncio.sleep(ping_freq)

    else:
        ic("False - Сетевое соединение недоступно")
        logger.info("False - Сетевое соединение недоступно")
        await asyncio.sleep(ping_freq)

    reply = asyncio.create_task(mon_net_connection())
    await reply


async def testing():
    n = 5
    while True:
        await asyncio.sleep(n)
        print(f'----{n} секунд прошло')
        n += 5


async def all_monitoring():
    tasks = [
        start_monitoring(),
        mon_net_connection(),
        testing()
    ]
    await asyncio.gather(*tasks)


asyncio.run(all_monitoring())