146 lines
5.3 KiB
Python
146 lines
5.3 KiB
Python
import re
|
|
import asyncio
|
|
import logging
|
|
from functools import lru_cache
|
|
|
|
from textual import work
|
|
from textual.app import ComposeResult
|
|
from textual.widgets import Label, Markdown
|
|
from textual.containers import ScrollableContainer
|
|
|
|
from widgets.EnvelopeHeader import EnvelopeHeader
|
|
|
|
class ContentContainer(ScrollableContainer):
|
|
"""A custom container that can switch between plaintext and markdown rendering."""
|
|
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.plaintext_mode = True
|
|
self.markup_worker = None
|
|
self.current_text = ""
|
|
self.current_id = None
|
|
self.message_cache = dict()
|
|
# LRU cache with a max size of 100 messages
|
|
|
|
|
|
def compose(self) -> ComposeResult:
|
|
"""Compose the container with a label for plaintext and markdown for rich content."""
|
|
yield EnvelopeHeader()
|
|
yield Label(id="plaintext_content", markup=False)
|
|
yield Markdown(id="markdown_content", classes="hidden")
|
|
|
|
def update_header(self, subject: str = "", date: str = "", from_: str = "", to: str = "", cc: str = "", bcc: str = "") -> None:
|
|
header = self.query_one(EnvelopeHeader)
|
|
header.subject = subject
|
|
header.date = date
|
|
header.from_ = from_
|
|
header.to = to
|
|
header.cc = cc
|
|
header.bcc = bcc
|
|
|
|
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 display_content(self, message_id: int) -> None:
|
|
"""Display content for the given message ID."""
|
|
self.current_id = message_id
|
|
|
|
# Show loading state
|
|
self.loading = True
|
|
# Check if the message is already cached
|
|
if message_id in self.message_cache:
|
|
self.current_text = self.message_cache[message_id]
|
|
plaintext = self.query_one("#plaintext_content", Label)
|
|
plaintext.update(self.current_text)
|
|
if not self.plaintext_mode:
|
|
# We're in markdown mode, so render the markdown
|
|
self.render_markdown()
|
|
else:
|
|
# Hide markdown, show plaintext
|
|
plaintext.remove_class("hidden")
|
|
self.query_one("#markdown_content").add_class("hidden")
|
|
|
|
self.loading = False
|
|
return self.current_text
|
|
else:
|
|
# Get message body (from cache or fetch new)
|
|
self.get_message_body(message_id)
|
|
|
|
|
|
@work(exclusive=True)
|
|
async def get_message_body(self, message_id: int) -> str:
|
|
"""Fetch the message body from Himalaya CLI."""
|
|
|
|
|
|
try:
|
|
process = await asyncio.create_subprocess_shell(
|
|
f"himalaya message read {str(message_id)} -p",
|
|
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:
|
|
# Process the email content
|
|
fixed_text = stdout.decode().replace("https://urldefense.com/v3/", "")
|
|
fixed_text = re.sub(r"atlOrigin.+?\w", "", fixed_text)
|
|
logging.info(f"rendering fixedText: {fixed_text[0:50]}")
|
|
|
|
self.current_text = fixed_text
|
|
self.message_cache[message_id] = fixed_text
|
|
|
|
# Update the plaintext content
|
|
plaintext = self.query_one("#plaintext_content", Label)
|
|
plaintext.update(fixed_text)
|
|
|
|
if not self.plaintext_mode:
|
|
# We're in markdown mode, so render the markdown
|
|
self.render_markdown()
|
|
else:
|
|
# Hide markdown, show plaintext
|
|
plaintext.remove_class("hidden")
|
|
self.query_one("#markdown_content").add_class("hidden")
|
|
|
|
self.loading = False
|
|
else:
|
|
logging.error(f"Error fetching message: {stderr.decode()}")
|
|
return f"Error fetching message content: {stderr.decode()}"
|
|
except Exception as e:
|
|
logging.error(f"Error fetching message content: {e}")
|
|
return f"Error fetching message content: {e}"
|
|
|
|
async def render_markdown(self) -> None:
|
|
"""Render the markdown content asynchronously."""
|
|
if self.markup_worker:
|
|
self.markup_worker.cancel()
|
|
|
|
markdown = self.query_one("#markdown_content", Markdown)
|
|
plaintext = self.query_one("#plaintext_content", Label)
|
|
|
|
await markdown.update(self.current_text)
|
|
|
|
# Show markdown, hide plaintext
|
|
markdown.remove_class("hidden")
|
|
plaintext.add_class("hidden")
|
|
|
|
async def toggle_mode(self) -> None:
|
|
"""Toggle between plaintext and markdown mode."""
|
|
self.plaintext_mode = not self.plaintext_mode
|
|
|
|
if self.plaintext_mode:
|
|
# Switch to plaintext
|
|
self.query_one("#plaintext_content").remove_class("hidden")
|
|
self.query_one("#markdown_content").add_class("hidden")
|
|
else:
|
|
# Switch to markdown
|
|
await self.render_markdown()
|
|
|
|
return self.plaintext_mode
|
|
|
|
|