Spaces:
Runtime error
Runtime error
| """ | |
| Spaces (Workspaces) manager for organizing chats and files by subject. | |
| Each space has its own vector DB and chat history. | |
| """ | |
| import json | |
| from pathlib import Path | |
| from typing import List, Dict, Optional | |
| from datetime import datetime | |
| import config | |
| class SpacesManager: | |
| """Manages workspaces (Spaces) for organizing study materials by subject.""" | |
| def __init__(self): | |
| self.spaces_file = config.DATA_DIR / "spaces.json" | |
| self.spaces_data = self._load_spaces() | |
| def _load_spaces(self) -> Dict: | |
| """Load spaces from file.""" | |
| if self.spaces_file.exists(): | |
| try: | |
| with open(self.spaces_file, 'r', encoding='utf-8') as f: | |
| return json.load(f) | |
| except Exception: | |
| return self._create_default_spaces() | |
| return self._create_default_spaces() | |
| def _create_default_spaces(self) -> Dict: | |
| """Create default spaces structure.""" | |
| return { | |
| "spaces": [ | |
| { | |
| "id": "general", | |
| "name": "General", | |
| "description": "General study materials", | |
| "created_at": datetime.now().isoformat(), | |
| "file_count": 0, | |
| "chat_count": 0 | |
| } | |
| ] | |
| } | |
| def save_spaces(self): | |
| """Save spaces to file.""" | |
| try: | |
| with open(self.spaces_file, 'w', encoding='utf-8') as f: | |
| json.dump(self.spaces_data, f, indent=2) | |
| except Exception as e: | |
| print(f"Error saving spaces: {e}") | |
| def get_all_spaces(self) -> List[Dict]: | |
| """Get all spaces.""" | |
| return self.spaces_data.get("spaces", []) | |
| def get_space(self, space_id: str) -> Optional[Dict]: | |
| """Get specific space by ID.""" | |
| for space in self.spaces_data.get("spaces", []): | |
| if space["id"] == space_id: | |
| return space | |
| return None | |
| def create_space(self, name: str, description: str = "") -> Dict: | |
| """Create a new space.""" | |
| space_id = name.lower().replace(" ", "_") | |
| # Check if space already exists | |
| if self.get_space(space_id): | |
| raise ValueError(f"Space '{name}' already exists") | |
| new_space = { | |
| "id": space_id, | |
| "name": name, | |
| "description": description, | |
| "created_at": datetime.now().isoformat(), | |
| "file_count": 0, | |
| "chat_count": 0 | |
| } | |
| self.spaces_data["spaces"].append(new_space) | |
| self.save_spaces() | |
| # Create dedicated directories for this space | |
| space_dir = config.DATA_DIR / "spaces" / space_id | |
| space_dir.mkdir(parents=True, exist_ok=True) | |
| (space_dir / "chats").mkdir(exist_ok=True) | |
| (space_dir / "vector_db").mkdir(exist_ok=True) | |
| (space_dir / "uploads").mkdir(exist_ok=True) | |
| return new_space | |
| def delete_space(self, space_id: str): | |
| """Delete a space (except General).""" | |
| if space_id == "general": | |
| raise ValueError("Cannot delete General space") | |
| self.spaces_data["spaces"] = [ | |
| s for s in self.spaces_data["spaces"] | |
| if s["id"] != space_id | |
| ] | |
| self.save_spaces() | |
| def update_space_counts(self, space_id: str, file_count: int = None, chat_count: int = None): | |
| """Update file/chat counts for a space.""" | |
| space = self.get_space(space_id) | |
| if space: | |
| if file_count is not None: | |
| space["file_count"] = file_count | |
| if chat_count is not None: | |
| space["chat_count"] = chat_count | |
| self.save_spaces() | |
| def get_space_chats_dir(self, space_id: str) -> Path: | |
| """Get chats directory for a space.""" | |
| return config.DATA_DIR / "spaces" / space_id / "chats" | |
| def get_space_vector_db_dir(self, space_id: str) -> Path: | |
| """Get vector DB directory for a space.""" | |
| return config.DATA_DIR / "spaces" / space_id / "vector_db" | |
| def get_space_uploads_dir(self, space_id: str) -> Path: | |
| """Get uploads directory for a space.""" | |
| return config.DATA_DIR / "spaces" / space_id / "uploads" | |