Add utility modules: file operations and logging
Some checks failed
CI / test (push) Failing after 10s
Some checks failed
CI / test (push) Failing after 10s
This commit is contained in:
155
dev_env_sync/utils/logging.py
Normal file
155
dev_env_sync/utils/logging.py
Normal file
@@ -0,0 +1,155 @@
|
||||
"""Logging utilities for dev-env-sync."""
|
||||
|
||||
import logging
|
||||
import os
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
|
||||
LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
||||
DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
|
||||
|
||||
|
||||
class ColoredFormatter(logging.Formatter):
|
||||
"""Custom formatter with colored output for console."""
|
||||
|
||||
COLORS = {
|
||||
"DEBUG": "\033[36m",
|
||||
"INFO": "\033[32m",
|
||||
"WARNING": "\033[33m",
|
||||
"ERROR": "\033[31m",
|
||||
"CRITICAL": "\033[35m",
|
||||
}
|
||||
RESET = "\033[0m"
|
||||
|
||||
def format(self, record: logging.LogRecord) -> str:
|
||||
if record.levelname in self.COLORS:
|
||||
color = self.COLORS[record.levelname]
|
||||
message = super().format(record)
|
||||
return f"{color}{message}{self.RESET}"
|
||||
return super().format(record)
|
||||
|
||||
|
||||
def setup_logging(
|
||||
verbose: bool = False,
|
||||
quiet: bool = False,
|
||||
log_file: Optional[str] = None,
|
||||
log_level: Optional[int] = None,
|
||||
) -> logging.Logger:
|
||||
"""Set up logging configuration."""
|
||||
logger = logging.getLogger("dev_env_sync")
|
||||
|
||||
if log_level is not None:
|
||||
logger.setLevel(log_level)
|
||||
elif verbose:
|
||||
logger.setLevel(logging.DEBUG)
|
||||
elif quiet:
|
||||
logger.setLevel(logging.WARNING)
|
||||
else:
|
||||
logger.setLevel(logging.INFO)
|
||||
|
||||
logger.handlers.clear()
|
||||
|
||||
console_handler = logging.StreamHandler()
|
||||
console_formatter = ColoredFormatter(LOG_FORMAT, datefmt=DATE_FORMAT)
|
||||
console_handler.setFormatter(console_formatter)
|
||||
logger.addHandler(console_handler)
|
||||
|
||||
if log_file:
|
||||
log_path = Path(log_file).expanduser()
|
||||
log_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
file_handler = logging.FileHandler(log_path, encoding="utf-8")
|
||||
file_formatter = logging.Formatter(LOG_FORMAT, datefmt=DATE_FORMAT)
|
||||
file_handler.setFormatter(file_formatter)
|
||||
logger.addHandler(file_handler)
|
||||
|
||||
return logger
|
||||
|
||||
|
||||
def get_logger(name: str = "dev_env_sync") -> logging.Logger:
|
||||
"""Get a logger instance."""
|
||||
return logging.getLogger(name)
|
||||
|
||||
|
||||
class OperationLogger:
|
||||
"""Logger for tracking operations in dry-run mode."""
|
||||
|
||||
def __init__(self, base_logger: logging.Logger, dry_run: bool = False):
|
||||
self.logger = base_logger
|
||||
self.dry_run = dry_run
|
||||
self.operations = []
|
||||
|
||||
def log_operation(self, operation_type: str, details: dict):
|
||||
"""Log an operation."""
|
||||
operation = {
|
||||
"type": operation_type,
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"details": details,
|
||||
}
|
||||
self.operations.append(operation)
|
||||
|
||||
if self.dry_run:
|
||||
self.logger.info(f"[DRY-RUN] {operation_type}: {details}")
|
||||
else:
|
||||
self.logger.debug(f"Executing {operation_type}: {details}")
|
||||
|
||||
def log_symlink_create(self, source: Path, target: Path):
|
||||
"""Log symlink creation."""
|
||||
self.log_operation("create_symlink", {
|
||||
"source": str(source),
|
||||
"target": str(target),
|
||||
})
|
||||
|
||||
def log_symlink_remove(self, target: Path):
|
||||
"""Log symlink removal."""
|
||||
self.log_operation("remove_symlink", {
|
||||
"target": str(target),
|
||||
})
|
||||
|
||||
def log_file_copy(self, source: Path, destination: Path):
|
||||
"""Log file copy."""
|
||||
self.log_operation("copy_file", {
|
||||
"source": str(source),
|
||||
"destination": str(destination),
|
||||
})
|
||||
|
||||
def log_file_backup(self, source: Path, backup_path: Path):
|
||||
"""Log file backup."""
|
||||
self.log_operation("backup_file", {
|
||||
"source": str(source),
|
||||
"backup_path": str(backup_path),
|
||||
})
|
||||
|
||||
def log_file_remove(self, path: Path):
|
||||
"""Log file removal."""
|
||||
self.log_operation("remove_file", {
|
||||
"path": str(path),
|
||||
})
|
||||
|
||||
def log_package_install(self, package_manager: str, package: str):
|
||||
"""Log package installation."""
|
||||
self.log_operation("install_package", {
|
||||
"package_manager": package_manager,
|
||||
"package": package,
|
||||
})
|
||||
|
||||
def log_editor_extension(self, editor: str, extension: str, action: str = "install"):
|
||||
"""Log editor extension installation."""
|
||||
self.log_operation("editor_extension", {
|
||||
"editor": editor,
|
||||
"extension": extension,
|
||||
"action": action,
|
||||
})
|
||||
|
||||
def get_operations_summary(self) -> str:
|
||||
"""Get a summary of all logged operations."""
|
||||
if not self.operations:
|
||||
return "No operations logged."
|
||||
|
||||
summary = []
|
||||
for op in self.operations:
|
||||
summary.append(f"- {op['type']}: {op['details']}")
|
||||
|
||||
return "\n".join(summary)
|
||||
Reference in New Issue
Block a user