aly.codes
Retro Bulletin Board Systems on atproto. Web app and TUI.
atbbs.xyz
python
tui
atproto
bbs
1import os
2
3import httpx
4from platformdirs import user_data_dir
5from textual.app import App, ComposeResult
6from textual.binding import Binding
7
8from core.auth.session import SessionStore
9from textual.containers import Vertical
10from textual.screen import Screen
11from textual.widgets import Button, Footer, Static
12
13from tui.screens.home import HomeScreen
14
15
16class LogoutConfirmScreen(Screen):
17 BINDINGS = [("escape", "app.pop_screen", "cancel")]
18
19 DEFAULT_CSS = """
20 LogoutConfirmScreen {
21 align: center middle;
22 }
23 LogoutConfirmScreen Vertical {
24 width: 40;
25 height: auto;
26 padding: 1 2;
27 }
28 LogoutConfirmScreen Button {
29 width: 100%;
30 margin-top: 1;
31 }
32 """
33
34 def compose(self) -> ComposeResult:
35 with Vertical():
36 yield Static("Log out?", classes="title")
37 yield Button("log out", id="logout-confirm", variant="error")
38 yield Button("cancel", id="logout-cancel")
39 yield Footer()
40
41 def on_mount(self) -> None:
42 self.query_one("#logout-confirm", Button).focus()
43
44 def on_button_pressed(self, event: Button.Pressed) -> None:
45 if event.button.id == "logout-confirm":
46 self.app.pop_screen()
47 self.app.do_logout()
48 else:
49 self.app.pop_screen()
50
51
52DATA_DIR = os.environ.get("ATBBS_DATA_DIR", user_data_dir("atbbs"))
53
54
55class AtbbsApp(App):
56 TITLE = "@bbs"
57 CSS_PATH = "app.tcss"
58 BINDINGS = [
59 Binding("ctrl+q", "quit", "quit"),
60 Binding("ctrl+l", "login", "account"),
61 Binding("ctrl+r", "refresh", "refresh", show=False),
62 Binding("ctrl+t", "inbox", "messages", show=False),
63 ]
64 SCREENS = {"home": HomeScreen}
65
66 def __init__(self, dial: str | None = None):
67 super().__init__()
68 self._dial = dial
69
70 def on_mount(self) -> None:
71 self.http_client = httpx.AsyncClient(timeout=10)
72 os.makedirs(DATA_DIR, exist_ok=True)
73 db_path = os.path.join(DATA_DIR, "atbbs.db")
74 self.session_store = SessionStore(db_path)
75 self.user_session = None
76
77 # Restore saved session
78 self._restore_session()
79
80 home = HomeScreen()
81 self.push_screen(home)
82
83 if self._dial:
84 home.connect(self._dial)
85
86 def _restore_session(self) -> None:
87 """Load the most recent session from the database."""
88 import sqlite3
89
90 try:
91 con = sqlite3.connect(self.session_store.db_path)
92 con.row_factory = sqlite3.Row
93 row = con.execute("SELECT * FROM oauth_session LIMIT 1").fetchone()
94 con.close()
95 if row:
96 self.user_session = dict(row)
97 self.sub_title = self.user_session.get("handle", "")
98 except Exception:
99 pass
100
101 def action_login(self) -> None:
102 if self.user_session:
103 self.push_screen(LogoutConfirmScreen())
104 else:
105 from tui.screens.login import LoginScreen
106
107 self.push_screen(LoginScreen())
108
109 def do_logout(self) -> None:
110 did = self.user_session.get("did")
111 if did:
112 self.session_store.delete_session(did)
113 handle = self.user_session.get("handle", "")
114 self.user_session = None
115 self.sub_title = ""
116 self.notify(f"Logged out of {handle}.")
117
118 def action_inbox(self) -> None:
119 if not self.user_session:
120 self.notify("Log in to see your messages.", severity="warning")
121 return
122 from tui.screens.activity import ActivityScreen
123
124 self.push_screen(ActivityScreen())
125
126 def watch_screen(self) -> None:
127 """Update title when returning from login."""
128 if self.user_session:
129 self.sub_title = self.user_session.get("handle", "")
130
131 def action_refresh(self) -> None:
132 screen = self.screen
133 if hasattr(screen, "refresh_data"):
134 screen.refresh_data()
135
136 async def on_unmount(self) -> None:
137 if hasattr(self, "http_client"):
138 await self.http_client.aclose()