File size: 2,468 Bytes
d9e0aef |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 |
# utils/storage_init.py
import os
import streamlit as st
from pathlib import Path
def initialize_storage_structure():
"""Initialize the storage structure and verify its existence."""
# Check if /data exists (Hugging Face persistent storage)
if os.path.exists('/data'):
base_path = Path('/data')
else:
# Fallback to local data directory in project
base_path = Path(os.getcwd()) / 'data'
try:
# Create main directories
directories = [
base_path / 'database',
base_path / 'files',
base_path / 'vectorstore',
base_path / 'metadata'
]
# Create each directory
for directory in directories:
directory.mkdir(parents=True, exist_ok=True)
# Print debug information
st.write("Storage directories initialized at:", str(base_path))
st.write("Available directories:")
for directory in directories:
st.write(f"- {directory}: {'Exists' if directory.exists() else 'Failed to create'}")
# Try to create a test file to verify write permissions
test_file = base_path / 'database' / '.test_write'
try:
test_file.touch()
test_file.unlink() # Remove the test file
st.success("Storage system initialized successfully!")
except Exception as e:
st.error(f"Write permission test failed: {e}")
return base_path
except Exception as e:
st.error(f"Error initializing storage structure: {e}")
return None
def verify_storage_access(base_path: Path):
"""Verify storage access and permissions."""
try:
# Check directory existence and permissions
checks = {
'exists': base_path.exists(),
'is_dir': base_path.is_dir(),
'readable': os.access(base_path, os.R_OK),
'writable': os.access(base_path, os.W_OK),
'executable': os.access(base_path, os.X_OK)
}
# Print debug information
st.write("Storage Access Verification:")
for check, result in checks.items():
st.write(f"- {check}: {'✅' if result else '❌'}")
# Return overall status
return all(checks.values())
except Exception as e:
st.error(f"Error verifying storage access: {e}")
return False |