Compare commits

...

2 Commits

Author SHA1 Message Date
2cdc4c6c42 refactor: tidy logger handler setup 2026-01-15 16:06:32 +01:00
07862f0ea2 refactor: simplify configuration create flow 2026-01-15 16:02:21 +01:00
2 changed files with 81 additions and 45 deletions

View File

@@ -56,65 +56,100 @@ class Configuration:
def exists(self) -> bool:
return self.config_file.exists()
def create(self):
logger = setup_logger(verbose=False)
if self.exists():
def _confirm_overwrite(self, logger) -> bool:
if not self.exists():
return True
overwrite = input(
"Configuration already exists. Overwrite? (y/N): ").strip().lower()
if overwrite not in ("y", "yes"):
if overwrite in ("y", "yes"):
return True
logger.info("Configuration creation cancelled.")
return
return False
def _ensure_config_dir(self) -> None:
config_dir = self.config_file.parent
config_dir.mkdir(parents=True, exist_ok=True)
print("Skywipe Configuration")
print("=" * 50)
print("Note: You should use an app password from Bluesky settings.")
def _prompt_handle(self, logger) -> str:
while True:
handle = input("Bluesky handle: ").strip()
is_valid, error_msg = _validate_handle(handle)
if is_valid:
break
return handle
logger.error(error_msg)
logger.info("Please enter a valid handle and try again.")
def _prompt_password(self, logger) -> str:
while True:
password = getpass(
"Bluesky (hopefully app) password: ").strip()
is_valid, error_msg = _validate_password(password)
if is_valid:
break
return password
logger.error(error_msg)
logger.info("Please check your password and try again.")
logger.info(
"Generate an app password at: https://bsky.app/settings/app-passwords")
def _parse_batch_size(self, logger) -> int | None:
batch_size = input("Batch size (default: 10): ").strip() or "10"
delay = input(
"Delay between batches in seconds (default: 1): ").strip() or "1"
verbose_input = input(
"Verbose mode (y/n, default: y): ").strip().lower() or "y"
verbose = verbose_input in ("y", "yes", "true", "1")
try:
batch_size = int(batch_size)
if batch_size < 1 or batch_size > 100:
logger.error("batch_size must be between 1 and 100")
return
batch_size_int = int(batch_size)
except ValueError:
logger.error("batch_size must be an integer")
return
return None
if batch_size_int < 1 or batch_size_int > 100:
logger.error("batch_size must be between 1 and 100")
return None
return batch_size_int
def _parse_delay(self, logger) -> int | None:
delay = input(
"Delay between batches in seconds (default: 1): ").strip() or "1"
try:
delay = int(delay)
if delay < 0 or delay > 60:
logger.error("delay must be between 0 and 60 seconds")
return
delay_int = int(delay)
except ValueError:
logger.error("delay must be an integer")
return None
if delay_int < 0 or delay_int > 60:
logger.error("delay must be between 0 and 60 seconds")
return None
return delay_int
def _parse_verbose(self) -> bool:
verbose_input = input(
"Verbose mode (y/n, default: y): ").strip().lower() or "y"
return verbose_input in ("y", "yes", "true", "1")
def _write_config(self, logger, config_data: dict) -> None:
try:
with open(self.config_file, "w") as f:
yaml.dump(config_data, f, default_flow_style=False)
except (IOError, OSError) as e:
logger.error(f"Failed to save configuration: {e}")
return
logger.info(f"Configuration saved to {self.config_file}")
def create(self):
logger = setup_logger(verbose=False)
if not self._confirm_overwrite(logger):
return
self._ensure_config_dir()
print("Skywipe Configuration")
print("=" * 50)
print("Note: You should use an app password from Bluesky settings.")
handle = self._prompt_handle(logger)
password = self._prompt_password(logger)
batch_size = self._parse_batch_size(logger)
if batch_size is None:
return
delay = self._parse_delay(logger)
if delay is None:
return
verbose = self._parse_verbose()
config_data = {
"handle": handle,
@@ -124,14 +159,7 @@ class Configuration:
"verbose": verbose
}
try:
with open(self.config_file, "w") as f:
yaml.dump(config_data, f, default_flow_style=False)
except (IOError, OSError) as e:
logger.error(f"Failed to save configuration: {e}")
return
logger.info(f"Configuration saved to {self.config_file}")
self._write_config(logger, config_data)
def load(self) -> dict:
if not self.exists():

View File

@@ -41,6 +41,7 @@ def setup_logger(verbose: bool = False, log_file: Path | None = None) -> logging
logger.setLevel(target_level)
progress_logger = logging.getLogger("skywipe.progress")
if not progress_logger.handlers:
progress_logger.propagate = True
info_handler = None
@@ -60,8 +61,11 @@ def setup_logger(verbose: bool = False, log_file: Path | None = None) -> logging
if info_handler is None:
info_handler = logging.StreamHandler(sys.stdout)
info_handler.addFilter(LevelFilter(logging.DEBUG, logging.INFO))
logger.addHandler(info_handler)
for existing in list(info_handler.filters):
if isinstance(existing, LevelFilter):
info_handler.removeFilter(existing)
info_handler.addFilter(LevelFilter(logging.DEBUG, logging.INFO))
info_handler.setFormatter(formatter)
info_handler.setLevel(target_level)
@@ -93,14 +97,18 @@ def get_logger() -> logging.Logger:
return logging.getLogger("skywipe")
def handle_error(error: Exception, logger: logging.Logger, exit_on_error: bool = False) -> None:
def _format_error_message(error: Exception) -> str:
if isinstance(error, KeyError):
message = error.args[0] if error.args else str(error)
logger.error(f"{message}")
elif isinstance(error, ValueError):
logger.error(f"{error}")
return str(error.args[0]) if error.args else str(error)
return str(error)
def handle_error(error: Exception, logger: logging.Logger, exit_on_error: bool = False) -> None:
if isinstance(error, (KeyError, ValueError)):
logger.error(_format_error_message(error))
else:
logger.error(f"Unexpected error: {error}", exc_info=True)
logger.error(
f"Unexpected error: {_format_error_message(error)}", exc_info=True)
if exit_on_error:
sys.exit(1)