tracker-test / config_manager.py
umangchaudhry's picture
Upload 11 files
b03a18c verified
import json
import os
from datetime import datetime
from google_drive_manager import GoogleDriveManager
import streamlit as st
class ConfigManager:
def __init__(self):
self.config_file = "wedding_config.json"
self.data_dir = "data"
self.app_config_file = "app_config.json"
self.demo_data_dir = "demo_data"
# Check if running on Hugging Face Spaces
self.is_huggingface = os.getenv('SPACE_ID') is not None
# Initialize Google Drive manager
self.drive_manager = GoogleDriveManager()
self.google_drive_enabled = False
# For Hugging Face Spaces, use in-memory storage
if self.is_huggingface:
self.use_memory_storage = True
self.memory_data = {}
self.data_loaded_from_drive = False # Track if data has been loaded
else:
self.use_memory_storage = False
self.data_loaded_from_drive = False
# Set up data directory for local development
try:
if not os.path.exists(self.data_dir):
os.makedirs(self.data_dir, exist_ok=True)
if not os.path.exists(self.demo_data_dir):
os.makedirs(self.demo_data_dir, exist_ok=True)
except Exception as e:
print(f"Error creating directories: {e}")
# Load app configuration
self.app_config = self.load_app_config()
# Initialize Google Drive if enabled
self._initialize_google_drive()
def get_user_folder(self):
"""Get the user-specific folder name"""
# Use demo_data folder if in demo mode, otherwise use laraandumang
if self.is_demo_mode():
return "demo_data"
return "laraandumang"
def set_user_folder(self, folder_name):
"""Set the user-specific folder name"""
self.user_folder = folder_name
def get_current_user_folder(self):
"""Get the currently set user folder"""
return getattr(self, 'user_folder', self.get_user_folder())
def _initialize_google_drive(self):
"""Initialize Google Drive connection"""
try:
folder_id = os.getenv('GOOGLE_DRIVE_FOLDER_ID')
if folder_id:
if self.drive_manager.initialize(folder_id):
self.google_drive_enabled = True
# Don't automatically sync on startup - let user choose
print("Google Drive initialized successfully. Manual sync required.")
except Exception as e:
print(f"Google Drive initialization failed: {e}")
self.google_drive_enabled = False
def _sync_from_google_drive(self):
"""Sync data files from Google Drive to local storage"""
if not self.google_drive_enabled:
return
try:
# Get user-specific folder
user_folder = self.get_current_user_folder()
# List of data files to sync
data_files = [
'guest_list_data.json',
'rsvp_data.json',
'tasks.json',
'vendors.json',
'wedding_party.json'
]
# Sync config file from user folder
config_content = self.drive_manager.download_file(f'{user_folder}/wedding_config.json')
if config_content:
if self.use_memory_storage:
# Store in memory for Hugging Face Spaces
self.memory_data['wedding_config.json'] = config_content
# Update cache in session state
st.session_state['cached_wedding_config'] = config_content
print("✅ Loaded wedding configuration from Google Drive")
else:
# Store in file for local development
config_path = self.get_config_file_path()
config_dir = os.path.dirname(config_path)
if config_dir:
os.makedirs(config_dir, exist_ok=True)
with open(config_path, 'w') as f:
json.dump(config_content, f, indent=2)
# Update cache in session state
st.session_state['cached_wedding_config'] = config_content
print(f"Stored wedding_config.json in file: {config_path}")
# Sync data files from user folder
for file_name in data_files:
content = self.drive_manager.download_file(f'{user_folder}/{file_name}')
if content:
if self.use_memory_storage:
# Store in memory for Hugging Face Spaces
self.memory_data[file_name] = content
# Update cache in session state
cache_key = f"cached_{file_name}"
st.session_state[cache_key] = content
# Only print once per sync operation
if file_name == data_files[0]: # First file
print(f"✅ Loaded {len(data_files)} data files from Google Drive")
else:
# Store in file for local development
file_path = self.get_data_file_path(file_name)
file_dir = os.path.dirname(file_path)
if file_dir:
os.makedirs(file_dir, exist_ok=True)
with open(file_path, 'w') as f:
json.dump(content, f, indent=2)
# Update cache in session state
cache_key = f"cached_{file_name}"
st.session_state[cache_key] = content
print(f"Stored {file_name} in file: {file_path}")
except Exception as e:
print(f"Error syncing from Google Drive: {e}")
# Don't fail the entire initialization if sync fails
# This is expected behavior - user can manually sync later
def _sync_to_google_drive(self):
"""Sync local data files to Google Drive"""
if not self.google_drive_enabled:
return
# Get user-specific folder
user_folder = self.get_user_folder()
# Sync config file to user folder
config_path = self.get_config_file_path()
if os.path.exists(config_path):
with open(config_path, 'r') as f:
config_content = json.load(f)
self.drive_manager.upload_file(f'{user_folder}/wedding_config.json', config_content)
# Sync data files to user folder
data_files = [
'guest_list_data.json',
'rsvp_data.json',
'tasks.json',
'vendors.json',
'wedding_party.json'
]
for file_name in data_files:
file_path = self.get_data_file_path(file_name)
if os.path.exists(file_path):
with open(file_path, 'r') as f:
content = json.load(f)
self.drive_manager.upload_file(f'{user_folder}/{file_name}', content)
def load_app_config(self):
"""Load app configuration from file"""
# For Hugging Face Spaces, check /tmp directory first
if self.is_huggingface:
tmp_config_path = f"/tmp/{self.app_config_file}"
if os.path.exists(tmp_config_path):
try:
with open(tmp_config_path, 'r') as f:
return json.load(f)
except (json.JSONDecodeError, FileNotFoundError):
pass
# Check original location
if os.path.exists(self.app_config_file):
try:
with open(self.app_config_file, 'r') as f:
return json.load(f)
except (json.JSONDecodeError, FileNotFoundError):
return self.get_default_app_config()
return self.get_default_app_config()
def get_default_app_config(self):
"""Get default app configuration"""
return {
"demo_mode": False,
"demo_data_path": "demo_data",
"app_settings": {
"theme": "green_adirondack",
"auto_save": True,
"backup_enabled": True
}
}
def is_demo_mode(self):
"""Check if demo mode is enabled"""
return self.app_config.get("demo_mode", False)
def get_data_directory(self):
"""Get the appropriate data directory based on demo mode"""
if self.is_demo_mode():
return self.demo_data_dir
return self.data_dir
def get_config_file_path(self):
"""Get the appropriate config file path based on demo mode"""
if self.is_demo_mode():
return os.path.join(self.demo_data_dir, "wedding_config.json")
# For Hugging Face Spaces, use app directory to avoid permission issues
if self.is_huggingface:
return "wedding_config.json"
# For local development, use the original path
if os.path.isabs(self.config_file):
return self.config_file
else:
return os.path.join(os.getcwd(), self.config_file)
def config_exists(self):
"""Check if configuration file exists"""
if self.use_memory_storage:
return 'wedding_config.json' in self.memory_data
else:
config_path = self.get_config_file_path()
return os.path.exists(config_path)
def load_config(self):
"""Load configuration from cache, file or memory"""
# Check if we have cached config in session state
if 'cached_wedding_config' in st.session_state:
return st.session_state['cached_wedding_config']
if self.config_exists():
try:
if self.use_memory_storage:
config = self.memory_data['wedding_config.json']
else:
config_path = self.get_config_file_path()
with open(config_path, 'r') as f:
config = json.load(f)
# Cache the config in session state
st.session_state['cached_wedding_config'] = config
return config
except (json.JSONDecodeError, FileNotFoundError, KeyError):
default_config = self.get_default_config()
st.session_state['cached_wedding_config'] = default_config
return default_config
default_config = self.get_default_config()
st.session_state['cached_wedding_config'] = default_config
return default_config
def save_config(self, config):
"""Save configuration to file or memory"""
try:
if self.use_memory_storage:
# Store in memory for Hugging Face Spaces
self.memory_data['wedding_config.json'] = config
else:
# Store in file for local development
config_path = self.get_config_file_path()
with open(config_path, 'w') as f:
json.dump(config, f, indent=2)
# Update cache in session state
st.session_state['cached_wedding_config'] = config
# Mark config as modified for manual sync
st.session_state['config_modified'] = True
return True
except Exception as e:
print(f"Error saving config: {e}")
return False
def reset_config(self):
"""Reset configuration by deleting the config file"""
try:
config_path = self.get_config_file_path()
if os.path.exists(config_path):
os.remove(config_path)
return True
except Exception as e:
print(f"Error resetting config: {e}")
return False
def get_default_config(self):
"""Get default configuration"""
return {
'wedding_info': {
'partner1_name': '',
'partner2_name': '',
'wedding_start_date': '',
'wedding_end_date': '',
'venue_city': ''
},
'custom_settings': {
'custom_tags': [
"Wedding Party", "Urgent", "Rehearsal", "Timeline",
"Maid of Honor", "Best Man", "Bridesmaid", "Groomsman",
"Flower Girl", "Ring Bearer", "Usher", "Reader",
"Venue", "Catering", "Photography", "Videography", "Music/DJ",
"Flowers", "Decor", "Attire", "Hair & Makeup", "Transportation",
"Invitations", "Cake", "Officiant", "Other",
"Decorations", "Centerpieces", "Favors", "Signage", "Linens",
"Tableware", "Lighting", "Accessories", "Stationery", "Gifts",
"Vendor", "Item", "Deposit Required", "Final Payment", "Installment",
"Food & Beverage", "Music", "Entertainment", "Lodging"
],
'task_assignees': []
},
'wedding_events': [
{"name": "Welcome Dinner", "date_offset": -1, "description": "Welcome dinner for out-of-town guests", "requires_meal_choice": True, "meal_options": ["Duck", "Surf & Turf", "Risotto (vegetarian)", "Stuffed Squash (vegetarian)"], "location": ""},
{"name": "Church Ceremony", "date_offset": 0, "description": "Main wedding ceremony", "requires_meal_choice": False, "meal_options": [], "location": ""},
{"name": "Reception", "date_offset": 0, "description": "Wedding reception with dinner", "requires_meal_choice": True, "meal_options": ["Duck", "Surf & Turf", "Risotto (vegetarian)", "Stuffed Squash (vegetarian)"], "location": ""},
{"name": "Mehndi Afterparty", "date_offset": 1, "description": "Mehndi celebration and afterparty", "requires_meal_choice": False, "meal_options": [], "location": ""},
{"name": "Indian Ceremony", "date_offset": 1, "description": "Traditional Indian wedding ceremony", "requires_meal_choice": False, "meal_options": [], "location": ""},
{"name": "Indian Reception", "date_offset": 1, "description": "Indian reception celebration", "requires_meal_choice": True, "meal_options": ["Duck", "Surf & Turf", "Risotto (vegetarian)", "Stuffed Squash (vegetarian)"], "location": ""}
]
}
def get_data_file_path(self, filename):
"""Get full path for data files"""
data_dir = self.get_data_directory()
return os.path.join(data_dir, filename)
def load_json_data(self, filename):
"""Load JSON data from cache, data directory, or memory"""
# Check if we have cached data in session state
cache_key = f"cached_{filename}"
if cache_key in st.session_state:
# Data loaded from cache
return st.session_state[cache_key]
# Load data from source
if self.use_memory_storage:
# Load from memory for Hugging Face Spaces
data = self.memory_data.get(filename, [])
else:
# Load from file for local development
filepath = self.get_data_file_path(filename)
if os.path.exists(filepath):
try:
with open(filepath, 'r') as f:
data = json.load(f)
except (json.JSONDecodeError, FileNotFoundError):
data = []
else:
data = []
# Cache the data in session state
st.session_state[cache_key] = data
return data
def save_json_data(self, filename, data):
"""Save JSON data to data directory or memory"""
try:
if self.use_memory_storage:
# Store in memory for Hugging Face Spaces
self.memory_data[filename] = data
else:
# Store in file for local development
filepath = self.get_data_file_path(filename)
with open(filepath, 'w') as f:
json.dump(data, f, indent=2)
# Update cache in session state
cache_key = f"cached_{filename}"
st.session_state[cache_key] = data
# Mark data as modified for manual sync
st.session_state[f'{filename}_modified'] = True
return True
except Exception as e:
print(f"Error saving data to {filename}: {e}")
return False
def toggle_demo_mode(self):
"""Toggle demo mode on/off"""
self.app_config["demo_mode"] = not self.app_config.get("demo_mode", False)
try:
# For Hugging Face Spaces, use /tmp directory to avoid permission issues
if self.is_huggingface:
config_path = f"/tmp/{self.app_config_file}"
else:
config_path = self.app_config_file
with open(config_path, 'w') as f:
json.dump(self.app_config, f, indent=2)
return True
except Exception as e:
print(f"Error saving app config: {e}")
return False
def set_demo_mode(self, enabled):
"""Set demo mode to specific value"""
self.app_config["demo_mode"] = enabled
try:
# For Hugging Face Spaces, use /tmp directory to avoid permission issues
if self.is_huggingface:
config_path = f"/tmp/{self.app_config_file}"
else:
config_path = self.app_config_file
with open(config_path, 'w') as f:
json.dump(self.app_config, f, indent=2)
return True
except Exception as e:
print(f"Error saving app config: {e}")
return False
def is_google_drive_enabled(self):
"""Check if Google Drive integration is enabled and working"""
return self.google_drive_enabled and self.drive_manager.is_online()
def manual_sync_to_drive(self):
"""Manually sync all data to Google Drive"""
if not self.google_drive_enabled:
return False
try:
self._sync_to_google_drive()
return True
except Exception as e:
print(f"Error syncing to Google Drive: {e}")
return False
def clear_cache(self):
"""Clear all cached data from session state"""
data_files = [
'guest_list_data.json',
'rsvp_data.json',
'tasks.json',
'vendors.json',
'wedding_party.json',
'guests.json'
]
for filename in data_files:
cache_key = f"cached_{filename}"
if cache_key in st.session_state:
del st.session_state[cache_key]
# Also clear config cache
if 'cached_wedding_config' in st.session_state:
del st.session_state['cached_wedding_config']
def get_cache_status(self):
"""Get information about what data is currently cached"""
data_files = [
'guest_list_data.json',
'rsvp_data.json',
'tasks.json',
'vendors.json',
'wedding_party.json',
'guests.json'
]
cached_files = []
for filename in data_files:
cache_key = f"cached_{filename}"
if cache_key in st.session_state:
cached_files.append(filename)
config_cached = 'cached_wedding_config' in st.session_state
return {
'cached_data_files': cached_files,
'config_cached': config_cached,
'total_cached': len(cached_files) + (1 if config_cached else 0)
}
def invalidate_cache_for_file(self, filename):
"""Invalidate cache for a specific file"""
cache_key = f"cached_{filename}"
if cache_key in st.session_state:
del st.session_state[cache_key]
def invalidate_config_cache(self):
"""Invalidate the config cache"""
if 'cached_wedding_config' in st.session_state:
del st.session_state['cached_wedding_config']
def manual_sync_from_drive(self):
"""Manually sync all data from Google Drive"""
if not self.google_drive_enabled:
return False
try:
self._sync_from_google_drive()
self.data_loaded_from_drive = True
# Clear modified flags since data is now in sync with Google Drive
self.clear_modified_flags()
return True
except Exception as e:
print(f"Error syncing from Google Drive: {e}")
return False
def load_existing_data_from_drive(self):
"""Load existing wedding data from Google Drive and create local config"""
if not self.google_drive_enabled:
return False
try:
# Disable demo mode when loading real data
self.set_demo_mode(False)
# Get user-specific folder
user_folder = self.get_current_user_folder()
# Check if wedding_config.json exists in Google Drive user folder
config_content = self.drive_manager.download_file(f'{user_folder}/wedding_config.json')
if config_content:
# Save config locally (in memory or file)
if self.use_memory_storage:
self.memory_data['wedding_config.json'] = config_content
# Update cache in session state
st.session_state['cached_wedding_config'] = config_content
print("Stored wedding_config.json in memory")
else:
config_path = self.get_config_file_path()
config_dir = os.path.dirname(config_path)
if config_dir:
os.makedirs(config_dir, exist_ok=True)
with open(config_path, 'w') as f:
json.dump(config_content, f, indent=2)
# Update cache in session state
st.session_state['cached_wedding_config'] = config_content
print(f"Stored wedding_config.json in file: {config_path}")
# Load data files from user folder
data_files = [
'guest_list_data.json',
'rsvp_data.json',
'tasks.json',
'vendors.json',
'wedding_party.json'
]
for file_name in data_files:
content = self.drive_manager.download_file(f'{user_folder}/{file_name}')
if content:
if self.use_memory_storage:
self.memory_data[file_name] = content
# Update cache in session state
cache_key = f"cached_{file_name}"
st.session_state[cache_key] = content
print(f"Stored {file_name} in memory")
else:
file_path = self.get_data_file_path(file_name)
file_dir = os.path.dirname(file_path)
if file_dir:
os.makedirs(file_dir, exist_ok=True)
with open(file_path, 'w') as f:
json.dump(content, f, indent=2)
# Update cache in session state
cache_key = f"cached_{file_name}"
st.session_state[cache_key] = content
print(f"Stored {file_name} in file: {file_path}")
self.data_loaded_from_drive = True
return True
return False
except Exception as e:
print(f"Error loading existing data from Google Drive: {e}")
# This is a common issue on Hugging Face Spaces due to network/SSL issues
# The error is expected and the user can retry manually
return False
def load_demo_data_from_drive(self):
"""Load demo wedding data from Google Drive and enable demo mode"""
if not self.google_drive_enabled:
return False
try:
# Enable demo mode first
self.set_demo_mode(True)
# Load demo data from demo folder in Google Drive
demo_folder = self.get_current_user_folder() # Will return "demo_data" since demo mode is enabled
# Check if wedding_config.json exists in demo folder
config_content = self.drive_manager.download_file(f'{demo_folder}/wedding_config.json')
if config_content:
# Save config locally (in memory or file)
if self.use_memory_storage:
self.memory_data['wedding_config.json'] = config_content
# Update cache in session state
st.session_state['cached_wedding_config'] = config_content
print("Stored demo wedding_config.json in memory")
else:
config_path = self.get_config_file_path()
config_dir = os.path.dirname(config_path)
if config_dir:
os.makedirs(config_dir, exist_ok=True)
with open(config_path, 'w') as f:
json.dump(config_content, f, indent=2)
# Update cache in session state
st.session_state['cached_wedding_config'] = config_content
print(f"Stored demo wedding_config.json in file: {config_path}")
# Load demo data files from demo folder
data_files = [
'guest_list_data.json',
'rsvp_data.json',
'tasks.json',
'vendors.json',
'wedding_party.json'
]
for file_name in data_files:
content = self.drive_manager.download_file(f'{demo_folder}/{file_name}')
if content:
if self.use_memory_storage:
self.memory_data[file_name] = content
# Update cache in session state
cache_key = f"cached_{file_name}"
st.session_state[cache_key] = content
print(f"Stored demo {file_name} in memory")
else:
file_path = self.get_data_file_path(file_name)
file_dir = os.path.dirname(file_path)
if file_dir:
os.makedirs(file_dir, exist_ok=True)
with open(file_path, 'w') as f:
json.dump(content, f, indent=2)
# Update cache in session state
cache_key = f"cached_{file_name}"
st.session_state[cache_key] = content
print(f"Stored demo {file_name} in file: {file_path}")
self.data_loaded_from_drive = True
return True
return False
except Exception as e:
print(f"Error loading demo data from Google Drive: {e}")
# This is a common issue on Hugging Face Spaces due to network/SSL issues
# The error is expected and the user can retry manually
return False
def get_google_drive_status(self):
"""Get status information about Google Drive integration"""
if not self.google_drive_enabled:
return {
'enabled': False,
'status': 'Disabled',
'message': 'Google Drive integration not configured'
}
if not self.drive_manager.is_online():
return {
'enabled': True,
'status': 'Offline',
'message': 'Google Drive service unavailable'
}
try:
files = self.drive_manager.list_files()
return {
'enabled': True,
'status': 'Online',
'message': f'Connected to Google Drive ({len(files)} files found)',
'files': [f['name'] for f in files]
}
except Exception as e:
return {
'enabled': True,
'status': 'Error',
'message': f'Error connecting to Google Drive: {str(e)}'
}
def get_modified_files(self):
"""Get list of files that have been modified since last sync"""
modified_files = []
# Check config
if st.session_state.get('config_modified', False):
modified_files.append('wedding_config.json')
# Check data files
data_files = [
'guest_list_data.json',
'rsvp_data.json',
'tasks.json',
'vendors.json',
'wedding_party.json'
]
for filename in data_files:
if st.session_state.get(f'{filename}_modified', False):
modified_files.append(filename)
return modified_files
def clear_modified_flags(self):
"""Clear all modified flags after successful sync"""
st.session_state['config_modified'] = False
data_files = [
'guest_list_data.json',
'rsvp_data.json',
'tasks.json',
'vendors.json',
'wedding_party.json'
]
for filename in data_files:
st.session_state[f'{filename}_modified'] = False
def manual_sync_to_drive(self):
"""Manually sync all current data to Google Drive"""
if not self.google_drive_enabled:
return False
try:
# Get user-specific folder
user_folder = self.get_current_user_folder()
modified_files = self.get_modified_files()
if not modified_files:
print("No modified files detected, but syncing all current data to ensure consistency")
else:
print(f"Syncing {len(modified_files)} modified files to Google Drive: {modified_files}")
# Always sync config (current state) to user folder
config = self.load_config()
if not self.drive_manager.upload_file(f'{user_folder}/wedding_config.json', config):
return False
# Always sync all data files (current state) to user folder
data_files = [
'guest_list_data.json',
'rsvp_data.json',
'tasks.json',
'vendors.json',
'wedding_party.json'
]
for filename in data_files:
data = self.load_json_data(filename)
if not self.drive_manager.upload_file(f'{user_folder}/{filename}', data):
return False
# Clear modified flags after successful sync
self.clear_modified_flags()
return True
except Exception as e:
print(f"Error syncing to Google Drive: {e}")
return False
def reset_app_state(self):
"""Reset the app state by clearing all cached data and memory storage"""
try:
# Clear memory data
if hasattr(self, 'memory_data'):
self.memory_data.clear()
# Clear session state cache
if 'cached_wedding_config' in st.session_state:
del st.session_state['cached_wedding_config']
# Clear other cached data files
for key in list(st.session_state.keys()):
if key.startswith('cached_'):
del st.session_state[key]
# Reset data loaded flag
self.data_loaded_from_drive = False
print("App state reset successfully")
return True
except Exception as e:
print(f"Error resetting app state: {e}")
return False