Created
August 6, 2025 14:36
-
-
Save sulincix/85b5552a38b0de0748cbe0b96c4bb60d to your computer and use it in GitHub Desktop.
Chat application uses gtk3 and udp broadcast
This file contains hidden or 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 os | |
| import gi | |
| import socket | |
| import threading | |
| import json | |
| import uuid | |
| import sqlite3 | |
| from time import sleep | |
| machine_id = "unknown" | |
| if os.path.isfile("/etc/machine-id"): | |
| with open("/etc/machine-id", "r") as f: | |
| machine_id = f.read() | |
| port=33333 | |
| database_file = "{}/.config/broadcast_chat_messages.db".format(os.environ["HOME"]) | |
| gi.require_version('Gtk', '3.0') | |
| from gi.repository import Gtk, GLib, Gdk | |
| class ChatApp(Gtk.Window): | |
| def __init__(self): | |
| super().__init__(title="LAN Broadcast Messenger") | |
| self.set_default_size(600, 500) | |
| self.set_border_width(10) | |
| self.last_id = None | |
| # Create a vertical box to hold widgets | |
| vbox = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=5) | |
| self.add(vbox) | |
| # Create a ListBox to display messages | |
| self.scrolled = Gtk.ScrolledWindow() | |
| self.listbox = Gtk.ListBox() | |
| self.listbox.set_margin_bottom(10) | |
| self.listbox.set_hexpand(True) | |
| self.listbox.set_vexpand(True) | |
| self.scrolled.add(self.listbox) | |
| vbox.pack_start(self.scrolled, True, True, 0) | |
| # Create an entry field for message input | |
| self.entry = Gtk.Entry() | |
| self.entry.set_placeholder_text("Type your message here...") | |
| self.entry.set_margin_bottom(5) | |
| vbox.pack_start(self.entry, False, False, 0) | |
| self.id_last = "" | |
| # Create a send button | |
| self.send_button = Gtk.Button(label="Send") | |
| self.send_button.set_margin_bottom(5) | |
| self.send_button.get_style_context().add_class("suggested-action") | |
| self.send_button.connect("clicked", self.on_send_button_clicked) | |
| vbox.pack_start(self.send_button, False, False, 0) | |
| # Initialize SQLite database | |
| GLib.idle_add(self.init_db) | |
| # Load messages from the database | |
| GLib.idle_add(self.load_messages) | |
| # Start the UDP listener in a separate thread | |
| self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) | |
| self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) | |
| self.sock.bind(("0.0.0.0", port)) | |
| threading.Thread(target=self.listen_for_messages, daemon=True).start() | |
| # Show the window | |
| self.connect("destroy", Gtk.main_quit) | |
| self.show_all() | |
| def scroll_to_bottom(self): | |
| adjustment = self.scrolled.get_vadjustment() | |
| adjustment.set_value(adjustment.get_upper()) | |
| def init_db(self): | |
| self.conn = sqlite3.connect(database_file) | |
| self.cursor = self.conn.cursor() | |
| self.cursor.execute("""CREATE TABLE IF NOT EXISTS messages ( | |
| id TEXT, | |
| username TEXT, | |
| hostname TEXT, | |
| text TEXT | |
| ) | |
| """) | |
| self.conn.commit() | |
| def load_messages(self): | |
| self.cursor.execute('SELECT * FROM messages') | |
| for row in self.cursor.fetchall(): | |
| message = { | |
| "id": row[0], | |
| "username": row[1], | |
| "hostname": row[2], | |
| "text": row[3] | |
| } | |
| self.add_message(message) | |
| def add_message(self, message): | |
| def generate_bubble(message, is_me): | |
| box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL) | |
| if is_me: | |
| box.pack_start(Gtk.Label(), True, True, 0) | |
| box.pack_start(Gtk.Label(label=message), False, False, 0) | |
| else: | |
| box.pack_start(Gtk.Label(label=message), False, False, 0) | |
| box.pack_start(Gtk.Label(), True, True, 0) | |
| return box | |
| if self.last_id != message["id"]: | |
| if self.last_id: | |
| self.listbox.add(Gtk.Label()) | |
| self.listbox.add(generate_bubble(f"{message['username']}@{message['hostname']}", message["id"] == machine_id)) | |
| self.listbox.add(generate_bubble(message["text"], message["id"] == machine_id)) | |
| self.last_id = message["id"] | |
| self.listbox.show_all() | |
| GLib.timeout_add(100,self.scroll_to_bottom) | |
| def listen_for_messages(self): | |
| while True: | |
| data, addr = self.sock.recvfrom(1024) | |
| try: | |
| message = json.loads(data.decode("utf-8")) | |
| # Update the ListBox with the incoming message | |
| valid = True | |
| for key in ["username", "hostname", "id", "text"]: | |
| if key not in message: | |
| print("Invalid message", message) | |
| valid = False | |
| break | |
| if valid: | |
| GLib.idle_add(self.add_message, message) | |
| GLib.idle_add(self.store_message, message) | |
| except json.JSONDecodeError: | |
| print("Received non-JSON message") | |
| def on_send_button_clicked(self, widget): | |
| message_text = self.entry.get_text() | |
| if message_text: | |
| # Create a JSON message with user information | |
| message = { | |
| "username": os.environ["USER"], | |
| "hostname": socket.gethostname(), | |
| "id": machine_id, | |
| "text": message_text | |
| } | |
| self.sock.sendto(json.dumps(message).encode("utf-8"), ("255.255.255.255", port)) | |
| self.entry.set_text("") | |
| def store_message(self, message): | |
| self.cursor.execute( | |
| """INSERT INTO messages (id, username, hostname, text) VALUES (?, ?, ?, ?)""", | |
| (message['id'], message['username'], message['hostname'], message['text']) | |
| ) | |
| self.conn.commit() | |
| def on_destroy(self, widget): | |
| self.conn.close() | |
| # Run the application | |
| if __name__ == "__main__": | |
| app = ChatApp() | |
| Gtk.main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment