Ваш месенджер на python — Python Idea на vc.ru

Хай,сегодня мы будем делать свой менджер на pywebio. Погнали писать код.

pip install asyncio pip install pywebio

Импортируем:

from pywebio import start_server from pywebio.input import * from pywebio.output import * from pywebio.session import run_async, run_js import asyncio chat_msgs = [] # Сохраняет сообщения чата online_users = set() # Список пользователей онлайн MAX_MESSAGES_COUNT = 100 # Максимальное количество сообщений в чате

Далее создаем асинхронную функцию main(), которая будет отвечать за основную логику чата. Внутри функции мы создаем интерфейс чата с помощью функций PyWebIO:

async def main(): global chat_msgs put_markdown("## Это чат") msg_box = output() put_scrollable(msg_box, height=300, keep_bottom=True) nickname = await input("Войти в чат", required=True, placeholder="Ваше имя", validate=lambda n: "Этот никнейм уже используется" if n in online_users or n == '' else None) online_users.add(nickname) chat_msgs.append(('', f"`{nickname}` присоединился к чату!")) msg_box.append(put_markdown(f"`{nickname}` присоединился к чату!")) refresh_task = run_async(refresh_msg(nickname, msg_box)) while True: data = await input_group("Новое сообщение", [ input(placeholder="Текст сообщения", name="msg"), actions(name="cmd", buttons=["Отправить", {'label': "Покинуть чат", 'type': 'cancel'}]) ], validate=lambda m: ('msg', "Введите ваше сообщение!") if m["cmd"] == "Отправить" and not m["msg"] else None) if data is None: break msg_box.append(put_markdown(f"`{nickname}`: {data['msg']}")) chat_msgs.append((nickname, data['msg'])) # выход из чата refresh_task.close() online_users.remove(nickname) toast("Вы покинули чат!") msg_box.append(put_markdown(f"Пользователь `{nickname}` покинул чат!")) chat_msgs.append(('', f"Пользователь `{nickname}` покинул чат!")) put_buttons(['Вернуться в чат'], onclick=lambda btn: run_js('window.location.reload()'))

Функция refresh_msg(nickname, msg_box) является асинхронной и предназначена для обновления сообщений в чате. Давайте рассмотрим её поэтапно:

async def refresh_msg(nickname, msg_box): global chat_msgs last_idx = len(chat_msgs) while True: await asyncio.sleep(1) for m in chat_msgs[last_idx:]: if m[0] != nickname: msg_box.append(put_markdown(f"`{m[0]}`: {m[1]}"))

Функция выполняется в бесконечном цикле с использованием while True. Внутри цикла мы вызываем await asyncio.sleep(1), чтобы задержать выполнение функции на 1 секунду перед каждой итерацией. Это позволяет осуществить плавное обновление сообщений в чате с задержкой в 1 секунду между обновлениями.

Затем мы проходим по сообщениям chat_msgs начиная с индекса last_idx. Мы проверяем, что отправитель сообщения (m[0]) не совпадает с текущим пользователем (nickname), и если это так, добавляем сообщение в msg_box для отображения в чате.

# удаление устаревших сообщений if len(chat_msgs) > MAX_MESSAGES_COUNT: chat_msgs = chat_msgs[len(chat_msgs) // 2:] last_idx = len(chat_msgs)

В конце каждой итерации обновляем last_idx для следующей итерации цикла. Мы также проверяем, если количество сообщений chat_msgs превышает MAX_MESSAGES_COUNT, то удаляем старые сообщения, оставляя только последнюю половину сообщений.

И наконец, в самом низу кода, мы вызываем функцию start_server(main, debug=True, port=8080, cdn=False), чтобы запустить веб-сервер и инициализировать чат с помощью функции main(). Веб-сервер будет запущен на порту 8080 и будет использовать локальные ресурсы, а отладочный режим (debug=True) позволяет выводить отладочную информацию при необходимости.

if __name__ == "__main__": start_server(main, debug=True, port=8080, cdn=False)

Ну вот:

Спасибо за просмотр статьй.