zenfyr.dev
social media crossposting tool. 3rd time's the charm
mastodon
misskey
crossposting
bluesky
add quotes to bsky
+34
-33
+3
-4
bluesky/info.py
+3
-4
bluesky/info.py
···
27
27
28
28
def _init_identity(self) -> None:
29
29
handle, did, pds = self.get_identity_options()
30
-
31
-
if did and pds:
30
+
if did:
32
31
self.did = did
32
+
if pds:
33
33
self.pds = pds
34
-
return
35
34
36
35
if not did:
37
36
if not handle:
···
40
39
self.did = handle_resolver.resolve_handle(handle)
41
40
42
41
if not pds:
43
-
self.log.info("Resolving PDS from %s DID document...", did)
42
+
self.log.info("Resolving PDS from %s DID document...", self.did)
44
43
atp_pds = did_resolver.resolve_did(self.did).get_atproto_pds()
45
44
if not atp_pds:
46
45
raise Exception("Failed to resolve atproto pds for %s")
+31
-29
bluesky/input.py
+31
-29
bluesky/input.py
···
13
13
LabelsAttachment,
14
14
LanguagesAttachment,
15
15
MediaAttachment,
16
+
QuoteAttachment,
16
17
RemoteUrlAttachment,
17
18
)
18
19
from cross.media import Blob, download_blob
···
81
82
RemoteUrlAttachment(url=f"https://bsky.app/profile/{did}/post/{rid}")
82
83
)
83
84
84
-
embed = record.get("embed", {})
85
-
if embed:
85
+
embed: dict[str, Any] = record.get("embed", {})
86
+
blob_urls: list[tuple[str, str, str | None]] = []
87
+
def handle_embeds(embed: dict[str, Any]):
88
+
nonlocal blob_urls, post
86
89
match cast(str, embed["$type"]):
87
90
case "app.bsky.embed.record" | "app.bsky.embed.recordWithMedia":
88
-
_, collection, _ = AtUri.record_uri(
89
-
cast(str, embed["record"]["uri"])
90
-
)
91
-
if collection == "app.bsky.feed.post":
92
-
self.log.info("Skipping '%s'! Quote..", post_uri)
91
+
rcrd = embed['record']['record'] if embed['record'].get('record') else embed['record']
92
+
did, collection, _ = AtUri.record_uri(rcrd["uri"])
93
+
if collection != "app.bsky.feed.post":
93
94
return
95
+
if did != self.did:
96
+
return
97
+
post.attachments.put(QuoteAttachment(quoted_id=rcrd["uri"], quoted_user=did))
98
+
99
+
if embed.get('media'):
100
+
handle_embeds(embed["media"])
94
101
case "app.bsky.embed.images":
95
-
blobs: list[Blob] = []
96
102
for image in embed["images"]:
97
103
blob_cid = image["image"]["ref"]["$link"]
98
104
url = f"{self.pds}/xrpc/com.atproto.sync.getBlob?did={self.did}&cid={blob_cid}"
99
-
self.log.info("Downloading %s...", blob_cid)
100
-
blob: Blob | None = download_blob(url, image.get("alt"))
101
-
if not blob:
102
-
self.log.error(
103
-
"Skipping %s! Failed to download blob %s.",
104
-
post_uri,
105
-
blob_cid,
106
-
)
107
-
return
108
-
blobs.append(blob)
109
-
post.attachments.put(MediaAttachment(blobs=blobs))
105
+
blob_urls.append((url, blob_cid, image.get("alt")))
110
106
case "app.bsky.embed.video":
111
107
blob_cid = embed["video"]["ref"]["$link"]
112
108
url = f"{self.pds}/xrpc/com.atproto.sync.getBlob?did={self.did}&cid={blob_cid}"
113
-
self.log.info("Downloading %s...", blob_cid)
114
-
blob: Blob | None = download_blob(url, embed.get("alt"))
115
-
if not blob:
116
-
self.log.error(
117
-
"Skipping %s! Failed to download blob %s.",
118
-
post_uri,
119
-
blob_cid,
120
-
)
121
-
return
122
-
post.attachments.put(MediaAttachment(blobs=[blob]))
109
+
blob_urls.append((url, blob_cid, embed.get("alt")))
123
110
case _:
124
111
self.log.warning(f"Unhandled embedd type {embed['$type']}")
125
112
pass
113
+
if embed:
114
+
handle_embeds(embed)
115
+
116
+
if blob_urls:
117
+
blobs: list[Blob] = []
118
+
for url, cid, alt in blob_urls:
119
+
self.log.info("Downloading %s...", cid)
120
+
blob: Blob | None = download_blob(url, alt)
121
+
if not blob:
122
+
self.log.error(
123
+
"Skipping %s! Failed to download blob %s.", post_uri, cid
124
+
)
125
+
return
126
+
blobs.append(blob)
127
+
post.attachments.put(MediaAttachment(blobs=blobs))
126
128
127
129
if "langs" in record:
128
130
post.attachments.put(LanguagesAttachment(langs=record["langs"]))