Add live search panel with debounced Himalaya search and help modal

This commit is contained in:
Bendt
2025-12-19 14:31:21 -05:00
parent 0cd7cf6984
commit 8be4b4785c
3 changed files with 424 additions and 53 deletions

View File

@@ -4,6 +4,7 @@ from .widgets.ContentContainer import ContentContainer
from .widgets.EnvelopeListItem import EnvelopeListItem, GroupHeader from .widgets.EnvelopeListItem import EnvelopeListItem, GroupHeader
from .screens.LinkPanel import LinkPanel from .screens.LinkPanel import LinkPanel
from .screens.ConfirmDialog import ConfirmDialog from .screens.ConfirmDialog import ConfirmDialog
from .screens.SearchPanel import SearchPanel
from .actions.task import action_create_task from .actions.task import action_create_task
from .actions.open import action_open from .actions.open import action_open
from .actions.delete import delete_current from .actions.delete import delete_current
@@ -11,7 +12,6 @@ from src.services.taskwarrior import client as taskwarrior_client
from src.services.himalaya import client as himalaya_client from src.services.himalaya import client as himalaya_client
from src.utils.shared_config import get_theme_name from src.utils.shared_config import get_theme_name
from src.utils.ipc import IPCListener, IPCMessage from src.utils.ipc import IPCListener, IPCMessage
from src.utils.search import SearchScreen
from textual.containers import Container, ScrollableContainer, Vertical, Horizontal from textual.containers import Container, ScrollableContainer, Vertical, Horizontal
from textual.timer import Timer from textual.timer import Timer
from textual.binding import Binding from textual.binding import Binding
@@ -75,6 +75,8 @@ class EmailViewerApp(App):
selected_messages: Reactive[set[int]] = reactive(set()) selected_messages: Reactive[set[int]] = reactive(set())
main_content_visible: Reactive[bool] = reactive(True) main_content_visible: Reactive[bool] = reactive(True)
search_query: Reactive[str] = reactive("") # Current search filter search_query: Reactive[str] = reactive("") # Current search filter
search_mode: Reactive[bool] = reactive(False) # True when showing search results
_cached_envelopes: List[Dict[str, Any]] = [] # Cached envelopes before search
def get_system_commands(self, screen: Screen) -> Iterable[SystemCommand]: def get_system_commands(self, screen: Screen) -> Iterable[SystemCommand]:
yield from super().get_system_commands(screen) yield from super().get_system_commands(screen)
@@ -133,6 +135,7 @@ class EmailViewerApp(App):
) )
def compose(self) -> ComposeResult: def compose(self) -> ComposeResult:
yield SearchPanel(id="search_panel")
yield Horizontal( yield Horizontal(
Vertical( Vertical(
ListView( ListView(
@@ -879,13 +882,30 @@ class EmailViewerApp(App):
self._update_list_view_subtitle() self._update_list_view_subtitle()
def action_clear_selection(self) -> None: def action_clear_selection(self) -> None:
"""Clear all selected messages.""" """Clear all selected messages and exit search mode."""
if self.selected_messages: if self.selected_messages:
self.selected_messages.clear() self.selected_messages.clear()
self.refresh_list_view_items() # Refresh all items to uncheck checkboxes self.refresh_list_view_items() # Refresh all items to uncheck checkboxes
self._update_list_view_subtitle() self._update_list_view_subtitle()
if self.search_query:
# Exit search mode if active
if self.search_mode:
search_panel = self.query_one("#search_panel", SearchPanel)
search_panel.hide()
self.search_mode = False
self.search_query = "" self.search_query = ""
# Restore cached envelopes
if self._cached_envelopes:
self.message_store.envelopes = self._cached_envelopes
self._cached_envelopes = []
self._populate_list_view()
# Restore envelope list title
sort_indicator = "" if self.sort_order_ascending else ""
self.query_one(
"#envelopes_list"
).border_title = f"1⃣ Emails {sort_indicator}"
self._update_list_view_subtitle() self._update_list_view_subtitle()
def action_oldest(self) -> None: def action_oldest(self) -> None:
@@ -897,32 +917,49 @@ class EmailViewerApp(App):
self.show_message(self.message_store.get_newest_id()) self.show_message(self.message_store.get_newest_id())
def action_search(self) -> None: def action_search(self) -> None:
"""Open search dialog to search messages via Himalaya.""" """Open the search panel."""
search_panel = self.query_one("#search_panel", SearchPanel)
if not search_panel.is_visible:
# Cache current envelopes before searching
self._cached_envelopes = list(self.message_store.envelopes)
search_panel.show(self.search_query)
def handle_search_result(query: str | None) -> None: def on_search_panel_search_requested(
if query is None: self, event: SearchPanel.SearchRequested
return # User cancelled ) -> None:
if not query.strip(): """Handle live search request from search panel."""
# Empty query - clear search self._perform_search(event.query, focus_results=False)
self.search_query = ""
self._update_list_view_subtitle()
return
self.search_query = query
self._perform_search(query)
self.push_screen( def on_search_panel_search_confirmed(
SearchScreen( self, event: SearchPanel.SearchConfirmed
title="Search Messages", ) -> None:
placeholder="Search by sender, recipient, subject, or body...", """Handle confirmed search (Enter key) - search and focus results."""
initial_value=self.search_query, self._perform_search(event.query, focus_results=True)
),
handle_search_result, def on_search_panel_search_cancelled(
) self, event: SearchPanel.SearchCancelled
) -> None:
"""Handle search cancellation - restore previous envelope list."""
self.search_mode = False
self.search_query = ""
# Restore cached envelopes
if self._cached_envelopes:
self.message_store.envelopes = self._cached_envelopes
self._cached_envelopes = []
self._populate_list_view()
# Restore envelope list title
sort_indicator = "" if self.sort_order_ascending else ""
self.query_one("#envelopes_list").border_title = f"1⃣ Emails {sort_indicator}"
self._update_list_view_subtitle()
self.query_one("#envelopes_list").focus()
@work(exclusive=True) @work(exclusive=True)
async def _perform_search(self, query: str) -> None: async def _perform_search(self, query: str, focus_results: bool = False) -> None:
"""Perform search using Himalaya and select first result.""" """Perform search using Himalaya and display results in envelope list."""
self.show_status(f"Searching for '{query}'...") search_panel = self.query_one("#search_panel", SearchPanel)
search_panel.update_status(-1, searching=True)
folder = self.folder if self.folder else None folder = self.folder if self.folder else None
account = self.current_account if self.current_account else None account = self.current_account if self.current_account else None
@@ -932,43 +969,74 @@ class EmailViewerApp(App):
) )
if not success: if not success:
search_panel.update_status(0, searching=False)
self.show_status("Search failed", "error") self.show_status("Search failed", "error")
return return
# Update search panel status
search_panel.update_status(len(results), searching=False)
if not results: if not results:
self.show_status(f"No messages found matching '{query}'") # Clear the envelope list and show "no results"
self._display_search_results([], query)
return return
# Get the first result's ID self.search_query = query
first_result = results[0] self.search_mode = True
result_id = int(first_result.get("id", 0)) self._display_search_results(results, query)
if result_id == 0: if focus_results:
self.show_status("Search returned invalid result", "error") # Focus the main content and select first result
return if results:
first_id = int(results[0].get("id", 0))
# Find this ID in our current envelope list and select it if first_id:
metadata = self.message_store.get_metadata(result_id) self.current_message_id = first_id
if metadata:
# Message is in current view - select it
self.current_message_id = result_id
self.current_message_index = metadata["index"]
# Update list view selection
list_view = self.query_one("#envelopes_list", ListView)
list_view.index = metadata["index"]
self.show_status(f"Found {len(results)} message(s) - showing first match")
self.action_focus_4()
else:
# Message not in current view (maybe filtered or not loaded)
# Just open it directly
self.current_message_id = result_id
self.show_status(
f"Found {len(results)} message(s) - ID {result_id} (not in current list)"
)
self.action_focus_4() self.action_focus_4()
def _display_search_results(
self, results: List[Dict[str, Any]], query: str
) -> None:
"""Display search results in the envelope list with a header."""
envelopes_list = self.query_one("#envelopes_list", ListView)
envelopes_list.clear()
config = get_config()
# Add search results header
header_label = f"Search: '{query}' ({len(results)} result{'s' if len(results) != 1 else ''})"
envelopes_list.append(ListItem(GroupHeader(label=header_label)))
# Create a temporary message store for search results
search_store = MessageStore()
search_store.load(results, self.sort_order_ascending)
# Store for navigation (replace main store temporarily)
self.message_store.envelopes = search_store.envelopes
self.total_messages = len(results)
for item in search_store.envelopes:
if item and item.get("type") == "header":
envelopes_list.append(ListItem(GroupHeader(label=item["label"])))
elif item:
message_id = int(item.get("id", 0))
is_selected = message_id in self.selected_messages
envelope_widget = EnvelopeListItem(
envelope=item,
config=config.envelope_display,
is_selected=is_selected,
)
envelopes_list.append(ListItem(envelope_widget))
# Update border title to show search mode
sort_indicator = "" if self.sort_order_ascending else ""
self.query_one(
"#envelopes_list"
).border_title = f"Search Results {sort_indicator}"
# Select first result if available
if len(envelopes_list.children) > 1:
envelopes_list.index = 1 # Skip header
def action_focus_1(self) -> None: def action_focus_1(self) -> None:
self.query_one("#envelopes_list").focus() self.query_one("#envelopes_list").focus()

View File

@@ -0,0 +1,300 @@
"""Docked search panel for mail app with live search.
Provides a search input docked to the top of the window with:
- Live search with 1 second debounce
- Cancel button to restore previous state
- Help button showing Himalaya search syntax
"""
from typing import Optional, List, Dict, Any
from textual.app import ComposeResult
from textual.binding import Binding
from textual.containers import Horizontal, Vertical
from textual.message import Message
from textual.reactive import reactive
from textual.screen import ModalScreen
from textual.timer import Timer
from textual.widget import Widget
from textual.widgets import Button, Input, Label, Static
HIMALAYA_SEARCH_HELP = """
## Himalaya Search Query Syntax
A filter query is composed of operators and conditions:
### Operators
- `not <condition>` - filter envelopes that do NOT match the condition
- `<condition> and <condition>` - filter envelopes matching BOTH conditions
- `<condition> or <condition>` - filter envelopes matching EITHER condition
### Conditions
- `date <yyyy-mm-dd>` - match the given date
- `before <yyyy-mm-dd>` - date strictly before the given date
- `after <yyyy-mm-dd>` - date strictly after the given date
- `from <pattern>` - senders matching the pattern
- `to <pattern>` - recipients matching the pattern
- `subject <pattern>` - subject matching the pattern
- `body <pattern>` - text body matching the pattern
- `flag <flag>` - envelopes with the given flag (e.g., `seen`, `flagged`)
### Examples
- `from john` - emails from anyone named John
- `subject meeting and after 2025-01-01` - meetings after Jan 1st
- `not flag seen` - unread emails
- `from boss or from manager` - emails from boss or manager
- `body urgent and before 2025-12-01` - urgent emails before Dec 1st
### Sort Query
Start with `order by`, followed by:
- `date [asc|desc]`
- `from [asc|desc]`
- `to [asc|desc]`
- `subject [asc|desc]`
Example: `from john order by date desc`
""".strip()
class SearchHelpModal(ModalScreen[None]):
"""Modal showing Himalaya search syntax help."""
DEFAULT_CSS = """
SearchHelpModal {
align: center middle;
}
SearchHelpModal > Vertical {
width: 80;
max-width: 90%;
height: auto;
max-height: 80%;
border: solid $primary;
background: $surface;
padding: 1 2;
}
SearchHelpModal > Vertical > Static {
height: auto;
max-height: 100%;
}
SearchHelpModal > Vertical > Horizontal {
height: auto;
align: center middle;
margin-top: 1;
}
"""
BINDINGS = [
Binding("escape", "close", "Close"),
]
def compose(self) -> ComposeResult:
from textual.widgets import Markdown
with Vertical():
yield Markdown(HIMALAYA_SEARCH_HELP)
with Horizontal():
yield Button("Close", variant="primary", id="close-btn")
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "close-btn":
self.dismiss(None)
def action_close(self) -> None:
self.dismiss(None)
class SearchPanel(Widget):
"""Docked search panel with live search capability."""
DEFAULT_CSS = """
SearchPanel {
dock: top;
height: auto;
width: 100%;
background: $surface;
border-bottom: solid $primary;
padding: 0 1;
display: none;
}
SearchPanel.visible {
display: block;
}
SearchPanel > Horizontal {
height: auto;
width: 100%;
align: left middle;
}
SearchPanel .search-label {
width: auto;
padding: 0 1;
color: $primary;
}
SearchPanel Input {
width: 1fr;
margin: 0 1;
}
SearchPanel Button {
width: auto;
min-width: 8;
margin: 0 0 0 1;
}
SearchPanel .search-status {
width: auto;
padding: 0 1;
color: $text-muted;
}
"""
BINDINGS = [
Binding("escape", "cancel", "Cancel search", show=False),
]
# Reactive to track search state
is_searching: reactive[bool] = reactive(False)
result_count: reactive[int] = reactive(-1) # -1 = no search yet
class SearchRequested(Message):
"""Fired when a search should be performed."""
def __init__(self, query: str) -> None:
super().__init__()
self.query = query
class SearchCancelled(Message):
"""Fired when the search is cancelled."""
pass
class SearchConfirmed(Message):
"""Fired when user presses Enter to confirm search and focus results."""
def __init__(self, query: str) -> None:
super().__init__()
self.query = query
def __init__(
self,
name: Optional[str] = None,
id: Optional[str] = None,
classes: Optional[str] = None,
) -> None:
super().__init__(name=name, id=id, classes=classes)
self._debounce_timer: Optional[Timer] = None
self._last_query: str = ""
def compose(self) -> ComposeResult:
with Horizontal():
yield Label("Search:", classes="search-label")
yield Input(
placeholder="from <name> or subject <text> or body <text>...",
id="search-input",
)
yield Label("", classes="search-status", id="search-status")
yield Button("?", variant="default", id="help-btn")
yield Button("Cancel", variant="warning", id="cancel-btn")
def show(self, initial_query: str = "") -> None:
"""Show the search panel and focus the input."""
self.add_class("visible")
input_widget = self.query_one("#search-input", Input)
input_widget.value = initial_query
self._last_query = initial_query
input_widget.focus()
def hide(self) -> None:
"""Hide the search panel."""
self.remove_class("visible")
self._cancel_debounce()
self.result_count = -1
@property
def is_visible(self) -> bool:
"""Check if the panel is visible."""
return self.has_class("visible")
@property
def search_query(self) -> str:
"""Get the current search query."""
return self.query_one("#search-input", Input).value
def _cancel_debounce(self) -> None:
"""Cancel any pending debounced search."""
if self._debounce_timer:
self._debounce_timer.stop()
self._debounce_timer = None
def _trigger_search(self) -> None:
"""Trigger the actual search after debounce."""
query = self.query_one("#search-input", Input).value.strip()
if query and query != self._last_query:
self._last_query = query
self.is_searching = True
self.post_message(self.SearchRequested(query))
def on_input_changed(self, event: Input.Changed) -> None:
"""Handle input changes with debounce."""
if event.input.id != "search-input":
return
# Cancel any existing timer
self._cancel_debounce()
# Don't search empty queries
if not event.value.strip():
return
# Set up new debounce timer (1 second)
self._debounce_timer = self.set_timer(1.0, self._trigger_search)
def on_input_submitted(self, event: Input.Submitted) -> None:
"""Handle Enter key - trigger search immediately and confirm."""
if event.input.id != "search-input":
return
self._cancel_debounce()
query = event.value.strip()
if query:
self._last_query = query
self.is_searching = True
self.post_message(self.SearchConfirmed(query))
def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button presses."""
if event.button.id == "cancel-btn":
self.action_cancel()
elif event.button.id == "help-btn":
self.app.push_screen(SearchHelpModal())
def action_cancel(self) -> None:
"""Cancel the search and restore previous state."""
self._cancel_debounce()
self.hide()
self.post_message(self.SearchCancelled())
def update_status(self, count: int, searching: bool = False) -> None:
"""Update the search status display."""
self.is_searching = searching
self.result_count = count
status = self.query_one("#search-status", Label)
if searching:
status.update("Searching...")
elif count >= 0:
status.update(f"{count} result{'s' if count != 1 else ''}")
else:
status.update("")
def watch_is_searching(self, searching: bool) -> None:
"""Update UI when searching state changes."""
status = self.query_one("#search-status", Label)
if searching:
status.update("Searching...")

View File

@@ -4,6 +4,7 @@ from .OpenMessage import OpenMessageScreen
from .DocumentViewer import DocumentViewerScreen from .DocumentViewer import DocumentViewerScreen
from .LinkPanel import LinkPanel, LinkItem, extract_links_from_content from .LinkPanel import LinkPanel, LinkItem, extract_links_from_content
from .ConfirmDialog import ConfirmDialog from .ConfirmDialog import ConfirmDialog
from .SearchPanel import SearchPanel, SearchHelpModal
__all__ = [ __all__ = [
"CreateTaskScreen", "CreateTaskScreen",
@@ -13,4 +14,6 @@ __all__ = [
"LinkItem", "LinkItem",
"extract_links_from_content", "extract_links_from_content",
"ConfirmDialog", "ConfirmDialog",
"SearchPanel",
"SearchHelpModal",
] ]