social media crossposting tool. 3rd time's the charm
zenfyr.dev
mastodon
misskey
crossposting
bluesky
1from typing import Any
2
3from atproto import AtUri, Client, IdResolver, client_utils
4from atproto_client import models
5
6from util.util import LOGGER
7
8
9def resolve_identity(
10 handle: str | None = None, did: str | None = None, pds: str | None = None
11):
12 """helper to try and resolve identity from provided parameters, a valid handle is enough"""
13
14 if did and pds:
15 return did, pds[:-1] if pds.endswith("/") else pds
16
17 resolver = IdResolver()
18 if not did:
19 if not handle:
20 raise Exception("ATP handle not specified!")
21 LOGGER.info("Resolving ATP identity for %s...", handle)
22 did = resolver.handle.resolve(handle)
23 if not did:
24 raise Exception("Failed to resolve DID!")
25
26 if not pds:
27 LOGGER.info("Resolving PDS from DID document...")
28 did_doc = resolver.did.resolve(did)
29 if not did_doc:
30 raise Exception("Failed to resolve DID doc for '%s'", did)
31 pds = did_doc.get_pds_endpoint()
32 if not pds:
33 raise Exception("Failed to resolve PDS!")
34
35 return did, pds[:-1] if pds.endswith("/") else pds
36
37
38class Client2(Client):
39 def __init__(self, base_url: str | None = None, *args: Any, **kwargs: Any) -> None:
40 super().__init__(base_url, *args, **kwargs)
41
42 def send_video(
43 self,
44 text: str | client_utils.TextBuilder,
45 video: bytes,
46 video_alt: str | None = None,
47 video_aspect_ratio: models.AppBskyEmbedDefs.AspectRatio | None = None,
48 reply_to: models.AppBskyFeedPost.ReplyRef | None = None,
49 langs: list[str] | None = None,
50 facets: list[models.AppBskyRichtextFacet.Main] | None = None,
51 labels: models.ComAtprotoLabelDefs.SelfLabels | None = None,
52 time_iso: str | None = None,
53 ) -> models.AppBskyFeedPost.CreateRecordResponse:
54 """same as send_video, but with labels"""
55
56 if video_alt is None:
57 video_alt = ""
58
59 upload = self.upload_blob(video)
60
61 return self.send_post(
62 text,
63 reply_to=reply_to,
64 embed=models.AppBskyEmbedVideo.Main(
65 video=upload.blob, alt=video_alt, aspect_ratio=video_aspect_ratio
66 ),
67 langs=langs,
68 facets=facets,
69 labels=labels,
70 time_iso=time_iso,
71 )
72
73 def send_images(
74 self,
75 text: str | client_utils.TextBuilder,
76 images: list[bytes],
77 image_alts: list[str] | None = None,
78 image_aspect_ratios: list[models.AppBskyEmbedDefs.AspectRatio] | None = None,
79 reply_to: models.AppBskyFeedPost.ReplyRef | None = None,
80 langs: list[str] | None = None,
81 facets: list[models.AppBskyRichtextFacet.Main] | None = None,
82 labels: models.ComAtprotoLabelDefs.SelfLabels | None = None,
83 time_iso: str | None = None,
84 ) -> models.AppBskyFeedPost.CreateRecordResponse:
85 """same as send_images, but with labels"""
86
87 if image_alts is None:
88 image_alts = [""] * len(images)
89 else:
90 diff = len(images) - len(image_alts)
91 image_alts = image_alts + [""] * diff
92
93 if image_aspect_ratios is None:
94 aligned_image_aspect_ratios = [None] * len(images)
95 else:
96 diff = len(images) - len(image_aspect_ratios)
97 aligned_image_aspect_ratios = image_aspect_ratios + [None] * diff
98
99 uploads = [self.upload_blob(image) for image in images]
100
101 embed_images = [
102 models.AppBskyEmbedImages.Image(
103 alt=alt, image=upload.blob, aspect_ratio=aspect_ratio
104 )
105 for alt, upload, aspect_ratio in zip(
106 image_alts, uploads, aligned_image_aspect_ratios
107 )
108 ]
109
110 return self.send_post(
111 text,
112 reply_to=reply_to,
113 embed=models.AppBskyEmbedImages.Main(images=embed_images),
114 langs=langs,
115 facets=facets,
116 labels=labels,
117 time_iso=time_iso,
118 )
119
120 def send_post(
121 self,
122 text: str | client_utils.TextBuilder,
123 reply_to: models.AppBskyFeedPost.ReplyRef | None = None,
124 embed: None
125 | models.AppBskyEmbedImages.Main
126 | models.AppBskyEmbedExternal.Main
127 | models.AppBskyEmbedRecord.Main
128 | models.AppBskyEmbedRecordWithMedia.Main
129 | models.AppBskyEmbedVideo.Main = None,
130 langs: list[str] | None = None,
131 facets: list[models.AppBskyRichtextFacet.Main] | None = None,
132 labels: models.ComAtprotoLabelDefs.SelfLabels | None = None,
133 time_iso: str | None = None,
134 ) -> models.AppBskyFeedPost.CreateRecordResponse:
135 """same as send_post, but with labels"""
136
137 if isinstance(text, client_utils.TextBuilder):
138 facets = text.build_facets()
139 text = text.build_text()
140
141 repo = self.me and self.me.did
142 if not repo:
143 raise Exception("Client not logged in!")
144
145 if not langs:
146 langs = ["en"]
147
148 record = models.AppBskyFeedPost.Record(
149 created_at=time_iso or self.get_current_time_iso(),
150 text=text,
151 reply=reply_to or None,
152 embed=embed or None,
153 langs=langs,
154 facets=facets or None,
155 labels=labels or None,
156 )
157 return self.app.bsky.feed.post.create(repo, record)
158
159 def create_gates(
160 self,
161 thread_gate_opts: list[str],
162 quote_gate: bool,
163 post_uri: str,
164 time_iso: str | None = None,
165 ):
166 account = self.me
167 if not account:
168 raise Exception("Client not logged in!")
169
170 rkey = AtUri.from_str(post_uri).rkey
171 time_iso = time_iso or self.get_current_time_iso()
172
173 if "everybody" not in thread_gate_opts:
174 allow = []
175 if thread_gate_opts:
176 if "following" in thread_gate_opts:
177 allow.append(models.AppBskyFeedThreadgate.FollowingRule())
178 if "followers" in thread_gate_opts:
179 allow.append(models.AppBskyFeedThreadgate.FollowerRule())
180 if "mentioned" in thread_gate_opts:
181 allow.append(models.AppBskyFeedThreadgate.MentionRule())
182
183 thread_gate = models.AppBskyFeedThreadgate.Record(
184 post=post_uri, created_at=time_iso, allow=allow
185 )
186
187 self.app.bsky.feed.threadgate.create(account.did, thread_gate, rkey)
188
189 if quote_gate:
190 post_gate = models.AppBskyFeedPostgate.Record(
191 post=post_uri,
192 created_at=time_iso,
193 embedding_rules=[models.AppBskyFeedPostgate.DisableRule()],
194 )
195
196 self.app.bsky.feed.postgate.create(account.did, post_gate, rkey)