🐝 Tiny CLI to post simultaneously to Mastodon and Bluesky
Uses `aspectRatio` when posting images to ATProto
Closes #3
+111
-39
+17
-6
not_my_ex/bluesky.py
+17
-6
not_my_ex/bluesky.py
···
58
58
if not self.is_authenticated:
59
59
await self.auth()
60
60
61
-
resp = await self.xrpc(
62
-
"com.atproto.repo.uploadBlob",
63
-
headers={"Content-type": media.mime},
64
-
data=media.content,
61
+
resp, dimensions = await gather(
62
+
self.xrpc(
63
+
"com.atproto.repo.uploadBlob",
64
+
headers={"Content-type": media.mime},
65
+
data=media.content,
66
+
),
67
+
media.dimensions(),
65
68
)
66
69
if resp.status_code != 200:
67
70
self.raise_from(resp)
68
71
69
-
data = resp.json()
70
-
return {"alt": media.alt or "", "image": data["blob"]}
72
+
embed = {"alt": media.alt or "", "image": resp.json().get("blob")}
73
+
if dimensions:
74
+
width, height = dimensions
75
+
embed["aspectRatio"] = {
76
+
"$type": "app.bsky.embed.defs",
77
+
"width": width,
78
+
"height": height,
79
+
}
80
+
81
+
return embed
71
82
72
83
async def data(self, post):
73
84
if not self.is_authenticated:
+55
not_my_ex/media.py
+55
not_my_ex/media.py
···
40
40
while self.path and not self.alt:
41
41
alt = input(f"Enter an alt text for {self.path}: ")
42
42
self.alt = alt.strip() or None
43
+
44
+
async def dimensions(self) -> tuple[int, int] | None:
45
+
if not self.path:
46
+
return None
47
+
48
+
if self.mime == "image/png":
49
+
async with open(self.path, "rb") as reader:
50
+
await reader.seek(16)
51
+
width_bytes = await reader.read(4)
52
+
height_bytes = await reader.read(4)
53
+
if len(width_bytes) == 4 and len(height_bytes) == 4:
54
+
width = int.from_bytes(width_bytes, "big")
55
+
height = int.from_bytes(height_bytes, "big")
56
+
return width, height
57
+
58
+
if self.mime == "image/jpeg":
59
+
async with open(self.path, "rb") as reader:
60
+
while True:
61
+
marker = await reader.read(2)
62
+
if not marker or len(marker) < 2:
63
+
break
64
+
65
+
if marker == b"\xff\xd8":
66
+
continue
67
+
68
+
if marker in (b"\xff\xc0", b"\xff\xc2"):
69
+
length_bytes = await reader.read(2)
70
+
if not length_bytes or len(length_bytes) < 2:
71
+
break
72
+
73
+
length = int.from_bytes(length_bytes, "big") - 2
74
+
if length < 0:
75
+
break
76
+
77
+
await reader.read(1)
78
+
height_bytes = await reader.read(2)
79
+
width_bytes = await reader.read(2)
80
+
if len(height_bytes) == 2 and len(width_bytes) == 2:
81
+
height = int.from_bytes(height_bytes, "big")
82
+
width = int.from_bytes(width_bytes, "big")
83
+
return width, height
84
+
break
85
+
86
+
else:
87
+
length_bytes = await reader.read(2)
88
+
if not length_bytes or len(length_bytes) < 2:
89
+
break
90
+
91
+
length = int.from_bytes(length_bytes, "big") - 2
92
+
if length < 0:
93
+
break
94
+
95
+
await reader.seek(length, 1)
96
+
97
+
return None
-16
tests/conftest.py
-16
tests/conftest.py
···
1
-
from base64 import b64decode
2
1
from os import environ
3
-
from pathlib import Path
4
-
from tempfile import NamedTemporaryFile
5
2
6
3
from pytest import fixture
7
4
8
5
from not_my_ex.auth import EnvAuth
9
6
10
-
ONE_PIXEL_IMAGE = (
11
-
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAAC0lEQVQIW2NgAAIAAAUAAR4"
12
-
"f7BQAAAAASUVORK5CYII="
13
-
)
14
7
SETTINGS = {
15
8
"BSKY_EMAIL": "python@mailinator.com",
16
9
"BSKY_PASSWORD": "forty2",
···
22
15
for key, value in SETTINGS.items():
23
16
environ[f"NOT_MY_EX_{key}"] = value
24
17
return config
25
-
26
-
27
-
@fixture
28
-
def image():
29
-
with NamedTemporaryFile(suffix=".png") as tmp:
30
-
content = b64decode(ONE_PIXEL_IMAGE)
31
-
path = Path(tmp.name)
32
-
path.write_bytes(content)
33
-
yield (path, content)
34
18
35
19
36
20
@fixture
tests/image.jpeg
tests/image.jpeg
This is a binary file and will not be displayed.
tests/image.png
tests/image.png
This is a binary file and will not be displayed.
+15
-3
tests/test_bluesky.py
+15
-3
tests/test_bluesky.py
···
1
+
from pathlib import Path
1
2
from unittest.mock import ANY, AsyncMock, Mock
2
3
3
4
from pytest import mark, raises
···
163
164
164
165
@mark.asyncio
165
166
async def test_bluesky_client_post_data_includes_images_blobs(auth):
167
+
png = Path(__file__).parent / "image.png"
166
168
client, response = AsyncMock(), Mock()
167
169
response.status_code = 200
168
170
response.json.return_value = {"blob": "42"}
···
171
173
bluesky.is_authenticated = True
172
174
bluesky.token = "token"
173
175
bluesky.did = "did"
174
-
media = Media("/tmp/42.png", b"42", "image/png", "my alt text")
176
+
media = await Media.from_img(str(png), "my alt text")
175
177
post = Post("hi", media=(media,))
176
178
data = await bluesky.data(post)
177
179
client.post.assert_any_call(
···
180
182
"Authorization": "Bearer token",
181
183
"Content-type": "image/png",
182
184
},
183
-
data=b"42",
185
+
data=png.read_bytes(),
184
186
)
185
187
assert data["record"]["embed"] == {
186
188
"$type": "app.bsky.embed.images",
187
-
"images": [{"alt": "my alt text", "image": "42"}],
189
+
"images": [
190
+
{
191
+
"image": "42",
192
+
"alt": "my alt text",
193
+
"aspectRatio": {
194
+
"$type": "app.bsky.embed.defs",
195
+
"width": 1,
196
+
"height": 2,
197
+
},
198
+
}
199
+
],
188
200
}
+17
-8
tests/test_media.py
+17
-8
tests/test_media.py
···
1
+
from pathlib import Path
1
2
from unittest.mock import patch
2
3
3
4
from pytest import mark, raises
···
6
7
7
8
8
9
@mark.asyncio
9
-
async def test_media_from_img(image):
10
-
path, content = image
11
-
media = await Media.from_img(path)
12
-
assert content == media.content
13
-
assert media.mime == "image/png"
10
+
@mark.parametrize("ext,width,height", (("png", 1, 2), ("jpeg", 2, 3)))
11
+
async def test_media_from_img(ext, width, height):
12
+
img = Path(__file__).parent / f"image.{ext}"
13
+
media = await Media.from_img(str(img))
14
+
assert media.content == img.read_bytes()
15
+
assert media.mime == f"image/{ext}"
16
+
assert await media.dimensions() == (width, height)
14
17
15
18
16
19
@mark.asyncio
···
20
23
21
24
22
25
@mark.asyncio
23
-
async def test_media_from_img_without_mime_type(image):
24
-
path, _ = image
26
+
async def test_media_from_img_without_mime_type():
27
+
img = Path(__file__).parent / "image.png"
25
28
with patch("not_my_ex.media.mime_for") as guess:
26
29
guess.return_value = None
27
30
with raises(ValueError):
28
-
await Media.from_img(path)
31
+
await Media.from_img(str(img))
29
32
30
33
31
34
def test_media_check_alt_text_with_existing_alt_text():
···
41
44
mock.return_value = "alt"
42
45
media.check_alt_text()
43
46
assert media.alt == "alt"
47
+
48
+
49
+
@mark.asyncio
50
+
async def test_media_dimensions_for_not_an_image():
51
+
media = await Media.from_img(__file__)
52
+
assert await media.dimensions() is None
+7
-6
tests/test_post.py
+7
-6
tests/test_post.py
···
1
+
from pathlib import Path
1
2
from unittest.mock import patch
2
3
3
4
from pytest import mark, raises
···
14
15
15
16
16
17
@mark.asyncio
17
-
async def test_post_with_media(image):
18
-
img, *_ = image
19
-
media = await Media.from_img(img)
18
+
async def test_post_with_media():
19
+
img = Path(__file__).parent / "image.png"
20
+
media = await Media.from_img(str(img))
20
21
post = Post("forty-two", media=(media,))
21
22
assert len(post.media) == 1
22
23
···
61
62
("", False, True),
62
63
),
63
64
)
64
-
async def test_post_is_empty(image, text, with_image, expected):
65
+
async def test_post_is_empty(text, with_image, expected):
66
+
img = Path(__file__).parent / "image.png"
65
67
kwargs = {"media": []}
66
68
if with_image:
67
-
img, *_ = image
68
-
media = await Media.from_img(img)
69
+
media = await Media.from_img(str(img))
69
70
kwargs["media"].append(media)
70
71
71
72
post = Post(text, **kwargs)