feat: implement our safeguard in commands
This commit is contained in:
@@ -10,9 +10,10 @@ from .quotes import delete_quotes_posts
|
||||
from .follows import unfollow_all
|
||||
from .bookmarks import delete_bookmarks
|
||||
from .logger import get_logger
|
||||
from .safeguard import require_confirmation
|
||||
|
||||
|
||||
CommandHandler = Callable[[], None]
|
||||
CommandHandler = Callable[..., None]
|
||||
|
||||
|
||||
class CommandRegistry:
|
||||
@@ -44,10 +45,13 @@ class CommandRegistry:
|
||||
def get_all_commands(self) -> Dict[str, str]:
|
||||
return self._help_texts.copy()
|
||||
|
||||
def execute(self, name: str):
|
||||
def execute(self, name: str, skip_confirmation: bool = False):
|
||||
handler = self.get_handler(name)
|
||||
if handler:
|
||||
handler()
|
||||
if name == "configure":
|
||||
handler()
|
||||
else:
|
||||
handler(skip_confirmation)
|
||||
else:
|
||||
raise ValueError(f"Unknown command: {name}")
|
||||
|
||||
@@ -60,43 +64,53 @@ def run_configure():
|
||||
config.create()
|
||||
|
||||
|
||||
def run_posts():
|
||||
def run_posts(skip_confirmation: bool = False):
|
||||
require_confirmation("delete all posts", skip_confirmation)
|
||||
delete_all_posts()
|
||||
|
||||
|
||||
def run_medias():
|
||||
def run_medias(skip_confirmation: bool = False):
|
||||
require_confirmation("delete all posts with media", skip_confirmation)
|
||||
delete_posts_with_medias()
|
||||
|
||||
|
||||
def run_likes():
|
||||
def run_likes(skip_confirmation: bool = False):
|
||||
require_confirmation("undo all likes", skip_confirmation)
|
||||
undo_likes()
|
||||
|
||||
|
||||
def run_reposts():
|
||||
def run_reposts(skip_confirmation: bool = False):
|
||||
require_confirmation("undo all reposts", skip_confirmation)
|
||||
undo_reposts()
|
||||
|
||||
|
||||
def run_quotes():
|
||||
def run_quotes(skip_confirmation: bool = False):
|
||||
require_confirmation("delete all quote posts", skip_confirmation)
|
||||
delete_quotes_posts()
|
||||
|
||||
|
||||
def run_follows():
|
||||
def run_follows(skip_confirmation: bool = False):
|
||||
require_confirmation("unfollow all accounts", skip_confirmation)
|
||||
unfollow_all()
|
||||
|
||||
|
||||
def run_bookmarks():
|
||||
def run_bookmarks(skip_confirmation: bool = False):
|
||||
require_confirmation("delete all bookmarks", skip_confirmation)
|
||||
delete_bookmarks()
|
||||
|
||||
|
||||
def run_all():
|
||||
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)
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user