commit c956a17faccf011f6c0dde4f81eed02869dfbe2a · zenfyr.dev/xpost

+31 -6

mastodon/info.py

···

       7
       7
        
       from cross.service import Service

     

       8
       8
        
       from util.util import normalize_service_url

     

       9
       9
        
       

     

       10
       10
       +
       

     

       10
       11
        
       def validate_and_transform(data: dict[str, Any]):

     

       11
       11
       -
           if 'token' not in data or 'instance' not in data:

     

       12
       12
       +
           if "token" not in data or "instance" not in data:

     

       12
       13
        
               raise KeyError("Missing required values 'token' or 'instance'")

     

       13
       14
        
       

     

       14
       15
        
           data["instance"] = normalize_service_url(data["instance"])

     

       16
       16
       +
       

     

       15
       17
        
       

     

       16
       18
        
       @dataclass(kw_only=True)

     

       17
       19
        
       class InstanceInfo:

     
···

       22
       24
        
           image_size_limit: int = 16777216

     

       23
       25
        
           video_size_limit: int = 103809024

     

       24
       26
        
       

     

       27
       27
       +
           text_format: str = "text/plain"

     

       28
       28
       +
       

     

       25
       29
        
           @classmethod

     

       26
       30
        
           def from_api(cls, data: dict[str, Any]) -> "InstanceInfo":

     

       27
       31
        
               config: dict[str, Any] = {}

     
···

       39
       43
        
                           "characters_reserved_per_url"

     

       40
       44
        
                       ]

     

       41
       45
        
       

     

       46
       46
       +
                   # glitch content type

     

       47
       47
       +
                   if "supported_mime_types" in statuses_config:

     

       48
       48
       +
                       text_mimes: list[str] = statuses_config["supported_mime_types"]

     

       49
       49
       +
       

     

       50
       50
       +
                       if "text/x.misskeymarkdown" in text_mimes:

     

       51
       51
       +
                           config["text_format"] = "text/x.misskeymarkdown"

     

       52
       52
       +
                       elif "text/markdown" in text_mimes:

     

       53
       53
       +
                           config["text_format"] = "text/markdown"

     

       54
       54
       +
       

     

       42
       55
        
               if "media_attachments" in data:

     

       43
       43
       -
                   media_config: dict[str, Any] = data.get("media_attachments", {})

     

       56
       56
       +
                   media_config: dict[str, Any] = data["media_attachments"]

     

       44
       57
        
                   if "image_size_limit" in media_config:

     

       45
       58
        
                       config["image_size_limit"] = media_config["image_size_limit"]

     

       46
       59
        
                   if "video_size_limit" in media_config:

     
···

       53
       66
        
                   config["image_size_limit"] = data["upload_limit"]

     

       54
       67
        
                   config["video_size_limit"] = data["upload_limit"]

     

       55
       68
        
       

     

       69
       69
       +
               if "pleroma" in data:

     

       70
       70
       +
                   pleroma: dict[str, Any] = data["pleroma"]

     

       71
       71
       +
                   if "metadata" in pleroma:

     

       72
       72
       +
                       metadata: dict[str, Any] = pleroma["metadata"]

     

       73
       73
       +
                       if "post_formats" in metadata:

     

       74
       74
       +
                           post_formats: list[str] = metadata["post_formats"]

     

       75
       75
       +
       

     

       76
       76
       +
                           if "text/x.misskeymarkdown" in post_formats:

     

       77
       77
       +
                               config["text_format"] = "text/x.misskeymarkdown"

     

       78
       78
       +
                           elif "text/markdown" in post_formats:

     

       79
       79
       +
                               config["text_format"] = "text/markdown"

     

       80
       80
       +
       

     

       56
       81
        
               return InstanceInfo(**config)

     

       57
       82
        
       

     

       58
       83
        
       

     

       59
       84
        
       class MastodonService(ABC, Service):

     

       60
       85
        
           def verify_credentials(self):

     

       61
       86
        
               token = self._get_token()

     

       62
       62
       -
               responce = requests.get(

     

       87
       87
       +
               response = requests.get(

     

       63
       88
        
                   f"{self.url}/api/v1/accounts/verify_credentials",

     

       64
       89
        
                   headers={"Authorization": f"Bearer {token}"},

     

       65
       90
        
               )

     

       66
       66
       -
               if responce.status_code != 200:

     

       91
       91
       +
               if response.status_code != 200:

     

       67
       92
        
                   self.log.error("Failed to validate user credentials!")

     

       68
       68
       -
                   responce.raise_for_status()

     

       69
       69
       -
               return dict(responce.json())

     

       93
       93
       +
                   response.raise_for_status()

     

       94
       94
       +
               return dict(response.json())

     

       70
       95
        
       

     

       71
       96
        
           def fetch_instance_info(self):

     

       72
       97
        
               token = self._get_token()

+4 -4

mastodon/input.py

···

       54
       54
        
               self.options: MastodonInputOptions = options

     

       55
       55
        
       

     

       56
       56
        
               self.log.info("Verifying %s credentails...", self.url)

     

       57
       57
       -
               responce = self.verify_credentials()

     

       58
       58
       -
               self.user_id: str = responce["id"]

     

       57
       57
       +
               response = self.verify_credentials()

     

       58
       58
       +
               self.user_id: str = response["id"]

     

       59
       59
        
       

     

       60
       60
        
               self.log.info("Getting %s configuration...", self.url)

     

       61
       61
       -
               responce = self.fetch_instance_info()

     

       62
       62
       -
               self.streaming_url: str = responce["urls"]["streaming_api"]

     

       61
       61
       +
               response = self.fetch_instance_info()

     

       62
       62
       +
               self.streaming_url: str = response["urls"]["streaming_api"]

     

       63
       63
        
       

     

       64
       64
        
           @override

     

       65
       65
        
           def _get_token(self) -> str:

+4 -4

mastodon/output.py

···

       32
       32
        
               self.options: MastodonOutputOptions = options

     

       33
       33
        
       

     

       34
       34
        
               self.log.info("Verifying %s credentails...", self.url)

     

       35
       35
       -
               responce = self.verify_credentials()

     

       36
       36
       -
               self.user_id: str = responce["id"]

     

       35
       35
       +
               response = self.verify_credentials()

     

       36
       36
       +
               self.user_id: str = response["id"]

     

       37
       37
        
       

     

       38
       38
        
               self.log.info("Getting %s configuration...", self.url)

     

       39
       39
       -
               responce = self.fetch_instance_info()

     

       40
       40
       -
               self.instance_info: InstanceInfo = InstanceInfo.from_api(responce)

     

       39
       39
       +
               response = self.fetch_instance_info()

     

       40
       40
       +
               self.instance_info: InstanceInfo = InstanceInfo.from_api(response)

     

       41
       41
        
       

     

       42
       42
        
           @override

     

       43
       43
        
           def _get_token(self) -> str:

+4 -4

misskey/info.py

···

       7
       7
        
       

     

       8
       8
        
       class MisskeyService(ABC, Service):

     

       9
       9
        
           def verify_credentials(self):

     

       10
       10
       -
               responce = requests.post(

     

       10
       10
       +
               response = requests.post(

     

       11
       11
        
                   f"{self.url}/api/i",

     

       12
       12
        
                   json={"i": self._get_token()},

     

       13
       13
        
                   headers={"Content-Type": "application/json"},

     

       14
       14
        
               )

     

       15
       15
       -
               if responce.status_code != 200:

     

       15
       15
       +
               if response.status_code != 200:

     

       16
       16
        
                   self.log.error("Failed to validate user credentials!")

     

       17
       17
       -
                   responce.raise_for_status()

     

       18
       18
       -
               return dict(responce.json())

     

       17
       17
       +
                   response.raise_for_status()

     

       18
       18
       +
               return dict(response.json())

     

       19
       19
        
       

     

       20
       20
        
           @abstractmethod

     

       21
       21
        
           def _get_token(self) -> str:

+2 -2

misskey/input.py

···

       55
       55
        
               self.options: MisskeyInputOptions = options

     

       56
       56
        
       

     

       57
       57
        
               self.log.info("Verifying %s credentails...", self.url)

     

       58
       58
       -
               responce = self.verify_credentials()

     

       59
       59
       -
               self.user_id: str = responce["id"]

     

       58
       58
       +
               response = self.verify_credentials()

     

       59
       59
       +
               self.user_id: str = response["id"]

     

       60
       60
        
       

     

       61
       61
        
           @override

     

       62
       62
        
           def _get_token(self) -> str: