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