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__), ".."))) import logging from typing import Iterable from textual import on from textual.widget import Widget 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.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 logging.basicConfig( level="NOTSET", handlers=[TextualHandler()], ) class StatusTitle(Static): total_messages: Reactive[int] = reactive(0) current_message_index: Reactive[int] = reactive(0) current_message_id: Reactive[int] = reactive(1) folder: Reactive[str] = reactive("INBOX") def render(self) -> RenderResult: 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" folder = reactive("INBOX") markdown: Reactive[str] = reactive("Loading...") def get_system_commands(self, screen: Screen) -> Iterable[SystemCommand]: yield from super().get_system_commands(screen) yield SystemCommand("Next Message", "Navigate to Next ID", self.action_next) yield SystemCommand("Previous Message", "Navigate to Previous ID", self.action_previous) yield SystemCommand("Delete Message", "Delete the current message", self.action_delete) yield SystemCommand("Archive Message", "Archive the current message", self.action_archive) yield SystemCommand("Open Message", "Open a specific message by ID", self.action_open) 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) BINDINGS = [ Binding("j", "next", "Next message"), Binding("k", "previous", "Previous message"), Binding("#", "delete", "Delete message"), Binding("e", "archive", "Archive message"), Binding("o", "open", "Open message", show=False), Binding("q", "quit", "Quit application"), Binding("t", "create_task", "Create Task") ] 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 Header(show_clock=True) yield StatusTitle().data_bind(EmailViewerApp.current_message_id) yield EnvelopeHeader() yield Markdown(id="main_content", markdown=self.markdown) yield Footer() 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 try: rawText = subprocess.run( ["himalaya", "message", "read", str(new_message_id)], capture_output=True, text=True ) if rawText.returncode == 0: # Render the email content as Markdown fixedText = rawText.stdout.replace("(https://urldefense.com/v3/", "(") fixedText = re.sub(r"atlOrigin.+?\)", ")", 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 ) if result.returncode == 0: import json envelopes = json.loads(result.stdout) if envelopes: 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.") except Exception as e: self.query_one("#main_content").update(f"Error: {e}") 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) 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) def action_next(self) -> None: action_next(self) def action_previous(self) -> None: action_previous(self) def action_delete(self) -> None: action_delete(self) def action_archive(self) -> None: action_archive(self) def action_open(self) -> None: action_open(self) def action_create_task(self) -> None: action_create_task(self) def action_scroll_down(self) -> None: """Scroll the main content down.""" self.query_one("#main_content").scroll_down() def action_scroll_up(self) -> None: """Scroll the main content up.""" self.query_one("#main_content").scroll_up() def action_scroll_page_down(self) -> None: """Scroll the main content down by a page.""" self.query_one("#main_content").scroll_page_down() def action_scroll_page_up(self) -> None: """Scroll the main content up by a page.""" self.query_one("#main_content").scroll_page_up() def action_quit(self) -> None: """Quit the application.""" self.exit() def action_oldest(self) -> None: action_oldest(self) def action_newest(self) -> None: action_newest(self) if __name__ == "__main__": app = EmailViewerApp() app.run()