commit 6bfd85e17ae6859b52fefe678d1c500b83959eb4 · zenfyr.dev/xpost

+52 -53

bluesky/common.py

···

       13
       13
        
       ADULT_PATTERN = re.compile(r"\b(sexual content|nsfw|erotic|adult only|18\+)\b", re.IGNORECASE)

     

       14
       14
        
       PORN_PATTERN  = re.compile(r"\b(porn|yiff|hentai|pornographic|fetish)\b", re.IGNORECASE)

     

       15
       15
        
       

     

       16
       16
       +
       class BlueskyPost(cross.Post):

     

       17
       17
       +
           def __init__(self, record: dict, tokens: list[cross.Token], attachments: list[MediaInfo]) -> None:

     

       18
       18
       +
               super().__init__()

     

       19
       19
       +
               self.uri = record['$xpost.strongRef']['uri']

     

       20
       20
       +
               self.parent_uri = None

     

       21
       21
       +
               if record.get('reply'):

     

       22
       22
       +
                   self.parent_uri = record['reply']['parent']['uri']

     

       23
       23
       +
                   

     

       24
       24
       +
               self.tokens = tokens

     

       25
       25
       +
               self.timestamp = record['createdAt']

     

       26
       26
       +
               labels = record.get('labels', {}).get('values')

     

       27
       27
       +
               if labels:

     

       28
       28
       +
                   self.spoiler = ', '.join([str(label['val']).replace('-', ' ') for label in labels])

     

       29
       29
       +
                   

     

       30
       30
       +
               self.attachments = attachments

     

       31
       31
       +
               self.languages = record.get('langs', [])

     

       32
       32
       +
           

     

       33
       33
       +
           # at:// of the post record

     

       34
       34
       +
           def get_id(self) -> str:

     

       35
       35
       +
               return self.uri

     

       36
       36
       +
           

     

       37
       37
       +
           def get_parent_id(self) -> str | None:

     

       38
       38
       +
               return self.parent_uri

     

       39
       39
       +
           

     

       40
       40
       +
           def get_tokens(self) -> list[cross.Token]:

     

       41
       41
       +
               return self.tokens

     

       42
       42
       +
           

     

       43
       43
       +
           def get_text_type(self) -> str:

     

       44
       44
       +
               return "text/plain"

     

       45
       45
       +
           

     

       46
       46
       +
           def get_timestamp(self) -> str:

     

       47
       47
       +
               return self.timestamp

     

       48
       48
       +
       

     

       49
       49
       +
           def get_attachments(self) -> list[MediaInfo]:

     

       50
       50
       +
               return self.attachments

     

       51
       51
       +
           

     

       52
       52
       +
           def get_spoiler(self) -> str | None:

     

       53
       53
       +
               return self.spoiler

     

       54
       54
       +
       

     

       55
       55
       +
           def get_languages(self) -> list[str]:

     

       56
       56
       +
               return self.languages

     

       57
       57
       +
           

     

       58
       58
       +
           def is_sensitive(self) -> bool:

     

       59
       59
       +
               return self.spoiler is not None

     

       60
       60
       +
       

     

       61
       61
       +
           def get_post_url(self) -> str | None:

     

       62
       62
       +
               parts = str(self.uri[len("at://"):]).split("/")

     

       63
       63
       +
               did, _, post_id = parts

     

       64
       64
       +
               

     

       65
       65
       +
               return f"https://bsky.app/profile/{did}/post/{post_id}"

     

       66
       66
       +
       

     

       16
       67
        
       def tokenize_post(post: dict) -> list[cross.Token]:

     

       17
       68
        
           text: str = post.get('text', '')

     

       18
       69
        
           if not text:

     
···

       86
       137
        
           if prev < len(text):

     

       87
       138
        
               tokens.append(cross.TextToken(text[prev:]))

     

       88
       139
        
           

     

       89
       89
       -
           return tokens

     

       90
       90
       -
       

     

       91
       91
       -
       class BlueskyPost(cross.Post):

     

       92
       92
       -
           def __init__(self, post: dict, tokens: list[cross.Token], attachments: list[MediaInfo]) -> None:

     

       93
       93
       -
               super().__init__()

     

       94
       94
       -
               self.post = post

     

       95
       95
       -
               self.tokens = tokens

     

       96
       96
       -
               

     

       97
       97
       -
               self.id = post['$xpost.strongRef']['uri']

     

       98
       98
       -
               

     

       99
       99
       -
               self.parent_id = None

     

       100
       100
       -
               if self.post.get('reply'):

     

       101
       101
       -
                   self.parent_id = self.post['reply']['parent']['uri']

     

       102
       102
       -
               

     

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

     

       104
       104
       -
               self.cw = ''

     

       105
       105
       -
               if labels:

     

       106
       106
       -
                   self.cw = ', '.join([str(label['val']).replace('-', ' ') for label in labels])

     

       107
       107
       -
               self.attachments = attachments

     

       108
       108
       -
           

     

       109
       109
       -
           def get_tokens(self) -> list[cross.Token]:

     

       110
       110
       -
               return self.tokens

     

       111
       111
       -
           

     

       112
       112
       -
           def get_parent_id(self) -> str | None:

     

       113
       113
       -
               return self.parent_id

     

       114
       114
       -
           

     

       115
       115
       -
           def get_post_date_iso(self) -> str:

     

       116
       116
       -
               return self.post.get('createdAt') or super().get_post_date_iso()

     

       117
       117
       -
           

     

       118
       118
       -
           def get_cw(self) -> str:

     

       119
       119
       -
               return self.cw or ''

     

       120
       120
       -
           

     

       121
       121
       -
           def get_id(self) -> str:

     

       122
       122
       -
               return self.id

     

       123
       123
       -
       

     

       124
       124
       -
           def get_languages(self) -> list[str]:

     

       125
       125
       -
               return self.post.get('langs', []) or []

     

       126
       126
       -
           

     

       127
       127
       -
           def is_sensitive(self) -> bool:

     

       128
       128
       -
               return self.post.get('labels', {}).get('values') or False

     

       129
       129
       -
           

     

       130
       130
       -
           def get_attachments(self) -> list[MediaInfo]:

     

       131
       131
       -
               return self.attachments

     

       132
       132
       -
           

     

       133
       133
       -
           def get_text_type(self) -> str:

     

       134
       134
       -
               return "text/plain"

     

       135
       135
       -
       

     

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

     

       137
       137
       -
               parts = str(self.id[len("at://"):]).split("/")

     

       138
       138
       -
               did, _, post_id = parts

     

       139
       139
       -
               

     

       140
       140
       -
               return f"https://bsky.app/profile/{did}/post/{post_id}"

     

       141
       141
       -
           

     

       140
       140
       +
           return tokens    

     

       142
       141
        
       

     

       143
       142
        
       def tokens_to_richtext(tokens: list[cross.Token]) -> client_utils.TextBuilder | None:

     

       144
       143
        
           builder = client_utils.TextBuilder()

+14 -14

bluesky/output.py

···

       173
       173
        
               tokens = post.get_tokens().copy()

     

       174
       174
        
               

     

       175
       175
        
               unique_labels: set[str] = set()

     

       176
       176
       -
               cw = post.get_cw()

     

       176
       176
       +
               cw = post.get_spoiler()

     

       177
       177
        
               if cw:

     

       178
       178
        
                   tokens.insert(0, cross.TextToken("CW: " + cw + "\n\n"))

     

       179
       179
        
                   unique_labels.add('graphic-media')

     

       180
       180
       -
               

     

       181
       181
       -
               # from bsky.app, a post can only have one of those labels

     

       182
       182
       -
               if PORN_PATTERN.search(cw):

     

       183
       183
       -
                   unique_labels.add('porn')

     

       184
       184
       -
               elif ADULT_PATTERN.search(cw):

     

       185
       185
       -
                   unique_labels.add('sexual')

     

       180
       180
       +
                   

     

       181
       181
       +
                   # from bsky.app, a post can only have one of those labels

     

       182
       182
       +
                   if PORN_PATTERN.search(cw):

     

       183
       183
       +
                       unique_labels.add('porn')

     

       184
       184
       +
                   elif ADULT_PATTERN.search(cw):

     

       185
       185
       +
                       unique_labels.add('sexual')

     

       186
       186
        
               

     

       187
       187
        
               if post.is_sensitive():

     

       188
       188
        
                   unique_labels.add('graphic-media')

     
···

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

     

       248
       248
        
                               parent=reply_ref,

     

       249
       249
        
                               root=root_ref

     

       250
       250
       -
                           ), labels=labels, time_iso=post.get_post_date_iso())

     

       250
       250
       +
                           ), labels=labels, time_iso=post.get_timestamp())

     

       251
       251
        
                       else:

     

       252
       252
       -
                           new_post = self.bsky.send_post(text, labels=labels, time_iso=post.get_post_date_iso())

     

       252
       252
       +
                           new_post = self.bsky.send_post(text, labels=labels, time_iso=post.get_timestamp())

     

       253
       253
        
                           root_ref = models.create_strong_ref(new_post)

     

       254
       254
        
                       

     

       255
       255
        
                       self.bsky.create_gates(

     

       256
       256
        
                           self.options.thread_gate, 

     

       257
       257
        
                           self.options.quote_gate, 

     

       258
       258
        
                           new_post.uri, 

     

       259
       259
       -
                           time_iso=post.get_post_date_iso()

     

       259
       259
       +
                           time_iso=post.get_timestamp()

     

       260
       260
        
                       )

     

       261
       261
        
                       reply_ref = models.create_strong_ref(new_post)

     

       262
       262
        
                       created_records.append(new_post)

     
···

       292
       292
        
                                   root=root_ref

     

       293
       293
        
                               ) if root_ref and reply_ref else None, 

     

       294
       294
        
                               labels=labels, 

     

       295
       295
       -
                               time_iso=post.get_post_date_iso()

     

       295
       295
       +
                               time_iso=post.get_timestamp()

     

       296
       296
        
                           )

     

       297
       297
        
                           if not root_ref:

     

       298
       298
        
                               root_ref = models.create_strong_ref(new_post)

     
···

       301
       301
        
                               self.options.thread_gate, 

     

       302
       302
        
                               self.options.quote_gate,

     

       303
       303
        
                               new_post.uri, 

     

       304
       304
       -
                               time_iso=post.get_post_date_iso()

     

       304
       304
       +
                               time_iso=post.get_timestamp()

     

       305
       305
        
                           )

     

       306
       306
        
                           reply_ref = models.create_strong_ref(new_post)

     

       307
       307
        
                           created_records.append(new_post)

     
···

       331
       331
        
                                   root=root_ref

     

       332
       332
        
                               ) if root_ref and reply_ref else None,

     

       333
       333
        
                               labels=labels,

     

       334
       334
       -
                               time_iso=post.get_post_date_iso()

     

       334
       334
       +
                               time_iso=post.get_timestamp()

     

       335
       335
        
                           )

     

       336
       336
        
                           if not root_ref:

     

       337
       337
        
                               root_ref = models.create_strong_ref(new_post)

     
···

       340
       340
        
                               self.options.thread_gate,

     

       341
       341
        
                               self.options.quote_gate, 

     

       342
       342
        
                               new_post.uri, 

     

       343
       343
       -
                               time_iso=post.get_post_date_iso()

     

       343
       343
       +
                               time_iso=post.get_timestamp()

     

       344
       344
        
                           )

     

       345
       345
        
                           reply_ref = models.create_strong_ref(new_post)

     

       346
       346
        
                           created_records.append(new_post)

+22 -19

cross.py

···

       1
       1
       +
       from abc import ABC, abstractmethod

     

       1
       2
        
       from typing import Callable, Any

     

       2
       3
        
       from util.database import DataBaseWorker

     

       3
       4
        
       from datetime import datetime, timezone

     
···

       52
       53
        
           def get_duration(self) -> float:

     

       53
       54
        
               return self.duration

     

       54
       55
        
           

     

       55
       55
       -
       class Post():

     

       56
       56
       -
           def __init__(self) -> None:

     

       57
       57
       -
               self.now_timestamp = datetime.now(timezone.utc).isoformat()

     

       56
       56
       +
       class Post(ABC):

     

       57
       57
       +
           @abstractmethod

     

       58
       58
       +
           def get_id(self) -> str:

     

       59
       59
       +
               return ''

     

       60
       60
       +
           

     

       61
       61
       +
           @abstractmethod

     

       62
       62
       +
           def get_parent_id(self) -> str | None:

     

       58
       63
        
               pass

     

       59
       64
        
           

     

       65
       65
       +
           @abstractmethod

     

       60
       66
        
           def get_tokens(self) -> list[Token]:

     

       61
       61
       -
               return []

     

       62
       62
       -
           

     

       63
       63
       -
           def get_parent_id(self) -> str | None:

     

       64
       64
       -
               return None

     

       67
       67
       +
               pass

     

       68
       68
       +
       

     

       69
       69
       +
           # returns input text type.

     

       70
       70
       +
           # text/plain, text/markdown, text/x.misskeymarkdown

     

       71
       71
       +
           @abstractmethod

     

       72
       72
       +
           def get_text_type(self) -> str:

     

       73
       73
       +
               pass

     

       65
       74
        
           

     

       66
       66
       -
           def get_post_date_iso(self) -> str:

     

       67
       67
       -
               return self.now_timestamp

     

       75
       75
       +
           # post iso timestamp

     

       76
       76
       +
           @abstractmethod

     

       77
       77
       +
           def get_timestamp(self) -> str:

     

       78
       78
       +
               pass

     

       68
       79
        
           

     

       69
       80
        
           def get_attachments(self) -> list[MediaInfo]:

     

       70
       81
        
               return []

     

       71
       82
        
           

     

       72
       72
       -
           def get_id(self) -> str:

     

       73
       73
       -
               return ''

     

       74
       74
       -
           

     

       75
       75
       -
           def get_cw(self) -> str:

     

       76
       76
       -
               return ''

     

       83
       83
       +
           def get_spoiler(self) -> str | None:

     

       84
       84
       +
               return None

     

       77
       85
        
           

     

       78
       86
        
           def get_languages(self) -> list[str]:

     

       79
       87
        
               return []

     

       80
       88
        
           

     

       81
       89
        
           def is_sensitive(self) -> bool:

     

       82
       90
        
               return False

     

       83
       83
       -
       

     

       84
       84
       -
           # returns input text type.

     

       85
       85
       -
           # text/plain, text/markdown, text/x.misskeymarkdown

     

       86
       86
       -
           def get_text_type(self) -> str:

     

       87
       87
       -
               return 'text/plain'

     

       88
       91
        
           

     

       89
       92
        
           def get_post_url(self) -> str | None:

     

       90
       93
        
               return None

+25 -19

mastodon/common.py

···

       4
       4
        
       class MastodonPost(cross.Post):

     

       5
       5
        
           def __init__(self, status: dict, tokens: list[cross.Token], media_attachments: list[MediaInfo]) -> None:

     

       6
       6
        
               super().__init__()

     

       7
       7
       -
               self.status = status

     

       8
       8
       -
               self.media_attachments = media_attachments

     

       7
       7
       +
               self.id = status['id']

     

       8
       8
       +
               self.parent_id = status.get('in_reply_to_id')

     

       9
       9
        
               self.tokens = tokens

     

       10
       10
        
               self.content_type = status.get('content_type', 'text/plain')

     

       11
       11
       +
               self.timestamp = status['created_at']

     

       12
       12
       +
               self.media_attachments = media_attachments

     

       13
       13
       +
               self.spoiler = status.get('spoiler_text')

     

       14
       14
       +
               self.language = [status['language']] if status.get('language') else []

     

       15
       15
       +
               self.sensitive = status.get('sensitive', False)

     

       16
       16
       +
               self.url = status.get('url')

     

       17
       17
       +
           

     

       18
       18
       +
           def get_id(self) -> str:

     

       19
       19
       +
               return self.id

     

       20
       20
       +
           

     

       21
       21
       +
           def get_parent_id(self) -> str | None:

     

       22
       22
       +
               return self.parent_id

     

       11
       23
        
           

     

       12
       24
        
           def get_tokens(self) -> list[cross.Token]:

     

       13
       25
        
               return self.tokens

     

       14
       26
        
           

     

       15
       15
       -
           def get_parent_id(self) -> str | None:

     

       16
       16
       -
               return self.status.get('in_reply_to_id')

     

       27
       27
       +
           def get_text_type(self) -> str:

     

       28
       28
       +
               return self.content_type

     

       17
       29
        
           

     

       18
       18
       -
           def get_post_date_iso(self) -> str:

     

       19
       19
       -
               return self.status.get('created_at') or self.now_timestamp

     

       30
       30
       +
           def get_timestamp(self) -> str:

     

       31
       31
       +
               return self.timestamp

     

       20
       32
        
           

     

       21
       21
       -
           def get_cw(self) -> str:

     

       22
       22
       -
               return self.status.get('spoiler_text') or ''

     

       33
       33
       +
           def get_attachments(self) -> list[MediaInfo]:

     

       34
       34
       +
               return self.media_attachments

     

       23
       35
        
           

     

       24
       24
       -
           def get_id(self) -> str:

     

       25
       25
       -
               return self.status['id']

     

       36
       36
       +
           def get_spoiler(self) -> str | None:

     

       37
       37
       +
               return self.spoiler

     

       26
       38
        
           

     

       27
       39
        
           def get_languages(self) -> list[str]:

     

       28
       28
       -
               return [self.status['language']] if self.status.get('language') else []

     

       40
       40
       +
               return self.language

     

       29
       41
        
           

     

       30
       42
        
           def is_sensitive(self) -> bool:

     

       31
       31
       -
               return self.status.get('sensitive', False)

     

       32
       32
       -
           

     

       33
       33
       -
           def get_attachments(self) -> list[MediaInfo]:

     

       34
       34
       -
               return self.media_attachments

     

       35
       35
       -
           

     

       36
       36
       -
           def get_text_type(self) -> str:

     

       37
       37
       -
               return self.content_type

     

       43
       43
       +
               return self.sensitive or (self.spoiler is not None)

     

       38
       44
        
           

     

       39
       45
        
           def get_post_url(self) -> str | None:

     

       40
       40
       -
               return self.status.get('url')
     

       46
       46
       +
               return self.url

+5 -4

mastodon/input.py

···

       167
       167
        
               database.delete_post(self.db, identifier, self.user_id, self.service)

     

       168
       168
        
           

     

       169
       169
        
           def _on_post(self, outputs: list[cross.Output], event: str, payload: str):

     

       170
       170
       -
               if event == 'update':

     

       171
       171
       -
                   self._on_create_post(outputs, json.loads(payload))

     

       172
       172
       -
               elif event == 'delete':

     

       173
       173
       -
                   self._on_delete_post(outputs, payload)          

     

       170
       170
       +
               match event:

     

       171
       171
       +
                   case 'update':

     

       172
       172
       +
                       self._on_create_post(outputs, json.loads(payload))

     

       173
       173
       +
                   case 'delete':

     

       174
       174
       +
                       self._on_delete_post(outputs, payload)

     

       174
       175
        
           

     

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

     

       176
       177
        
               uri = f"{self.streaming}/api/v1/streaming?stream=user&access_token={self.token}"

+2 -2

mastodon/output.py

···

       282
       282
        
                   payload = {

     

       283
       283
        
                       'status': status,

     

       284
       284
        
                       'media_ids': media or [],

     

       285
       285
       -
                       'spoiler_text': post.get_cw(),

     

       285
       285
       +
                       'spoiler_text': post.get_spoiler() or '',

     

       286
       286
        
                       'visibility': self.options.get('visibility', 'public'),

     

       287
       287
        
                       'content_type': self.text_format,

     

       288
       288
        
                       'language': lang

     
···

       291
       291
        
                   if media:

     

       292
       292
        
                       payload['sensitive'] = post.is_sensitive()

     

       293
       293
        
                       

     

       294
       294
       -
                       if post.get_cw():

     

       294
       294
       +
                       if post.get_spoiler():

     

       295
       295
        
                           payload['sensitive'] = True

     

       296
       296
        
                       

     

       297
       297
        
                       if not status:

+20 -17

misskey/common.py

···

       5
       5
        
           def __init__(self, instance_url: str, note: dict, tokens: list[cross.Token], files: list[MediaInfo]) -> None:

     

       6
       6
        
               super().__init__()

     

       7
       7
        
               self.note = note

     

       8
       8
       -
               self.sensitive = any([a.get('isSensitive', False) for a in note.get('files', [])])

     

       9
       9
       -
               self.media_attachments = files

     

       8
       8
       +
               self.id = note['id']

     

       9
       9
       +
               self.parent_id = note.get('replyId')

     

       10
       10
        
               self.tokens = tokens

     

       11
       11
       +
               self.timestamp = note['createdAt']

     

       12
       12
       +
               self.media_attachments = files

     

       13
       13
       +
               self.spoiler = note.get('cw')

     

       14
       14
       +
               self.sensitive = any([a.get('isSensitive', False) for a in note.get('files', [])])

     

       11
       15
        
               self.url = instance_url + '/notes/' + note['id']

     

       12
       16
        
           

     

       17
       17
       +
           def get_id(self) -> str:

     

       18
       18
       +
               return self.id

     

       19
       19
       +
           

     

       20
       20
       +
           def get_parent_id(self) -> str | None:

     

       21
       21
       +
               return self.parent_id

     

       22
       22
       +
           

     

       13
       23
        
           def get_tokens(self) -> list[cross.Token]:

     

       14
       24
        
               return self.tokens

     

       25
       25
       +
       

     

       26
       26
       +
           def get_text_type(self) -> str:

     

       27
       27
       +
               return "text/x.misskeymarkdown"

     

       15
       28
        
           

     

       16
       16
       -
           def get_parent_id(self) -> str | None:

     

       17
       17
       -
               return self.note.get('replyId')

     

       18
       18
       -
           

     

       19
       19
       -
           def get_post_date_iso(self) -> str:

     

       20
       20
       -
               date = self.note.get('createdAt')

     

       21
       21
       -
               return date or super().get_post_date_iso()

     

       29
       29
       +
           def get_timestamp(self) -> str:

     

       30
       30
       +
               return self.timestamp

     

       22
       31
        
           

     

       23
       32
        
           def get_attachments(self) -> list[MediaInfo]:

     

       24
       33
        
               return self.media_attachments

     

       25
       34
        
           

     

       26
       26
       -
           def get_id(self) -> str:

     

       27
       27
       -
               return self.note['id']

     

       28
       28
       -
           

     

       29
       29
       -
           def get_cw(self) -> str:

     

       30
       30
       -
               return self.note.get('cw') or ''

     

       35
       35
       +
           def get_spoiler(self) -> str | None:

     

       36
       36
       +
               return self.spoiler

     

       31
       37
        
           

     

       32
       38
        
           def get_languages(self) -> list[str]:

     

       33
       39
        
               return []

     

       34
       40
        
           

     

       35
       41
        
           def is_sensitive(self) -> bool:

     

       36
       36
       -
               return self.sensitive

     

       37
       37
       -
           

     

       38
       38
       -
           def get_text_type(self) -> str:

     

       39
       39
       -
               return "text/x.misskeymarkdown"

     

       42
       42
       +
               return self.sensitive or (self.spoiler is not None)

     

       40
       43
        
           

     

       41
       44
        
           def get_post_url(self) -> str | None:

     

       42
       45
        
               return self.url