222 lines
8.6 KiB
Python
222 lines
8.6 KiB
Python
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...")
|
|
header_expanded = False
|
|
|
|
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("h", "toggle_header", "Toggle Envelope Header"),
|
|
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_toggle_header(self) -> None:
|
|
"""Toggle the visibility of the EnvelopeHeader panel."""
|
|
header = self.query_one(EnvelopeHeader)
|
|
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)
|
|
|
|
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()
|