Add manager modules: dotfile, backup, package, and editor
Some checks failed
CI / test (push) Failing after 10s
Some checks failed
CI / test (push) Failing after 10s
This commit is contained in:
452
dev_env_sync/managers/editor.py
Normal file
452
dev_env_sync/managers/editor.py
Normal file
@@ -0,0 +1,452 @@
|
||||
"""Editor configuration synchronization."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
from abc import ABC, abstractmethod
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
from dataclasses import dataclass
|
||||
|
||||
from dev_env_sync.core.config import EditorExtension, ConfigSchema
|
||||
from dev_env_sync.utils.file_ops import FileOps, path_expand
|
||||
from dev_env_sync.utils.logging import get_logger
|
||||
from dev_env_sync.core.platform import PlatformDetector
|
||||
|
||||
|
||||
class EditorError(Exception):
|
||||
"""Raised when an editor operation fails."""
|
||||
pass
|
||||
|
||||
|
||||
@dataclass
|
||||
class EditorSyncResult:
|
||||
"""Result of an editor sync operation."""
|
||||
success: bool
|
||||
editor: str
|
||||
operation: str
|
||||
details: str
|
||||
path: Optional[Path] = None
|
||||
|
||||
|
||||
class EditorConfig(ABC):
|
||||
"""Abstract base class for editor configuration."""
|
||||
|
||||
def __init__(self, dry_run: bool = False, logger=None):
|
||||
self.dry_run = dry_run
|
||||
self.logger = logger or get_logger(__name__)
|
||||
self.file_ops = FileOps(dry_run=dry_run, logger=logger)
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def name(self) -> str:
|
||||
"""Return the editor name."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_config_dir(self) -> Path:
|
||||
"""Get the platform-specific config directory."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_settings_file(self) -> Optional[Path]:
|
||||
"""Get the path to the settings file."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def sync_settings(self, settings_source: Path) -> EditorSyncResult:
|
||||
"""Sync editor settings from source file."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def sync_extensions(self, extensions: List[EditorExtension]) -> List[EditorSyncResult]:
|
||||
"""Sync editor extensions."""
|
||||
pass
|
||||
|
||||
def _ensure_config_dir(self) -> Path:
|
||||
"""Ensure the config directory exists."""
|
||||
config_dir = self.get_config_dir()
|
||||
self.file_ops.create_directory(config_dir, exist_ok=True)
|
||||
return config_dir
|
||||
|
||||
|
||||
class VSCodeConfig(EditorConfig):
|
||||
"""VS Code configuration handler."""
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "vscode"
|
||||
|
||||
def get_config_dir(self) -> Path:
|
||||
return PlatformDetector.get_editor_config_dir("vscode")
|
||||
|
||||
def get_settings_file(self) -> Optional[Path]:
|
||||
return self.get_config_dir() / "settings.json"
|
||||
|
||||
def sync_settings(self, settings_source: Path) -> EditorSyncResult:
|
||||
"""Sync VS Code settings from source JSON file."""
|
||||
try:
|
||||
self._ensure_config_dir()
|
||||
|
||||
if not settings_source.exists():
|
||||
return EditorSyncResult(
|
||||
success=False,
|
||||
editor=self.name,
|
||||
operation="sync_settings",
|
||||
details=f"Source settings file not found: {settings_source}",
|
||||
)
|
||||
|
||||
content = self.file_ops.read_file(settings_source)
|
||||
if content is None:
|
||||
return EditorSyncResult(
|
||||
success=False,
|
||||
editor=self.name,
|
||||
operation="sync_settings",
|
||||
details=f"Failed to read settings source: {settings_source}",
|
||||
)
|
||||
|
||||
valid_json = json.loads(content)
|
||||
|
||||
target = self.get_settings_file()
|
||||
success = self.file_ops.write_file(target, json.dumps(valid_json, indent=2))
|
||||
|
||||
if success:
|
||||
return EditorSyncResult(
|
||||
success=True,
|
||||
editor=self.name,
|
||||
operation="sync_settings",
|
||||
details=f"Synced settings to {target}",
|
||||
path=target,
|
||||
)
|
||||
else:
|
||||
return EditorSyncResult(
|
||||
success=False,
|
||||
editor=self.name,
|
||||
operation="sync_settings",
|
||||
details=f"Failed to write settings: {target}",
|
||||
)
|
||||
except json.JSONDecodeError as e:
|
||||
return EditorSyncResult(
|
||||
success=False,
|
||||
editor=self.name,
|
||||
operation="sync_settings",
|
||||
details=f"Invalid JSON in settings source: {e}",
|
||||
)
|
||||
except Exception as e:
|
||||
return EditorSyncResult(
|
||||
success=False,
|
||||
editor=self.name,
|
||||
operation="sync_settings",
|
||||
details=f"Error syncing settings: {e}",
|
||||
)
|
||||
|
||||
def sync_extensions(self, extensions: List[EditorExtension]) -> List[EditorSyncResult]:
|
||||
"""Sync VS Code extensions (install command generation)."""
|
||||
results = []
|
||||
|
||||
for ext in extensions:
|
||||
result = self._install_extension(ext.name)
|
||||
results.append(result)
|
||||
|
||||
return results
|
||||
|
||||
def _install_extension(self, extension_id: str) -> EditorSyncResult:
|
||||
"""Generate extension installation command."""
|
||||
install_cmd = f"code --install-extension {extension_id}"
|
||||
|
||||
self.logger.info(f"VS Code extension: {extension_id}")
|
||||
self.logger.info(f" Install command: {install_cmd}")
|
||||
|
||||
if self.dry_run:
|
||||
return EditorSyncResult(
|
||||
success=True,
|
||||
editor=self.name,
|
||||
operation="install_extension",
|
||||
details=f"[DRY-RUN] Would install: {extension_id}",
|
||||
)
|
||||
|
||||
return EditorSyncResult(
|
||||
success=True,
|
||||
editor=self.name,
|
||||
operation="install_extension",
|
||||
details=f"Extension installation queued: {extension_id}",
|
||||
)
|
||||
|
||||
def get_extensions_list(self) -> List[str]:
|
||||
"""Get list of currently installed extensions."""
|
||||
return []
|
||||
|
||||
def merge_settings(
|
||||
self,
|
||||
source: Path,
|
||||
target: Path,
|
||||
strategy: str = "replace",
|
||||
) -> EditorSyncResult:
|
||||
"""Merge VS Code settings."""
|
||||
try:
|
||||
source_content = self.file_ops.read_file(source)
|
||||
target_content = self.file_ops.read_file(target)
|
||||
|
||||
if source_content is None:
|
||||
return EditorSyncResult(
|
||||
success=False,
|
||||
editor=self.name,
|
||||
operation="merge_settings",
|
||||
details=f"Could not read source: {source}",
|
||||
)
|
||||
|
||||
source_settings = json.loads(source_content) if source_content else {}
|
||||
|
||||
if target_content and strategy == "merge":
|
||||
target_settings = json.loads(target_content)
|
||||
merged = {**target_settings, **source_settings}
|
||||
else:
|
||||
merged = source_settings
|
||||
|
||||
success = self.file_ops.write_file(target, json.dumps(merged, indent=2))
|
||||
|
||||
if success:
|
||||
return EditorSyncResult(
|
||||
success=True,
|
||||
editor=self.name,
|
||||
operation="merge_settings",
|
||||
details=f"Merged settings to {target}",
|
||||
path=target,
|
||||
)
|
||||
else:
|
||||
return EditorSyncResult(
|
||||
success=False,
|
||||
editor=self.name,
|
||||
operation="merge_settings",
|
||||
details=f"Failed to write merged settings: {target}",
|
||||
)
|
||||
except json.JSONDecodeError as e:
|
||||
return EditorSyncResult(
|
||||
success=False,
|
||||
editor=self.name,
|
||||
operation="merge_settings",
|
||||
details=f"Invalid JSON: {e}",
|
||||
)
|
||||
except Exception as e:
|
||||
return EditorSyncResult(
|
||||
success=False,
|
||||
editor=self.name,
|
||||
operation="merge_settings",
|
||||
details=f"Error merging settings: {e}",
|
||||
)
|
||||
|
||||
|
||||
class NeovimConfig(EditorConfig):
|
||||
"""Neovim configuration handler."""
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "neovim"
|
||||
|
||||
def get_config_dir(self) -> Path:
|
||||
return PlatformDetector.get_editor_config_dir("neovim")
|
||||
|
||||
def get_init_file(self) -> Path:
|
||||
"""Get the path to init.lua or init.vim."""
|
||||
init_lua = self.get_config_dir() / "init.lua"
|
||||
init_vim = self.get_config_dir() / "init.vim"
|
||||
|
||||
if init_lua.exists():
|
||||
return init_lua
|
||||
elif init_vim.exists():
|
||||
return init_vim
|
||||
return init_lua
|
||||
|
||||
def get_settings_file(self) -> Optional[Path]:
|
||||
return self.get_init_file()
|
||||
|
||||
def sync_settings(self, settings_source: Path) -> EditorSyncResult:
|
||||
"""Sync Neovim config from source file."""
|
||||
try:
|
||||
self._ensure_config_dir()
|
||||
|
||||
if not settings_source.exists():
|
||||
return EditorSyncResult(
|
||||
success=False,
|
||||
editor=self.name,
|
||||
operation="sync_settings",
|
||||
details=f"Source config not found: {settings_source}",
|
||||
)
|
||||
|
||||
content = self.file_ops.read_file(settings_source)
|
||||
if content is None:
|
||||
return EditorSyncResult(
|
||||
success=False,
|
||||
editor=self.name,
|
||||
operation="sync_settings",
|
||||
details=f"Failed to read config source: {settings_source}",
|
||||
)
|
||||
|
||||
target = self.get_init_file()
|
||||
success = self.file_ops.write_file(target, content)
|
||||
|
||||
if success:
|
||||
return EditorSyncResult(
|
||||
success=True,
|
||||
editor=self.name,
|
||||
operation="sync_settings",
|
||||
details=f"Synced config to {target}",
|
||||
path=target,
|
||||
)
|
||||
else:
|
||||
return EditorSyncResult(
|
||||
success=False,
|
||||
editor=self.name,
|
||||
operation="sync_settings",
|
||||
details=f"Failed to write config: {target}",
|
||||
)
|
||||
except Exception as e:
|
||||
return EditorSyncResult(
|
||||
success=False,
|
||||
editor=self.name,
|
||||
operation="sync_settings",
|
||||
details=f"Error syncing config: {e}",
|
||||
)
|
||||
|
||||
def sync_extensions(self, extensions: List[EditorExtension]) -> List[EditorSyncResult]:
|
||||
"""Sync Neovim plugins (vim-plug installation commands)."""
|
||||
results = []
|
||||
|
||||
self.logger.info("Neovim plugins detected:")
|
||||
for ext in extensions:
|
||||
result = self._install_plugin(ext.name)
|
||||
results.append(result)
|
||||
|
||||
return results
|
||||
|
||||
def _install_plugin(self, plugin: str) -> EditorSyncResult:
|
||||
"""Generate plugin installation information."""
|
||||
install_cmd = f"Plug '{plugin}' # Add to init.lua, then run :PlugInstall"
|
||||
|
||||
self.logger.info(f" Neovim plugin: {plugin}")
|
||||
self.logger.info(f" Add to init.lua: {install_cmd}")
|
||||
|
||||
return EditorSyncResult(
|
||||
success=True,
|
||||
editor=self.name,
|
||||
operation="install_plugin",
|
||||
details=f"Plugin configuration: {plugin}",
|
||||
)
|
||||
|
||||
def generate_vim_plug_setup(self) -> str:
|
||||
"""Generate vim-plug setup code."""
|
||||
return r'''
|
||||
" Install vim-plug if not installed
|
||||
if empty(glob('~/.config/nvim/autoload/plug.vim'))
|
||||
silent !curl -fLo ~/.config/nvim/autoload/plug.vim --create-dirs
|
||||
\ https://raw.githubusercontent.com/junegunn/vim-plug/master/plug.vim
|
||||
autocmd VimEnter * PlugInstall --sync | source $MYVIMRC
|
||||
endif
|
||||
|
||||
" Add your plugins between plug#begin and plug#end
|
||||
call plug#begin('~/.config/nvim/plugged')
|
||||
|
||||
" Plugins go here
|
||||
" Plug 'plugin_author/plugin_name'
|
||||
|
||||
call plug#end()
|
||||
'''
|
||||
|
||||
def sync_init_lua(self, source: Path) -> EditorSyncResult:
|
||||
"""Sync init.lua configuration."""
|
||||
return self.sync_settings(source)
|
||||
|
||||
def sync_lua_config(self, config_dir: Path) -> List[EditorSyncResult]:
|
||||
"""Sync entire lua config directory."""
|
||||
results = []
|
||||
self._ensure_config_dir()
|
||||
|
||||
lua_target_dir = self.get_config_dir() / "lua"
|
||||
if config_dir.exists():
|
||||
for lua_file in config_dir.rglob("*.lua"):
|
||||
relative_path = lua_file.relative_to(config_dir)
|
||||
target = lua_target_dir / relative_path
|
||||
result = self.sync_settings(lua_file)
|
||||
results.append(result)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
class EditorManager:
|
||||
"""Manager for all editor configurations."""
|
||||
|
||||
EDITORS = {
|
||||
"vscode": VSCodeConfig,
|
||||
"neovim": NeovimConfig,
|
||||
}
|
||||
|
||||
def __init__(self, config: ConfigSchema, dry_run: bool = False, logger=None):
|
||||
self.config = config
|
||||
self.dry_run = dry_run
|
||||
self.logger = logger or get_logger(__name__)
|
||||
self._editors: Dict[str, EditorConfig] = {}
|
||||
|
||||
def _get_editor(self, editor_name: str) -> Optional[EditorConfig]:
|
||||
"""Get or create an editor configuration handler."""
|
||||
if editor_name not in self.EDITORS:
|
||||
self.logger.warning(f"Unknown editor: {editor_name}")
|
||||
return None
|
||||
|
||||
if editor_name not in self._editors:
|
||||
self._editors[editor_name] = self.EDITORS[editor_name](
|
||||
dry_run=self.dry_run,
|
||||
logger=self.logger,
|
||||
)
|
||||
|
||||
return self._editors[editor_name]
|
||||
|
||||
def sync_editor(self, editor_name: str) -> List[EditorSyncResult]:
|
||||
"""Sync a specific editor's configuration."""
|
||||
results = []
|
||||
|
||||
if editor_name not in self.config.editors:
|
||||
self.logger.warning(f"Editor '{editor_name}' not in configuration")
|
||||
return results
|
||||
|
||||
editor_config = self._get_editor(editor_name)
|
||||
if editor_config is None:
|
||||
return results
|
||||
|
||||
editor_settings = self.config.editors[editor_name]
|
||||
|
||||
if "settings" in editor_settings:
|
||||
if isinstance(editor_settings["settings"], dict):
|
||||
settings_file = editor_settings["settings"].get("settings_file")
|
||||
if settings_file:
|
||||
result = editor_config.sync_settings(Path(settings_file))
|
||||
results.append(result)
|
||||
elif isinstance(editor_settings["settings"], str):
|
||||
result = editor_config.sync_settings(Path(editor_settings["settings"]))
|
||||
results.append(result)
|
||||
|
||||
if "extensions" in editor_settings:
|
||||
extensions = editor_settings["extensions"]
|
||||
ext_results = editor_config.sync_extensions(extensions)
|
||||
results.extend(ext_results)
|
||||
|
||||
return results
|
||||
|
||||
def sync_all(self) -> Dict[str, List[EditorSyncResult]]:
|
||||
"""Sync all editor configurations."""
|
||||
results = {}
|
||||
|
||||
for editor_name in self.config.editors:
|
||||
editor_results = self.sync_editor(editor_name)
|
||||
results[editor_name] = editor_results
|
||||
|
||||
return results
|
||||
|
||||
def print_summary(self, results: Dict[str, List[EditorSyncResult]]):
|
||||
"""Print a summary of editor sync results."""
|
||||
self.logger.info("Editor Configuration Summary:")
|
||||
|
||||
for editor, editor_results in results.items():
|
||||
self.logger.info(f" {editor}:")
|
||||
for result in editor_results:
|
||||
status = "OK" if result.success else "FAIL"
|
||||
self.logger.info(f" [{status}] {result.operation}: {result.details}")
|
||||
Reference in New Issue
Block a user