Something went wrong. Try again.
🌉 A bridge between decentralized social networks fed.brid.gy
Something went wrong. Try again.
6.0 kB · 172 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173"""Convert all stored Activity entities to Objects.
https://github.com/snarfed/bridgy-fed/issues/286
Run from repo top level directory:
source local/bin/activate.cshenv PYTHONPATH=. GOOGLE_APPLICATION_CREDENTIALS=service_account_creds.json \ python scripts/migrate_activity_to_object.py"""from datetime import datetimeimport jsonimport sys
import dateutil.parserfrom google.cloud import ndbfrom granary import as1from oauth_dropins.webutil import appengine_config, util
import commonfrom models import Activity, Object, Target
seen = {}latest_updated = datetime(1900, 1, 1)
with open('seen.json') as f: # maps object id to updated timestamp, or None seen = {k: datetime.fromisoformat(v) for k, v in json.load(f).items()} latest_updated = seen['_latest_updated']
def run(): global latest_updated
query = Activity.query().order(Activity.key) if len(sys.argv) > 1: print(f'Starting at {sys.argv[1]}') query = query.filter(Activity.key >= ndb.Key(Activity, sys.argv[1])) elif latest_updated: print(f'Starting at {latest_updated}') query = list(Activity.query(Activity.updated > latest_updated)) query.sort(key=lambda a: a.key) # print(query.filter(Activity.updated > latest_updated).count()) # print(len(query)) # sys.exit() else: print('Starting at the beginning')
id = obj = None obj = None num_activities = count = 0
for a in query: if a.source() != id: # finished the current Object if obj: print(f'{num_activities} total', flush=True) obj.status = ('in progress' if obj.undelivered else 'failed' if obj.failed else 'complete' if obj.delivered else 'new') print(f' Storing object', flush=True) obj.put() seen[obj.key.id()] = obj.updated
for field in 'actor', 'object': inner = obj_as1.get(field) if isinstance(inner, dict) and inner.get('id'): id = inner['id'] updated = inner.get('updated') if updated: updated = dateutil.parser.parse(updated) published = inner.get('published') if published: published = dateutil.parser.parse(published)
inner_obj = Object( id=id, source_protocol=obj.source_protocol, as1=json.dumps(inner), type=as1.object_type(inner), created=(published or updated or obj.created ).replace(tzinfo=None), updated=(updated or published or obj.updated ).replace(tzinfo=None), ) if id not in seen or inner_obj.updated > seen[id]: print(f' Storing inner {field} {id}') inner_obj.put() seen[id] = inner_obj.updated
count += 1
id = a.source() if id == 'UI': id = json.loads(a.source_as2)['id']
# start a new Object num_activities = 0 print(f'Collecting {id} ..', end='', flush=True) assert util.is_web(id)
obj_as1 = a.to_as1() type = as1.object_type(obj_as1)
labels = [] if obj_as1.get('objectType') == 'activity': labels.append('activity') if a.direction == 'out': labels.append('user') elif a.domain: if type in ('like', 'share', 'follow'): labels.append('notification') elif type in ('note', 'article', 'post'): labels.append('feed')
obj = Object( id=id, domains=a.domain, source_protocol=('ui' if a.source() == 'UI' else 'webmention' if a.direction == 'out' else a.protocol), labels=labels, as1=json.dumps(obj_as1), # bsky=None, as2=a.source_as2, mf2=a.source_mf2, type=type, # deleted=None, delivered=[], undelivered=[], failed=[], created=a.created, updated=a.updated, )
# add this Activity to current Object status = a.status if a.protocol == 'ostatus': # only 26 'complete' ostatus Activitys, all with different source URLs obj.status = 'ignored' if a.status == 'error' else a.status elif status != 'ignored': dest = (obj.delivered if a.status == 'complete' else obj.failed if a.status == 'error' else obj.undelivered) dest.append(Target(uri=a.target(), protocol='activitypub'))
if a.created < obj.created: obj.created = a.created if a.updated > obj.updated: obj.updated = a.updated
if a.updated > latest_updated: latest_updated = a.updated
# if count == 20: # break
num_activities += 1 print('.', end='', flush=True)
with appengine_config.ndb_client.context(): try: run() finally: print(f'\n\nLatest updated: {latest_updated}', flush=True) if seen: seen['_latest_updated'] = latest_updated with open('seen.json', 'w') as f: json.dump({id: dt.isoformat() for id, dt in seen.items()}, f, indent=2)