social media crossposting tool. 3rd time's the charm
zenfyr.dev
mastodon
misskey
crossposting
bluesky
1from abc import ABC, abstractmethod
2from dataclasses import dataclass
3from typing import Any
4
5import requests
6
7from cross.service import Service
8from util.util import LOGGER
9
10
11@dataclass(kw_only=True)
12class InstanceInfo:
13 max_characters: int = 500
14 max_media_attachments: int = 4
15 characters_reserved_per_url: int = 23
16
17 image_size_limit: int = 16777216
18 video_size_limit: int = 103809024
19
20 @classmethod
21 def from_api(cls, data: dict[str, Any]) -> "InstanceInfo":
22 config: dict[str, Any] = {}
23
24 if "statuses" in data:
25 statuses_config: dict[str, Any] = data.get("statuses", {})
26 if "max_characters" in statuses_config:
27 config["max_characters"] = statuses_config["max_characters"]
28 if "max_media_attachments" in statuses_config:
29 config["max_media_attachments"] = statuses_config[
30 "max_media_attachments"
31 ]
32 if "characters_reserved_per_url" in statuses_config:
33 config["characters_reserved_per_url"] = statuses_config[
34 "characters_reserved_per_url"
35 ]
36
37 if "media_attachments" in data:
38 media_config: dict[str, Any] = data.get("media_attachments", {})
39 if "image_size_limit" in media_config:
40 config["image_size_limit"] = media_config["image_size_limit"]
41 if "video_size_limit" in media_config:
42 config["video_size_limit"] = media_config["video_size_limit"]
43
44 # *oma extensions
45 if "max_toot_chars" in data:
46 config["max_characters"] = data["max_toot_chars"]
47 if "upload_limit" in data:
48 config["image_size_limit"] = data["upload_limit"]
49 config["video_size_limit"] = data["upload_limit"]
50
51 return InstanceInfo(**config)
52
53
54class MastodonService(ABC, Service):
55 def verify_credentials(self):
56 token = self._get_token()
57 responce = requests.get(
58 f"{self.url}/api/v1/accounts/verify_credentials",
59 headers={"Authorization": f"Bearer {token}"},
60 )
61 if responce.status_code != 200:
62 LOGGER.error("Failed to validate user credentials!")
63 responce.raise_for_status()
64 return dict(responce.json())
65
66 def fetch_instance_info(self):
67 token = self._get_token()
68 responce = requests.get(
69 f"{self.url}/api/v1/instance",
70 headers={"Authorization": f"Bearer {token}"},
71 )
72 if responce.status_code != 200:
73 LOGGER.error("Failed to get instance info!")
74 responce.raise_for_status()
75 return dict(responce.json())
76
77 @abstractmethod
78 def _get_token(self) -> str:
79 pass