diff --git a/tui/screens/activity.py b/tui/screens/activity.py index 69a3e86..f87b27e 100644 --- a/tui/screens/activity.py +++ b/tui/screens/activity.py @@ -119,16 +119,16 @@ class ActivityScreen(Screen): await scroll.mount(Static("No messages yet.", classes="subtitle")) return - for a in self._items[:50]: - title = a["thread_title"] if a["type"] == "reply" else "quoted your reply" - if a["type"] == "reply": + for item in self._items[:50]: + title = item["thread_title"] if item["type"] == "reply" else "quoted your reply" + if item["type"] == "reply": title = f"on: {title}" await scroll.mount( Post( - author=a["handle"], - date=a["created_at"], + author=item["handle"], + date=item["created_at"], title=title, - body=a["body"], + body=item["body"], ) ) diff --git a/tui/screens/board.py b/tui/screens/board.py index 249937a..4a04640 100644 --- a/tui/screens/board.py +++ b/tui/screens/board.py @@ -67,16 +67,14 @@ class BoardScreen(Screen): self.notify("Could not fetch threads.", severity="error") return - lv = self.query_one("#thread-list", ListView) - lv.clear() - for t in self.threads: - label = ( - f" {t.title} — {t.author.handle} · {format_datetime(t.created_at)}" - ) - await lv.append(ListItem(Static(label), name=t.uri)) + thread_list = self.query_one("#thread-list", ListView) + thread_list.clear() + for thread in self.threads: + label = f" {thread.title} — {thread.author.handle} · {format_datetime(thread.created_at)}" + await thread_list.append(ListItem(Static(label), name=thread.uri)) if self.threads: - lv.index = 0 + thread_list.index = 0 # Remove old next page button if present for btn in self.query("#next-page"): diff --git a/tui/screens/compose.py b/tui/screens/compose.py deleted file mode 100644 index f31de95..0000000 --- a/tui/screens/compose.py +++ /dev/null @@ -1,318 +0,0 @@ -import mimetypes -from pathlib import Path - -from textual import work -from textual.app import ComposeResult -from textual.containers import Vertical -from textual.screen import Screen -from textual.widgets import Footer, Input, Static, TextArea - -from core import lexicon, limits -from core.models import AtUri, AuthError, BBS, Board, Reply, Thread -from core.records import ( - create_news_record, - create_reply_record, - create_thread_record, - upload_blob, -) -from tui.util import require_session -from tui.widgets.breadcrumb import Breadcrumb - - -async def _upload_file(screen, file_path: str, session: dict) -> list[dict] | None: - """Upload a file and return attachments list, or None on error.""" - p = Path(file_path).expanduser().resolve() - if not p.exists(): - screen.notify(f"File not found: {p}", severity="error") - return None - if not p.is_file(): - screen.notify(f"Not a file: {p}", severity="error") - return None - data = p.read_bytes() - mime = mimetypes.guess_type(str(p))[0] or "application/octet-stream" - - async def _update_nonce(did, field, value): - if hasattr(screen.app, "user_session") and screen.app.user_session: - screen.app.user_session[field] = value - - try: - blob_ref = await upload_blob( - screen.app.http_client, session, data, mime, session_updater=_update_nonce - ) - return [{"file": blob_ref, "name": p.name}] - except Exception as e: - screen.notify(f"Failed to upload file: {e}", severity="error") - return None - - -class ComposeThreadScreen(Screen): - BINDINGS = [ - ("escape", "app.pop_screen", "back"), - ("ctrl+s", "post", "post"), - ] - - def __init__(self, bbs: BBS, handle: str, board: Board) -> None: - super().__init__() - self.bbs = bbs - self.handle = handle - self.board = board - - def compose(self) -> ComposeResult: - yield Breadcrumb( - ("@bbs", 3), - (self.bbs.site.name, 2), - (self.board.name, 1), - ("new thread", 0), - ) - with Vertical(): - yield Static("new thread", classes="title") - yield Input( - placeholder="Thread title", - id="thread-title", - max_length=limits.THREAD_TITLE, - ) - yield TextArea(id="thread-body", language=None) - yield Input(placeholder="attach file (path, optional)", id="thread-file") - yield Footer() - - def on_mount(self) -> None: - self.query_one("#thread-title", Input).focus() - - def action_post(self) -> None: - self.post_thread() - - @work(exclusive=True) - async def post_thread(self) -> None: - session = require_session(self) - if not session: - return - - title = self.query_one("#thread-title", Input).value.strip() - body = self.query_one("#thread-body", TextArea).text.strip() - if not title or not body: - self.notify("Title and body cannot be empty.", severity="error") - return - if len(body) > limits.THREAD_BODY: - self.notify( - f"Body too long ({len(body)}/{limits.THREAD_BODY}).", severity="error" - ) - return - - board_uri = str(AtUri(self.bbs.identity.did, lexicon.BOARD, self.board.slug)) - - # Handle file attachment - attachments = [] - file_path = self.query_one("#thread-file", Input).value.strip() - if file_path: - attachments = await _upload_file(self, file_path, session) - if attachments is None: - return - - try: - resp = await create_thread_record( - self.app.http_client, - session, - board_uri, - title, - body, - attachments=attachments or None, - ) - resp.raise_for_status() - except AuthError: - self.notify("Session expired. Please log in again.", severity="error") - return - except Exception as e: - self.notify(f"Failed to post thread: {e}", severity="error") - return - - self.app.pop_screen() - - -class ComposeReplyScreen(Screen): - BINDINGS = [ - ("escape", "app.pop_screen", "back"), - ("ctrl+s", "post", "post"), - ("ctrl+g", "toggle_quote", "toggle quote"), - ] - - def __init__( - self, bbs: BBS, handle: str, thread: Thread, quote: Reply | None = None - ) -> None: - super().__init__() - self.bbs = bbs - self.handle = handle - self._original_quote = quote - self.quote = quote # Reply object or None - self.thread = thread - - def compose(self) -> ComposeResult: - yield Breadcrumb( - ("@bbs", 3), - (self.bbs.site.name, 2), - (self.thread.title, 1), - ("reply", 0), - ) - with Vertical(): - yield Static(f"reply to: {self.thread.title}", classes="title") - if self.quote: - body_preview = self.quote.body[:60] + ( - "..." if len(self.quote.body) > 60 else "" - ) - yield Static( - f"quoting {self.quote.author.handle}: {body_preview}", - classes="subtitle", - id="quote-info", - ) - yield TextArea(id="reply-body", language=None) - yield Input(placeholder="attach file (path, optional)", id="reply-file") - yield Footer() - - def on_mount(self) -> None: - self.query_one("#reply-body", TextArea).focus() - - def action_toggle_quote(self) -> None: - if not self._original_quote: - return - if self.quote: - self.quote = None - for w in self.query("#quote-info"): - w.remove() - else: - self.quote = self._original_quote - body_preview = self.quote.body[:60] + ( - "..." if len(self.quote.body) > 60 else "" - ) - scroll = self.query_one(Vertical) - scroll.mount( - Static( - f"quoting {self.quote.author.handle}: {body_preview}", - classes="subtitle", - id="quote-info", - ), - before=self.query_one("#reply-body"), - ) - - def action_post(self) -> None: - self.post_reply() - - @work(exclusive=True) - async def post_reply(self) -> None: - session = require_session(self) - if not session: - return - - body = self.query_one("#reply-body", TextArea).text.strip() - if not body: - self.notify("Message body cannot be empty.", severity="error") - return - if len(body) > limits.REPLY_BODY: - self.notify( - f"Body too long ({len(body)}/{limits.REPLY_BODY}).", severity="error" - ) - return - - # Handle file attachment - attachments = [] - file_path = self.query_one("#reply-file", Input).value.strip() - if file_path: - attachments = await _upload_file(self, file_path, session) - if attachments is None: - return - - try: - resp = await create_reply_record( - self.app.http_client, - session, - self.thread.uri, - body, - attachments=attachments or None, - quote=self.quote.uri if self.quote else None, - ) - resp.raise_for_status() - except AuthError: - self.notify("Session expired. Please log in again.", severity="error") - return - except Exception as e: - self.notify(f"Failed to post reply: {e}", severity="error") - return - - self.app.pop_screen() - - -class ComposeNewsScreen(Screen): - BINDINGS = [ - ("escape", "app.pop_screen", "back"), - ("ctrl+s", "post", "post"), - ] - - def __init__(self, bbs: BBS, handle: str) -> None: - super().__init__() - self.bbs = bbs - self.handle = handle - - def compose(self) -> ComposeResult: - yield Breadcrumb( - ("@bbs", 2), - (self.bbs.site.name, 1), - ("news", 0), - ) - with Vertical(): - yield Static("news", classes="title") - yield Input( - placeholder="Title", id="news-title", max_length=limits.NEWS_TITLE - ) - yield TextArea(id="news-body", language=None) - yield Input(placeholder="attach file (path, optional)", id="news-file") - yield Footer() - - def on_mount(self) -> None: - self.query_one("#news-title", Input).focus() - - def action_post(self) -> None: - self.post_news() - - @work(exclusive=True) - async def post_news(self) -> None: - session = require_session(self) - if not session: - return - - title = self.query_one("#news-title", Input).value.strip() - body = self.query_one("#news-body", TextArea).text.strip() - if not title or not body: - self.notify("Title and body cannot be empty.", severity="error") - return - if len(body) > limits.NEWS_BODY: - self.notify( - f"Body too long ({len(body)}/{limits.NEWS_BODY}).", severity="error" - ) - return - - site_uri = str(AtUri(self.bbs.identity.did, lexicon.SITE, "self")) - - # Handle file attachment - attachments = [] - file_path = self.query_one("#news-file", Input).value.strip() - if file_path: - attachments = await _upload_file(self, file_path, session) - if attachments is None: - return - - try: - resp = await create_news_record( - self.app.http_client, - session, - site_uri, - title, - body, - attachments=attachments or None, - ) - resp.raise_for_status() - except AuthError: - self.notify("Session expired. Please log in again.", severity="error") - return - except Exception as e: - self.notify(f"Failed to post news: {e}", severity="error") - return - - self.app.pop_screen() diff --git a/tui/screens/compose/__init__.py b/tui/screens/compose/__init__.py new file mode 100644 index 0000000..3a41e16 --- /dev/null +++ b/tui/screens/compose/__init__.py @@ -0,0 +1,5 @@ +from tui.screens.compose.thread import ComposeThreadScreen +from tui.screens.compose.reply import ComposeReplyScreen +from tui.screens.compose.news import ComposeNewsScreen + +__all__ = ["ComposeThreadScreen", "ComposeReplyScreen", "ComposeNewsScreen"] diff --git a/tui/screens/compose/news.py b/tui/screens/compose/news.py new file mode 100644 index 0000000..4594980 --- /dev/null +++ b/tui/screens/compose/news.py @@ -0,0 +1,90 @@ +from textual import work +from textual.app import ComposeResult +from textual.containers import Vertical +from textual.screen import Screen +from textual.widgets import Footer, Input, Static, TextArea + +from core import lexicon, limits +from core.models import AtUri, AuthError, BBS +from core.records import create_news_record +from tui.util import require_session +from tui.widgets.breadcrumb import Breadcrumb +from tui.screens.compose.upload import upload_file + + +class ComposeNewsScreen(Screen): + BINDINGS = [ + ("escape", "app.pop_screen", "back"), + ("ctrl+s", "post", "post"), + ] + + def __init__(self, bbs: BBS, handle: str) -> None: + super().__init__() + self.bbs = bbs + self.handle = handle + + def compose(self) -> ComposeResult: + yield Breadcrumb( + ("@bbs", 2), + (self.bbs.site.name, 1), + ("news", 0), + ) + with Vertical(): + yield Static("news", classes="title") + yield Input( + placeholder="Title", id="news-title", max_length=limits.NEWS_TITLE + ) + yield TextArea(id="news-body", language=None) + yield Input(placeholder="attach file (path, optional)", id="news-file") + yield Footer() + + def on_mount(self) -> None: + self.query_one("#news-title", Input).focus() + + def action_post(self) -> None: + self.post_news() + + @work(exclusive=True) + async def post_news(self) -> None: + session = require_session(self) + if not session: + return + + title = self.query_one("#news-title", Input).value.strip() + body = self.query_one("#news-body", TextArea).text.strip() + if not title or not body: + self.notify("Title and body cannot be empty.", severity="error") + return + if len(body) > limits.NEWS_BODY: + self.notify( + f"Body too long ({len(body)}/{limits.NEWS_BODY}).", severity="error" + ) + return + + site_uri = str(AtUri(self.bbs.identity.did, lexicon.SITE, "self")) + + attachments = [] + file_path = self.query_one("#news-file", Input).value.strip() + if file_path: + attachments = await upload_file(self, file_path, session) + if attachments is None: + return + + try: + resp = await create_news_record( + self.app.http_client, + session, + site_uri, + title, + body, + attachments=attachments or None, + ) + resp.raise_for_status() + except AuthError: + self.notify("Session expired. Please log in again.", severity="error") + return + except Exception as error: + self.notify(f"Failed to post news: {error}", severity="error") + return + + self.app.pop_screen() diff --git a/tui/screens/compose/reply.py b/tui/screens/compose/reply.py new file mode 100644 index 0000000..9e45af4 --- /dev/null +++ b/tui/screens/compose/reply.py @@ -0,0 +1,122 @@ +from textual import work +from textual.app import ComposeResult +from textual.containers import Vertical +from textual.screen import Screen +from textual.widgets import Footer, Input, Static, TextArea + +from core import limits +from core.models import AuthError, BBS, Reply, Thread +from core.records import create_reply_record +from tui.util import require_session +from tui.widgets.breadcrumb import Breadcrumb +from tui.screens.compose.upload import upload_file + + +class ComposeReplyScreen(Screen): + BINDINGS = [ + ("escape", "app.pop_screen", "back"), + ("ctrl+s", "post", "post"), + ("ctrl+g", "toggle_quote", "toggle quote"), + ] + + def __init__( + self, bbs: BBS, handle: str, thread: Thread, quote: Reply | None = None + ) -> None: + super().__init__() + self.bbs = bbs + self.handle = handle + self._original_quote = quote + self.quote = quote + self.thread = thread + + def compose(self) -> ComposeResult: + yield Breadcrumb( + ("@bbs", 3), + (self.bbs.site.name, 2), + (self.thread.title, 1), + ("reply", 0), + ) + with Vertical(): + yield Static(f"reply to: {self.thread.title}", classes="title") + if self.quote: + body_preview = self.quote.body[:60] + ( + "..." if len(self.quote.body) > 60 else "" + ) + yield Static( + f"quoting {self.quote.author.handle}: {body_preview}", + classes="subtitle", + id="quote-info", + ) + yield TextArea(id="reply-body", language=None) + yield Input(placeholder="attach file (path, optional)", id="reply-file") + yield Footer() + + def on_mount(self) -> None: + self.query_one("#reply-body", TextArea).focus() + + def action_toggle_quote(self) -> None: + if not self._original_quote: + return + if self.quote: + self.quote = None + for widget in self.query("#quote-info"): + widget.remove() + else: + self.quote = self._original_quote + body_preview = self.quote.body[:60] + ( + "..." if len(self.quote.body) > 60 else "" + ) + container = self.query_one(Vertical) + container.mount( + Static( + f"quoting {self.quote.author.handle}: {body_preview}", + classes="subtitle", + id="quote-info", + ), + before=self.query_one("#reply-body"), + ) + + def action_post(self) -> None: + self.post_reply() + + @work(exclusive=True) + async def post_reply(self) -> None: + session = require_session(self) + if not session: + return + + body = self.query_one("#reply-body", TextArea).text.strip() + if not body: + self.notify("Message body cannot be empty.", severity="error") + return + if len(body) > limits.REPLY_BODY: + self.notify( + f"Body too long ({len(body)}/{limits.REPLY_BODY}).", severity="error" + ) + return + + attachments = [] + file_path = self.query_one("#reply-file", Input).value.strip() + if file_path: + attachments = await upload_file(self, file_path, session) + if attachments is None: + return + + try: + resp = await create_reply_record( + self.app.http_client, + session, + self.thread.uri, + body, + attachments=attachments or None, + quote=self.quote.uri if self.quote else None, + ) + resp.raise_for_status() + except AuthError: + self.notify("Session expired. Please log in again.", severity="error") + return + except Exception as error: + self.notify(f"Failed to post reply: {error}", severity="error") + return + + self.app.pop_screen() diff --git a/tui/screens/compose/thread.py b/tui/screens/compose/thread.py new file mode 100644 index 0000000..a455e47 --- /dev/null +++ b/tui/screens/compose/thread.py @@ -0,0 +1,94 @@ +from textual import work +from textual.app import ComposeResult +from textual.containers import Vertical +from textual.screen import Screen +from textual.widgets import Footer, Input, Static, TextArea + +from core import lexicon, limits +from core.models import AtUri, AuthError, BBS, Board +from core.records import create_thread_record +from tui.util import require_session +from tui.widgets.breadcrumb import Breadcrumb +from tui.screens.compose.upload import upload_file + + +class ComposeThreadScreen(Screen): + BINDINGS = [ + ("escape", "app.pop_screen", "back"), + ("ctrl+s", "post", "post"), + ] + + def __init__(self, bbs: BBS, handle: str, board: Board) -> None: + super().__init__() + self.bbs = bbs + self.handle = handle + self.board = board + + def compose(self) -> ComposeResult: + yield Breadcrumb( + ("@bbs", 3), + (self.bbs.site.name, 2), + (self.board.name, 1), + ("new thread", 0), + ) + with Vertical(): + yield Static("new thread", classes="title") + yield Input( + placeholder="Thread title", + id="thread-title", + max_length=limits.THREAD_TITLE, + ) + yield TextArea(id="thread-body", language=None) + yield Input(placeholder="attach file (path, optional)", id="thread-file") + yield Footer() + + def on_mount(self) -> None: + self.query_one("#thread-title", Input).focus() + + def action_post(self) -> None: + self.post_thread() + + @work(exclusive=True) + async def post_thread(self) -> None: + session = require_session(self) + if not session: + return + + title = self.query_one("#thread-title", Input).value.strip() + body = self.query_one("#thread-body", TextArea).text.strip() + if not title or not body: + self.notify("Title and body cannot be empty.", severity="error") + return + if len(body) > limits.THREAD_BODY: + self.notify( + f"Body too long ({len(body)}/{limits.THREAD_BODY}).", severity="error" + ) + return + + board_uri = str(AtUri(self.bbs.identity.did, lexicon.BOARD, self.board.slug)) + + attachments = [] + file_path = self.query_one("#thread-file", Input).value.strip() + if file_path: + attachments = await upload_file(self, file_path, session) + if attachments is None: + return + + try: + resp = await create_thread_record( + self.app.http_client, + session, + board_uri, + title, + body, + attachments=attachments or None, + ) + resp.raise_for_status() + except AuthError: + self.notify("Session expired. Please log in again.", severity="error") + return + except Exception as error: + self.notify(f"Failed to post thread: {error}", severity="error") + return + + self.app.pop_screen() diff --git a/tui/screens/compose/upload.py b/tui/screens/compose/upload.py new file mode 100644 index 0000000..5aa58df --- /dev/null +++ b/tui/screens/compose/upload.py @@ -0,0 +1,31 @@ +import mimetypes +from pathlib import Path + +from core.records import upload_blob + + +async def upload_file(screen, file_path: str, session: dict) -> list[dict] | None: + """Upload a file and return attachments list, or None on error.""" + path = Path(file_path).expanduser().resolve() + if not path.exists(): + screen.notify(f"File not found: {path}", severity="error") + return None + if not path.is_file(): + screen.notify(f"Not a file: {path}", severity="error") + return None + + data = path.read_bytes() + mime = mimetypes.guess_type(str(path))[0] or "application/octet-stream" + + async def nonce_updater(did, field, value): + if hasattr(screen.app, "user_session") and screen.app.user_session: + screen.app.user_session[field] = value + + try: + blob_ref = await upload_blob( + screen.app.http_client, session, data, mime, session_updater=nonce_updater + ) + return [{"file": blob_ref, "name": path.name}] + except Exception as error: + screen.notify(f"Failed to upload file: {error}", severity="error") + return None diff --git a/tui/screens/home.py b/tui/screens/home.py index 9979231..73fd1bf 100644 --- a/tui/screens/home.py +++ b/tui/screens/home.py @@ -116,27 +116,29 @@ class HomeScreen(Screen): if len(raw) > 5: raw = random.sample(raw, 5) - dids = [r["did"] for r in raw] + dids = [record["did"] for record in raw] authors = await resolve_identities_batch(client, dids) items = [] - for r in raw: - did = r["did"] + for record in raw: + did = record["did"] if did in authors: - name = r["record"].get("name", "") - desc = r["record"].get("description", "") + name = record["record"].get("name", "") + desc = record["record"].get("description", "") handle = authors[did].handle items.append((handle, name, desc)) if not items: return - lv = self.query_one("#discover-list", ListView) + discover_list = self.query_one("#discover-list", ListView) for handle, name, desc in items: - await lv.append(ListItem(Static(f" {name or handle}"), name=handle)) + await discover_list.append( + ListItem(Static(f" {name or handle}"), name=handle) + ) self.query_one("#discover-label").display = True - lv.display = True + discover_list.display = True lv.index = 0 # select first bbs except Exception: diff --git a/tui/screens/sysop.py b/tui/screens/sysop.py deleted file mode 100644 index 613c6f9..0000000 --- a/tui/screens/sysop.py +++ /dev/null @@ -1,536 +0,0 @@ -import asyncio - -from textual import work -from textual.app import ComposeResult -from textual.containers import Vertical, VerticalScroll -from textual.screen import Screen -from textual.widgets import Button, Footer, Input, ListItem, ListView, Static, TextArea - -from core import lexicon, limits -from core.models import AtUri, AuthError, BBS -from core.records import ( - create_ban_record, - create_hidden_record, - delete_record, - list_pds_records, - put_board_record, - put_site_record, -) -from core.constellation import get_news -from core.slingshot import resolve_identities_batch, resolve_identity -from core.util import now_iso -from tui.util import make_session_updater, require_session -from tui.widgets.breadcrumb import Breadcrumb - - -class SysopScreen(Screen): - BINDINGS = [("escape", "app.pop_screen", "back")] - - def __init__(self, bbs: BBS, handle: str) -> None: - super().__init__() - self.bbs = bbs - self.handle = handle - - def compose(self) -> ComposeResult: - yield ListView( - ListItem(Static(" Edit BBS"), name="edit"), - ListItem(Static(" Moderation"), name="moderate"), - ListItem(Static(" Delete BBS"), name="delete"), - id="sysop-menu", - ) - yield Footer() - - def on_mount(self) -> None: - self.query_one("#sysop-menu", ListView).focus() - - def on_list_view_selected(self, event: ListView.Selected) -> None: - name = event.item.name - if name == "edit": - self.app.push_screen(SysopEditScreen(self.bbs, self.handle)) - elif name == "moderate": - self.app.push_screen(SysopModerateScreen(self.bbs, self.handle)) - elif name == "delete": - self.app.push_screen(SysopDeleteScreen(self.bbs, self.handle)) - - -class SysopEditScreen(Screen): - BINDINGS = [ - ("escape", "app.pop_screen", "back"), - ("ctrl+s", "save", "save"), - ("ctrl+n", "add_board", "add board"), - ("ctrl+d", "remove_board", "remove board"), - ] - - def __init__(self, bbs: BBS, handle: str) -> None: - super().__init__() - self.bbs = bbs - self.handle = handle - self._boards = [ - { - "slug": b.slug, - "name": b.name, - "description": b.description, - "created_at": b.created_at, - } - for b in bbs.site.boards - ] - - def compose(self) -> ComposeResult: - yield Breadcrumb( - ("@bbs", 3), - (self.bbs.site.name, 2), - ("sysop", 1), - ("edit", 0), - ) - with VerticalScroll(id="edit-scroll"): - yield Static("NAME", classes="section-label") - yield Input( - value=self.bbs.site.name, id="edit-name", max_length=limits.SITE_NAME - ) - yield Static("DESCRIPTION", classes="section-label") - yield Input( - value=self.bbs.site.description, - id="edit-desc", - max_length=limits.SITE_DESCRIPTION, - ) - yield Static("INTRO", classes="section-label") - yield TextArea(self.bbs.site.intro, id="edit-intro", language=None) - yield Static( - "BOARDS (ctrl+n add, ctrl+d remove)", - classes="section-label", - id="boards-label", - ) - for b in self._boards: - yield Static( - f" {b['slug']}", classes="subtitle", id=f"board-label-{b['slug']}" - ) - yield Input( - value=b["name"], - id=f"board-name-{b['slug']}", - max_length=limits.BOARD_NAME, - ) - yield Input( - value=b["description"], - id=f"board-desc-{b['slug']}", - max_length=limits.BOARD_DESCRIPTION, - ) - yield Footer() - - def on_mount(self) -> None: - self.query_one("#edit-name", Input).focus() - - def action_add_board(self) -> None: - # Find a unique slug - i = len(self._boards) + 1 - while any(b["slug"] == f"board-{i}" for b in self._boards): - i += 1 - slug = f"board-{i}" - self._boards.append( - {"slug": slug, "name": slug, "description": "", "created_at": now_iso()} - ) - - scroll = self.query_one("#edit-scroll", VerticalScroll) - label = Static(f" {slug}", classes="subtitle", id=f"board-label-{slug}") - name_input = Input( - value=slug, id=f"board-name-{slug}", max_length=limits.BOARD_NAME - ) - desc_input = Input( - value="", id=f"board-desc-{slug}", max_length=limits.BOARD_DESCRIPTION - ) - scroll.mount(label) - scroll.mount(name_input) - scroll.mount(desc_input) - name_input.focus() - - def action_remove_board(self) -> None: - if len(self._boards) <= 1: - self.notify("Must have at least one board.", severity="warning") - return - - # Remove the last board - board = self._boards.pop() - slug = board["slug"] - for widget_id in ( - f"board-label-{slug}", - f"board-name-{slug}", - f"board-desc-{slug}", - ): - try: - self.query_one(f"#{widget_id}").remove() - except Exception: - pass - - def action_save(self) -> None: - self._do_save() - - @work(exclusive=True) - async def _do_save(self) -> None: - session = require_session(self) - if not session: - return - - updater = make_session_updater(self.app.session_store) - - name = self.query_one("#edit-name", Input).value.strip() - description = self.query_one("#edit-desc", Input).value.strip() - intro = self.query_one("#edit-intro", TextArea).text - - if not name: - self.notify("Name cannot be empty.", severity="error") - return - if len(intro) > limits.SITE_INTRO: - self.notify( - f"Intro too long ({len(intro)}/{limits.SITE_INTRO}).", severity="error" - ) - return - - now = now_iso() - - try: - # Update/create board records - for b in self._boards: - board_name = self.query_one( - f"#board-name-{b['slug']}", Input - ).value.strip() - board_desc = self.query_one( - f"#board-desc-{b['slug']}", Input - ).value.strip() - await put_board_record( - self.app.http_client, - session, - b["slug"], - board_name or b["slug"], - board_desc, - b["created_at"], - updater, - ) - - # Delete removed boards - current_slugs = {b["slug"] for b in self._boards} - for board in self.bbs.site.boards: - if board.slug not in current_slugs: - await delete_record( - self.app.http_client, - session, - lexicon.BOARD, - board.slug, - updater, - ) - - # Update site record - await put_site_record( - self.app.http_client, - session, - { - "$type": lexicon.SITE, - "name": name, - "description": description, - "intro": intro, - "boards": [b["slug"] for b in self._boards], - "createdAt": self.bbs.site.created_at or now, - "updatedAt": now, - }, - updater, - ) - self.notify("BBS updated.") - self.app.pop_screen() - except AuthError: - self.notify("Session expired. Please log in again.", severity="error") - except Exception as e: - self.notify(f"Could not update BBS: {e}", severity="error") - - -class SysopModerateScreen(Screen): - BINDINGS = [ - ("escape", "app.pop_screen", "back"), - ("ctrl+d", "remove", "remove"), - ("ctrl+b", "add_ban", "ban"), - ("ctrl+x", "add_hide", "hide"), - ] - - def __init__(self, bbs: BBS, handle: str) -> None: - super().__init__() - self.bbs = bbs - self.handle = handle - self._ban_rkeys: dict[str, str] = {} - self._hide_rkeys: dict[str, str] = {} - - def compose(self) -> ComposeResult: - yield Breadcrumb( - ("@bbs", 3), - (self.bbs.site.name, 2), - ("sysop", 1), - ("moderation", 0), - ) - with VerticalScroll(): - yield Static("BANNED USERS", classes="section-label") - yield Input(placeholder="handle or DID to ban", id="ban-input") - yield ListView(id="ban-list") - yield Static("HIDDEN POSTS", classes="section-label") - yield Input(placeholder="at:// URI to hide", id="hide-input") - yield ListView(id="hide-list") - yield Footer() - - def on_mount(self) -> None: - self._load_data() - - @work(exclusive=True) - async def _load_data(self) -> None: - client = self.app.http_client - session = self.app.user_session - - # Fetch ban and hide records in parallel - ban_result, hide_result = await asyncio.gather( - list_pds_records(client, session["pds_url"], session["did"], lexicon.BAN), - list_pds_records(client, session["pds_url"], session["did"], lexicon.HIDE), - return_exceptions=True, - ) - - if isinstance(ban_result, BaseException): - self._ban_rkeys = {} - else: - self._ban_rkeys = { - r["value"]["did"]: AtUri.parse(r["uri"]).rkey - for r in ban_result - if r.get("value", {}).get("did") - } - - if isinstance(hide_result, BaseException): - self._hide_rkeys = {} - else: - self._hide_rkeys = { - r["value"]["uri"]: AtUri.parse(r["uri"]).rkey - for r in hide_result - if r.get("value", {}).get("uri") - } - - # Resolve banned handles - banned_dids = list(self._ban_rkeys.keys()) - banned_handles: dict[str, str] = {} - if banned_dids: - try: - authors = await resolve_identities_batch(client, banned_dids) - banned_handles = {did: authors[did].handle for did in authors} - except Exception: - pass - - ban_list = self.query_one("#ban-list", ListView) - ban_list.clear() - for did in banned_dids: - label = banned_handles.get(did, did) - await ban_list.append(ListItem(Static(f" {label}"), name=f"ban:{did}")) - - hide_list = self.query_one("#hide-list", ListView) - hide_list.clear() - for uri in self._hide_rkeys: - await hide_list.append(ListItem(Static(f" {uri}"), name=f"hide:{uri}")) - - # Focus first list with items - if banned_dids: - ban_list.focus() - elif self._hide_rkeys: - hide_list.focus() - - def action_remove(self) -> None: - for lv_id in ("ban-list", "hide-list"): - lv = self.query_one(f"#{lv_id}", ListView) - if lv.index is not None and lv.has_focus: - item = lv.children[lv.index] - if item.name: - self._do_remove(item.name, item) - return - - @work - async def _do_remove(self, key: str, item) -> None: - session = self.app.user_session - updater = make_session_updater(self.app.session_store) - - kind, _, value = key.partition(":") - try: - if kind == "ban" and value in self._ban_rkeys: - rkey = self._ban_rkeys[value] - await delete_record( - self.app.http_client, session, lexicon.BAN, rkey, updater - ) - del self._ban_rkeys[value] - self.notify(f"Unbanned {value}.") - elif kind == "hide" and value in self._hide_rkeys: - rkey = self._hide_rkeys[value] - await delete_record( - self.app.http_client, session, lexicon.HIDE, rkey, updater - ) - del self._hide_rkeys[value] - self.notify("Post unhidden.") - await item.remove() - except AuthError: - self.notify("Session expired. Please log in again.", severity="error") - except Exception: - self.notify("Could not remove record.", severity="error") - - def action_add_ban(self) -> None: - identifier = self.query_one("#ban-input", Input).value.strip() - if not identifier: - self.notify("Enter a handle or DID.", severity="warning") - return - self._do_add_ban(identifier) - - @work - async def _do_add_ban(self, identifier: str) -> None: - session = self.app.user_session - store = self.app.session_store - client = self.app.http_client - - async def updater(d, field, value): - store.update_session_field(d, field, value) - - # Resolve handle to DID if needed - did = identifier - if not identifier.startswith("did:"): - try: - identity = await resolve_identity(client, identifier) - did = identity.did - except Exception: - self.notify(f"Could not resolve {identifier}.", severity="error") - return - - if did in self._ban_rkeys: - self.notify("Already banned.", severity="warning") - return - - try: - await create_ban_record(client, session, did, updater) - self.notify(f"Banned {did}.") - self.query_one("#ban-input", Input).value = "" - self._load_data() - except AuthError: - self.notify("Session expired. Please log in again.", severity="error") - except Exception: - self.notify("Could not ban user.", severity="error") - - def action_add_hide(self) -> None: - uri = self.query_one("#hide-input", Input).value.strip() - if not uri or not uri.startswith("at://"): - self.notify("Enter a valid AT-URI.", severity="warning") - return - self._do_add_hide(uri) - - @work - async def _do_add_hide(self, uri: str) -> None: - session = self.app.user_session - updater = make_session_updater(self.app.session_store) - - if uri in self._hide_rkeys: - self.notify("Already hidden.", severity="warning") - return - - try: - await create_hidden_record(self.app.http_client, session, uri, updater) - self.notify("Post hidden.") - self.query_one("#hide-input", Input).value = "" - self._load_data() - except AuthError: - self.notify("Session expired. Please log in again.", severity="error") - except Exception: - self.notify("Could not hide post.", severity="error") - - def refresh_data(self) -> None: - self._load_data() - - -class SysopDeleteScreen(Screen): - BINDINGS = [("escape", "app.pop_screen", "cancel")] - - def __init__(self, bbs: BBS, handle: str) -> None: - super().__init__() - self.bbs = bbs - self.handle = handle - - def compose(self) -> ComposeResult: - with Vertical(): - yield Static("Delete your BBS?", classes="title") - yield Static( - "This will delete your site record, all boards, news, " - "bans, and hidden post records. Threads and replies from " - "users will remain in their repos.", - ) - yield Button("delete", id="delete-confirm", variant="error") - yield Button("cancel", id="delete-cancel") - yield Footer() - - def on_mount(self) -> None: - self.query_one("#delete-cancel", Button).focus() - - def on_button_pressed(self, event: Button.Pressed) -> None: - if event.button.id == "delete-confirm": - self._do_delete() - else: - self.app.pop_screen() - - @work(exclusive=True) - async def _do_delete(self) -> None: - session = self.app.user_session - store = self.app.session_store - client = self.app.http_client - - async def updater(d, field, value): - store.update_session_field(d, field, value) - - failed = [] - - # Delete boards - for board in self.bbs.site.boards: - try: - await delete_record(client, session, lexicon.BOARD, board.slug, updater) - except Exception: - failed.append(f"board/{board.slug}") - - # Delete news - - site_uri = str(AtUri(session["did"], lexicon.SITE, "self")) - try: - backlinks = await get_news(client, site_uri) - for ref in backlinks.records: - if ref.did == session["did"]: - try: - await delete_record( - client, session, lexicon.NEWS, ref.rkey, updater - ) - except Exception: - failed.append(f"news/{ref.rkey}") - except Exception: - failed.append("news lookup") - - # Delete ban and hide records - for collection in (lexicon.BAN, lexicon.HIDE): - try: - records = await list_pds_records( - client, session["pds_url"], session["did"], collection - ) - for r in records: - rkey = AtUri.parse(r["uri"]).rkey - try: - await delete_record(client, session, collection, rkey, updater) - except Exception: - failed.append(f"{collection}/{rkey}") - except Exception: - failed.append(f"{collection} lookup") - - if failed: - self.notify( - f"Could not delete: {', '.join(failed)}. Site record not deleted.", - severity="error", - ) - return - - # Delete site record - try: - await delete_record(client, session, lexicon.SITE, "self", updater) - except Exception: - self.notify("Could not delete site record.", severity="error") - return - - self.notify("BBS deleted.") - # Pop delete screen + sysop screen + site screen - self.app.pop_screen() - self.app.pop_screen() - self.app.pop_screen() diff --git a/tui/screens/sysop/__init__.py b/tui/screens/sysop/__init__.py new file mode 100644 index 0000000..7668c45 --- /dev/null +++ b/tui/screens/sysop/__init__.py @@ -0,0 +1,6 @@ +from tui.screens.sysop.menu import SysopScreen +from tui.screens.sysop.edit import SysopEditScreen +from tui.screens.sysop.moderate import SysopModerateScreen +from tui.screens.sysop.delete import SysopDeleteScreen + +__all__ = ["SysopScreen", "SysopEditScreen", "SysopModerateScreen", "SysopDeleteScreen"] diff --git a/tui/screens/sysop/delete.py b/tui/screens/sysop/delete.py new file mode 100644 index 0000000..f75ad95 --- /dev/null +++ b/tui/screens/sysop/delete.py @@ -0,0 +1,103 @@ +from textual import work +from textual.app import ComposeResult +from textual.containers import Vertical +from textual.screen import Screen +from textual.widgets import Button, Footer, Static + +from core import lexicon +from core.models import AtUri, BBS +from core.constellation import get_news +from core.records import delete_record, list_pds_records +from tui.util import make_session_updater + + +class SysopDeleteScreen(Screen): + BINDINGS = [("escape", "app.pop_screen", "cancel")] + + def __init__(self, bbs: BBS, handle: str) -> None: + super().__init__() + self.bbs = bbs + self.handle = handle + + def compose(self) -> ComposeResult: + with Vertical(): + yield Static("Delete your BBS?", classes="title") + yield Static( + "This will delete your site record, all boards, news, " + "bans, and hidden post records. Threads and replies from " + "users will remain in their repos.", + ) + yield Button("delete", id="delete-confirm", variant="error") + yield Button("cancel", id="delete-cancel") + yield Footer() + + def on_mount(self) -> None: + self.query_one("#delete-cancel", Button).focus() + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "delete-confirm": + self._do_delete() + else: + self.app.pop_screen() + + @work(exclusive=True) + async def _do_delete(self) -> None: + session = self.app.user_session + client = self.app.http_client + updater = make_session_updater(self.app.session_store) + + failed = [] + + for board in self.bbs.site.boards: + try: + await delete_record(client, session, lexicon.BOARD, board.slug, updater) + except Exception: + failed.append(f"board/{board.slug}") + + site_uri = str(AtUri(session["did"], lexicon.SITE, "self")) + try: + backlinks = await get_news(client, site_uri) + for ref in backlinks.records: + if ref.did == session["did"]: + try: + await delete_record( + client, session, lexicon.NEWS, ref.rkey, updater + ) + except Exception: + failed.append(f"news/{ref.rkey}") + except Exception: + failed.append("news lookup") + + for collection in (lexicon.BAN, lexicon.HIDE): + try: + records = await list_pds_records( + client, session["pds_url"], session["did"], collection + ) + for record in records: + rkey = AtUri.parse(record["uri"]).rkey + try: + await delete_record( + client, session, collection, rkey, updater + ) + except Exception: + failed.append(f"{collection}/{rkey}") + except Exception: + failed.append(f"{collection} lookup") + + if failed: + self.notify( + f"Could not delete: {', '.join(failed)}. Site record not deleted.", + severity="error", + ) + return + + try: + await delete_record(client, session, lexicon.SITE, "self", updater) + except Exception: + self.notify("Could not delete site record.", severity="error") + return + + self.notify("BBS deleted.") + self.app.pop_screen() + self.app.pop_screen() + self.app.pop_screen() diff --git a/tui/screens/sysop/edit.py b/tui/screens/sysop/edit.py new file mode 100644 index 0000000..0afd196 --- /dev/null +++ b/tui/screens/sysop/edit.py @@ -0,0 +1,196 @@ +from textual import work +from textual.app import ComposeResult +from textual.containers import VerticalScroll +from textual.screen import Screen +from textual.widgets import Footer, Input, Static, TextArea + +from core import lexicon, limits +from core.models import AuthError, BBS +from core.records import delete_record, put_board_record, put_site_record +from core.util import now_iso +from tui.util import make_session_updater, require_session +from tui.widgets.breadcrumb import Breadcrumb + + +class SysopEditScreen(Screen): + BINDINGS = [ + ("escape", "app.pop_screen", "back"), + ("ctrl+s", "save", "save"), + ("ctrl+n", "add_board", "add board"), + ("ctrl+d", "remove_board", "remove board"), + ] + + def __init__(self, bbs: BBS, handle: str) -> None: + super().__init__() + self.bbs = bbs + self.handle = handle + self._boards = [ + { + "slug": board.slug, + "name": board.name, + "description": board.description, + "created_at": board.created_at, + } + for board in bbs.site.boards + ] + + def compose(self) -> ComposeResult: + yield Breadcrumb( + ("@bbs", 3), + (self.bbs.site.name, 2), + ("sysop", 1), + ("edit", 0), + ) + with VerticalScroll(id="edit-scroll"): + yield Static("NAME", classes="section-label") + yield Input( + value=self.bbs.site.name, id="edit-name", max_length=limits.SITE_NAME + ) + yield Static("DESCRIPTION", classes="section-label") + yield Input( + value=self.bbs.site.description, + id="edit-desc", + max_length=limits.SITE_DESCRIPTION, + ) + yield Static("INTRO", classes="section-label") + yield TextArea(self.bbs.site.intro, id="edit-intro", language=None) + yield Static( + "BOARDS (ctrl+n add, ctrl+d remove)", + classes="section-label", + id="boards-label", + ) + for board in self._boards: + yield Static( + f" {board['slug']}", + classes="subtitle", + id=f"board-label-{board['slug']}", + ) + yield Input( + value=board["name"], + id=f"board-name-{board['slug']}", + max_length=limits.BOARD_NAME, + ) + yield Input( + value=board["description"], + id=f"board-desc-{board['slug']}", + max_length=limits.BOARD_DESCRIPTION, + ) + yield Footer() + + def on_mount(self) -> None: + self.query_one("#edit-name", Input).focus() + + def action_add_board(self) -> None: + index = len(self._boards) + 1 + while any(board["slug"] == f"board-{index}" for board in self._boards): + index += 1 + slug = f"board-{index}" + self._boards.append( + {"slug": slug, "name": slug, "description": "", "created_at": now_iso()} + ) + + scroll = self.query_one("#edit-scroll", VerticalScroll) + label = Static(f" {slug}", classes="subtitle", id=f"board-label-{slug}") + name_input = Input( + value=slug, id=f"board-name-{slug}", max_length=limits.BOARD_NAME + ) + desc_input = Input( + value="", id=f"board-desc-{slug}", max_length=limits.BOARD_DESCRIPTION + ) + scroll.mount(label) + scroll.mount(name_input) + scroll.mount(desc_input) + name_input.focus() + + def action_remove_board(self) -> None: + if len(self._boards) <= 1: + self.notify("Must have at least one board.", severity="warning") + return + board = self._boards.pop() + slug = board["slug"] + for widget_id in ( + f"board-label-{slug}", + f"board-name-{slug}", + f"board-desc-{slug}", + ): + try: + self.query_one(f"#{widget_id}").remove() + except Exception: + pass + + def action_save(self) -> None: + self._do_save() + + @work(exclusive=True) + async def _do_save(self) -> None: + session = require_session(self) + if not session: + return + + updater = make_session_updater(self.app.session_store) + + name = self.query_one("#edit-name", Input).value.strip() + description = self.query_one("#edit-desc", Input).value.strip() + intro = self.query_one("#edit-intro", TextArea).text + + if not name: + self.notify("Name cannot be empty.", severity="error") + return + if len(intro) > limits.SITE_INTRO: + self.notify( + f"Intro too long ({len(intro)}/{limits.SITE_INTRO}).", + severity="error", + ) + return + + now = now_iso() + + try: + for board in self._boards: + board_name = self.query_one( + f"#board-name-{board['slug']}", Input + ).value.strip() + board_desc = self.query_one( + f"#board-desc-{board['slug']}", Input + ).value.strip() + await put_board_record( + self.app.http_client, + session, + board["slug"], + board_name or board["slug"], + board_desc, + board["created_at"], + updater, + ) + + current_slugs = {board["slug"] for board in self._boards} + for board in self.bbs.site.boards: + if board.slug not in current_slugs: + await delete_record( + self.app.http_client, + session, + lexicon.BOARD, + board.slug, + updater, + ) + + await put_site_record( + self.app.http_client, + session, + { + "$type": lexicon.SITE, + "name": name, + "description": description, + "intro": intro, + "boards": [board["slug"] for board in self._boards], + "createdAt": self.bbs.site.created_at or now, + "updatedAt": now, + }, + updater, + ) + self.notify("BBS updated.") + self.app.pop_screen() + except AuthError: + self.notify("Session expired. Please log in again.", severity="error") + except Exception as e: + self.notify(f"Could not update BBS: {e}", severity="error") diff --git a/tui/screens/sysop/menu.py b/tui/screens/sysop/menu.py new file mode 100644 index 0000000..fd4b820 --- /dev/null +++ b/tui/screens/sysop/menu.py @@ -0,0 +1,38 @@ +from textual.app import ComposeResult +from textual.screen import Screen +from textual.widgets import Footer, ListItem, ListView, Static + +from core.models import BBS + + +class SysopScreen(Screen): + BINDINGS = [("escape", "app.pop_screen", "back")] + + def __init__(self, bbs: BBS, handle: str) -> None: + super().__init__() + self.bbs = bbs + self.handle = handle + + def compose(self) -> ComposeResult: + yield ListView( + ListItem(Static(" Edit BBS"), name="edit"), + ListItem(Static(" Moderation"), name="moderate"), + ListItem(Static(" Delete BBS"), name="delete"), + id="sysop-menu", + ) + yield Footer() + + def on_mount(self) -> None: + self.query_one("#sysop-menu", ListView).focus() + + def on_list_view_selected(self, event: ListView.Selected) -> None: + name = event.item.name + if name == "edit": + from tui.screens.sysop.edit import SysopEditScreen + self.app.push_screen(SysopEditScreen(self.bbs, self.handle)) + elif name == "moderate": + from tui.screens.sysop.moderate import SysopModerateScreen + self.app.push_screen(SysopModerateScreen(self.bbs, self.handle)) + elif name == "delete": + from tui.screens.sysop.delete import SysopDeleteScreen + self.app.push_screen(SysopDeleteScreen(self.bbs, self.handle)) diff --git a/tui/screens/sysop/moderate.py b/tui/screens/sysop/moderate.py new file mode 100644 index 0000000..c264cb4 --- /dev/null +++ b/tui/screens/sysop/moderate.py @@ -0,0 +1,209 @@ +import asyncio + +from textual import work +from textual.app import ComposeResult +from textual.containers import VerticalScroll +from textual.screen import Screen +from textual.widgets import Footer, Input, ListItem, ListView, Static + +from core import lexicon +from core.models import AtUri, AuthError, BBS +from core.records import ( + create_ban_record, + create_hidden_record, + delete_record, + list_pds_records, +) +from core.slingshot import resolve_identities_batch, resolve_identity +from tui.util import make_session_updater +from tui.widgets.breadcrumb import Breadcrumb + + +class SysopModerateScreen(Screen): + BINDINGS = [ + ("escape", "app.pop_screen", "back"), + ("ctrl+d", "remove", "remove"), + ("ctrl+b", "add_ban", "ban"), + ("ctrl+x", "add_hide", "hide"), + ] + + def __init__(self, bbs: BBS, handle: str) -> None: + super().__init__() + self.bbs = bbs + self.handle = handle + self._ban_rkeys: dict[str, str] = {} + self._hide_rkeys: dict[str, str] = {} + + def compose(self) -> ComposeResult: + yield Breadcrumb( + ("@bbs", 3), + (self.bbs.site.name, 2), + ("sysop", 1), + ("moderation", 0), + ) + with VerticalScroll(): + yield Static("BANNED USERS", classes="section-label") + yield Input(placeholder="handle or DID to ban", id="ban-input") + yield ListView(id="ban-list") + yield Static("HIDDEN POSTS", classes="section-label") + yield Input(placeholder="at:// URI to hide", id="hide-input") + yield ListView(id="hide-list") + yield Footer() + + def on_mount(self) -> None: + self._load_data() + + @work(exclusive=True) + async def _load_data(self) -> None: + client = self.app.http_client + session = self.app.user_session + + ban_result, hide_result = await asyncio.gather( + list_pds_records(client, session["pds_url"], session["did"], lexicon.BAN), + list_pds_records(client, session["pds_url"], session["did"], lexicon.HIDE), + return_exceptions=True, + ) + + if isinstance(ban_result, BaseException): + self._ban_rkeys = {} + else: + self._ban_rkeys = { + record["value"]["did"]: AtUri.parse(record["uri"]).rkey + for record in ban_result + if record.get("value", {}).get("did") + } + + if isinstance(hide_result, BaseException): + self._hide_rkeys = {} + else: + self._hide_rkeys = { + record["value"]["uri"]: AtUri.parse(record["uri"]).rkey + for record in hide_result + if record.get("value", {}).get("uri") + } + + banned_dids = list(self._ban_rkeys.keys()) + banned_handles: dict[str, str] = {} + if banned_dids: + try: + authors = await resolve_identities_batch(client, banned_dids) + banned_handles = {did: authors[did].handle for did in authors} + except Exception: + pass + + ban_list = self.query_one("#ban-list", ListView) + ban_list.clear() + for did in banned_dids: + label = banned_handles.get(did, did) + await ban_list.append(ListItem(Static(f" {label}"), name=f"ban:{did}")) + + hide_list = self.query_one("#hide-list", ListView) + hide_list.clear() + for uri in self._hide_rkeys: + await hide_list.append(ListItem(Static(f" {uri}"), name=f"hide:{uri}")) + + if banned_dids: + ban_list.focus() + elif self._hide_rkeys: + hide_list.focus() + + def action_remove(self) -> None: + for list_id in ("ban-list", "hide-list"): + list_view = self.query_one(f"#{list_id}", ListView) + if list_view.index is not None and list_view.has_focus: + item = list_view.children[list_view.index] + if item.name: + self._do_remove(item.name, item) + return + + @work + async def _do_remove(self, key: str, item) -> None: + session = self.app.user_session + updater = make_session_updater(self.app.session_store) + + kind, _, value = key.partition(":") + try: + if kind == "ban" and value in self._ban_rkeys: + rkey = self._ban_rkeys[value] + await delete_record( + self.app.http_client, session, lexicon.BAN, rkey, updater + ) + del self._ban_rkeys[value] + self.notify(f"Unbanned {value}.") + elif kind == "hide" and value in self._hide_rkeys: + rkey = self._hide_rkeys[value] + await delete_record( + self.app.http_client, session, lexicon.HIDE, rkey, updater + ) + del self._hide_rkeys[value] + self.notify("Post unhidden.") + await item.remove() + except AuthError: + self.notify("Session expired. Please log in again.", severity="error") + except Exception: + self.notify("Could not remove record.", severity="error") + + def action_add_ban(self) -> None: + identifier = self.query_one("#ban-input", Input).value.strip() + if not identifier: + self.notify("Enter a handle or DID.", severity="warning") + return + self._do_add_ban(identifier) + + @work + async def _do_add_ban(self, identifier: str) -> None: + session = self.app.user_session + client = self.app.http_client + updater = make_session_updater(self.app.session_store) + + did = identifier + if not identifier.startswith("did:"): + try: + identity = await resolve_identity(client, identifier) + did = identity.did + except Exception: + self.notify(f"Could not resolve {identifier}.", severity="error") + return + + if did in self._ban_rkeys: + self.notify("Already banned.", severity="warning") + return + + try: + await create_ban_record(client, session, did, updater) + self.notify(f"Banned {did}.") + self.query_one("#ban-input", Input).value = "" + self._load_data() + except AuthError: + self.notify("Session expired. Please log in again.", severity="error") + except Exception: + self.notify("Could not ban user.", severity="error") + + def action_add_hide(self) -> None: + uri = self.query_one("#hide-input", Input).value.strip() + if not uri or not uri.startswith("at://"): + self.notify("Enter a valid AT-URI.", severity="warning") + return + self._do_add_hide(uri) + + @work + async def _do_add_hide(self, uri: str) -> None: + session = self.app.user_session + updater = make_session_updater(self.app.session_store) + + if uri in self._hide_rkeys: + self.notify("Already hidden.", severity="warning") + return + + try: + await create_hidden_record(self.app.http_client, session, uri, updater) + self.notify("Post hidden.") + self.query_one("#hide-input", Input).value = "" + self._load_data() + except AuthError: + self.notify("Session expired. Please log in again.", severity="error") + except Exception: + self.notify("Could not hide post.", severity="error") + + def refresh_data(self) -> None: + self._load_data() diff --git a/tui/screens/thread.py b/tui/screens/thread.py index 6862d44..8a2845a 100644 --- a/tui/screens/thread.py +++ b/tui/screens/thread.py @@ -116,57 +116,57 @@ class ThreadScreen(Screen): scroll = self.query_one("#thread-scroll") - for r in result.replies: - self._replies_map[r.uri] = r + for reply in result.replies: + self._replies_map[reply.uri] = reply # Fetch any quoted replies not already known (in parallel) missing = [ - r.quote - for r in result.replies - if r.quote and r.quote not in self._replies_map + reply.quote + for reply in result.replies + if reply.quote and reply.quote not in self._replies_map ] async def fetch_quote(uri: str): parsed = AtUri.parse(uri) - rec, author = await asyncio.gather( + record, author = await asyncio.gather( get_record(client, parsed.did, parsed.collection, parsed.rkey), resolve_identity(client, parsed.did), ) - return uri, reply_from_record(rec, author) + return uri, reply_from_record(record, author) if missing: - results = await asyncio.gather( + quote_results = await asyncio.gather( *[fetch_quote(uri) for uri in missing], return_exceptions=True, ) - for r in results: - if isinstance(r, tuple): - self._replies_map[r[0]] = r[1] + for quote_result in quote_results: + if isinstance(quote_result, tuple): + self._replies_map[quote_result[0]] = quote_result[1] - for r in result.replies: + for reply in result.replies: quote_text = None - if r.quote and r.quote in self._replies_map: - q = self._replies_map[r.quote] - body_preview = q.body[:200] + ("..." if len(q.body) > 200 else "") - quote_text = f"{q.author.handle}: {body_preview}" + if reply.quote and reply.quote in self._replies_map: + quoted = self._replies_map[reply.quote] + body_preview = quoted.body[:200] + ("..." if len(quoted.body) > 200 else "") + quote_text = f"{quoted.author.handle}: {body_preview}" await scroll.mount( Post( - author=r.author.handle, - date=r.created_at, - body=r.body, - author_did=r.author.did, - author_pds=r.author.pds, - record_uri=r.uri, + author=reply.author.handle, + date=reply.created_at, + body=reply.body, + author_did=reply.author.did, + author_pds=reply.author.pds, + record_uri=reply.uri, collection=lexicon.REPLY, - attachments=r.attachments, + attachments=reply.attachments, quote_text=quote_text, ), before=self.query_one("#page-status-bottom"), ) # Focus first reply - replies = [p for p in self.query(Post) if p.collection == lexicon.REPLY] + replies = [post for post in self.query(Post) if post.collection == lexicon.REPLY] if replies: replies[0].focus() diff --git a/tui/widgets/post.py b/tui/widgets/post.py index 649e745..b86831b 100644 --- a/tui/widgets/post.py +++ b/tui/widgets/post.py @@ -113,9 +113,9 @@ class Post(Widget, can_focus=True): if self._quote_text: yield Markdown(self._quote_text, classes="post-quote") yield Markdown(self._body, classes="post-body") - for att in self.attachments: - name = att.get("name", "file") - cid = att.get("file", {}).get("ref", {}).get("$link", "") + for attachment in self.attachments: + name = attachment.get("name", "file") + cid = attachment.get("file", {}).get("ref", {}).get("$link", "") if cid and self.author_pds and self.author_did: url = f"{self.author_pds}/xrpc/com.atproto.sync.getBlob?did={self.author_did}&cid={cid}" yield AttachmentLink(name, url)