mirror of
https://github.com/PR0M3TH3AN/SeedPass.git
synced 2026-11-06 12:52:04 +00:00
3598 lines
145 KiB
Python
3598 lines
145 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import shlex
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def check_tui2_runtime() -> dict[str, Any]:
|
|
"""Return runtime capability diagnostics for TUI v2."""
|
|
textual_available = importlib.util.find_spec("textual") is not None
|
|
return {
|
|
"status": "ok" if textual_available else "unavailable",
|
|
"backend": "textual",
|
|
"textual_available": textual_available,
|
|
"message": (
|
|
"Textual runtime available."
|
|
if textual_available
|
|
else "Textual is not installed. Install `textual` to run tui2."
|
|
),
|
|
}
|
|
|
|
|
|
def parse_palette_command(command: str) -> tuple[str, list[str]]:
|
|
"""Parse a palette command into ``(cmd, args)``."""
|
|
raw = command.strip()
|
|
if not raw:
|
|
raise ValueError("Palette: command required")
|
|
try:
|
|
parts = shlex.split(raw)
|
|
except ValueError as exc:
|
|
raise ValueError(f"Palette parse error: {exc}") from exc
|
|
if not parts:
|
|
raise ValueError("Palette: command required")
|
|
return parts[0].lower(), parts[1:]
|
|
|
|
|
|
def pagination_window(
|
|
total_rows: int, page_size: int, page_index: int
|
|
) -> tuple[int, int, int, int]:
|
|
"""Return normalized pagination tuple.
|
|
|
|
Returns ``(normalized_page_index, start, end, total_pages)``.
|
|
"""
|
|
if page_size <= 0:
|
|
raise ValueError("page_size must be > 0")
|
|
total = max(0, int(total_rows))
|
|
total_pages = 1 if total == 0 else (total + page_size - 1) // page_size
|
|
page = min(max(0, int(page_index)), total_pages - 1)
|
|
start = page * page_size
|
|
end = min(total, start + page_size)
|
|
return page, start, end, total_pages
|
|
|
|
|
|
def truncate_entry_for_display(
|
|
entry: dict[str, Any], content_limit: int
|
|
) -> dict[str, Any]:
|
|
"""Return an entry payload suitable for responsive display in TUI details."""
|
|
payload = dict(entry)
|
|
content = payload.get("content")
|
|
if content_limit <= 0:
|
|
return payload
|
|
if isinstance(content, str) and len(content) > content_limit:
|
|
head = content[:content_limit]
|
|
payload["content"] = (
|
|
f"{head}\n\n...[truncated {len(content) - len(head)} chars]"
|
|
)
|
|
payload["content_truncated"] = True
|
|
return payload
|
|
|
|
|
|
def render_qr_ascii(data: str) -> str:
|
|
"""Render ``data`` as an ASCII QR code."""
|
|
import qrcode
|
|
|
|
qr = qrcode.QRCode(border=1)
|
|
qr.add_data(data)
|
|
qr.make(fit=True)
|
|
matrix = qr.get_matrix()
|
|
lines: list[str] = []
|
|
for row in matrix:
|
|
lines.append("".join("##" if cell else " " for cell in row))
|
|
return "\n".join(lines)
|
|
|
|
|
|
def launch_tui2(
|
|
*,
|
|
fingerprint: str | None = None,
|
|
entry_service_factory: Any | None = None,
|
|
profile_service_factory: Any | None = None,
|
|
config_service_factory: Any | None = None,
|
|
nostr_service_factory: Any | None = None,
|
|
sync_service_factory: Any | None = None,
|
|
utility_service_factory: Any | None = None,
|
|
vault_service_factory: Any | None = None,
|
|
app_hook: Any | None = None,
|
|
) -> bool:
|
|
"""Launch TUI v2 when runtime dependencies are available.
|
|
|
|
Returns ``True`` when launch succeeds, ``False`` when runtime is unavailable.
|
|
"""
|
|
runtime = check_tui2_runtime()
|
|
if not runtime["textual_available"]:
|
|
return False
|
|
|
|
from textual.app import App, ComposeResult
|
|
from textual.containers import Horizontal, Vertical
|
|
from textual.reactive import reactive
|
|
from textual.widgets import Footer, Header, Input, Label, ListItem, ListView, Static
|
|
|
|
try:
|
|
from textual.widgets import TextArea
|
|
except Exception: # pragma: no cover - runtime fallback for older textual
|
|
TextArea = None
|
|
|
|
class EntryListItem(ListItem):
|
|
def __init__(self, entry_index: int, text: str) -> None:
|
|
super().__init__(Label(text))
|
|
self.entry_index = int(entry_index)
|
|
|
|
class SeedPassTuiV2(App[None]):
|
|
RESULT_PAGE_SIZE = 200
|
|
DETAIL_CONTENT_PREVIEW_LIMIT = 4000
|
|
|
|
CSS = """
|
|
Screen {
|
|
background: #080a0c;
|
|
color: #97b8a6;
|
|
}
|
|
Header {
|
|
background: #0b0f13;
|
|
color: #daf2e5;
|
|
text-style: bold;
|
|
border-bottom: solid #274533;
|
|
}
|
|
Footer {
|
|
background: #0b0f13;
|
|
color: #97b8a6;
|
|
}
|
|
Input {
|
|
background: #0d1114;
|
|
color: #daf2e5;
|
|
border: solid #1a3024;
|
|
}
|
|
Input:focus {
|
|
border: heavy #58f29d;
|
|
}
|
|
ListView {
|
|
background: #0d1114;
|
|
border: solid #1a3024;
|
|
}
|
|
ListItem {
|
|
color: #97b8a6;
|
|
}
|
|
ListItem.-highlight {
|
|
background: #122019;
|
|
color: #daf2e5;
|
|
text-style: bold;
|
|
}
|
|
Static {
|
|
color: #97b8a6;
|
|
}
|
|
#command-palette {
|
|
height: 2;
|
|
margin: 0 1 1 1;
|
|
border: solid #2abf75;
|
|
background: #0d1114;
|
|
color: #daf2e5;
|
|
}
|
|
#body { height: 1fr; }
|
|
#status {
|
|
height: 1;
|
|
padding: 0 1;
|
|
border: heavy #58f29d;
|
|
background: #0d1114;
|
|
color: #daf2e5;
|
|
}
|
|
#left { width: 34; border: solid #1a3024; padding: 1; background: #0d1114; }
|
|
#center { width: 1fr; border: solid #1a3024; padding: 1; background: #0d1114; }
|
|
#right { width: 1fr; border: solid #1a3024; padding: 1; background: #0d1114; }
|
|
#filters {
|
|
height: 1fr;
|
|
overflow: auto;
|
|
}
|
|
#search { margin-bottom: 1; }
|
|
#quick-jump { margin-bottom: 1; }
|
|
#entry-list { height: 1fr; }
|
|
#entry-detail {
|
|
height: 1fr;
|
|
overflow: auto;
|
|
border: solid #274533;
|
|
padding: 1;
|
|
}
|
|
#link-detail {
|
|
height: 12;
|
|
overflow: auto;
|
|
border: solid #2abf75;
|
|
padding: 1;
|
|
margin-top: 1;
|
|
}
|
|
#secret-detail {
|
|
height: 12;
|
|
overflow: auto;
|
|
border: solid #58f29d;
|
|
padding: 1;
|
|
margin-top: 1;
|
|
text-wrap: nowrap;
|
|
}
|
|
#totp-board {
|
|
height: 1fr;
|
|
overflow: auto;
|
|
border: solid #2abf75;
|
|
padding: 1;
|
|
}
|
|
#right-view { height: 1fr; }
|
|
#right-editor { height: 1fr; }
|
|
#doc-edit-label { margin-bottom: 1; }
|
|
#doc-edit-file-type { margin-bottom: 1; }
|
|
#doc-edit-tags { margin-bottom: 1; }
|
|
#doc-edit-help { margin-top: 1; }
|
|
#doc-edit-content { height: 1fr; }
|
|
#doc-edit-content-single { height: 1fr; }
|
|
#help-overlay {
|
|
layer: overlay;
|
|
dock: top;
|
|
margin: 1 2;
|
|
border: solid #58f29d;
|
|
padding: 1;
|
|
background: #0d1114;
|
|
color: #daf2e5;
|
|
}
|
|
#activity {
|
|
height: 12;
|
|
border: solid #274533;
|
|
padding: 1;
|
|
margin-top: 1;
|
|
overflow: auto;
|
|
}
|
|
.pane-focus { border: heavy #58f29d; }
|
|
.hidden { display: none; }
|
|
"""
|
|
BINDINGS = [
|
|
("q", "quit", "Quit"),
|
|
("r", "refresh", "Refresh"),
|
|
("x", "retry_last_error", "Retry"),
|
|
("question_mark", "toggle_help", "Help"),
|
|
("slash", "focus_search", "Search"),
|
|
("j", "focus_jump", "Jump"),
|
|
("1", "focus_left", "Left"),
|
|
("2", "focus_center", "Center"),
|
|
("3", "focus_right", "Right"),
|
|
("f", "cycle_filter", "Filter"),
|
|
("h", "cycle_archive_scope", "Archive View"),
|
|
("p", "prev_page", "Prev Page"),
|
|
("n", "next_page", "Next Page"),
|
|
("l", "cycle_link_filter", "Link Filter"),
|
|
("[", "prev_link", "Prev Link"),
|
|
("]", "next_link", "Next Link"),
|
|
("o", "open_link_target", "Open Link"),
|
|
("v", "reveal_selected", "Reveal"),
|
|
("g", "show_qr", "QR"),
|
|
("6", "toggle_totp_board", "2FA Board"),
|
|
("a", "toggle_archive", "Archive/Restore"),
|
|
("e", "edit_document", "Edit Document"),
|
|
("ctrl+s", "save_document", "Save"),
|
|
("ctrl+p", "open_palette", "Palette"),
|
|
("escape", "cancel_document_edit", "Cancel"),
|
|
]
|
|
|
|
filter_kind: reactive[str] = reactive("all")
|
|
archive_scope: reactive[str] = reactive("active")
|
|
link_relation_filter: reactive[str] = reactive("all")
|
|
editing_document: reactive[bool] = reactive(False)
|
|
palette_open: reactive[bool] = reactive(False)
|
|
help_open: reactive[bool] = reactive(False)
|
|
totp_board_open: reactive[bool] = reactive(False)
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Header(show_clock=True)
|
|
yield Input(
|
|
placeholder=(
|
|
"Command palette: help | open <id> | search <q> | "
|
|
"filter <kind> | archive | restore | "
|
|
"archive-filter <active|all|archived> | "
|
|
"add-password <label> <length> [username] [url] | "
|
|
"add-totp <label> [period] [digits] [secret] | "
|
|
"add-key-value <label> <key> <value> | "
|
|
"add-document <label> <file_type> <content> | "
|
|
"add-ssh <label> [index] | add-pgp <label> [index] [key_type] [user_id] | "
|
|
"add-nostr <label> [index] | add-seed <label> [words] [index] | "
|
|
"add-managed-account <label> [index] | "
|
|
"notes-set <text> | notes-clear | tag-add <tag> | tag-rm <tag> | "
|
|
"tags-set <comma-list> | tags-clear | field-add <label> <value> [hidden] | "
|
|
"field-rm <label> | set-field <name> <value> | clear-field <name> | "
|
|
"2fa-board | 2fa-hide | 2fa-refresh | 2fa-copy <entry_id> | "
|
|
"profiles-list | profile-switch <fp> [password] | profile-add | "
|
|
"profile-remove <fp> | profile-rename <fp> <name> | "
|
|
"setting-secret <on|off> [delay] | setting-offline <on|off> | "
|
|
"setting-quick-unlock <on|off> | setting-timeout <seconds> | "
|
|
"setting-kdf-iterations <n> | setting-kdf-mode <mode> | "
|
|
"relay-list | relay-add <url> | relay-rm <index> | "
|
|
"relay-reset | npub | nostr-reset-sync-state | nostr-fresh-namespace | "
|
|
"sync-now | sync-bg | "
|
|
"checksum-verify | checksum-update | db-export <path> | db-import <path> | "
|
|
"totp-export <path> | parent-seed-backup [path] [password] | "
|
|
"managed-load [entry_id] | managed-exit | "
|
|
"doc-export [output_path] | "
|
|
"link-add <target> [relation] [note] | "
|
|
"link-rm <target> [relation] | "
|
|
"reveal [confirm] | qr [public|private] [confirm] | "
|
|
"link-filter <relation|all> | link-next | link-prev | link-open | "
|
|
"page-next | page-prev | page <n> | retry"
|
|
),
|
|
id="command-palette",
|
|
classes="hidden",
|
|
)
|
|
yield Static("", id="help-overlay", classes="hidden")
|
|
with Horizontal(id="body"):
|
|
with Vertical(id="left"):
|
|
yield Static("", id="filters")
|
|
yield Static("", id="activity")
|
|
with Vertical(id="center"):
|
|
yield Input(
|
|
placeholder="Search entries (Enter to apply)", id="search"
|
|
)
|
|
yield Input(placeholder="Jump to entry id (Enter)", id="quick-jump")
|
|
yield ListView(id="entry-list")
|
|
with Vertical(id="right"):
|
|
with Vertical(id="right-view"):
|
|
yield Static("", id="entry-detail")
|
|
yield Static("", id="link-detail")
|
|
yield Static("", id="secret-detail")
|
|
yield Static("", id="totp-board", classes="hidden")
|
|
with Vertical(id="right-editor", classes="hidden"):
|
|
yield Input(placeholder="Document title", id="doc-edit-label")
|
|
yield Input(
|
|
placeholder="File extension (txt, md, py, ...)",
|
|
id="doc-edit-file-type",
|
|
)
|
|
yield Input(
|
|
placeholder="Tags (comma-separated)", id="doc-edit-tags"
|
|
)
|
|
if TextArea is not None:
|
|
yield TextArea("", id="doc-edit-content")
|
|
else:
|
|
yield Input(
|
|
placeholder="Document content",
|
|
id="doc-edit-content-single",
|
|
)
|
|
yield Static(
|
|
"Edit mode: Ctrl+S save, Esc cancel",
|
|
id="doc-edit-help",
|
|
)
|
|
yield Static("Ready", id="status")
|
|
yield Footer()
|
|
|
|
def on_mount(self) -> None:
|
|
self._service = None
|
|
self._profile_service = None
|
|
self._config_service = None
|
|
self._nostr_service = None
|
|
self._sync_service = None
|
|
self._utility_service = None
|
|
self._vault_service = None
|
|
self._selected_entry_id: int | None = None
|
|
self._selected_entry: dict[str, Any] | None = None
|
|
self._last_query = ""
|
|
self._entry_ids_in_view: list[int] = []
|
|
self._all_results: list[tuple[Any, ...]] = []
|
|
self._result_page = 0
|
|
self._current_links: list[dict[str, Any]] = []
|
|
self._current_link_cursor = 0
|
|
self._last_error: str | None = None
|
|
self._retry_action: Any | None = None
|
|
self._activity_log: list[str] = []
|
|
self._focus_pane = "center"
|
|
self._doc_dirty = False
|
|
self._doc_snapshot: dict[str, Any] = {}
|
|
self._last_status_message = ""
|
|
self._time_now = time.time
|
|
self._totp_rows: list[dict[str, Any]] = []
|
|
self._totp_tick = self.set_interval(1.0, self._tick_totp_board)
|
|
try:
|
|
self._service = (
|
|
entry_service_factory() if callable(entry_service_factory) else None
|
|
)
|
|
except Exception as exc:
|
|
self._service = None
|
|
self._record_failure(
|
|
"Unable to initialize entry service",
|
|
exc,
|
|
retry=self._retry_initialize_service,
|
|
hint="Press 'x' to retry initialization.",
|
|
)
|
|
self.query_one("#entry-detail", Static).update(
|
|
f"Unable to initialize entry service: {exc}"
|
|
)
|
|
try:
|
|
self._profile_service = (
|
|
profile_service_factory()
|
|
if callable(profile_service_factory)
|
|
else None
|
|
)
|
|
except Exception as exc:
|
|
self._profile_service = None
|
|
self._log_activity(f"Profile service unavailable: {exc}")
|
|
try:
|
|
self._config_service = (
|
|
config_service_factory()
|
|
if callable(config_service_factory)
|
|
else None
|
|
)
|
|
except Exception as exc:
|
|
self._config_service = None
|
|
self._log_activity(f"Config service unavailable: {exc}")
|
|
try:
|
|
self._nostr_service = (
|
|
nostr_service_factory() if callable(nostr_service_factory) else None
|
|
)
|
|
except Exception as exc:
|
|
self._nostr_service = None
|
|
self._log_activity(f"Nostr service unavailable: {exc}")
|
|
try:
|
|
self._sync_service = (
|
|
sync_service_factory() if callable(sync_service_factory) else None
|
|
)
|
|
except Exception as exc:
|
|
self._sync_service = None
|
|
self._log_activity(f"Sync service unavailable: {exc}")
|
|
try:
|
|
self._utility_service = (
|
|
utility_service_factory()
|
|
if callable(utility_service_factory)
|
|
else None
|
|
)
|
|
except Exception as exc:
|
|
self._utility_service = None
|
|
self._log_activity(f"Utility service unavailable: {exc}")
|
|
try:
|
|
self._vault_service = (
|
|
vault_service_factory() if callable(vault_service_factory) else None
|
|
)
|
|
except Exception as exc:
|
|
self._vault_service = None
|
|
self._log_activity(f"Vault service unavailable: {exc}")
|
|
self._update_filters_panel()
|
|
self._update_help_overlay()
|
|
self._update_activity_panel()
|
|
self._apply_focus_style()
|
|
self._set_secret_panel(
|
|
"Sensitive data hidden. Use 'v' to reveal 🔑 or 'g' for QR ▦."
|
|
)
|
|
self._load_entries()
|
|
|
|
def _set_status(self, message: str) -> None:
|
|
if message == self._last_status_message:
|
|
return
|
|
self._last_status_message = message
|
|
mode = (
|
|
"PALETTE"
|
|
if self.palette_open
|
|
else ("EDIT" if self.editing_document else "VIEW")
|
|
)
|
|
text = f"[{mode} | {self._focus_pane.upper()}] {message}"
|
|
self.query_one("#status", Static).update(text)
|
|
self._log_activity(message)
|
|
|
|
def _log_activity(self, message: str) -> None:
|
|
msg = message.strip()
|
|
if not msg:
|
|
return
|
|
self._activity_log.append(msg)
|
|
self._activity_log = self._activity_log[-5:]
|
|
self._update_activity_panel()
|
|
|
|
def _update_activity_panel(self) -> None:
|
|
if not self._activity_log:
|
|
text = "Activity Log\n\nNo actions yet."
|
|
else:
|
|
lines = ["Activity Log", ""]
|
|
for i, item in enumerate(reversed(self._activity_log), start=1):
|
|
short = item if len(item) <= 88 else f"{item[:85]}..."
|
|
lines.append(f"{i}. {short}")
|
|
text = "\n".join(lines)
|
|
self.query_one("#activity", Static).update(text)
|
|
|
|
def _set_secret_panel(self, text: str) -> None:
|
|
self.query_one("#secret-detail", Static).update(text)
|
|
|
|
def _update_help_overlay(self) -> None:
|
|
box = self.query_one("#help-overlay", Static)
|
|
if not self.help_open:
|
|
box.add_class("hidden")
|
|
return
|
|
box.remove_class("hidden")
|
|
text = "\n".join(
|
|
[
|
|
"TUI v2 Quick Help (Esc to close)",
|
|
"",
|
|
"Core : / search j jump-id p/n page f kind h archive scope r refresh",
|
|
"Modes : Ctrl+P palette e edit-doc Ctrl+S save Esc cancel/close",
|
|
"Graph ⚯ : l relation filter brackets link select o open link target",
|
|
"Secrets 🔑: v reveal selected secret g QR for selected entry",
|
|
"Resilience: x retry last error",
|
|
"Pane Focus: 1 left 2 center 3 right",
|
|
"",
|
|
"Palette examples",
|
|
(
|
|
'help | 2fa-board | 2fa-copy 12 | add-password "Site" 20 '
|
|
"user https://x | profiles-list | checksum-verify | db-export backup.enc"
|
|
),
|
|
]
|
|
)
|
|
box.update(text)
|
|
|
|
def _apply_focus_style(self) -> None:
|
|
for pane_id in ("left", "center", "right"):
|
|
pane = self.query_one(f"#{pane_id}", Vertical)
|
|
if pane_id == self._focus_pane:
|
|
pane.add_class("pane-focus")
|
|
else:
|
|
pane.remove_class("pane-focus")
|
|
|
|
def _refresh_doc_edit_help(self) -> None:
|
|
marker = "*" if self._doc_dirty else "clean"
|
|
self.query_one("#doc-edit-help", Static).update(
|
|
f"Edit mode [{marker}]: Ctrl+S save, Esc cancel"
|
|
)
|
|
|
|
def _record_failure(
|
|
self,
|
|
context: str,
|
|
exc: Exception,
|
|
*,
|
|
retry: Any | None = None,
|
|
hint: str = "",
|
|
) -> None:
|
|
self._last_error = f"{context}: {exc}"
|
|
self._retry_action = retry
|
|
suffix = f" {hint}" if hint else ""
|
|
self._set_status(f"{self._last_error}.{suffix}")
|
|
|
|
def _clear_failure(self) -> None:
|
|
self._last_error = None
|
|
self._retry_action = None
|
|
|
|
def _retry_initialize_service(self) -> None:
|
|
try:
|
|
self._service = (
|
|
entry_service_factory() if callable(entry_service_factory) else None
|
|
)
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"Unable to initialize entry service",
|
|
exc,
|
|
retry=self._retry_initialize_service,
|
|
hint="Press 'x' to retry initialization.",
|
|
)
|
|
return
|
|
self._clear_failure()
|
|
self._set_status("Entry service initialized")
|
|
self._load_entries(self._last_query, reset_page=False)
|
|
|
|
def _current_filter_kinds(self) -> list[str] | None:
|
|
if self.filter_kind == "all":
|
|
return None
|
|
return [self.filter_kind]
|
|
|
|
def _current_archive_scope_flags(self) -> tuple[bool, bool]:
|
|
if self.archive_scope == "all":
|
|
return True, False
|
|
if self.archive_scope == "archived":
|
|
return True, True
|
|
return False, False
|
|
|
|
def _selected_summary(self) -> str:
|
|
if (
|
|
not isinstance(self._selected_entry, dict)
|
|
or self._selected_entry_id is None
|
|
):
|
|
return "Selected: (none)"
|
|
kind = str(
|
|
self._selected_entry.get("kind")
|
|
or self._selected_entry.get("type")
|
|
or "unknown"
|
|
)
|
|
label = str(self._selected_entry.get("label", ""))
|
|
return f"Selected: #{self._selected_entry_id} [{kind}] {label}"
|
|
|
|
def _update_filters_panel(self) -> None:
|
|
fp_line = (
|
|
f"Fingerprint: {fingerprint}"
|
|
if fingerprint
|
|
else "Fingerprint: (default)"
|
|
)
|
|
text = "\n".join(
|
|
[
|
|
"SeedPass ◈ TUI v2",
|
|
fp_line,
|
|
"",
|
|
self._selected_summary(),
|
|
f"Filter : {self.filter_kind}",
|
|
f"Archive: {self.archive_scope}",
|
|
f"Links : {self.link_relation_filter}",
|
|
(
|
|
f"Results: {len(self._all_results)} | "
|
|
f"Page: {self._result_page + 1}/{self._total_pages()}"
|
|
),
|
|
"",
|
|
"Nav ⌘",
|
|
"- / search",
|
|
"- j jump to id",
|
|
"- f cycle kind filter",
|
|
"- h cycle archive scope",
|
|
"- p/n prev/next page",
|
|
"- 1/2/3 focus pane",
|
|
"- ? help overlay",
|
|
"- r refresh",
|
|
"- x retry last error",
|
|
"",
|
|
"Actions ⚡",
|
|
"- a archive/restore",
|
|
"- e edit document",
|
|
"- Ctrl+S save doc",
|
|
"- Ctrl+P command palette",
|
|
"- add-* create entries",
|
|
"- tags/notes/fields via palette",
|
|
"- docs: edit/save/export",
|
|
"- 6 toggle dedicated 2FA board",
|
|
"- profile/relay/sync/settings",
|
|
"- npub + nostr reset/fresh namespace",
|
|
"- checksum/db/totp/seed backup",
|
|
"- managed-load/managed-exit session",
|
|
"- l cycle link relation",
|
|
"- [ / ] select link",
|
|
"- o open link target",
|
|
"- v reveal selected secret",
|
|
"- g show selected QR",
|
|
"- Esc cancel/close",
|
|
]
|
|
)
|
|
self.query_one("#filters", Static).update(text)
|
|
|
|
def _render_entry_label(
|
|
self, idx: int, label: str, etype: str, archived: bool
|
|
) -> str:
|
|
arch = " [archived]" if archived else ""
|
|
return f"{idx:>4} {etype:<15} {label}{arch}"
|
|
|
|
def _total_pages(self) -> int:
|
|
_page, _start, _end, total_pages = pagination_window(
|
|
len(self._all_results), self.RESULT_PAGE_SIZE, self._result_page
|
|
)
|
|
return total_pages
|
|
|
|
def _render_current_page(self, *, preserve_selected: bool = True) -> None:
|
|
self._entry_ids_in_view = []
|
|
list_view = self.query_one("#entry-list", ListView)
|
|
list_view.clear()
|
|
total = len(self._all_results)
|
|
if total == 0:
|
|
self._selected_entry_id = None
|
|
self._selected_entry = None
|
|
self.query_one("#entry-detail", Static).update("No entries match.")
|
|
self.query_one("#link-detail", Static).update(
|
|
"Links: select an entry first."
|
|
)
|
|
self._set_secret_panel(
|
|
"Sensitive data hidden. Select an entry, then use 'v' (reveal) or 'g' (QR)."
|
|
)
|
|
self._current_links = []
|
|
self._current_link_cursor = 0
|
|
self._set_status("No entries match current filter/search")
|
|
self._update_filters_panel()
|
|
return
|
|
|
|
self._result_page, start, end, _total_pages = pagination_window(
|
|
total, self.RESULT_PAGE_SIZE, self._result_page
|
|
)
|
|
page_rows = self._all_results[start:end]
|
|
for idx, label, _username, _url, archived, etype in page_rows:
|
|
kind = getattr(etype, "value", str(etype))
|
|
item = EntryListItem(
|
|
idx,
|
|
self._render_entry_label(idx, label, kind, bool(archived)),
|
|
)
|
|
self._entry_ids_in_view.append(int(idx))
|
|
list_view.append(item)
|
|
|
|
self._update_filters_panel()
|
|
chosen_id = None
|
|
if preserve_selected and self._selected_entry_id in self._entry_ids_in_view:
|
|
chosen_id = self._selected_entry_id
|
|
elif self._entry_ids_in_view:
|
|
chosen_id = self._entry_ids_in_view[0]
|
|
if chosen_id is not None:
|
|
self._show_entry(chosen_id)
|
|
|
|
def _entry_detail_text(self, entry: dict[str, Any]) -> str:
|
|
payload = truncate_entry_for_display(
|
|
entry, self.DETAIL_CONTENT_PREVIEW_LIMIT
|
|
)
|
|
entry_id = (
|
|
self._selected_entry_id
|
|
if self._selected_entry_id is not None
|
|
else payload.get("id", "?")
|
|
)
|
|
kind = payload.get("kind") or payload.get("type") or "unknown"
|
|
header = f"Entry #{entry_id} [{kind}]"
|
|
return f"{header}\n{'-' * len(header)}\n{json.dumps(payload, indent=2, sort_keys=True)}"
|
|
|
|
def _load_entries(self, query: str = "", *, reset_page: bool = False) -> None:
|
|
self._last_query = query
|
|
if self._service is None:
|
|
self.query_one("#entry-detail", Static).update(
|
|
"Entry service unavailable in this runtime."
|
|
)
|
|
self.query_one("#link-detail", Static).update("Links unavailable.")
|
|
self._set_secret_panel(
|
|
"Sensitive data unavailable: entry service missing."
|
|
)
|
|
self._all_results = []
|
|
self._result_page = 0
|
|
self._current_links = []
|
|
self._current_link_cursor = 0
|
|
self._update_filters_panel()
|
|
return
|
|
|
|
try:
|
|
include_archived, archived_only = self._current_archive_scope_flags()
|
|
results = self._service.search_entries(
|
|
query,
|
|
kinds=self._current_filter_kinds(),
|
|
include_archived=include_archived,
|
|
archived_only=archived_only,
|
|
)
|
|
except Exception as exc:
|
|
self.query_one("#entry-detail", Static).update(
|
|
f"Failed to load entries: {exc}"
|
|
)
|
|
self.query_one("#link-detail", Static).update("Links unavailable.")
|
|
self._set_secret_panel(
|
|
"Sensitive data unavailable: failed to load entries."
|
|
)
|
|
self._all_results = []
|
|
self._result_page = 0
|
|
self._current_links = []
|
|
self._current_link_cursor = 0
|
|
self._update_filters_panel()
|
|
self._record_failure(
|
|
"Failed to load entries",
|
|
exc,
|
|
retry=lambda: self._load_entries(
|
|
self._last_query, reset_page=False
|
|
),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
self._all_results = list(results)
|
|
self._clear_failure()
|
|
if reset_page:
|
|
self._result_page = 0
|
|
self._render_current_page(preserve_selected=not reset_page)
|
|
|
|
def _show_entry(self, entry_index: int) -> None:
|
|
if self._service is None:
|
|
return
|
|
try:
|
|
entry = self._service.retrieve_entry(entry_index)
|
|
if not isinstance(entry, dict):
|
|
self.query_one("#entry-detail", Static).update("Entry not found.")
|
|
self.query_one("#link-detail", Static).update(
|
|
"Links: entry not found."
|
|
)
|
|
self._current_links = []
|
|
self._current_link_cursor = 0
|
|
self._set_status(f"Entry {entry_index} not found")
|
|
return
|
|
self._selected_entry_id = int(entry_index)
|
|
self._selected_entry = dict(entry)
|
|
body = self._entry_detail_text(entry)
|
|
self.query_one("#entry-detail", Static).update(body)
|
|
self._set_secret_panel(
|
|
"Sensitive data hidden. Use 'v' to reveal 🔑 or 'g' for QR ▦."
|
|
)
|
|
self._update_links_panel()
|
|
self._update_filters_panel()
|
|
self._set_status(f"Selected entry {entry_index}")
|
|
except Exception as exc:
|
|
self.query_one("#entry-detail", Static).update(
|
|
f"Failed to load entry {entry_index}: {exc}"
|
|
)
|
|
self.query_one("#link-detail", Static).update("Links unavailable.")
|
|
self._set_secret_panel(
|
|
"Sensitive data unavailable: failed to load selected entry."
|
|
)
|
|
self._current_links = []
|
|
self._current_link_cursor = 0
|
|
self._record_failure(
|
|
f"Failed to load entry {entry_index}",
|
|
exc,
|
|
retry=lambda: self._show_entry(entry_index),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
|
|
def _update_links_panel(self) -> None:
|
|
if self._service is None or self._selected_entry_id is None:
|
|
self.query_one("#link-detail", Static).update(
|
|
"Links: select an entry first."
|
|
)
|
|
self._current_links = []
|
|
self._current_link_cursor = 0
|
|
return
|
|
try:
|
|
links = self._service.get_links(self._selected_entry_id)
|
|
except Exception as exc:
|
|
self.query_one("#link-detail", Static).update(
|
|
f"Links unavailable: {exc}"
|
|
)
|
|
self._current_links = []
|
|
self._current_link_cursor = 0
|
|
self._record_failure(
|
|
"Failed to load entry links",
|
|
exc,
|
|
retry=self._update_links_panel,
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if self.link_relation_filter == "all":
|
|
filtered = [lnk for lnk in links if isinstance(lnk, dict)]
|
|
else:
|
|
filtered = [
|
|
lnk
|
|
for lnk in links
|
|
if isinstance(lnk, dict)
|
|
and str(lnk.get("relation", "related_to")).lower()
|
|
== self.link_relation_filter
|
|
]
|
|
self._current_links = filtered
|
|
if self._current_link_cursor >= len(self._current_links):
|
|
self._current_link_cursor = 0
|
|
|
|
if not links:
|
|
self.query_one("#link-detail", Static).update(
|
|
f"Links for #{self._selected_entry_id}\n\nNo graph links for this entry.\n"
|
|
"Use Ctrl+P and run: link-add <target_id> [relation] [note]"
|
|
)
|
|
return
|
|
if not filtered:
|
|
self.query_one("#link-detail", Static).update(
|
|
f"Links for #{self._selected_entry_id}\n\n"
|
|
f"Relation filter: {self.link_relation_filter}\n\n"
|
|
"No links match this relation filter.\n"
|
|
"Press 'l' to cycle relation filter."
|
|
)
|
|
return
|
|
|
|
lines = [
|
|
f"Links for #{self._selected_entry_id}",
|
|
"",
|
|
f"Relation filter: {self.link_relation_filter}",
|
|
(
|
|
f"Selected link: {self._current_link_cursor + 1}/"
|
|
f"{len(self._current_links)}"
|
|
),
|
|
"",
|
|
"Format: relation -> target_id (note)",
|
|
"",
|
|
]
|
|
for i, link in enumerate(self._current_links):
|
|
target = link.get("target")
|
|
relation = link.get("relation", "related_to")
|
|
note = str(link.get("note", "")).strip()
|
|
prefix = ">" if i == self._current_link_cursor else " "
|
|
if note:
|
|
lines.append(f"{prefix} {relation} -> {target} ({note})")
|
|
else:
|
|
lines.append(f"{prefix} {relation} -> {target}")
|
|
self.query_one("#link-detail", Static).update("\n".join(lines))
|
|
self._clear_failure()
|
|
|
|
def _is_selected_document(self) -> bool:
|
|
if not isinstance(self._selected_entry, dict):
|
|
return False
|
|
kind = str(
|
|
self._selected_entry.get("kind") or self._selected_entry.get("type")
|
|
)
|
|
return kind == "document"
|
|
|
|
def _selected_kind(self) -> str:
|
|
if not isinstance(self._selected_entry, dict):
|
|
return ""
|
|
return (
|
|
str(
|
|
self._selected_entry.get("kind")
|
|
or self._selected_entry.get("type")
|
|
or ""
|
|
)
|
|
.strip()
|
|
.lower()
|
|
)
|
|
|
|
@staticmethod
|
|
def _kind_icon(kind: str) -> str:
|
|
return {
|
|
"password": "🗝",
|
|
"totp": "📱",
|
|
"ssh": "🖧",
|
|
"pgp": "🔒",
|
|
"nostr": "⚡",
|
|
"document": "📄",
|
|
"managed_account": "👥",
|
|
"seed": "🌱",
|
|
}.get(kind, "•")
|
|
|
|
def _is_secret_mode_enabled(self) -> bool:
|
|
if self._service is None:
|
|
return False
|
|
getter = getattr(self._service, "get_secret_mode_enabled", None)
|
|
if not callable(getter):
|
|
return False
|
|
try:
|
|
return bool(getter())
|
|
except Exception:
|
|
return False
|
|
|
|
def _clipboard_clear_delay(self) -> int:
|
|
if self._service is None:
|
|
return 30
|
|
getter = getattr(self._service, "get_clipboard_clear_delay", None)
|
|
if not callable(getter):
|
|
return 30
|
|
try:
|
|
return int(getter())
|
|
except Exception:
|
|
return 30
|
|
|
|
def _copy_to_clipboard(self, value: str) -> bool:
|
|
if self._service is None:
|
|
return False
|
|
copier = getattr(self._service, "copy_to_clipboard", None)
|
|
if not callable(copier):
|
|
return False
|
|
try:
|
|
return bool(copier(value))
|
|
except Exception:
|
|
return False
|
|
|
|
def _resolve_selected_sensitive_payload(
|
|
self, *, qr_mode: str = "default"
|
|
) -> tuple[str, str, str | None, str | None, str]:
|
|
if self._service is None:
|
|
raise ValueError("Entry service unavailable")
|
|
if self._selected_entry_id is None:
|
|
raise ValueError("No entry selected")
|
|
|
|
entry = self._selected_entry or self._service.retrieve_entry(
|
|
self._selected_entry_id
|
|
)
|
|
if not isinstance(entry, dict) or not entry:
|
|
raise ValueError("Selected entry not found")
|
|
|
|
kind = (
|
|
self._selected_kind()
|
|
or str(entry.get("kind") or entry.get("type") or "").strip().lower()
|
|
)
|
|
label = str(entry.get("label", f"entry-{self._selected_entry_id}"))
|
|
icon = self._kind_icon(kind)
|
|
|
|
if kind == "password":
|
|
length = int(entry.get("length", 16))
|
|
password = self._service.generate_password(
|
|
length, self._selected_entry_id
|
|
)
|
|
return (
|
|
f"Sensitive {icon}: Password #{self._selected_entry_id}",
|
|
f"Label : {label}\nPassword : {password}",
|
|
None,
|
|
password,
|
|
kind,
|
|
)
|
|
|
|
if kind == "seed":
|
|
phrase = self._service.get_seed_phrase(self._selected_entry_id)
|
|
from seedpass.core.seedqr import encode_seedqr
|
|
|
|
seedqr = encode_seedqr(phrase)
|
|
return (
|
|
f"Sensitive {icon}: Seed #{self._selected_entry_id}",
|
|
f"Label : {label}\nSeed Phrase : {phrase}",
|
|
seedqr,
|
|
phrase,
|
|
kind,
|
|
)
|
|
|
|
if kind == "managed_account":
|
|
phrase = self._service.get_managed_account_seed(self._selected_entry_id)
|
|
from seedpass.core.seedqr import encode_seedqr
|
|
|
|
seedqr = encode_seedqr(phrase)
|
|
return (
|
|
f"Sensitive {icon}: Managed Account Seed #{self._selected_entry_id}",
|
|
f"Label : {label}\nSeed Phrase : {phrase}",
|
|
seedqr,
|
|
phrase,
|
|
kind,
|
|
)
|
|
|
|
if kind == "totp":
|
|
secret = self._service.get_totp_secret(self._selected_entry_id)
|
|
code = self._service.get_totp_code(self._selected_entry_id)
|
|
period = int(entry.get("period", 30))
|
|
digits = int(entry.get("digits", 6))
|
|
from seedpass.core.totp import TotpManager
|
|
|
|
uri = TotpManager.make_otpauth_uri(
|
|
label, secret, period=period, digits=digits
|
|
)
|
|
return (
|
|
f"Sensitive {icon}: TOTP #{self._selected_entry_id}",
|
|
(
|
|
f"Label : {label}\n"
|
|
f"Code : {code}\n"
|
|
f"Secret : {secret}\n"
|
|
f"Period : {period}s\n"
|
|
f"Digits : {digits}"
|
|
),
|
|
uri,
|
|
code,
|
|
kind,
|
|
)
|
|
|
|
if kind == "ssh":
|
|
priv_key, pub_key = self._service.get_ssh_key_pair(
|
|
self._selected_entry_id
|
|
)
|
|
return (
|
|
f"Sensitive {icon}: SSH #{self._selected_entry_id}",
|
|
f"Label : {label}\nPublic Key : {pub_key}\nPrivate Key : {priv_key}",
|
|
None,
|
|
priv_key,
|
|
kind,
|
|
)
|
|
|
|
if kind == "pgp":
|
|
priv_key, fingerprint_text = self._service.get_pgp_key(
|
|
self._selected_entry_id
|
|
)
|
|
return (
|
|
f"Sensitive {icon}: PGP #{self._selected_entry_id}",
|
|
(
|
|
f"Label : {label}\n"
|
|
f"Fingerprint : {fingerprint_text}\n"
|
|
f"Private Key :\n{priv_key}"
|
|
),
|
|
None,
|
|
priv_key,
|
|
kind,
|
|
)
|
|
|
|
if kind == "nostr":
|
|
npub, nsec = self._service.get_nostr_key_pair(self._selected_entry_id)
|
|
if qr_mode == "private":
|
|
qr_payload = nsec
|
|
else:
|
|
qr_payload = f"nostr:{npub}"
|
|
return (
|
|
f"Sensitive {icon}: Nostr #{self._selected_entry_id}",
|
|
f"Label : {label}\nnpub : {npub}\nnsec : {nsec}",
|
|
qr_payload,
|
|
nsec,
|
|
kind,
|
|
)
|
|
|
|
raise ValueError(
|
|
"Reveal unsupported for kind "
|
|
f"'{kind or 'unknown'}'. Supported: password, seed, managed_account, "
|
|
"totp, ssh, pgp, nostr."
|
|
)
|
|
|
|
def _requires_confirm(
|
|
self, *, kind: str, include_qr: bool, qr_mode: str
|
|
) -> bool:
|
|
if include_qr:
|
|
return kind == "nostr" and qr_mode == "private"
|
|
return kind in {"seed", "managed_account", "ssh", "pgp"}
|
|
|
|
def _show_sensitive_panel(
|
|
self, *, include_qr: bool, qr_mode: str = "default", confirm: bool = False
|
|
) -> None:
|
|
try:
|
|
title, body, qr_data, secret_value, kind = (
|
|
self._resolve_selected_sensitive_payload(qr_mode=qr_mode)
|
|
)
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"Sensitive reveal failed",
|
|
exc,
|
|
retry=(
|
|
(lambda: self.action_show_qr(mode=qr_mode, confirm=confirm))
|
|
if include_qr
|
|
else (lambda: self.action_reveal_selected(confirm=confirm))
|
|
),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if (
|
|
self._requires_confirm(
|
|
kind=kind, include_qr=include_qr, qr_mode=qr_mode
|
|
)
|
|
and not confirm
|
|
):
|
|
if include_qr:
|
|
self._set_secret_panel(
|
|
"Sensitive QR hidden.\n"
|
|
"Confirmation required for this action.\n"
|
|
"Run: qr private confirm"
|
|
)
|
|
self._set_status(
|
|
"Sensitive QR requires confirmation. Run: qr private confirm"
|
|
)
|
|
else:
|
|
self._set_secret_panel(
|
|
"Sensitive reveal hidden.\n"
|
|
"Confirmation required for this action.\n"
|
|
"Run: reveal confirm"
|
|
)
|
|
self._set_status(
|
|
"Sensitive reveal requires confirmation. Run: reveal confirm"
|
|
)
|
|
return
|
|
|
|
if include_qr:
|
|
if not qr_data:
|
|
self._set_secret_panel(
|
|
f"{title}\n\nQR not supported for this entry."
|
|
)
|
|
self._clear_failure()
|
|
self._set_status("QR not available for selected entry")
|
|
return
|
|
try:
|
|
qr_ascii = render_qr_ascii(qr_data)
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"QR render failed",
|
|
exc,
|
|
retry=self.action_show_qr,
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
self._set_secret_panel(
|
|
f"{title} QR\n\n{qr_ascii}\n\nPayload: {qr_data}"
|
|
)
|
|
self._clear_failure()
|
|
self._set_status("Rendered QR for selected entry")
|
|
return
|
|
|
|
self._set_secret_panel(f"{title}\n\n{body}")
|
|
self._clear_failure()
|
|
if self._is_secret_mode_enabled() and secret_value:
|
|
if self._copy_to_clipboard(secret_value):
|
|
delay = self._clipboard_clear_delay()
|
|
self._set_secret_panel(
|
|
f"{title}\n\nSecret mode is enabled.\n"
|
|
f"Sensitive value copied to clipboard (auto-clear in {delay}s)."
|
|
)
|
|
self._set_status("Sensitive value copied to clipboard")
|
|
return
|
|
self._set_status("Revealed selected sensitive data")
|
|
|
|
def _set_document_editor_visible(self, visible: bool) -> None:
|
|
view = self.query_one("#right-view", Vertical)
|
|
editor = self.query_one("#right-editor", Vertical)
|
|
if visible:
|
|
view.add_class("hidden")
|
|
editor.remove_class("hidden")
|
|
else:
|
|
editor.add_class("hidden")
|
|
view.remove_class("hidden")
|
|
self.editing_document = visible
|
|
self._focus_pane = "right"
|
|
self._apply_focus_style()
|
|
|
|
def _set_palette_visible(self, visible: bool) -> None:
|
|
palette = self.query_one("#command-palette", Input)
|
|
if visible:
|
|
palette.remove_class("hidden")
|
|
palette.focus()
|
|
else:
|
|
palette.value = ""
|
|
palette.add_class("hidden")
|
|
self.palette_open = visible
|
|
self._update_help_overlay()
|
|
|
|
def _set_totp_board_visible(self, visible: bool) -> None:
|
|
board = self.query_one("#totp-board", Static)
|
|
entry_detail = self.query_one("#entry-detail", Static)
|
|
link_detail = self.query_one("#link-detail", Static)
|
|
secret_detail = self.query_one("#secret-detail", Static)
|
|
if visible:
|
|
entry_detail.add_class("hidden")
|
|
link_detail.add_class("hidden")
|
|
secret_detail.add_class("hidden")
|
|
board.remove_class("hidden")
|
|
self.totp_board_open = True
|
|
self._refresh_totp_board(force_reload=True)
|
|
else:
|
|
board.add_class("hidden")
|
|
entry_detail.remove_class("hidden")
|
|
link_detail.remove_class("hidden")
|
|
secret_detail.remove_class("hidden")
|
|
self.totp_board_open = False
|
|
|
|
@staticmethod
|
|
def _totp_source(entry: dict[str, Any]) -> str:
|
|
deterministic = entry.get("deterministic")
|
|
if isinstance(deterministic, bool):
|
|
return "det" if deterministic else "imp"
|
|
secret = str(entry.get("secret", "")).strip()
|
|
return "imp" if secret else "det"
|
|
|
|
def _load_totp_rows(self) -> None:
|
|
if self._service is None:
|
|
self._totp_rows = []
|
|
return
|
|
rows: list[dict[str, Any]] = []
|
|
results = self._service.search_entries("", kinds=["totp"])
|
|
for row in results:
|
|
entry_id = int(row[0])
|
|
label = str(row[1])
|
|
entry = self._service.retrieve_entry(entry_id)
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
period = int(entry.get("period", 30))
|
|
digits = int(entry.get("digits", 6))
|
|
rows.append(
|
|
{
|
|
"id": entry_id,
|
|
"label": label,
|
|
"period": period,
|
|
"digits": digits,
|
|
"source": self._totp_source(entry),
|
|
}
|
|
)
|
|
self._totp_rows = rows
|
|
|
|
def _refresh_totp_board(self, *, force_reload: bool = False) -> None:
|
|
if not self.totp_board_open:
|
|
return
|
|
if self._service is None:
|
|
self.query_one("#totp-board", Static).update(
|
|
"2FA Board\n\nEntry service unavailable."
|
|
)
|
|
return
|
|
try:
|
|
if force_reload or not self._totp_rows:
|
|
self._load_totp_rows()
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"Failed to load 2FA board",
|
|
exc,
|
|
retry=lambda: self._refresh_totp_board(force_reload=True),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
now = int(self._time_now())
|
|
secret_mode = self._is_secret_mode_enabled()
|
|
lines = [
|
|
"2FA Board",
|
|
"",
|
|
"id src rem code label",
|
|
"-- --- --- -------- -----",
|
|
]
|
|
if not self._totp_rows:
|
|
lines.extend(
|
|
["", "No TOTP entries found.", "Use add-totp to create one."]
|
|
)
|
|
else:
|
|
for row in self._totp_rows:
|
|
period = max(1, int(row["period"]))
|
|
rem = period - (now % period)
|
|
if rem <= 0:
|
|
rem = period
|
|
code = (
|
|
"******"
|
|
if secret_mode
|
|
else self._service.get_totp_code(int(row["id"]))
|
|
)
|
|
lines.append(
|
|
f"{row['id']:>2} {row['source']:<3} {rem:>3} {str(code):<8} {row['label']}"
|
|
)
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"source: det=deterministic, imp=imported",
|
|
"Commands: 2fa-copy <entry_id>, 2fa-refresh, 2fa-hide",
|
|
]
|
|
)
|
|
self.query_one("#totp-board", Static).update("\n".join(lines))
|
|
self._clear_failure()
|
|
|
|
def _tick_totp_board(self) -> None:
|
|
if self.totp_board_open:
|
|
self._refresh_totp_board(force_reload=False)
|
|
|
|
def _get_document_editor_text(self) -> str:
|
|
if TextArea is not None:
|
|
area = self.query_one("#doc-edit-content")
|
|
return str(getattr(area, "text", ""))
|
|
return self.query_one("#doc-edit-content-single", Input).value
|
|
|
|
def _set_document_editor_text(self, content: str) -> None:
|
|
if TextArea is not None:
|
|
area = self.query_one("#doc-edit-content")
|
|
if hasattr(area, "load_text"):
|
|
area.load_text(content)
|
|
elif hasattr(area, "text"):
|
|
area.text = content
|
|
return
|
|
self.query_one("#doc-edit-content-single", Input).value = content
|
|
|
|
def _mark_doc_dirty(self, dirty: bool = True) -> None:
|
|
if not self.editing_document:
|
|
return
|
|
if self._doc_dirty == dirty:
|
|
return
|
|
self._doc_dirty = dirty
|
|
self._refresh_doc_edit_help()
|
|
|
|
def _selected_entry_payload(self) -> dict[str, Any]:
|
|
if self._service is None:
|
|
raise ValueError("Entry service unavailable")
|
|
if self._selected_entry_id is None:
|
|
raise ValueError("No entry selected")
|
|
entry = self._selected_entry or self._service.retrieve_entry(
|
|
self._selected_entry_id
|
|
)
|
|
if not isinstance(entry, dict) or not entry:
|
|
raise ValueError("Selected entry not found")
|
|
return dict(entry)
|
|
|
|
def _apply_selected_modify(
|
|
self,
|
|
*,
|
|
raw_command: str,
|
|
success_message: str,
|
|
failure_context: str,
|
|
hint: str = "Press 'x' to retry.",
|
|
**kwargs: Any,
|
|
) -> None:
|
|
if self._service is None or self._selected_entry_id is None:
|
|
self._set_status("No entry selected")
|
|
return
|
|
try:
|
|
self._service.modify_entry(self._selected_entry_id, **kwargs)
|
|
current_id = self._selected_entry_id
|
|
self._load_entries(self._last_query, reset_page=False)
|
|
if current_id in self._entry_ids_in_view:
|
|
self._show_entry(current_id)
|
|
self._clear_failure()
|
|
self._set_status(success_message)
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
failure_context,
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw_command),
|
|
hint=hint,
|
|
)
|
|
|
|
@staticmethod
|
|
def _parse_toggle_token(token: str) -> bool:
|
|
value = token.strip().lower()
|
|
if value in {"1", "true", "yes", "y", "on", "enable", "enabled"}:
|
|
return True
|
|
if value in {"0", "false", "no", "n", "off", "disable", "disabled"}:
|
|
return False
|
|
raise ValueError(f"invalid toggle value '{token}'")
|
|
|
|
def _run_palette_command(self, command: str) -> None:
|
|
raw = command.strip()
|
|
try:
|
|
cmd, args = parse_palette_command(raw)
|
|
except ValueError as exc:
|
|
self._set_status(str(exc))
|
|
return
|
|
|
|
if cmd == "help":
|
|
self._set_status(
|
|
"Palette commands: help, open, search, filter, archive, "
|
|
"restore, archive-filter, edit-doc, save-doc, cancel-edit, link-add, link-rm, "
|
|
"add-password, add-totp, add-key-value, add-document, add-ssh, "
|
|
"add-pgp, add-nostr, add-seed, add-managed-account, "
|
|
"notes-set, notes-clear, tag-add, tag-rm, tags-set, tags-clear, "
|
|
"field-add, field-rm, set-field, clear-field, doc-export, "
|
|
"2fa-board, 2fa-hide, 2fa-refresh, 2fa-copy, "
|
|
"profiles-list, profile-switch, profile-add, profile-remove, profile-rename, "
|
|
"setting-secret, setting-offline, setting-quick-unlock, setting-timeout, "
|
|
"setting-kdf-iterations, setting-kdf-mode, "
|
|
"relay-list, relay-add, relay-rm, relay-reset, "
|
|
"npub, nostr-reset-sync-state, nostr-fresh-namespace, "
|
|
"sync-now, sync-bg, "
|
|
"checksum-verify, checksum-update, db-export, db-import, "
|
|
"totp-export, parent-seed-backup, managed-load, managed-exit, "
|
|
"reveal [confirm], qr [public|private] [confirm], link-filter, "
|
|
"link-next, link-prev, link-open, page-next, page-prev, page <n>, "
|
|
"retry, jump <id>"
|
|
)
|
|
return
|
|
if cmd == "retry":
|
|
self.action_retry_last_error()
|
|
return
|
|
|
|
if cmd == "open":
|
|
if len(args) != 1:
|
|
self._set_status("Usage: open <entry_id>")
|
|
return
|
|
try:
|
|
entry_id = int(args[0])
|
|
except ValueError:
|
|
self._set_status("open requires integer entry_id")
|
|
return
|
|
self._show_entry(entry_id)
|
|
return
|
|
if cmd == "jump":
|
|
if len(args) != 1:
|
|
self._set_status("Usage: jump <entry_id>")
|
|
return
|
|
try:
|
|
entry_id = int(args[0])
|
|
except ValueError:
|
|
self._set_status("jump requires integer entry_id")
|
|
return
|
|
self._show_entry(entry_id)
|
|
return
|
|
|
|
if cmd == "search":
|
|
query = " ".join(args)
|
|
self.query_one("#search", Input).value = query
|
|
self._load_entries(query=query, reset_page=True)
|
|
self._set_status(f"Applied search: {query}")
|
|
return
|
|
|
|
if cmd == "filter":
|
|
if len(args) != 1:
|
|
self._set_status("Usage: filter <kind|all>")
|
|
return
|
|
self.filter_kind = args[0].strip().lower()
|
|
self._update_filters_panel()
|
|
self._load_entries(query=self._last_query, reset_page=True)
|
|
self._set_status(f"Applied filter: {self.filter_kind}")
|
|
return
|
|
|
|
if cmd == "archive-filter":
|
|
if len(args) != 1:
|
|
self._set_status("Usage: archive-filter <active|all|archived>")
|
|
return
|
|
mode = args[0].strip().lower()
|
|
if mode not in {"active", "all", "archived"}:
|
|
self._set_status(
|
|
"archive-filter must be one of: active, all, archived"
|
|
)
|
|
return
|
|
self.archive_scope = mode
|
|
self._update_filters_panel()
|
|
self._load_entries(query=self._last_query, reset_page=True)
|
|
self._set_status(f"Applied archive filter: {self.archive_scope}")
|
|
return
|
|
|
|
if cmd == "add-password":
|
|
if self._service is None:
|
|
self._set_status("Entry service unavailable")
|
|
return
|
|
if len(args) < 2 or len(args) > 4:
|
|
self._set_status(
|
|
"Usage: add-password <label> <length> [username] [url]"
|
|
)
|
|
return
|
|
label = args[0].strip()
|
|
if not label:
|
|
self._set_status("add-password label is required")
|
|
return
|
|
try:
|
|
length = int(args[1])
|
|
except ValueError:
|
|
self._set_status("add-password length must be an integer")
|
|
return
|
|
username = args[2] if len(args) >= 3 else None
|
|
url = args[3] if len(args) >= 4 else None
|
|
try:
|
|
new_id = self._service.add_entry(
|
|
label,
|
|
length,
|
|
username=username,
|
|
url=url,
|
|
)
|
|
self._load_entries(self._last_query, reset_page=False)
|
|
self._show_entry(int(new_id))
|
|
self._clear_failure()
|
|
self._set_status(f"Added password entry #{int(new_id)}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"add-password failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "add-totp":
|
|
if self._service is None:
|
|
self._set_status("Entry service unavailable")
|
|
return
|
|
if len(args) < 1 or len(args) > 4:
|
|
self._set_status(
|
|
"Usage: add-totp <label> [period] [digits] [secret]"
|
|
)
|
|
return
|
|
label = args[0].strip()
|
|
if not label:
|
|
self._set_status("add-totp label is required")
|
|
return
|
|
period = 30
|
|
digits = 6
|
|
secret: str | None = None
|
|
if len(args) >= 2:
|
|
try:
|
|
period = int(args[1])
|
|
except ValueError:
|
|
self._set_status("add-totp period must be an integer")
|
|
return
|
|
if len(args) >= 3:
|
|
try:
|
|
digits = int(args[2])
|
|
except ValueError:
|
|
self._set_status("add-totp digits must be an integer")
|
|
return
|
|
if len(args) >= 4:
|
|
secret = args[3].strip() or None
|
|
deterministic = secret is None
|
|
try:
|
|
self._service.add_totp(
|
|
label,
|
|
period=period,
|
|
digits=digits,
|
|
deterministic=deterministic,
|
|
secret=secret,
|
|
)
|
|
self._load_entries(self._last_query, reset_page=False)
|
|
try:
|
|
matches = self._service.search_entries(label, kinds=["totp"])
|
|
except Exception:
|
|
matches = []
|
|
selected = None
|
|
for row in matches:
|
|
idx, entry_label, _u, _url, _arch, _etype = row
|
|
if str(entry_label) == label:
|
|
selected = int(idx)
|
|
if selected is None and matches:
|
|
selected = int(matches[-1][0])
|
|
if selected is not None:
|
|
self._show_entry(selected)
|
|
self._set_status(f"Added TOTP entry #{selected}")
|
|
else:
|
|
self._set_status("Added TOTP entry")
|
|
self._clear_failure()
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"add-totp failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "add-key-value":
|
|
if self._service is None:
|
|
self._set_status("Entry service unavailable")
|
|
return
|
|
if len(args) != 3:
|
|
self._set_status("Usage: add-key-value <label> <key> <value>")
|
|
return
|
|
label = args[0].strip()
|
|
key = args[1]
|
|
value = args[2]
|
|
if not label:
|
|
self._set_status("add-key-value label is required")
|
|
return
|
|
try:
|
|
new_id = self._service.add_key_value(label, key, value)
|
|
self._load_entries(self._last_query, reset_page=False)
|
|
self._show_entry(int(new_id))
|
|
self._clear_failure()
|
|
self._set_status(f"Added key/value entry #{int(new_id)}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"add-key-value failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "add-document":
|
|
if self._service is None:
|
|
self._set_status("Entry service unavailable")
|
|
return
|
|
if len(args) != 3:
|
|
self._set_status(
|
|
"Usage: add-document <label> <file_type> <content>"
|
|
)
|
|
return
|
|
label = args[0].strip()
|
|
file_type = args[1].strip().lstrip(".") or "txt"
|
|
content = args[2]
|
|
if not label:
|
|
self._set_status("add-document label is required")
|
|
return
|
|
try:
|
|
new_id = self._service.add_document(
|
|
label,
|
|
content,
|
|
file_type=file_type,
|
|
)
|
|
self._load_entries(self._last_query, reset_page=False)
|
|
self._show_entry(int(new_id))
|
|
self._clear_failure()
|
|
self._set_status(f"Added document entry #{int(new_id)}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"add-document failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "add-ssh":
|
|
if self._service is None:
|
|
self._set_status("Entry service unavailable")
|
|
return
|
|
if len(args) < 1 or len(args) > 2:
|
|
self._set_status("Usage: add-ssh <label> [index]")
|
|
return
|
|
label = args[0].strip()
|
|
if not label:
|
|
self._set_status("add-ssh label is required")
|
|
return
|
|
index = None
|
|
if len(args) == 2:
|
|
try:
|
|
index = int(args[1])
|
|
except ValueError:
|
|
self._set_status("add-ssh index must be an integer")
|
|
return
|
|
try:
|
|
new_id = self._service.add_ssh_key(label, index=index)
|
|
self._load_entries(self._last_query, reset_page=False)
|
|
self._show_entry(int(new_id))
|
|
self._clear_failure()
|
|
self._set_status(f"Added SSH entry #{int(new_id)}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"add-ssh failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "add-pgp":
|
|
if self._service is None:
|
|
self._set_status("Entry service unavailable")
|
|
return
|
|
if len(args) < 1 or len(args) > 4:
|
|
self._set_status(
|
|
"Usage: add-pgp <label> [index] [key_type] [user_id]"
|
|
)
|
|
return
|
|
label = args[0].strip()
|
|
if not label:
|
|
self._set_status("add-pgp label is required")
|
|
return
|
|
index = None
|
|
key_type = "ed25519"
|
|
user_id = ""
|
|
if len(args) >= 2:
|
|
try:
|
|
index = int(args[1])
|
|
except ValueError:
|
|
self._set_status("add-pgp index must be an integer")
|
|
return
|
|
if len(args) >= 3:
|
|
key_type = args[2].strip() or "ed25519"
|
|
if len(args) >= 4:
|
|
user_id = args[3]
|
|
try:
|
|
new_id = self._service.add_pgp_key(
|
|
label,
|
|
index=index,
|
|
key_type=key_type,
|
|
user_id=user_id,
|
|
)
|
|
self._load_entries(self._last_query, reset_page=False)
|
|
self._show_entry(int(new_id))
|
|
self._clear_failure()
|
|
self._set_status(f"Added PGP entry #{int(new_id)}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"add-pgp failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "add-nostr":
|
|
if self._service is None:
|
|
self._set_status("Entry service unavailable")
|
|
return
|
|
if len(args) < 1 or len(args) > 2:
|
|
self._set_status("Usage: add-nostr <label> [index]")
|
|
return
|
|
label = args[0].strip()
|
|
if not label:
|
|
self._set_status("add-nostr label is required")
|
|
return
|
|
index = None
|
|
if len(args) == 2:
|
|
try:
|
|
index = int(args[1])
|
|
except ValueError:
|
|
self._set_status("add-nostr index must be an integer")
|
|
return
|
|
try:
|
|
new_id = self._service.add_nostr_key(label, index=index)
|
|
self._load_entries(self._last_query, reset_page=False)
|
|
self._show_entry(int(new_id))
|
|
self._clear_failure()
|
|
self._set_status(f"Added Nostr entry #{int(new_id)}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"add-nostr failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "add-seed":
|
|
if self._service is None:
|
|
self._set_status("Entry service unavailable")
|
|
return
|
|
if len(args) < 1 or len(args) > 3:
|
|
self._set_status("Usage: add-seed <label> [words] [index]")
|
|
return
|
|
label = args[0].strip()
|
|
if not label:
|
|
self._set_status("add-seed label is required")
|
|
return
|
|
words = 24
|
|
index = None
|
|
if len(args) >= 2:
|
|
try:
|
|
words = int(args[1])
|
|
except ValueError:
|
|
self._set_status("add-seed words must be an integer")
|
|
return
|
|
if len(args) == 3:
|
|
try:
|
|
index = int(args[2])
|
|
except ValueError:
|
|
self._set_status("add-seed index must be an integer")
|
|
return
|
|
try:
|
|
new_id = self._service.add_seed(
|
|
label,
|
|
words=words,
|
|
index=index,
|
|
)
|
|
self._load_entries(self._last_query, reset_page=False)
|
|
self._show_entry(int(new_id))
|
|
self._clear_failure()
|
|
self._set_status(f"Added seed entry #{int(new_id)}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"add-seed failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd in ("add-managed-account", "add-managed"):
|
|
if self._service is None:
|
|
self._set_status("Entry service unavailable")
|
|
return
|
|
if len(args) < 1 or len(args) > 2:
|
|
self._set_status("Usage: add-managed-account <label> [index]")
|
|
return
|
|
label = args[0].strip()
|
|
if not label:
|
|
self._set_status("add-managed-account label is required")
|
|
return
|
|
index = None
|
|
if len(args) == 2:
|
|
try:
|
|
index = int(args[1])
|
|
except ValueError:
|
|
self._set_status("add-managed-account index must be an integer")
|
|
return
|
|
try:
|
|
new_id = self._service.add_managed_account(label, index=index)
|
|
self._load_entries(self._last_query, reset_page=False)
|
|
self._show_entry(int(new_id))
|
|
self._clear_failure()
|
|
self._set_status(f"Added managed account entry #{int(new_id)}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"add-managed-account failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "notes-set":
|
|
if not args:
|
|
self._set_status("Usage: notes-set <text>")
|
|
return
|
|
notes_text = " ".join(args).strip()
|
|
if not notes_text:
|
|
self._set_status("notes-set text is required")
|
|
return
|
|
self._apply_selected_modify(
|
|
raw_command=raw,
|
|
success_message="Updated notes for selected entry",
|
|
failure_context="notes-set failed",
|
|
notes=notes_text,
|
|
)
|
|
return
|
|
|
|
if cmd == "notes-clear":
|
|
if args:
|
|
self._set_status("Usage: notes-clear")
|
|
return
|
|
self._apply_selected_modify(
|
|
raw_command=raw,
|
|
success_message="Cleared notes for selected entry",
|
|
failure_context="notes-clear failed",
|
|
notes="",
|
|
)
|
|
return
|
|
|
|
if cmd == "tag-add":
|
|
if len(args) != 1:
|
|
self._set_status("Usage: tag-add <tag>")
|
|
return
|
|
tag = args[0].strip()
|
|
if not tag:
|
|
self._set_status("tag-add tag is required")
|
|
return
|
|
try:
|
|
entry = self._selected_entry_payload()
|
|
except Exception as exc:
|
|
self._set_status(str(exc))
|
|
return
|
|
current_tags = entry.get("tags")
|
|
tags = (
|
|
[str(item).strip() for item in current_tags if str(item).strip()]
|
|
if isinstance(current_tags, list)
|
|
else []
|
|
)
|
|
if tag not in tags:
|
|
tags.append(tag)
|
|
self._apply_selected_modify(
|
|
raw_command=raw,
|
|
success_message=f"Added tag '{tag}'",
|
|
failure_context="tag-add failed",
|
|
tags=tags,
|
|
)
|
|
return
|
|
|
|
if cmd == "tag-rm":
|
|
if len(args) != 1:
|
|
self._set_status("Usage: tag-rm <tag>")
|
|
return
|
|
tag = args[0].strip()
|
|
if not tag:
|
|
self._set_status("tag-rm tag is required")
|
|
return
|
|
try:
|
|
entry = self._selected_entry_payload()
|
|
except Exception as exc:
|
|
self._set_status(str(exc))
|
|
return
|
|
current_tags = entry.get("tags")
|
|
tags = (
|
|
[str(item).strip() for item in current_tags if str(item).strip()]
|
|
if isinstance(current_tags, list)
|
|
else []
|
|
)
|
|
next_tags = [item for item in tags if item != tag]
|
|
self._apply_selected_modify(
|
|
raw_command=raw,
|
|
success_message=f"Removed tag '{tag}'",
|
|
failure_context="tag-rm failed",
|
|
tags=next_tags,
|
|
)
|
|
return
|
|
|
|
if cmd == "tags-set":
|
|
if not args:
|
|
self._set_status("Usage: tags-set <comma-separated tags>")
|
|
return
|
|
raw_tags = " ".join(args).strip()
|
|
tags = [item.strip() for item in raw_tags.split(",") if item.strip()]
|
|
self._apply_selected_modify(
|
|
raw_command=raw,
|
|
success_message="Updated tags for selected entry",
|
|
failure_context="tags-set failed",
|
|
tags=tags,
|
|
)
|
|
return
|
|
|
|
if cmd == "tags-clear":
|
|
if args:
|
|
self._set_status("Usage: tags-clear")
|
|
return
|
|
self._apply_selected_modify(
|
|
raw_command=raw,
|
|
success_message="Cleared tags for selected entry",
|
|
failure_context="tags-clear failed",
|
|
tags=[],
|
|
)
|
|
return
|
|
|
|
if cmd == "field-add":
|
|
if len(args) < 2 or len(args) > 3:
|
|
self._set_status(
|
|
"Usage: field-add <label> <value> (optional: hidden)"
|
|
)
|
|
return
|
|
field_label = args[0].strip()
|
|
field_value = args[1]
|
|
if not field_label:
|
|
self._set_status("field-add label is required")
|
|
return
|
|
hidden = False
|
|
if len(args) == 3:
|
|
token = args[2].strip().lower()
|
|
hidden = token in {"hidden", "1", "true", "yes", "y"}
|
|
try:
|
|
entry = self._selected_entry_payload()
|
|
except Exception as exc:
|
|
self._set_status(str(exc))
|
|
return
|
|
current_fields = entry.get("custom_fields")
|
|
fields = (
|
|
[dict(item) for item in current_fields if isinstance(item, dict)]
|
|
if isinstance(current_fields, list)
|
|
else []
|
|
)
|
|
fields.append(
|
|
{"label": field_label, "value": field_value, "is_hidden": hidden}
|
|
)
|
|
self._apply_selected_modify(
|
|
raw_command=raw,
|
|
success_message=f"Added custom field '{field_label}'",
|
|
failure_context="field-add failed",
|
|
custom_fields=fields,
|
|
)
|
|
return
|
|
|
|
if cmd == "field-rm":
|
|
if len(args) != 1:
|
|
self._set_status("Usage: field-rm <label>")
|
|
return
|
|
field_label = args[0].strip()
|
|
if not field_label:
|
|
self._set_status("field-rm label is required")
|
|
return
|
|
try:
|
|
entry = self._selected_entry_payload()
|
|
except Exception as exc:
|
|
self._set_status(str(exc))
|
|
return
|
|
current_fields = entry.get("custom_fields")
|
|
fields = (
|
|
[dict(item) for item in current_fields if isinstance(item, dict)]
|
|
if isinstance(current_fields, list)
|
|
else []
|
|
)
|
|
normalized = field_label.lower()
|
|
next_fields = [
|
|
item
|
|
for item in fields
|
|
if str(item.get("label", "")).strip().lower() != normalized
|
|
]
|
|
self._apply_selected_modify(
|
|
raw_command=raw,
|
|
success_message=f"Removed custom field '{field_label}'",
|
|
failure_context="field-rm failed",
|
|
custom_fields=next_fields,
|
|
)
|
|
return
|
|
|
|
if cmd in {"set-field", "field-set"}:
|
|
if len(args) < 2:
|
|
self._set_status("Usage: set-field <name> <value>")
|
|
return
|
|
field_raw = args[0].strip().lower()
|
|
field = field_raw.replace("-", "_")
|
|
value_raw = " ".join(args[1:]).strip()
|
|
if not field:
|
|
self._set_status("set-field name is required")
|
|
return
|
|
if not value_raw:
|
|
self._set_status("set-field value is required")
|
|
return
|
|
alias = {
|
|
"name": "label",
|
|
"user": "username",
|
|
"pass_length": "length",
|
|
"filetype": "file_type",
|
|
}
|
|
field = alias.get(field, field)
|
|
try:
|
|
entry = self._selected_entry_payload()
|
|
except Exception as exc:
|
|
self._set_status(str(exc))
|
|
return
|
|
kind = str(entry.get("kind") or entry.get("type") or "").lower()
|
|
allowed_fields = {
|
|
"password": {"label", "notes", "username", "url", "length"},
|
|
"totp": {"label", "notes", "period", "digits"},
|
|
"key_value": {"label", "notes", "key", "value"},
|
|
"document": {"label", "notes", "file_type", "content"},
|
|
"ssh": {"label", "notes"},
|
|
"pgp": {"label", "notes"},
|
|
"nostr": {"label", "notes"},
|
|
"seed": {"label", "notes"},
|
|
"managed_account": {"label", "notes"},
|
|
}
|
|
allowed = allowed_fields.get(kind, {"label", "notes"})
|
|
if field not in allowed:
|
|
self._set_status(
|
|
f"set-field '{field}' not supported for kind '{kind or 'unknown'}'"
|
|
)
|
|
return
|
|
value: Any = value_raw
|
|
if field in {"length", "period", "digits"}:
|
|
try:
|
|
value = int(value_raw)
|
|
except ValueError:
|
|
self._set_status(f"set-field {field} must be an integer")
|
|
return
|
|
if field == "file_type":
|
|
value = str(value).strip().lstrip(".") or "txt"
|
|
self._apply_selected_modify(
|
|
raw_command=raw,
|
|
success_message=f"Updated {field} for selected entry",
|
|
failure_context="set-field failed",
|
|
**{field: value},
|
|
)
|
|
return
|
|
|
|
if cmd in {"clear-field", "field-clear"}:
|
|
if len(args) != 1:
|
|
self._set_status("Usage: clear-field <name>")
|
|
return
|
|
field_raw = args[0].strip().lower()
|
|
field = field_raw.replace("-", "_")
|
|
if not field:
|
|
self._set_status("clear-field name is required")
|
|
return
|
|
alias = {
|
|
"user": "username",
|
|
"filetype": "file_type",
|
|
}
|
|
field = alias.get(field, field)
|
|
try:
|
|
entry = self._selected_entry_payload()
|
|
except Exception as exc:
|
|
self._set_status(str(exc))
|
|
return
|
|
kind = str(entry.get("kind") or entry.get("type") or "").lower()
|
|
clearable = {
|
|
"password": {"username", "url", "notes"},
|
|
"totp": {"notes"},
|
|
"key_value": {"notes", "key", "value"},
|
|
"document": {"notes", "content"},
|
|
"ssh": {"notes"},
|
|
"pgp": {"notes"},
|
|
"nostr": {"notes"},
|
|
"seed": {"notes"},
|
|
"managed_account": {"notes"},
|
|
}
|
|
allowed = clearable.get(kind, {"notes"})
|
|
if field not in allowed:
|
|
self._set_status(
|
|
f"clear-field '{field}' not supported for kind '{kind or 'unknown'}'"
|
|
)
|
|
return
|
|
self._apply_selected_modify(
|
|
raw_command=raw,
|
|
success_message=f"Cleared {field} for selected entry",
|
|
failure_context="clear-field failed",
|
|
**{field: ""},
|
|
)
|
|
return
|
|
|
|
if cmd == "2fa-board":
|
|
if args:
|
|
self._set_status("Usage: 2fa-board")
|
|
return
|
|
self._set_totp_board_visible(True)
|
|
self._set_status("2FA board opened")
|
|
return
|
|
|
|
if cmd == "2fa-hide":
|
|
if args:
|
|
self._set_status("Usage: 2fa-hide")
|
|
return
|
|
self._set_totp_board_visible(False)
|
|
self._set_status("2FA board closed")
|
|
return
|
|
|
|
if cmd == "2fa-refresh":
|
|
if args:
|
|
self._set_status("Usage: 2fa-refresh")
|
|
return
|
|
self._refresh_totp_board(force_reload=True)
|
|
self._set_status("2FA board refreshed")
|
|
return
|
|
|
|
if cmd == "2fa-copy":
|
|
if len(args) != 1:
|
|
self._set_status("Usage: 2fa-copy <entry_id>")
|
|
return
|
|
if self._service is None:
|
|
self._set_status("Entry service unavailable")
|
|
return
|
|
try:
|
|
entry_id = int(args[0])
|
|
except ValueError:
|
|
self._set_status("2fa-copy entry_id must be an integer")
|
|
return
|
|
try:
|
|
code = self._service.get_totp_code(entry_id)
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"2fa-copy failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
if not self._copy_to_clipboard(code):
|
|
self._set_status("Clipboard copy unavailable")
|
|
return
|
|
delay = self._clipboard_clear_delay()
|
|
if self.totp_board_open:
|
|
self._refresh_totp_board(force_reload=False)
|
|
self._set_status(
|
|
f"Copied TOTP code for entry {entry_id} to clipboard ({delay}s clear)"
|
|
)
|
|
self._clear_failure()
|
|
return
|
|
|
|
if cmd == "profiles-list":
|
|
if args:
|
|
self._set_status("Usage: profiles-list")
|
|
return
|
|
if self._profile_service is None:
|
|
self._set_status("Profile service unavailable")
|
|
return
|
|
try:
|
|
rows = self._profile_service.list_profiles()
|
|
profiles = [str(item) for item in rows]
|
|
if profiles:
|
|
self._set_status(
|
|
f"Profiles ({len(profiles)}): " + ", ".join(profiles[:4])
|
|
)
|
|
else:
|
|
self._set_status("No profiles found")
|
|
self._clear_failure()
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"profiles-list failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "profile-switch":
|
|
if self._profile_service is None:
|
|
self._set_status("Profile service unavailable")
|
|
return
|
|
if len(args) < 1 or len(args) > 2:
|
|
self._set_status(
|
|
"Usage: profile-switch <fingerprint> (optional: password)"
|
|
)
|
|
return
|
|
fp = args[0].strip()
|
|
if not fp:
|
|
self._set_status("profile-switch fingerprint is required")
|
|
return
|
|
password = args[1] if len(args) == 2 else None
|
|
try:
|
|
from seedpass.core.api import ProfileSwitchRequest
|
|
|
|
self._profile_service.switch_profile(
|
|
ProfileSwitchRequest(fingerprint=fp, password=password)
|
|
)
|
|
self._clear_failure()
|
|
self._load_entries(self._last_query, reset_page=True)
|
|
self._set_status(f"Switched profile to {fp}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"profile-switch failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "profile-add":
|
|
if self._profile_service is None:
|
|
self._set_status("Profile service unavailable")
|
|
return
|
|
if args:
|
|
self._set_status("Usage: profile-add")
|
|
return
|
|
try:
|
|
fp = self._profile_service.add_profile()
|
|
self._clear_failure()
|
|
if fp:
|
|
self._set_status(f"Created profile {fp}")
|
|
else:
|
|
self._set_status("Created profile")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"profile-add failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "profile-remove":
|
|
if self._profile_service is None:
|
|
self._set_status("Profile service unavailable")
|
|
return
|
|
if len(args) != 1:
|
|
self._set_status("Usage: profile-remove <fingerprint>")
|
|
return
|
|
fp = args[0].strip()
|
|
if not fp:
|
|
self._set_status("profile-remove fingerprint is required")
|
|
return
|
|
try:
|
|
from seedpass.core.api import ProfileRemoveRequest
|
|
|
|
self._profile_service.remove_profile(
|
|
ProfileRemoveRequest(fingerprint=fp)
|
|
)
|
|
self._clear_failure()
|
|
self._set_status(f"Removed profile {fp}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"profile-remove failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "profile-rename":
|
|
if self._profile_service is None:
|
|
self._set_status("Profile service unavailable")
|
|
return
|
|
if len(args) < 2:
|
|
self._set_status("Usage: profile-rename <fingerprint> <name>")
|
|
return
|
|
fp = args[0].strip()
|
|
name = " ".join(args[1:]).strip()
|
|
if not fp:
|
|
self._set_status("profile-rename fingerprint is required")
|
|
return
|
|
if not name:
|
|
self._set_status("profile-rename name is required")
|
|
return
|
|
try:
|
|
self._profile_service.rename_profile(fp, name)
|
|
self._clear_failure()
|
|
self._set_status(f"Renamed profile {fp} to '{name}'")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"profile-rename failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "setting-secret":
|
|
if self._config_service is None:
|
|
self._set_status("Config service unavailable")
|
|
return
|
|
if len(args) < 1 or len(args) > 2:
|
|
self._set_status("Usage: setting-secret <on|off> [delay]")
|
|
return
|
|
try:
|
|
enabled = self._parse_toggle_token(args[0])
|
|
except ValueError as exc:
|
|
self._set_status(str(exc))
|
|
return
|
|
delay = self._clipboard_clear_delay()
|
|
if len(args) == 2:
|
|
try:
|
|
delay = int(args[1])
|
|
except ValueError:
|
|
self._set_status("setting-secret delay must be an integer")
|
|
return
|
|
setter = getattr(self._config_service, "set_secret_mode", None)
|
|
if not callable(setter):
|
|
self._set_status("Config service does not support secret mode")
|
|
return
|
|
try:
|
|
setter(enabled, int(delay))
|
|
self._clear_failure()
|
|
state = "on" if enabled else "off"
|
|
self._set_status(f"Secret mode {state} (delay {int(delay)}s)")
|
|
if self.totp_board_open:
|
|
self._refresh_totp_board(force_reload=False)
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"setting-secret failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "setting-offline":
|
|
if self._config_service is None:
|
|
self._set_status("Config service unavailable")
|
|
return
|
|
if len(args) != 1:
|
|
self._set_status("Usage: setting-offline <on|off>")
|
|
return
|
|
try:
|
|
enabled = self._parse_toggle_token(args[0])
|
|
except ValueError as exc:
|
|
self._set_status(str(exc))
|
|
return
|
|
setter = getattr(self._config_service, "set_offline_mode", None)
|
|
if not callable(setter):
|
|
self._set_status("Config service does not support offline mode")
|
|
return
|
|
try:
|
|
setter(enabled)
|
|
self._clear_failure()
|
|
self._set_status(f"Offline mode {'on' if enabled else 'off'}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"setting-offline failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "setting-quick-unlock":
|
|
if self._config_service is None:
|
|
self._set_status("Config service unavailable")
|
|
return
|
|
if len(args) != 1:
|
|
self._set_status("Usage: setting-quick-unlock <on|off>")
|
|
return
|
|
try:
|
|
enabled = self._parse_toggle_token(args[0])
|
|
except ValueError as exc:
|
|
self._set_status(str(exc))
|
|
return
|
|
setter = getattr(self._config_service, "set", None)
|
|
if not callable(setter):
|
|
self._set_status("Config service does not support quick unlock")
|
|
return
|
|
try:
|
|
setter("quick_unlock", "true" if enabled else "false")
|
|
self._clear_failure()
|
|
self._set_status(f"Quick unlock {'on' if enabled else 'off'}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"setting-quick-unlock failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "setting-timeout":
|
|
if self._config_service is None:
|
|
self._set_status("Config service unavailable")
|
|
return
|
|
if len(args) != 1:
|
|
self._set_status("Usage: setting-timeout <seconds>")
|
|
return
|
|
try:
|
|
value = float(args[0])
|
|
except ValueError:
|
|
self._set_status("setting-timeout requires numeric seconds")
|
|
return
|
|
setter = getattr(self._config_service, "set", None)
|
|
if not callable(setter):
|
|
self._set_status("Config service does not support timeout setting")
|
|
return
|
|
try:
|
|
setter("inactivity_timeout", str(value))
|
|
self._clear_failure()
|
|
self._set_status(f"Inactivity timeout set to {value}s")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"setting-timeout failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "setting-kdf-iterations":
|
|
if self._config_service is None:
|
|
self._set_status("Config service unavailable")
|
|
return
|
|
if len(args) != 1:
|
|
self._set_status("Usage: setting-kdf-iterations <n>")
|
|
return
|
|
try:
|
|
value = int(args[0])
|
|
except ValueError:
|
|
self._set_status("setting-kdf-iterations requires integer value")
|
|
return
|
|
setter = getattr(self._config_service, "set", None)
|
|
if not callable(setter):
|
|
self._set_status("Config service does not support KDF settings")
|
|
return
|
|
try:
|
|
setter("kdf_iterations", str(value))
|
|
self._clear_failure()
|
|
self._set_status(f"KDF iterations set to {value}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"setting-kdf-iterations failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "setting-kdf-mode":
|
|
if self._config_service is None:
|
|
self._set_status("Config service unavailable")
|
|
return
|
|
if len(args) != 1:
|
|
self._set_status("Usage: setting-kdf-mode <mode>")
|
|
return
|
|
mode = args[0].strip()
|
|
if not mode:
|
|
self._set_status("setting-kdf-mode mode is required")
|
|
return
|
|
setter = getattr(self._config_service, "set", None)
|
|
if not callable(setter):
|
|
self._set_status("Config service does not support KDF settings")
|
|
return
|
|
try:
|
|
setter("kdf_mode", mode)
|
|
self._clear_failure()
|
|
self._set_status(f"KDF mode set to {mode}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"setting-kdf-mode failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "relay-list":
|
|
if self._nostr_service is None:
|
|
self._set_status("Nostr service unavailable")
|
|
return
|
|
if args:
|
|
self._set_status("Usage: relay-list")
|
|
return
|
|
try:
|
|
relays = [str(item) for item in self._nostr_service.list_relays()]
|
|
self._clear_failure()
|
|
if relays:
|
|
self._set_status(
|
|
f"Relays ({len(relays)}): " + ", ".join(relays[:3])
|
|
)
|
|
else:
|
|
self._set_status("No relays configured")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"relay-list failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "relay-add":
|
|
if self._nostr_service is None:
|
|
self._set_status("Nostr service unavailable")
|
|
return
|
|
if len(args) != 1:
|
|
self._set_status("Usage: relay-add <url>")
|
|
return
|
|
url = args[0].strip()
|
|
if not url:
|
|
self._set_status("relay-add url is required")
|
|
return
|
|
try:
|
|
self._nostr_service.add_relay(url)
|
|
self._clear_failure()
|
|
self._set_status(f"Added relay {url}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"relay-add failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd in {"npub", "nostr-pubkey"}:
|
|
if self._nostr_service is None:
|
|
self._set_status("Nostr service unavailable")
|
|
return
|
|
if args:
|
|
self._set_status("Usage: npub")
|
|
return
|
|
getter = getattr(self._nostr_service, "get_pubkey", None)
|
|
if not callable(getter):
|
|
self._set_status("Nostr service does not support pubkey display")
|
|
return
|
|
try:
|
|
pubkey = str(getter()).strip()
|
|
if not pubkey:
|
|
self._set_status("No active Nostr pubkey available")
|
|
return
|
|
panel = f"Active Nostr pubkey\n\nnpub: {pubkey}"
|
|
try:
|
|
panel = (
|
|
f"{panel}\n\nQR (nostr:npub)\n\n"
|
|
f"{render_qr_ascii(f'nostr:{pubkey}')}"
|
|
)
|
|
except Exception:
|
|
pass
|
|
self._set_secret_panel(panel)
|
|
self._clear_failure()
|
|
self._set_status("Displayed active Nostr pubkey")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"npub failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd in {"relay-rm", "relay-remove"}:
|
|
if self._nostr_service is None:
|
|
self._set_status("Nostr service unavailable")
|
|
return
|
|
if len(args) != 1:
|
|
self._set_status("Usage: relay-rm <index>")
|
|
return
|
|
try:
|
|
idx = int(args[0])
|
|
except ValueError:
|
|
self._set_status("relay-rm index must be an integer")
|
|
return
|
|
try:
|
|
self._nostr_service.remove_relay(idx)
|
|
self._clear_failure()
|
|
self._set_status(f"Removed relay #{idx}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"relay-rm failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "relay-reset":
|
|
if self._nostr_service is None:
|
|
self._set_status("Nostr service unavailable")
|
|
return
|
|
if args:
|
|
self._set_status("Usage: relay-reset")
|
|
return
|
|
resetter = getattr(self._nostr_service, "reset_relays", None)
|
|
if not callable(resetter):
|
|
self._set_status("Nostr service does not support relay reset")
|
|
return
|
|
try:
|
|
relays = list(resetter())
|
|
self._clear_failure()
|
|
self._set_status(f"Relays reset ({len(relays)})")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"relay-reset failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "nostr-reset-sync-state":
|
|
if self._nostr_service is None:
|
|
self._set_status("Nostr service unavailable")
|
|
return
|
|
if args:
|
|
self._set_status("Usage: nostr-reset-sync-state")
|
|
return
|
|
reset_sync = getattr(self._nostr_service, "reset_sync_state", None)
|
|
if not callable(reset_sync):
|
|
self._set_status("Nostr service does not support sync-state reset")
|
|
return
|
|
try:
|
|
current_idx = int(reset_sync())
|
|
self._clear_failure()
|
|
self._set_status(
|
|
f"Nostr sync state reset (account index {current_idx})"
|
|
)
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"nostr-reset-sync-state failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "nostr-fresh-namespace":
|
|
if self._nostr_service is None:
|
|
self._set_status("Nostr service unavailable")
|
|
return
|
|
if args:
|
|
self._set_status("Usage: nostr-fresh-namespace")
|
|
return
|
|
fresh_ns = getattr(self._nostr_service, "start_fresh_namespace", None)
|
|
if not callable(fresh_ns):
|
|
self._set_status(
|
|
"Nostr service does not support fresh namespace rotation"
|
|
)
|
|
return
|
|
try:
|
|
next_idx = int(fresh_ns())
|
|
self._clear_failure()
|
|
self._set_status(
|
|
f"Started fresh Nostr namespace at account index {next_idx}"
|
|
)
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"nostr-fresh-namespace failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "sync-now":
|
|
if self._sync_service is None:
|
|
self._set_status("Sync service unavailable")
|
|
return
|
|
if args:
|
|
self._set_status("Usage: sync-now")
|
|
return
|
|
try:
|
|
result = self._sync_service.sync()
|
|
self._clear_failure()
|
|
if result is None:
|
|
self._set_status("Sync completed (no publish result)")
|
|
else:
|
|
manifest = getattr(result, "manifest_id", "unknown")
|
|
self._set_status(f"Sync completed manifest {manifest}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"sync-now failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "sync-bg":
|
|
if self._sync_service is None:
|
|
self._set_status("Sync service unavailable")
|
|
return
|
|
if args:
|
|
self._set_status("Usage: sync-bg")
|
|
return
|
|
try:
|
|
self._sync_service.start_background_vault_sync()
|
|
self._clear_failure()
|
|
self._set_status("Background sync started")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"sync-bg failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "checksum-verify":
|
|
if self._utility_service is None:
|
|
self._set_status("Utility service unavailable")
|
|
return
|
|
if args:
|
|
self._set_status("Usage: checksum-verify")
|
|
return
|
|
verifier = getattr(self._utility_service, "verify_checksum", None)
|
|
if not callable(verifier):
|
|
self._set_status("Utility service does not support checksum verify")
|
|
return
|
|
try:
|
|
verifier()
|
|
self._clear_failure()
|
|
self._set_status("Checksum verification complete")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"checksum-verify failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "checksum-update":
|
|
if self._utility_service is None:
|
|
self._set_status("Utility service unavailable")
|
|
return
|
|
if args:
|
|
self._set_status("Usage: checksum-update")
|
|
return
|
|
updater = getattr(self._utility_service, "update_checksum", None)
|
|
if not callable(updater):
|
|
self._set_status("Utility service does not support checksum update")
|
|
return
|
|
try:
|
|
updater()
|
|
self._clear_failure()
|
|
self._set_status("Checksum updated")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"checksum-update failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "db-export":
|
|
if self._vault_service is None:
|
|
self._set_status("Vault service unavailable")
|
|
return
|
|
if len(args) != 1:
|
|
self._set_status("Usage: db-export <path>")
|
|
return
|
|
try:
|
|
from seedpass.core.api import VaultExportRequest
|
|
|
|
path = Path(args[0]).expanduser()
|
|
self._vault_service.export_vault(VaultExportRequest(path=path))
|
|
self._clear_failure()
|
|
self._set_status(f"Database exported to {path}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"db-export failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "db-import":
|
|
if self._vault_service is None:
|
|
self._set_status("Vault service unavailable")
|
|
return
|
|
if len(args) != 1:
|
|
self._set_status("Usage: db-import <path>")
|
|
return
|
|
try:
|
|
from seedpass.core.api import VaultImportRequest
|
|
|
|
path = Path(args[0]).expanduser()
|
|
self._vault_service.import_vault(VaultImportRequest(path=path))
|
|
self._clear_failure()
|
|
self._load_entries(self._last_query, reset_page=True)
|
|
self._set_status(f"Database imported from {path}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"db-import failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "totp-export":
|
|
if self._service is None:
|
|
self._set_status("Entry service unavailable")
|
|
return
|
|
if len(args) != 1:
|
|
self._set_status("Usage: totp-export <path>")
|
|
return
|
|
path = Path(args[0]).expanduser()
|
|
try:
|
|
payload = self._service.export_totp_entries()
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
json.dumps(payload, indent=2, sort_keys=True),
|
|
encoding="utf-8",
|
|
)
|
|
self._clear_failure()
|
|
self._set_status(f"Exported TOTP entries to {path}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"totp-export failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "parent-seed-backup":
|
|
if self._vault_service is None:
|
|
self._set_status("Vault service unavailable")
|
|
return
|
|
if len(args) > 2:
|
|
self._set_status(
|
|
"Usage: parent-seed-backup (optional: path) (optional: password)"
|
|
)
|
|
return
|
|
path = Path(args[0]).expanduser() if len(args) >= 1 else None
|
|
password = args[1] if len(args) >= 2 else None
|
|
try:
|
|
from seedpass.core.api import BackupParentSeedRequest
|
|
|
|
self._vault_service.backup_parent_seed(
|
|
BackupParentSeedRequest(path=path, password=password)
|
|
)
|
|
self._clear_failure()
|
|
if path is not None:
|
|
self._set_status(f"Parent seed backup written to {path}")
|
|
else:
|
|
self._set_status("Parent seed backup/reveal completed")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"parent-seed-backup failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "doc-export":
|
|
if self._service is None or self._selected_entry_id is None:
|
|
self._set_status("No entry selected")
|
|
return
|
|
if len(args) > 1:
|
|
self._set_status("Usage: doc-export (optional: output_path)")
|
|
return
|
|
try:
|
|
entry = self._selected_entry_payload()
|
|
except Exception as exc:
|
|
self._set_status(str(exc))
|
|
return
|
|
kind = str(entry.get("kind") or entry.get("type") or "").strip().lower()
|
|
if kind != "document":
|
|
self._set_status("doc-export requires selected document entry")
|
|
return
|
|
output_path = args[0] if args else None
|
|
try:
|
|
dest = self._service.export_document_file(
|
|
self._selected_entry_id, output_path=output_path
|
|
)
|
|
self._clear_failure()
|
|
self._set_status(f"Exported document to {dest}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"doc-export failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd in ("archive", "restore"):
|
|
if self._service is None or self._selected_entry_id is None:
|
|
self._set_status("No entry selected")
|
|
return
|
|
try:
|
|
if cmd == "archive":
|
|
self._service.archive_entry(self._selected_entry_id)
|
|
action = "archived"
|
|
else:
|
|
self._service.restore_entry(self._selected_entry_id)
|
|
action = "restored"
|
|
current_id = self._selected_entry_id
|
|
self._load_entries(self._last_query, reset_page=False)
|
|
if current_id in self._entry_ids_in_view:
|
|
self._show_entry(current_id)
|
|
else:
|
|
self._selected_entry_id = int(current_id)
|
|
entry = self._service.retrieve_entry(int(current_id))
|
|
self._selected_entry = (
|
|
dict(entry) if isinstance(entry, dict) else {}
|
|
)
|
|
self._update_filters_panel()
|
|
self._clear_failure()
|
|
self._set_status(f"Entry {current_id} {action}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
f"{cmd} failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(cmd),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
return
|
|
|
|
if cmd == "managed-load":
|
|
if self._service is None:
|
|
self._set_status("Entry service unavailable")
|
|
return
|
|
if len(args) > 1:
|
|
self._set_status("Usage: managed-load (optional: entry_id)")
|
|
return
|
|
if len(args) == 1:
|
|
try:
|
|
entry_id = int(args[0])
|
|
except ValueError:
|
|
self._set_status("managed-load entry_id must be an integer")
|
|
return
|
|
else:
|
|
if self._selected_entry_id is None:
|
|
self._set_status(
|
|
"managed-load requires a selected managed account entry"
|
|
)
|
|
return
|
|
entry_id = int(self._selected_entry_id)
|
|
|
|
loader = getattr(self._service, "load_managed_account", None)
|
|
if not callable(loader):
|
|
self._set_status(
|
|
"Entry service does not support managed account session loading"
|
|
)
|
|
return
|
|
entry = self._service.retrieve_entry(entry_id)
|
|
kind = (
|
|
str(entry.get("kind") or entry.get("type") or "").strip().lower()
|
|
if isinstance(entry, dict)
|
|
else ""
|
|
)
|
|
if kind != "managed_account":
|
|
self._set_status("Selected entry is not a managed account")
|
|
return
|
|
try:
|
|
loader(entry_id)
|
|
self._load_entries(query="", reset_page=True)
|
|
self._clear_failure()
|
|
self._set_status(
|
|
f"Loaded managed account session from entry #{entry_id}"
|
|
)
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"managed-load failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "managed-exit":
|
|
if self._service is None:
|
|
self._set_status("Entry service unavailable")
|
|
return
|
|
if args:
|
|
self._set_status("Usage: managed-exit")
|
|
return
|
|
exiter = getattr(self._service, "exit_managed_account", None)
|
|
if not callable(exiter):
|
|
self._set_status(
|
|
"Entry service does not support managed account session exit"
|
|
)
|
|
return
|
|
try:
|
|
exiter()
|
|
self._load_entries(query="", reset_page=True)
|
|
self._clear_failure()
|
|
self._set_status("Exited managed account session")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"managed-exit failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "edit-doc":
|
|
self.action_edit_document()
|
|
return
|
|
|
|
if cmd == "save-doc":
|
|
self.action_save_document()
|
|
return
|
|
|
|
if cmd == "cancel-edit":
|
|
self.action_cancel_document_edit()
|
|
return
|
|
if cmd == "reveal":
|
|
if len(args) > 1 or (len(args) == 1 and args[0].lower() != "confirm"):
|
|
self._set_status("Usage: reveal [confirm]")
|
|
return
|
|
confirm = len(args) == 1 and args[0].lower() == "confirm"
|
|
self.action_reveal_selected(confirm=confirm)
|
|
return
|
|
if cmd == "qr":
|
|
mode = "default"
|
|
confirm = False
|
|
for token in args:
|
|
value = token.strip().lower()
|
|
if value in {"public", "private"}:
|
|
mode = value
|
|
continue
|
|
if value == "confirm":
|
|
confirm = True
|
|
continue
|
|
self._set_status("Usage: qr [public|private] [confirm]")
|
|
return
|
|
self.action_show_qr(mode=mode, confirm=confirm)
|
|
return
|
|
|
|
if cmd == "link-add":
|
|
if self._service is None or self._selected_entry_id is None:
|
|
self._set_status("No source entry selected")
|
|
return
|
|
if len(args) < 1:
|
|
self._set_status(
|
|
"Usage: link-add <target_id> [relation] [note text]"
|
|
)
|
|
return
|
|
try:
|
|
target = int(args[0])
|
|
except ValueError:
|
|
self._set_status("link-add target_id must be an integer")
|
|
return
|
|
relation = args[1] if len(args) >= 2 else "related_to"
|
|
note = " ".join(args[2:]) if len(args) >= 3 else ""
|
|
try:
|
|
self._service.add_link(
|
|
self._selected_entry_id,
|
|
target,
|
|
relation=relation,
|
|
note=note,
|
|
)
|
|
self._update_links_panel()
|
|
self._clear_failure()
|
|
self._set_status(
|
|
f"Link added: {self._selected_entry_id} {relation} {target}"
|
|
)
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"link-add failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "link-rm":
|
|
if self._service is None or self._selected_entry_id is None:
|
|
self._set_status("No source entry selected")
|
|
return
|
|
if len(args) < 1:
|
|
self._set_status("Usage: link-rm <target_id> [relation]")
|
|
return
|
|
try:
|
|
target = int(args[0])
|
|
except ValueError:
|
|
self._set_status("link-rm target_id must be an integer")
|
|
return
|
|
relation = args[1] if len(args) >= 2 else None
|
|
try:
|
|
self._service.remove_link(
|
|
self._selected_entry_id,
|
|
target,
|
|
relation=relation,
|
|
)
|
|
self._update_links_panel()
|
|
self._clear_failure()
|
|
self._set_status(
|
|
f"Link removed: {self._selected_entry_id} -> {target}"
|
|
)
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"link-rm failed",
|
|
exc,
|
|
retry=lambda: self._run_palette_command(raw),
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
return
|
|
|
|
if cmd == "refresh":
|
|
self.action_refresh()
|
|
return
|
|
if cmd == "page-next":
|
|
self.action_next_page()
|
|
return
|
|
if cmd == "page-prev":
|
|
self.action_prev_page()
|
|
return
|
|
if cmd == "page":
|
|
if len(args) != 1:
|
|
self._set_status("Usage: page <page_number>")
|
|
return
|
|
try:
|
|
page = int(args[0])
|
|
except ValueError:
|
|
self._set_status("page requires integer page number")
|
|
return
|
|
if page < 1:
|
|
self._set_status("page must be >= 1")
|
|
return
|
|
self._result_page = min(page - 1, self._total_pages() - 1)
|
|
self._render_current_page(preserve_selected=False)
|
|
self._set_status(
|
|
f"Moved to page {self._result_page + 1}/{self._total_pages()}"
|
|
)
|
|
return
|
|
if cmd == "link-filter":
|
|
if len(args) != 1:
|
|
self._set_status("Usage: link-filter <relation|all>")
|
|
return
|
|
relation = args[0].strip().lower() or "all"
|
|
self.link_relation_filter = relation
|
|
self._current_link_cursor = 0
|
|
self._update_links_panel()
|
|
self._set_status(f"Applied link relation filter: {relation}")
|
|
return
|
|
if cmd == "link-next":
|
|
self.action_next_link()
|
|
return
|
|
if cmd == "link-prev":
|
|
self.action_prev_link()
|
|
return
|
|
if cmd == "link-open":
|
|
self.action_open_link_target()
|
|
return
|
|
|
|
self._set_status(f"Unknown command: {cmd}")
|
|
|
|
def action_refresh(self) -> None:
|
|
search = self.query_one("#search", Input).value
|
|
self._update_filters_panel()
|
|
self._load_entries(query=search, reset_page=False)
|
|
if self.totp_board_open:
|
|
self._refresh_totp_board(force_reload=True)
|
|
|
|
def action_toggle_help(self) -> None:
|
|
if self.palette_open:
|
|
self._set_palette_visible(False)
|
|
self.help_open = not self.help_open
|
|
self._update_help_overlay()
|
|
self._set_status("Help opened" if self.help_open else "Help closed")
|
|
|
|
def action_focus_left(self) -> None:
|
|
self._focus_pane = "left"
|
|
self._apply_focus_style()
|
|
self._set_status("Focused left pane")
|
|
|
|
def action_focus_center(self) -> None:
|
|
self._focus_pane = "center"
|
|
self._apply_focus_style()
|
|
self.query_one("#search", Input).focus()
|
|
self._set_status("Focused center pane")
|
|
|
|
def action_focus_right(self) -> None:
|
|
self._focus_pane = "right"
|
|
self._apply_focus_style()
|
|
self._set_status("Focused right pane")
|
|
|
|
def action_focus_jump(self) -> None:
|
|
if self.editing_document:
|
|
self._set_status("Finish document edit before jumping")
|
|
return
|
|
if self.palette_open:
|
|
self._set_palette_visible(False)
|
|
self._focus_pane = "center"
|
|
self._apply_focus_style()
|
|
self.query_one("#quick-jump", Input).focus()
|
|
self._set_status("Focused jump-to-id input")
|
|
|
|
def action_retry_last_error(self) -> None:
|
|
retry = self._retry_action
|
|
if retry is None:
|
|
self._set_status("No retry action available")
|
|
return
|
|
try:
|
|
retry()
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"Retry failed",
|
|
exc,
|
|
retry=retry,
|
|
hint="Press 'x' to retry again.",
|
|
)
|
|
|
|
def action_focus_search(self) -> None:
|
|
if self.editing_document:
|
|
self._set_status("Finish document edit before searching")
|
|
return
|
|
if self.palette_open:
|
|
self._set_palette_visible(False)
|
|
self._focus_pane = "center"
|
|
self._apply_focus_style()
|
|
self.query_one("#search", Input).focus()
|
|
|
|
def action_cycle_filter(self) -> None:
|
|
order = [
|
|
"all",
|
|
"password",
|
|
"totp",
|
|
"document",
|
|
"key_value",
|
|
"ssh",
|
|
"pgp",
|
|
"nostr",
|
|
"seed",
|
|
"managed_account",
|
|
]
|
|
idx = order.index(self.filter_kind) if self.filter_kind in order else 0
|
|
self.filter_kind = order[(idx + 1) % len(order)]
|
|
self._load_entries(query=self._last_query, reset_page=True)
|
|
self._set_status(f"Applied filter: {self.filter_kind}")
|
|
|
|
def action_cycle_archive_scope(self) -> None:
|
|
if self.editing_document:
|
|
self._set_status("Finish document edit before changing archive scope")
|
|
return
|
|
order = ["active", "all", "archived"]
|
|
idx = order.index(self.archive_scope) if self.archive_scope in order else 0
|
|
self.archive_scope = order[(idx + 1) % len(order)]
|
|
self._load_entries(query=self._last_query, reset_page=True)
|
|
self._set_status(f"Applied archive filter: {self.archive_scope}")
|
|
|
|
def action_next_page(self) -> None:
|
|
if not self._all_results:
|
|
self._set_status("No entries loaded")
|
|
return
|
|
if self._result_page >= self._total_pages() - 1:
|
|
self._set_status("Already on last page")
|
|
return
|
|
self._result_page += 1
|
|
self._render_current_page(preserve_selected=False)
|
|
self._set_status(f"Page {self._result_page + 1}/{self._total_pages()}")
|
|
|
|
def action_prev_page(self) -> None:
|
|
if not self._all_results:
|
|
self._set_status("No entries loaded")
|
|
return
|
|
if self._result_page <= 0:
|
|
self._set_status("Already on first page")
|
|
return
|
|
self._result_page -= 1
|
|
self._render_current_page(preserve_selected=False)
|
|
self._set_status(f"Page {self._result_page + 1}/{self._total_pages()}")
|
|
|
|
def action_cycle_link_filter(self) -> None:
|
|
if self._service is None or self._selected_entry_id is None:
|
|
self._set_status("Select an entry before filtering links")
|
|
return
|
|
if self.editing_document:
|
|
self._set_status("Finish document edit before link operations")
|
|
return
|
|
relation_order = [
|
|
"all",
|
|
"related_to",
|
|
"depends_on",
|
|
"references",
|
|
"contains",
|
|
"derived_from",
|
|
]
|
|
idx = (
|
|
relation_order.index(self.link_relation_filter)
|
|
if self.link_relation_filter in relation_order
|
|
else 0
|
|
)
|
|
self.link_relation_filter = relation_order[(idx + 1) % len(relation_order)]
|
|
self._current_link_cursor = 0
|
|
self._update_links_panel()
|
|
self._set_status(f"Link relation filter: {self.link_relation_filter}")
|
|
|
|
def action_next_link(self) -> None:
|
|
if not self._current_links:
|
|
self._set_status("No links to select")
|
|
return
|
|
self._current_link_cursor = (self._current_link_cursor + 1) % len(
|
|
self._current_links
|
|
)
|
|
self._update_links_panel()
|
|
self._set_status(
|
|
f"Selected link {self._current_link_cursor + 1}/{len(self._current_links)}"
|
|
)
|
|
|
|
def action_prev_link(self) -> None:
|
|
if not self._current_links:
|
|
self._set_status("No links to select")
|
|
return
|
|
self._current_link_cursor = (self._current_link_cursor - 1) % len(
|
|
self._current_links
|
|
)
|
|
self._update_links_panel()
|
|
self._set_status(
|
|
f"Selected link {self._current_link_cursor + 1}/{len(self._current_links)}"
|
|
)
|
|
|
|
def action_open_link_target(self) -> None:
|
|
if not self._current_links:
|
|
self._set_status("No links to open")
|
|
return
|
|
link = self._current_links[self._current_link_cursor]
|
|
target = link.get("target")
|
|
try:
|
|
target_id = int(target)
|
|
except Exception:
|
|
self._set_status(f"Invalid link target: {target}")
|
|
return
|
|
self._show_entry(target_id)
|
|
self._set_status(f"Opened linked entry {target_id}")
|
|
|
|
def action_open_palette(self) -> None:
|
|
if self.editing_document:
|
|
self._set_status("Finish document edit before opening palette")
|
|
return
|
|
if self.help_open:
|
|
self.help_open = False
|
|
self._update_help_overlay()
|
|
self._set_palette_visible(not self.palette_open)
|
|
self._focus_pane = "center"
|
|
self._apply_focus_style()
|
|
if self.palette_open:
|
|
self._set_status("Palette opened")
|
|
else:
|
|
self._set_status("Palette closed")
|
|
|
|
def action_reveal_selected(self, confirm: bool = False) -> None:
|
|
self._show_sensitive_panel(include_qr=False, confirm=confirm)
|
|
|
|
def action_show_qr(self, mode: str = "default", confirm: bool = False) -> None:
|
|
self._show_sensitive_panel(include_qr=True, qr_mode=mode, confirm=confirm)
|
|
|
|
def action_toggle_totp_board(self) -> None:
|
|
if self.editing_document:
|
|
self._set_status("Finish document edit before opening 2FA board")
|
|
return
|
|
target = not self.totp_board_open
|
|
self._set_totp_board_visible(target)
|
|
self._focus_pane = "right"
|
|
self._apply_focus_style()
|
|
if target:
|
|
self._set_status("2FA board opened")
|
|
else:
|
|
self._set_status("2FA board closed")
|
|
|
|
def action_toggle_archive(self) -> None:
|
|
if self._service is None or self._selected_entry_id is None:
|
|
self._set_status("No entry selected")
|
|
return
|
|
if self.editing_document:
|
|
self._set_status("Finish document edit before archive/restore")
|
|
return
|
|
|
|
try:
|
|
archived = bool(
|
|
self._selected_entry and self._selected_entry.get("archived")
|
|
)
|
|
if archived:
|
|
self._service.restore_entry(self._selected_entry_id)
|
|
action = "restored"
|
|
else:
|
|
self._service.archive_entry(self._selected_entry_id)
|
|
action = "archived"
|
|
current_id = self._selected_entry_id
|
|
self._load_entries(self._last_query)
|
|
if current_id in self._entry_ids_in_view:
|
|
self._show_entry(current_id)
|
|
else:
|
|
self._selected_entry_id = int(current_id)
|
|
entry = self._service.retrieve_entry(int(current_id))
|
|
self._selected_entry = (
|
|
dict(entry) if isinstance(entry, dict) else {}
|
|
)
|
|
self._update_filters_panel()
|
|
self._clear_failure()
|
|
self._set_status(f"Entry {current_id} {action}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"Archive/restore failed",
|
|
exc,
|
|
retry=self.action_toggle_archive,
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
|
|
def action_edit_document(self) -> None:
|
|
if self._service is None:
|
|
self._set_status("Entry service unavailable")
|
|
return
|
|
if self._selected_entry_id is None or not self._is_selected_document():
|
|
self._set_status("Select a document entry to edit")
|
|
return
|
|
if self.editing_document:
|
|
self._set_status("Document editor already open")
|
|
return
|
|
|
|
entry = self._selected_entry or {}
|
|
self.query_one("#doc-edit-label", Input).value = str(entry.get("label", ""))
|
|
self.query_one("#doc-edit-file-type", Input).value = str(
|
|
entry.get("file_type", "txt")
|
|
)
|
|
tags = entry.get("tags")
|
|
if isinstance(tags, list):
|
|
tags_text = ", ".join(str(tag) for tag in tags)
|
|
else:
|
|
tags_text = ""
|
|
self.query_one("#doc-edit-tags", Input).value = tags_text
|
|
self._set_document_editor_text(str(entry.get("content", "")))
|
|
self._doc_snapshot = {
|
|
"label": str(entry.get("label", "")),
|
|
"file_type": str(entry.get("file_type", "txt")).lstrip("."),
|
|
"tags": tags_text,
|
|
"content": str(entry.get("content", "")),
|
|
}
|
|
self._doc_dirty = False
|
|
self._set_document_editor_visible(True)
|
|
self._refresh_doc_edit_help()
|
|
if TextArea is not None:
|
|
self.query_one("#doc-edit-content").focus()
|
|
else:
|
|
self.query_one("#doc-edit-content-single", Input).focus()
|
|
self._set_status(f"Editing document {self._selected_entry_id}")
|
|
|
|
def action_save_document(self) -> None:
|
|
if not self.editing_document:
|
|
self._set_status("Document editor not active")
|
|
return
|
|
if self._service is None or self._selected_entry_id is None:
|
|
self._set_status("No document selected")
|
|
return
|
|
|
|
label = self.query_one("#doc-edit-label", Input).value.strip()
|
|
file_type = (
|
|
self.query_one("#doc-edit-file-type", Input).value.strip().lstrip(".")
|
|
)
|
|
tags_raw = self.query_one("#doc-edit-tags", Input).value.strip()
|
|
tags = [part.strip() for part in tags_raw.split(",") if part.strip()]
|
|
content = self._get_document_editor_text()
|
|
current_form = {
|
|
"label": label,
|
|
"file_type": file_type,
|
|
"tags": tags_raw,
|
|
"content": content,
|
|
}
|
|
self._doc_dirty = current_form != self._doc_snapshot
|
|
self._refresh_doc_edit_help()
|
|
if not self._doc_dirty:
|
|
self._set_status("No document changes to save")
|
|
return
|
|
|
|
try:
|
|
self._service.modify_entry(
|
|
self._selected_entry_id,
|
|
label=label or None,
|
|
content=content,
|
|
file_type=file_type or None,
|
|
tags=tags,
|
|
)
|
|
current_id = self._selected_entry_id
|
|
self._set_document_editor_visible(False)
|
|
self._doc_dirty = False
|
|
self._load_entries(self._last_query, reset_page=False)
|
|
if current_id in self._entry_ids_in_view:
|
|
self._show_entry(current_id)
|
|
self._clear_failure()
|
|
self._set_status(f"Saved document {current_id}")
|
|
except Exception as exc:
|
|
self._record_failure(
|
|
"Failed to save document",
|
|
exc,
|
|
retry=self.action_save_document,
|
|
hint="Press 'x' to retry.",
|
|
)
|
|
|
|
def action_cancel_document_edit(self) -> None:
|
|
if self.palette_open:
|
|
self._set_palette_visible(False)
|
|
self._set_status("Palette closed")
|
|
return
|
|
if self.help_open:
|
|
self.help_open = False
|
|
self._update_help_overlay()
|
|
self._set_status("Help closed")
|
|
return
|
|
if self.totp_board_open:
|
|
self._set_totp_board_visible(False)
|
|
self._set_status("2FA board closed")
|
|
return
|
|
if not self.editing_document:
|
|
return
|
|
self._set_document_editor_visible(False)
|
|
self._doc_dirty = False
|
|
self._set_status("Canceled document edit")
|
|
|
|
def on_input_changed(self, event: Input.Changed) -> None:
|
|
if event.input.id in {
|
|
"doc-edit-label",
|
|
"doc-edit-file-type",
|
|
"doc-edit-tags",
|
|
"doc-edit-content-single",
|
|
}:
|
|
self._mark_doc_dirty(True)
|
|
|
|
def on_text_area_changed(self, _event: Any) -> None:
|
|
self._mark_doc_dirty(True)
|
|
|
|
def on_input_submitted(self, event: Input.Submitted) -> None:
|
|
if event.input.id == "search":
|
|
if self.editing_document:
|
|
self._set_status("Finish document edit before searching")
|
|
return
|
|
self._load_entries(query=event.value.strip(), reset_page=True)
|
|
return
|
|
if event.input.id == "quick-jump":
|
|
raw = event.value.strip()
|
|
if not raw:
|
|
self._set_status("Jump requires an entry id")
|
|
return
|
|
try:
|
|
entry_id = int(raw)
|
|
except ValueError:
|
|
self._set_status("Jump requires an integer entry id")
|
|
return
|
|
self._show_entry(entry_id)
|
|
return
|
|
if event.input.id == "command-palette":
|
|
command = event.value
|
|
self._set_palette_visible(False)
|
|
self._run_palette_command(command)
|
|
|
|
def on_list_view_selected(self, event: ListView.Selected) -> None:
|
|
if self.editing_document:
|
|
self._set_status("Finish document edit before selecting another entry")
|
|
return
|
|
self._focus_pane = "center"
|
|
self._apply_focus_style()
|
|
item = event.item
|
|
if isinstance(item, EntryListItem):
|
|
self._show_entry(item.entry_index)
|
|
|
|
def on_list_view_highlighted(self, event: ListView.Highlighted) -> None:
|
|
if self.editing_document:
|
|
return
|
|
item = event.item
|
|
if isinstance(item, EntryListItem):
|
|
self._show_entry(item.entry_index)
|
|
|
|
app = SeedPassTuiV2()
|
|
if callable(app_hook):
|
|
app_hook(app)
|
|
return True
|
|
app.run()
|
|
return True
|