import gradio as gr import asyncio import websockets import json import uuid import argparse import urllib.parse from datetime import datetime import logging import sys import os import shutil from pathlib import Path import time # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[logging.StreamHandler(sys.stdout)] ) logger = logging.getLogger("chat-node") # Dictionary to store active connections active_connections = {} # Dictionary to store message history for each chat room (in-memory cache) chat_history = {} # Dictionary to store message history for logs log_history = [] # Dictionary to track file modification times file_modification_times = {} # Dictionary to track users in each room/sector sector_users = {} # Grid dimensions for 2D sector map GRID_WIDTH = 10 GRID_HEIGHT = 10 # Directories for persistent storage HISTORY_DIR = "chat_history" LOG_DIR = "server_logs" os.makedirs(HISTORY_DIR, exist_ok=True) os.makedirs(LOG_DIR, exist_ok=True) # README files README_PATH = os.path.join(HISTORY_DIR, "README.md") if not os.path.exists(README_PATH): with open(README_PATH, "w") as f: f.write("# Chat History\n\nThis directory contains persistent chat history files.\n") LOG_README_PATH = os.path.join(LOG_DIR, "README.md") if not os.path.exists(LOG_README_PATH): with open(LOG_README_PATH, "w") as f: f.write("# Server Logs\n\nThis directory contains server log files.\n") # Get node name from URL or command line def get_node_name(): parser = argparse.ArgumentParser(description='Start a chat node with a specific name') parser.add_argument('--node-name', type=str, default=None, help='Name for this chat node') parser.add_argument('--port', type=int, default=7860, help='Port to run the Gradio interface on') args = parser.parse_args() node_name = args.node_name port = args.port if not node_name: node_name = f"node-{uuid.uuid4().hex[:8]}" return node_name, port def get_room_history_file(room_id): timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") return os.path.join(HISTORY_DIR, f"{room_id}_{timestamp}.jsonl") def get_log_file(): timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") return os.path.join(LOG_DIR, f"log_{timestamp}.jsonl") def get_all_room_history_files(room_id): files = [] for file in os.listdir(HISTORY_DIR): if file.startswith(f"{room_id}_") and file.endswith(".jsonl"): files.append(os.path.join(HISTORY_DIR, file)) files.sort(key=lambda x: os.path.getmtime(x), reverse=True) return files def list_all_history_files(): """List all history files with room IDs and modification times.""" files = [] for file in os.listdir(HISTORY_DIR): if file.endswith(".jsonl") and file != "README.md": parts = file.split('_', 1) if len(parts) > 0: room_id = parts[0] file_path = os.path.join(HISTORY_DIR, file) mod_time = os.path.getmtime(file_path) files.append((room_id, file_path, mod_time)) files.sort(key=lambda x: x[2], reverse=True) # Sort by modification time return files def load_room_history(room_id): if room_id not in chat_history: chat_history[room_id] = [] history_files = get_all_room_history_files(room_id) for file in history_files: if file not in file_modification_times: file_modification_times[file] = os.path.getmtime(file) messages = [] for history_file in history_files: try: with open(history_file, 'r') as f: for line in f: line = line.strip() if line: try: data = json.loads(line) messages.append(data) except json.JSONDecodeError: logger.error(f"Error parsing JSON line in {history_file}") except Exception as e: logger.error(f"Error loading history from {history_file}: {e}") messages.sort(key=lambda x: x.get("timestamp", ""), reverse=False) chat_history[room_id] = messages logger.info(f"Loaded {len(messages)} messages from {len(history_files)} files for room {room_id}") if room_id not in sector_users: sector_users[room_id] = set() return chat_history[room_id] def save_message_to_history(room_id, message): history_files = get_all_room_history_files(room_id) if not history_files: history_file = get_room_history_file(room_id) else: newest_file = history_files[0] if os.path.getsize(newest_file) > 1024 * 1024: # 1 MB history_file = get_room_history_file(room_id) else: history_file = newest_file try: with open(history_file, 'a') as f: f.write(json.dumps(message) + '\n') file_modification_times[history_file] = os.path.getmtime(history_file) logger.debug(f"Saved message to {history_file}") except Exception as e: logger.error(f"Error saving message to {history_file}: {e}") def save_log_entry(entry): """Save a log entry to the newest log file.""" log_files = [f for f in os.listdir(LOG_DIR) if f.startswith("log_") and f.endswith(".jsonl")] log_files.sort(key=lambda x: os.path.getmtime(os.path.join(LOG_DIR, x)), reverse=True) if not log_files: log_file = get_log_file() else: newest_log = os.path.join(LOG_DIR, log_files[0]) if os.path.getsize(newest_log) > 1024 * 1024: # 1 MB log_file = get_log_file() else: log_file = newest_log try: with open(log_file, 'a') as f: f.write(json.dumps(entry) + '\n') file_modification_times[log_file] = os.path.getmtime(log_file) except Exception as e: logger.error(f"Error saving log to {log_file}: {e}") def check_for_new_messages(): updated_rooms = set() for file in os.listdir(HISTORY_DIR): if file.endswith(".jsonl"): file_path = os.path.join(HISTORY_DIR, file) current_mtime = os.path.getmtime(file_path) if file_path not in file_modification_times or current_mtime > file_modification_times[file_path]: parts = file.split('_', 1) if len(parts) > 0: room_id = parts[0] updated_rooms.add(room_id) file_modification_times[file_path] = current_mtime for room_id in updated_rooms: if room_id in chat_history: old_history_len = len(chat_history[room_id]) chat_history[room_id] = [] load_room_history(room_id) new_history_len = len(chat_history[room_id]) if new_history_len > old_history_len: logger.info(f"Found {new_history_len - old_history_len} new messages for room {room_id}") return updated_rooms def check_for_new_logs(): """Check for new log entries and broadcast them.""" global log_history updated = False for file in os.listdir(LOG_DIR): if file.endswith(".jsonl"): file_path = os.path.join(LOG_DIR, file) current_mtime = os.path.getmtime(file_path) if file_path not in file_modification_times or current_mtime > file_modification_times[file_path]: updated = True file_modification_times[file_path] = current_mtime if updated: log_history = [] for file in sorted([f for f in os.listdir(LOG_DIR) if f.endswith(".jsonl")], key=lambda x: os.path.getmtime(os.path.join(LOG_DIR, x))): file_path = os.path.join(LOG_DIR, file) try: with open(file_path, 'r') as f: for line in f: line = line.strip() if line: try: log_history.append(json.loads(line)) except json.JSONDecodeError: logger.error(f"Error parsing log line in {file_path}") except Exception as e: logger.error(f"Error loading logs from {file_path}: {e}") # Broadcast logs to all connected clients in a "logs" room log_msg = { "type": "log", "content": "\n".join([entry["message"] for entry in log_history[-10:]]), # Last 10 entries "timestamp": datetime.now().isoformat(), "sender": "system", "room_id": "logs" } asyncio.create_task(broadcast_message(log_msg, "logs")) def get_sector_coordinates(room_id): try: if ',' in room_id: x, y = map(int, room_id.split(',')) return max(0, min(x, GRID_WIDTH-1)), max(0, min(y, GRID_HEIGHT-1)) except: pass hash_val = hash(room_id) x = abs(hash_val) % GRID_WIDTH y = abs(hash_val >> 8) % GRID_HEIGHT return x, y def generate_sector_map(): grid = [[' ' for _ in range(GRID_WIDTH)] for _ in range(GRID_HEIGHT)] for room_id, users in sector_users.items(): if users: x, y = get_sector_coordinates(room_id) user_count = len(users) grid[y][x] = str(min(user_count, 9)) if user_count < 10 else '+' header = ' ' + ''.join([str(i % 10) for i in range(GRID_WIDTH)]) map_str = header + '\n' for y in range(GRID_HEIGHT): row = f"{y % 10}|" + ''.join(grid[y]) + '|' map_str += row + '\n' map_str += header return f"```\n{map_str}\n```\n\nLegend: Number indicates users in sector. '+' means 10+ users." async def clear_all_history(): global chat_history chat_history = {} for file in os.listdir(HISTORY_DIR): if file.endswith(".jsonl"): os.remove(os.path.join(HISTORY_DIR, file)) clear_msg = { "type": "system", "content": "🧹 All chat history has been cleared by a user", "timestamp": datetime.now().isoformat(), "sender": "system" } for room_id in list(active_connections.keys()): clear_msg["room_id"] = room_id await broadcast_message(clear_msg, room_id) logger.info("All chat history cleared") return "All chat history cleared" async def websocket_handler(websocket, path): try: path_parts = path.strip('/').split('/') room_id = path_parts[0] if path_parts else "default" client_id = str(uuid.uuid4()) if room_id not in active_connections: active_connections[room_id] = {} active_connections[room_id][client_id] = websocket if room_id not in sector_users: sector_users[room_id] = set() sector_users[room_id].add(client_id) x, y = get_sector_coordinates(room_id) room_history = load_room_history(room_id) welcome_msg = { "type": "system", "content": f"Welcome to room '{room_id}' (Sector {x},{y})! Connected from node '{NODE_NAME}'", "timestamp": datetime.now().isoformat(), "sender": "system", "room_id": room_id } await websocket.send(json.dumps(welcome_msg)) map_msg = { "type": "system", "content": f"Sector Map:\n{generate_sector_map()}", "timestamp": datetime.now().isoformat(), "sender": "system", "room_id": room_id } await websocket.send(json.dumps(map_msg)) for msg in room_history: await websocket.send(json.dumps(msg)) join_msg = { "type": "system", "content": f"User joined the room (Sector {x},{y}) - {len(sector_users[room_id])} users now present", "timestamp": datetime.now().isoformat(), "sender": "system", "room_id": room_id } await broadcast_message(join_msg, room_id) save_message_to_history(room_id, join_msg) logger.info(f"New client {client_id} connected to room {room_id} (Sector {x},{y})") if room_id == "logs": for log_entry in log_history[-10:]: # Send last 10 log entries log_msg = { "type": "log", "content": log_entry["message"], "timestamp": log_entry["timestamp"], "sender": "system", "room_id": "logs" } await websocket.send(json.dumps(log_msg)) async for message in websocket: try: data = json.loads(message) if data.get("type") == "command" and data.get("command") == "clear_history": result = await clear_all_history() continue if data.get("type") == "command" and data.get("command") == "show_map": map_msg = { "type": "system", "content": f"Sector Map:\n{generate_sector_map()}", "timestamp": datetime.now().isoformat(), "sender": "system", "room_id": room_id } await websocket.send(json.dumps(map_msg)) continue data["timestamp"] = datetime.now().isoformat() data["sender_node"] = NODE_NAME data["room_id"] = room_id chat_history[room_id].append(data) if len(chat_history[room_id]) > 500: chat_history[room_id] = chat_history[room_id][-500:] save_message_to_history(room_id, data) await broadcast_message(data, room_id) except json.JSONDecodeError: error_msg = { "type": "error", "content": "Invalid JSON format", "timestamp": datetime.now().isoformat(), "sender": "system", "room_id": room_id } await websocket.send(json.dumps(error_msg)) except websockets.exceptions.ConnectionClosed: logger.info(f"Client {client_id} disconnected from room {room_id}") finally: if room_id in active_connections and client_id in active_connections[room_id]: del active_connections[room_id][client_id] if room_id in sector_users and client_id in sector_users[room_id]: sector_users[room_id].remove(client_id) x, y = get_sector_coordinates(room_id) leave_msg = { "type": "system", "content": f"User left the room (Sector {x},{y}) - {len(sector_users.get(room_id, set()))} users remaining", "timestamp": datetime.now().isoformat(), "sender": "system", "room_id": room_id } await broadcast_message(leave_msg, room_id) save_message_to_history(room_id, leave_msg) if not active_connections[room_id]: del active_connections[room_id] async def broadcast_message(message, room_id): if room_id in active_connections: disconnected_clients = [] for client_id, websocket in active_connections[room_id].items(): try: await websocket.send(json.dumps(message)) except websockets.exceptions.ConnectionClosed: disconnected_clients.append(client_id) for client_id in disconnected_clients: del active_connections[room_id][client_id] async def start_websocket_server(host='0.0.0.0', port=8765): server = await websockets.serve(websocket_handler, host, port) logger.info(f"WebSocket server started on ws://{host}:{port}") return server main_event_loop = None message_queue = [] def send_message(message, username, room_id): if not message.strip(): return None global message_queue msg_data = { "type": "chat", "content": message, "username": username, "room_id": room_id } message_queue.append(msg_data) formatted_msg = f"{username}: {message}" return formatted_msg def join_room(room_id, chat_history_output): if not room_id.strip(): return "Please enter a valid room ID", chat_history_output room_id = urllib.parse.quote(room_id.strip()) history = load_room_history(room_id) formatted_history = [] for msg in history: if msg.get("type") == "chat": sender_node = f" [{msg.get('sender_node', 'unknown')}]" if "sender_node" in msg else "" time_str = "" if "timestamp" in msg: try: dt = datetime.fromisoformat(msg["timestamp"]) time_str = f"[{dt.strftime('%H:%M:%S')}] " except: pass formatted_history.append(f"{time_str}{msg.get('username', 'Anonymous')}{sender_node}: {msg.get('content', '')}") elif msg.get("type") == "system": formatted_history.append(f"System: {msg.get('content', '')}") return f"Joined room: {room_id}", formatted_history def send_clear_command(): global message_queue msg_data = { "type": "command", "command": "clear_history", "username": "System" } message_queue.append(msg_data) return "🧹 Clearing all chat history..." def list_available_rooms(): history_files = list_all_history_files() if not history_files: return "No chat rooms available yet. Create one by joining a room!" room_list = "### Available Chat Rooms\n\n" for room_id, file_path, mod_time in history_files: last_activity = datetime.fromtimestamp(mod_time).strftime("%Y-%m-%d %H:%M:%S") room_list += f"- **{room_id}**: Last activity {last_activity}\n" return room_list def create_gradio_interface(): with gr.Blocks(title=f"Chat Node: {NODE_NAME}") as interface: gr.Markdown(f"# Chat Node: {NODE_NAME}") gr.Markdown("Join a room by entering a room ID below or create a new one.") with gr.Row(): with gr.Column(scale=3): room_list = gr.Markdown(value="Loading available rooms...") refresh_button = gr.Button("🔄 Refresh Room List") with gr.Column(scale=1): clear_button = gr.Button("🧹 Clear All Chat History", variant="stop") with gr.Row(): with gr.Column(scale=2): room_id_input = gr.Textbox(label="Room ID", placeholder="Enter room ID or use x,y coordinates") join_button = gr.Button("Join Room") with gr.Column(scale=1): with gr.Row(): x_coord = gr.Number(label="X", value=0, minimum=0, maximum=GRID_WIDTH-1, step=1) y_coord = gr.Number(label="Y", value=0, minimum=0, maximum=GRID_HEIGHT-1, step=1) grid_join_button = gr.Button("Join by Coordinates") chat_history_output = gr.Textbox(label="Chat History", lines=20, max_lines=20) with gr.Row(): username_input = gr.Textbox(label="Username", placeholder="Enter your username", value="User") with gr.Column(scale=3): message_input = gr.Textbox( label="Message", placeholder="Type your message here. Press Shift+Enter for new line, Enter to send.", lines=3 ) with gr.Column(scale=1): send_button = gr.Button("Send") map_button = gr.Button("🗺️ Show Map") current_room_display = gr.Textbox(label="Current Room", value="Not joined any room yet") refresh_button.click(list_available_rooms, inputs=[], outputs=[room_list]) clear_button.click(send_clear_command, inputs=[], outputs=[room_list]) def join_by_coordinates(x, y): return f"{int(x)},{int(y)}" grid_join_button.click(join_by_coordinates, inputs=[x_coord, y_coord], outputs=[room_id_input]).then( join_room, inputs=[room_id_input, chat_history_output], outputs=[current_room_display, chat_history_output] ) join_button.click(join_room, inputs=[room_id_input, chat_history_output], outputs=[current_room_display, chat_history_output]) def send_and_clear(message, username, room_id): if not room_id.startswith("Joined room:"): return "Please join a room first", message actual_room_id = room_id.replace("Joined room: ", "").strip() message_lines = message.strip().split("\n") formatted_msg = "" for line in message_lines: if line.strip(): sent_msg = send_message(line.strip(), username, actual_room_id) if sent_msg: formatted_msg += sent_msg + "\n" return "", formatted_msg if formatted_msg else message, None send_button.click( send_and_clear, inputs=[message_input, username_input, current_room_display], outputs=[message_input, chat_history_output] ) def show_sector_map(room_id): if not room_id.startswith("Joined room:"): return "Please join a room first to view the map" return generate_sector_map() map_button.click(show_sector_map, inputs=[current_room_display], outputs=[chat_history_output]) def on_message_submit(message, username, room_id): return send_and_clear(message, username, room_id) message_input.submit( on_message_submit, inputs=[message_input, username_input, current_room_display], outputs=[message_input, chat_history_output] ) interface.load(list_available_rooms, inputs=[], outputs=[room_list]) return interface async def process_message_queue(): global message_queue while True: if message_queue: msg_data = message_queue.pop(0) await broadcast_message(msg_data, msg_data["room_id"]) await asyncio.sleep(0.1) async def process_logs(): """Periodically check and broadcast new log entries.""" while True: check_for_new_logs() await asyncio.sleep(1) # Check every second # Custom logging handler to save logs and broadcast them class LogBroadcastHandler(logging.Handler): def emit(self, record): log_entry = { "timestamp": datetime.now().isoformat(), "level": record.levelname, "message": self.format(record), "name": record.name } save_log_entry(log_entry) log_history.append(log_entry) logger.addHandler(LogBroadcastHandler()) async def main(): global NODE_NAME, main_event_loop NODE_NAME, port = get_node_name() main_event_loop = asyncio.get_running_loop() server = await start_websocket_server() asyncio.create_task(process_message_queue()) asyncio.create_task(process_logs()) # Start log processor interface = create_gradio_interface() from starlette.middleware.base import BaseHTTPMiddleware class NodeNameMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): global NODE_NAME query_params = dict(request.query_params) if "node_name" in query_params: NODE_NAME = query_params["node_name"] logger.info(f"Node name set to {NODE_NAME} from URL parameter") response = await call_next(request) return response app = gr.routes.App.create_app(interface) app.add_middleware(NodeNameMiddleware) import uvicorn config = uvicorn.Config(app, host="0.0.0.0", port=port) server = uvicorn.Server(config) logger.info(f"Starting Gradio interface on http://0.0.0.0:{port} with node name '{NODE_NAME}'") await server.serve() if __name__ == "__main__": asyncio.run(main())