docs: update roadmap

README.md

@@ -1,16 +1,16 @@
 # Skywipe
 
-Skywipe is a work-in-progress Python 3.13+ CLI that helps you wipe data from your Bluesky account using the AT Protocol.
+Skywipe is a work-in-progress Python 3.13+ CLI that helps you wipe data from your Bluesky account using the AT Protocol SDK.
 
-## Warning
+**Warning:** This tool performs _**destructive operations**_.
 
-This tool performs destructive operations. Only use it if you intend to erase data from your Bluesky account.
+Only use it if you intend to permanently erase data from your Bluesky account.
 
 ## Requirements
 
-We're using [`uv`](https://github.com/astral-sh/uv) for dependency and virtual environment management.
+Check [pyproject.toml](pyproject.toml).
 
-You can setup the project (aka create a virtual environment and install dependencies) with :
+You can use `uv` to install dependencies:
 
 ```bash
 git clone https://git.kharec.info/Kharec/skywipe.git
@@ -20,37 +20,31 @@ uv sync
 
 ## How to run
 
-When installation will be worked out, you'll be able to :
-
-```bash
-skywipe all            # target everything
-skywipe configure      # create configuration
-skywipe posts          # only posts
-skywipe medias         # only posts with medias
-skywipe likes          # only likes
-skywipe reposts        # only reposts
-skywipe follows        # only follows
-```
-
 While it's being developed, you can use the tool using `uv` :
 
 ```bash
-uv run main.py all            # target everything
+uv run python -m skywipe.cli all            # target everything
-uv run main.py configure      # create configuration
+uv run python -m skywipe.cli configure      # create configuration
-uv run main.py posts          # only posts
+uv run python -m skywipe.cli posts          # delete posts
-uv run main.py medias         # only posts with medias
+uv run python -m skywipe.cli medias         # delete posts with medias
-uv run main.py likes          # only likes
+uv run python -m skywipe.cli likes          # undo likes
-uv run main.py reposts        # only reposts
+uv run python -m skywipe.cli reposts        # undo reposts
-uv run main.py follows        # only follows
+uv run python -m skywipe.cli quotes         # delete quotes
+uv run python -m skywipe.cli follows        # unfollow all
+uv run python -m skywipe.cli bookmarks      # delete bookmarks
 ```
 
-### Configuration
+Use the `--yes` flag to skip the confirmation prompt and proceed with the operation.
+
+A log of the operations will be saved in `~/.cache/skywipe/skywipe.log`.
+
+## Configuration
 
 If you run the tool for the first time, it will prompt you to use `skywipe configure` to create the configuration file, which is located in `~/.config/skywipe/config.yml` :
 
 ```yaml
 handle: your_handle
-password: your_password_encrypted
+password: your_password
 batch_size: 10
 delay: 1
 verbose: true
@@ -60,17 +54,20 @@ BE SURE TO USE A [BLUESKY APP PASSWORD](https://blueskyfeeds.com/faq-app-passwor
 
 ## Roadmap
 
-- [ ] build cli parameter management
+- [x] build cli parameter management
-- [ ] handle configuration logic
+- [x] handle configuration logic
-- [ ] sign in to at protocol
+- [x] sign in to at protocol
-- [ ] delete posts in groups
+- [x] delete posts in batch
-- [ ] only delete posts with media
+- [x] only delete posts with media
-- [ ] remove likes
+- [x] undo likes
-- [ ] remove reposts
+- [x] undo reposts
-- [ ] unfollow accounts
+- [x] delete quotes
-- [ ] make `all` run the other commands
+- [x] unfollow accounts
-- [ ] add simple progress and logging
+- [x] remove bookmarks
-- [ ] add safeguards like confirmations and clear dry-run info
+- [x] make `all` run the other commands
+- [x] add simple progress and logging
+- [x] add safeguards (confirmation, dry-run flag)
+- [x] decent code architecture
 - [ ] installation and run process
 
 ## License

main.py

@@ -1,26 +0,0 @@
-"""Main entry point for Skywipe"""
-
-import sys
-import argparse
-
-from skywipe.commands import *
-
-
-def main():
-    parser = argparse.ArgumentParser(
-        description="Clean your bluesky account with style")
-    parser.add_argument(
-        "command",
-        choices=["all", "configure", "posts",
-                 "medias", "likes", "reposts", "follows"],
-        help="Command to execute"
-    )
-
-    args = parser.parse_args()
-
-    if args.command == "configure":
-        run_configure()
-    
-
-if __name__ == '__main__':
-    main()

pyproject.toml

@@ -1,11 +1,10 @@
 [project]
 name = "skywipe"
 version = "0.1.0"
-description = "Clean your bluesky account with style"
+description = "Clean your bluesky account"
 readme = "README.md"
 requires-python = ">=3.13"
-dependencies = [
+dependencies = ["atproto>=0.0.65", "pyyaml>=6.0"]
-    "atproto>=0.0.65",
+
-    "pyyaml>=6.0",
+[project.scripts]
-    "cryptography>=42.0.0",
+skywipe = "skywipe.cli:main"
-]

skywipe/auth.py

@@ -0,0 +1,28 @@
+"""Authentication module for Skywipe"""
+
+from atproto import Client
+from .configure import Configuration
+
+
+class Auth:
+    def __init__(self):
+        self.config = Configuration()
+        self.client = None
+
+    def login(self) -> Client:
+        config_data = self.config.load()
+
+        handle = config_data.get("handle")
+        password = config_data.get("password")
+
+        if not handle or not password:
+            raise ValueError(
+                "handle and password must be set in configuration")
+
+        self.client = Client()
+        self.client.login(handle, password)
+
+        return self.client
+
+    def is_logged(self) -> bool:
+        return bool(getattr(self.client, "me", None))

skywipe/cli.py

@@ -0,0 +1,76 @@
+"""Main entry point for Skywipe"""
+
+import sys
+import argparse
+from pathlib import Path
+
+from .commands import registry
+from .configure import Configuration
+from .logger import setup_logger
+
+
+def create_parser():
+    commands = registry.get_all_commands()
+
+    parser = argparse.ArgumentParser(
+        description="Clean your bluesky account with style.",
+        epilog="WARNING: This tool performs destructive operations. Only use it if you intend to erase data from your Bluesky account."
+    )
+
+    parser.add_argument(
+        "--yes",
+        action="store_true",
+        help="Skip confirmation prompt and proceed with destructive operations"
+    )
+
+    subparsers = parser.add_subparsers(
+        dest="command",
+        help="Command to execute",
+        metavar="COMMAND",
+        required=True
+    )
+
+    for cmd, help_text in commands.items():
+        subparsers.add_parser(cmd, help=help_text)
+
+    return parser
+
+
+def require_config():
+    config = Configuration()
+    if not config.exists():
+        logger = setup_logger(verbose=False)
+        logger.error("Configuration file not found.")
+        logger.error("You must run 'skywipe configure' first.")
+        sys.exit(1)
+
+
+def main():
+    parser = create_parser()
+    args = parser.parse_args()
+
+    if registry.requires_config(args.command):
+        require_config()
+        config = Configuration()
+        config_data = config.load()
+        verbose = config_data.get("verbose", False)
+        log_file = Path.home() / ".cache" / "skywipe" / "skywipe.log"
+        setup_logger(verbose=verbose, log_file=log_file)
+    else:
+        setup_logger(verbose=False)
+
+    try:
+        registry.execute(
+            args.command, skip_confirmation=getattr(args, "yes", False))
+    except ValueError as e:
+        logger = setup_logger(verbose=False)
+        logger.error(f"{e}")
+        sys.exit(1)
+    except Exception as e:
+        logger = setup_logger(verbose=False)
+        logger.error(f"Unexpected error: {e}", exc_info=True)
+        sys.exit(1)
+
+
+if __name__ == '__main__':
+    main()

skywipe/commands.py

@@ -0,0 +1,130 @@
+"""Command implementations for Skywipe"""
+
+from typing import Callable, Dict, Optional
+from .configure import Configuration
+from .operations import Operation
+from .post_analysis import PostAnalyzer
+from .logger import get_logger
+from .safeguard import require_confirmation
+
+
+CommandHandler = Callable[..., None]
+
+
+class CommandRegistry:
+    def __init__(self):
+        self._commands = {}
+        self._help_texts = {}
+        self._requires_config = {}
+
+    def register(
+        self,
+        name: str,
+        handler: CommandHandler,
+        help_text: str,
+        requires_config: bool = True
+    ):
+        self._commands[name] = handler
+        self._help_texts[name] = help_text
+        self._requires_config[name] = requires_config
+
+    def get_handler(self, name: str) -> Optional[CommandHandler]:
+        return self._commands.get(name)
+
+    def get_help_text(self, name: str) -> Optional[str]:
+        return self._help_texts.get(name)
+
+    def requires_config(self, name: str) -> bool:
+        return self._requires_config.get(name, True)
+
+    def get_all_commands(self) -> Dict[str, str]:
+        return self._help_texts.copy()
+
+    def execute(self, name: str, skip_confirmation: bool = False):
+        handler = self.get_handler(name)
+        if handler:
+            if name == "configure":
+                handler()
+            else:
+                handler(skip_confirmation)
+        else:
+            raise ValueError(f"Unknown command: {name}")
+
+
+registry = CommandRegistry()
+
+
+def run_configure():
+    config = Configuration()
+    config.create()
+
+
+def run_posts(skip_confirmation: bool = False):
+    require_confirmation("delete all posts", skip_confirmation)
+    Operation("Deleting posts").run()
+
+
+def run_medias(skip_confirmation: bool = False):
+    require_confirmation("delete all posts with media", skip_confirmation)
+    Operation("Deleting posts with media",
+              filter_fn=lambda post: PostAnalyzer.has_media(post.post)).run()
+
+
+def run_likes(skip_confirmation: bool = False):
+    require_confirmation("undo all likes", skip_confirmation)
+    Operation("Undoing likes", strategy_type="record",
+              collection="app.bsky.feed.like").run()
+
+
+def run_reposts(skip_confirmation: bool = False):
+    require_confirmation("undo all reposts", skip_confirmation)
+    Operation("Undoing reposts", strategy_type="record",
+              collection="app.bsky.feed.repost").run()
+
+
+def run_quotes(skip_confirmation: bool = False):
+    require_confirmation("delete all quote posts", skip_confirmation)
+    Operation("Deleting quote posts",
+              filter_fn=lambda post: PostAnalyzer.has_quote(post.post)).run()
+
+
+def run_follows(skip_confirmation: bool = False):
+    require_confirmation("unfollow all accounts", skip_confirmation)
+    Operation("Unfollowing accounts", strategy_type="record",
+              collection="app.bsky.graph.follow").run()
+
+
+def run_bookmarks(skip_confirmation: bool = False):
+    require_confirmation("delete all bookmarks", skip_confirmation)
+    Operation("Deleting bookmarks", strategy_type="bookmark").run()
+
+
+def run_all(skip_confirmation: bool = False):
+    logger = get_logger()
+    require_confirmation(
+        "run all cleanup commands (posts, likes, reposts, follows, bookmarks)", skip_confirmation)
+
+    commands = ["posts", "likes", "reposts", "follows", "bookmarks"]
+
+    logger.info("Running all cleanup commands...")
+    for cmd in commands:
+        try:
+            logger.info(f"Starting command: {cmd}")
+            registry.execute(cmd, skip_confirmation=True)
+            logger.info(f"Completed command: {cmd}")
+        except Exception as e:
+            logger.error(f"Error running '{cmd}': {e}", exc_info=True)
+            continue
+    logger.info("All commands completed.")
+
+
+registry.register("configure", run_configure,
+                  "create configuration", requires_config=False)
+registry.register("posts", run_posts, "only posts")
+registry.register("medias", run_medias, "only posts with medias")
+registry.register("likes", run_likes, "only likes")
+registry.register("reposts", run_reposts, "only reposts")
+registry.register("quotes", run_quotes, "only quotes")
+registry.register("follows", run_follows, "only follows")
+registry.register("bookmarks", run_bookmarks, "only bookmarks")
+registry.register("all", run_all, "target everything")

skywipe/configure.py

@@ -0,0 +1,64 @@
+"""Core configuration module for Skywipe"""
+
+import getpass
+from pathlib import Path
+import yaml
+from .logger import setup_logger
+
+
+class Configuration:
+    def __init__(self):
+        self.config_file = Path.home() / ".config" / "skywipe" / "config.yml"
+
+    def exists(self) -> bool:
+        return self.config_file.exists()
+
+    def create(self):
+        logger = setup_logger(verbose=False)
+        if self.exists():
+            overwrite = input(
+                "Configuration already exists. Overwrite? (y/N): ").strip().lower()
+            if overwrite not in ("y", "yes"):
+                logger.info("Configuration creation cancelled.")
+                return
+
+        config_dir = self.config_file.parent
+        config_dir.mkdir(parents=True, exist_ok=True)
+
+        print("Skywipe Configuration")
+        print("=" * 50)
+        handle = input("Bluesky handle: ").strip()
+        password = getpass.getpass("Bluesky app password: ").strip()
+        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)
+            delay = int(delay)
+        except ValueError:
+            logger.error("batch_size and delay must be integers")
+            return
+
+        config_data = {
+            "handle": handle,
+            "password": password,
+            "batch_size": batch_size,
+            "delay": delay,
+            "verbose": verbose
+        }
+
+        with open(self.config_file, "w") as f:
+            yaml.dump(config_data, f, default_flow_style=False)
+
+        logger.info(f"Configuration saved to {self.config_file}")
+
+    def load(self) -> dict:
+        if not self.exists():
+            raise FileNotFoundError(
+                f"Configuration file not found: {self.config_file}")
+        with open(self.config_file, "r") as f:
+            return yaml.safe_load(f)

skywipe/logger.py

@@ -0,0 +1,73 @@
+"""Centralized logging module for Skywipe"""
+
+import logging
+import sys
+from pathlib import Path
+from typing import Optional
+
+
+class ProgressTracker:
+    def __init__(self, operation: str = "Processing"):
+        self.current = 0
+        self.operation = operation
+
+    def update(self, count: int = 1):
+        self.current += count
+
+    def batch(self, batch_num: int, batch_size: int, total_batches: Optional[int] = None):
+        logger = logging.getLogger("skywipe.progress")
+        if total_batches:
+            logger.info(
+                f"{self.operation} - batch {batch_num}/{total_batches} ({batch_size} items)"
+            )
+        else:
+            logger.info(
+                f"{self.operation} - batch {batch_num} ({batch_size} items)")
+
+
+class LevelFilter(logging.Filter):
+    def __init__(self, min_level: int, max_level: int):
+        super().__init__()
+        self.min_level = min_level
+        self.max_level = max_level
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        return self.min_level <= record.levelno <= self.max_level
+
+
+def setup_logger(verbose: bool = False, log_file: Optional[Path] = None) -> logging.Logger:
+    logger = logging.getLogger("skywipe")
+    logger.setLevel(logging.DEBUG if verbose else logging.INFO)
+
+    if logger.handlers:
+        return logger
+
+    formatter = logging.Formatter(fmt="%(levelname)s: %(message)s")
+
+    info_handler = logging.StreamHandler(sys.stdout)
+    info_handler.setLevel(logging.DEBUG if verbose else logging.INFO)
+    info_handler.addFilter(LevelFilter(logging.DEBUG, logging.INFO))
+    info_handler.setFormatter(formatter)
+    logger.addHandler(info_handler)
+
+    error_handler = logging.StreamHandler(sys.stderr)
+    error_handler.setLevel(logging.WARNING)
+    error_handler.setFormatter(formatter)
+    logger.addHandler(error_handler)
+
+    if log_file:
+        log_file.parent.mkdir(parents=True, exist_ok=True)
+        file_handler = logging.FileHandler(log_file, encoding="utf-8")
+        file_handler.setLevel(logging.DEBUG)
+        file_formatter = logging.Formatter(
+            fmt="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+            datefmt="%Y-%m-%d %H:%M:%S"
+        )
+        file_handler.setFormatter(file_formatter)
+        logger.addHandler(file_handler)
+
+    return logger
+
+
+def get_logger() -> logging.Logger:
+    return logging.getLogger("skywipe")

skywipe/operations.py

@@ -0,0 +1,222 @@
+"""Shared operation utilities and strategies for Skywipe"""
+
+import time
+from typing import Callable, Optional, Any
+from atproto import models
+
+from .auth import Auth
+from .configure import Configuration
+from .logger import get_logger, ProgressTracker
+
+
+class OperationContext:
+    def __init__(self):
+        self.logger = get_logger()
+        self.auth = Auth()
+        self.client = self.auth.login()
+        self.config = Configuration()
+        self.config_data = self.config.load()
+        self.batch_size = self.config_data.get("batch_size", 10)
+        self.delay = self.config_data.get("delay", 1)
+        self.did = self.client.me.did
+
+
+class RecordDeletionStrategy:
+    def __init__(self, collection: str):
+        self.collection = collection
+
+    def fetch(self, context: OperationContext, cursor: Optional[str] = None):
+        list_params = models.ComAtprotoRepoListRecords.Params(
+            repo=context.did,
+            collection=self.collection,
+            limit=context.batch_size,
+            cursor=cursor
+        )
+        return context.client.com.atproto.repo.list_records(params=list_params)
+
+    def extract_items(self, response):
+        return response.records
+
+    def get_cursor(self, response):
+        return response.cursor
+
+    def process_item(self, record, context: OperationContext):
+        record_uri = record.uri
+        rkey = record_uri.rsplit("/", 1)[-1]
+        delete_data = {
+            "repo": context.did,
+            "collection": self.collection,
+            "rkey": rkey
+        }
+        context.client.com.atproto.repo.delete_record(data=delete_data)
+        context.logger.debug(f"Deleted: {record_uri}")
+
+
+class FeedStrategy:
+    def fetch(self, context: OperationContext, cursor: Optional[str] = None):
+        if cursor:
+            return context.client.get_author_feed(
+                actor=context.did, limit=context.batch_size, cursor=cursor
+            )
+        return context.client.get_author_feed(actor=context.did, limit=context.batch_size)
+
+    def extract_items(self, response):
+        return response.feed
+
+    def get_cursor(self, response):
+        return response.cursor
+
+    def process_item(self, post, context: OperationContext):
+        uri = post.post.uri
+        context.client.delete_post(uri)
+        context.logger.debug(f"Deleted post: {uri}")
+
+
+class BookmarkStrategy:
+    def fetch(self, context: OperationContext, cursor: Optional[str] = None):
+        get_params = models.AppBskyBookmarkGetBookmarks.Params(
+            limit=context.batch_size,
+            cursor=cursor
+        )
+        return context.client.app.bsky.bookmark.get_bookmarks(params=get_params)
+
+    def extract_items(self, response):
+        return response.bookmarks
+
+    def get_cursor(self, response):
+        return response.cursor
+
+    def process_item(self, bookmark, context: OperationContext):
+        bookmark_uri = self._extract_bookmark_uri(bookmark)
+        if not bookmark_uri:
+            raise ValueError("Unable to find bookmark URI")
+
+        delete_data = models.AppBskyBookmarkDeleteBookmark.Data(
+            uri=bookmark_uri)
+        context.client.app.bsky.bookmark.delete_bookmark(data=delete_data)
+        context.logger.debug(f"Deleted bookmark: {bookmark_uri}")
+
+    def _extract_bookmark_uri(self, bookmark):
+        if hasattr(bookmark, "uri"):
+            return bookmark.uri
+
+        for attr_name in ("subject", "record", "post", "item"):
+            if hasattr(bookmark, attr_name):
+                nested = getattr(bookmark, attr_name)
+                if hasattr(nested, "uri"):
+                    return nested.uri
+        return None
+
+
+class Operation:
+    def __init__(
+        self,
+        operation_name: str,
+        strategy_type: str = "feed",
+        collection: Optional[str] = None,
+        filter_fn: Optional[Callable[[Any], bool]] = None
+    ):
+        self.operation_name = operation_name
+        self.filter_fn = filter_fn
+
+        if strategy_type == "record":
+            if not collection:
+                raise ValueError("Collection is required for record strategy")
+            self.strategy = RecordDeletionStrategy(collection)
+        elif strategy_type == "bookmark":
+            self.strategy = BookmarkStrategy()
+        else:
+            self.strategy = FeedStrategy()
+
+    def run(self) -> int:
+        context = OperationContext()
+        progress = ProgressTracker(operation=self.operation_name)
+
+        context.logger.info(
+            f"Starting {self.operation_name} with batch_size={context.batch_size}, delay={context.delay}s"
+        )
+
+        cursor = None
+        total_processed = 0
+        batch_num = 0
+
+        while True:
+            batch_num += 1
+            response = self.strategy.fetch(context, cursor)
+            items = self.strategy.extract_items(response)
+
+            if not items:
+                break
+
+            progress.batch(batch_num, len(items))
+
+            for item in items:
+                if self.filter_fn and not self.filter_fn(item):
+                    continue
+
+                try:
+                    self.strategy.process_item(item, context)
+                    total_processed += 1
+                    progress.update(1)
+                except Exception as e:
+                    context.logger.error(f"Error processing item: {e}")
+
+            cursor = self.strategy.get_cursor(response)
+            if not cursor:
+                break
+
+            if context.delay > 0:
+                time.sleep(context.delay)
+
+        context.logger.info(
+            f"{self.operation_name}: {total_processed} items processed.")
+        return total_processed
+
+
+def run_operation(
+    strategy,
+    operation_name: str,
+    filter_fn: Optional[Callable[[Any], bool]] = None
+) -> int:
+    context = OperationContext()
+    progress = ProgressTracker(operation=operation_name)
+
+    context.logger.info(
+        f"Starting {operation_name} with batch_size={context.batch_size}, delay={context.delay}s"
+    )
+
+    cursor = None
+    total_processed = 0
+    batch_num = 0
+
+    while True:
+        batch_num += 1
+        response = strategy.fetch(context, cursor)
+        items = strategy.extract_items(response)
+
+        if not items:
+            break
+
+        progress.batch(batch_num, len(items))
+
+        for item in items:
+            if filter_fn and not filter_fn(item):
+                continue
+
+            try:
+                strategy.process_item(item, context)
+                total_processed += 1
+                progress.update(1)
+            except Exception as e:
+                context.logger.error(f"Error processing item: {e}")
+
+        cursor = strategy.get_cursor(response)
+        if not cursor:
+            break
+
+        if context.delay > 0:
+            time.sleep(context.delay)
+
+    context.logger.info(
+        f"{operation_name}: {total_processed} items processed.")
+    return total_processed

skywipe/post_analysis.py

@@ -0,0 +1,60 @@
+"""Post analysis utilities for Skywipe"""
+
+
+class PostAnalyzer:
+    MEDIA_TYPES = {
+        'app.bsky.embed.images',
+        'app.bsky.embed.video',
+        'app.bsky.embed.external'
+    }
+
+    QUOTE_TYPES = {
+        'app.bsky.embed.record',
+        'app.bsky.embed.recordWithMedia',
+        'app.bsky.embed.record_with_media'
+    }
+
+    @staticmethod
+    def has_media(post_record):
+        embed = getattr(post_record, 'embed', None)
+        if not embed:
+            return False
+
+        embed_type = getattr(embed, 'py_type', None)
+        if embed_type:
+            embed_type_base = embed_type.split('#')[0]
+
+            if embed_type_base in PostAnalyzer.MEDIA_TYPES:
+                return True
+
+            if embed_type_base in ('app.bsky.embed.recordWithMedia', 'app.bsky.embed.record_with_media'):
+                media = getattr(embed, 'media', None)
+                if media:
+                    media_type = getattr(media, 'py_type', None)
+                    if media_type:
+                        media_type_base = media_type.split('#')[0]
+                        if media_type_base in PostAnalyzer.MEDIA_TYPES:
+                            return True
+
+        for attr in ('images', 'video', 'external'):
+            if hasattr(embed, attr):
+                return True
+            if isinstance(embed, dict) and embed.get(attr):
+                return True
+
+        return False
+
+    @staticmethod
+    def has_quote(post_record):
+        embed = getattr(post_record, 'embed', None)
+        if not embed:
+            return False
+
+        embed_type = getattr(embed, 'py_type', None)
+        if embed_type:
+            embed_type_base = embed_type.split('#')[0]
+            if embed_type_base in PostAnalyzer.QUOTE_TYPES:
+                return True
+
+        return (hasattr(embed, 'record') or
+                (isinstance(embed, dict) and embed.get('record')))

skywipe/safeguard.py

@@ -0,0 +1,27 @@
+"""Safeguard module for Skywipe"""
+
+import sys
+from .logger import get_logger
+
+
+CONFIRM_RESPONSES = {"yes", "y"}
+
+
+def require_confirmation(operation: str, skip_confirmation: bool = False) -> None:
+    if skip_confirmation:
+        return
+
+    logger = get_logger()
+    logger.warning(f"This will {operation}")
+    logger.warning("This operation is DESTRUCTIVE and cannot be undone!")
+
+    try:
+        response = input(
+            "Are you sure you want to continue? (y/N): ").strip().lower()
+    except (EOFError, KeyboardInterrupt):
+        logger.info("\nOperation cancelled.")
+        sys.exit(0)
+
+    if response not in CONFIRM_RESPONSES:
+        logger.info("Operation cancelled.")
+        sys.exit(0)

uv.lock

@@ -381,14 +381,12 @@ version = "0.1.0"
 source = { virtual = "." }
 dependencies = [
     { name = "atproto" },
-    { name = "cryptography" },
     { name = "pyyaml" },
 ]
 
 [package.metadata]
 requires-dist = [
     { name = "atproto", specifier = ">=0.0.65" },
-    { name = "cryptography", specifier = ">=42.0.0" },
     { name = "pyyaml", specifier = ">=6.0" },
 ]