Kharec/skywipe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: use our new logger

Browse Source
This commit is contained in:
Sandro CURY CAZZANIGA
2025-12-20 15:46:33 +01:00
parent 58ab6cfafa
commit 06d70edaf1
10 changed files with 122 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
skywipe/bookmarks.py
View File

@@ -4,9 +4,11 @@ import time
from atproto import models
from .auth import Auth
from .configure import Configuration
from .logger import get_logger, ProgressTracker
def delete_bookmarks():
logger = get_logger()
auth = Auth()
client = auth.login()
config = Configuration()
@@ -14,16 +16,17 @@ def delete_bookmarks():
batch_size = config_data.get("batch_size", 10)
delay = config_data.get("delay", 1)
verbose = config_data.get("verbose", False)
if verbose:
print(
f"Starting bookmark deletion with batch_size={batch_size}, delay={delay}s")
logger.info(
f"Starting bookmark deletion with batch_size={batch_size}, delay={delay}s")
cursor = None
total_deleted = 0
batch_num = 0
progress = ProgressTracker(operation="Deleting bookmarks")
while True:
batch_num += 1
get_params = models.AppBskyBookmarkGetBookmarks.Params(
limit=batch_size,
cursor=cursor
@@ -35,6 +38,8 @@ def delete_bookmarks():
if not bookmarks:
break
progress.batch(batch_num, len(bookmarks))
for bookmark in bookmarks:
try:
bookmark_uri = None
@@ -49,8 +54,7 @@ def delete_bookmarks():
break
if not bookmark_uri:
if verbose:
print(f"Skipping bookmark: unable to find uri")
logger.debug("Skipping bookmark: unable to find uri")
continue
delete_data = models.AppBskyBookmarkDeleteBookmark.Data(
@@ -58,12 +62,11 @@ def delete_bookmarks():
)
client.app.bsky.bookmark.delete_bookmark(data=delete_data)
total_deleted += 1
if verbose:
print(f"Deleted bookmark: {bookmark_uri}")
progress.update(1)
logger.debug(f"Deleted bookmark: {bookmark_uri}")
except Exception as e:
bookmark_uri = getattr(bookmark, "uri", "unknown")
if verbose:
print(f"Error deleting bookmark {bookmark_uri}: {e}")
logger.error(f"Error deleting bookmark {bookmark_uri}: {e}")
cursor = response.cursor
if not cursor:
@@ -72,4 +75,4 @@ def delete_bookmarks():
if delay > 0:
time.sleep(delay)
print(f"Deleted {total_deleted} bookmarks.")
logger.info(f"Deleted {total_deleted} bookmarks.")

21
skywipe/cli.py
View File

@@ -2,9 +2,11 @@
import sys
import argparse
from pathlib import Path
from .commands import registry
from .configure import Configuration
from .logger import setup_logger
def create_parser():
@@ -31,8 +33,9 @@ def create_parser():
def require_config():
config = Configuration()
if not config.exists():
print("Error: Configuration file not found.")
print("You must run 'skywipe configure' first.")
logger = setup_logger(verbose=False)
logger.error("Configuration file not found.")
logger.error("You must run 'skywipe configure' first.")
sys.exit(1)
@@ -42,11 +45,23 @@ def main():
if registry.requires_config(args.command):
require_config()
config = Configuration()
config_data = config.load()
verbose = config_data.get("verbose", False)
log_file = Path.home() / ".config" / "skywipe" / "skywipe.log"
setup_logger(verbose=verbose, log_file=log_file)
else:
setup_logger(verbose=False)
try:
registry.execute(args.command)
except ValueError as e:
print(f"Error: {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)

8
skywipe/commands.py
View File

@@ -9,6 +9,7 @@ from .reposts import undo_reposts
from .quotes import delete_quotes_posts
from .follows import unfollow_all
from .bookmarks import delete_bookmarks
from .logger import get_logger
CommandHandler = Callable[[], None]
@@ -88,14 +89,19 @@ def run_bookmarks():
def run_all():
logger = get_logger()
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)
logger.info(f"Completed command: {cmd}")
except Exception as e:
print(f"Error running '{cmd}': {e}")
logger.error(f"Error running '{cmd}': {e}", exc_info=True)
continue
logger.info("All commands completed.")
registry.register("configure", run_configure,

7
skywipe/configure.py
View File

@@ -3,6 +3,7 @@
import getpass
from pathlib import Path
import yaml
from .logger import setup_logger, get_logger
class Configuration:
@@ -13,10 +14,12 @@ class Configuration:
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
@@ -37,7 +40,7 @@ class Configuration:
batch_size = int(batch_size)
delay = int(delay)
except ValueError:
print("Error: batch_size and delay must be integers")
logger.error("batch_size and delay must be integers")
return
config_data = {
@@ -51,7 +54,7 @@ class Configuration:
with open(self.config_file, "w") as f:
yaml.dump(config_data, f, default_flow_style=False)
print(f"\nConfiguration saved to {self.config_file}")
logger.info(f"Configuration saved to {self.config_file}")
def load(self) -> dict:
if not self.exists():

22
skywipe/follows.py
View File

@@ -4,12 +4,14 @@ import time
from atproto import models
from .auth import Auth
from .configure import Configuration
from .logger import get_logger, ProgressTracker
FOLLOW_COLLECTION = "app.bsky.graph.follow"
def unfollow_all():
logger = get_logger()
auth = Auth()
client = auth.login()
config = Configuration()
@@ -17,17 +19,18 @@ def unfollow_all():
batch_size = config_data.get("batch_size", 10)
delay = config_data.get("delay", 1)
verbose = config_data.get("verbose", False)
if verbose:
print(
f"Starting unfollow operation with batch_size={batch_size}, delay={delay}s")
logger.info(
f"Starting unfollow operation with batch_size={batch_size}, delay={delay}s")
did = client.me.did
cursor = None
total_unfollowed = 0
batch_num = 0
progress = ProgressTracker(operation="Unfollowing accounts")
while True:
batch_num += 1
list_params = models.ComAtprotoRepoListRecords.Params(
repo=did,
collection=FOLLOW_COLLECTION,
@@ -41,6 +44,8 @@ def unfollow_all():
if not records:
break
progress.batch(batch_num, len(records))
for record in records:
try:
record_uri = record.uri
@@ -52,12 +57,11 @@ def unfollow_all():
}
client.com.atproto.repo.delete_record(data=delete_data)
total_unfollowed += 1
if verbose:
print(f"Unfollowed: {record_uri}")
progress.update(1)
logger.debug(f"Unfollowed: {record_uri}")
except Exception as e:
record_uri = getattr(record, "uri", "unknown")
if verbose:
print(f"Error unfollowing {record_uri}: {e}")
logger.error(f"Error unfollowing {record_uri}: {e}")
cursor = response.cursor
if not cursor:
@@ -66,4 +70,4 @@ def unfollow_all():
if delay > 0:
time.sleep(delay)
print(f"Unfollowed {total_unfollowed} accounts.")
logger.info(f"Unfollowed {total_unfollowed} accounts.")

22
skywipe/likes.py
View File

@@ -4,12 +4,14 @@ import time
from atproto import models
from .auth import Auth
from .configure import Configuration
from .logger import get_logger, ProgressTracker
LIKE_COLLECTION = "app.bsky.feed.like"
def undo_likes():
logger = get_logger()
auth = Auth()
client = auth.login()
config = Configuration()
@@ -17,17 +19,18 @@ def undo_likes():
batch_size = config_data.get("batch_size", 10)
delay = config_data.get("delay", 1)
verbose = config_data.get("verbose", False)
if verbose:
print(
f"Starting like deletion with batch_size={batch_size}, delay={delay}s")
logger.info(
f"Starting like deletion with batch_size={batch_size}, delay={delay}s")
did = client.me.did
cursor = None
total_undone = 0
batch_num = 0
progress = ProgressTracker(operation="Undoing likes")
while True:
batch_num += 1
list_params = models.ComAtprotoRepoListRecords.Params(
repo=did,
collection=LIKE_COLLECTION,
@@ -41,6 +44,8 @@ def undo_likes():
if not records:
break
progress.batch(batch_num, len(records))
for record in records:
try:
record_uri = record.uri
@@ -52,12 +57,11 @@ def undo_likes():
}
client.com.atproto.repo.delete_record(data=delete_data)
total_undone += 1
if verbose:
print(f"Undone like: {record_uri}")
progress.update(1)
logger.debug(f"Undone like: {record_uri}")
except Exception as e:
record_uri = getattr(record, "uri", "unknown")
if verbose:
print(f"Error undoing like {record_uri}: {e}")
logger.error(f"Error undoing like {record_uri}: {e}")
cursor = response.cursor
if not cursor:
@@ -66,4 +70,4 @@ def undo_likes():
if delay > 0:
time.sleep(delay)
print(f"Undone {total_undone} likes.")
logger.info(f"Undone {total_undone} likes.")

25
skywipe/medias.py
View File

@@ -3,9 +3,11 @@
import time
from .auth import Auth
from .configure import Configuration
from .logger import get_logger, ProgressTracker
def delete_posts_with_medias():
logger = get_logger()
auth = Auth()
client = auth.login()
config = Configuration()
@@ -13,17 +15,18 @@ def delete_posts_with_medias():
batch_size = config_data.get("batch_size", 10)
delay = config_data.get("delay", 1)
verbose = config_data.get("verbose", False)
if verbose:
print(
f"Starting media post deletion with batch_size={batch_size}, delay={delay}s")
logger.info(
f"Starting media post deletion with batch_size={batch_size}, delay={delay}s")
did = client.me.did
cursor = None
total_deleted = 0
batch_num = 0
progress = ProgressTracker(operation="Deleting posts with media")
while True:
batch_num += 1
response = client.get_author_feed(
actor=did, limit=batch_size, cursor=cursor)
@@ -31,6 +34,8 @@ def delete_posts_with_medias():
if not posts:
break
progress.batch(batch_num, len(posts))
for post in posts:
post_record = post.post
@@ -69,18 +74,16 @@ def delete_posts_with_medias():
break
if not has_media:
if verbose:
print(f"Skipping post without media: {post_record.uri}")
logger.debug(f"Skipping post without media: {post_record.uri}")
continue
try:
client.delete_post(post_record.uri)
total_deleted += 1
if verbose:
print(f"Deleted post with media: {post_record.uri}")
progress.update(1)
logger.debug(f"Deleted post with media: {post_record.uri}")
except Exception as e:
if verbose:
print(f"Error deleting post {post_record.uri}: {e}")
logger.error(f"Error deleting post {post_record.uri}: {e}")
cursor = response.cursor
if not cursor:
@@ -89,4 +92,4 @@ def delete_posts_with_medias():
if delay > 0:
time.sleep(delay)
print(f"Deleted {total_deleted} posts with media.")
logger.info(f"Deleted {total_deleted} posts with media.")

21
skywipe/posts.py
View File

@@ -3,9 +3,11 @@
import time
from .auth import Auth
from .configure import Configuration
from .logger import get_logger, ProgressTracker
def delete_all_posts():
logger = get_logger()
auth = Auth()
client = auth.login()
config = Configuration()
@@ -13,17 +15,18 @@ def delete_all_posts():
batch_size = config_data.get("batch_size", 10)
delay = config_data.get("delay", 1)
verbose = config_data.get("verbose", False)
if verbose:
print(
f"Starting post deletion with batch_size={batch_size}, delay={delay}s")
logger.info(
f"Starting post deletion with batch_size={batch_size}, delay={delay}s")
did = client.me.did
cursor = None
total_deleted = 0
batch_num = 0
progress = ProgressTracker(operation="Deleting posts")
while True:
batch_num += 1
if cursor:
response = client.get_author_feed(
actor=did, limit=batch_size, cursor=cursor)
@@ -35,16 +38,16 @@ def delete_all_posts():
break
post_uris = [post.post.uri for post in posts]
progress.batch(batch_num, len(post_uris))
for uri in post_uris:
try:
client.delete_post(uri)
total_deleted += 1
if verbose:
print(f"Deleted post: {uri}")
progress.update(1)
logger.debug(f"Deleted post: {uri}")
except Exception as e:
if verbose:
print(f"Error deleting post {uri}: {e}")
logger.error(f"Error deleting post {uri}: {e}")
cursor = response.cursor
if not cursor:
@@ -53,4 +56,4 @@ def delete_all_posts():
if delay > 0:
time.sleep(delay)
print(f"Deleted {total_deleted} posts.")
logger.info(f"Deleted {total_deleted} posts.")

24
skywipe/quotes.py
View File

@@ -3,9 +3,11 @@
import time
from .auth import Auth
from .configure import Configuration
from .logger import get_logger, ProgressTracker
def delete_quotes_posts():
logger = get_logger()
auth = Auth()
client = auth.login()
config = Configuration()
@@ -13,17 +15,17 @@ def delete_quotes_posts():
batch_size = config_data.get("batch_size", 10)
delay = config_data.get("delay", 1)
verbose = config_data.get("verbose", False)
if verbose:
print(
f"Starting quote post deletion with batch_size={batch_size}, delay={delay}s")
logger.info(f"Starting quote post deletion with batch_size={batch_size}, delay={delay}s")
did = client.me.did
cursor = None
total_deleted = 0
batch_num = 0
progress = ProgressTracker(operation="Deleting quote posts")
while True:
batch_num += 1
response = client.get_author_feed(
actor=did, limit=batch_size, cursor=cursor)
@@ -31,6 +33,8 @@ def delete_quotes_posts():
if not posts:
break
progress.batch(batch_num, len(posts))
for post in posts:
post_record = post.post
@@ -52,18 +56,16 @@ def delete_quotes_posts():
has_quote = True
if not has_quote:
if verbose:
print(f"Skipping post without quote: {post_record.uri}")
logger.debug(f"Skipping post without quote: {post_record.uri}")
continue
try:
client.delete_post(post_record.uri)
total_deleted += 1
if verbose:
print(f"Deleted quote post: {post_record.uri}")
progress.update(1)
logger.debug(f"Deleted quote post: {post_record.uri}")
except Exception as e:
if verbose:
print(f"Error deleting quote post {post_record.uri}: {e}")
logger.error(f"Error deleting quote post {post_record.uri}: {e}")
cursor = response.cursor
if not cursor:
@@ -72,4 +74,4 @@ def delete_quotes_posts():
if delay > 0:
time.sleep(delay)
print(f"Deleted {total_deleted} quote posts.")
logger.info(f"Deleted {total_deleted} quote posts.")

22
skywipe/reposts.py
View File

@@ -4,12 +4,14 @@ import time
from atproto import models
from .auth import Auth
from .configure import Configuration
from .logger import get_logger, ProgressTracker
REPOST_COLLECTION = "app.bsky.feed.repost"
def undo_reposts():
logger = get_logger()
auth = Auth()
client = auth.login()
config = Configuration()
@@ -17,17 +19,18 @@ def undo_reposts():
batch_size = config_data.get("batch_size", 10)
delay = config_data.get("delay", 1)
verbose = config_data.get("verbose", False)
if verbose:
print(
f"Starting repost deletion with batch_size={batch_size}, delay={delay}s")
logger.info(
f"Starting repost deletion with batch_size={batch_size}, delay={delay}s")
did = client.me.did
cursor = None
total_undone = 0
batch_num = 0
progress = ProgressTracker(operation="Undoing reposts")
while True:
batch_num += 1
list_params = models.ComAtprotoRepoListRecords.Params(
repo=did,
collection=REPOST_COLLECTION,
@@ -41,6 +44,8 @@ def undo_reposts():
if not records:
break
progress.batch(batch_num, len(records))
for record in records:
try:
record_uri = record.uri
@@ -52,12 +57,11 @@ def undo_reposts():
}
client.com.atproto.repo.delete_record(data=delete_data)
total_undone += 1
if verbose:
print(f"Undone repost: {record_uri}")
progress.update(1)
logger.debug(f"Undone repost: {record_uri}")
except Exception as e:
record_uri = getattr(record, "uri", "unknown")
if verbose:
print(f"Error undoing repost {record_uri}: {e}")
logger.error(f"Error undoing repost {record_uri}: {e}")
cursor = response.cursor
if not cursor:
@@ -66,4 +70,4 @@ def undo_reposts():
if delay > 0:
time.sleep(delay)
print(f"Undone {total_undone} reposts.")
logger.info(f"Undone {total_undone} reposts.")