Retro Bulletin Board Systems on atproto. Web app and TUI.
lazy mirror of alyraffauf/atbbs
atbbs.xyz
forums
python
tui
atproto
bbs
tui: fix attachment downloads not saving file extension
+95
-25
+8
-16
tui/screens/thread.py
+8
-16
tui/screens/thread.py
···
1
1
import asyncio
2
-
from pathlib import Path
3
2
4
-
from platformdirs import user_downloads_dir
5
3
from textual import work
6
4
from textual.app import ComposeResult
7
5
from textual.binding import Binding
···
19
17
from core.slingshot import get_record, resolve_identity
20
18
from core.util import attachment_cid, blob_url
21
19
from tui.screens.compose import ComposeReplyScreen
22
-
from tui.util import ban_user, hide_post, require_session, require_sysop
20
+
from tui.util import (
21
+
ban_user,
22
+
download_blob,
23
+
hide_post,
24
+
require_session,
25
+
require_sysop,
26
+
)
23
27
from tui.widgets.breadcrumb import Breadcrumb
24
28
from tui.widgets.post import Post
25
29
···
302
306
303
307
@work(exclusive=True)
304
308
async def _do_save(self, post: Post) -> None:
305
-
downloads = Path(user_downloads_dir())
306
-
downloads.mkdir(parents=True, exist_ok=True)
307
-
308
309
client = self.app.http_client
309
310
for attachment in post.attachments:
310
311
name = attachment.get("name", "file")
···
314
315
315
316
url = blob_url(post.author_pds, post.author_did, cid)
316
317
try:
317
-
resp = await client.get(url)
318
-
resp.raise_for_status()
319
-
path = downloads / name
320
-
if path.exists():
321
-
stem, suffix = path.stem, path.suffix
322
-
i = 1
323
-
while path.exists():
324
-
path = downloads / f"{stem}_{i}{suffix}"
325
-
i += 1
326
-
path.write_bytes(resp.content)
318
+
path = await download_blob(client, url, name, downloads)
327
319
self.notify(f"Saved to {path}")
328
320
except Exception:
329
321
self.notify(f"Failed to download {name}.", severity="error")
+31
tui/util.py
+31
tui/util.py
···
1
1
"""TUI utilities."""
2
2
3
+
from pathlib import Path
4
+
5
+
import httpx
6
+
from platformdirs import user_downloads_dir
7
+
3
8
from core.auth.session import SessionStore
4
9
from core.models import AuthError, BBS
5
10
from core.records import create_ban_record, create_hidden_record
6
11
from core.resolver import invalidate_bbs_cache
12
+
13
+
14
+
def unique_path(path: Path) -> Path:
15
+
"""Return path, or path with a `_N` suffix if it already exists."""
16
+
if not path.exists():
17
+
return path
18
+
stem, suffix = path.stem, path.suffix
19
+
counter = 1
20
+
while True:
21
+
candidate = path.parent / f"{stem}_{counter}{suffix}"
22
+
if not candidate.exists():
23
+
return candidate
24
+
counter += 1
25
+
26
+
27
+
async def download_blob(
28
+
client: httpx.AsyncClient, url: str, filename: str
29
+
) -> Path:
30
+
"""Fetch a blob URL and save it to the user's Downloads folder."""
31
+
downloads = Path(user_downloads_dir())
32
+
downloads.mkdir(parents=True, exist_ok=True)
33
+
resp = await client.get(url)
34
+
resp.raise_for_status()
35
+
path = unique_path(downloads / filename)
36
+
path.write_bytes(resp.content)
37
+
return path
7
38
8
39
9
40
def require_session(screen) -> dict | None:
+56
-9
tui/widgets/post.py
+56
-9
tui/widgets/post.py
···
2
2
import webbrowser
3
3
from urllib.parse import unquote
4
4
5
+
from textual import work
5
6
from textual.app import ComposeResult
6
7
from textual.widget import Widget
7
8
from textual.widgets import Markdown, Static
···
12
13
blob_url,
13
14
format_datetime_local as format_datetime,
14
15
)
16
+
from tui.util import download_blob
15
17
16
18
ATTACHMENT_LINK_RE = re.compile(r"!?\[([^\]]*)\]\(attachment:([^)\s]+)\)")
17
19
ATTACHMENT_REF_RE = re.compile(r"attachment:([^)\s>\"']+)")
···
72
74
}
73
75
"""
74
76
75
-
def __init__(self, display: str, url: str, **kwargs) -> None:
77
+
def __init__(
78
+
self,
79
+
display: str,
80
+
url: str,
81
+
filename: str | None = None,
82
+
**kwargs,
83
+
) -> None:
76
84
super().__init__(display, markup=False, **kwargs)
77
85
self._url = url
86
+
self._filename = filename
78
87
79
88
def on_click(self) -> None:
80
-
webbrowser.open(self._url)
89
+
self._activate()
81
90
82
91
def key_enter(self) -> None:
83
-
webbrowser.open(self._url)
92
+
self._activate()
93
+
94
+
def _activate(self) -> None:
95
+
if self._filename:
96
+
self._save()
97
+
else:
98
+
webbrowser.open(self._url)
99
+
100
+
@work
101
+
async def _save(self) -> None:
102
+
try:
103
+
path = await download_blob(
104
+
self.app.http_client, self._url, self._filename
105
+
)
106
+
self.notify(f"Saved to {path}")
107
+
except Exception:
108
+
self.notify(
109
+
f"Failed to download {self._filename}.", severity="error"
110
+
)
84
111
85
112
86
113
class Post(Widget, can_focus=True):
···
163
190
self._body, self.attachments, self.author_pds, self.author_did
164
191
)
165
192
yield Markdown(body, classes="post-body")
193
+
194
+
downloadable, undownloadable = self._partition_attachments()
195
+
filename_by_url = {url: name for name, url in downloadable}
196
+
166
197
for number, (label, url) in enumerate(extract_body_links(body), 1):
167
-
yield AttachmentLink(f"[{number}] {label}", url)
198
+
yield AttachmentLink(
199
+
f"[{number}] {label}", url, filename=filename_by_url.get(url)
200
+
)
201
+
168
202
embedded = referenced_attachment_names(self._body)
203
+
for name, url in downloadable:
204
+
if name not in embedded:
205
+
yield AttachmentLink(f"[{name}]", url, filename=name)
206
+
for name in undownloadable:
207
+
if name not in embedded:
208
+
yield Static(f"[{name}]", classes="post-attachment", markup=False)
209
+
210
+
def _partition_attachments(
211
+
self,
212
+
) -> tuple[list[tuple[str, str]], list[str]]:
213
+
"""Split attachments into (name, url) we can fetch and names we can't."""
214
+
downloadable: list[tuple[str, str]] = []
215
+
undownloadable: list[str] = []
169
216
for attachment in self.attachments:
170
217
name = attachment.get("name", "file")
171
-
if name in embedded:
172
-
continue
173
218
cid = attachment_cid(attachment)
174
219
if cid and self.author_pds and self.author_did:
175
-
url = blob_url(self.author_pds, self.author_did, cid)
176
-
yield AttachmentLink(f"[{name}]", url)
220
+
downloadable.append(
221
+
(name, blob_url(self.author_pds, self.author_did, cid))
222
+
)
177
223
else:
178
-
yield Static(f"[{name}]", classes="post-attachment", markup=False)
224
+
undownloadable.append(name)
225
+
return downloadable, undownloadable