Skip to content

Instantly share code, notes, and snippets.

@m2288
Forked from Priler/pywebio_online_chat.py
Last active November 4, 2022 20:25
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save m2288/fc389667bc8dec3da528fb9b87e74726 to your computer and use it in GitHub Desktop.
Save m2288/fc389667bc8dec3da528fb9b87e74726 to your computer and use it in GitHub Desktop.
Simple online chat on Python in less than 100 lines of code
import asyncio
from pywebio import start_server
from pywebio.input import *
from pywebio.output import *
from pywebio.session import defer_call, info as session_info, run_async, run_js
chat_msgs = []
online_users = set()
MAX_MESSAGES_COUNT = 100
async def main():
global chat_msgs
put_markdown("## 🧊 Online çata xoş gəlmisiz!\nÇatda maksimum 100 mesaj saxlanılır!")
msg_box = output()
put_scrollable(msg_box, height=300, keep_bottom=True)
nickname = await input("Çata daxil olun", required=True, placeholder="Adınız", validate=lambda n: "Bu ad artıq istifadə olunub!" if n in online_users or n == '📢' else None)
online_users.add(nickname)
chat_msgs.append(('📢', f'`{nickname}` çata qoşuldu!'))
msg_box.append(put_markdown(f'📢 `{nickname}` çata qoşuldu'))
refresh_task = run_async(refresh_msg(nickname, msg_box))
while True:
data = await input_group("💭 Yeni mesaj", [
input(placeholder="Mesajın mətni ...", name="msg"),
actions(name="cmd", buttons=["Göndər", {'label': "Çatdan çıxış", 'type': 'cancel'}])
], validate = lambda m: ('msg', "Mesajın mətnini daxil edin!") if m["cmd"] == "Göndər" 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("Çatdan çıxdınız!")
msg_box.append(put_markdown(f'📢 `{nickname}` çatdan çıxdı!'))
chat_msgs.append(('📢', f'📢 `{nickname}` çatdan çıxdı!'))
put_buttons(['Təkrar daxil ol'], onclick=lambda btn:run_js('window.location.reload()'))
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: # if not a message from current user
msg_box.append(put_markdown(f"`{m[0]}`: {m[1]}"))
# remove expired
if len(chat_msgs) > MAX_MESSAGES_COUNT:
chat_msgs = chat_msgs[len(chat_msgs) // 2:]
last_idx = len(chat_msgs)
if __name__ == "__main__":
start_server(main, debug=True, port=8080, cdn=False)
@m2288
Copy link
Author

m2288 commented Nov 4, 2022

AZ translate

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment