social media crossposting tool. 3rd time's the charm
zenfyr.dev
mastodon
misskey
crossposting
bluesky
1from abc import ABC, abstractmethod
2from typing import Any, override
3
4from atproto.identity import did_resolver, handle_resolver
5from cross.service import Service
6from util.util import normalize_service_url
7
8SERVICE = "https://bsky.app"
9
10
11def validate_and_transform(data: dict[str, Any]):
12 if not data["handle"] and not data["did"]:
13 raise KeyError("no 'handle' or 'did' specified for bluesky input!")
14
15 if "did" in data:
16 did = str(data["did"]) # only did:web and did:plc are supported
17 if not did.startswith("did:plc:") and not did.startswith("did:web:"):
18 raise ValueError(f"Invalid handle {did}!")
19
20 if "pds" in data:
21 data["pds"] = normalize_service_url(data["pds"])
22
23
24class BlueskyService(ABC, Service):
25 pds: str
26 did: str
27
28 def _init_identity(self) -> None:
29 handle, did, pds = self.get_identity_options()
30
31 if did and pds:
32 self.did = did
33 self.pds = pds
34 return
35
36 if not did:
37 if not handle:
38 raise KeyError("No did: or atproto handle provided!")
39 self.log.info("Resolving ATP identity for %s...", handle)
40 self.did = handle_resolver.resolve_handle(handle)
41
42 if not pds:
43 self.log.info("Resolving PDS from %s DID document...", did)
44 atp_pds = did_resolver.resolve_did(self.did).get_atproto_pds()
45 if not atp_pds:
46 raise Exception("Failed to resolve atproto pds for %s")
47 self.pds = atp_pds
48
49 @abstractmethod
50 def get_identity_options(self) -> tuple[str | None, str | None, str | None]:
51 pass