-
-
Save m2288/fc389667bc8dec3da528fb9b87e74726 to your computer and use it in GitHub Desktop.
Simple online chat on Python in less than 100 lines of code
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
AZ translate