import streamlit as st import asyncio import websockets import uuid import argparse from datetime import datetime import os import random import time import hashlib from PIL import Image import glob import base64 import io import streamlit.components.v1 as components import edge_tts from audio_recorder_streamlit import audio_recorder import nest_asyncio import re from streamlit_paste_button import paste_image_button import pytz import shutil # Patch for nested async - sneaky fix! πβ¨ nest_asyncio.apply() # Static config - constants rule! ππ icons = 'π€π§ π¬π' START_ROOM = "Sector π" # Page setup - dressing up the window! πΌοΈπ st.set_page_config( page_title="π€π§ MMO Chat Brainππ¬", page_icon=icons, layout="wide", initial_sidebar_state="auto" ) # Funky usernames - whoβs who in the zoo with unique voices! ππΎποΈ FUN_USERNAMES = { "CosmicJester π": "en-US-AriaNeural", "PixelPanda πΌ": "en-US-JennyNeural", "QuantumQuack π¦": "en-GB-SoniaNeural", "StellarSquirrel πΏοΈ": "en-AU-NatashaNeural", "GizmoGuru βοΈ": "en-CA-ClaraNeural", "NebulaNinja π ": "en-US-GuyNeural", "ByteBuster πΎ": "en-GB-RyanNeural", "GalacticGopher π": "en-AU-WilliamNeural", "RocketRaccoon π": "en-CA-LiamNeural", "EchoElf π§": "en-US-AnaNeural", "PhantomFox π¦": "en-US-BrandonNeural", "WittyWizard π§": "en-GB-ThomasNeural", "LunarLlama π": "en-AU-FreyaNeural", "SolarSloth βοΈ": "en-CA-LindaNeural", "AstroAlpaca π¦": "en-US-ChristopherNeural", "CyberCoyote πΊ": "en-GB-ElliotNeural", "MysticMoose π¦": "en-AU-JamesNeural", "GlitchGnome π§": "en-CA-EthanNeural", "VortexViper π": "en-US-AmberNeural", "ChronoChimp π": "en-GB-LibbyNeural" } # Folders galore - organizing chaos! ππ CHAT_DIR = "chat_logs" VOTE_DIR = "vote_logs" STATE_FILE = "user_state.txt" AUDIO_DIR = "audio_logs" HISTORY_DIR = "history_logs" MEDIA_DIR = "media_files" os.makedirs(CHAT_DIR, exist_ok=True) os.makedirs(VOTE_DIR, exist_ok=True) os.makedirs(AUDIO_DIR, exist_ok=True) os.makedirs(HISTORY_DIR, exist_ok=True) os.makedirs(MEDIA_DIR, exist_ok=True) CHAT_FILE = os.path.join(CHAT_DIR, "global_chat.md") QUOTE_VOTES_FILE = os.path.join(VOTE_DIR, "quote_votes.md") MEDIA_VOTES_FILE = os.path.join(VOTE_DIR, "media_votes.md") HISTORY_FILE = os.path.join(HISTORY_DIR, "chat_history.md") # Fancy digits - numbers got style! π’π UNICODE_DIGITS = {i: f"{i}\uFE0Fβ£" for i in range(10)} # Massive font collection - typography bonanza! ποΈπ¨ UNICODE_FONTS = [ ("Normal", lambda x: x), ("Bold", lambda x: "".join(chr(ord(c) + 0x1D400 - 0x41) if 'A' <= c <= 'Z' else chr(ord(c) + 0x1D41A - 0x61) if 'a' <= c <= 'z' else c for c in x)), ("Italic", lambda x: "".join(chr(ord(c) + 0x1D434 - 0x41) if 'A' <= c <= 'Z' else chr(ord(c) + 0x1D44E - 0x61) if 'a' <= c <= 'z' else c for c in x)), ("Bold Italic", lambda x: "".join(chr(ord(c) + 0x1D468 - 0x41) if 'A' <= c <= 'Z' else chr(ord(c) + 0x1D482 - 0x61) if 'a' <= c <= 'z' else c for c in x)), ("Script", lambda x: "".join(chr(ord(c) + 0x1D49C - 0x41) if 'A' <= c <= 'Z' else chr(ord(c) + 0x1D4B6 - 0x61) if 'a' <= c <= 'z' else c for c in x)), ("Bold Script", lambda x: "".join(chr(ord(c) + 0x1D4D0 - 0x41) if 'A' <= c <= 'Z' else chr(ord(c) + 0x1D4EA - 0x61) if 'a' <= c <= 'z' else c for c in x)), ("Fraktur", lambda x: "".join(chr(ord(c) + 0x1D504 - 0x41) if 'A' <= c <= 'Z' else chr(ord(c) + 0x1D51E - 0x61) if 'a' <= c <= 'z' else c for c in x)), ("Bold Fraktur", lambda x: "".join(chr(ord(c) + 0x1D56C - 0x41) if 'A' <= c <= 'Z' else chr(ord(c) + 0x1D586 - 0x61) if 'a' <= c <= 'z' else c for c in x)), ("Double Struck", lambda x: "".join(chr(ord(c) + 0x1D538 - 0x41) if 'A' <= c <= 'Z' else chr(ord(c) + 0x1D552 - 0x61) if 'a' <= c <= 'z' else c for c in x)), ("Sans Serif", lambda x: "".join(chr(ord(c) + 0x1D5A0 - 0x41) if 'A' <= c <= 'Z' else chr(ord(c) + 0x1D5BA - 0x61) if 'a' <= c <= 'z' else c for c in x)), ("Sans Serif Bold", lambda x: "".join(chr(ord(c) + 0x1D5D4 - 0x41) if 'A' <= c <= 'Z' else chr(ord(c) + 0x1D5EE - 0x61) if 'a' <= c <= 'z' else c for c in x)), ("Sans Serif Italic", lambda x: "".join(chr(ord(c) + 0x1D608 - 0x41) if 'A' <= c <= 'Z' else chr(ord(c) + 0x1D622 - 0x61) if 'a' <= c <= 'z' else c for c in x)), ("Sans Serif Bold Italic", lambda x: "".join(chr(ord(c) + 0x1D63C - 0x41) if 'A' <= c <= 'Z' else chr(ord(c) + 0x1D656 - 0x61) if 'a' <= c <= 'z' else c for c in x)), ("Monospace", lambda x: "".join(chr(ord(c) + 0x1D670 - 0x41) if 'A' <= c <= 'Z' else chr(ord(c) + 0x1D68A - 0x61) if 'a' <= c <= 'z' else c for c in x)), ("Circled", lambda x: "".join(chr(ord(c) - 0x41 + 0x24B6) if 'A' <= c <= 'Z' else chr(ord(c) - 0x61 + 0x24D0) if 'a' <= c <= 'z' else c for c in x)), ("Squared", lambda x: "".join(chr(ord(c) - 0x41 + 0x1F130) if 'A' <= c <= 'Z' else c for c in x)), ("Negative Circled", lambda x: "".join(chr(ord(c) - 0x41 + 0x1F150) if 'A' <= c <= 'Z' else c for c in x)), ("Negative Squared", lambda x: "".join(chr(ord(c) - 0x41 + 0x1F170) if 'A' <= c <= 'Z' else c for c in x)), ("Regional Indicator", lambda x: "".join(chr(ord(c) - 0x41 + 0x1F1E6) if 'A' <= c <= 'Z' else c for c in x)), ] # Global state - keeping tabs! ππ if 'server_running' not in st.session_state: st.session_state.server_running = False if 'server_task' not in st.session_state: st.session_state.server_task = None if 'active_connections' not in st.session_state: st.session_state.active_connections = {} if 'media_notifications' not in st.session_state: st.session_state.media_notifications = [] if 'last_chat_update' not in st.session_state: st.session_state.last_chat_update = 0 if 'displayed_chat_lines' not in st.session_state: st.session_state.displayed_chat_lines = [] if 'old_val' not in st.session_state: st.session_state.old_val = "" if 'last_query' not in st.session_state: st.session_state.last_query = "" if 'message_text' not in st.session_state: st.session_state.message_text = "" if 'audio_cache' not in st.session_state: st.session_state.audio_cache = {} if 'pasted_image_data' not in st.session_state: st.session_state.pasted_image_data = None if 'quote_line' not in st.session_state: st.session_state.quote_line = None if 'refresh_rate' not in st.session_state: st.session_state.refresh_rate = 5 if 'base64_cache' not in st.session_state: st.session_state.base64_cache = {} if 'transcript_history' not in st.session_state: st.session_state.transcript_history = [] if 'last_transcript' not in st.session_state: st.session_state.last_transcript = "" if 'image_hashes' not in st.session_state: st.session_state.image_hashes = set() # Timestamp wizardry - clock ticks with flair! β°π© def format_timestamp_prefix(username): central = pytz.timezone('US/Central') now = datetime.now(central) return f"{now.strftime('%I-%M-%p-ct-%m-%d-%Y')}-by-{username}" # Compute image hash from binary data def compute_image_hash(image_data): if isinstance(image_data, Image.Image): img_byte_arr = io.BytesIO() image_data.save(img_byte_arr, format='PNG') img_bytes = img_byte_arr.getvalue() else: img_bytes = image_data return hashlib.md5(img_bytes).hexdigest()[:8] # Node naming - christening the beast! ππΌ 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) parser.add_argument('--port', type=int, default=8501) args = parser.parse_args() username = st.session_state.get('username', 'System π') log_action(username, "ππΌ - Node naming - christening the beast!") return args.node_name or f"node-{uuid.uuid4().hex[:8]}", args.port # Action logger - spying on deeds! π΅οΈπ def log_action(username, action): if 'action_log' not in st.session_state: st.session_state.action_log = {} user_log = st.session_state.action_log.setdefault(username, {}) current_time = time.time() user_log = {k: v for k, v in user_log.items() if current_time - v < 10} st.session_state.action_log[username] = user_log if action not in user_log: central = pytz.timezone('US/Central') with open(HISTORY_FILE, 'a') as f: f.write(f"[{datetime.now(central).strftime('%Y-%m-%d %H:%M:%S')}] {username}: {action}\n") user_log[action] = current_time # Clean text - strip the fancy stuff! π§Ήπ def clean_text_for_tts(text): cleaned = re.sub(r'[#*!\[\]]+', '', text) cleaned = ' '.join(cleaned.split()) return cleaned[:200] if cleaned else "No text to speak" # Chat saver - words locked tight! π¬π async def save_chat_entry(username, message, is_markdown=False): await asyncio.to_thread(log_action, username, "π¬π - Chat saver - words locked tight!") central = pytz.timezone('US/Central') timestamp = datetime.now(central).strftime("%Y-%m-%d %H:%M:%S") if is_markdown: entry = f"[{timestamp}] {username}:\n```markdown\n{message}\n```" else: entry = f"[{timestamp}] {username}: {message}" await asyncio.to_thread(lambda: open(CHAT_FILE, 'a').write(f"{entry}\n")) voice = FUN_USERNAMES.get(username, "en-US-AriaNeural") cleaned_message = clean_text_for_tts(message) audio_file = await async_edge_tts_generate(cleaned_message, voice) if audio_file: with open(HISTORY_FILE, 'a') as f: f.write(f"[{timestamp}] {username}: Audio generated - {audio_file}\n") await broadcast_message(f"{username}|{message}", "chat") st.session_state.last_chat_update = time.time() return audio_file # Chat loader - history unleashed! ππ async def load_chat(): username = st.session_state.get('username', 'System π') await asyncio.to_thread(log_action, username, "ππ - Chat loader - history unleashed!") if not os.path.exists(CHAT_FILE): await asyncio.to_thread(lambda: open(CHAT_FILE, 'a').write(f"# {START_ROOM} Chat\n\nWelcome to the cosmic hub - start chatting! π€\n")) with open(CHAT_FILE, 'r') as f: content = await asyncio.to_thread(f.read) return content # User lister - whoβs in the gang! π₯π async def get_user_list(chat_content): username = st.session_state.get('username', 'System π') await asyncio.to_thread(log_action, username, "π₯π - User lister - whoβs in the gang!") users = set() for line in chat_content.split('\n'): if line.strip() and ': ' in line: user = line.split(': ')[1].split(' ')[0] users.add(user) return sorted(list(users)) # Join checker - been here before? πͺπ async def has_joined_before(client_id, chat_content): username = st.session_state.get('username', 'System π') await asyncio.to_thread(log_action, username, "πͺπ - Join checker - been here before?") return any(f"Client-{client_id}" in line for line in chat_content.split('\n')) # Suggestion maker - old quips resurface! π‘π async def get_message_suggestions(chat_content, prefix): username = st.session_state.get('username', 'System π') await asyncio.to_thread(log_action, username, "π‘π - Suggestion maker - old quips resurface!") lines = chat_content.split('\n') messages = [line.split(': ', 1)[1] for line in lines if ': ' in line and line.strip()] return [msg for msg in messages if msg.lower().startswith(prefix.lower())][:5] # Vote saver - cheers recorded! ππ async def save_vote(file, item, user_hash, username, comment=""): await asyncio.to_thread(log_action, username, "ππ - Vote saver - cheers recorded!") central = pytz.timezone('US/Central') timestamp = datetime.now(central).strftime("%Y-%m-%d %H:%M:%S") entry = f"[{timestamp}] {user_hash} voted for {item}" await asyncio.to_thread(lambda: open(file, 'a').write(f"{entry}\n")) await asyncio.to_thread(lambda: open(HISTORY_FILE, "a").write(f"- {timestamp} - User {user_hash} voted for {item}\n")) chat_message = f"{username} upvoted: \"{item}\"" if comment: chat_message += f" - {comment}" await save_chat_entry(username, chat_message) # Vote counter - tallying the love! ππ async def load_votes(file): username = st.session_state.get('username', 'System π') await asyncio.to_thread(log_action, username, "ππ - Vote counter - tallying the love!") if not os.path.exists(file): await asyncio.to_thread(lambda: open(file, 'w').write("# Vote Tally\n\nNo votes yet - get clicking! π±οΈ\n")) with open(file, 'r') as f: content = await asyncio.to_thread(f.read) lines = content.strip().split('\n')[2:] votes = {} user_votes = set() for line in lines: if line.strip() and 'voted for' in line: user_hash = line.split('] ')[1].split(' voted for ')[0] item = line.split('voted for ')[1] vote_key = f"{user_hash}-{item}" if vote_key not in user_votes: votes[item] = votes.get(item, 0) + 1 user_votes.add(vote_key) return votes # Hash generator - secret codes ahoy! ππ΅οΈ async def generate_user_hash(): username = st.session_state.get('username', 'System π') await asyncio.to_thread(log_action, username, "ππ΅οΈ - Hash generator - secret codes ahoy!") if 'user_hash' not in st.session_state: st.session_state.user_hash = hashlib.md5(str(random.getrandbits(128)).encode()).hexdigest()[:8] return st.session_state.user_hash # Audio maker - voices come alive! πΆπ async def async_edge_tts_generate(text, voice, rate=0, pitch=0, file_format="mp3"): username = st.session_state.get('username', 'System π') await asyncio.to_thread(log_action, username, "πΆπ - Audio maker - voices come alive!") timestamp = format_timestamp_prefix(username) filename = f"{timestamp}.{file_format}" filepath = os.path.join(AUDIO_DIR, filename) communicate = edge_tts.Communicate(text, voice, rate=f"{rate:+d}%", pitch=f"{pitch:+d}Hz") try: await communicate.save(filepath) return filepath if os.path.exists(filepath) else None except edge_tts.exceptions.NoAudioReceived: with open(HISTORY_FILE, 'a') as f: central = pytz.timezone('US/Central') f.write(f"[{datetime.now(central).strftime('%Y-%m-%d %H:%M:%S')}] {username}: Audio failed - No audio received for '{text}'\n") return None # Audio player - tunes blast off! ππ def play_and_download_audio(file_path): if file_path and os.path.exists(file_path): st.audio(file_path) if file_path not in st.session_state.base64_cache: with open(file_path, "rb") as f: b64 = base64.b64encode(f.read()).decode() st.session_state.base64_cache[file_path] = b64 b64 = st.session_state.base64_cache[file_path] dl_link = f'π΅ Download {os.path.basename(file_path)}' st.markdown(dl_link, unsafe_allow_html=True) # Image saver - pics preserved with naming! πΈπΎ async def save_pasted_image(image, username): await asyncio.to_thread(log_action, username, "πΈπΎ - Image saver - pics preserved!") img_hash = compute_image_hash(image) if img_hash in st.session_state.image_hashes: return None timestamp = format_timestamp_prefix(username) filename = f"{timestamp}-{img_hash}.png" filepath = os.path.join(MEDIA_DIR, filename) await asyncio.to_thread(image.save, filepath, "PNG") st.session_state.image_hashes.add(img_hash) return filepath # Video renderer - movies roll with autoplay! π₯π¬ async def get_video_html(video_path, width="100%"): username = st.session_state.get('username', 'System π') await asyncio.to_thread(log_action, username, "π₯π¬ - Video renderer - movies roll!") with open(video_path, 'rb') as f: video_data = await asyncio.to_thread(f.read) video_url = f"data:video/mp4;base64,{base64.b64encode(video_data).decode()}" return f'' # Audio renderer - sounds soar! πΆβοΈ async def get_audio_html(audio_path, width="100%"): username = st.session_state.get('username', 'System π') await asyncio.to_thread(log_action, username, "πΆβοΈ - Audio renderer - sounds soar!") audio_url = f"data:audio/mpeg;base64,{base64.b64encode(await asyncio.to_thread(open, audio_path, 'rb').read()).decode()}" return f'' # Websocket handler - chat links up! ππ async def websocket_handler(websocket, path): username = st.session_state.get('username', 'System π') await asyncio.to_thread(log_action, username, "ππ - Websocket handler - chat links up!") try: client_id = str(uuid.uuid4()) room_id = "chat" st.session_state.active_connections.setdefault(room_id, {})[client_id] = websocket chat_content = await load_chat() username = st.session_state.get('username', random.choice(list(FUN_USERNAMES.keys()))) if not await has_joined_before(client_id, chat_content): await save_chat_entry(f"Client-{client_id}", f"{username} has joined {START_ROOM}!") async for message in websocket: parts = message.split('|', 1) if len(parts) == 2: username, content = parts await save_chat_entry(username, content) except websockets.ConnectionClosed: pass finally: if room_id in st.session_state.active_connections and client_id in st.session_state.active_connections[room_id]: del st.session_state.active_connections[room_id][client_id] # Message broadcaster - words fly far! π’βοΈ async def broadcast_message(message, room_id): username = st.session_state.get('username', 'System π') await asyncio.to_thread(log_action, username, "π’βοΈ - Message broadcaster - words fly far!") if room_id in st.session_state.active_connections: disconnected = [] for client_id, ws in st.session_state.active_connections[room_id].items(): try: await ws.send(message) except websockets.ConnectionClosed: disconnected.append(client_id) for client_id in disconnected: del st.session_state.active_connections[room_id][client_id] # Server starter - web spins up! π₯οΈπ async def run_websocket_server(): username = st.session_state.get('username', 'System π') await asyncio.to_thread(log_action, username, "π₯οΈπ - Server starter - web spins up!") if not st.session_state.server_running: server = await websockets.serve(websocket_handler, '0.0.0.0', 8765) st.session_state.server_running = True await server.wait_closed() # Voice processor - speech to text! π€π async def process_voice_input(audio_bytes): username = st.session_state.get('username', 'System π') await asyncio.to_thread(log_action, username, "π€π - Voice processor - speech to text!") if audio_bytes: text = "Voice input simulation" await save_chat_entry(username, text) # Dummy AI lookup function (replace with actual implementation) async def perform_ai_lookup(query, vocal_summary=True, extended_refs=False, titles_summary=True, full_audio=False, useArxiv=True, useArxivAudio=False): username = st.session_state.get('username', 'System π') result = f"AI Lookup Result for '{query}' (Arxiv: {useArxiv}, Audio: {useArxivAudio})" await save_chat_entry(username, result) if useArxivAudio: audio_file = await async_edge_tts_generate(result, FUN_USERNAMES.get(username, "en-US-AriaNeural")) if audio_file: st.audio(audio_file) # Delete all user files function def delete_user_files(): protected_files = {'app.py', 'requirements.txt', 'README.md'} deleted_files = [] # Directories to clear directories = [MEDIA_DIR, AUDIO_DIR, CHAT_DIR, VOTE_DIR, HISTORY_DIR] for directory in directories: if os.path.exists(directory): for root, _, files in os.walk(directory): for file in files: file_path = os.path.join(root, file) if os.path.basename(file_path) not in protected_files: try: os.remove(file_path) deleted_files.append(file_path) except Exception as e: st.error(f"Failed to delete {file_path}: {e}") # Remove empty directories try: shutil.rmtree(directory, ignore_errors=True) os.makedirs(directory, exist_ok=True) # Recreate empty directory except Exception as e: st.error(f"Failed to remove directory {directory}: {e}") # Clear session state caches st.session_state.image_hashes.clear() st.session_state.audio_cache.clear() st.session_state.base64_cache.clear() st.session_state.displayed_chat_lines.clear() return deleted_files # ASR Component HTML ASR_HTML = """
β³ {font_func('Refresh in:')} {countdown_str}
", unsafe_allow_html=True) time.sleep(1) st.rerun() st.sidebar.subheader("Chat History π") with open(HISTORY_FILE, 'r') as f: history_content = f.read() st.sidebar.markdown(history_content) loop.run_until_complete(async_interface()) if __name__ == "__main__": main()