Retro Bulletin Board Systems on atproto. Web app and TUI. lazy mirror of alyraffauf/atbbs atbbs.xyz
forums python tui atproto bbs
3
fork

Configure Feed

Select the types of activity you want to include in your feed.

telnet: init

+270
telnet/__init__.py

This is a binary file and will not be displayed.

+270
telnet/server.py
··· 1 + import asyncio 2 + 3 + import httpx 4 + 5 + import re 6 + import textwrap 7 + 8 + from core.records import hydrate_replies, hydrate_threads 9 + from core.resolver import resolve_bbs 10 + from core.util import format_datetime_utc 11 + 12 + AMBER = "\033[38;5;208m" 13 + RESET = "\033[0m" 14 + DIM = "\033[2m" 15 + 16 + LOGO = ( 17 + f" {AMBER}▞▀▖{RESET}▌ ▌\r\n" 18 + f" {AMBER}▌▙▌{RESET}▛▀▖▛▀▖▞▀▘\r\n" 19 + f" {AMBER}▌▀ {RESET}▌ ▌▌ ▌▝▀▖\r\n" 20 + f" {AMBER}▝▀ {RESET}▀▀ ▀▀ ▀▀\r\n" 21 + ) 22 + 23 + 24 + LINE_WIDTH = 75 25 + 26 + READ_TIMEOUT = 120 # seconds per prompt 27 + MAX_CONNECTIONS = 20 28 + 29 + _connections = asyncio.Semaphore(MAX_CONNECTIONS) 30 + 31 + 32 + def wrap(text: str) -> str: 33 + """Wrap text to LINE_WIDTH without breaking indentation.""" 34 + out = [] 35 + for line in text.split("\r\n"): 36 + visible = re.sub(r"\033\[[0-9;]*m", "", line) # Strip ANSI codes 37 + if len(visible) <= LINE_WIDTH: 38 + out.append(line) 39 + else: 40 + indent = len(visible) - len(visible.lstrip()) # Preserve indentation 41 + wrapped = textwrap.fill( 42 + visible, 43 + width=LINE_WIDTH, 44 + initial_indent="", 45 + subsequent_indent=" " * indent, 46 + ) 47 + out.append(wrapped) 48 + return "\r\n".join(out) 49 + 50 + 51 + async def write(writer: asyncio.StreamWriter, text: str): 52 + writer.write(wrap(text).encode()) 53 + await writer.drain() 54 + 55 + 56 + async def prompt( 57 + reader: asyncio.streamReader, writer: asyncio.StreamWriter, label: str = "> " 58 + ) -> str: 59 + """Handle prompt + inactivity timeouts safely.""" 60 + await write(writer, label) 61 + try: 62 + data = await asyncio.wait_for(reader.readline(), timeout=READ_TIMEOUT) 63 + except asyncio.TimeoutError: 64 + await write(writer, "\r\n Connection inactive.\r\n") 65 + return "" 66 + if not data: 67 + return "" 68 + text = data.decode(errors="ignore").strip() 69 + return re.sub(r"[^\x20-\x7e]", "", text) 70 + 71 + async def show_bbs(writer, bbs): 72 + await write(writer, f"\r\n {bbs.site.name}\r\n") 73 + await write(writer, f" {bbs.site.description}\r\n") 74 + if bbs.site.intro: 75 + await write(writer, "\r\n") 76 + for line in bbs.site.intro.splitlines(): 77 + await write(writer, f" {line}\r\n") 78 + await write(writer, "\r\n") 79 + if bbs.site.boards: 80 + await write(writer, " Boards\r\n") 81 + for i, board in enumerate(bbs.site.boards, 1): 82 + await write(writer, f" {i}. {board.name}: {board.description}\r\n") 83 + await write(writer, "\r\n") 84 + 85 + if bbs.news: 86 + await write(writer, f" Latest News: {bbs.news[0].title}\r\n\r\n") 87 + 88 + await write(writer, "[#] open board [n] news [q] quit\r\n") 89 + 90 + async def show_board(writer, board, threads, has_next): 91 + await write(writer, f"\r\n {board.name}\r\n") 92 + await write(writer, f" {board.description}\r\n\r\n") 93 + 94 + if not threads: 95 + await write(writer, " No threads yet.\r\n") 96 + else: 97 + for i, t in enumerate(threads, 1): 98 + date = format_datetime_utc(t.created_at) 99 + await write(writer, f" {i}. {t.title} · {t.author.handle} · {date}\r\n") 100 + 101 + cmds = ["[#] open thread"] 102 + if has_next: 103 + cmds.append("[n] next") 104 + cmds.extend(["[b] back", "[q] quit"]) 105 + await write(writer, f"\r\n{' '.join(cmds)}\r\n") 106 + 107 + 108 + async def show_thread_header(writer, thread): 109 + await write(writer, f"\r\n {thread.title}\r\n") 110 + await write( 111 + writer, 112 + f" by {thread.author.handle} · {format_datetime_utc(thread.created_at)}\r\n", 113 + ) 114 + await write(writer, "\r\n") 115 + for line in thread.body.splitlines(): 116 + await write(writer, f" {line}\r\n") 117 + await write(writer, "\r\n") 118 + 119 + 120 + async def show_replies(writer, replies): 121 + for r in replies: 122 + await write( 123 + writer, f" {r.author.handle} · {format_datetime_utc(r.created_at)}\r\n" 124 + ) 125 + for line in r.body.splitlines(): 126 + await write(writer, f" {line}\r\n") 127 + await write(writer, "\r\n") 128 + 129 + 130 + async def show_thread_prompt(writer, has_more): 131 + cmds = ["[b] back"] 132 + if has_more: 133 + cmds.append("[n] show more") 134 + cmds.append("[q] quit") 135 + await write(writer, f"{' '.join(cmds)}\r\n") 136 + 137 + 138 + async def show_news(writer, news): 139 + await write(writer, "\r\n News:\r\n\r\n") 140 + for item in news: 141 + await write( 142 + writer, f" {item.title} · {format_datetime_utc(item.created_at)}\r\n" 143 + ) 144 + for line in item.body.splitlines(): 145 + await write(writer, f" {line}\r\n") 146 + await write(writer, "\r\n") 147 + await write(writer, "[b] back [q] quit\r\n") 148 + 149 + 150 + async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter): 151 + if _connections.locked(): 152 + writer.write(b" Server is full. Try again later.\r\n") 153 + await writer.drain() 154 + writer.close() 155 + return 156 + async with _connections, httpx.AsyncClient() as client: 157 + await write(writer, f"\r\n{LOGO}\r\n") 158 + await write( 159 + writer, 160 + " This is a read-only telnet gateway for Atmosphere BBSes.\r\n Please dial a BBS.\r\n\r\n", 161 + ) 162 + 163 + handle = await prompt(reader, writer, "handle> ") 164 + if not handle: 165 + writer.close() 166 + return 167 + 168 + try: 169 + bbs = await resolve_bbs(client, handle) 170 + except Exception as e: 171 + await write(writer, " Could not reach that BBS.\r\n") 172 + await write(writer, f"{e}\r\n") 173 + writer.close() 174 + return 175 + 176 + state = "bbs" 177 + board = None 178 + threads = [] 179 + thread_cursor = None 180 + thread = None 181 + reply_result = None 182 + 183 + while True: 184 + if state == "bbs": 185 + await show_bbs(writer, bbs) 186 + elif state == "board": 187 + await show_board(writer, board, threads, thread_cursor is not None) 188 + elif state == "news": 189 + await show_news(writer, bbs.news) 190 + 191 + # Thread state renders inline below — prompt only 192 + if state == "thread": 193 + await show_thread_prompt( 194 + writer, reply_result.page < reply_result.total_pages 195 + ) 196 + 197 + cmd = await prompt(reader, writer) 198 + if not cmd or cmd == "q": 199 + break 200 + 201 + if state == "bbs": 202 + if cmd == "n" and bbs.news: 203 + state = "news" 204 + elif cmd.isdigit(): 205 + idx = int(cmd) - 1 206 + if 0 <= idx < len(bbs.site.boards): 207 + board = bbs.site.boards[idx] 208 + try: 209 + threads, thread_cursor = await hydrate_threads( 210 + client, bbs, board 211 + ) 212 + except Exception: 213 + await write(writer, " Could not load threads.\r\n") 214 + continue 215 + state = "board" 216 + 217 + elif state == "board": 218 + if cmd == "b": 219 + state = "bbs" 220 + elif cmd == "n" and thread_cursor: 221 + try: 222 + threads, thread_cursor = await hydrate_threads( 223 + client, bbs, board, cursor=thread_cursor 224 + ) 225 + except Exception: 226 + await write(writer, " Could not load threads.\r\n") 227 + elif cmd.isdigit(): 228 + idx = int(cmd) - 1 229 + if 0 <= idx < len(threads): 230 + thread = threads[idx] 231 + await show_thread_header(writer, thread) 232 + try: 233 + reply_result = await hydrate_replies( 234 + client, bbs, thread.uri 235 + ) 236 + except Exception: 237 + await write(writer, " Could not load replies.\r\n") 238 + continue 239 + await show_replies(writer, reply_result.replies) 240 + state = "thread" 241 + 242 + elif state == "thread": 243 + if cmd == "b": 244 + state = "board" 245 + elif cmd == "n" and reply_result.page < reply_result.total_pages: 246 + try: 247 + reply_result = await hydrate_replies( 248 + client, bbs, thread.uri, page=reply_result.page + 1 249 + ) 250 + await show_replies(writer, reply_result.replies) 251 + except Exception: 252 + await write(writer, " Could not load replies.\r\n") 253 + 254 + elif state == "news": 255 + if cmd == "b": 256 + state = "bbs" 257 + 258 + await write(writer, "\r\n Goodbye!\r\n") 259 + writer.close() 260 + 261 + 262 + async def main(host: str = "0.0.0.0", port: int = 2323): 263 + server = await asyncio.start_server(handle_client, host, port, limit=256) 264 + print(f"Telnet BBS gateway listening on {host}:{port}") 265 + async with server: 266 + await server.serve_forever() 267 + 268 + 269 + if __name__ == "__main__": 270 + asyncio.run(main())