Ein Chat in Python
LAN_chat.pywimport datetime
import io
import math
import os
import pathlib
import re
import socket
import sys
import threading
import time
import tkinter
import tkinter.messagebox
from tkinter import font
root = tkinter.Tk()
root.withdraw()
progname="Stefan Ram's LAN chat"
versionname=progname +" (Version 2020-09-14/--24)"
def startup():
global file
shared_dir_ = r"X:\Alle"
shared_dir = pathlib.Path( shared_dir_ )
if not shared_dir.is_dir():
tkinter.messagebox.showerror( progname, f'There is no directory "{shared_dir_}"!' )
sys.exit( 99 )
file = shared_dir_ + r"\chatfile_fvmchmpqtx.txt"
startup()
def follow():
first_opening = True
while True:
try:
with open( file, encoding="utf_8" ) as input:
if first_opening:
# SEEK_END = 2; input.seek( 0, SEEK_END )
input.seek( stream_pos, io.SEEK_SET )
first_opening = False
latest_data = input.read()
while True:
if '\n' not in latest_data:
latest_data += input.read()
if '\n' not in latest_data:
yield ''
if not os.path.isfile( file ):
break
continue
latest_lines = latest_data.split( '\n' )
if latest_data[ -1 ]!= '\n':
latest_data = latest_lines[ -1 ]
else:
latest_data = input.read()
for line in latest_lines[:-1]:
yield line + '\n'
except OSError as oserror:
if oserror.errno == 2:
time.sleep( 3 )
def receive():
for msg in follow():
if exiting: return
if msg == "":
if exiting: return
root.config( cursor="" )
time.sleep( 1 )
else:
root.config( cursor="wait" )
try:
#print( f"received: \"{msg}\"." )
if exiting: return
listbox.insert( tkinter.END, msg )
listbox.yview( tkinter.END )
if re.match( r"^\d\d\d\d-\d\d-\d\dT.*", msg ):
listbox.itemconfig( tkinter.END, bg='lightgray' )
except OSError as oserror:
print( oserror )
break
def entrybox_send( event=None ):
msg = entrybox.get( "1.0", "end-1c" )
entrybox.delete( "1.0", tkinter.END )
try:
with open( file, "a", encoding="utf_8" ) as output:
output.write( datetime.datetime.now( datetime.timezone.utc ).astimezone().isoformat( 'T' ) + "\n" )
output.write( msg + "\n" )
os.system('attrib +H ' + file + ' /S')
except exception as e:
print( tkinter.END, "\nentrybox_send: append failed." )
if msg == "exit":
root.quit()
def on_closing( event=None ):
global exiting
exiting = True
root.destroy()
height=16
width=16
image=tkinter.PhotoImage(height=16, width=width)
image.blank()
for x in range( width ):
for y in range( height ):
image.put( "#ffffff",( x, y ))
for x in range( width ):
for y in range( 11, 14 ):
image.put( "#000000",( x, y ))
root.wm_iconphoto('True', image)
root.title( versionname )
def f():
while not exiting:
root.title( versionname + " " + datetime.datetime.now( datetime.timezone.utc ).astimezone().isoformat( 'T' ) )
time.sleep( 1 )
messages = tkinter.Frame( root )
scrollbar = tkinter.Scrollbar( messages )
small_font = font.nametofont("TkFixedFont")
small_font.configure(size=20,weight="bold")
listbox = tkinter.Listbox( messages, height=1, width=1, yscrollcommand=scrollbar.set, selectmode=tkinter.EXTENDED, font=small_font )
scrollbar.pack( side=tkinter.RIGHT, fill=tkinter.Y )
scrollbar.config(command=listbox.yview) # call list.yview when I move
listbox.config(yscrollcommand=scrollbar.set) # call sbar.set when I move
listbox.pack( side=tkinter.LEFT, expand=1, fill=tkinter.BOTH )
messages.pack( expand=1, fill=tkinter.BOTH )
from tkinter.scrolledtext import ScrolledText
entrybox = ScrolledText( root, height=5 )
entrybox.bind( "<Control-KeyRelease-Return>", entrybox_send )
entrybox.pack( side=tkinter.LEFT, expand=1, fill=tkinter.X )
send_button = tkinter.Button( root, text="Send", command=entrybox_send )
send_button.pack( side=tkinter.RIGHT )
entrybox.focus_set()
root.geometry( "512x383" )
root.protocol( "WM_DELETE_WINDOW", on_closing )
root.deiconify()
stream_pos = 0
# on the main thread, to suppress screen updating
try:
with open( file, encoding="utf_8" ) as input:
latest_data = input.read()
stream_pos = input.tell()
latest_lines = latest_data.split( '\n' )
root.config( cursor="wait" )
for line in latest_lines[:-1]:
listbox.insert( tkinter.END, line )
listbox.yview( tkinter.END )
if re.match( r"^\d\d\d\d-\d\d-\d\dT.*", line ):
listbox.itemconfig( tkinter.END, bg='lightgray' )
root.config( cursor="" )
except OSError as oserror:
pass
exiting = False
freceive_thread = threading.Thread( target=f )
freceive_thread.start()
receive_thread = threading.Thread( target=receive )
receive_thread.start()
tkinter.mainloop()- socket chat server
import socket
import threading
def serve():
while True:
client, address = server_socket.accept()
print( f"{address} has connected." )
client.send( bytes( "Connection accepted.\r\n", "utf_8") )
clientdict[ client ]= address
threading.Thread( target=handle_client, args=( client, )).start()
def handle_client( client ):
while True:
try:
data = client.recv( size )
text = data.decode()
print( f"received: \"{text}\"" )
if text == "exit":
dismiss( client )
break
else:
distribute( data )
except:
print( "handle_client: error in send." )
dismiss( client )
break
def distribute( data ):
for client in clientdict.copy():
try:
client.send( data )
except:
print( "distribute: error in send." )
dismiss( client )
continue
def dismiss( client ):
print( "dimissing client." )
try:
client.send( bytes( "exit", "utf_8" ))
except:
pass
client.close()
del clientdict[ client ]
print( f"{socket.gethostbyname_ex( socket.gethostname() ) = }" )
clientdict = {}
size = 1024
server_socket = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
server_socket.bind( ( '', 33001 ))
if __name__ == "__main__":
server_socket.listen( 5 )
print( "Server is running." )
thread = threading.Thread( target=serve )
thread.start()
thread.join()
server_socket.close()
- socket chat client
import math
import socket
import threading
import tkinter
def receive():
while True:
try:
msg = server.recv( size ).decode( "utf_8 ")
print( f"received: \"{msg}\"." )
listbox.insert( tkinter.END, msg )
except OSError:
break
def entrybox_send( event=None ):
msg = inputvar.get()
inputvar.set( "" )
try:
server.send( bytes( msg, "utf_8" ))
except:
print( tkinter.END, "\nentrybox_send: server.send failed." )
if msg == "exit":
server.close()
tk.quit()
def on_closing( event=None ):
try:
server.send( bytes( "exit", "utf_8" ))
except:
print( "\non_closing: server.send failed." )
try:
server.close()
except:
print( "\non_closing: server.close failed." )
tk.destroy()
tk = tkinter.Tk()
height=16
width=16
image=tkinter.PhotoImage(height=16, width=width)
image.blank()
for x in range( width ):
for y in range( height ):
image.put( "#ffffff",( x, y ))
for x in range( width ):
for y in range( 11, 14 ):
image.put( "#000000",( x, y ))
tk.wm_iconphoto('True', image)
tk.title( "Chat" )
messages = tkinter.Frame( tk )
inputvar = tkinter.StringVar()
scrollbar = tkinter.Scrollbar( messages )
listbox = tkinter.Listbox( messages, height=1, width=1, yscrollcommand=scrollbar.set )
scrollbar.pack( side=tkinter.RIGHT, fill=tkinter.Y )
listbox.pack( side=tkinter.LEFT, expand=1, fill=tkinter.BOTH )
messages.pack( expand=1, fill=tkinter.BOTH )
entrybox = tkinter.Entry( tk, textvariable=inputvar )
entrybox.bind( "<Return>", entrybox_send )
entrybox.pack( side=tkinter.LEFT, expand=1, fill=tkinter.X )
send_button = tkinter.Button( tk, text="Send", command=entrybox_send )
send_button.pack( side=tkinter.RIGHT )
tk.geometry( "512x383" )
tk.protocol( "WM_DELETE_WINDOW", on_closing )
host = input( 'server address: ' )
if not host:
host = "localhost"
size = 1024
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.connect( ( host, 33001 ))
receive_thread = threading.Thread( target=receive )
receive_thread.start()
tkinter.mainloop()