Retro Bulletin Board Systems on atproto. Web app and TUI.
lazy mirror of alyraffauf/atbbs
atbbs.xyz
forums
python
tui
atproto
bbs
tui: add ban/hide helpers
+77
-80
+2
-4
tui/screens/news.py
+2
-4
tui/screens/news.py
···
7
7
from core import lexicon
8
8
from core.models import AtUri, AuthError, BBS, Post as PostModel
9
9
from core.records import delete_record
10
-
from tui.util import make_session_updater
10
+
from tui.util import make_session_updater, require_sysop
11
11
from tui.widgets.breadcrumb import Breadcrumb
12
12
from tui.widgets.post import Post
13
13
···
43
43
yield Footer()
44
44
45
45
def action_delete(self) -> None:
46
-
session = self.app.user_session
47
-
if not session or session["did"] != self.bbs.identity.did:
48
-
self.notify("Only the sysop can delete news.", severity="error")
46
+
if not require_sysop(self, self.bbs):
49
47
return
50
48
self._do_delete()
51
49
+3
-11
tui/screens/site.py
+3
-11
tui/screens/site.py
···
11
11
from tui.screens.compose import ComposeNewsScreen
12
12
from tui.screens.news import NewsScreen
13
13
from tui.screens.sysop import SysopScreen
14
-
from tui.util import require_session
14
+
from tui.util import require_sysop
15
15
from tui.widgets.breadcrumb import Breadcrumb
16
16
17
17
···
79
79
self.notify("Could not refresh.", severity="error")
80
80
81
81
def action_sysop(self) -> None:
82
-
session = require_session(self)
83
-
if not session:
84
-
return
85
-
if session["did"] != self.bbs.identity.did:
86
-
self.notify("Only the sysop can manage this BBS.", severity="error")
82
+
if not require_sysop(self, self.bbs):
87
83
return
88
84
self.app.push_screen(SysopScreen(self.bbs, self.handle))
89
85
90
86
def action_new_news(self) -> None:
91
-
session = require_session(self)
92
-
if not session:
93
-
return
94
-
if session["did"] != self.bbs.identity.did:
95
-
self.notify("Only the sysop can post news.", severity="error")
87
+
if not require_sysop(self, self.bbs):
96
88
return
97
89
self.app.push_screen(ComposeNewsScreen(self.bbs, self.handle))
98
90
+5
-31
tui/screens/sysop/moderate.py
+5
-31
tui/screens/sysop/moderate.py
···
8
8
9
9
from core import lexicon
10
10
from core.models import AtUri, AuthError, BBS
11
-
from core.records import (
12
-
create_ban_record,
13
-
create_hidden_record,
14
-
delete_record,
15
-
list_pds_records,
16
-
)
11
+
from core.records import delete_record, list_pds_records
17
12
from core.resolver import invalidate_bbs_cache
18
13
from core.slingshot import resolve_identities_batch, resolve_identity
19
-
from tui.util import make_session_updater
14
+
from tui.util import ban_user, hide_post, make_session_updater
20
15
from tui.widgets.breadcrumb import Breadcrumb
21
16
22
17
···
154
149
155
150
@work
156
151
async def _do_add_ban(self, identifier: str) -> None:
157
-
session = self.app.user_session
158
-
client = self.app.http_client
159
-
updater = make_session_updater(self.app.session_store)
160
-
161
152
did = identifier
162
153
if not identifier.startswith("did:"):
163
154
try:
164
-
identity = await resolve_identity(client, identifier)
155
+
identity = await resolve_identity(self.app.http_client, identifier)
165
156
did = identity.did
166
157
except Exception:
167
158
self.notify(f"Could not resolve {identifier}.", severity="error")
···
171
162
self.notify("Already banned.", severity="warning")
172
163
return
173
164
174
-
try:
175
-
await create_ban_record(client, session, did, updater)
176
-
invalidate_bbs_cache()
177
-
self.notify(f"Banned {did}.")
165
+
if await ban_user(self, did):
178
166
self.query_one("#ban-input", Input).value = ""
179
167
self._load_data()
180
-
except AuthError:
181
-
self.notify("Session expired. Please log in again.", severity="error")
182
-
except Exception:
183
-
self.notify("Could not ban user.", severity="error")
184
168
185
169
def action_add_hide(self) -> None:
186
170
uri = self.query_one("#hide-input", Input).value.strip()
···
191
175
192
176
@work
193
177
async def _do_add_hide(self, uri: str) -> None:
194
-
session = self.app.user_session
195
-
updater = make_session_updater(self.app.session_store)
196
-
197
178
if uri in self._hide_rkeys:
198
179
self.notify("Already hidden.", severity="warning")
199
180
return
200
181
201
-
try:
202
-
await create_hidden_record(self.app.http_client, session, uri, updater)
203
-
invalidate_bbs_cache()
204
-
self.notify("Post hidden.")
182
+
if await hide_post(self, uri):
205
183
self.query_one("#hide-input", Input).value = ""
206
184
self._load_data()
207
-
except AuthError:
208
-
self.notify("Session expired. Please log in again.", severity="error")
209
-
except Exception:
210
-
self.notify("Could not hide post.", severity="error")
211
185
212
186
def refresh_data(self) -> None:
213
187
self._load_data()
+7
-34
tui/screens/thread.py
+7
-34
tui/screens/thread.py
···
11
11
12
12
from core import lexicon
13
13
from core.models import BBS, AtUri, AuthError, Post as PostModel
14
-
from core.records import (
15
-
create_ban_record,
16
-
create_hidden_record,
17
-
delete_record,
18
-
post_from_record,
19
-
)
20
-
from core.resolver import invalidate_bbs_cache
21
-
from core.records import hydrate_replies as fetch_replies
14
+
from core.records import delete_record, hydrate_replies as fetch_replies, post_from_record
22
15
from core.slingshot import get_record, resolve_identity
23
16
from tui.screens.compose import ComposeReplyScreen
24
-
from tui.util import make_session_updater, require_session
17
+
from tui.util import ban_user, hide_post, require_session, require_sysop
25
18
from tui.widgets.breadcrumb import Breadcrumb
26
19
from tui.widgets.post import Post
27
20
···
180
173
replies[0].focus()
181
174
182
175
def action_ban(self) -> None:
183
-
session = self.app.user_session
184
-
if not session or session["did"] != self.bbs.identity.did:
185
-
self.notify("User not authorized.", severity="error")
176
+
if not require_sysop(self, self.bbs):
186
177
return
187
178
focused = self.focused
188
179
if not isinstance(focused, Post) or not focused.author_did:
189
180
return
190
-
if focused.author_did == session["did"]:
181
+
if focused.author_did == self.app.user_session["did"]:
191
182
self.notify("Cannot ban yourself.", severity="warning")
192
183
return
193
184
self._do_ban(focused.author_did)
194
185
195
186
@work
196
187
async def _do_ban(self, did: str) -> None:
197
-
session = self.app.user_session
198
-
updater = make_session_updater(self.app.session_store)
199
-
try:
200
-
await create_ban_record(self.app.http_client, session, did, updater)
201
-
invalidate_bbs_cache()
202
-
self.notify(f"Banned {did}.")
203
-
except Exception:
204
-
self.notify("Could not ban user.", severity="error")
188
+
await ban_user(self, did)
205
189
206
190
def action_hide(self) -> None:
207
-
session = self.app.user_session
208
-
if not session or session["did"] != self.bbs.identity.did:
209
-
self.notify("User not authorized.", severity="error")
191
+
if not require_sysop(self, self.bbs):
210
192
return
211
193
focused = self.focused
212
194
if not isinstance(focused, Post) or not focused.record_uri:
···
215
197
216
198
@work
217
199
async def _do_hide(self, post: Post) -> None:
218
-
session = self.app.user_session
219
-
updater = make_session_updater(self.app.session_store)
220
-
try:
221
-
await create_hidden_record(
222
-
self.app.http_client, session, post.record_uri, updater
223
-
)
224
-
invalidate_bbs_cache()
200
+
if await hide_post(self, post.record_uri):
225
201
await post.remove()
226
-
self.notify("Post hidden.")
227
-
except Exception:
228
-
self.notify("Could not hide post.", severity="error")
229
202
230
203
def action_next_page(self) -> None:
231
204
if self._page < self._total_pages:
+60
tui/util.py
+60
tui/util.py
···
1
1
"""TUI utilities."""
2
2
3
3
from core.auth.session import SessionStore
4
+
from core.models import AuthError, BBS
5
+
from core.records import create_ban_record, create_hidden_record
6
+
from core.resolver import invalidate_bbs_cache
4
7
5
8
6
9
def require_session(screen) -> dict | None:
···
12
15
return session
13
16
14
17
18
+
def require_sysop(screen, bbs: BBS) -> dict | None:
19
+
"""Return the user session if logged in AND is the BBS sysop.
20
+
21
+
Shows an error notification and returns None otherwise.
22
+
"""
23
+
session = screen.app.user_session
24
+
if not session:
25
+
screen.notify("You must be logged in to do that.", severity="error")
26
+
return None
27
+
if session["did"] != bbs.identity.did:
28
+
screen.notify("Only the sysop can do that.", severity="error")
29
+
return None
30
+
return session
31
+
32
+
15
33
def make_session_updater(store: SessionStore):
16
34
"""Create a session_updater callback for PDS write operations."""
17
35
···
19
37
store.update_session_field(did, field, value)
20
38
21
39
return updater
40
+
41
+
42
+
async def ban_user(screen, did: str) -> bool:
43
+
"""Ban a user by DID. Returns True on success, False on failure.
44
+
45
+
Handles the full workflow: create ban record, invalidate cache,
46
+
and show a success or error notification.
47
+
"""
48
+
session = screen.app.user_session
49
+
updater = make_session_updater(screen.app.session_store)
50
+
try:
51
+
await create_ban_record(screen.app.http_client, session, did, updater)
52
+
invalidate_bbs_cache()
53
+
screen.notify(f"Banned {did}.")
54
+
return True
55
+
except AuthError:
56
+
screen.notify("Session expired. Please log in again.", severity="error")
57
+
return False
58
+
except Exception:
59
+
screen.notify("Could not ban user.", severity="error")
60
+
return False
61
+
62
+
63
+
async def hide_post(screen, uri: str) -> bool:
64
+
"""Hide a post by AT-URI. Returns True on success, False on failure.
65
+
66
+
Handles the full workflow: create hidden record, invalidate cache,
67
+
and show a success or error notification.
68
+
"""
69
+
session = screen.app.user_session
70
+
updater = make_session_updater(screen.app.session_store)
71
+
try:
72
+
await create_hidden_record(screen.app.http_client, session, uri, updater)
73
+
invalidate_bbs_cache()
74
+
screen.notify("Post hidden.")
75
+
return True
76
+
except AuthError:
77
+
screen.notify("Session expired. Please log in again.", severity="error")
78
+
return False
79
+
except Exception:
80
+
screen.notify("Could not hide post.", severity="error")
81
+
return False