

简介
使用PyWebIO库,仅126行代码。
可实现实时聊天,在线人数统计,显IP,时间。
需求
- Python>=3.5.2
- PyWebIO
- asyncio
代码
import asyncio
import json
import time
import pywebio
from pywebio.output import *
from pywebio.session import download, run_js, local, run_async, defer_call
from pywebio.input import *
from pywebio.session import local
msg_history = []
online_ips = set()
def about():
popup('说明', [
put_markdown('* V0.18更新:异步更新,减少冗余信息。'),
put_markdown('* V0.17更新:增加在线IP查看。'),
put_markdown('* V0.16更新:聊天记录重加载,界面美化。'),
put_markdown('* V0.15更新:重塑代码结构,界面美化。'),
put_markdown('* V0.14更新:增加对markdown语法支持。'),
put_markdown('* V0.13更新:增加在线人数,nickname改为IP地址,增加欢迎语与结束语。'),
put_markdown('* V0.12更新:支持nickname。'),
put_markdown('* V0.11更新:支持实时更新聊天框。'),
put_markdown('* V0.10更新:支持消息发送,刷新可接收。'),
put_markdown('竟何@lylelove').style('text-align:right')
])
def cut_foot():
pywebio.session.run_js('document.getElementsByClassName("footer")[0].style.display="none"')
put_button('说明', onclick=about).style("position: fixed;bottom: 10px;right: 10px")
def msg_print(local_msg_his):
with use_scope('msg_content', clear=True):
for i in range(len(local_msg_his)):
put_markdown(local_msg_his[i][0]).style('color: red;font-size:12px')
put_markdown(local_msg_his[i][1])
def head():
cut_foot()
pywebio.config(theme='sketchy')
pywebio.session.set_env(title='聊天室', output_max_width='1080px')
put_html('<h1><center>聊天室<sup>V0.18</sup></center></h1>')
def content():
put_scrollable(put_scope('msg_content'), height=300, keep_bottom=True)
usr_ip = pywebio.session.info.user_ip
online_ips.add(usr_ip)
localtime = time.asctime(time.localtime(time.time()))
msg_history.append(
[localtime, '📢 : `' + str(usr_ip) + '` join the room. ' + str(len(online_ips)) + ' users currently online'])
print(localtime + ' ' + str(online_ips))
@defer_call
def on_close():
localtime = time.asctime(time.localtime(time.time()))
online_ips.remove(usr_ip)
msg_history.append(
[localtime,
'📢 : `' + str(usr_ip) + '` have left the chat room. ' + str(len(online_ips)) + ' users currently online'])
print(localtime + ' ' + str(online_ips))
async def msg_input():
while True:
def check_talk(p):
if len(p) == 0:
return '内容不可为空!'
if len(p) > 500:
return '超出500字符限制!'
msg_item = await input('发送消息', type=TEXT, validate=check_talk,
placeholder='支持markdown语法,最多输入500字符。')
localtime = time.asctime(time.localtime(time.time()))
usr_ip = pywebio.session.info.user_ip
msg_item = '📨 : `' + str(usr_ip) + "` : " + msg_item
msg_history.append([localtime, msg_item, usr_ip])
print([localtime, msg_item])
with open('data.json', 'w') as f:
json.dump(msg_history, f)
def online_ips_show():
with popup('当前在线IP') as ip_show:
for i in range(len(online_ips)):
ip_list=list(online_ips)
put_markdown('`'+str(ip_list[i])+'`')
async def re_msg():
while True:
if len(local.l_msg_history) != len(msg_history):
pass
local.l_msg_history = []
for i in range(len(msg_history)):
local.l_msg_history.append(msg_history[i])
msg_print(local_msg_his=local.l_msg_history)
if len(online_ips) != local.line:
local.line = len(online_ips)
with use_scope('oline_ips', clear=True):
put_button('当前在线:' + str(len(online_ips)), onclick=online_ips_show,color='info')
if len(msg_history) >= 500:
msg_history.pop(0)
await asyncio.sleep(1)
async def main():
local.l_msg_history = []
with open('data.json', 'r') as file_read:
temp = json.load(file_read)
if len(msg_history) != len(temp):
msg_history.clear()
for i in range(len(temp)):
msg_history.append(temp[i])
local.line = len(online_ips)
head()
content()
run_async(re_msg())
run_async(msg_input())
if __name__ == '__main__':
pywebio.start_server(main, port=2239)
Comments