Spaces:
Sleeping
Sleeping
# app.py (Full Code - Including process_pdf_tab fix) | |
import streamlit as st | |
import asyncio | |
# import websockets # Removed - Not using WS for state sync | |
import uuid | |
from datetime import datetime | |
import os | |
import random | |
import time | |
import hashlib | |
import glob | |
import base64 | |
import io | |
import streamlit.components.v1 as components | |
import edge_tts # Keep for TTS | |
import nest_asyncio | |
import re | |
import pytz | |
import shutil | |
from PyPDF2 import PdfReader # Keep if PDF tab is kept | |
import threading | |
import json | |
import zipfile | |
from dotenv import load_dotenv | |
# from streamlit_marquee import streamlit_marquee # Keep import if used | |
from collections import defaultdict, Counter, deque # Use deque for action log | |
# import pandas as pd # Removed dependency | |
from streamlit_js_eval import streamlit_js_eval, sync # Use sync was removed, re-checking - use streamlit_js_eval | |
from PIL import Image # Needed for paste_image_component | |
# ============================================================================== | |
# Configuration & Constants | |
# ============================================================================== | |
# ๐ ๏ธ Patch asyncio for nesting | |
nest_asyncio.apply() | |
# ๐จ Page Config | |
st.set_page_config( | |
page_title="๐ค๐๏ธ Shared World Builder ๐", | |
page_icon="๐๏ธ", | |
layout="wide", | |
initial_sidebar_state="expanded" | |
) | |
# General Constants | |
icons = '๐ค๐๏ธ๐ฃ๏ธ๐พ' | |
Site_Name = '๐ค๐๏ธ Shared World Builder ๐ฃ๏ธ' | |
START_ROOM = "World Lobby ๐" # Used? Maybe just for chat title | |
MEDIA_DIR = "." # Base directory for general files | |
STATE_FILE = "user_state.txt" # For remembering username | |
# User/Chat Constants | |
FUN_USERNAMES = { | |
"BuilderBot ๐ค": "en-US-AriaNeural", "WorldWeaver ๐ธ๏ธ": "en-US-JennyNeural", | |
"Terraformer ๐ฑ": "en-GB-SoniaNeural", "SkyArchitect โ๏ธ": "en-AU-NatashaNeural", | |
"PixelPainter ๐จ": "en-CA-ClaraNeural", "VoxelVortex ๐ช๏ธ": "en-US-GuyNeural", | |
"CosmicCrafter โจ": "en-GB-RyanNeural", "GeoGuru ๐บ๏ธ": "en-AU-WilliamNeural", | |
"BlockBard ๐งฑ": "en-CA-LiamNeural", "SoundSculptor ๐": "en-US-AnaNeural", | |
} | |
EDGE_TTS_VOICES = list(set(FUN_USERNAMES.values())) # Define available voices | |
DEFAULT_TTS_VOICE = "en-US-AriaNeural" | |
CHAT_DIR = "chat_logs" | |
# Audio Constants | |
AUDIO_CACHE_DIR = "audio_cache" | |
AUDIO_DIR = "audio_logs" | |
# World Builder Constants | |
SAVED_WORLDS_DIR = "saved_worlds" # Directory for MD world files | |
PLOT_WIDTH = 50.0 # Needed for JS injection | |
PLOT_DEPTH = 50.0 # Needed for JS injection | |
WORLD_STATE_FILE_MD_PREFIX = "๐_" # Prefix for world save files | |
MAX_ACTION_LOG_SIZE = 30 # Max entries in sidebar action log | |
# File Emojis | |
FILE_EMOJIS = {"md": "๐", "mp3": "๐ต", "png": "๐ผ๏ธ", "mp4": "๐ฅ", "zip": "๐ฆ", "json": "๐"} | |
# Primitives Map (Tool Name -> Emoji for Radio Button Display) | |
PRIMITIVE_MAP = { | |
"Tree": "๐ณ", "Rock": "๐ฟ", "Simple House": "๐๏ธ", "Pine Tree": "๐ฒ", "Brick Wall": "๐งฑ", | |
"Sphere": "๐ต", "Cube": "๐ฆ", "Cylinder": "๐งด", "Cone": "๐ฆ", "Torus": "๐ฉ", | |
"Mushroom": "๐", "Cactus": "๐ต", "Campfire": "๐ฅ", "Star": "โญ", "Gem": "๐", | |
"Tower": "๐ผ", "Barrier": "๐ง", "Fountain": "โฒ", "Lantern": "๐ฎ", "Sign Post": "ํป" | |
} | |
# Add None option for clearing tool | |
TOOLS_MAP = {"None": "๐ซ"} | |
TOOLS_MAP.update({name: emoji for emoji, name in PRIMITIVE_MAP.items()}) # Combine for radio options | |
# --- Directories --- | |
for d in [CHAT_DIR, AUDIO_DIR, AUDIO_CACHE_DIR, SAVED_WORLDS_DIR]: | |
os.makedirs(d, exist_ok=True) | |
# --- API Keys (Placeholder) --- | |
load_dotenv() | |
# --- Global State & Locks (Removed world_objects global, only client tracking if WS re-added) --- | |
# Lock might still be useful if background tasks modify shared resources (like file lists?) - Keep for now | |
app_lock = threading.Lock() | |
# connected_clients = set() # Removed WS client tracking | |
# ============================================================================== | |
# Utility Functions | |
# ============================================================================== | |
def get_current_time_str(tz='UTC'): | |
try: timezone = pytz.timezone(tz); now_aware = datetime.now(timezone) | |
except Exception: now_aware = datetime.now(pytz.utc) | |
return now_aware.strftime('%Y%m%d_%H%M%S') | |
def clean_filename_part(text, max_len=25): | |
if not isinstance(text, str): text = "invalid_name" | |
text = re.sub(r'\s+', '_', text); text = re.sub(r'[^\w\-.]', '', text) | |
return text[:max_len] | |
def run_async(async_func, *args, **kwargs): | |
"""Runs an async function safely from a sync context using create_task or asyncio.run.""" | |
try: loop = asyncio.get_running_loop(); return loop.create_task(async_func(*args, **kwargs)) | |
except RuntimeError: | |
# print(f"Warning: Running async func {async_func.__name__} in new event loop.") | |
try: return asyncio.run(async_func(*args, **kwargs)) | |
except Exception as e: print(f"Error run_async new loop: {e}"); return None | |
except Exception as e: print(f"Error run_async schedule task: {e}"); return None | |
def ensure_dir(dir_path): os.makedirs(dir_path, exist_ok=True) | |
# ============================================================================== | |
# World State File Handling (Markdown + JSON) | |
# ============================================================================== | |
def generate_world_save_filename(username="User", world_name="World"): | |
"""Generates filename including username, world name, timestamp.""" | |
timestamp = get_current_time_str(); clean_user = clean_filename_part(username, 15); | |
clean_world = clean_filename_part(world_name, 20); | |
rand_hash = hashlib.md5(str(time.time()).encode()+username.encode()+world_name.encode()).hexdigest()[:4] | |
return f"{WORLD_STATE_FILE_MD_PREFIX}{clean_world}_by_{clean_user}_{timestamp}_{rand_hash}.md" | |
def parse_world_filename(filename): | |
"""Extracts info from filename (Name, User, Time, Hash).""" | |
basename = os.path.basename(filename) | |
if basename.startswith(WORLD_STATE_FILE_MD_PREFIX) and basename.endswith(".md"): | |
core = basename[len(WORLD_STATE_FILE_MD_PREFIX):-3]; parts = core.split('_') | |
if len(parts) >= 5 and parts[-3] == "by": | |
timestamp_str = parts[-2]; username = parts[-4]; world_name = " ".join(parts[:-4]); dt_obj = None | |
try: dt_obj = pytz.utc.localize(datetime.strptime(timestamp_str, '%Y%m%d_%H%M%S')) | |
except Exception: dt_obj = None | |
return {"name": world_name or "Untitled", "user": username, "timestamp": timestamp_str, "dt": dt_obj, "filename": filename} | |
# Fallback | |
dt_fallback = None; try: mtime = os.path.getmtime(filename); dt_fallback = datetime.fromtimestamp(mtime, tz=pytz.utc) | |
except Exception: pass | |
return {"name": basename.replace('.md','').replace(WORLD_STATE_FILE_MD_PREFIX, ''), "user": "Unknown", "timestamp": "Unknown", "dt": dt_fallback, "filename": filename} | |
def save_world_to_md(target_filename_base, world_data_dict): | |
"""Saves the provided world state dictionary to a specific MD file.""" | |
# No global state modification, just write the provided dict | |
save_path = os.path.join(SAVED_WORLDS_DIR, target_filename_base) | |
print(f"Saving {len(world_data_dict)} objects to MD file: {save_path}...") | |
success = False | |
parsed_info = parse_world_filename(save_path) # Parse final path | |
timestamp_save = get_current_time_str() | |
md_content = f"""# World State: {parsed_info['name']} by {parsed_info['user']} | |
* **File Saved:** {timestamp_save} (UTC) | |
* **Source Timestamp:** {parsed_info['timestamp']} | |
* **Objects:** {len(world_data_dict)} | |
```json | |
{json.dumps(world_data_dict, indent=2)} | |
```""" | |
try: | |
ensure_dir(SAVED_WORLDS_DIR); | |
with open(save_path, 'w', encoding='utf-8') as f: f.write(md_content) | |
print(f"World state saved successfully to {target_filename_base}"); success = True | |
except Exception as e: print(f"Error saving world state to {save_path}: {e}") | |
return success | |
def load_world_from_md(filename_base): | |
"""Loads world state dict from an MD file (basename), returns dict or None.""" | |
load_path = os.path.join(SAVED_WORLDS_DIR, filename_base) | |
print(f"Loading world state dictionary from MD file: {load_path}...") | |
if not os.path.exists(load_path): st.error(f"World file not found: {filename_base}"); return None | |
try: | |
with open(load_path, 'r', encoding='utf-8') as f: content = f.read() | |
json_match = re.search(r"```json\s*(\{[\s\S]*?\})\s*```", content, re.IGNORECASE) | |
if not json_match: st.error(f"Could not find valid JSON block in {filename_base}"); return None | |
world_data_dict = json.loads(json_match.group(1)) | |
print(f"Parsed {len(world_data_dict)} objects from {filename_base}.") | |
return world_data_dict # Return the dictionary | |
except json.JSONDecodeError as e: st.error(f"Invalid JSON found in {filename_base}: {e}"); return None | |
except Exception as e: st.error(f"Error loading world state from {filename_base}: {e}"); st.exception(e); return None | |
def get_saved_worlds(): | |
"""Scans the saved worlds directory for world MD files and parses them.""" | |
try: | |
ensure_dir(SAVED_WORLDS_DIR); | |
world_files = glob.glob(os.path.join(SAVED_WORLDS_DIR, f"{WORLD_STATE_FILE_MD_PREFIX}*.md")) | |
parsed_worlds = [parse_world_filename(f) for f in world_files] | |
parsed_worlds.sort(key=lambda x: x['dt'] if x['dt'] else datetime.min.replace(tzinfo=pytz.utc), reverse=True) | |
return parsed_worlds | |
except Exception as e: print(f"Error scanning saved worlds: {e}"); st.error(f"Could not scan saved worlds: {e}"); return [] | |
# ============================================================================== | |
# User State & Session Init | |
# ============================================================================== | |
def save_username(username): | |
try: | |
with open(STATE_FILE, 'w') as f: f.write(username) | |
except Exception as e: print(f"Failed save username: {e}") | |
def load_username(): | |
if os.path.exists(STATE_FILE): | |
try: | |
with open(STATE_FILE, 'r') as f: return f.read().strip() | |
except Exception as e: print(f"Failed load username: {e}") | |
return None | |
def init_session_state(): | |
"""Initializes Streamlit session state variables.""" | |
defaults = { | |
'last_chat_update': 0, 'message_input': "", 'audio_cache': {}, | |
'tts_voice': DEFAULT_TTS_VOICE, 'chat_history': [], 'enable_audio': True, | |
'download_link_cache': {}, 'username': None, 'autosend': False, | |
'last_message': "", | |
'selected_object': 'None', # Current building tool | |
'current_world_file': None, # Track loaded world filename (basename) | |
'new_world_name': "MyWorld", | |
'action_log': deque(maxlen=MAX_ACTION_LOG_SIZE), | |
'world_to_load_data': None, # Temp storage for state loaded from file before sending to JS | |
'js_object_placed_data': None # Temp storage for data coming from JS place event | |
} | |
for k, v in defaults.items(): | |
if k not in st.session_state: | |
if k == 'action_log': st.session_state[k] = deque(maxlen=MAX_ACTION_LOG_SIZE) | |
else: st.session_state[k] = v | |
# Ensure complex types initialized correctly if state is reloaded badly | |
if not isinstance(st.session_state.chat_history, list): st.session_state.chat_history = [] | |
if not isinstance(st.session_state.audio_cache, dict): st.session_state.audio_cache = {} | |
if not isinstance(st.session_state.download_link_cache, dict): st.session_state.download_link_cache = {} | |
if not isinstance(st.session_state.action_log, deque): st.session_state.action_log = deque(maxlen=MAX_ACTION_LOG_SIZE) | |
# ============================================================================== | |
# Action Log Helper | |
# ============================================================================== | |
def add_action_log(message): | |
"""Adds a message to the session's action log.""" | |
if 'action_log' not in st.session_state: | |
st.session_state.action_log = deque(maxlen=MAX_ACTION_LOG_SIZE) | |
timestamp = datetime.now().strftime("%H:%M:%S") | |
st.session_state.action_log.appendleft(f"[{timestamp}] {message}") # Add to front | |
# ============================================================================== | |
# JS Communication Handler Function | |
# ============================================================================== | |
# Define this at the top level so streamlit_js_eval can find it | |
def handle_js_object_placed(data): # Renamed arg for clarity | |
"""Callback triggered by JS when an object is placed.""" | |
print(f"Python received object placed event data: {type(data)}") # Debug type | |
# Data from streamlit_js_eval might already be a dict if json_data='...' was NOT used | |
# Let's assume data is already a dict/list from JS object sent. | |
# If it comes as a string, we need json.loads() | |
processed_data = None | |
if isinstance(data, str): | |
try: | |
processed_data = json.loads(data) | |
except json.JSONDecodeError: | |
print("Failed to decode JSON data from JS object place event.") | |
return False # Indicate failure | |
elif isinstance(data, dict): | |
processed_data = data # Assume it's already a dict | |
else: | |
print(f"Received unexpected data type from JS place event: {type(data)}") | |
return False | |
if processed_data and 'obj_id' in processed_data and 'type' in processed_data: | |
# Store data in session state to be processed in the main script flow | |
st.session_state.js_object_placed_data = processed_data | |
add_action_log(f"Placed {processed_data.get('type', 'object')} ({processed_data.get('obj_id', 'N/A')[:6]}...)") | |
else: | |
print("Received invalid object placement data structure from JS.") | |
return False | |
return True # Acknowledge receipt to JS | |
# ============================================================================== | |
# Audio / TTS / Chat / File Handling Helpers (Keep implementations) | |
# ============================================================================== | |
# --- Placeholder for brevity --- | |
def clean_text_for_tts(text): # ... implementation ... | |
if not isinstance(text, str): return "No text" | |
text = re.sub(r'\[([^\]]+)\]\([^\)]+\)', r'\1', text); text = re.sub(r'[#*_`!]', '', text) | |
text = ' '.join(text.split()); return text[:250] or "No text" | |
def create_file(content, username, file_type="md", save_path=None): # ... implementation ... | |
if not save_path: filename = generate_filename(content, username, file_type); save_path = os.path.join(MEDIA_DIR, filename) | |
ensure_dir(os.path.dirname(save_path)) | |
try: | |
with open(save_path, 'w', encoding='utf-8') as f: f.write(content); return save_path | |
except Exception as e: print(f"Error creating file {save_path}: {e}"); return None | |
def get_download_link(file_path, file_type="md"): # ... implementation ... | |
if not file_path or not os.path.exists(file_path): basename = os.path.basename(file_path) if file_path else "N/A"; return f"<small>Not found: {basename}</small>" | |
try: mtime = os.path.getmtime(file_path) | |
except OSError: mtime = 0 | |
cache_key = f"dl_{file_path}_{mtime}"; | |
if 'download_link_cache' not in st.session_state: st.session_state.download_link_cache = {} | |
if cache_key not in st.session_state.download_link_cache: | |
try: | |
with open(file_path, "rb") as f: b64 = base64.b64encode(f.read()).decode() | |
mime_types = {"md": "text/markdown", "mp3": "audio/mpeg", "png": "image/png", "mp4": "video/mp4", "zip": "application/zip", "json": "application/json"} | |
basename = os.path.basename(file_path) | |
link_html = f'<a href="data:{mime_types.get(file_type, "application/octet-stream")};base64,{b64}" download="{basename}" title="Download {basename}">{FILE_EMOJIS.get(file_type, "๐")}</a>' | |
st.session_state.download_link_cache[cache_key] = link_html | |
except Exception as e: print(f"Error generating DL link for {file_path}: {e}"); return f"<small>Err</small>" | |
return st.session_state.download_link_cache.get(cache_key, "<small>CacheErr</small>") | |
async def async_edge_tts_generate(text, voice, username): # ... implementation ... | |
if not text: return None | |
cache_key = hashlib.md5(f"{text[:150]}_{voice}".encode()).hexdigest(); | |
if 'audio_cache' not in st.session_state: st.session_state.audio_cache = {} | |
cached_path = st.session_state.audio_cache.get(cache_key); | |
if cached_path and os.path.exists(cached_path): return cached_path | |
text_cleaned = clean_text_for_tts(text); | |
if not text_cleaned or text_cleaned == "No text": return None | |
filename_base = generate_filename(text_cleaned, username, "mp3"); save_path = os.path.join(AUDIO_DIR, filename_base); | |
ensure_dir(AUDIO_DIR) | |
try: | |
communicate = edge_tts.Communicate(text_cleaned, voice); await communicate.save(save_path); | |
if os.path.exists(save_path) and os.path.getsize(save_path) > 0: st.session_state.audio_cache[cache_key] = save_path; return save_path | |
else: print(f"Audio file {save_path} failed generation."); return None | |
except Exception as e: print(f"Edge TTS Error: {e}"); return None | |
def play_and_download_audio(file_path): # ... implementation ... | |
if file_path and os.path.exists(file_path): | |
try: st.audio(file_path); file_type = file_path.split('.')[-1]; st.markdown(get_download_link(file_path, file_type), unsafe_allow_html=True) | |
except Exception as e: st.error(f"Audio display error for {os.path.basename(file_path)}: {e}") | |
async def save_chat_entry(username, message, voice, is_markdown=False): # ... implementation ... | |
if not message.strip(): return None, None | |
timestamp_str = get_current_time_str(); entry = f"[{timestamp_str}] {username} ({voice}): {message}" if not is_markdown else f"[{timestamp_str}] {username} ({voice}):\n```markdown\n{message}\n```" | |
md_filename_base = generate_filename(message, username, "md"); md_file_path = os.path.join(CHAT_DIR, md_filename_base); | |
md_file = create_file(entry, username, "md", save_path=md_file_path) | |
if 'chat_history' not in st.session_state: st.session_state.chat_history = []; | |
st.session_state.chat_history.append(entry) | |
audio_file = None; | |
if st.session_state.get('enable_audio', True): tts_message = message ; audio_file = await async_edge_tts_generate(tts_message, voice, username) | |
return md_file, audio_file | |
async def load_chat_history(): # ... implementation ... | |
if 'chat_history' not in st.session_state: st.session_state.chat_history = [] | |
if not st.session_state.chat_history: | |
ensure_dir(CHAT_DIR); print("Loading chat history from files...") | |
chat_files = sorted(glob.glob(os.path.join(CHAT_DIR, "*.md")), key=os.path.getmtime); loaded_count = 0 | |
temp_history = [] | |
for f_path in chat_files: | |
try: | |
with open(f_path, 'r', encoding='utf-8') as file: temp_history.append(file.read().strip()); loaded_count += 1 | |
except Exception as e: print(f"Err read chat {f_path}: {e}") | |
st.session_state.chat_history = temp_history | |
print(f"Loaded {loaded_count} chat entries from files.") | |
return st.session_state.chat_history | |
def create_zip_of_files(files_to_zip, prefix="Archive"): # ... implementation ... | |
if not files_to_zip: st.warning("No files provided to zip."); return None | |
timestamp = format_timestamp_prefix(f"Zip_{prefix}"); zip_name = f"{prefix}_{timestamp}.zip" | |
try: | |
print(f"Creating zip: {zip_name}..."); | |
with zipfile.ZipFile(zip_name, 'w', zipfile.ZIP_DEFLATED) as z: | |
for f in files_to_zip: | |
if os.path.exists(f): z.write(f, os.path.basename(f)) | |
else: print(f"Skip zip missing: {f}") | |
print("Zip success."); st.success(f"Created {zip_name}"); return zip_name | |
except Exception as e: print(f"Zip failed: {e}"); st.error(f"Zip failed: {e}"); return None | |
def delete_files(file_patterns, exclude_files=None): # ... implementation ... | |
protected = [STATE_FILE, "app.py", "index.html", "requirements.txt", "README.md"] | |
current_world_base = st.session_state.get('current_world_file') | |
if current_world_base: protected.append(current_world_base) | |
if exclude_files: protected.extend(exclude_files) | |
deleted_count = 0; errors = 0 | |
for pattern in file_patterns: | |
pattern_path = pattern | |
print(f"Attempting to delete files matching: {pattern_path}") | |
try: | |
files_to_delete = glob.glob(pattern_path) | |
if not files_to_delete: print(f"No files found for pattern: {pattern}"); continue | |
for f_path in files_to_delete: | |
basename = os.path.basename(f_path) | |
if os.path.isfile(f_path) and basename not in protected: | |
try: os.remove(f_path); print(f"Deleted: {f_path}"); deleted_count += 1 | |
except Exception as e: print(f"Failed delete {f_path}: {e}"); errors += 1 | |
elif os.path.isdir(f_path): print(f"Skipping directory: {f_path}") | |
except Exception as glob_e: print(f"Error matching pattern {pattern}: {glob_e}"); errors += 1 | |
msg = f"Deleted {deleted_count} files."; | |
if errors > 0: msg += f" Encountered {errors} errors."; st.warning(msg) | |
elif deleted_count > 0: st.success(msg) | |
else: st.info("No matching files found to delete.") | |
st.session_state['download_link_cache'] = {}; st.session_state['audio_cache'] = {} | |
async def save_pasted_image(image, username): # ... implementation ... | |
if not image: return None | |
try: img_hash = hashlib.md5(image.tobytes()).hexdigest()[:8]; timestamp = format_timestamp_prefix(username); filename = f"{timestamp}_pasted_{img_hash}.png"; filepath = os.path.join(MEDIA_DIR, filename); image.save(filepath, "PNG"); print(f"Pasted image saved: {filepath}"); return filepath | |
except Exception as e: print(f"Failed image save: {e}"); return None | |
def paste_image_component(): # ... implementation ... | |
pasted_img = None; img_type = None | |
paste_input_value = st.text_area("Paste Image Data Here", key="paste_input_area", height=50, value=st.session_state.get('paste_image_base64_input', "")) | |
if st.button("Process Pasted Image ๐", key="process_paste_button"): | |
st.session_state.paste_image_base64_input = paste_input_value | |
if paste_input_value and paste_input_value.startswith('data:image'): | |
try: | |
mime_type = paste_input_value.split(';')[0].split(':')[1]; base64_str = paste_input_value.split(',')[1]; img_bytes = base64.b64decode(base64_str); pasted_img = Image.open(io.BytesIO(img_bytes)); img_type = mime_type.split('/')[1] | |
st.image(pasted_img, caption=f"Pasted ({img_type.upper()})", width=150); st.session_state.paste_image_base64 = base64_str | |
st.session_state.paste_image_base64_input = "" | |
st.rerun() | |
except ImportError: st.error("Pillow library needed for image pasting.") | |
except Exception as e: st.error(f"Img decode err: {e}"); st.session_state.paste_image_base64 = ""; st.session_state.paste_image_base64_input = paste_input_value | |
else: st.warning("No valid image data pasted."); st.session_state.paste_image_base64 = ""; st.session_state.paste_image_base64_input = paste_input_value | |
processed_b64 = st.session_state.get('paste_image_base64', '') | |
if processed_b64: | |
try: img_bytes = base64.b64decode(processed_b64); return Image.open(io.BytesIO(img_bytes)) | |
except Exception: return None | |
return None | |
class AudioProcessor: # ... implementation ... | |
def __init__(self): self.cache_dir=AUDIO_CACHE_DIR; ensure_dir(self.cache_dir); self.metadata=json.load(open(f"{self.cache_dir}/metadata.json", 'r')) if os.path.exists(f"{self.cache_dir}/metadata.json") else {} | |
def _save_metadata(self): | |
try: | |
with open(f"{self.cache_dir}/metadata.json", 'w') as f: json.dump(self.metadata, f, indent=2) | |
except Exception as e: print(f"Failed metadata save: {e}") | |
async def create_audio(self, text, voice='en-US-AriaNeural'): | |
cache_key=hashlib.md5(f"{text[:150]}:{voice}".encode()).hexdigest(); cache_path=os.path.join(self.cache_dir, f"{cache_key}.mp3"); | |
if cache_key in self.metadata and os.path.exists(cache_path): return cache_path | |
text_cleaned=clean_text_for_tts(text); | |
if not text_cleaned: return None | |
ensure_dir(os.path.dirname(cache_path)) | |
try: | |
communicate=edge_tts.Communicate(text_cleaned,voice); await communicate.save(cache_path) | |
if os.path.exists(cache_path) and os.path.getsize(cache_path) > 0: self.metadata[cache_key]={'timestamp': datetime.now().isoformat(), 'text_length': len(text_cleaned), 'voice': voice}; self._save_metadata(); return cache_path | |
else: return None | |
except Exception as e: print(f"TTS Create Audio Error: {e}"); return None | |
def process_pdf_tab(pdf_file, max_pages, voice): # ... implementation ... | |
st.subheader("PDF Processing Results"); | |
if pdf_file is None: st.info("Upload a PDF file and click 'Process PDF' to begin."); return | |
audio_processor = AudioProcessor(); | |
try: | |
reader=PdfReader(pdf_file); | |
if reader.is_encrypted: st.warning("PDF is encrypted."); return | |
total_pages=min(len(reader.pages),max_pages); | |
st.write(f"Processing first {total_pages} pages of '{pdf_file.name}'..."); | |
texts, audios={}, {}; page_threads = []; results_lock = threading.Lock() | |
def process_page_sync(page_num, page_text): | |
async def run_async_audio(): return await audio_processor.create_audio(page_text, voice) | |
try: audio_path = asyncio.run(run_async_audio()) | |
if audio_path: | |
with results_lock: audios[page_num] = audio_path | |
except Exception as page_e: print(f"Err process page {page_num+1}: {page_e}") | |
for i in range(total_pages): | |
try: page = reader.pages[i]; text = page.extract_text(); | |
if text and text.strip(): texts[i]=text; thread = threading.Thread(target=process_page_sync, args=(i, text)); page_threads.append(thread); thread.start() | |
else: texts[i] = "[No text extracted]" | |
except Exception as extract_e: texts[i] = f"[Error extract: {extract_e}]"; print(f"Error page {i+1} extract: {extract_e}") | |
progress_bar = st.progress(0.0, text="Processing pages...") | |
total_threads = len(page_threads); start_join_time = time.time() | |
while any(t.is_alive() for t in page_threads): | |
completed_threads = total_threads - sum(t.is_alive() for t in page_threads); progress = completed_threads / total_threads if total_threads > 0 else 1.0 | |
progress_bar.progress(min(progress, 1.0), text=f"Processed {completed_threads}/{total_threads} pages...") | |
if time.time() - start_join_time > 600: print("PDF processing timed out."); break | |
time.sleep(0.5) | |
progress_bar.progress(1.0, text="Processing complete.") | |
for i in range(total_pages): | |
with st.expander(f"Page {i+1}"): | |
st.markdown(texts.get(i, "[Error getting text]")) | |
audio_file = audios.get(i) | |
if audio_file: play_and_download_audio(audio_file) | |
else: st.caption("Audio generation failed or was skipped.") | |
except ImportError: st.error("PyPDF2 library needed for PDF processing.") | |
except Exception as pdf_e: st.error(f"Err read PDF: {pdf_e}"); st.exception(pdf_e) | |
# ============================================================================== | |
# Streamlit UI Layout Functions | |
# ============================================================================== | |
def render_sidebar(): | |
"""Renders the Streamlit sidebar contents.""" | |
with st.sidebar: | |
st.header("๐พ World Management") | |
# --- World Save --- | |
current_file = st.session_state.get('current_world_file') | |
current_world_name = "Live State" | |
if current_file: | |
parsed = parse_world_filename(os.path.join(SAVED_WORLDS_DIR, current_file)) # Parse full path for name | |
current_world_name = parsed.get("name", current_file) | |
# Input for Save name (used for both Save and Save As New) | |
world_save_name = st.text_input( | |
"World Name for Save:", | |
key="world_save_name_input", | |
value=current_world_name if current_file else st.session_state.get('new_world_name', 'MyWorld'), | |
help="Enter name to save as new, or keep current name to overwrite." | |
) | |
if st.button("๐พ Save Current World View", key="sidebar_save_world", help="Saves the current 3D view state to a file."): | |
if not world_save_name.strip(): | |
st.warning("Please enter a World Name before saving.") | |
else: | |
with st.spinner("Requesting world state & saving..."): | |
js_world_state_str = streamlit_js_eval("getWorldStateForSave();", key="get_world_state_sidebar_save", want_result=True) | |
if js_world_state_str: | |
try: | |
world_data_dict = json.loads(js_world_state_str) | |
if isinstance(world_data_dict, dict): | |
# Decide filename: Overwrite if name matches current loaded file's derived name, else new file | |
filename_to_save = "" | |
is_overwrite = False | |
if current_file: | |
parsed_current = parse_world_filename(os.path.join(SAVED_WORLDS_DIR, current_file)) | |
# Check if input name matches the name part of the current file | |
if world_save_name == parsed_current.get('name', ''): | |
filename_to_save = current_file | |
is_overwrite = True | |
if not filename_to_save: # Create new filename if not overwriting | |
filename_to_save = generate_world_save_filename(st.session_state.username, world_save_name) | |
if save_world_to_md(filename_to_save, world_data_dict): | |
action = "Overwritten" if is_overwrite else "Saved new" | |
st.success(f"World {action}: {filename_to_save}") | |
add_action_log(f"Saved world: {filename_to_save}") | |
st.session_state.current_world_file = filename_to_save # Track saved file | |
st.session_state.new_world_name = "MyWorld" # Reset default potentially | |
# Clear the input only if it was a new save? Optional. | |
# st.session_state.world_save_name_input = current_world_name if is_overwrite else "MyWorld" | |
st.rerun() # Refresh sidebar list | |
else: st.error("Failed to save world state.") | |
else: st.error("Invalid state format from client.") | |
except json.JSONDecodeError: st.error("Failed to decode state from client.") | |
except Exception as e: st.error(f"Save error: {e}") | |
else: st.warning("Did not receive world state from client.") | |
# --- World Load --- | |
st.markdown("---") | |
st.header("๐ Load World") | |
saved_worlds = get_saved_worlds() | |
if not saved_worlds: st.caption("No saved worlds found.") | |
else: | |
st.caption("Click button to load state.") | |
cols_header = st.columns([3, 1, 1]) | |
with cols_header[0]: st.write("**Name** (User, Time)") | |
with cols_header[1]: st.write("**Load**") | |
with cols_header[2]: st.write("**DL**") | |
display_limit = 15; displayed_count = 0 | |
for world_info in saved_worlds: | |
f_basename = os.path.basename(world_info['filename']) | |
f_fullpath = os.path.join(SAVED_WORLDS_DIR, f_basename) | |
display_name = world_info.get('name', f_basename); user = world_info.get('user', 'N/A'); timestamp = world_info.get('timestamp', 'N/A') | |
display_text = f"{display_name} ({user}, {timestamp})" | |
# Use expander for items beyond limit | |
container = st | |
if displayed_count >= display_limit: | |
if 'expander_open' not in st.session_state: st.session_state.expander_open = False | |
container = st.expander(f"Show {len(saved_worlds)-display_limit} more...", expanded=st.session_state.expander_open) | |
if not container: break # Should not happen, safety | |
with container: # Display inside sidebar or expander | |
col1, col2, col3 = st.columns([3, 1, 1]) | |
with col1: st.write(f"<small>{display_text}</small>", unsafe_allow_html=True) | |
with col2: | |
# Disable button if already loaded | |
is_current = (st.session_state.get('current_world_file') == f_basename) | |
btn_load = st.button("Load", key=f"load_{f_basename}", help=f"Load {f_basename}", disabled=is_current) | |
with col3: st.markdown(get_download_link(f_fullpath, "md"), unsafe_allow_html=True) | |
if btn_load: # Handle click if not disabled | |
print(f"Load button clicked for: {f_basename}") | |
world_dict = load_world_from_md(f_basename) | |
if world_dict is not None: | |
st.session_state.world_to_load_data = world_dict | |
st.session_state.current_world_file = f_basename | |
add_action_log(f"Loading world: {f_basename}") | |
st.rerun() | |
else: st.error(f"Failed to parse world file: {f_basename}") | |
displayed_count += 1 | |
if displayed_count >= display_limit and len(saved_worlds) > display_limit: | |
# Toggle expander state if needed for next items - complex logic, maybe remove expander for simplicity? | |
# For now, just rely on the single expander after the limit is hit. | |
pass | |
# --- Build Tools --- | |
st.markdown("---") | |
st.header("๐๏ธ Build Tools") | |
st.caption("Select an object to place.") | |
tool_options = list(TOOLS_MAP.keys()) | |
current_tool_name = st.session_state.get('selected_object', 'None') | |
try: tool_index = tool_options.index(current_tool_name) | |
except ValueError: tool_index = 0 # Default to None/index 0 | |
selected_tool = st.radio( | |
"Select Tool:", options=tool_options, index=tool_index, | |
format_func=lambda name: f"{TOOLS_MAP.get(name, '')} {name}", | |
key="tool_selector_radio", horizontal=True | |
) | |
if selected_tool != current_tool_name: | |
st.session_state.selected_object = selected_tool | |
add_action_log(f"Selected tool: {selected_tool}") | |
run_async(lambda tool=selected_tool: streamlit_js_eval(f"updateSelectedObjectType({json.dumps(tool)});", key=f"update_tool_js_{tool}")) | |
st.rerun() | |
# --- Action Log --- | |
st.markdown("---") | |
st.header("๐ Action Log") | |
log_container = st.container(height=200) | |
with log_container: | |
log_entries = st.session_state.get('action_log', []) | |
if log_entries: st.code('\n'.join(log_entries), language="log") | |
else: st.caption("No actions recorded yet.") | |
# --- Voice/User --- | |
st.markdown("---") | |
st.header("๐ฃ๏ธ Voice & User") | |
current_username = st.session_state.get('username', "DefaultUser") | |
username_options = list(FUN_USERNAMES.keys()) if FUN_USERNAMES else [current_username] | |
current_index = 0; | |
if current_username in username_options: try: current_index = username_options.index(current_username) | |
except ValueError: pass | |
new_username = st.selectbox("Change Name/Voice", options=username_options, index=current_index, key="username_select", format_func=lambda x: x.split(" ")[0]) | |
if new_username != st.session_state.username: | |
st.session_state.username = new_username; st.session_state.tts_voice = FUN_USERNAMES.get(new_username, DEFAULT_TTS_VOICE); save_username(st.session_state.username) | |
add_action_log(f"Username changed to {new_username}") | |
st.rerun() | |
st.session_state['enable_audio'] = st.toggle("Enable TTS Audio", value=st.session_state.get('enable_audio', True)) | |
def render_main_content(): | |
"""Renders the main content area with tabs.""" | |
st.title(f"{Site_Name} - User: {st.session_state.username}") | |
# Check if world data needs to be sent to JS | |
world_data_to_load = st.session_state.pop('world_to_load_data', None) | |
if world_data_to_load is not None: | |
print(f"Sending loaded world state ({len(world_data_to_load)} objects) to JS...") | |
# Use sync version of streamlit_js_eval if we need confirmation or it simplifies flow | |
# However, just calling the JS function might be enough if JS handles errors. | |
streamlit_js_eval(js_code=f"loadWorldState({json.dumps(world_data_to_load)});", key="load_world_js") | |
st.toast("World loaded in 3D view.", icon="๐") | |
# Set up the mechanism for JS to call Python when an object is placed | |
streamlit_js_eval( | |
js_code=""" | |
if (!window.sendPlacedObjectToPython) { | |
window.sendPlacedObjectToPython = (objectData) => { | |
// Call Python function 'handle_js_object_placed' via its unique key | |
streamlit_js_eval(`handle_js_object_placed(${JSON.stringify(objectData)})`, key='js_place_event_handler'); | |
} | |
} | |
""", | |
key="setup_js_place_event_handler" # Key for the setup code itself | |
) | |
# Check if the Python handler function was triggered in the previous interaction | |
# Note: The Python function `handle_js_object_placed` should have been defined globally | |
# and it likely stored data in st.session_state.js_object_placed_data | |
placed_data = st.session_state.pop('js_object_placed_data', None) # Use pop to consume | |
if placed_data: | |
print(f"Processed stored placed object data: {placed_data.get('obj_id')}") | |
# Action log already added in handle_js_object_placed. | |
# No further action needed here in this simplified model (no server state update). | |
pass | |
# Define Tabs | |
tab_world, tab_chat, tab_pdf, tab_files = st.tabs(["๐๏ธ World Builder", "๐ฃ๏ธ Chat", "๐ PDF Tools", "๐ Files & Settings"]) | |
# --- World Builder Tab --- | |
with tab_world: | |
st.header("Shared 3D World") | |
st.caption("Place objects using sidebar tools. Use Sidebar/Files tab to Save/Load.") | |
current_file_basename = st.session_state.get('current_world_file', None) | |
if current_file_basename: | |
full_path = os.path.join(SAVED_WORLDS_DIR, current_file_basename) | |
if os.path.exists(full_path): parsed = parse_world_filename(full_path); st.info(f"Current World: **{parsed['name']}** (`{current_file_basename}`)") | |
else: st.warning(f"Loaded file '{current_file_basename}' missing."); st.session_state.current_world_file = None | |
else: st.info("Live State Active (Save to persist)") | |
# Embed HTML Component | |
html_file_path = 'index.html' | |
try: | |
with open(html_file_path, 'r', encoding='utf-8') as f: html_template = f.read() | |
# Inject state needed by JS | |
js_injection_script = f"""<script> | |
window.USERNAME = {json.dumps(st.session_state.username)}; | |
window.SELECTED_OBJECT_TYPE = {json.dumps(st.session_state.selected_object)}; | |
window.PLOT_WIDTH = {json.dumps(PLOT_WIDTH)}; | |
window.PLOT_DEPTH = {json.dumps(PLOT_DEPTH)}; | |
// Initial world state is now loaded via loadWorldState() JS call triggered by Python | |
console.log("Streamlit State Injected:", {{ username: window.USERNAME, selectedObject: window.SELECTED_OBJECT_TYPE }}); | |
</script>""" | |
html_content_with_state = html_template.replace('</head>', js_injection_script + '\n</head>', 1) | |
components.html(html_content_with_state, height=700, scrolling=False) | |
except FileNotFoundError: st.error(f"CRITICAL ERROR: Could not find '{html_file_path}'.") | |
except Exception as e: st.error(f"Error loading 3D component: {e}"); st.exception(e) | |
# --- Chat Tab --- | |
with tab_chat: | |
st.header(f"๐ฌ Chat") | |
chat_history_list = st.session_state.get('chat_history', []) | |
if not chat_history_list: chat_history_list = asyncio.run(load_chat_history()) | |
chat_container = st.container(height=500) | |
with chat_container: | |
if chat_history_list: st.markdown("----\n".join(reversed(chat_history_list[-50:]))) | |
else: st.caption("No chat messages yet.") | |
# Define callback for chat input clear | |
def clear_chat_input_callback(): | |
st.session_state.message_input = "" | |
message_value = st.text_input("Your Message:", key="message_input", label_visibility="collapsed") | |
send_button_clicked = st.button("Send Chat", key="send_chat_button", on_click=clear_chat_input_callback) | |
# Removed autosend for simplicity | |
if send_button_clicked: # Process only on button click now | |
message_to_send = message_value # Value before potential clear by callback | |
if message_to_send.strip() and message_to_send != st.session_state.get('last_message', ''): | |
st.session_state.last_message = message_to_send | |
voice = st.session_state.get('tts_voice', DEFAULT_TTS_VOICE) | |
# Use run_async for background tasks | |
run_async(save_chat_entry, st.session_state.username, message_to_send, voice) | |
# Rerun is handled implicitly by button click + callback | |
elif send_button_clicked: st.toast("Message empty or same as last.") | |
# --- PDF Tab --- | |
with tab_pdf: | |
st.header("๐ PDF Tools") | |
pdf_file = st.file_uploader("Upload PDF for Audio Conversion", type="pdf", key="pdf_upload") | |
max_pages = st.slider('Max Pages to Process', 1, 50, 10, key="pdf_pages") | |
if pdf_file: | |
if st.button("Process PDF to Audio", key="process_pdf_button"): | |
with st.spinner("Processing PDF... This may take time."): | |
process_pdf_tab(pdf_file, max_pages, st.session_state.tts_voice) | |
# --- Files & Settings Tab --- | |
with tab_files: | |
st.header("๐ Files & Settings") | |
st.subheader("๐พ World Management") | |
current_file_basename = st.session_state.get('current_world_file', None) | |
if current_file_basename: | |
full_path_for_parse = os.path.join(SAVED_WORLDS_DIR, current_file_basename) | |
save_label = f"Save Changes to '{current_file_basename}'" | |
if os.path.exists(full_path_for_parse): parsed = parse_world_filename(full_path_for_parse); save_label = f"Save Changes to '{parsed['name']}'" | |
if st.button(save_label, key="save_current_world_files", help=f"Overwrite '{current_file_basename}'"): | |
if not os.path.exists(full_path_for_parse): st.error(f"Cannot save, file missing.") | |
else: | |
with st.spinner("Requesting state & saving..."): | |
js_world_state_str = streamlit_js_eval("getWorldStateForSave();", key="get_world_state_save_current", want_result=True) | |
if js_world_state_str: | |
try: | |
world_data_dict = json.loads(js_world_state_str) | |
if isinstance(world_data_dict, dict): | |
if save_world_to_md(current_file_basename, world_data_dict): st.success("Current world saved!"); add_action_log(f"Saved world: {current_file_basename}") | |
else: st.error("Failed to save.") | |
else: st.error("Invalid format from client.") | |
except json.JSONDecodeError: st.error("Failed to decode state from client.") | |
else: st.warning("Did not receive world state from client.") | |
else: st.info("Load a world or use 'Save As New Version' below.") | |
st.subheader("Save As New Version") | |
new_name_files = st.text_input("World Name:", key="new_world_name_files_tab", value=st.session_state.get('new_world_name', 'MyWorld')) | |
if st.button("๐พ Save Current View as New Version", key="save_new_version_files"): | |
if new_name_files.strip(): | |
with st.spinner(f"Requesting state & saving as '{new_name_files}'..."): | |
js_world_state_str = streamlit_js_eval("getWorldStateForSave();", key="get_world_state_save_new", want_result=True) | |
if js_world_state_str: | |
try: | |
world_data_dict = json.loads(js_world_state_str) | |
if isinstance(world_data_dict, dict): | |
new_filename_base = generate_world_save_filename(st.session_state.username, new_name_files) | |
if save_world_to_md(new_filename_base, world_data_dict): | |
st.success(f"Saved as {new_filename_base}") | |
st.session_state.current_world_file = new_filename_base; st.session_state.new_world_name = "MyWorld"; | |
add_action_log(f"Saved new world: {new_filename_base}") | |
st.rerun() | |
else: st.error("Failed to save new version.") | |
else: st.error("Invalid format from client.") | |
except json.JSONDecodeError: st.error("Failed to decode state from client.") | |
else: st.warning("Did not receive world state from client.") | |
else: st.warning("Please enter a name.") | |
# Removed Server Status Section | |
st.subheader("๐๏ธ Delete Files") | |
st.warning("Deletion is permanent!", icon="โ ๏ธ") | |
col_del1, col_del2, col_del3, col_del4 = st.columns(4) | |
with col_del1: | |
if st.button("๐๏ธ Chats", key="del_chat_md"): delete_files([os.path.join(CHAT_DIR, "*.md")]); st.session_state.chat_history = []; st.rerun() | |
with col_del2: | |
if st.button("๐๏ธ Audio", key="del_audio_mp3"): delete_files([os.path.join(AUDIO_DIR, "*.mp3"), os.path.join(AUDIO_CACHE_DIR, "*.mp3")]); st.session_state.audio_cache = {}; st.rerun() | |
with col_del3: | |
if st.button("๐๏ธ Worlds", key="del_worlds_md"): delete_files([os.path.join(SAVED_WORLDS_DIR, f"{WORLD_STATE_FILE_MD_PREFIX}*.md")]); st.session_state.current_world_file = None; st.rerun() | |
with col_del4: | |
if st.button("๐๏ธ All Gen", key="del_all_gen"): delete_files([os.path.join(CHAT_DIR, "*.md"), os.path.join(AUDIO_DIR, "*.mp3"), os.path.join(AUDIO_CACHE_DIR, "*.mp3"), os.path.join(SAVED_WORLDS_DIR, "*.md"), os.path.join(MEDIA_DIR, "*.zip")]); st.session_state.chat_history = []; st.session_state.audio_cache = {}; st.session_state.current_world_file = None; st.rerun() | |
st.subheader("๐ฆ Download Archives") | |
col_zip1, col_zip2, col_zip3 = st.columns(3) | |
with col_zip1: | |
if st.button("Zip Worlds"): create_zip_of_files(glob.glob(os.path.join(SAVED_WORLDS_DIR, "*.md")), "Worlds") | |
with col_zip2: | |
if st.button("Zip Chats"): create_zip_of_files(glob.glob(os.path.join(CHAT_DIR, "*.md")), "Chats") | |
with col_zip3: | |
if st.button("Zip Audio"): create_zip_of_files(glob.glob(os.path.join(AUDIO_DIR, "*.mp3")) + glob.glob(os.path.join(AUDIO_CACHE_DIR, "*.mp3")), "Audio") | |
zip_files = sorted(glob.glob(os.path.join(MEDIA_DIR,"*.zip")), key=os.path.getmtime, reverse=True) | |
if zip_files: | |
st.caption("Existing Zip Files:") | |
for zip_file in zip_files: st.markdown(get_download_link(zip_file, "zip"), unsafe_allow_html=True) | |
else: | |
st.caption("No zip archives found.") | |
# ============================================================================== | |
# Main Execution Logic | |
# ============================================================================== | |
def initialize_app(): | |
"""Handles session init and initial world load.""" | |
init_session_state() | |
# Load username | |
if not st.session_state.username: | |
loaded_user = load_username() | |
if loaded_user and loaded_user in FUN_USERNAMES: st.session_state.username = loaded_user; st.session_state.tts_voice = FUN_USERNAMES[loaded_user] | |
else: st.session_state.username = random.choice(list(FUN_USERNAMES.keys())) if FUN_USERNAMES else "User"; st.session_state.tts_voice = FUN_USERNAMES.get(st.session_state.username, DEFAULT_TTS_VOICE); save_username(st.session_state.username) | |
# Load initial world state (most recent) if no specific world load is pending from a button click | |
if st.session_state.get('world_to_load_data') is None and st.session_state.get('current_world_file') is None: | |
print("Attempting initial load of most recent world...") | |
saved_worlds = get_saved_worlds() | |
if saved_worlds: | |
latest_world_file_basename = os.path.basename(saved_worlds[0]['filename']) | |
print(f"Loading most recent world on startup: {latest_world_file_basename}") | |
world_dict = load_world_from_md(latest_world_file_basename) | |
if world_dict is not None: | |
st.session_state.world_to_load_data = world_dict # Set data to be sent to JS | |
st.session_state.current_world_file = latest_world_file_basename # Update selection state | |
else: print("Failed to load most recent world.") | |
else: print("No saved worlds found for initial load.") | |
if __name__ == "__main__": | |
initialize_app() # Initialize state, user, potentially queue initial world load data | |
render_sidebar() # Render sidebar UI (includes load buttons) | |
render_main_content() # Render main UI (includes logic to send loaded world data to JS) |