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