Compare commits
4 Commits
054dc01813
...
ecbee7a8ac
| Author | SHA1 | Date | |
|---|---|---|---|
| ecbee7a8ac | |||
| eaf4e94d24 | |||
| 97e166d5f7 | |||
| 61e2d7f731 |
@@ -6,7 +6,7 @@ from pathlib import Path
|
|||||||
|
|
||||||
from .commands import registry
|
from .commands import registry
|
||||||
from .configure import Configuration
|
from .configure import Configuration
|
||||||
from .logger import setup_logger, get_logger
|
from .logger import setup_logger, get_logger, handle_error
|
||||||
|
|
||||||
|
|
||||||
LOG_FILE = Path.home() / ".cache" / "skywipe" / "skywipe.log"
|
LOG_FILE = Path.home() / ".cache" / "skywipe" / "skywipe.log"
|
||||||
@@ -39,10 +39,9 @@ def create_parser():
|
|||||||
return parser
|
return parser
|
||||||
|
|
||||||
|
|
||||||
def require_config():
|
def require_config(logger):
|
||||||
config = Configuration()
|
config = Configuration()
|
||||||
if not config.exists():
|
if not config.exists():
|
||||||
logger = get_logger()
|
|
||||||
logger.error("Configuration file not found.")
|
logger.error("Configuration file not found.")
|
||||||
logger.error("You must run 'skywipe configure' first.")
|
logger.error("You must run 'skywipe configure' first.")
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
@@ -53,9 +52,10 @@ def main():
|
|||||||
args = parser.parse_args()
|
args = parser.parse_args()
|
||||||
|
|
||||||
setup_logger(verbose=False, log_file=LOG_FILE)
|
setup_logger(verbose=False, log_file=LOG_FILE)
|
||||||
|
logger = get_logger()
|
||||||
|
|
||||||
if registry.requires_config(args.command):
|
if registry.requires_config(args.command):
|
||||||
require_config()
|
require_config(logger)
|
||||||
config = Configuration()
|
config = Configuration()
|
||||||
config_data = config.load()
|
config_data = config.load()
|
||||||
verbose = config_data.get("verbose", False)
|
verbose = config_data.get("verbose", False)
|
||||||
@@ -64,14 +64,8 @@ def main():
|
|||||||
try:
|
try:
|
||||||
registry.execute(
|
registry.execute(
|
||||||
args.command, skip_confirmation=getattr(args, "yes", False))
|
args.command, skip_confirmation=getattr(args, "yes", False))
|
||||||
except ValueError as e:
|
except (ValueError, Exception) as e:
|
||||||
logger = get_logger()
|
handle_error(e, logger, exit_on_error=True)
|
||||||
logger.error(f"{e}")
|
|
||||||
sys.exit(1)
|
|
||||||
except Exception as e:
|
|
||||||
logger = get_logger()
|
|
||||||
logger.error(f"Unexpected error: {e}", exc_info=True)
|
|
||||||
sys.exit(1)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ from typing import Callable, Dict, Optional, Any
|
|||||||
from .configure import Configuration
|
from .configure import Configuration
|
||||||
from .operations import Operation
|
from .operations import Operation
|
||||||
from .post_analysis import PostAnalyzer
|
from .post_analysis import PostAnalyzer
|
||||||
from .logger import get_logger
|
from .logger import get_logger, handle_error
|
||||||
from .safeguard import require_confirmation
|
from .safeguard import require_confirmation
|
||||||
|
|
||||||
|
|
||||||
@@ -131,8 +131,10 @@ def _create_operation_handler(
|
|||||||
collection: Optional[str] = None,
|
collection: Optional[str] = None,
|
||||||
filter_fn: Optional[Callable[[Any], bool]] = None
|
filter_fn: Optional[Callable[[Any], bool]] = None
|
||||||
) -> CommandHandler:
|
) -> CommandHandler:
|
||||||
|
logger = get_logger()
|
||||||
|
|
||||||
def handler(skip_confirmation: bool = False):
|
def handler(skip_confirmation: bool = False):
|
||||||
require_confirmation(confirmation_message, skip_confirmation)
|
require_confirmation(confirmation_message, skip_confirmation, logger)
|
||||||
try:
|
try:
|
||||||
Operation(
|
Operation(
|
||||||
operation_name,
|
operation_name,
|
||||||
@@ -140,15 +142,8 @@ def _create_operation_handler(
|
|||||||
collection=collection,
|
collection=collection,
|
||||||
filter_fn=filter_fn
|
filter_fn=filter_fn
|
||||||
).run()
|
).run()
|
||||||
except ValueError as e:
|
except (ValueError, Exception) as e:
|
||||||
logger = get_logger()
|
handle_error(e, logger)
|
||||||
logger.error(f"{e}")
|
|
||||||
raise
|
|
||||||
except Exception as e:
|
|
||||||
logger = get_logger()
|
|
||||||
logger.error(
|
|
||||||
f"Unexpected error during operation: {e}", exc_info=True)
|
|
||||||
raise
|
|
||||||
return handler
|
return handler
|
||||||
|
|
||||||
|
|
||||||
@@ -195,7 +190,7 @@ def run_all(skip_confirmation: bool = False):
|
|||||||
|
|
||||||
commands_str = ", ".join(commands)
|
commands_str = ", ".join(commands)
|
||||||
all_confirmation = f"run all cleanup commands ({commands_str})"
|
all_confirmation = f"run all cleanup commands ({commands_str})"
|
||||||
require_confirmation(all_confirmation, skip_confirmation)
|
require_confirmation(all_confirmation, skip_confirmation, logger)
|
||||||
|
|
||||||
logger.info("Running all cleanup commands...")
|
logger.info("Running all cleanup commands...")
|
||||||
|
|
||||||
|
|||||||
@@ -89,3 +89,15 @@ def setup_logger(verbose: bool = False, log_file: Optional[Path] = None) -> logg
|
|||||||
|
|
||||||
def get_logger() -> logging.Logger:
|
def get_logger() -> logging.Logger:
|
||||||
return logging.getLogger("skywipe")
|
return logging.getLogger("skywipe")
|
||||||
|
|
||||||
|
|
||||||
|
def handle_error(error: Exception, logger: logging.Logger, exit_on_error: bool = False) -> None:
|
||||||
|
if isinstance(error, ValueError):
|
||||||
|
logger.error(f"{error}")
|
||||||
|
else:
|
||||||
|
logger.error(f"Unexpected error: {error}", exc_info=True)
|
||||||
|
|
||||||
|
if exit_on_error:
|
||||||
|
sys.exit(1)
|
||||||
|
else:
|
||||||
|
raise error
|
||||||
|
|||||||
@@ -1,17 +1,21 @@
|
|||||||
"""Safeguard module for Skywipe"""
|
"""Safeguard module for Skywipe"""
|
||||||
|
|
||||||
import sys
|
import sys
|
||||||
|
import logging
|
||||||
|
from typing import Optional
|
||||||
from .logger import get_logger
|
from .logger import get_logger
|
||||||
|
|
||||||
|
|
||||||
CONFIRM_RESPONSES = {"yes", "y"}
|
CONFIRM_RESPONSES = {"yes", "y"}
|
||||||
|
|
||||||
|
|
||||||
def require_confirmation(operation: str, skip_confirmation: bool = False) -> None:
|
def require_confirmation(operation: str, skip_confirmation: bool = False, logger: Optional[logging.Logger] = None) -> None:
|
||||||
if skip_confirmation:
|
if skip_confirmation:
|
||||||
return
|
return
|
||||||
|
|
||||||
logger = get_logger()
|
if logger is None:
|
||||||
|
logger = get_logger()
|
||||||
|
|
||||||
logger.warning(f"This will {operation}")
|
logger.warning(f"This will {operation}")
|
||||||
logger.warning("This operation is DESTRUCTIVE and cannot be undone!")
|
logger.warning("This operation is DESTRUCTIVE and cannot be undone!")
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user