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

excellent

Browse Source
This commit is contained in:
Tim Bendt
2025-05-02 01:16:13 -04:00
parent fe22010a68
commit 615aeda3b9
20 changed files with 313 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
maildir_gtd/actions/__pycache__/archive.cpython-311.pyc
View File

Binary file not shown.

BIN
maildir_gtd/actions/__pycache__/delete.cpython-311.pyc
View File

Binary file not shown.

BIN
maildir_gtd/actions/__pycache__/newest.cpython-311.pyc
View File

Binary file not shown.

BIN
maildir_gtd/actions/__pycache__/next.cpython-311.pyc
View File

Binary file not shown.

BIN
maildir_gtd/actions/__pycache__/oldest.cpython-311.pyc
View File

Binary file not shown.

BIN
maildir_gtd/actions/__pycache__/open.cpython-311.pyc
View File

Binary file not shown.

BIN
maildir_gtd/actions/__pycache__/previous.cpython-311.pyc
View File

Binary file not shown.

BIN
maildir_gtd/actions/__pycache__/show_message.cpython-311.pyc
View File

Binary file not shown.

41
maildir_gtd/actions/archive.py
View File

@@ -1,19 +1,36 @@
from textual.widgets import Static
import subprocess
import asyncio
import logging
from maildir_gtd.actions.next import action_next
def action_archive(app) -> None:
from textual import work
from textual.logging import TextualHandler
from textual.widgets import ListView
logging.basicConfig(
level="NOTSET",
handlers=[TextualHandler()],
)
@work(exclusive=False)
async def archive_current(app) -> None:
"""Archive the current email message."""
app.show_status(f"Archiving message {app.current_message_id}...")
try:
result = subprocess.run(
["himalaya", "message", "move", "Archives", str(app.current_message_id)],
capture_output=True,
text=True
index = app.current_message_index
logging.info("Archiving message ID: " + str(app.current_message_id))
process = await asyncio.create_subprocess_shell(
f"himalaya message move Archives {app.current_message_id}",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
if result.returncode == 0:
action_next(app) # Automatically show the next message
stdout, stderr = await process.communicate()
# app.reload_needed = True
app.show_status(f"{stdout.decode()}", "info")
logging.info(stdout.decode())
if process.returncode == 0:
await app.query_one(ListView).pop(index)
app.query_one(ListView).index = index + 1
app.action_next() # Automatically show the next message
else:
app.show_status(f"Error archiving message: {result.stderr}", "error")
app.show_status(f"Error archiving message: {stderr.decode()}", "error")
except Exception as e:
app.show_status(f"Error: {e}", "error")

33
maildir_gtd/actions/delete.py
View File

@@ -1,22 +1,25 @@
import subprocess
from textual.widgets import Static
from maildir_gtd.actions.next import action_next
import asyncio
from textual import work
from textual.widgets import ListView
def action_delete(app) -> None:
"""Delete the current email message."""
@work(exclusive=False)
async def delete_current(app) -> None:
app.show_status(f"Deleting message {app.current_message_id}...")
app.query_one("#main_content", Static).loading = True
try:
result = subprocess.run(
["himalaya", "message", "delete", str(app.current_message_id)],
capture_output=True,
text=True
index = app.current_message_index
process = await asyncio.create_subprocess_shell(
f"himalaya message delete {app.current_message_id}",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
if result.returncode == 0:
app.query_one("#main_content").loading = False
action_next(app) # Automatically show the next message
stdout, stderr = await process.communicate()
# app.reload_needed = True
app.show_status(f"{stdout.decode()}", "info")
if process.returncode == 0:
await app.query_one(ListView).pop(index)
app.query_one(ListView).index = index + 1
app.action_next() # Automatically show the next message
else:
app.show_status(f"Failed to delete message {app.current_message_id}.", "error")
app.show_status(f"Failed to delete message {app.current_message_id}. {stderr.decode()}", "error")
except Exception as e:
app.show_status(f"Error: {e}", "error")

17
maildir_gtd/actions/newest.py
View File

@@ -1,18 +1,13 @@
import subprocess
import asyncio
def action_newest(app) -> None:
async def action_newest(app) -> None:
"""Show the previous email message by finding the next lower ID from the list of envelope IDs."""
try:
result = subprocess.run(
["himalaya", "envelope", "list", "-o", "json", "-s", "9999"],
capture_output=True,
text=True
)
if result.returncode == 0:
import json
envelopes = json.loads(result.stdout)
ids = sorted((int(envelope['id']) for envelope in envelopes), reverse=True)
if (app.reload_needed):
await app.action_fetch_list()
ids = sorted((int(envelope['id']) for envelope in app.all_envelopes), reverse=True)
app.current_message_id = ids[0]
app.show_message(app.current_message_id)
return

16
maildir_gtd/actions/next.py
View File

@@ -9,20 +9,14 @@ from textual.reactive import Reactive
from textual.binding import Binding
from textual.timer import Timer
from textual.containers import ScrollableContainer, Horizontal, Vertical, Grid
import subprocess
import asyncio
def action_next(app) -> None:
async def action_next(app) -> None:
"""Show the next email message by finding the next higher ID from the list of envelope IDs."""
try:
result = subprocess.run(
["himalaya", "envelope", "list", "-o", "json", "-s", "9999"],
capture_output=True,
text=True
)
if result.returncode == 0:
import json
envelopes = json.loads(result.stdout)
ids = sorted(int(envelope['id']) for envelope in envelopes)
if (app.reload_needed):
app.action_fetch_list()
ids = sorted(int(envelope['id']) for envelope in app.all_envelopes)
for envelope_id in ids:
if envelope_id > int(app.current_message_id):
app.show_message(envelope_id)

15
maildir_gtd/actions/oldest.py
View File

@@ -1,18 +1,13 @@
import subprocess
import asyncio
def action_oldest(app) -> None:
"""Show the previous email message by finding the next lower ID from the list of envelope IDs."""
try:
result = subprocess.run(
["himalaya", "envelope", "list", "-o", "json", "-s", "9999"],
capture_output=True,
text=True
)
if result.returncode == 0:
import json
envelopes = json.loads(result.stdout)
ids = sorted((int(envelope['id']) for envelope in envelopes))
if (app.reload_needed):
app.action_fetch_list()
ids = sorted((int(envelope['id']) for envelope in app.all_envelopes))
app.current_message_id = ids[0]
app.show_message(app.current_message_id)
return

4
maildir_gtd/actions/open.py
View File

@@ -6,9 +6,9 @@ def action_open(app) -> None:
def check_id(message_id: str) -> bool:
try:
int(message_id)
app.current_message_id = message_id
app.show_message(app.current_message_id)
app.show_message(message_id)
except ValueError:
app.bell()
app.show_status("Invalid message ID. Please enter an integer.", severity="error")
return True
return False

13
maildir_gtd/actions/previous.py
View File

@@ -4,15 +4,10 @@ import subprocess
def action_previous(app) -> None:
"""Show the previous email message by finding the next lower ID from the list of envelope IDs."""
try:
result = subprocess.run(
["himalaya", "envelope", "list", "-o", "json", "-s", "9999"],
capture_output=True,
text=True
)
if result.returncode == 0:
import json
envelopes = json.loads(result.stdout)
ids = sorted((int(envelope['id']) for envelope in envelopes), reverse=True)
if (app.reload_needed):
app.action_fetch_list()
ids = sorted((int(envelope['id']) for envelope in app.all_envelopes), reverse=True)
for envelope_id in ids:
if envelope_id < int(app.current_message_id):
app.current_message_id = envelope_id

2
maildir_gtd/actions/show_message.py
View File

@@ -10,5 +10,5 @@ logging.basicConfig(
def show_message(app, message_id: int) -> None:
"""Fetch and display the email message by ID."""
app.current_message_id = message_id
logging.info("Showing message ID: " + str(message_id))
app.current_message_id = message_id

255
maildir_gtd/app.py
View File

@@ -1,33 +1,31 @@
import re
import sys
import os
from datetime import datetime # Add this import at the top of the file
from actions.newest import action_newest
from actions.oldest import action_oldest
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from datetime import datetime
import asyncio
import logging
from typing import Iterable
from textual import on
from textual.widget import Widget
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from textual import work
from textual.worker import Worker
from textual.app import App, ComposeResult, SystemCommand, RenderResult
from textual.logging import TextualHandler
from textual.screen import Screen
from textual.widgets import Header, Footer, Static, Label, Input, Button, Markdown
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, Horizontal, Vertical, Grid
import subprocess
from maildir_gtd.actions.archive import action_archive
from maildir_gtd.actions.delete import action_delete
from maildir_gtd.actions.open import action_open
from maildir_gtd.actions.show_message import show_message
from maildir_gtd.actions.next import action_next
from maildir_gtd.actions.previous import action_previous
from maildir_gtd.actions.task import action_create_task
from maildir_gtd.widgets.EnvelopeHeader import EnvelopeHeader
from textual.containers import ScrollableContainer, Grid
from actions.archive import archive_current
from actions.delete import delete_current
from actions.open import action_open
from actions.task import action_create_task
from widgets.EnvelopeHeader import EnvelopeHeader
logging.basicConfig(
level="NOTSET",
@@ -46,17 +44,23 @@ class StatusTitle(Static):
return f"{self.folder} | ID: {self.current_message_id} | [b]{self.current_message_index}[/b]/{self.total_messages}"
class EmailViewerApp(App):
"""A simple email viewer app using the Himalaya CLI."""
title = "Maildir GTD Reader"
current_message_id: Reactive[int] = reactive(1)
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")
markdown: Reactive[str] = reactive("Loading...")
header_expanded = False
header_expanded = reactive(False)
reload_needed = reactive(True)
all_envelopes = reactive([])
next_id: Reactive[int] = reactive(0)
previous_id: Reactive[int] = reactive(0)
oldest_id: Reactive[int] = reactive(0)
newest_id: Reactive[int] = reactive(0)
msg_worker: Worker | None = None
message_body_cache: dict[int, str] = {}
def get_system_commands(self, screen: Screen) -> Iterable[SystemCommand]:
yield from super().get_system_commands(screen)
@@ -68,6 +72,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)
BINDINGS = [
Binding("j", "next", "Next message"),
@@ -77,7 +82,8 @@ class EmailViewerApp(App):
Binding("o", "open", "Open message", show=False),
Binding("q", "quit", "Quit application"),
Binding("h", "toggle_header", "Toggle Envelope Header"),
Binding("t", "create_task", "Create Task")
Binding("t", "create_task", "Create Task"),
Binding("ctrl-r", "reload", "Reload message list")
]
BINDINGS.extend([
@@ -89,80 +95,162 @@ class EmailViewerApp(App):
def compose(self) -> ComposeResult:
"""Create child widgets for the app."""
# yield Header(show_clock=True)
yield StatusTitle().data_bind(EmailViewerApp.current_message_id)
yield EnvelopeHeader()
yield Markdown(id="main_content", markdown=self.markdown)
yield Grid(
ListView(ListItem(Label("All emails...")), id="list_view", initial_index=0),
ScrollableContainer(
StatusTitle().data_bind(EmailViewerApp.current_message_id),
EnvelopeHeader(),
Markdown(),
id="main_content",
)
)
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(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()
await worker.wait()
self.action_oldest()
def compute_newest_id(self) -> None:
if not self.all_envelopes:
return 0
return sorted((int(envelope['id']) for envelope in self.all_envelopes))[-1]
def compute_oldest_id(self) -> None:
if not self.all_envelopes:
return 0
return sorted((int(envelope['id']) for envelope in self.all_envelopes))[0]
def compute_next_id(self) -> None:
if not self.all_envelopes:
return 0
for envelope_id in sorted(int(envelope['id']) for envelope in self.all_envelopes):
if envelope_id > int(self.current_message_id):
return envelope_id
return self.newest_id
def compute_previous_id(self) -> None:
if not self.all_envelopes:
return 0
for envelope_id in sorted((int(envelope['id']) for envelope in self.all_envelopes), reverse=True):
if envelope_id < int(self.current_message_id):
return envelope_id
return self.oldest_id
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()
def watch_current_message_id(self, old_message_id: int, new_message_id: int) -> None:
"""Called when the current message ID changes."""
logging.info(f"Current message ID changed from {old_message_id} to {new_message_id}")
self.query_one("#main_content").loading = True
self.markdown = ""
if (new_message_id == old_message_id):
return
self.msg_worker.cancel() if self.msg_worker else None
headers = self.query_one(EnvelopeHeader)
for index, envelope in enumerate(self.all_envelopes):
if int(envelope['id']) == new_message_id:
self.current_message_index = index
headers.subject = str(envelope['subject']).strip()
headers.from_ = envelope['from']['addr']
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(ListView).index = index
break
if (self.message_body_cache.get(new_message_id)):
# If the message body is already cached, use it
msg = self.query_one(Markdown)
msg.update(self.message_body_cache[new_message_id])
return
else:
self.query_one("#main_content").loading = True
self.msg_worker = self.fetch_one_message(new_message_id)
def on_list_view_selected(self, event: ListView.Selected) -> None:
"""Called when an item in the list view is selected."""
logging.info(f"Selected item: {self.all_envelopes[event.list_view.index]}")
self.current_message_id = int(self.all_envelopes[event.list_view.index]['id'])
@work(exclusive=False)
async def fetch_one_message(self, new_message_id:int) -> None:
msg = self.query_one(Markdown)
try:
rawText = subprocess.run(
["himalaya", "message", "read", str(new_message_id)],
capture_output=True,
text=True
process = await asyncio.create_subprocess_shell(
f"himalaya message read {str(new_message_id)}",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
if rawText.returncode == 0:
stdout, stderr = await process.communicate()
logging.info(f"stdout: {stdout.decode()[0:50]}...")
if process.returncode == 0:
# Render the email content as Markdown
fixedText = rawText.stdout.replace("(https://urldefense.com/v3/", "(")
fixedText = stdout.decode().replace("(https://urldefense.com/v3/", "(")
fixedText = re.sub(r"atlOrigin.+?\)", ")", fixedText)
logging.info(f"rendering fixedText: {fixedText[0:50]}")
self.message_body_cache[new_message_id] = fixedText
await msg.update(fixedText)
self.query_one("#main_content").loading = False
self.query_one("#main_content").update(markdown = str(fixedText))
logging.info(fixedText)
result = subprocess.run(
["himalaya", "envelope", "list", "-o", "json", "-s", "9999"],
capture_output=True,
text=True
except Exception as e:
self.show_status(f"Error fetching message content: {e}", "error")
logging.error(f"Error fetching message content: {e}")
@work(exclusive=False)
async def action_fetch_list(self) -> None:
msglist = self.query_one(ListView)
try:
msglist.loading = True
process = await asyncio.create_subprocess_shell(
"himalaya envelope list -o json -s 9999",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
if result.returncode == 0:
stdout, stderr = await process.communicate()
logging.info(f"stdout: {stdout.decode()[0:50]}")
if process.returncode == 0:
import json
envelopes = json.loads(result.stdout)
envelopes = json.loads(stdout.decode())
if envelopes:
self.reload_needed = False
status = self.query_one(StatusTitle)
status.total_messages = len(envelopes)
headers = self.query_one(EnvelopeHeader)
# Find the index of the envelope that matches the current_message_id
for index, envelope in enumerate(sorted(envelopes, key=lambda x: int(x['id']))):
if int(envelope['id']) == new_message_id:
headers.subject = envelope['subject']
headers.from_ = envelope['from']['addr']
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")
status.current_message_index = index + 1 # 1-based index
break
status.update()
headers.update()
else:
self.query_one("#main_content").update("Failed to fetch the most recent message ID.")
msglist.clear()
envelopes = sorted(envelopes, key=lambda x: int(x['id']))
self.all_envelopes = envelopes
for envelope in envelopes:
item = ListItem(Label(str(envelope['subject']).strip(), classes="email_subject", markup=False))
msglist.append(item)
msglist.index = self.current_message_index
else:
self.show_status("Failed to fetch the most recent message ID.", "error")
except Exception as e:
self.query_one("#main_content").update(f"Error: {e}")
self.show_status(f"Error fetching message list: {e}", "error")
finally:
msglist.loading = False
def on_mount(self) -> None:
self.alert_timer: Timer | None = None # Timer to throttle alerts
self.theme = "monokai"
self.title = "MaildirGTD"
# self.watch(self.query_one(StatusTitle), "current_message_id", update_progress)
# Fetch the ID of the most recent message using the Himalaya CLI
self.action_oldest()
def show_message(self, message_id: int) -> None:
show_message(self, message_id)
self.current_message_id = message_id
def show_status(self, message: str, severity: str = "information") -> None:
"""Display a status message using the built-in notify function."""
self.notify(message, title="Status", severity=severity, timeout=1, markup=True)
self.notify(message, title="Status", severity=severity, timeout=1.6, markup=True)
def action_toggle_header(self) -> None:
"""Toggle the visibility of the EnvelopeHeader panel."""
@@ -170,18 +258,23 @@ class EmailViewerApp(App):
header.styles.height = "1" if self.header_expanded else "auto"
self.header_expanded = not self.header_expanded
def action_next(self) -> None:
action_next(self)
self.show_message(self.next_id)
self.action_fetch_list() if self.reload_needed else None
def action_previous(self) -> None:
action_previous(self)
self.action_fetch_list() if self.reload_needed else None
self.show_message(self.previous_id)
def action_delete(self) -> None:
action_delete(self)
self.all_envelopes.remove(self.all_envelopes[self.current_message_index])
self.message_body_cache.pop(self.current_message_id, None)
delete_current(self)
def action_archive(self) -> None:
action_archive(self)
self.all_envelopes.remove(self.all_envelopes[self.current_message_index])
self.message_body_cache.pop(self.current_message_id, None)
archive_current(self)
def action_open(self) -> None:
action_open(self)
@@ -211,10 +304,12 @@ class EmailViewerApp(App):
self.exit()
def action_oldest(self) -> None:
action_oldest(self)
self.action_fetch_list() if self.reload_needed else None
self.show_message(self.oldest_id)
def action_newest(self) -> None:
action_newest(self)
self.action_fetch_list() if self.reload_needed else None
self.show_message(self.newest_id)
if __name__ == "__main__":
app = EmailViewerApp()

22
maildir_gtd/email_viewer.tcss
View File

@@ -29,10 +29,28 @@ EnvelopeHeader {
width: 100%;
height: 1;
tint: $primary 10%;
}
#main_content {
Markdown {
padding: 1 2;
}
ListView {
dock: left;
width: 30%;
height: 100%;
padding: 0;
}
.email_subject {
width: 100%;
padding: 0
}
.header_key {
tint: gray 20%;
}
.header_value {
padding:0 1 0 0;
}

56
maildir_gtd/widgets/EnvelopeHeader.py
View File

@@ -1,8 +1,9 @@
from textual.reactive import Reactive
from textual.app import RenderResult
from textual.app import RenderResult, ComposeResult
from textual.widgets import Static, Label
from textual.containers import Vertical, Horizontal, Container, ScrollableContainer
class EnvelopeHeader(Static):
class EnvelopeHeader(ScrollableContainer):
subject = Reactive("")
from_ = Reactive("")
@@ -15,12 +16,51 @@ class EnvelopeHeader(Static):
def on_mount(self) -> None:
"""Mount the header."""
def render(self) -> RenderResult:
return f"[b]{self.subject}[/b] [dim]({self.date})[/] \r\n" \
f"[dim]From:[/dim] {self.from_} [dim]To:[/dim] {self.to} \r\n" \
f"[dim]Date:[/dim] {self.date} \r\n" \
f"[dim]CC:[/dim] {self.cc} \r\n" \
f"[dim]BCC:[/dim] {self.bcc} \r\n" \
def compose(self) -> ComposeResult:
yield Horizontal(
Label("Subject:", classes="header_key"),
Label(self.subject, classes="header_value", markup=False, id="subject"),
Label("Date:", classes="header_key"),
Label(self.date, classes="header_value", markup=False, id="date"),
)
# yield Horizontal(
# Label("From:", classes="header_key"),
# Label(self.from_, classes="header_value", markup=False, id="from"),
# )
# yield Horizontal(
# Label("To:", classes="header_key"),
# Label(self.to, classes="header_value", markup=False, id="to"),
# )
# yield Horizontal(
# )
# yield Horizontal(
# Label("CC:", classes="header_key"),
# Label(self.cc, classes="header_value", markup=False, id="cc"),
# )
def watch_subject(self, subject: str) -> None:
"""Watch the subject for changes."""
self.query_one("#subject").update(subject)
# def watch_to(self, to: str) -> None:
# """Watch the to field for changes."""
# self.query_one("#to").update(to)
# def watch_from(self, from_: str) -> None:
# """Watch the from field for changes."""
# self.query_one("#from").update(from_)
def watch_date(self, date: str) -> None:
"""Watch the date for changes."""
self.query_one("#date").update(date)
# def watch_cc(self, cc: str) -> None:
# """Watch the cc field for changes."""
# self.query_one("#cc").update(cc)

BIN
maildir_gtd/widgets/__pycache__/EnvelopeHeader.cpython-311.pyc
View File

Binary file not shown.