Something went wrong. Try again.
Something went wrong. Try again.
6.3 kB · 148 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149#!/usr/bin/env python3"""Rewrites the references between Mediatracker records from strongRefs to AT-URI strings (one-off).
Usage: migrate_strongrefs.py --handle you.example.com --backup mediatracker-export-YYYY-MM-DD.zip \ [--pds https://pds.example.com] [--dry-run]
Before the change a record referred to its item with a `com.atproto.repo.strongRef` (`{ "uri": ..., "cid": ... }`);now the field holds the AT-URI string. This rewrites, under its own record key and with nothing else changed:
- `subject` of movie.tracked, movie.watch, movie.rating, tvShow.tracked, tvShow.watch, tvShow.rating and tvShow.episode.rating,- `show` of tvShow.episode.item.
Records that hold the string already are left alone, so it can be run again at any time (after an interruption, or tocheck that nothing is left). A PDS that limits writes is waited out until the time it reports.
NOTHING IS WRITTEN WITHOUT A BACKUP: `--backup` is the ZIP of the app's export (Settings -> Account -> Export, orGET /tracking/export). It must belong to this account and hold at least as many records per collection as the PDSdoes now. A dry run (`--dry-run`) needs none and writes nothing.
The app password is read from the ATPROTO_APP_PASSWORD environment variable, or asked for. Afterwards use "Read dataagain" in the app's settings, so the record index reads the new records."""import argparseimport getpassimport jsonimport osimport sysimport zipfile
from import_to_pds import ( NAMESPACE, MOVIE_TRACKED, MOVIE_RATING, MOVIE_WATCH, SHOW_TRACKED, SHOW_RATING, SHOW_WATCH, EPISODE_ITEM, EPISODE_RATING, MAX_WRITES_PER_CALL, Pds, XrpcError, rkey_of, _chunks,)
# collection -> the field that refers to another recordREFERENCE_FIELDS = { MOVIE_TRACKED: "subject", MOVIE_WATCH: "subject", MOVIE_RATING: "subject", SHOW_TRACKED: "subject", SHOW_WATCH: "subject", SHOW_RATING: "subject", EPISODE_ITEM: "show", EPISODE_RATING: "subject",}
def check_backup(path, did, counts): """Stops unless `path` is an export of this account that holds every record the PDS has.""" try: with zipfile.ZipFile(path) as archive: manifest = json.loads(archive.read("manifest.json")) except (OSError, KeyError, ValueError, zipfile.BadZipFile) as error: sys.exit(f"{path} is not an export of the app (no readable manifest.json): {error}")
if manifest.get("did") != did: sys.exit(f"The backup is of {manifest.get('did')}, not of {did}.")
saved = {entry["collection"]: entry["recordCount"] for entry in manifest.get("collections", [])} for collection, count in counts.items(): if saved.get(collection, 0) < count: sys.exit(f"The backup holds {saved.get(collection, 0)} records of {collection}, the PDS {count}. " "Export again (after \"Read data again\") and use that file.")
print(f"Backup {path} of {manifest.get('exportedAt')} covers all {sum(counts.values())} records.")
def migrate(pds, dry_run, backup): pending = {} unchanged = failed = 0 counts = {}
for collection, field in REFERENCE_FIELDS.items(): records = pds.list_records(collection) counts[collection] = len(records) for record in records: reference = record["value"].get(field) if isinstance(reference, str): unchanged += 1 elif isinstance(reference, dict) and isinstance(reference.get("uri"), str): pending.setdefault(collection, []).append( (rkey_of(record["uri"]), {**record["value"], field: reference["uri"]})) else: print(f" cannot rewrite {record['uri']}: no readable {field}") failed += 1
total = sum(len(records) for records in pending.values()) print(f"{sum(counts.values())} records read: {total} to rewrite, {unchanged} in the new shape already, " f"{failed} that can't be rewritten.")
if dry_run or total == 0: return failed
check_backup(backup, pds.did, counts)
written = 0 for collection, records in pending.items(): short = collection.removeprefix(NAMESPACE + ".") for chunk in _chunks(records, MAX_WRITES_PER_CALL): try: pds.apply_updates([(collection, rkey, value) for rkey, value in chunk]) except XrpcError as error: print(f" {len(chunk)} records of {short} failed: {error}") failed += len(chunk) continue written += len(chunk) print(f" {written}/{total} written ({short})")
print(f"Done: {written} rewritten, {unchanged} unchanged, {failed} failed.")
return failed
def main(): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--handle", required=True, help="handle (or DID) of the account") parser.add_argument("--pds", default="https://bsky.social", help="URL of the account's PDS") parser.add_argument("--backup", help="ZIP of the app's export; required to write anything") parser.add_argument("--dry-run", action="store_true", help="only read the PDS and say what would be written") arguments = parser.parse_args()
if not arguments.dry_run and not arguments.backup: sys.exit("Export your data first (Settings -> Account -> Export) and pass the ZIP as --backup. " "Nothing is written without it.")
try: password = os.environ.get("ATPROTO_APP_PASSWORD") or getpass.getpass("App password: ") pds = Pds(arguments.pds, arguments.handle, password) print(f"Logged in as {pds.did}") failed = migrate(pds, arguments.dry_run, arguments.backup) except XrpcError as error: sys.exit(f"The PDS answered with an error: {error}") except KeyboardInterrupt: sys.exit("\nStopped. Run it again to carry on - what is rewritten already is skipped.")
if arguments.dry_run: print("Dry run: nothing was written.")
if failed: sys.exit(1)
print("Afterwards use \"Read data again\" in the app's settings, so the index reads the new records.")
if __name__ == "__main__": main()