757 lines
23 KiB
Python
757 lines
23 KiB
Python
from __future__ import annotations
|
|
|
|
import curses
|
|
import os
|
|
import sys
|
|
from collections.abc import Iterator
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import TextIO
|
|
|
|
from app.navigator import Navigator, VisibleEntry
|
|
from app.opener import (
|
|
OpenWithOption,
|
|
available_openers,
|
|
is_blocking_option,
|
|
launch_background_option,
|
|
launch_blocking_option,
|
|
)
|
|
|
|
ESCAPE_KEY = 27
|
|
BACKSPACE_KEYS = (curses.KEY_BACKSPACE, 127, 8)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RenderLine:
|
|
"""A rendered navigator row ready for terminal output."""
|
|
|
|
text: str
|
|
is_selected: bool
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class HistoryLine:
|
|
"""A rendered history row ready for terminal output."""
|
|
|
|
text: str
|
|
is_selected: bool
|
|
|
|
|
|
def move_selection(current_index: int, step: int, entry_count: int) -> int:
|
|
"""Move the selected index within the available entry range."""
|
|
|
|
if entry_count <= 0:
|
|
return 0
|
|
|
|
return max(0, min(current_index + step, entry_count - 1))
|
|
|
|
|
|
def format_entry(entry: VisibleEntry) -> str:
|
|
"""Format a visible entry for terminal display."""
|
|
|
|
if entry.is_parent:
|
|
return entry.name
|
|
|
|
indent = " " * entry.depth
|
|
|
|
if entry.is_file:
|
|
indicator = "•"
|
|
elif entry.has_children:
|
|
indicator = "▼" if entry.is_expanded else "▶"
|
|
else:
|
|
indicator = " "
|
|
|
|
return f"{indent}{indicator} {entry.name}"
|
|
|
|
|
|
def build_render_lines(navigator: Navigator) -> list[RenderLine]:
|
|
"""Build all visible lines for the current navigator state."""
|
|
|
|
return [
|
|
RenderLine(
|
|
text=format_entry(entry),
|
|
is_selected=index == navigator.selected_index,
|
|
)
|
|
for index, entry in enumerate(navigator.visible_entries)
|
|
]
|
|
|
|
|
|
def build_history_lines(
|
|
entries: tuple[Path, ...], selected_index: int
|
|
) -> list[HistoryLine]:
|
|
"""Build all visible lines for history mode."""
|
|
|
|
return [
|
|
HistoryLine(text=str(path), is_selected=index == selected_index)
|
|
for index, path in enumerate(entries)
|
|
]
|
|
|
|
|
|
def clamp_scroll_offset(
|
|
selected_index: int, scroll_offset: int, visible_count: int, total_count: int
|
|
) -> int:
|
|
"""Keep the selected row within the visible scrolling window."""
|
|
|
|
if visible_count <= 0 or total_count <= visible_count:
|
|
return 0
|
|
|
|
max_offset = total_count - visible_count
|
|
offset = max(0, min(scroll_offset, max_offset))
|
|
|
|
if selected_index < offset:
|
|
return selected_index
|
|
|
|
if selected_index >= offset + visible_count:
|
|
return selected_index - visible_count + 1
|
|
|
|
return offset
|
|
|
|
|
|
class TerminalUI:
|
|
"""Minimal curses UI for the directory navigator MVP."""
|
|
|
|
def __init__(self) -> None:
|
|
self._history_mode = False
|
|
self._history_selected_index = 0
|
|
self._tree_scroll_offset = 0
|
|
self._history_scroll_offset = 0
|
|
self._filter_mode = False
|
|
self._filter_query = ""
|
|
self._filter_selected_index = 0
|
|
self._filter_scroll_offset = 0
|
|
self._status_message: str | None = None
|
|
|
|
def run(self, navigator: Navigator) -> Path:
|
|
"""Start the interactive session and return the final directory path."""
|
|
|
|
with open_terminal_streams() as (input_stream, output_stream):
|
|
with redirect_standard_streams(input_stream, output_stream):
|
|
return curses.wrapper(self._run_session, navigator)
|
|
|
|
def _run_session(self, stdscr: curses.window, navigator: Navigator) -> Path:
|
|
curses.curs_set(0)
|
|
stdscr.keypad(True)
|
|
original_path = navigator.current_path
|
|
|
|
while True:
|
|
self._render(stdscr, navigator)
|
|
key = stdscr.getch()
|
|
|
|
if self._history_mode:
|
|
self._handle_history_key(key, navigator, stdscr=stdscr)
|
|
continue
|
|
|
|
if self._filter_mode:
|
|
filter_result = self._handle_filter_key(key, navigator, stdscr)
|
|
if filter_result is not None:
|
|
return filter_result
|
|
continue
|
|
|
|
if key == ESCAPE_KEY:
|
|
return original_path
|
|
|
|
if key == ord("h"):
|
|
self._toggle_history_mode(navigator)
|
|
continue
|
|
|
|
tree_result = self._handle_tree_key(key, navigator, stdscr=stdscr)
|
|
if tree_result is not None:
|
|
return tree_result
|
|
|
|
def _render(self, stdscr: curses.window, navigator: Navigator) -> None:
|
|
stdscr.erase()
|
|
height, width = stdscr.getmaxyx()
|
|
|
|
if self._history_mode:
|
|
self._render_history_mode(stdscr, navigator, width=width, height=height)
|
|
stdscr.refresh()
|
|
return
|
|
|
|
hidden_status = "on" if navigator.show_hidden else "off"
|
|
file_status = "on" if navigator.show_files else "off"
|
|
path_text = (
|
|
f"{navigator.current_path} [hidden: {hidden_status}] [files: {file_status}]"
|
|
)
|
|
stdscr.addnstr(0, 0, path_text, max(0, width - 1))
|
|
|
|
start_row = 2
|
|
if self._filter_mode:
|
|
stdscr.addnstr(
|
|
1,
|
|
0,
|
|
f"Filter: {self._filter_query}",
|
|
max(0, width - 1),
|
|
curses.A_BOLD,
|
|
)
|
|
start_row = 3
|
|
|
|
lines, selected_index = self._build_tree_lines_for_render(navigator)
|
|
footer_rows = 1 if self._status_message else 0
|
|
visible_count = max(0, height - start_row - footer_rows)
|
|
|
|
if self._filter_mode:
|
|
self._filter_scroll_offset = clamp_scroll_offset(
|
|
selected_index,
|
|
self._filter_scroll_offset,
|
|
visible_count,
|
|
len(lines),
|
|
)
|
|
scroll_offset = self._filter_scroll_offset
|
|
else:
|
|
self._tree_scroll_offset = clamp_scroll_offset(
|
|
selected_index,
|
|
self._tree_scroll_offset,
|
|
visible_count,
|
|
len(lines),
|
|
)
|
|
scroll_offset = self._tree_scroll_offset
|
|
|
|
if self._filter_mode and self._filter_query and not lines:
|
|
stdscr.addnstr(start_row, 0, "No matches", max(0, width - 1), curses.A_DIM)
|
|
else:
|
|
for row, line in enumerate(
|
|
lines[scroll_offset : scroll_offset + visible_count],
|
|
start=start_row,
|
|
):
|
|
if row >= height:
|
|
break
|
|
|
|
attributes = curses.A_REVERSE if line.is_selected else curses.A_NORMAL
|
|
stdscr.addnstr(row, 0, line.text, max(0, width - 1), attributes)
|
|
|
|
if self._status_message and height > 0:
|
|
stdscr.addnstr(
|
|
height - 1,
|
|
0,
|
|
self._status_message,
|
|
max(0, width - 1),
|
|
curses.A_DIM,
|
|
)
|
|
|
|
stdscr.refresh()
|
|
|
|
def _build_tree_lines_for_render(
|
|
self, navigator: Navigator
|
|
) -> tuple[list[RenderLine], int]:
|
|
if not self._filter_mode:
|
|
return build_render_lines(navigator), navigator.selected_index
|
|
|
|
entries = navigator.visible_entries
|
|
matches = self._matching_indices(entries)
|
|
if not matches:
|
|
self._filter_selected_index = 0
|
|
return [], 0
|
|
|
|
self._filter_selected_index = max(
|
|
0, min(self._filter_selected_index, len(matches) - 1)
|
|
)
|
|
|
|
return [
|
|
RenderLine(
|
|
text=format_entry(entries[entry_index]),
|
|
is_selected=match_index == self._filter_selected_index,
|
|
)
|
|
for match_index, entry_index in enumerate(matches)
|
|
], self._filter_selected_index
|
|
|
|
def _render_history_mode(
|
|
self, stdscr: curses.window, navigator: Navigator, width: int, height: int
|
|
) -> None:
|
|
history_lines = build_history_lines(
|
|
navigator.history.entries(), self._history_selected_index
|
|
)
|
|
start_row = 1
|
|
footer_rows = 1 if self._status_message else 0
|
|
visible_count = max(0, height - start_row - 1 - footer_rows)
|
|
self._history_scroll_offset = clamp_scroll_offset(
|
|
self._history_selected_index,
|
|
self._history_scroll_offset,
|
|
visible_count,
|
|
len(history_lines),
|
|
)
|
|
|
|
stdscr.addnstr(start_row, 0, "History:", max(0, width - 1), curses.A_BOLD)
|
|
|
|
for index, line in enumerate(
|
|
history_lines[
|
|
self._history_scroll_offset : self._history_scroll_offset
|
|
+ visible_count
|
|
],
|
|
start=1,
|
|
):
|
|
row = start_row + index
|
|
if row >= height:
|
|
break
|
|
|
|
attributes = curses.A_REVERSE if line.is_selected else curses.A_NORMAL
|
|
stdscr.addnstr(row, 0, line.text, max(0, width - 1), attributes)
|
|
|
|
if self._status_message and height > 0:
|
|
stdscr.addnstr(
|
|
height - 1,
|
|
0,
|
|
self._status_message,
|
|
max(0, width - 1),
|
|
curses.A_DIM,
|
|
)
|
|
|
|
def _toggle_history_mode(self, navigator: Navigator) -> None:
|
|
self._history_mode = not self._history_mode
|
|
self._status_message = None
|
|
if self._history_mode:
|
|
self._history_selected_index = self._get_current_history_index(navigator)
|
|
self._history_scroll_offset = 0
|
|
|
|
def _handle_history_key(
|
|
self, key: int, navigator: Navigator, stdscr: curses.window | None = None
|
|
) -> None:
|
|
history_entries = navigator.history.entries()
|
|
|
|
if key in (ESCAPE_KEY, ord("h")):
|
|
self._history_mode = False
|
|
return
|
|
|
|
if key == curses.KEY_UP:
|
|
self._history_selected_index = move_selection(
|
|
self._history_selected_index,
|
|
step=-1,
|
|
entry_count=len(history_entries),
|
|
)
|
|
return
|
|
|
|
if key == curses.KEY_DOWN:
|
|
self._history_selected_index = move_selection(
|
|
self._history_selected_index,
|
|
step=1,
|
|
entry_count=len(history_entries),
|
|
)
|
|
return
|
|
|
|
if key == curses.KEY_PPAGE:
|
|
step = -self._page_step(stdscr, start_row=1)
|
|
self._history_selected_index = move_selection(
|
|
self._history_selected_index,
|
|
step=step,
|
|
entry_count=len(history_entries),
|
|
)
|
|
return
|
|
|
|
if key == curses.KEY_NPAGE:
|
|
step = self._page_step(stdscr, start_row=1)
|
|
self._history_selected_index = move_selection(
|
|
self._history_selected_index,
|
|
step=step,
|
|
entry_count=len(history_entries),
|
|
)
|
|
return
|
|
|
|
if key in (curses.KEY_HOME, ord("g")):
|
|
self._history_selected_index = 0
|
|
return
|
|
|
|
if key in (curses.KEY_END, ord("G")) and history_entries:
|
|
self._history_selected_index = len(history_entries) - 1
|
|
return
|
|
|
|
if key in (curses.KEY_ENTER, 10, 13) and history_entries:
|
|
navigator.select_history_entry(self._history_selected_index)
|
|
self._history_mode = False
|
|
self._tree_scroll_offset = 0
|
|
|
|
def _handle_filter_key(
|
|
self, key: int, navigator: Navigator, stdscr: curses.window | None = None
|
|
) -> Path | None:
|
|
entries = navigator.visible_entries
|
|
matches = self._matching_indices(entries)
|
|
|
|
if key == ESCAPE_KEY:
|
|
self._filter_mode = False
|
|
self._filter_query = ""
|
|
self._filter_selected_index = 0
|
|
self._filter_scroll_offset = 0
|
|
return None
|
|
|
|
if key in (curses.KEY_ENTER, 10, 13):
|
|
if not matches:
|
|
return None
|
|
|
|
selected_entry_index = matches[self._filter_selected_index]
|
|
navigator.set_selected_index(selected_entry_index)
|
|
selected = entries[selected_entry_index]
|
|
|
|
self._filter_mode = False
|
|
self._filter_query = ""
|
|
self._filter_selected_index = 0
|
|
self._filter_scroll_offset = 0
|
|
|
|
if selected.is_file:
|
|
self._open_with_selected(navigator, stdscr)
|
|
return None
|
|
|
|
navigator.enter_selected_directory()
|
|
self._tree_scroll_offset = 0
|
|
return None
|
|
|
|
if key == curses.KEY_UP:
|
|
self._filter_selected_index = move_selection(
|
|
self._filter_selected_index,
|
|
step=-1,
|
|
entry_count=len(matches),
|
|
)
|
|
return None
|
|
|
|
if key == curses.KEY_DOWN:
|
|
self._filter_selected_index = move_selection(
|
|
self._filter_selected_index,
|
|
step=1,
|
|
entry_count=len(matches),
|
|
)
|
|
return None
|
|
|
|
if key == curses.KEY_PPAGE:
|
|
step = -self._page_step(stdscr, start_row=3)
|
|
self._filter_selected_index = move_selection(
|
|
self._filter_selected_index,
|
|
step=step,
|
|
entry_count=len(matches),
|
|
)
|
|
return None
|
|
|
|
if key == curses.KEY_NPAGE:
|
|
step = self._page_step(stdscr, start_row=3)
|
|
self._filter_selected_index = move_selection(
|
|
self._filter_selected_index,
|
|
step=step,
|
|
entry_count=len(matches),
|
|
)
|
|
return None
|
|
|
|
if key == curses.KEY_HOME:
|
|
self._filter_selected_index = 0
|
|
return None
|
|
|
|
if key == curses.KEY_END and matches:
|
|
self._filter_selected_index = len(matches) - 1
|
|
return None
|
|
|
|
if key in BACKSPACE_KEYS:
|
|
self._filter_query = self._filter_query[:-1]
|
|
self._filter_selected_index = 0
|
|
self._filter_scroll_offset = 0
|
|
return None
|
|
|
|
if 32 <= key <= 126:
|
|
self._filter_query += chr(key)
|
|
self._filter_selected_index = 0
|
|
self._filter_scroll_offset = 0
|
|
|
|
return None
|
|
|
|
def _handle_tree_key(
|
|
self,
|
|
key: int,
|
|
navigator: Navigator,
|
|
stdscr: curses.window | None = None,
|
|
) -> Path | None:
|
|
self._status_message = None
|
|
|
|
if key == curses.KEY_UP:
|
|
navigator.set_selected_index(
|
|
move_selection(
|
|
navigator.selected_index,
|
|
step=-1,
|
|
entry_count=len(navigator.visible_entries),
|
|
)
|
|
)
|
|
return None
|
|
|
|
if key == curses.KEY_DOWN:
|
|
navigator.set_selected_index(
|
|
move_selection(
|
|
navigator.selected_index,
|
|
step=1,
|
|
entry_count=len(navigator.visible_entries),
|
|
)
|
|
)
|
|
return None
|
|
|
|
if key == curses.KEY_PPAGE:
|
|
navigator.set_selected_index(
|
|
move_selection(
|
|
navigator.selected_index,
|
|
step=-self._page_step(stdscr, start_row=2),
|
|
entry_count=len(navigator.visible_entries),
|
|
)
|
|
)
|
|
return None
|
|
|
|
if key == curses.KEY_NPAGE:
|
|
navigator.set_selected_index(
|
|
move_selection(
|
|
navigator.selected_index,
|
|
step=self._page_step(stdscr, start_row=2),
|
|
entry_count=len(navigator.visible_entries),
|
|
)
|
|
)
|
|
return None
|
|
|
|
if key in (curses.KEY_HOME, ord("g")):
|
|
navigator.set_selected_index(0)
|
|
return None
|
|
|
|
if key in (curses.KEY_END, ord("G")):
|
|
navigator.set_selected_index(len(navigator.visible_entries) - 1)
|
|
return None
|
|
|
|
if key == curses.KEY_RIGHT:
|
|
selected = navigator.selected_entry
|
|
if selected.is_file:
|
|
return None
|
|
|
|
was_expanded = selected.is_expanded
|
|
|
|
if not navigator.expand_selected_directory():
|
|
navigator.enter_selected_directory()
|
|
self._tree_scroll_offset = 0
|
|
return None
|
|
|
|
if was_expanded:
|
|
navigator.enter_selected_directory()
|
|
self._tree_scroll_offset = 0
|
|
|
|
return None
|
|
|
|
if key == curses.KEY_LEFT:
|
|
if not navigator.collapse_selected_directory():
|
|
navigator.go_to_parent_directory()
|
|
self._tree_scroll_offset = 0
|
|
|
|
return None
|
|
|
|
if key == ord("b"):
|
|
navigator.go_back()
|
|
self._tree_scroll_offset = 0
|
|
return None
|
|
|
|
if key == ord("f"):
|
|
navigator.go_forward()
|
|
self._tree_scroll_offset = 0
|
|
return None
|
|
|
|
if key == ord("."):
|
|
navigator.toggle_hidden()
|
|
self._tree_scroll_offset = 0
|
|
return None
|
|
|
|
if key == ord("a"):
|
|
navigator.toggle_files()
|
|
self._tree_scroll_offset = 0
|
|
return None
|
|
|
|
if key == ord("/"):
|
|
self._filter_mode = True
|
|
self._filter_query = ""
|
|
self._filter_selected_index = 0
|
|
self._filter_scroll_offset = 0
|
|
return None
|
|
|
|
if key == ord("o"):
|
|
self._open_with_selected(navigator, stdscr)
|
|
return None
|
|
|
|
if key in (curses.KEY_ENTER, 10, 13):
|
|
return navigator.selected_entry.path
|
|
|
|
return None
|
|
|
|
def _matching_indices(self, entries: tuple[VisibleEntry, ...]) -> list[int]:
|
|
if not self._filter_query:
|
|
return list(range(len(entries)))
|
|
|
|
query = self._filter_query.casefold()
|
|
return [
|
|
index
|
|
for index, entry in enumerate(entries)
|
|
if query in entry.name.casefold()
|
|
]
|
|
|
|
def _page_step(self, stdscr: curses.window | None, start_row: int) -> int:
|
|
if stdscr is None:
|
|
return 10
|
|
|
|
height, _width = stdscr.getmaxyx()
|
|
footer_rows = 1 if self._status_message else 0
|
|
return max(1, height - start_row - footer_rows)
|
|
|
|
def _open_with_selected(
|
|
self, navigator: Navigator, stdscr: curses.window | None
|
|
) -> None:
|
|
selected = navigator.selected_entry
|
|
options = available_openers(selected.path)
|
|
if not options:
|
|
self._status_message = "No applications available for this entry"
|
|
return
|
|
|
|
option_index = 0
|
|
if stdscr is not None:
|
|
choice = self._prompt_open_with_choice(stdscr, options, selected.name)
|
|
if choice is None:
|
|
self._status_message = "Open cancelled"
|
|
return
|
|
|
|
option_index = choice
|
|
|
|
option = options[option_index]
|
|
if is_blocking_option(option):
|
|
success, message = self._launch_blocking_with_terminal_restore(
|
|
stdscr, selected.path, option
|
|
)
|
|
else:
|
|
success, message = launch_background_option(selected.path, option)
|
|
|
|
self._status_message = message
|
|
if not success:
|
|
self._status_message = message
|
|
|
|
def _prompt_open_with_choice(
|
|
self,
|
|
stdscr: curses.window,
|
|
options: list[OpenWithOption],
|
|
entry_name: str,
|
|
) -> int | None:
|
|
selected_index = 0
|
|
|
|
while True:
|
|
height, width = stdscr.getmaxyx()
|
|
title_row = max(0, height - len(options) - 2)
|
|
title = f"Open '{entry_name}' with (↑/↓, Enter, Esc):"
|
|
|
|
stdscr.addnstr(
|
|
title_row,
|
|
0,
|
|
" " * max(0, width - 1),
|
|
max(0, width - 1),
|
|
)
|
|
stdscr.addnstr(
|
|
title_row,
|
|
0,
|
|
title,
|
|
max(0, width - 1),
|
|
curses.A_BOLD,
|
|
)
|
|
|
|
for index, option in enumerate(options):
|
|
row = title_row + index + 1
|
|
if row >= height:
|
|
break
|
|
|
|
attributes = (
|
|
curses.A_REVERSE if index == selected_index else curses.A_NORMAL
|
|
)
|
|
label = f" {option.label}"
|
|
stdscr.addnstr(
|
|
row,
|
|
0,
|
|
" " * max(0, width - 1),
|
|
max(0, width - 1),
|
|
)
|
|
stdscr.addnstr(row, 0, label, max(0, width - 1), attributes)
|
|
|
|
stdscr.refresh()
|
|
key = stdscr.getch()
|
|
|
|
if key == ESCAPE_KEY:
|
|
return None
|
|
|
|
if key == curses.KEY_UP:
|
|
selected_index = move_selection(
|
|
selected_index,
|
|
step=-1,
|
|
entry_count=len(options),
|
|
)
|
|
continue
|
|
|
|
if key == curses.KEY_DOWN:
|
|
selected_index = move_selection(
|
|
selected_index,
|
|
step=1,
|
|
entry_count=len(options),
|
|
)
|
|
continue
|
|
|
|
if key in (curses.KEY_ENTER, 10, 13):
|
|
return selected_index
|
|
|
|
def _launch_blocking_with_terminal_restore(
|
|
self,
|
|
stdscr: curses.window | None,
|
|
path: Path,
|
|
option: OpenWithOption,
|
|
) -> tuple[bool, str]:
|
|
if stdscr is None:
|
|
return launch_blocking_option(path, option)
|
|
|
|
curses.def_prog_mode()
|
|
curses.endwin()
|
|
|
|
try:
|
|
success, message = launch_blocking_option(path, option)
|
|
finally:
|
|
curses.reset_prog_mode()
|
|
stdscr.keypad(True)
|
|
stdscr.clear()
|
|
stdscr.refresh()
|
|
|
|
return success, message
|
|
|
|
def _get_current_history_index(self, navigator: Navigator) -> int:
|
|
history_entries = navigator.history.entries()
|
|
|
|
for index in range(len(history_entries) - 1, -1, -1):
|
|
if history_entries[index] == navigator.current_path:
|
|
return index
|
|
|
|
return 0
|
|
|
|
|
|
@contextmanager
|
|
def open_terminal_streams() -> Iterator[tuple[TextIO, TextIO]]:
|
|
"""Open terminal streams suitable for interactive rendering."""
|
|
|
|
try:
|
|
with (
|
|
open("/dev/tty", encoding="utf-8") as input_stream,
|
|
open("/dev/tty", "w", encoding="utf-8") as output_stream,
|
|
):
|
|
yield input_stream, output_stream
|
|
return
|
|
except OSError:
|
|
if not sys.stdin.isatty() or not sys.stderr.isatty():
|
|
raise RuntimeError("Interactive terminal is required") from None
|
|
|
|
yield sys.stdin, sys.stderr
|
|
|
|
|
|
@contextmanager
|
|
def redirect_standard_streams(
|
|
input_stream: TextIO, output_stream: TextIO
|
|
) -> Iterator[None]:
|
|
"""Temporarily redirect stdin and stdout to terminal-backed streams."""
|
|
|
|
sys.stdout.flush()
|
|
sys.stderr.flush()
|
|
|
|
original_stdin = os.dup(0)
|
|
original_stdout = os.dup(1)
|
|
|
|
try:
|
|
os.dup2(input_stream.fileno(), 0)
|
|
os.dup2(output_stream.fileno(), 1)
|
|
yield
|
|
finally:
|
|
os.dup2(original_stdin, 0)
|
|
os.dup2(original_stdout, 1)
|
|
os.close(original_stdin)
|
|
os.close(original_stdout)
|