285 lines
8.5 KiB
Python
285 lines
8.5 KiB
Python
"""Live notmuch search launcher.
|
|
|
|
A FuzzyMenu variant that runs `notmuch search` per keystroke (debounced),
|
|
renders thread summaries, and on activation opens the thread in emacs notmuch.
|
|
A bare-query handoff item is always appended so the user can defer to
|
|
notmuch-search inside emacs without having a matching result selected.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from fabric.widgets.box import Box
|
|
from fabric.widgets.label import Label
|
|
from gi.repository import GLib, Gtk
|
|
from loguru import logger
|
|
|
|
from sims.config import NOTMUCH
|
|
|
|
from .base import FuzzyMenu
|
|
|
|
|
|
DEBOUNCE_MS = 120
|
|
MIN_QUERY_LEN = 2
|
|
SEARCH_LIMIT = 30
|
|
|
|
|
|
@dataclass
|
|
class NotmuchHit:
|
|
thread: str
|
|
subject: str
|
|
authors: str
|
|
date_relative: str
|
|
|
|
@property
|
|
def query(self) -> str:
|
|
return f"thread:{self.thread}"
|
|
|
|
|
|
@dataclass
|
|
class SavedSearch:
|
|
name: str
|
|
query: str
|
|
|
|
|
|
@dataclass
|
|
class BareQueryHandoff:
|
|
query: str
|
|
|
|
|
|
def _elisp_escape(s: str) -> str:
|
|
return s.replace("\\", "\\\\").replace('"', '\\"')
|
|
|
|
|
|
class NotmuchSearchProvider:
|
|
def __init__(self):
|
|
self._notmuch_path = NOTMUCH.get("notmuch_path", "notmuch")
|
|
self._emacsclient = NOTMUCH.get("emacsclient_command", "emacsclient")
|
|
raw_saved = NOTMUCH.get("saved_searches", []) or []
|
|
self._saved: list[SavedSearch] = []
|
|
for entry in raw_saved:
|
|
query = (entry.get("query") or "").strip()
|
|
if not query:
|
|
continue
|
|
name = entry.get("name") or query
|
|
self._saved.append(SavedSearch(name=name, query=query))
|
|
|
|
def saved_searches(self) -> list[SavedSearch]:
|
|
return list(self._saved)
|
|
|
|
def search(self, query: str) -> list[NotmuchHit]:
|
|
cmd = [
|
|
self._notmuch_path,
|
|
"search",
|
|
"--format=json",
|
|
"--output=summary",
|
|
f"--limit={SEARCH_LIMIT}",
|
|
query,
|
|
]
|
|
try:
|
|
proc = subprocess.run(
|
|
cmd, capture_output=True, text=True, check=True, timeout=2
|
|
)
|
|
except subprocess.CalledProcessError as e:
|
|
logger.warning(
|
|
f"[NotmuchSearch] search failed for {query!r}: {e.stderr.strip()}"
|
|
)
|
|
return []
|
|
except (subprocess.TimeoutExpired, FileNotFoundError) as e:
|
|
logger.warning(f"[NotmuchSearch] {e}")
|
|
return []
|
|
try:
|
|
raw = json.loads(proc.stdout) or []
|
|
except json.JSONDecodeError:
|
|
return []
|
|
hits: list[NotmuchHit] = []
|
|
for r in raw:
|
|
hits.append(
|
|
NotmuchHit(
|
|
thread=str(r.get("thread", "")),
|
|
subject=r.get("subject") or "(no subject)",
|
|
authors=r.get("authors") or "",
|
|
date_relative=r.get("date_relative") or "",
|
|
)
|
|
)
|
|
return hits
|
|
|
|
def render_hit(self, hit: NotmuchHit) -> Gtk.Widget:
|
|
text = Box(name="notmuch-text", orientation="v", spacing=0, h_expand=True)
|
|
text.add(
|
|
Label(
|
|
label=hit.subject,
|
|
name="notmuch-subject",
|
|
h_align="start",
|
|
ellipsization="end",
|
|
max_chars_width=80,
|
|
)
|
|
)
|
|
text.add(
|
|
Label(
|
|
label=hit.authors,
|
|
name="notmuch-authors",
|
|
h_align="start",
|
|
ellipsization="end",
|
|
max_chars_width=80,
|
|
)
|
|
)
|
|
meta = Label(
|
|
label=hit.date_relative, name="notmuch-date", h_align="end"
|
|
)
|
|
return Box(
|
|
name="slot-box",
|
|
orientation="h",
|
|
spacing=10,
|
|
children=[text, meta],
|
|
)
|
|
|
|
def render_saved(self, item: SavedSearch) -> Gtk.Widget:
|
|
text = Box(name="notmuch-text", orientation="v", spacing=0, h_expand=True)
|
|
text.add(
|
|
Label(label=item.name, name="notmuch-saved-name", h_align="start")
|
|
)
|
|
text.add(
|
|
Label(
|
|
label=item.query,
|
|
name="notmuch-saved-query",
|
|
h_align="start",
|
|
ellipsization="end",
|
|
max_chars_width=80,
|
|
)
|
|
)
|
|
return Box(
|
|
name="slot-box", orientation="h", spacing=10, children=[text]
|
|
)
|
|
|
|
def render_bare(self, item: BareQueryHandoff) -> Gtk.Widget:
|
|
return Box(
|
|
name="slot-box",
|
|
orientation="h",
|
|
children=[
|
|
Label(
|
|
label=f"→ Search '{item.query}' in emacs",
|
|
name="notmuch-bare",
|
|
h_align="start",
|
|
),
|
|
],
|
|
)
|
|
|
|
def open_thread(self, hit: NotmuchHit) -> None:
|
|
self._emacs_eval(f'(notmuch-show "{_elisp_escape(hit.query)}")')
|
|
|
|
def open_search(self, query: str) -> None:
|
|
self._emacs_eval(f'(notmuch-search "{_elisp_escape(query)}")')
|
|
|
|
def _emacs_eval(self, sexp: str) -> None:
|
|
try:
|
|
subprocess.Popen(
|
|
[self._emacsclient, "-c", "-e", sexp],
|
|
start_new_session=True,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"[NotmuchSearch] failed to launch emacsclient: {e}")
|
|
|
|
|
|
class _MenuProvider:
|
|
"""LauncherProvider that defers to NotmuchSearchProvider.
|
|
|
|
items() returns saved searches (shown when the entry is empty).
|
|
filter() is identity — NotmuchSearchMenu manages self._items directly
|
|
on every keystroke (live search instead of in-memory filtering).
|
|
"""
|
|
|
|
def __init__(self, search: NotmuchSearchProvider):
|
|
self._search = search
|
|
|
|
def items(self) -> list[Any]:
|
|
return list(self._search.saved_searches())
|
|
|
|
def filter(self, items: list[Any], query: str) -> list[Any]:
|
|
return items
|
|
|
|
def render(self, item: Any) -> Gtk.Widget:
|
|
if isinstance(item, NotmuchHit):
|
|
return self._search.render_hit(item)
|
|
if isinstance(item, SavedSearch):
|
|
return self._search.render_saved(item)
|
|
if isinstance(item, BareQueryHandoff):
|
|
return self._search.render_bare(item)
|
|
return Box()
|
|
|
|
def activate(self, item: Any) -> None:
|
|
if isinstance(item, NotmuchHit):
|
|
self._search.open_thread(item)
|
|
elif isinstance(item, SavedSearch):
|
|
self._search.open_search(item.query)
|
|
elif isinstance(item, BareQueryHandoff):
|
|
self._search.open_search(item.query)
|
|
|
|
|
|
class NotmuchSearchMenu(FuzzyMenu):
|
|
def __init__(self, monitor: int = 0):
|
|
self._search = NotmuchSearchProvider()
|
|
super().__init__(
|
|
provider=_MenuProvider(self._search),
|
|
monitor=monitor,
|
|
placeholder="notmuch search…",
|
|
window_name="notmuch-search",
|
|
max_results=12,
|
|
)
|
|
self._debounce_id: int | None = None
|
|
|
|
def show(self):
|
|
self._cancel_debounce()
|
|
super().show()
|
|
|
|
def hide(self):
|
|
self._cancel_debounce()
|
|
super().hide()
|
|
|
|
def _on_text_changed(self, entry, *_):
|
|
text = entry.get_text()
|
|
self._cancel_debounce()
|
|
self._selected_index = 0
|
|
self._scroll_offset = 0
|
|
|
|
if not text:
|
|
# Empty query: re-show saved searches via the provider.
|
|
self._items = self._search.saved_searches()
|
|
self._refresh_viewport(text)
|
|
return
|
|
|
|
if len(text) < MIN_QUERY_LEN:
|
|
# Too short to query notmuch; offer just the bare-query handoff.
|
|
self._items = [BareQueryHandoff(query=text)]
|
|
self._refresh_viewport(text)
|
|
return
|
|
|
|
# Show the bare-query handoff immediately while we wait for the search
|
|
# — gives the user a way to commit before the debounce fires.
|
|
self._items = [BareQueryHandoff(query=text)]
|
|
self._refresh_viewport(text)
|
|
self._debounce_id = GLib.timeout_add(
|
|
DEBOUNCE_MS, self._on_debounce_fire, text
|
|
)
|
|
|
|
def _cancel_debounce(self):
|
|
if self._debounce_id is not None:
|
|
GLib.source_remove(self._debounce_id)
|
|
self._debounce_id = None
|
|
|
|
def _on_debounce_fire(self, text: str):
|
|
self._debounce_id = None
|
|
if self.search_entry.get_text() != text:
|
|
return False
|
|
hits = self._search.search(text)
|
|
items: list[Any] = list(hits)
|
|
items.append(BareQueryHandoff(query=text))
|
|
self._items = items
|
|
self._selected_index = 0
|
|
self._scroll_offset = 0
|
|
self._refresh_viewport(text)
|
|
return False
|