sirtimbly/luk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

layout of panels tweaked

Browse Source
This commit is contained in:
Tim Bendt
2025-05-04 14:50:58 -06:00
parent 08eb4ee0cf
commit b26674ff4e
9 changed files with 201 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
maildir_gtd/app.py
View File

@@ -19,7 +19,7 @@ from textual.widgets import Footer, Static, Label, Markdown, ListView, ListItem
from textual.reactive import reactive, Reactive
from textual.binding import Binding
from textual.timer import Timer
from textual.containers import ScrollableContainer, Grid
from textual.containers import ScrollableContainer, Grid, Vertical, Horizontal
from actions.archive import archive_current
from actions.delete import delete_current
@@ -48,7 +48,6 @@ class EmailViewerApp(App):
"""A simple email viewer app using the Himalaya CLI."""
CSS_PATH = "email_viewer.tcss"
title = "Maildir GTD Reader"
current_message_id: Reactive[int] = reactive(0)
current_message_index: Reactive[int] = reactive(0)
folder = reactive("INBOX")
@@ -61,6 +60,8 @@ class EmailViewerApp(App):
newest_id: Reactive[int] = reactive(0)
msg_worker: Worker | None = None
message_body_cache: dict[int, str] = {}
total_messages: Reactive[int] = reactive(0)
status_title = reactive("Message View")
def get_system_commands(self, screen: Screen) -> Iterable[SystemCommand]:
yield from super().get_system_commands(screen)
@@ -72,7 +73,7 @@ class EmailViewerApp(App):
yield SystemCommand("Create Task", "Create a task using the task CLI", self.action_create_task)
yield SystemCommand("Oldest Message", "Show the oldest message", self.action_oldest)
yield SystemCommand("Newest Message", "Show the newest message", self.action_newest)
yield SystemCommand("Reload", "Reload the message list", self.action_fetch_list)
yield SystemCommand("Reload", "Reload the message list", self.fetch_envelopes)
BINDINGS = [
Binding("j", "next", "Next message"),
@@ -83,40 +84,60 @@ class EmailViewerApp(App):
Binding("q", "quit", "Quit application"),
Binding("h", "toggle_header", "Toggle Envelope Header"),
Binding("t", "create_task", "Create Task"),
Binding("ctrl-r", "reload", "Reload message list")
Binding("%", "reload", "Reload message list"),
Binding("1", "focus_1", "Focus Accounts Panel"),
Binding("2", "focus_2", "Focus Folders Panel"),
Binding("3", "focus_3", "Focus Envelopes Panel")
]
BINDINGS.extend([
Binding("down", "scroll_down", "Scroll down"),
Binding("up", "scroll_up", "Scroll up"),
Binding("space", "scroll_page_down", "Scroll page down"),
Binding("b", "scroll_page_up", "Scroll page up")
])
def compose(self) -> ComposeResult:
"""Create child widgets for the app."""
yield Grid(
ListView(ListItem(Label("All emails...")), id="list_view", initial_index=0),
ScrollableContainer(
StatusTitle().data_bind(EmailViewerApp.current_message_id),
yield Horizontal(
Vertical(
ListView(ListItem(Label("All emails...")), id="envelopes_list", classes="list_view", initial_index=0),
ListView(id="accounts_list", classes="list_view"),
ListView(id="folders_list", classes="list_view"),
id="sidebar"
),
ScrollableContainer(
EnvelopeHeader(),
Markdown(),
id="main_content",
)
)
id="main_content"
),
id="outer-wrapper"
)
yield Footer()
async def on_mount(self) -> None:
self.alert_timer: Timer | None = None # Timer to throttle alerts
self.theme = "monokai"
self.title = "MaildirGTD"
self.query_one("#main_content").border_title = self.status_title
self.query_one("#envelopes_list").border_title = "\[1] Emails"
self.query_one("#accounts_list").border_title = "\[2] Accounts"
self.query_one("#folders_list").border_title = "\[3] Folders"
# self.query_one(ListView).data_bind(index=EmailViewerApp.current_message_index)
# self.watch(self.query_one(StatusTitle), "current_message_id", update_progress)
# Fetch the ID of the most recent message using the Himalaya CLI
worker = self.action_fetch_list()
self.fetch_accounts()
self.fetch_folders()
worker = self.fetch_envelopes()
await worker.wait()
self.query_one("#envelopes_list").focus()
self.action_oldest()
def compute_status_title(self) -> None:
return f"Message ID: {self.current_message_id} | [b]{self.current_message_index}[/b]/{self.total_messages}"
def watch_status_title(self, old_status_title: str, new_status_title: str) -> None:
self.query_one("#main_content").border_title = new_status_title
def compute_newest_id(self) -> None:
if not self.all_envelopes:
return 0
@@ -146,7 +167,7 @@ class EmailViewerApp(App):
def watch_reload_needed(self, old_reload_needed: bool, new_reload_needed: bool) -> None:
logging.info(f"Reload needed: {new_reload_needed}")
if (old_reload_needed == False and new_reload_needed == True):
self.action_fetch_list()
self.fetch_envelopes()
def watch_current_message_id(self, old_message_id: int, new_message_id: int) -> None:
@@ -165,7 +186,7 @@ class EmailViewerApp(App):
headers.to = envelope['to']['addr']
headers.date = datetime.strptime(envelope['date'].replace("+00:00", ""), "%Y-%m-%d %H:%M").strftime("%a %b %d %H:%M")
headers.cc = envelope['cc']['addr'] if 'cc' in envelope else ""
self.query_one(StatusTitle).current_message_index = index
# self.query_one(StatusTitle).current_message_index = index
self.query_one(ListView).index = index
break
@@ -211,8 +232,8 @@ class EmailViewerApp(App):
logging.error(f"Error fetching message content: {e}")
@work(exclusive=False)
async def action_fetch_list(self) -> None:
msglist = self.query_one(ListView)
async def fetch_envelopes(self) -> None:
msglist = self.query_one("#envelopes_list")
try:
msglist.loading = True
process = await asyncio.create_subprocess_shell(
@@ -227,8 +248,7 @@ class EmailViewerApp(App):
envelopes = json.loads(stdout.decode())
if envelopes:
self.reload_needed = False
status = self.query_one(StatusTitle)
status.total_messages = len(envelopes)
self.total_messages = len(envelopes)
msglist.clear()
envelopes = sorted(envelopes, key=lambda x: int(x['id']))
self.all_envelopes = envelopes
@@ -243,7 +263,55 @@ class EmailViewerApp(App):
finally:
msglist.loading = False
@work(exclusive=False)
async def fetch_accounts(self) -> None:
accounts_list = self.query_one("#accounts_list")
try:
accounts_list.loading = True
process = await asyncio.create_subprocess_shell(
"himalaya account list -o json",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
logging.info(f"stdout: {stdout.decode()[0:50]}")
if process.returncode == 0:
import json
accounts = json.loads(stdout.decode())
if accounts:
for account in accounts:
item = ListItem(Label(str(account['name']).strip(), classes="account_name", markup=False))
accounts_list.append(item)
except Exception as e:
self.show_status(f"Error fetching account list: {e}", "error")
finally:
accounts_list.loading = False
@work(exclusive=False)
async def fetch_folders(self) -> None:
folders_list = self.query_one("#folders_list")
folders_list.clear()
folders_list.append(ListItem(Label("INBOX", classes="folder_name", markup=False)))
try:
folders_list.loading = True
process = await asyncio.create_subprocess_shell(
"himalaya folder list -o json",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
logging.info(f"stdout: {stdout.decode()[0:50]}")
if process.returncode == 0:
import json
folders = json.loads(stdout.decode())
if folders:
for folder in folders:
item = ListItem(Label(str(folder['name']).strip(), classes="folder_name", markup=False))
folders_list.append(item)
except Exception as e:
self.show_status(f"Error fetching folder list: {e}", "error")
finally:
folders_list.loading = False
def show_message(self, message_id: int) -> None:
self.current_message_id = message_id
@@ -260,22 +328,22 @@ class EmailViewerApp(App):
def action_next(self) -> None:
self.show_message(self.next_id)
self.action_fetch_list() if self.reload_needed else None
self.fetch_envelopes() if self.reload_needed else None
def action_previous(self) -> None:
self.action_fetch_list() if self.reload_needed else None
self.fetch_envelopes() if self.reload_needed else None
self.show_message(self.previous_id)
def action_delete(self) -> None:
self.all_envelopes.remove(self.all_envelopes[self.current_message_index])
self.message_body_cache.pop(self.current_message_id, None)
self.query_one(StatusTitle).total_messages = len(self.all_envelopes)
self.total_messages = len(self.all_envelopes)
delete_current(self)
def action_archive(self) -> None:
self.all_envelopes.remove(self.all_envelopes[self.current_message_index])
self.message_body_cache.pop(self.current_message_id, None)
self.query_one(StatusTitle).total_messages = len(self.all_envelopes)
self.total_messages = len(self.all_envelopes)
archive_current(self)
def action_open(self) -> None:
@@ -306,13 +374,22 @@ class EmailViewerApp(App):
self.exit()
def action_oldest(self) -> None:
self.action_fetch_list() if self.reload_needed else None
self.fetch_envelopes() if self.reload_needed else None
self.show_message(self.oldest_id)
def action_newest(self) -> None:
self.action_fetch_list() if self.reload_needed else None
self.fetch_envelopes() if self.reload_needed else None
self.show_message(self.newest_id)
def action_focus_1(self) -> None:
self.query_one("#envelopes_list").focus()
def action_focus_2(self) -> None:
self.query_one("#accounts_list").focus()
def action_focus_3(self) -> None:
self.query_one("#folders_list").focus()
if __name__ == "__main__":
app = EmailViewerApp()
app.run()