""" ProjectManager - Project Registry Service Manages the server-side registry of projects, including: - Project registration with path validation - Root path storage and retrieval - Project lifecycle management """ import os from pathlib import Path from typing import Optional, List, Dict, Any import logging logger = logging.getLogger(__name__) class ProjectManager: """ Manages project registry with root path validation. Works with the existing Projects database class to add root_path support. Validates paths exist and are accessible before registration. """ def __init__(self, projects_db, config_service=None): """ Initialize project manager. Args: projects_db: Projects database class (from storage.database) config_service: Optional ConfigService for config initialization """ self.db = projects_db self.config_service = config_service logger.info("ProjectManager initialized") def register_project( self, name: str, root_path: str, description: str = "", figma_file_key: str = "" ) -> Dict[str, Any]: """ Register a new project with validated root path. Args: name: Human-readable project name root_path: Absolute path to project directory description: Optional project description figma_file_key: Optional Figma file key Returns: Created project dict Raises: ValueError: If path doesn't exist or isn't a directory PermissionError: If no write access to path """ # Resolve and validate path root_path = os.path.abspath(root_path) if not os.path.isdir(root_path): raise ValueError(f"Path does not exist or is not a directory: {root_path}") if not os.access(root_path, os.W_OK): raise PermissionError(f"No write access to path: {root_path}") # Check if path already registered existing = self.get_by_path(root_path) if existing: raise ValueError(f"Path already registered as project: {existing['name']}") # Generate project ID import uuid project_id = str(uuid.uuid4())[:8] # Create project in database project = self.db.create( id=project_id, name=name, description=description, figma_file_key=figma_file_key ) # Update with root_path (need to add this column) self._update_root_path(project_id, root_path) project['root_path'] = root_path # Initialize .dss folder and config if config_service available if self.config_service: try: self.config_service.init_config(root_path) logger.info(f"Initialized .dss config for project {name}") except Exception as e: logger.warning(f"Failed to init config for {name}: {e}") logger.info(f"Registered project: {name} at {root_path}") return project def get_project(self, project_id: str) -> Optional[Dict[str, Any]]: """ Get project by ID with path validation. Args: project_id: Project UUID Returns: Project dict or None if not found Raises: ValueError: If project path no longer exists """ project = self.db.get(project_id) if not project: return None root_path = project.get('root_path') if root_path and not os.path.isdir(root_path): logger.warning(f"Project path no longer exists: {root_path}") # Don't raise, just mark it project['path_valid'] = False else: project['path_valid'] = True return project def list_projects(self, status: str = None, valid_only: bool = False) -> List[Dict[str, Any]]: """ List all projects with optional filtering. Args: status: Filter by status (active, archived, etc.) valid_only: Only return projects with valid paths Returns: List of project dicts """ projects = self.db.list(status=status) # Add path validation status for project in projects: root_path = project.get('root_path') project['path_valid'] = bool(root_path and os.path.isdir(root_path)) if valid_only: projects = [p for p in projects if p.get('path_valid', False)] return projects def get_by_path(self, root_path: str) -> Optional[Dict[str, Any]]: """ Find project by root path. Args: root_path: Absolute path to search for Returns: Project dict or None if not found """ root_path = os.path.abspath(root_path) projects = self.list_projects() for project in projects: if project.get('root_path') == root_path: return project return None def update_project( self, project_id: str, name: str = None, description: str = None, root_path: str = None, figma_file_key: str = None, status: str = None ) -> Optional[Dict[str, Any]]: """ Update project fields. Args: project_id: Project UUID name: Optional new name description: Optional new description root_path: Optional new root path (validated) figma_file_key: Optional new Figma key status: Optional new status Returns: Updated project dict or None if not found """ project = self.db.get(project_id) if not project: return None # Validate new root_path if provided if root_path: root_path = os.path.abspath(root_path) if not os.path.isdir(root_path): raise ValueError(f"Path does not exist: {root_path}") if not os.access(root_path, os.W_OK): raise PermissionError(f"No write access: {root_path}") self._update_root_path(project_id, root_path) # Update other fields via existing update method updates = {} if name is not None: updates['name'] = name if description is not None: updates['description'] = description if figma_file_key is not None: updates['figma_file_key'] = figma_file_key if status is not None: updates['status'] = status if updates: self.db.update(project_id, **updates) return self.get_project(project_id) def delete_project(self, project_id: str, delete_config: bool = False) -> bool: """ Delete a project from registry. Args: project_id: Project UUID delete_config: If True, also delete .dss folder Returns: True if deleted, False if not found """ project = self.db.get(project_id) if not project: return False if delete_config and project.get('root_path'): import shutil dss_path = Path(project['root_path']) / '.dss' if dss_path.exists(): shutil.rmtree(dss_path) logger.info(f"Deleted .dss folder at {dss_path}") self.db.delete(project_id) logger.info(f"Deleted project: {project_id}") return True def _update_root_path(self, project_id: str, root_path: str) -> None: """ Update root_path in database. Uses raw SQL since the column may not be in the existing model. """ from storage.database import get_connection with get_connection() as conn: # Ensure column exists try: conn.execute(""" ALTER TABLE projects ADD COLUMN root_path TEXT DEFAULT '' """) logger.info("Added root_path column to projects table") except Exception: # Column already exists pass # Update the value conn.execute( "UPDATE projects SET root_path = ? WHERE id = ?", (root_path, project_id) ) @staticmethod def ensure_schema(): """ Ensure database schema has root_path column. Call this on startup to migrate existing databases. """ from storage.database import get_connection with get_connection() as conn: cursor = conn.cursor() # Check if column exists cursor.execute("PRAGMA table_info(projects)") columns = [col[1] for col in cursor.fetchall()] if 'root_path' not in columns: cursor.execute(""" ALTER TABLE projects ADD COLUMN root_path TEXT DEFAULT '' """) logger.info("Migration: Added root_path column to projects table") else: logger.debug("Schema check: root_path column exists")