commit 7b2d0f108f60528a409988f9d71d02a06df41f9b · zenfyr.dev/xpost

+196 -116

bluesky.py

···

       46
       46
        
               self.client.login(did, util.get_or_envvar(settings, 'app-password'))

     

       47
       47
        
               self.bsky = Bluesky(self.client)

     

       48
       48
        
           

     

       49
       49
       +
           def _find_parent(self, parent_id: str):

     

       50
       50
       +
               login = self.client.me

     

       51
       51
       +
               if not login:

     

       52
       52
       +
                   raise Exception("Client not logged in!")

     

       53
       53
       +
               

     

       54
       54
       +
               reply_data = database.find_post(self.db, parent_id, self.input.user_id, self.input.service)

     

       55
       55
       +
               assert reply_data, "reply_data requested, but doesn't exist in db (should've been skipped bt firehose)"

     

       56
       56
       +
                   

     

       57
       57
       +
               reply_mappings = [json.loads(data[0]) for data in database.find_mappings(self.db, reply_data['id'], SERVICE, login.did)]

     

       58
       58
       +
               if not reply_mappings:

     

       59
       59
       +
                   LOGGER.error("Failed to find mappings for a post in the db!")

     

       60
       60
       +
                   return None

     

       61
       61
       +
                   

     

       62
       62
       +
               reply_record = models.AppBskyFeedPost.CreateRecordResponse(uri=str(reply_mappings[-1]['uri']), cid=str(reply_mappings[-1]['cid']))

     

       63
       63
       +
               root_record = models.AppBskyFeedPost.CreateRecordResponse(uri=str(reply_mappings[0]['uri']), cid=str(reply_mappings[0]['cid']))

     

       64
       64
       +
               if reply_data['root_id']:

     

       65
       65
       +
                   root_data = database.find_post_by_id(self.db, reply_data['root_id'])

     

       66
       66
       +
                   assert root_data, "root_data requested but doesn't exist in db"

     

       67
       67
       +
                       

     

       68
       68
       +
                   root_mappings = [json.loads(data[0]) for data in database.find_mappings(self.db, reply_data['root_id'], SERVICE, login.did)]

     

       69
       69
       +
                   if not root_mappings:

     

       70
       70
       +
                       LOGGER.error("Failed to find mappings for a post in the db!")

     

       71
       71
       +
                       return None

     

       72
       72
       +
                   root_record = models.AppBskyFeedPost.CreateRecordResponse(uri=str(root_mappings[0]['uri']), cid=str(root_mappings[0]['cid']))

     

       73
       73
       +
               

     

       74
       74
       +
               return (

     

       75
       75
       +
                   models.create_strong_ref(root_record),

     

       76
       76
       +
                   models.create_strong_ref(reply_record),

     

       77
       77
       +
                   reply_data['root_id'],

     

       78
       78
       +
                   reply_data['id']

     

       79
       79
       +
               )

     

       80
       80
       +
           

     

       81
       81
       +
           def _split_attachments(self, attachments: list[cross.MediaAttachment]):

     

       82
       82
       +
               sup_media: list[cross.MediaAttachment] = []

     

       83
       83
       +
               unsup_media: list[cross.MediaAttachment] = []

     

       84
       84
       +
               

     

       85
       85
       +
               for attachment in attachments:

     

       86
       86
       +
                   attachment_type = attachment.get_type()

     

       87
       87
       +
                   if not attachment_type:

     

       88
       88
       +
                       continue

     

       89
       89
       +
       

     

       90
       90
       +
                   if attachment_type in {'video', 'image'}: # TODO convert gifs to videos

     

       91
       91
       +
                       sup_media.append(attachment)

     

       92
       92
       +
                   else:

     

       93
       93
       +
                       unsup_media.append(attachment)

     

       94
       94
       +
               

     

       95
       95
       +
               return (sup_media, unsup_media)

     

       96
       96
       +
       

     

       97
       97
       +
           def _split_media_per_post(

     

       98
       98
       +
               self, 

     

       99
       99
       +
               tokens: list[client_utils.TextBuilder], 

     

       100
       100
       +
               media: list[cross.MediaAttachment]):

     

       101
       101
       +
               

     

       102
       102
       +
               posts: list[dict] = [{"tokens": tokens, "attachments": []} for tokens in tokens]

     

       103
       103
       +
               available_indices: list[int] = list(range(len(posts)))

     

       104
       104
       +
               

     

       105
       105
       +
               current_image_post_idx: int | None = None

     

       106
       106
       +
       

     

       107
       107
       +
               def make_blank_post() -> dict:

     

       108
       108
       +
                   return {

     

       109
       109
       +
                       "tokens": [client_utils.TextBuilder().text('')],

     

       110
       110
       +
                       "attachments": []

     

       111
       111
       +
                   }

     

       112
       112
       +
               

     

       113
       113
       +
               def pop_next_empty_index() -> int:

     

       114
       114
       +
                   if available_indices:

     

       115
       115
       +
                       return available_indices.pop(0)

     

       116
       116
       +
                   else:

     

       117
       117
       +
                       new_idx = len(posts)

     

       118
       118
       +
                       posts.append(make_blank_post())

     

       119
       119
       +
                       return new_idx

     

       120
       120
       +
               

     

       121
       121
       +
               for att in media:

     

       122
       122
       +
                   if att.get_type() == 'video':

     

       123
       123
       +
                       current_image_post_idx = None

     

       124
       124
       +
                       idx = pop_next_empty_index()

     

       125
       125
       +
                       posts[idx]["attachments"].append(att)

     

       126
       126
       +
                   elif att.get_type() == 'image':

     

       127
       127
       +
                       if (

     

       128
       128
       +
                           current_image_post_idx is not None

     

       129
       129
       +
                           and len(posts[current_image_post_idx]["attachments"]) < 4

     

       130
       130
       +
                       ):

     

       131
       131
       +
                           posts[current_image_post_idx]["attachments"].append(att)

     

       132
       132
       +
                       else:

     

       133
       133
       +
                           idx = pop_next_empty_index()

     

       134
       134
       +
                           posts[idx]["attachments"].append(att)

     

       135
       135
       +
                           current_image_post_idx = idx

     

       136
       136
       +
               

     

       137
       137
       +
               result: list[tuple[client_utils.TextBuilder, list[cross.MediaAttachment]]] = []

     

       138
       138
       +
               for p in posts:

     

       139
       139
       +
                   result.append((p["tokens"], p["attachments"]))

     

       140
       140
       +
               return result

     

       141
       141
       +
           

     

       49
       142
        
           def accept_post(self, post: cross.Post):

     

       50
       143
        
               login = self.client.me

     

       51
       144
        
               if not login:

     
···

       60
       153
        
               root_ref = None

     

       61
       154
        
               reply_ref = None

     

       62
       155
        
               if parent_id:

     

       63
       63
       -
                   # parentless posts are skipped by the input

     

       64
       64
       -
                   reply_data = database.find_post(self.db, parent_id, self.input.user_id, self.input.service)

     

       65
       65
       -
                   assert reply_data, "reply_data requested, but doesn't exist in db (should've been skipped bt firehose)"

     

       66
       66
       -
                   

     

       67
       67
       -
                   reply_mappings = [json.loads(data[0]) for data in database.find_mappings(self.db, reply_data['id'], SERVICE, login.did)]

     

       68
       68
       -
                   if not reply_mappings:

     

       69
       69
       -
                       LOGGER.error("Failed to find mappings for a post in the db!")

     

       156
       156
       +
                   parents = self._find_parent(parent_id)

     

       157
       157
       +
                   if not parents:

     

       70
       158
        
                       return

     

       71
       71
       -
                   

     

       72
       72
       -
                   reply_record = models.AppBskyFeedPost.CreateRecordResponse(uri=str(reply_mappings[-1]['uri']), cid=str(reply_mappings[-1]['cid']))

     

       73
       73
       -
                   root_record = models.AppBskyFeedPost.CreateRecordResponse(uri=str(reply_mappings[0]['uri']), cid=str(reply_mappings[0]['cid']))

     

       74
       74
       -
                   if reply_data['root_id']:

     

       75
       75
       -
                       root_data = database.find_post_by_id(self.db, reply_data['root_id'])

     

       76
       76
       -
                       assert root_data, "root_data requested but doesn't exist in db"

     

       77
       77
       -
                       

     

       78
       78
       -
                       root_mappings = [json.loads(data[0]) for data in database.find_mappings(self.db, reply_data['root_id'], SERVICE, login.did)]

     

       79
       79
       -
                       if not root_mappings:

     

       80
       80
       -
                           LOGGER.error("Failed to find mappings for a post in the db!")

     

       81
       81
       -
                           return

     

       82
       82
       -
                       root_record = models.AppBskyFeedPost.CreateRecordResponse(uri=str(root_mappings[0]['uri']), cid=str(root_mappings[0]['cid']))

     

       83
       83
       -
                   

     

       84
       84
       -
                   new_root_id = reply_data['root_id']

     

       85
       85
       -
                   new_parent_id = reply_data['id']

     

       86
       86
       -
                   

     

       87
       87
       -
                   root_ref = models.create_strong_ref(root_record)

     

       88
       88
       -
                   reply_ref = models.create_strong_ref(reply_record)

     

       159
       159
       +
                   root_ref, reply_ref, new_root_id, new_parent_id = parents

     

       89
       160
        
               

     

       90
       161
        
               tokens = post.get_tokens()

     

       91
       162
        
               

     
···

       105
       176
        
                   unique_labels.add('graphic-media')

     

       106
       177
        
               

     

       107
       178
        
               labels = models.ComAtprotoLabelDefs.SelfLabels(values=[models.ComAtprotoLabelDefs.SelfLabel(val=label) for label in unique_labels])

     

       179
       179
       +
       

     

       180
       180
       +
               sup_media, unsup_media = self._split_attachments(post.get_attachments())

     

       181
       181
       +
       

     

       182
       182
       +
               if unsup_media:

     

       183
       183
       +
                   if tokens:

     

       184
       184
       +
                       tokens.append(cross.TextToken('\n'))

     

       185
       185
       +
                   for i, attachment in enumerate(unsup_media):

     

       186
       186
       +
                       tokens.append(cross.LinkToken(

     

       187
       187
       +
                               attachment.get_url(),

     

       188
       188
       +
                               f"[{media_util.get_filename_from_url(attachment.get_url())}]"

     

       189
       189
       +
                       ))

     

       190
       190
       +
                       tokens.append(cross.TextToken(' '))

     

       191
       191
       +
       

     

       108
       192
        
               

     

       109
       193
        
               split_tokens: list[list[cross.Token]] = util.split_tokens(post.get_tokens(), 300)

     

       110
       194
        
               post_text: list[client_utils.TextBuilder] = []

     
···

       121
       205
        
               if not post_text:

     

       122
       206
        
                   post_text = [client_utils.TextBuilder().text('')]

     

       123
       207
        
               

     

       208
       208
       +
               # download media first. increased RAM usage, but more reliable

     

       209
       209
       +
               for m in sup_media:

     

       210
       210
       +
                   if not m.bytes:

     

       211
       211
       +
                       if m.get_type() == 'image':

     

       212
       212
       +
                           image_bytes = media_util.download_blob(m.get_url(), max_bytes=2_000_000)

     

       213
       213
       +
                           if not image_bytes:

     

       214
       214
       +
                               LOGGER.error("Skipping post_id '%s', failed to download attachment! File too large?", post.get_id())

     

       215
       215
       +
                               return

     

       216
       216
       +
                           m.bytes = image_bytes

     

       217
       217
       +
                       elif m.get_type() == 'video':

     

       218
       218
       +
                           video_bytes = media_util.download_blob(m.get_url(), max_bytes=100_000_000)

     

       219
       219
       +
                           if not video_bytes:

     

       220
       220
       +
                               LOGGER.error("Skipping post_id '%s', failed to download attachment! File too large?", post.get_id())

     

       221
       221
       +
                               return

     

       222
       222
       +
                           m.bytes = video_bytes

     

       223
       223
       +
               

     

       124
       224
        
               created_records: list[models.AppBskyFeedPost.CreateRecordResponse] = []

     

       125
       125
       -
               attachments = post.get_attachments()

     

       126
       126
       -
               if not attachments:

     

       127
       127
       -
                   for text in post_text:

     

       225
       225
       +
               baked_media = self._split_media_per_post(post_text, sup_media)

     

       226
       226
       +
               

     

       227
       227
       +
               for text, attachments in baked_media:

     

       228
       228
       +
                   if not attachments:

     

       128
       229
        
                       if reply_ref and root_ref:

     

       129
       230
        
                           new_post = self.bsky.send_post(text, reply_to=models.AppBskyFeedPost.ReplyRef(

     

       130
       231
        
                               parent=reply_ref,

     
···

       137
       238
        
                       self.bsky.create_gates(self.options, new_post.uri)

     

       138
       239
        
                       reply_ref = models.create_strong_ref(new_post)

     

       139
       240
        
                       created_records.append(new_post)

     

       140
       140
       -
               elif len(attachments) <= 4:

     

       141
       141
       -
                   if len(attachments) == 1 and attachments[0].get_type() == 'video':

     

       142
       142
       -
                       video_data = attachments[0]

     

       143
       143
       -
                       

     

       144
       144
       -
                       video_io = media_util.download_blob(video_data.get_url(), max_bytes=100_000_000)

     

       145
       145
       -
                       if not video_io:

     

       146
       146
       -
                           LOGGER.error("Skipping post_id '%s', failed to download attachment! File too large?", post.get_id())

     

       147
       147
       -
                           return

     

       148
       148
       -
                       

     

       149
       149
       -
                       metadata = video_data.create_meta(video_io)

     

       150
       150
       -
                       if metadata.get_duration() > 180:

     

       151
       151
       -
                           LOGGER.info("Skipping post_id '%s', video attachment too long!", post.get_id())

     

       152
       152
       -
                           return

     

       153
       153
       -
                       

     

       154
       154
       -
                       aspect_ratio = models.AppBskyEmbedDefs.AspectRatio(

     

       155
       155
       -
                           width=metadata.get_width(), 

     

       156
       156
       -
                           height=metadata.get_height()

     

       157
       157
       -
                       )

     

       158
       158
       -
                       

     

       159
       159
       -
                       new_post = self.bsky.send_video(

     

       160
       160
       -
                           text=post_text[0],

     

       161
       161
       -
                           video=video_io,

     

       162
       162
       -
                           video_aspect_ratio=aspect_ratio,

     

       163
       163
       -
                           video_alt=video_data.get_alt(),

     

       164
       164
       -
                           reply_to= models.AppBskyFeedPost.ReplyRef(

     

       165
       165
       -
                               parent=reply_ref,

     

       166
       166
       -
                               root=root_ref

     

       167
       167
       -
                           ) if root_ref and reply_ref else None,

     

       168
       168
       -
                           labels=labels

     

       169
       169
       -
                       )

     

       170
       170
       -
                       if not root_ref:

     

       171
       171
       -
                           root_ref = models.create_strong_ref(new_post)

     

       172
       172
       -
                           

     

       173
       173
       -
                       self.bsky.create_gates(self.options, new_post.uri)

     

       174
       174
       -
                       reply_ref = models.create_strong_ref(new_post)

     

       175
       241
        
                   else:

     

       176
       176
       -
                       for attachment in attachments:

     

       177
       177
       -
                           if attachment.get_type() != 'image':

     

       178
       178
       -
                               LOGGER.info("Skipping post_id '%s'. Attachment type mismatch. got: '%s' expected: 'image'", post.get_id(), attachment.get_type())

     

       179
       179
       -
                               return

     

       180
       180
       -
                       

     

       181
       181
       -
                       images: list[bytes] = []

     

       182
       182
       -
                       image_alts: list[str] = []

     

       183
       183
       -
                       image_aspect_ratios: list[models.AppBskyEmbedDefs.AspectRatio] = []

     

       184
       184
       -
                       for attachment in attachments:

     

       185
       185
       -
                           image_io = media_util.download_blob(attachment.get_url(), max_bytes=2_000_000)

     

       186
       186
       -
                           if not image_io:

     

       242
       242
       +
                       # if a single post is an image - everything else is an image

     

       243
       243
       +
                       if attachments[0].get_type() == 'image':

     

       244
       244
       +
                           images: list[bytes] = []

     

       245
       245
       +
                           image_alts: list[str] = []

     

       246
       246
       +
                           image_aspect_ratios: list[models.AppBskyEmbedDefs.AspectRatio] = []

     

       247
       247
       +
                           

     

       248
       248
       +
                           for attachment in attachments:

     

       249
       249
       +
                               assert attachment.bytes

     

       250
       250
       +
                               image_io = media_util.compress_image(attachment.bytes, quality=100)

     

       251
       251
       +
                               metadata = attachment.create_meta(image_io)

     

       252
       252
       +
                           

     

       253
       253
       +
                               if len(image_io) > 1_000_000:

     

       254
       254
       +
                                   LOGGER.info("Compressing %s...", attachment.get_url())

     

       255
       255
       +
                           

     

       256
       256
       +
                               images.append(image_io)

     

       257
       257
       +
                               image_alts.append(attachment.get_alt())

     

       258
       258
       +
                               image_aspect_ratios.append(models.AppBskyEmbedDefs.AspectRatio(

     

       259
       259
       +
                                   width=metadata.get_width(), 

     

       260
       260
       +
                                   height=metadata.get_height()

     

       261
       261
       +
                               ))

     

       262
       262
       +
                           

     

       263
       263
       +
                           new_post = self.bsky.send_images(

     

       264
       264
       +
                               text=post_text[0],

     

       265
       265
       +
                               images=images,

     

       266
       266
       +
                               image_alts=image_alts,

     

       267
       267
       +
                               image_aspect_ratios=image_aspect_ratios,

     

       268
       268
       +
                               reply_to= models.AppBskyFeedPost.ReplyRef(

     

       269
       269
       +
                                   parent=reply_ref,

     

       270
       270
       +
                                   root=root_ref

     

       271
       271
       +
                               ) if root_ref and reply_ref else None, 

     

       272
       272
       +
                               labels=labels

     

       273
       273
       +
                           )

     

       274
       274
       +
                           if not root_ref:

     

       275
       275
       +
                               root_ref = models.create_strong_ref(new_post)

     

       276
       276
       +
                           

     

       277
       277
       +
                           self.bsky.create_gates(self.options, new_post.uri)

     

       278
       278
       +
                           reply_ref = models.create_strong_ref(new_post)

     

       279
       279
       +
                           created_records.append(new_post)

     

       280
       280
       +
                       else: # video is guarantedd to be one

     

       281
       281
       +
                           video_data = attachments[0]

     

       282
       282
       +
                           assert attachment.bytes

     

       283
       283
       +
                           

     

       284
       284
       +
                           video_io = attachment.bytes

     

       285
       285
       +
                           if not video_io:

     

       187
       286
        
                               LOGGER.error("Skipping post_id '%s', failed to download attachment! File too large?", post.get_id())

     

       188
       287
        
                               return

     

       189
       189
       -
                           LOGGER.info("Converting %s to .webp...", attachment.get_url())

     

       190
       190
       -
                           image_io = media_util.compress_image(image_io, quality=100)

     

       191
       191
       -
                           metadata = attachment.create_meta(image_io)

     

       192
       288
        
                           

     

       193
       193
       -
                           if len(image_io) > 1_000_000:

     

       194
       194
       -
                               LOGGER.info("Compressing %s...", attachment.get_url())

     

       195
       195
       -
                           

     

       196
       196
       -
                           images.append(image_io)

     

       197
       197
       -
                           image_alts.append(attachment.get_alt())

     

       198
       198
       -
                           image_aspect_ratios.append(models.AppBskyEmbedDefs.AspectRatio(

     

       289
       289
       +
                           metadata = video_data.create_meta(video_io)

     

       290
       290
       +
                           if metadata.get_duration() > 180:

     

       291
       291
       +
                               LOGGER.info("Skipping post_id '%s', video attachment too long!", post.get_id())

     

       292
       292
       +
                               return

     

       293
       293
       +
                       

     

       294
       294
       +
                           aspect_ratio = models.AppBskyEmbedDefs.AspectRatio(

     

       199
       295
        
                               width=metadata.get_width(), 

     

       200
       296
        
                               height=metadata.get_height()

     

       201
       201
       -
                           ))

     

       297
       297
       +
                           )

     

       202
       298
        
                           

     

       203
       203
       -
                       new_post = self.bsky.send_images(

     

       204
       204
       -
                           text=post_text[0],

     

       205
       205
       -
                           images=images,

     

       206
       206
       -
                           image_alts=image_alts,

     

       207
       207
       -
                           image_aspect_ratios=image_aspect_ratios,

     

       208
       208
       -
                           reply_to= models.AppBskyFeedPost.ReplyRef(

     

       209
       209
       -
                               parent=reply_ref,

     

       210
       210
       -
                               root=root_ref

     

       211
       211
       -
                           ) if root_ref and reply_ref else None, 

     

       212
       212
       -
                           labels=labels

     

       213
       213
       -
                       )

     

       214
       214
       -
                       if not root_ref:

     

       215
       215
       -
                           root_ref = models.create_strong_ref(new_post)

     

       299
       299
       +
                           new_post = self.bsky.send_video(

     

       300
       300
       +
                               text=post_text[0],

     

       301
       301
       +
                               video=video_io,

     

       302
       302
       +
                               video_aspect_ratio=aspect_ratio,

     

       303
       303
       +
                               video_alt=video_data.get_alt(),

     

       304
       304
       +
                               reply_to= models.AppBskyFeedPost.ReplyRef(

     

       305
       305
       +
                                   parent=reply_ref,

     

       306
       306
       +
                                   root=root_ref

     

       307
       307
       +
                               ) if root_ref and reply_ref else None,

     

       308
       308
       +
                               labels=labels

     

       309
       309
       +
                           )

     

       310
       310
       +
                           if not root_ref:

     

       311
       311
       +
                               root_ref = models.create_strong_ref(new_post)

     

       216
       312
        
                           

     

       217
       217
       -
                       self.bsky.create_gates(self.options, new_post.uri)

     

       218
       218
       -
                       reply_ref = models.create_strong_ref(new_post)

     

       219
       219
       -
                       

     

       220
       220
       -
                   created_records.append(new_post)

     

       221
       221
       -
                   for text in post_text[1:]:

     

       222
       222
       -
                       new_post = self.bsky.send_post(text, reply_to=models.AppBskyFeedPost.ReplyRef(

     

       223
       223
       -
                           parent=reply_ref,

     

       224
       224
       -
                           root=root_ref

     

       225
       225
       -
                       ), labels=labels)

     

       226
       226
       -
                       self.bsky.create_gates(self.options, new_post.uri)

     

       227
       227
       -
                       

     

       228
       228
       -
                       reply_ref = models.create_strong_ref(new_post)

     

       229
       229
       -
                       created_records.append(new_post)

     

       230
       230
       -
               else:

     

       231
       231
       -
                   LOGGER.info("Skipping post_id '%s', too many attachments!", post.get_id())

     

       232
       232
       -
                   return 

     

       233
       233
       -
               

     

       234
       234
       -
               if not created_records:

     

       235
       235
       -
                   LOGGER.info("Skipped post_id '%s', for some reason...")

     

       313
       313
       +
                           self.bsky.create_gates(self.options, new_post.uri)

     

       314
       314
       +
                           reply_ref = models.create_strong_ref(new_post)

     

       315
       315
       +
                           created_records.append(new_post)

     

       236
       316
        
               

     

       237
       317
        
               db_post = database.find_post(self.db, post.get_id(), self.input.user_id, self.input.service)

     

       238
       318
        
               assert db_post, "ghghghhhhh"

cross.py

···

       48
       48
        
       

     

       49
       49
        
       class MediaAttachment():

     

       50
       50
        
           def __init__(self) -> None:

     

       51
       51
       +
               self.bytes: bytes | None = None # filled-in later

     

       51
       52
        
               pass

     

       52
       53
        
           

     

       53
       54
        
           def create_meta(self, bytes: bytes) -> MediaMeta:

+1 -1

mastodon.py

···

       135
       135
        
           

     

       136
       136
        
           # type of attachment

     

       137
       137
        
           def get_type(self) -> str | None:

     

       138
       138
       -
               return FORMATS[self.attachment.get('type', 'other')]

     

       138
       138
       +
               return FORMATS.get(self.attachment.get('type', 'other'), 'other')

     

       139
       139
        
               

     

       140
       140
        
           # create file metadata from bytes or other

     

       141
       141
        
           def create_meta(self, bytes: bytes) -> cross.MediaMeta:

+17

media_util.py

···

       1
       1
        
       import requests

     

       2
       2
        
       import subprocess

     

       3
       3
        
       import json

     

       4
       4
       +
       import re, urllib.parse, os

     

       4
       5
        
       from util import LOGGER

     

       6
       6
       +
       

     

       7
       7
       +
       FILENAME = re.compile(r'filename="?([^\";]*)"?')

     

       8
       8
       +
       

     

       9
       9
       +
       def get_filename_from_url(url):

     

       10
       10
       +
           try:

     

       11
       11
       +
               response = requests.head(url, allow_redirects=True)

     

       12
       12
       +
               disposition = response.headers.get('Content-Disposition')

     

       13
       13
       +
               if disposition:

     

       14
       14
       +
                   filename = FILENAME.findall(disposition)

     

       15
       15
       +
                   if filename:

     

       16
       16
       +
                       return filename[0]

     

       17
       17
       +
           except requests.RequestException:

     

       18
       18
       +
               pass

     

       19
       19
       +
       

     

       20
       20
       +
           parsed_url = urllib.parse.urlparse(url)

     

       21
       21
       +
           return os.path.basename(parsed_url.path)

     

       5
       22
        
       

     

       6
       23
        
       def probe_bytes(bytes: bytes) -> dict:

     

       7
       24
        
           cmd = [