commit 31a048eee2609ca6de43733088c24d2b2da90fe9 · zenfyr.dev/xpost

+3 -3

bluesky/common.py

···

       92
       92
        
               self.post = post

     

       93
       93
        
               self.tokens = tokens

     

       94
       94
        
               

     

       95
       95
       -
               self.id = json.dumps(self.post['$xpost.strongRef'], sort_keys=True)

     

       95
       95
       +
               self.id = post['$xpost.strongRef']['uri']

     

       96
       96
        
               

     

       97
       97
        
               self.parent_id = None

     

       98
       98
        
               if self.post.get('reply'):

     

       99
       99
       -
                   self.parent_id = json.dumps(self.post['reply']['parent'], sort_keys=True)

     

       99
       99
       +
                   self.parent_id = self.post['reply']['parent']['uri']

     

       100
       100
        
               

     

       101
       101
        
               labels = self.post.get('labels', {}).get('values')

     

       102
       102
        
               self.cw = ''

     
···

       132
       132
        
               return "text/plain"

     

       133
       133
        
       

     

       134
       134
        
           def get_post_url(self) -> str | None:

     

       135
       135
       -
               at_uri: str = self.post['$xpost.strongRef']['uri'][len("at://"):]

     

       135
       135
       +
               at_uri: str = self.post['$xpost.uri'][len("at://"):]

     

       136
       136
        
               

     

       137
       137
        
               parts = at_uri.split("/")

     

       138
       138
        
               did, _, post_id = parts

+72 -107

bluesky/input.py

···

       1
       1
       -
       import re, json

     

       1
       1
       +
       import re, json, websockets, asyncio

     

       2
       2
        
       

     

       3
       3
       -
       from atproto import AsyncFirehoseSubscribeReposClient, CAR

     

       4
       3
        
       from atproto_client import models

     

       5
       4
        
       from atproto_client.models.utils import get_or_create as get_model_or_create

     

       6
       6
       -
       from atproto_firehose import models as firehose_models, parse_subscribe_repos_message as parse_firehose

     

       7
       5
        
       from bluesky.atproto2 import resolve_identity

     

       8
       6
        
       

     

       9
       7
        
       from bluesky.common import BlueskyPost, SERVICE, tokenize_post

     
···

       33
       31
        
               super().__init__(SERVICE, did, settings, db)

     

       34
       32
        
           

     

       35
       33
        
           def _on_post(self, outputs: list[cross.Output], post: dict[str, Any]):

     

       36
       36
       -
               post_ref = json.dumps(post['$xpost.strongRef'], sort_keys=True)

     

       34
       34
       +
               post_uri = post['$xpost.strongRef']['uri']

     

       35
       35
       +
               post_cid = post['$xpost.strongRef']['cid']

     

       37
       36
        
               

     

       38
       38
       -
               parent_ref = None

     

       37
       37
       +
               parent_uri = None

     

       39
       38
        
               if post.get('reply'):

     

       40
       40
       -
                   parent_ref = json.dumps(post['reply']['parent'], sort_keys=True)

     

       39
       39
       +
                   parent_uri = post['reply']['parent']['uri']

     

       41
       40
        
               

     

       42
       41
        
               embed = post.get('embed', {})

     

       43
       42
        
               if embed.get('$type') in ('app.bsky.embed.record', 'app.bsky.embed.recordWithMedia'):

     

       44
       43
        
                   did, collection, rid = str(embed['record']['uri'][len('at://'):]).split('/')

     

       45
       44
        
                   if collection == 'app.bsky.feed.post':

     

       46
       46
       -
                       LOGGER.info("Skipping '%s'! Quote..", post_ref)

     

       45
       45
       +
                       LOGGER.info("Skipping '%s'! Quote..", post_uri)

     

       47
       46
        
                       return

     

       48
       47
        
                   

     

       49
       49
       -
               success = database.try_insert_post(self.db, post_ref, parent_ref, self.user_id, self.service)

     

       48
       48
       +
               success = database.try_insert_post(self.db, post_uri, parent_uri, self.user_id, self.service)

     

       50
       49
        
               if not success:

     

       51
       51
       -
                   LOGGER.info("Skipping '%s' as parent post was not found in db!", post_ref)

     

       50
       50
       +
                   LOGGER.info("Skipping '%s' as parent post was not found in db!", post_uri)

     

       52
       51
        
                   return

     

       52
       52
       +
               database.store_data(self.db, post_uri, self.user_id, self.service, {'cid': post_cid})

     

       53
       53
        
               

     

       54
       54
        
               tokens = tokenize_post(post)

     

       55
       55
        
               if not cross.test_filters(tokens, self.options.filters):

     

       56
       56
       -
                   LOGGER.info("Skipping '%s'. Matched a filter!", post_ref)

     

       56
       56
       +
                   LOGGER.info("Skipping '%s'. Matched a filter!", post_uri)

     

       57
       57
        
                   return

     

       58
       58
        
               

     

       59
       59
       -
               LOGGER.info("Crossposting '%s'...", post_ref)

     

       59
       59
       +
               LOGGER.info("Crossposting '%s'...", post_uri)

     

       60
       60
        
               

     

       61
       61
        
               def get_blob_url(blob: str):

     

       62
       62
        
                   return f'{self.pds}/xrpc/com.atproto.sync.getBlob?did={self.user_id}&cid={blob}'

     
···

       71
       71
        
                       LOGGER.info("Downloading %s...", url)

     

       72
       72
        
                       io = download_media(url, image.alt)

     

       73
       73
        
                       if not io:

     

       74
       74
       -
                           LOGGER.error("Skipping '%s'. Failed to download media!", post_ref)

     

       74
       74
       +
                           LOGGER.error("Skipping '%s'. Failed to download media!", post_uri)

     

       75
       75
        
                           return

     

       76
       76
        
                       attachments.append(io)

     

       77
       77
        
               elif embed.get('$type') == 'app.bsky.embed.video':

     
···

       81
       81
        
                   LOGGER.info("Downloading %s...", url)

     

       82
       82
        
                   io = download_media(url, model.alt if model.alt else '')

     

       83
       83
        
                   if not io:

     

       84
       84
       -
                       LOGGER.error("Skipping '%s'. Failed to download media!", post_ref)

     

       84
       84
       +
                       LOGGER.error("Skipping '%s'. Failed to download media!", post_uri)

     

       85
       85
        
                       return

     

       86
       86
        
                   attachments.append(io)

     

       87
       87
        
                   

     
···

       89
       89
        
               for output in outputs:

     

       90
       90
        
                   output.accept_post(cross_post)

     

       91
       91
        
       

     

       92
       92
       -
           def _on_delete_post(self, outputs: list[cross.Output], post_id: dict, repost: bool):

     

       93
       93
       -
               identifier = json.dumps(post_id, sort_keys=True)

     

       94
       94
       -
               post = database.find_post(self.db, identifier, self.user_id, self.service)

     

       92
       92
       +
           def _on_delete_post(self, outputs: list[cross.Output], post_id: str, repost: bool):

     

       93
       93
       +
               post = database.find_post(self.db, post_id, self.user_id, self.service)

     

       95
       94
        
               if not post:

     

       96
       95
        
                   return

     

       97
       96
        
               

     

       98
       98
       -
               LOGGER.info("Deleting '%s'...", identifier)

     

       97
       97
       +
               LOGGER.info("Deleting '%s'...", post_id)

     

       99
       98
        
               if repost:

     

       100
       99
        
                   for output in outputs:

     

       101
       101
       -
                       output.delete_repost(identifier)

     

       100
       100
       +
                       output.delete_repost(post_id)

     

       102
       101
        
               else:

     

       103
       102
        
                   for output in outputs:

     

       104
       104
       -
                       output.delete_post(identifier)

     

       105
       105
       -
               database.delete_post(self.db, identifier, self.user_id, self.service)

     

       103
       103
       +
                       output.delete_post(post_id)

     

       104
       104
       +
               database.delete_post(self.db, post_id, self.user_id, self.service)

     

       106
       105
        
           

     

       107
       106
        
           def _on_repost(self, outputs: list[cross.Output], post: dict[str, Any]):

     

       108
       108
       -
               post_ref = json.dumps(post['$xpost.strongRef'], sort_keys=True)

     

       107
       107
       +
               post_uri = post['$xpost.strongRef']['uri']

     

       108
       108
       +
               post_cid = post['$xpost.strongRef']['cid']

     

       109
       109
        
               

     

       110
       110
       -
               reposted_ref = {

     

       111
       111
       -
                   'cid': post['subject']['cid'],

     

       112
       112
       -
                   'uri': post['subject']['uri']

     

       113
       113
       -
               }

     

       114
       114
       -
               reposted_ref = json.dumps(reposted_ref, sort_keys=True)

     

       110
       110
       +
               reposted_uri = post['subject']['uri']

     

       115
       111
        
               

     

       116
       116
       -
               success = database.try_insert_repost(self.db, post_ref, reposted_ref, self.user_id, self.service)

     

       112
       112
       +
               success = database.try_insert_repost(self.db, post_uri, reposted_uri, self.user_id, self.service)

     

       117
       113
        
               if not success:

     

       118
       118
       -
                   LOGGER.info("Skipping '%s' as reposted post was not found in db!", post_ref)

     

       114
       114
       +
                   LOGGER.info("Skipping '%s' as reposted post was not found in db!", post_uri)

     

       119
       115
        
                   return

     

       116
       116
       +
               database.store_data(self.db, post_uri, self.user_id, self.service, {'cid': post_cid})

     

       120
       117
        
               

     

       121
       121
       -
               LOGGER.info("Crossposting '%s'...", post_ref)

     

       118
       118
       +
               LOGGER.info("Crossposting '%s'...", post_uri)

     

       122
       119
        
               for output in outputs:

     

       123
       123
       -
                   output.accept_repost(post_ref, reposted_ref)

     

       124
       124
       -
               

     

       120
       120
       +
                   output.accept_repost(post_uri, reposted_uri)

     

       125
       121
        
       

     

       126
       126
       -
       class BlueskyPdsInput(BlueskyInput):

     

       122
       122
       +
       class BlueskyJetstreamInput(BlueskyInput):

     

       127
       123
        
           def __init__(self, settings: dict, db: DataBaseWorker) -> None:

     

       128
       124
        
               super().__init__(settings, db)

     

       125
       125
       +
               self.jetstream = settings.get("jetstream", "wss://jetstream2.us-east.bsky.network/subscribe")

     

       129
       126
        
           

     

       130
       130
       -
           def __on_commit(self, outputs: list[cross.Output], message: firehose_models.MessageFrame):

     

       131
       131
       -
               blocks = message.body.get('blocks')

     

       132
       132
       -
               if not blocks:

     

       127
       127
       +
           def __on_commit(self, outputs: list[cross.Output], msg: dict):

     

       128
       128
       +
               if msg.get('did') != self.user_id:

     

       133
       129
        
                   return

     

       134
       130
        
               

     

       135
       135
       -
               parsed = parse_firehose(message)

     

       136
       136
       -
               if not isinstance(parsed, models.ComAtprotoSyncSubscribeRepos.Commit):

     

       131
       131
       +
               commit: dict = msg.get('commit', {})

     

       132
       132
       +
               if not commit:

     

       137
       133
        
                   return

     

       138
       138
       -
               blocks = parsed.blocks

     

       139
       139
       -
                   

     

       140
       140
       -
               car = None

     

       141
       141
       -
               def get_lazy_repo() -> CAR:

     

       142
       142
       -
                   nonlocal car, blocks

     

       134
       134
       +
               

     

       135
       135
       +
               commit_type = commit['operation']

     

       136
       136
       +
               match commit_type:

     

       137
       137
       +
                   case 'create':

     

       138
       138
       +
                       record = dict(commit.get('record', {}))

     

       139
       139
       +
                       record['$xpost.strongRef'] = {

     

       140
       140
       +
                           'cid': commit['cid'],

     

       141
       141
       +
                           'uri': f'at://{self.user_id}/{commit['collection']}/{commit['rkey']}'

     

       142
       142
       +
                       }

     

       143
       143
        
                       

     

       144
       144
       -
                   if isinstance(blocks, str):

     

       145
       145
       -
                       blocks = blocks.encode()

     

       146
       146
       -
                   assert blocks

     

       147
       147
       -
                       

     

       148
       148
       -
                   if car:

     

       149
       149
       -
                       return car

     

       150
       150
       -
                   car = CAR.from_bytes(blocks)

     

       151
       151
       -
                   return car

     

       152
       152
       -
                   

     

       153
       153
       -
               for op in parsed.ops:

     

       154
       154
       -
                   if op.action == 'delete':

     

       155
       155
       -
                       if not op.prev:

     

       156
       156
       -
                           continue

     

       157
       157
       -
       

     

       158
       158
       -
                       if op.path.startswith('app.bsky.feed.post'):

     

       159
       159
       -
                           self._on_delete_post(outputs, {

     

       160
       160
       -
                               'cid': op.prev.encode(),

     

       161
       161
       -
                               'uri': f'at://{parsed.repo}/{op.path}' 

     

       162
       162
       -
                           }, False)

     

       163
       163
       -
                       elif op.path.startswith('app.bsky.feed.repost'):

     

       164
       164
       -
                           self._on_delete_post(outputs, {

     

       165
       165
       -
                               'cid': op.prev.encode(),

     

       166
       166
       -
                               'uri': f'at://{parsed.repo}/{op.path}' 

     

       167
       167
       -
                           }, True)

     

       168
       168
       -
                       

     

       169
       169
       -
                       continue

     

       170
       170
       -
                       

     

       171
       171
       -
                   if op.action != 'create':

     

       172
       172
       -
                       continue

     

       173
       173
       -
       

     

       174
       174
       -
                   if not op.cid:

     

       175
       175
       -
                       continue

     

       176
       176
       -
                       

     

       177
       177
       -
                   record_data = get_lazy_repo().blocks.get(op.cid)

     

       178
       178
       -
                   if not record_data:

     

       179
       179
       -
                       continue

     

       180
       180
       -
                   

     

       181
       181
       -
                   record_dict = dict(record_data)

     

       182
       182
       -
                   record_dict['$xpost.strongRef'] = {

     

       183
       183
       -
                       'cid': op.cid.encode(),

     

       184
       184
       -
                       'uri': f'at://{parsed.repo}/{op.path}'

     

       185
       185
       -
                   }

     

       186
       186
       -
                   if record_dict['$type'] == 'app.bsky.feed.post':

     

       187
       187
       -
                       self._on_post(outputs, record_dict)

     

       188
       188
       -
                   elif record_dict['$type'] == 'app.bsky.feed.repost':

     

       189
       189
       -
                       self._on_repost(outputs, record_dict)

     

       190
       190
       -
                       

     

       144
       144
       +
                       match commit['collection']:

     

       145
       145
       +
                           case 'app.bsky.feed.post':

     

       146
       146
       +
                               self._on_post(outputs, record)

     

       147
       147
       +
                           case 'app.bsky.feed.repost':

     

       148
       148
       +
                               self._on_repost(outputs, record)

     

       149
       149
       +
                   case 'delete':

     

       150
       150
       +
                       post_id: str = f'at://{self.user_id}/{commit['collection']}/{commit['rkey']}'

     

       151
       151
       +
                       match commit['collection']:

     

       152
       152
       +
                           case 'app.bsky.feed.post':

     

       153
       153
       +
                               self._on_delete_post(outputs, post_id, False)

     

       154
       154
       +
                           case 'app.bsky.feed.repost':

     

       155
       155
       +
                               self._on_delete_post(outputs, post_id, True)

     

       191
       156
        
           

     

       192
       157
        
           async def listen(self, outputs: list[cross.Output], submit: Callable[[Callable[[], Any]], Any]):

     

       193
       193
       -
               streaming: str = f"wss://{self.pds.split("://", 1)[1]}/xrpc"

     

       194
       194
       -
               

     

       195
       195
       -
               client = AsyncFirehoseSubscribeReposClient(base_uri=streaming)

     

       158
       158
       +
               uri = self.jetstream + '?'

     

       159
       159
       +
               uri += "wantedCollections=app.bsky.feed.post"

     

       160
       160
       +
               uri += "&wantedCollections=app.bsky.feed.repost"

     

       161
       161
       +
               uri += f"&wantedDids={self.user_id}"

     

       196
       162
        
               

     

       197
       197
       -
               async def on_message(message: firehose_models.MessageFrame):

     

       198
       198
       -
                   if message.header.t != '#commit':

     

       199
       199
       -
                       return

     

       200
       200
       -
                   

     

       201
       201
       -
                   if message.body.get('repo') != self.user_id:

     

       202
       202
       -
                       return

     

       163
       163
       +
               async for ws in websockets.connect(uri, extra_headers={"User-Agent": "XPost/0.0.3"}):

     

       164
       164
       +
                   try:

     

       165
       165
       +
                       LOGGER.info("Listening to %s...", self.jetstream)

     

       166
       166
       +
                       

     

       167
       167
       +
                       async def listen_for_messages():

     

       168
       168
       +
                           async for msg in ws:

     

       169
       169
       +
                               submit(lambda: self.__on_commit(outputs, json.loads(msg)))

     

       203
       170
        
                   

     

       204
       204
       -
                   if message.body.get('tooBig'):

     

       205
       205
       -
                       LOGGER.error("#commit message is tooBig!")

     

       206
       206
       -
                       return

     

       207
       207
       -
       

     

       208
       208
       -
                   submit(lambda: self.__on_commit(outputs, message))

     

       209
       209
       -
                   return

     

       210
       210
       -
               

     

       211
       211
       -
               LOGGER.info("Listening to %s...", streaming + '/com.atproto.sync.subscribeRepos')

     

       212
       212
       -
               await client.start(on_message)
     

       171
       171
       +
                       listen = asyncio.create_task(listen_for_messages())

     

       172
       172
       +
                       

     

       173
       173
       +
                       await asyncio.gather(listen)

     

       174
       174
       +
                   except websockets.ConnectionClosedError as e:

     

       175
       175
       +
                       LOGGER.error(e, stack_info=True, exc_info=True)

     

       176
       176
       +
                       LOGGER.info("Reconnecting to %s...", self.jetstream)

     

       177
       177
       +
                       continue

+38 -15

bluesky/output.py

···

       80
       80
        
                   LOGGER.error("Failed to find thread tuple in the database!")

     

       81
       81
        
                   return None

     

       82
       82
        
               

     

       83
       83
       -
               root_ref = json.loads(thread_tuple[0])

     

       84
       84
       -
               reply_ref = json.loads(thread_tuple[1])

     

       83
       83
       +
               root_uri: str = thread_tuple[0]

     

       84
       84
       +
               reply_uri: str = thread_tuple[1]

     

       85
       85
       +
               

     

       86
       86
       +
               root_cid = database.fetch_data(self.db, root_uri, login.did, SERVICE)['cid']

     

       87
       87
       +
               reply_cid = database.fetch_data(self.db, root_uri, login.did, SERVICE)['cid']

     

       85
       88
        
               

     

       86
       86
       -
               root_record = models.AppBskyFeedPost.CreateRecordResponse(uri=str(root_ref['uri']), cid=str(root_ref['cid']))

     

       87
       87
       -
               reply_record = models.AppBskyFeedPost.CreateRecordResponse(uri=str(reply_ref['uri']), cid=str(reply_ref['cid']))

     

       89
       89
       +
               root_record = models.AppBskyFeedPost.CreateRecordResponse(uri=root_uri, cid=root_cid)

     

       90
       90
       +
               reply_record = models.AppBskyFeedPost.CreateRecordResponse(uri=reply_uri, cid=reply_cid)

     

       88
       91
        
               

     

       89
       92
        
               return (

     

       90
       93
        
                   models.create_strong_ref(root_record),

     
···

       345
       348
        
               db_post = database.find_post(self.db, post.get_id(), self.input.user_id, self.input.service)

     

       346
       349
        
               assert db_post, "ghghghhhhh"

     

       347
       350
        
               

     

       348
       348
       -
               db_identifiers = [json.dumps(cr.model_dump(), sort_keys=True) for cr in created_records]

     

       349
       349
       -
               

     

       350
       351
        
               if new_root_id is None or  new_parent_id is None:

     

       351
       352
        
                   new_root_id = database.insert_post(

     

       352
       353
        
                       self.db,

     

       353
       353
       -
                       db_identifiers[0],

     

       354
       354
       +
                       created_records[0].uri,

     

       354
       355
        
                       login.did,

     

       355
       356
        
                       SERVICE

     

       356
       357
        
                   )

     

       358
       358
       +
                   database.store_data(

     

       359
       359
       +
                       self.db, 

     

       360
       360
       +
                       created_records[0].uri, 

     

       361
       361
       +
                       login.did, 

     

       362
       362
       +
                       SERVICE,

     

       363
       363
       +
                       {'cid': created_records[0].cid}

     

       364
       364
       +
                   )

     

       365
       365
       +
                   

     

       357
       366
        
                   new_parent_id = new_root_id

     

       358
       367
        
                   database.insert_mapping(self.db, db_post['id'], new_parent_id)

     

       359
       359
       -
                   db_identifiers = db_identifiers[1:]

     

       368
       368
       +
                   created_records = created_records[1:]

     

       360
       369
        
               

     

       361
       361
       -
               for db_id in db_identifiers:

     

       370
       370
       +
               for record in created_records:

     

       362
       371
        
                   new_parent_id = database.insert_reply(

     

       363
       372
        
                       self.db, 

     

       364
       364
       -
                       db_id,

     

       373
       373
       +
                       record.uri,

     

       365
       374
        
                       login.did,

     

       366
       375
        
                       SERVICE,

     

       367
       376
        
                       new_parent_id,

     

       368
       377
        
                       new_root_id

     

       369
       378
        
                   )

     

       379
       379
       +
                   database.store_data(

     

       380
       380
       +
                       self.db, 

     

       381
       381
       +
                       record.uri, 

     

       382
       382
       +
                       login.did, 

     

       383
       383
       +
                       SERVICE,

     

       384
       384
       +
                       {'cid': record.cid}

     

       385
       385
       +
                   )

     

       370
       386
        
                   database.insert_mapping(self.db, db_post['id'], new_parent_id)

     

       371
       387
        
                   

     

       372
       388
        
           def delete_post(self, identifier: str):

     
···

       379
       395
        
               mappings = database.find_mappings(self.db, post['id'], SERVICE, login.did)

     

       380
       396
        
               for mapping in mappings[::-1]:

     

       381
       397
        
                   LOGGER.info("Deleting '%s'...", mapping[0])

     

       382
       382
       -
                   self.bsky.delete_post(json.loads(mapping[0])['uri'])

     

       398
       398
       +
                   self.bsky.delete_post(mapping[0])

     

       383
       399
        
                   database.delete_post(self.db, mapping[0], SERVICE, login.did)

     

       384
       400
        
           

     

       385
       401
        
           def accept_repost(self, repost_id: str, reposted_id: str):

     
···

       394
       410
        
               # mappings of the reposted post

     

       395
       411
        
               mappings = database.find_mappings(self.db, reposted['id'], SERVICE, login.did)

     

       396
       412
        
               if mappings:

     

       397
       397
       -
                   id = json.loads(mappings[0][0])

     

       398
       398
       -
                   rsp = self.bsky.repost(id['uri'], id['cid'])

     

       413
       413
       +
                   cid = database.fetch_data(self.db, mappings[0][0], login.did, SERVICE)['cid']

     

       414
       414
       +
                   rsp = self.bsky.repost(mappings[0][0], cid)

     

       399
       415
        
                   

     

       400
       416
        
                   internal_id = database.insert_repost(

     

       401
       417
        
                       self.db, 

     

       402
       402
       -
                       json.dumps(rsp.model_dump(), sort_keys=True),

     

       418
       418
       +
                       rsp.uri,

     

       403
       419
        
                       reposted['id'],

     

       404
       420
        
                       login.did,

     

       405
       421
        
                       SERVICE)

     

       422
       422
       +
                   database.store_data(

     

       423
       423
       +
                       self.db, 

     

       424
       424
       +
                       rsp.uri, 

     

       425
       425
       +
                       login.did, 

     

       426
       426
       +
                       SERVICE,

     

       427
       427
       +
                       {'cid': rsp.cid}

     

       428
       428
       +
                   )

     

       406
       429
        
                   database.insert_mapping(self.db, repost['id'], internal_id)

     

       407
       430
        
           

     

       408
       431
        
           def __delete_repost(self, repost_id: str) -> tuple[models.AppBskyActorDefs.ProfileViewDetailed | None, dict | None]:

     
···

       415
       438
        
               mappings = database.find_mappings(self.db, repost['id'], SERVICE, login.did)

     

       416
       439
        
               if mappings:

     

       417
       440
        
                   LOGGER.info("Deleting '%s'...", mappings[0][0])

     

       418
       418
       -
                   self.bsky.unrepost(json.loads(mappings[0][0])['uri'])

     

       441
       441
       +
                   self.bsky.unrepost(mappings[0][0])

     

       419
       442
        
                   database.delete_post(self.db, mappings[0][0], login.did, SERVICE)

     

       420
       443
        
               return login, repost

     

       421
       444

+7 -2

main.py

···

       5
       5
        
       from util.util import LOGGER, as_json

     

       6
       6
        
       import cross, util.database as database

     

       7
       7
        
       

     

       8
       8
       -
       from bluesky.input import BlueskyPdsInput

     

       8
       8
       +
       from bluesky.input import BlueskyJetstreamInput

     

       9
       9
        
       from bluesky.output import BlueskyOutputOptions, BlueskyOutput

     

       10
       10
        
       

     

       11
       11
        
       from mastodon.input import MastodonInputOptions, MastodonInput

     
···

       33
       33
        
       INPUTS = {

     

       34
       34
        
           "mastodon-wss": lambda settings, db: MastodonInput(settings, db),

     

       35
       35
        
           "misskey-wss": lambda settigs, db: MisskeyInput(settigs, db),

     

       36
       36
       -
           "bluesky-pds-wss": lambda settings, db: BlueskyPdsInput(settings, db)

     

       36
       36
       +
           "bluesky-jetstream-wss": lambda settings, db: BlueskyJetstreamInput(settings, db)

     

       37
       37
        
       }

     

       38
       38
        
       

     

       39
       39
        
       OUTPUTS = {

     
···

       89
       89
        
               db_worker.execute("""

     

       90
       90
        
                   ALTER TABLE posts

     

       91
       91
        
                   ADD COLUMN reposted_id INTEGER NULL REFERENCES posts(id) ON DELETE SET NULL

     

       92
       92
       +
               """)

     

       93
       93
       +
           if "extra_data" not in column_names:

     

       94
       94
       +
               db_worker.execute("""

     

       95
       95
       +
                   ALTER TABLE posts

     

       96
       96
       +
                   ADD COLUMN extra_data TEXT NULL

     

       92
       97
        
               """)

     

       93
       98
        
           

     

       94
       99
        
           # create the mappings table

+25

util/database.py

···

       2
       2
        
       from concurrent.futures import Future

     

       3
       3
        
       import threading

     

       4
       4
        
       import queue

     

       5
       5
       +
       import json

     

       5
       6
        
       

     

       6
       7
        
       class DataBaseWorker():

     

       7
       8
        
           def __init__(self, database: str) -> None:

     
···

       125
       126
        
                 AND user_id = ?

     

       126
       127
        
               """, (identifier, serivce, user_id))

     

       127
       128
        
           

     

       129
       129
       +
       def fetch_data(db: DataBaseWorker, identifier: str, user_id: str, service: str) -> dict:

     

       130
       130
       +
           result = db.execute(

     

       131
       131
       +
               """

     

       132
       132
       +
               SELECT extra_data

     

       133
       133
       +
               FROM posts 

     

       134
       134
       +
               WHERE identifier = ? 

     

       135
       135
       +
                 AND user_id = ? 

     

       136
       136
       +
                 AND service = ?

     

       137
       137
       +
               """, (identifier, user_id, service))

     

       138
       138
       +
           if not result or not result[0]:

     

       139
       139
       +
               return {}

     

       140
       140
       +
           return json.loads(result[0][0])

     

       141
       141
       +
       

     

       142
       142
       +
       def store_data(db: DataBaseWorker, identifier: str, user_id: str, service: str, extra_data: dict) -> None:

     

       143
       143
       +
           db.execute(

     

       144
       144
       +
               """

     

       145
       145
       +
               UPDATE posts

     

       146
       146
       +
               SET extra_data = ?

     

       147
       147
       +
               WHERE identifier = ?

     

       148
       148
       +
                 AND user_id = ?

     

       149
       149
       +
                 AND service = ?

     

       150
       150
       +
               """,

     

       151
       151
       +
               (json.dumps(extra_data), identifier, user_id, service)

     

       152
       152
       +
           )

     

       128
       153
        
       

     

       129
       154
        
       def find_mappings(db: DataBaseWorker, original_post: int, service: str, user_id: str) -> list[str]:

     

       130
       155
        
           return db.execute(