Alte Skeets löschen
Ich bin ein Freund davon, meine Posts auf Social Media nicht zu alt werden zu lassen. Telegram und Mastodon, bieten das mit eigenen Werkzeugen an. Für Bluesky gibt es onlinedienste, wie den Skysweeper oder Repocleaner. Eine richtig saubere Löschung machen die aber nicht.
Insbesondere dann nicht, wen Threads weggeräumt werden sollen.
Deswegen suchte ich nach Lösungen, möglichst mit PHP, die das Ganze erledigten. Ich bin kein wirklich guter Programmierer und nehme gerne das, was andere entwickelt haben.
Mit PHP fand ich nichts, was ich verstanden hätte, dafür aber einige python-Scripts.
Python habe ich auf dem heimischen Windows-Rechner am Laufen und so sprach alles für einen Versuch.
Für beide Scripts ist es nötig, dass ihr für Euren Account ein App-Passwort generiert habt. Deleteskee ist der ältere. Skycleaner ist der Fork.
Beide laufen nicht ohne Modifikation, da sie den gleichen Fehler bei der Behandlung der Daten aufweisen.
Die bei Github bereitgestellten Requirements-Dateien müsst ihr in Eure python-Umgebung einlesen mit: pip install -r requirements.txt
Zusätzlich: pip install pytz
In der Praxis hat sich gezeigt, dass es mit deleteskee gerade bei den Threads am besten läuft. Die Ausgabe ist halt gewöhnungsbedürftig. Aber das Ergebnis zählt am Meisten. 😄
Konfigurationsdatei Config.json
Eine config.json, die für beide Skripte funktioniert sieht so aus:
{
"username": "[Dein Handle].bsky.social",
"password": "Dein App-Passwort",
"days_to_keep": {
"posts": 180,
"reposts": 30,
"boosts": 30
}
}
Korrigiertes Deleteskee-Script (deleteskee.py)
from atproto import Client, AtUri
from datetime import datetime, timedelta
from pathlib import Path
import json
import os
import pytz
from datetime import datetime, timezone
class Config():
def __init__(self):
self.days_to_keep = {"posts": 14, "boosts": 7}
cfgfile = Path.cwd() / "config.json"
if cfgfile.exists():
with cfgfile.open() as handle:
cfg = json.load(handle)
self.username = cfg.get("username", None)
self.password = cfg.get("password", None)
if "days_to_keep" in cfg:
self.days_to_keep = cfg["days_to_keep"]
else:
self.username = os.environ["DELETESKEE_USERNAME"]
self.password = os.environ["DELETESKEE_PASSWORD"]
if "DELETESKEE_KEEP_POSTS" in os.environ:
self.days_to_keep["posts"] = int(os.environ["DELETESKEE_KEEP_POSTS"])
if "DELETESKEE_KEEP_BOOSTS" in os.environ:
self.days_to_keep["boosts"] = int(os.envioron["DELETESKEE_KEEP_BOOSTS"])
config = Config()
cli = Client()
profile = cli.login(config.username, config.password)
def paginated_list_records(cli, repo, collection):
params = {
"repo": repo,
"collection": collection,
"limit": 100,
}
records = []
while True:
resp = cli.com.atproto.repo.list_records(params)
records.extend(resp.records)
if resp.cursor:
params["cursor"] = resp.cursor
else:
break
return records
now_aware = datetime.now(timezone.utc)
now = datetime.now()
post_delta = timedelta(days=config.days_to_keep["posts"])
post_check = now_aware - post_delta
boost_delta = timedelta(days=config.days_to_keep["boosts"])
boost_check = now_aware - boost_delta
records = {}
for collection in ["app.bsky.feed.post", "app.bsky.feed.repost"]:
records[collection] = paginated_list_records(cli, config.username, collection)
print(f"{collection}: {len(records[collection])}")
deletes = []
for collection, posts in records.items():
if collection == "app.bsky.feed.post":
check_date = post_check
elif collection == "app.bsky.feed.repost":
check_date = boost_check
for post in reversed(posts):
postdate = post.value.created_at
if postdate.endswith("Z"):
postdate = datetime.fromisoformat(post.value.created_at[:-1])
else:
postdate = datetime.fromisoformat(post.value.created_at)
if postdate.astimezone(timezone.utc) < check_date.astimezone(timezone.utc):
uri = AtUri.from_str(post.uri)
deletes.append({
"$type": "com.atproto.repo.applyWrites#delete",
"rkey": uri.rkey,
"collection": collection,
})
else:
break
print("going to delete", len(deletes), "posts/reposts")
if len(deletes) > 0:
for i in range(0, len(deletes), 200):
print(cli.com.atproto.repo.apply_writes({"repo": config.username, "writes": deletes[i:i+200]}))
for collection in ["app.bsky.feed.post", "app.bsky.feed.repost"]:
records[collection] = paginated_list_records(cli, config.username, collection)
print(f"{collection}: {len(records[collection])}")
Korrigiertes Skycleaner Script (cleaner.py)
from atproto import Client, AtUri
from datetime import datetime, timedelta, timezone
from pathlib import Path
import json
import pytz
class Config():
def __init__(self):
self.days_to_keep = {"posts": 0, "reposts": 0}
cfgfile = Path.cwd() / "config.json"
if cfgfile.exists():
with cfgfile.open() as handle:
cfg = json.load(handle)
self.username = cfg.get("username", None)
self.password = cfg.get("password", None)
self.days_to_keep = cfg.get("days_to_keep", None)
config = Config()
cli = Client()
profile = cli.login(config.username, config.password)
def paginated_list_records(cli, repo, collection):
params = {
"repo": repo,
"collection": collection,
"limit": 100,
}
records = []
while True:
resp = cli.com.atproto.repo.list_records(params)
records.extend(resp.records)
if resp.cursor:
params["cursor"] = resp.cursor
else:
break
return records
now = datetime.now(timezone.utc)
post_delta = timedelta(days=config.days_to_keep["posts"])
post_hold_datetime = now - post_delta
repost_delta = timedelta(days=config.days_to_keep["reposts"])
repost_hold_datetime = now - repost_delta
records = {}
for collection in ["app.bsky.feed.post", "app.bsky.feed.repost"]:
records[collection] = paginated_list_records(cli, config.username, collection)
print(f"{collection}: {len(records[collection])}")
deletes = []
for collection, posts in records.items():
if collection == "app.bsky.feed.post":
hold_datetime = post_hold_datetime
elif collection == "app.bsky.feed.repost":
hold_datetime = repost_hold_datetime
else:
break
for post in reversed(posts):
# remove charactors on `created_at` behined of `Z`
# z_index_in_created_at = post.value.created_at.index('Z')
# post_created_at = datetime.fromisoformat(post.value.created_at[:z_index_in_created_at+1])
post_created_at = post.value.created_at
if post_created_at.endswith("Z"):
post_created_at = datetime.fromisoformat(post.value.created_at[:-1])
else:
post_created_at = datetime.fromisoformat(post.value.created_at)
if post_created_at.astimezone(timezone.utc) <= hold_datetime.astimezone(timezone.utc):
uri = AtUri.from_str(post.uri)
deletes.append({
"$type": "com.atproto.repo.applyWrites#delete",
"rkey": uri.rkey,
"collection": collection,
})
else:
pass
print(f'{datetime.now()} COMMENCE DELETE: {len(deletes)} posts/reposts')
if len(deletes) > 0:
for i in range(0, len(deletes), 200):
cli.com.atproto.repo.apply_writes({"repo": config.username, "writes": deletes[i:i+200]})
print(f'{datetime.now()} DELETE COMPLETED')
Aufruf : python deleteskee.py
Ergebnis:
app.bsky.feed.post: 12
app.bsky.feed.repost: 10
going to delete 0 posts/reposts
app.bsky.feed.post: 12
app.bsky.feed.repost: 10
Ich hoffe, dass dieser Artikel dem Einen oder der Anderen hilft.
Kommentare
Kommentar veröffentlichen