Progress bar!

Signed-off-by: Markus Birth <markus@birth-online.de>
This commit is contained in:
2026-05-12 01:15:59 +01:00
parent 9bc3d916f2
commit a96b2cc007
+17 -6
View File
@@ -1,7 +1,9 @@
import configparser
import requests
from collections.abc import Callable
from atproto import Client, client_utils
from rich import print
from rich.progress import Progress, ProgressColumn, TextColumn, BarColumn, MofNCompleteColumn, TimeRemainingColumn
from time import sleep
class BlueBackBlocker():
@@ -13,7 +15,7 @@ class BlueBackBlocker():
def get_lists(self):
return self.agent.app.bsky.graph.get_lists({'actor': self.did})
def get_list(self, list_id: str):
def get_list(self, list_id: str, progress: Callable[[int, int], None] = None):
list_uri = f"at://{self.did}/app.bsky.graph.list/{list_id}"
cursor = None
members = []
@@ -21,8 +23,10 @@ class BlueBackBlocker():
r = self.agent.app.bsky.graph.get_list({"list": list_uri, "cursor": cursor})
# returns keys "items", "list", and "cursor"
members += r["items"]
if progress:
progress(len(members), r["list"]["list_item_count"]-1)
#print(r)
sleep(0.20)
sleep(0.20) # Poor man's way of making sure to not go over 5 requests per second
if not r["cursor"]:
break
cursor = r["cursor"]
@@ -78,10 +82,17 @@ def main():
blocker = BlueBackBlocker(config["BlueSky"]["Username"], config["BlueSky"]["Password"])
print(f"Logged into BlueSky, display name: [bold blue]{blocker.profile.display_name}[/bold blue], ", end="")
print(f"DID: [bold yellow]{blocker.did}")
print(f"Loading list with id [bold white]{config['BlueSky']['ListId']}")
blocklist = blocker.get_list(config["BlueSky"]["ListId"])
#print(blocklist)
print(f"Found list: [bold green]{blocklist['list']['name']}[/bold green] with [bold yellow]{len(blocklist['items'])}[/bold yellow] entries.")
with Progress(
TextColumn("[progress.description]{task.description}"),
BarColumn(),
MofNCompleteColumn(),
TimeRemainingColumn()
) as prog:
task_bsky_list = prog.add_task(f"Loading list with id [bold white]{config['BlueSky']['ListId']}")
blocklist = blocker.get_list(config["BlueSky"]["ListId"], lambda cur, tot: prog.update(task_bsky_list, completed=cur, total=tot))
#print(blocklist)
print(f"Found list: [bold green]{blocklist['list']['name']}[/bold green] with [bold yellow]{len(blocklist['items'])}[/bold yellow] entries.")
"""
cs = ClearSky(blocker.did)