mirror of
https://github.com/markqvist/NomadNet.git
synced 2026-10-05 21:17:22 +00:00
1155 lines
44 KiB
Python
1155 lines
44 KiB
Python
import os
|
|
import json
|
|
import shutil
|
|
import urwid
|
|
import RNS
|
|
import nomadnet
|
|
|
|
from .ReadlineEdit import ReadlineEdit
|
|
from .Browser import Browser
|
|
from nomadnet.vendor.additional_urwid_widgets import IndicativeListBox
|
|
|
|
PREVIEW_DELAY = 0.4
|
|
PANE_WEIGHTS = {"files": 0.22, "editor": 0.40, "preview": 0.38}
|
|
PANE_ORDER = ["files", "editor", "preview"]
|
|
MAX_HIGHLIGHT_CHARS = 50000
|
|
MAX_PREVIEW_CHARS = 100000
|
|
|
|
|
|
|
|
|
|
SNIPPETS = [
|
|
("Heading", ">Heading\n"),
|
|
("Sub-heading", ">>Sub heading\n"),
|
|
("Collapsible", "`+>Section\nbody\n<\n"),
|
|
("Bold", "`!bold`!"),
|
|
("Italic", "`*italic`*"),
|
|
("Underline", "`_underline`_"),
|
|
("Foreground col", "`F00ftext`f"),
|
|
("Background col", "`B400text`b"),
|
|
("Link", "`[label`:/page/index.mu]"),
|
|
("Anchor", "`:anchor-name"),
|
|
("Anchor link", "`[label`#anchor-name]"),
|
|
("Input field", "`<24|name`>"),
|
|
("Masked field", "`<!24|name`>"),
|
|
("Checkbox", "`<?|name|value`>"),
|
|
("Radio", "`<^|name|value`>"),
|
|
("Divider", "-\n"),
|
|
("Table", "`t\n| A | B |\n| 1 | 2 |\n`t\n"),
|
|
("Literal block", "`=\nliteral text\n`=\n"),
|
|
]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
EDITOR_STYLES = [
|
|
("editor_comment", "dark gray", "default", None, "#888", "default"),
|
|
("editor_heading", "light cyan,bold", "default", None, "#0cc", "default"),
|
|
("editor_tag", "light magenta", "default", None, "#c6c", "default"),
|
|
("editor_field", "yellow", "default", None, "#cc0", "default"),
|
|
("editor_link", "light green", "default", None, "#0c6", "default"),
|
|
("editor_dir", "light blue,bold", "default", None, "#39f", "default"),
|
|
("editor_exec", "dark gray", "default", None, "#666", "default"),
|
|
("editor_lineno", "dark gray", "default", None, "#555", "default"),
|
|
]
|
|
|
|
|
|
def is_executable(path):
|
|
try:
|
|
return (not RNS.vendor.platformutils.is_windows()) and os.path.isfile(path) and os.access(path, os.X_OK)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _scan_tag(text, i, n):
|
|
# text[i] == "`"; return (end_index, attr)
|
|
if i + 1 >= n:
|
|
return i + 1, "editor_tag"
|
|
c = text[i + 1]
|
|
if c == "<":
|
|
j = text.find(">", i + 2); return (n if j < 0 else j + 1), "editor_field"
|
|
if c == "[":
|
|
j = text.find("]", i + 2); return (n if j < 0 else j + 1), "editor_link"
|
|
if c == "{":
|
|
j = text.find("}", i + 2); return (n if j < 0 else j + 1), "editor_link"
|
|
if c in "FB":
|
|
if i + 2 < n and text[i + 2] == "T":
|
|
return min(i + 9, n), "editor_tag"
|
|
return min(i + 5, n), "editor_tag"
|
|
if c == ":":
|
|
j = i + 2
|
|
while j < n and (text[j].isalnum() or text[j] in "_-"):
|
|
j += 1
|
|
return j, "editor_tag"
|
|
return i + 2, "editor_tag"
|
|
|
|
|
|
def highlight_micron(text):
|
|
n = len(text)
|
|
attrs = [None] * n
|
|
i = 0
|
|
line_start = True
|
|
while i < n:
|
|
c = text[i]
|
|
if line_start and c == "#":
|
|
j = text.find("\n", i); j = n if j < 0 else j
|
|
for k in range(i, j): attrs[k] = "editor_comment"
|
|
i = j; continue
|
|
if line_start and c == ">":
|
|
j = i
|
|
while j < n and text[j] == ">": j += 1
|
|
for k in range(i, j): attrs[k] = "editor_heading"
|
|
i = j; line_start = False; continue
|
|
if line_start and c == "`" and i + 2 < n and text[i + 1] in "+-" and text[i + 2] == ">":
|
|
j = i + 2
|
|
while j < n and text[j] == ">": j += 1
|
|
for k in range(i, j): attrs[k] = "editor_heading"
|
|
i = j; line_start = False; continue
|
|
if c == "\n":
|
|
i += 1; line_start = True; continue
|
|
if c == "\\" and i + 1 < n:
|
|
i += 2; line_start = False; continue
|
|
if c == "`":
|
|
j, attr = _scan_tag(text, i, n)
|
|
for k in range(i, min(j, n)): attrs[k] = attr
|
|
i = j; line_start = False; continue
|
|
i += 1; line_start = False
|
|
|
|
out = []
|
|
run_attr = attrs[0] if n else None
|
|
run_len = 0
|
|
for a in attrs:
|
|
if a == run_attr:
|
|
run_len += 1
|
|
else:
|
|
|
|
out.append((run_attr, run_len)); run_attr = a; run_len = 1
|
|
if run_len:
|
|
out.append((run_attr, run_len))
|
|
return out
|
|
|
|
|
|
class MicronEdit(ReadlineEdit):
|
|
def __init__(self, *args, **kwargs):
|
|
self._hl_text = None
|
|
self._hl_attrib = []
|
|
super().__init__(*args, **kwargs)
|
|
|
|
def get_text(self):
|
|
text = self.caption + self.edit_text
|
|
if text != self._hl_text:
|
|
self._hl_text = text
|
|
self._hl_attrib = [] if len(text) > MAX_HIGHLIGHT_CHARS else highlight_micron(text)
|
|
return text, self._hl_attrib
|
|
|
|
|
|
class GutterEdit(MicronEdit):
|
|
# MicronEdit with a wrap-aware line-number gutter on the left
|
|
def _gutter_width(self, maxcol):
|
|
gw = max(3, len(str(self.edit_text.count("\n") + 1)) + 2)
|
|
if maxcol - gw < 1:
|
|
return 0
|
|
return gw
|
|
|
|
def rows(self, size, focus=False):
|
|
gw = self._gutter_width(size[0])
|
|
return super().rows((max(1, size[0] - gw),), focus)
|
|
|
|
def get_cursor_coords(self, size):
|
|
gw = self._gutter_width(size[0])
|
|
coords = super().get_cursor_coords((max(1, size[0] - gw),))
|
|
if coords is None:
|
|
return None
|
|
return coords[0] + gw, coords[1]
|
|
|
|
def move_cursor_to_coords(self, size, x, y):
|
|
gw = self._gutter_width(size[0])
|
|
if isinstance(x, int):
|
|
x = max(0, x - gw)
|
|
return super().move_cursor_to_coords((max(1, size[0] - gw),), x, y)
|
|
|
|
def render(self, size, focus=False):
|
|
maxcol = size[0]
|
|
gw = self._gutter_width(maxcol)
|
|
if gw == 0:
|
|
return super().render(size, focus)
|
|
tw = max(1, maxcol - gw)
|
|
|
|
cheap = len(self.edit_text) > MAX_HIGHLIGHT_CHARS
|
|
wrap = "any" if cheap else "space"
|
|
if self.wrap != wrap:
|
|
self.set_wrap_mode(wrap)
|
|
|
|
edit_canv = super().render((tw,), focus=False)
|
|
h = edit_canv.rows()
|
|
rows = []
|
|
for i, line in enumerate(self.edit_text.split("\n"), 1):
|
|
rows.append(str(i).rjust(gw - 1) + " ")
|
|
if cheap:
|
|
wrapped = max(1, (len(line) + tw - 1) // tw)
|
|
else:
|
|
wrapped = max(1, urwid.Text(line).rows((tw,)))
|
|
rows.extend([""] * (wrapped - 1))
|
|
rows = (rows + [""] * h)[:h]
|
|
|
|
|
|
|
|
|
|
gutter_canv = urwid.Text([("editor_lineno", "\n".join(r.ljust(gw) for r in rows))], wrap="clip").render((gw,))
|
|
joined = urwid.CanvasJoin([(gutter_canv, None, False, gw), (edit_canv, None, focus, tw)])
|
|
if focus:
|
|
coords = super().get_cursor_coords((tw,))
|
|
if coords is not None:
|
|
joined.cursor = (coords[0] + gw, coords[1])
|
|
return joined
|
|
|
|
def mouse_event(self, size, event, button, x, y, focus):
|
|
gw = self._gutter_width(size[0])
|
|
if x < gw:
|
|
return True
|
|
return super().mouse_event((max(1, size[0] - gw),), event, button, x - gw, y, focus)
|
|
|
|
|
|
class FileEntry(urwid.WidgetWrap):
|
|
signals = ["click"]
|
|
|
|
def __init__(self, label, path, depth, executable=False):
|
|
self.path = path
|
|
self.executable = executable
|
|
text = (" " * depth) + "· " + label + (" *" if executable else "")
|
|
style = "editor_exec" if executable else None
|
|
super().__init__(urwid.AttrMap(urwid.Text(text), style, focus_map="list_focus"))
|
|
|
|
def selectable(self):
|
|
return True
|
|
|
|
def keypress(self, size, key):
|
|
if self._command_map[key] == urwid.ACTIVATE:
|
|
self._emit("click")
|
|
return None
|
|
return key
|
|
|
|
def mouse_event(self, size, event, button, x, y, focus):
|
|
if button == 1 and urwid.util.is_mouse_press(event):
|
|
self._emit("click")
|
|
return True
|
|
return False
|
|
|
|
|
|
class FolderEntry(urwid.WidgetWrap):
|
|
signals = ["click"]
|
|
|
|
def __init__(self, label, path, depth, expanded, glyphs):
|
|
self.path = path
|
|
icon = glyphs.get("folder_open" if expanded else "folder", "[-]" if expanded else "[+]")
|
|
text = (" " * depth) + icon + " " + label + "/"
|
|
super().__init__(urwid.AttrMap(urwid.Text(text), "editor_dir", focus_map="list_focus"))
|
|
|
|
def selectable(self):
|
|
return True
|
|
|
|
def keypress(self, size, key):
|
|
if self._command_map[key] == urwid.ACTIVATE:
|
|
self._emit("click")
|
|
return None
|
|
return key
|
|
|
|
def mouse_event(self, size, event, button, x, y, focus):
|
|
if button == 1 and urwid.util.is_mouse_press(event):
|
|
self._emit("click")
|
|
return True
|
|
return False
|
|
|
|
|
|
class PathDialog(urwid.WidgetWrap):
|
|
def __init__(self, title, default, on_ok, on_cancel):
|
|
self.on_ok = on_ok
|
|
self.on_cancel = on_cancel
|
|
self.edit = ReadlineEdit(caption="", edit_text=default)
|
|
pile = urwid.Pile([
|
|
urwid.Text(title),
|
|
urwid.Divider(),
|
|
urwid.AttrMap(self.edit, "list_off_focus", "list_focus"),
|
|
urwid.Divider(),
|
|
urwid.Columns([
|
|
urwid.Button("OK", on_press=lambda b: self.on_ok(self.edit.edit_text)),
|
|
urwid.Button("Cancel", on_press=lambda b: self.on_cancel()),
|
|
], dividechars=2),
|
|
])
|
|
super().__init__(urwid.LineBox(urwid.Filler(pile, valign="top"), title=title))
|
|
|
|
def keypress(self, size, key):
|
|
if key == "esc":
|
|
self.on_cancel(); return None
|
|
if key == "enter" and self._w.original_widget.original_widget.focus_position == 2:
|
|
self.on_ok(self.edit.edit_text); return None
|
|
return super().keypress(size, key)
|
|
|
|
|
|
class InfoDialog(urwid.WidgetWrap):
|
|
def __init__(self, title, message, on_ok):
|
|
self.on_ok = on_ok
|
|
pile = urwid.Pile([
|
|
urwid.Text(message),
|
|
urwid.Divider(),
|
|
urwid.Button("OK", on_press=lambda b: on_ok()),
|
|
])
|
|
super().__init__(urwid.LineBox(urwid.Filler(pile, valign="top"), title=title))
|
|
|
|
def keypress(self, size, key):
|
|
if key in ("esc", "enter"):
|
|
self.on_ok(); return None
|
|
return super().keypress(size, key)
|
|
|
|
|
|
class ConfirmDialog(urwid.WidgetWrap):
|
|
def __init__(self, title, message, buttons, on_cancel=None):
|
|
self.on_cancel = on_cancel
|
|
cols = [urwid.Button(label, on_press=lambda b, c=cb: c()) for label, cb in buttons]
|
|
pile = urwid.Pile([urwid.Text(message), urwid.Divider(), urwid.Columns(cols, dividechars=2)])
|
|
super().__init__(urwid.LineBox(urwid.Filler(pile, valign="top"), title=title))
|
|
|
|
def keypress(self, size, key):
|
|
if key == "esc" and self.on_cancel:
|
|
self.on_cancel(); return None
|
|
return super().keypress(size, key)
|
|
|
|
|
|
class SnippetDialog(urwid.WidgetWrap):
|
|
def __init__(self, on_pick, on_cancel):
|
|
self.on_cancel = on_cancel
|
|
items = [urwid.AttrMap(urwid.Button(name, on_press=lambda b, t=text: on_pick(t)), None, focus_map="list_focus")
|
|
for name, text in SNIPPETS]
|
|
super().__init__(urwid.LineBox(urwid.ListBox(urwid.SimpleFocusListWalker(items)), title="Insert Micron snippet"))
|
|
|
|
def keypress(self, size, key):
|
|
if key == "esc":
|
|
self.on_cancel(); return None
|
|
return super().keypress(size, key)
|
|
|
|
|
|
class FindDialog(urwid.WidgetWrap):
|
|
def __init__(self, ed, on_close):
|
|
self.ed = ed
|
|
self.on_close = on_close
|
|
self.find_edit = ReadlineEdit(caption="Find: ")
|
|
self.repl_edit = ReadlineEdit(caption="Replace: ")
|
|
pile = urwid.Pile([
|
|
urwid.AttrMap(self.find_edit, "list_off_focus", "list_focus"),
|
|
urwid.AttrMap(self.repl_edit, "list_off_focus", "list_focus"),
|
|
urwid.Divider(),
|
|
urwid.Columns([
|
|
urwid.Button("Next", on_press=lambda b: ed.find_next(self.find_edit.edit_text)),
|
|
urwid.Button("Replace", on_press=lambda b: ed.replace_one(self.find_edit.edit_text, self.repl_edit.edit_text)),
|
|
urwid.Button("All", on_press=lambda b: ed.replace_all(self.find_edit.edit_text, self.repl_edit.edit_text)),
|
|
urwid.Button("Close", on_press=lambda b: on_close()),
|
|
], dividechars=1),
|
|
])
|
|
super().__init__(urwid.LineBox(urwid.Filler(pile, valign="top"), title="Find / Replace"))
|
|
|
|
def keypress(self, size, key):
|
|
if key == "esc":
|
|
self.on_close(); return None
|
|
return super().keypress(size, key)
|
|
|
|
|
|
class PermissionsDialog(urwid.WidgetWrap):
|
|
def __init__(self, title, note, content, on_ok, on_cancel):
|
|
self.on_ok = on_ok
|
|
self.on_cancel = on_cancel
|
|
self.edit = ReadlineEdit(caption="", edit_text=content, multiline=True)
|
|
pile = urwid.Pile([
|
|
urwid.Text(note),
|
|
urwid.Divider(),
|
|
urwid.BoxAdapter(urwid.AttrMap(urwid.ListBox(urwid.SimpleListWalker([self.edit])), "list_off_focus", "list_focus"), 7),
|
|
urwid.Divider(),
|
|
urwid.Columns([
|
|
urwid.Button("Save", on_press=lambda b: self.on_ok(self.edit.edit_text)),
|
|
urwid.Button("Cancel", on_press=lambda b: self.on_cancel()),
|
|
], dividechars=2),
|
|
])
|
|
super().__init__(urwid.LineBox(urwid.Filler(pile, valign="top"), title=title))
|
|
|
|
def keypress(self, size, key):
|
|
if key == "esc":
|
|
self.on_cancel(); return None
|
|
return super().keypress(size, key)
|
|
|
|
|
|
class EditorColumns(urwid.Columns):
|
|
def keypress(self, size, key):
|
|
ed = self.editor_display
|
|
if key == "ctrl s":
|
|
ed.save_or_saveas(); return None
|
|
if key == "ctrl n":
|
|
ed.new_file_dialog(); return None
|
|
if key == "ctrl f":
|
|
ed.find_dialog(); return None
|
|
if key == "ctrl g":
|
|
ed.snippet_dialog(); return None
|
|
if key == "ctrl p":
|
|
ed.permissions_dialog(); return None
|
|
if key == "f2" and ed.files_focused():
|
|
ed.rename_dialog(); return None
|
|
if key == "delete" and ed.files_focused():
|
|
ed.delete_dialog(); return None
|
|
if key == "ctrl t":
|
|
ed.toggle_pane("files"); return None
|
|
if key == "ctrl b":
|
|
ed.toggle_pane("editor"); return None
|
|
if key == "ctrl r":
|
|
ed.toggle_pane("preview"); return None
|
|
if key == "tab":
|
|
self.focus_position = (self.focus_position + 1) % len(self.contents); return None
|
|
if key == "shift tab":
|
|
self.focus_position = (self.focus_position - 1) % len(self.contents); return None
|
|
if key == "esc":
|
|
nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.frame.focus_position = "header"
|
|
return None
|
|
result = super().keypress(size, key)
|
|
if key == "up" and result == "up":
|
|
nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.frame.focus_position = "header"
|
|
return None
|
|
return result
|
|
|
|
|
|
class PageEditorShortcuts():
|
|
def __init__(self, app):
|
|
self.app = app
|
|
self.status = urwid.Text("", align=urwid.RIGHT)
|
|
base = "[C-n]New [C-s]Save [C-f]Find [C-g]Snippet [C-p]Perms [F2]Rename [Del]Delete [Tab]Pane [C-t/b/r]Toggle [Esc]Menu"
|
|
self.widget = urwid.AttrMap(urwid.Columns([urwid.Text(base), self.status], dividechars=2), "shortcutbar")
|
|
|
|
def set_status(self, msg):
|
|
self.status.set_text(msg)
|
|
|
|
|
|
class PageEditorDisplay():
|
|
def __init__(self, app):
|
|
self.app = app
|
|
self.current_path = None
|
|
self.editable = False
|
|
self.is_exec = False
|
|
self.dirty = False
|
|
self.preview_alarm = None
|
|
self._loading = False
|
|
self._find = None
|
|
self._last_file = None
|
|
self.expanded = set(p for p in (app.pagespath, app.filespath) if p)
|
|
self.filter_mu = False
|
|
self.pane_visible = {"files": True, "editor": True, "preview": True}
|
|
self.load_state()
|
|
self.register_styles()
|
|
|
|
self.preview_browser = Browser(self.app, "nomadnetwork", "node", delegate=None)
|
|
self.preview_browser.loopback = self.app.node.destination.hash
|
|
self.preview_browser.destination_hash = self.app.node.destination.hash
|
|
self.preview_browser.handle_link = self._preview_handle_link
|
|
self.preview_browser.marked_link = lambda *a, **k: None
|
|
self.preview_browser._content_cols = self._preview_cols
|
|
# don't fetch partials in the preview
|
|
self.preview_browser.detect_partials = lambda: None
|
|
|
|
self.editor = GutterEdit(caption="", edit_text="", multiline=True)
|
|
urwid.connect_signal(self.editor, "postchange", self.on_edit_change)
|
|
|
|
self.filter_checkbox = urwid.CheckBox("Show only .mu", state=self.filter_mu, on_state_change=self.on_filter_change)
|
|
self.tree_walker = urwid.SimpleFocusListWalker(self.build_entries())
|
|
self.tree_listbox = IndicativeListBox(self.tree_walker, initialization_is_selection_change=False, highlight_offFocus="list_off_focus")
|
|
self.files_pile = urwid.Pile([
|
|
(urwid.PACK, urwid.AttrMap(self.filter_checkbox, None, focus_map="list_focus")),
|
|
(urwid.WEIGHT, 1, self.tree_listbox),
|
|
])
|
|
try: self.files_pile.focus_position = 1
|
|
except Exception: pass
|
|
|
|
self.pane_widgets = {
|
|
"files": urwid.LineBox(self.files_pile, title="Files"),
|
|
"editor": urwid.LineBox(urwid.ListBox(urwid.SimpleListWalker([self.editor])), title="Editor"),
|
|
"preview": urwid.LineBox(urwid.SolidFill(" "), title="Preview"),
|
|
}
|
|
|
|
self.columns = EditorColumns([], dividechars=0)
|
|
self.columns.editor_display = self
|
|
self._visible = []
|
|
self.rebuild_columns()
|
|
|
|
self.shortcuts_display = PageEditorShortcuts(self.app)
|
|
self.body = urwid.WidgetPlaceholder(self.columns)
|
|
self.widget = self.body
|
|
|
|
def register_styles(self):
|
|
screen = self.app.ui.screen
|
|
for entry in EDITOR_STYLES:
|
|
try: screen.register_palette_entry(*entry)
|
|
except Exception: pass
|
|
|
|
def _preview_handle_link(self, target, link_data=None):
|
|
# in the preview follow in page anchors but ignore external navigation
|
|
# TODO
|
|
|
|
if isinstance(target, str) and target.startswith("#"):
|
|
name = target[1:]
|
|
self.app.ui.loop.set_alarm_in(0.0, lambda l, d: self.preview_browser._jump_to_anchor(name))
|
|
|
|
def _preview_cols(self):
|
|
try:
|
|
cols = self.app.ui.loop.screen.get_cols_rows()[0]
|
|
visible = self.visible_panes()
|
|
total = sum(PANE_WEIGHTS[n] for n in visible)
|
|
frac = PANE_WEIGHTS["preview"] / total if "preview" in visible else PANE_WEIGHTS["preview"]
|
|
return max(20, int(cols * frac) - 3)
|
|
except Exception:
|
|
return 80
|
|
|
|
def shortcuts(self):
|
|
return self.shortcuts_display
|
|
|
|
def start(self):
|
|
self.refresh_tree()
|
|
if self._last_file and os.path.isfile(self._last_file) and self.current_path is None:
|
|
self.load_file(self._last_file)
|
|
self._last_file = None
|
|
|
|
def files_focused(self):
|
|
try: return self.visible_panes()[self.columns.focus_position] == "files"
|
|
except Exception: return False
|
|
|
|
def _state_file(self):
|
|
d = getattr(self.app, "configdir", None)
|
|
return os.path.join(d, "editor_state.json") if d else None
|
|
|
|
def load_state(self):
|
|
p = self._state_file()
|
|
if not p or not os.path.isfile(p):
|
|
return
|
|
try:
|
|
with open(p) as f:
|
|
st = json.load(f)
|
|
self.filter_mu = bool(st.get("filter_mu", self.filter_mu))
|
|
if isinstance(st.get("expanded"), list):
|
|
self.expanded = set(st["expanded"])
|
|
pv = st.get("pane_visible")
|
|
if isinstance(pv, dict):
|
|
for k in self.pane_visible:
|
|
if k in pv: self.pane_visible[k] = bool(pv[k])
|
|
self._last_file = st.get("last_file")
|
|
except Exception:
|
|
pass
|
|
|
|
def save_state(self):
|
|
p = self._state_file()
|
|
if not p:
|
|
return
|
|
try:
|
|
self._atomic_write(p, json.dumps({"filter_mu": self.filter_mu, "expanded": sorted(self.expanded),
|
|
"pane_visible": self.pane_visible, "last_file": self.current_path}))
|
|
except Exception:
|
|
pass
|
|
|
|
def _selected_tree(self):
|
|
try:
|
|
sel = self.tree_listbox.get_selected_item()
|
|
return sel.original_widget if sel else None
|
|
except Exception:
|
|
return None
|
|
|
|
def _roots(self):
|
|
return [r for r in (self.app.pagespath, self.app.filespath) if r]
|
|
|
|
def _is_root(self, path):
|
|
try:
|
|
rp = os.path.realpath(path)
|
|
except Exception:
|
|
return False
|
|
return any(rp == os.path.realpath(r) for r in self._roots())
|
|
|
|
def _confine(self, path, base=None):
|
|
# Never save outside of the Nomad storage directory
|
|
path = (path or "").strip()
|
|
if not path:
|
|
return None
|
|
path = os.path.expanduser(path)
|
|
if not os.path.isabs(path):
|
|
base = base or self.app.pagespath or self.app.filespath
|
|
if not base:
|
|
return None
|
|
path = os.path.join(base, path)
|
|
norm = os.path.normpath(path)
|
|
for root in self._roots():
|
|
rp = os.path.normpath(root)
|
|
if norm == rp or norm.startswith(rp + os.sep):
|
|
return norm
|
|
return None
|
|
|
|
def _atomic_write(self, path, text):
|
|
d = os.path.dirname(path) or "."
|
|
tmp = os.path.join(d, "." + os.path.basename(path) + ".tmp")
|
|
try:
|
|
existing_mode = None
|
|
try:
|
|
existing_mode = os.stat(path).st_mode
|
|
except OSError:
|
|
existing_mode = None
|
|
with open(tmp, "w", encoding="utf-8") as f:
|
|
f.write(text)
|
|
f.flush()
|
|
try: os.fsync(f.fileno())
|
|
except Exception: pass
|
|
if existing_mode is not None:
|
|
try: os.chmod(tmp, existing_mode)
|
|
except Exception: pass
|
|
os.replace(tmp, path)
|
|
except Exception:
|
|
try: os.remove(tmp)
|
|
except Exception: pass
|
|
raise
|
|
|
|
def _confirm_overwrite(self, dest, proceed):
|
|
self.show_dialog(ConfirmDialog("Overwrite",
|
|
"%s already exists.\nOverwrite it?" % os.path.basename(dest),
|
|
[("Overwrite", proceed), ("Cancel", self.close_dialog)],
|
|
on_cancel=self.close_dialog))
|
|
|
|
def visible_panes(self):
|
|
return [n for n in PANE_ORDER if self.pane_visible[n]] or ["editor"]
|
|
|
|
def rebuild_columns(self):
|
|
focused = None
|
|
try: focused = self._visible[self.columns.focus_position]
|
|
except Exception: pass
|
|
visible = self.visible_panes()
|
|
self.columns.contents = [
|
|
(self.pane_widgets[n], self.columns.options(urwid.WEIGHT, PANE_WEIGHTS[n], box_widget=True))
|
|
for n in visible
|
|
]
|
|
self._visible = visible
|
|
# the rpeview starts non-selectable
|
|
target = focused if focused in visible else ("files" if "files" in visible else visible[0])
|
|
try: self.columns.focus_position = visible.index(target)
|
|
except Exception: pass
|
|
|
|
def toggle_pane(self, name):
|
|
visible = self.visible_panes()
|
|
if self.pane_visible[name] and len(visible) <= 1:
|
|
return
|
|
self.pane_visible[name] = not self.pane_visible[name]
|
|
self.rebuild_columns()
|
|
self.save_state()
|
|
|
|
def refresh_tree(self):
|
|
focus_path = None
|
|
try:
|
|
sel = self.tree_listbox.get_selected_item()
|
|
if sel is not None:
|
|
focus_path = getattr(sel.original_widget, "path", None)
|
|
except Exception:
|
|
pass
|
|
entries = self.build_entries()
|
|
# indicativelistbox wraps items in attrmap for highlighting
|
|
self.tree_walker[:] = [urwid.AttrMap(e, None) for e in entries]
|
|
if focus_path is not None:
|
|
for i, e in enumerate(entries):
|
|
if getattr(e, "path", None) == focus_path:
|
|
try: self.tree_listbox.select_item(i)
|
|
except Exception: pass
|
|
break
|
|
|
|
def build_entries(self):
|
|
entries = []
|
|
for label, root in (("pages", self.app.pagespath), ("files", self.app.filespath)):
|
|
if not root or not os.path.isdir(root):
|
|
continue
|
|
entries.extend(self.walk_dir(label, root, 0))
|
|
if not entries:
|
|
entries = [urwid.Text("no files")]
|
|
return entries
|
|
|
|
def walk_dir(self, name, path, depth):
|
|
out = []
|
|
expanded = path in self.expanded
|
|
folder = FolderEntry(name, path, depth, expanded, self.app.ui.glyphs)
|
|
urwid.connect_signal(folder, "click", self.on_folder_click)
|
|
out.append(folder)
|
|
if not expanded:
|
|
return out
|
|
try:
|
|
names = sorted(os.listdir(path), key=str.lower)
|
|
except Exception:
|
|
return out
|
|
dirs = [n for n in names if os.path.isdir(os.path.join(path, n)) and not n.startswith(".") and n != "__pycache__"]
|
|
files = [n for n in names if not os.path.isdir(os.path.join(path, n)) and not n.startswith(".")]
|
|
if self.filter_mu:
|
|
files = [n for n in files if n.endswith(".mu")]
|
|
dirs = [d for d in dirs if self._dir_has_mu(os.path.join(path, d))]
|
|
for d in dirs:
|
|
out.extend(self.walk_dir(d, os.path.join(path, d), depth + 1))
|
|
for fn in files:
|
|
full = os.path.join(path, fn)
|
|
entry = FileEntry(fn, full, depth + 1, executable=is_executable(full))
|
|
urwid.connect_signal(entry, "click", self.on_file_click)
|
|
out.append(entry)
|
|
return out
|
|
|
|
def _dir_has_mu(self, path):
|
|
for root, dirs, files in os.walk(path):
|
|
dirs[:] = [d for d in dirs if not d.startswith(".") and d != "__pycache__"]
|
|
if any(f.endswith(".mu") for f in files):
|
|
return True
|
|
return False
|
|
|
|
def on_filter_change(self, checkbox, state):
|
|
self.filter_mu = state
|
|
self.refresh_tree()
|
|
self.save_state()
|
|
|
|
def on_folder_click(self, folder):
|
|
if folder.path in self.expanded:
|
|
self.expanded.discard(folder.path)
|
|
else:
|
|
self.expanded.add(folder.path)
|
|
self.refresh_tree()
|
|
self.save_state()
|
|
|
|
def on_file_click(self, entry):
|
|
self.open_path(entry.path)
|
|
|
|
def open_path(self, path):
|
|
if self.dirty:
|
|
self.confirm_discard(lambda: self._do_open(path))
|
|
else:
|
|
self._do_open(path)
|
|
|
|
def _do_open(self, path):
|
|
self.load_file(path)
|
|
if self.pane_visible["editor"]:
|
|
try: self.columns.focus_position = self.visible_panes().index("editor")
|
|
except Exception: pass
|
|
|
|
def confirm_discard(self, proceed):
|
|
def save_then():
|
|
self.close_dialog()
|
|
if self.current_path and self.editable:
|
|
self.save_file(self.current_path)
|
|
proceed()
|
|
name = os.path.basename(self.current_path) if self.current_path else "this buffer"
|
|
if self.current_path and self.editable:
|
|
buttons = [("Save", save_then), ("Discard", lambda: (self.close_dialog(), proceed())), ("Cancel", self.close_dialog)]
|
|
else:
|
|
buttons = [("Discard", lambda: (self.close_dialog(), proceed())), ("Cancel", self.close_dialog)]
|
|
self.show_dialog(ConfirmDialog("Unsaved changes", "Discard unsaved changes to %s?" % name,
|
|
buttons, on_cancel=self.close_dialog))
|
|
|
|
def load_file(self, path):
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
self.editable = True
|
|
except UnicodeDecodeError:
|
|
content = ""
|
|
self.editable = False
|
|
except Exception as e:
|
|
self.set_status("open failed: %s" % e); return
|
|
|
|
self.current_path = path
|
|
self.is_exec = is_executable(path)
|
|
self.dirty = False
|
|
self._loading = True
|
|
self.editor.set_edit_text(content)
|
|
self.editor.set_edit_pos(0)
|
|
self._loading = False
|
|
|
|
name = os.path.basename(path)
|
|
self.update_title()
|
|
self.save_state()
|
|
if not self.editable:
|
|
self.set_status("binary file (not editable)")
|
|
elif self.is_exec:
|
|
self.set_preview_note(name + " is an executable page. the node runs it and serves its output.")
|
|
self.set_status("executable page")
|
|
self.show_info_dialog("Executable page", name + " is marked executable (chmod +x). NomadNet runs it and serves its output as the page, so the preview shows the source rather than the rendered output.")
|
|
else:
|
|
self.render_preview()
|
|
self.set_status("opened " + name)
|
|
|
|
def update_title(self, modified=False):
|
|
|
|
|
|
|
|
if self.current_path is None:
|
|
self.pane_widgets["editor"].set_title("Editor"); return
|
|
suffix = " (binary)" if not self.editable else (" (executable)" if self.is_exec else "")
|
|
if modified:
|
|
suffix += " *"
|
|
self.pane_widgets["editor"].set_title("Editor: " + os.path.basename(self.current_path) + suffix)
|
|
|
|
def set_preview_note(self, msg):
|
|
self.pane_widgets["preview"].original_widget = urwid.Filler(urwid.Text(msg), urwid.TOP)
|
|
|
|
def show_info_dialog(self, title, message):
|
|
self.show_dialog(InfoDialog(title, message, self.close_dialog))
|
|
|
|
def save_or_saveas(self):
|
|
if self.current_path is None:
|
|
self.save_as_dialog(); return
|
|
self.save_file(self.current_path)
|
|
|
|
def save_file(self, path):
|
|
if not self.editable and self.current_path is not None:
|
|
self.set_status("cannot save binary file"); return
|
|
try:
|
|
self._atomic_write(path, self.editor.edit_text)
|
|
self.current_path = path
|
|
self.editable = True
|
|
self.dirty = False
|
|
self.set_modified(False)
|
|
self.set_status("saved " + os.path.basename(path))
|
|
self.refresh_tree()
|
|
self.save_state()
|
|
except Exception as e:
|
|
self.set_status("save failed: %s" % e)
|
|
|
|
def new_file_dialog(self):
|
|
default = (self.app.pagespath or "") + os.sep
|
|
self.show_dialog(PathDialog("New (end with / for a folder)", default, self._do_new, self.close_dialog))
|
|
|
|
def _do_new(self, path):
|
|
self.close_dialog()
|
|
raw = (path or "").strip()
|
|
if not raw:
|
|
return
|
|
is_folder = raw.endswith("/") or raw.endswith(os.sep)
|
|
target = self._confine(raw)
|
|
if target is None:
|
|
self.set_status("path must be inside pages/ or files/"); return
|
|
try:
|
|
if is_folder:
|
|
os.makedirs(target, exist_ok=True)
|
|
self.expanded.add(target)
|
|
self.refresh_tree(); self.save_state()
|
|
self.set_status("created folder " + os.path.basename(target))
|
|
else:
|
|
os.makedirs(os.path.dirname(target), exist_ok=True)
|
|
if not os.path.exists(target):
|
|
self._atomic_write(target, "")
|
|
self.refresh_tree()
|
|
self.open_path(target)
|
|
except Exception as e:
|
|
self.set_status("new failed: %s" % e)
|
|
|
|
def rename_dialog(self):
|
|
w = self._selected_tree()
|
|
if not w or not getattr(w, "path", None):
|
|
self.set_status("select a file/folder to rename"); return
|
|
if self._is_root(w.path):
|
|
self.set_status("cannot rename the pages/files root"); return
|
|
self.show_dialog(PathDialog("Rename", w.path, lambda p, old=w.path: self._do_rename(old, p), self.close_dialog))
|
|
|
|
def _do_rename(self, old, new):
|
|
self.close_dialog()
|
|
target = self._confine(new, base=os.path.dirname(old))
|
|
if target is None:
|
|
self.set_status("path must be inside pages/ or files/"); return
|
|
if target == os.path.normpath(old):
|
|
return
|
|
|
|
def proceed():
|
|
self.close_dialog()
|
|
try:
|
|
os.makedirs(os.path.dirname(target), exist_ok=True)
|
|
os.rename(old, target)
|
|
if os.path.isfile(old + ".allowed"):
|
|
try: os.replace(old + ".allowed", target + ".allowed")
|
|
except Exception: pass
|
|
old_prefix = old + os.sep
|
|
if self.current_path == old:
|
|
self.current_path = target; self.update_title()
|
|
elif self.current_path and self.current_path.startswith(old_prefix):
|
|
self.current_path = target + self.current_path[len(old):]; self.update_title()
|
|
self.expanded = set(
|
|
target if p == old else (target + p[len(old):] if p.startswith(old_prefix) else p)
|
|
for p in self.expanded
|
|
)
|
|
self.refresh_tree(); self.save_state()
|
|
self.set_status("renamed to " + os.path.basename(target))
|
|
except Exception as e:
|
|
self.set_status("rename failed: %s" % e)
|
|
|
|
if os.path.exists(target) and os.path.realpath(target) != os.path.realpath(old):
|
|
self._confirm_overwrite(target, proceed)
|
|
else:
|
|
proceed()
|
|
|
|
def delete_dialog(self):
|
|
w = self._selected_tree()
|
|
if not w or not getattr(w, "path", None):
|
|
self.set_status("select a file/folder to delete"); return
|
|
path = w.path
|
|
if self._is_root(path):
|
|
self.set_status("cannot delete the pages/files root"); return
|
|
self.show_dialog(ConfirmDialog("Delete", "Delete %s?" % os.path.basename(path),
|
|
[("Delete", lambda: self._do_delete(path)), ("Cancel", self.close_dialog)], on_cancel=self.close_dialog))
|
|
|
|
def _do_delete(self, path):
|
|
self.close_dialog()
|
|
try:
|
|
prefix = path + os.sep
|
|
if os.path.isdir(path):
|
|
shutil.rmtree(path)
|
|
self.expanded = set(p for p in self.expanded if p != path and not p.startswith(prefix))
|
|
else:
|
|
os.remove(path)
|
|
if os.path.isfile(path + ".allowed"):
|
|
try: os.remove(path + ".allowed")
|
|
except Exception: pass
|
|
if self.current_path is not None and (self.current_path == path or self.current_path.startswith(prefix)):
|
|
self.current_path = None; self.editable = False; self.is_exec = False; self.dirty = False
|
|
self._loading = True; self.editor.set_edit_text(""); self._loading = False
|
|
self.update_title(); self.set_preview_note("")
|
|
self.refresh_tree(); self.save_state()
|
|
self.set_status("deleted " + os.path.basename(path))
|
|
except Exception as e:
|
|
self.set_status("delete failed: %s" % e)
|
|
|
|
def _read_allowed(self, path):
|
|
ap = path + ".allowed"
|
|
if not os.path.isfile(ap):
|
|
return ""
|
|
try:
|
|
with open(ap, encoding="utf-8") as f:
|
|
return "\n".join(ln.strip() for ln in f.read().splitlines() if ln.strip())
|
|
except Exception:
|
|
return ""
|
|
|
|
def _walk_pages(self, path):
|
|
for root, dirs, files in os.walk(path):
|
|
dirs[:] = [d for d in dirs if not d.startswith(".") and d != "__pycache__"]
|
|
for fn in files:
|
|
if fn.endswith(".allowed"):
|
|
continue
|
|
yield os.path.join(root, fn)
|
|
|
|
def _folder_permissions(self, path):
|
|
bodies = {}
|
|
total = 0
|
|
for page in self._walk_pages(path):
|
|
total += 1
|
|
body = self._read_allowed(page)
|
|
bodies[body] = bodies.get(body, 0) + 1
|
|
restricted = sum(count for body, count in bodies.items() if body)
|
|
distinct = [body for body in bodies if body]
|
|
common = distinct[0] if len(bodies) == 1 and distinct else None
|
|
return common, restricted, total
|
|
|
|
def permissions_dialog(self):
|
|
w = self._selected_tree()
|
|
if not w or not getattr(w, "path", None):
|
|
self.set_status("select a file/folder for permissions"); return
|
|
path = w.path
|
|
is_dir = os.path.isdir(path)
|
|
|
|
if is_dir:
|
|
common, restricted, total = self._folder_permissions(path)
|
|
content = common or ""
|
|
if not total:
|
|
state = "This folder contains no files."
|
|
elif not restricted:
|
|
state = "All %d files here are public." % total
|
|
elif common is not None:
|
|
state = "All %d files here share these entries." % total
|
|
else:
|
|
state = "Entries differ across files (%d of %d restricted)." % (restricted, total)
|
|
note = ("Allowed identity hashes per line.\n"
|
|
+ state + "\nSaving applies these entries to every file in this folder and its subfolders.")
|
|
else:
|
|
content = self._read_allowed(path)
|
|
note = "Allowed identity hashes per line. Empty = public.\nApplies to this page or file."
|
|
|
|
title = "Permissions: " + os.path.basename(path) + ("/" if is_dir else "")
|
|
self.show_dialog(PermissionsDialog(title, note, content,
|
|
lambda txt: self._save_permissions(path, is_dir, txt), self.close_dialog))
|
|
|
|
def _save_permissions(self, path, is_dir, text):
|
|
body = "\n".join(ln.strip() for ln in text.splitlines() if ln.strip())
|
|
|
|
if is_dir and not body:
|
|
restricted = self._folder_permissions(path)[1]
|
|
if restricted:
|
|
self.show_dialog(ConfirmDialog("Make public",
|
|
"This removes the access restrictions from %d page(s)\nunder %s, making them public.\n\nContinue?"
|
|
% (restricted, os.path.basename(path)),
|
|
[("Make public", lambda: self._apply_permissions(path, is_dir, body)), ("Cancel", self.close_dialog)],
|
|
on_cancel=self.close_dialog))
|
|
return
|
|
|
|
self._apply_permissions(path, is_dir, body)
|
|
|
|
def _apply_permissions(self, path, is_dir, body):
|
|
self.close_dialog()
|
|
try:
|
|
if is_dir:
|
|
count = 0
|
|
for page in self._walk_pages(path):
|
|
self._write_allowed(page, body)
|
|
count += 1
|
|
self.set_status(("permissions applied to %d files" if body else "permissions cleared on %d files") % count)
|
|
else:
|
|
self._write_allowed(path, body)
|
|
self.set_status("permissions saved" if body else "permissions cleared (public)")
|
|
self.refresh_tree()
|
|
except Exception as e:
|
|
self.set_status("permissions failed: %s" % e)
|
|
|
|
def _write_allowed(self, path, body):
|
|
ap = path + ".allowed"
|
|
if body:
|
|
self._atomic_write(ap, body + "\n")
|
|
elif os.path.exists(ap):
|
|
os.remove(ap)
|
|
|
|
def snippet_dialog(self):
|
|
self.show_dialog(SnippetDialog(self._insert_snippet, self.close_dialog))
|
|
|
|
def _insert_snippet(self, text):
|
|
self.close_dialog()
|
|
if self.pane_visible["editor"]:
|
|
try: self.columns.focus_position = self.visible_panes().index("editor")
|
|
except Exception: pass
|
|
pos = self.editor.edit_pos
|
|
t = self.editor.edit_text
|
|
self.editor.set_edit_text(t[:pos] + text + t[pos:])
|
|
self.editor.set_edit_pos(pos + len(text))
|
|
|
|
def find_dialog(self):
|
|
if self._find is None:
|
|
self._find = FindDialog(self, self.close_dialog)
|
|
self.show_dialog(self._find)
|
|
|
|
def find_next(self, term):
|
|
if not term:
|
|
return
|
|
t = self.editor.edit_text
|
|
idx = t.find(term, self.editor.edit_pos)
|
|
if idx < 0:
|
|
idx = t.find(term, 0)
|
|
if idx < 0:
|
|
self.set_status("'%s' not found" % term); return
|
|
self.editor.set_edit_pos(idx + len(term))
|
|
self.set_status("found at %d" % idx)
|
|
|
|
def replace_one(self, term, repl):
|
|
if not term:
|
|
return
|
|
t = self.editor.edit_text
|
|
end = self.editor.edit_pos
|
|
start = end - len(term)
|
|
if start >= 0 and t[start:end] == term:
|
|
self.editor.set_edit_text(t[:start] + repl + t[end:])
|
|
self.editor.set_edit_pos(start + len(repl))
|
|
self.set_status("replaced")
|
|
else:
|
|
self.find_next(term)
|
|
|
|
def replace_all(self, term, repl):
|
|
if not term:
|
|
return
|
|
cnt = self.editor.edit_text.count(term)
|
|
if cnt:
|
|
self.editor.set_edit_text(self.editor.edit_text.replace(term, repl))
|
|
self.set_status("replaced %d" % cnt)
|
|
else:
|
|
self.set_status("'%s' not found" % term)
|
|
|
|
def save_as_dialog(self):
|
|
default = self.current_path or ((self.app.pagespath or "") + os.sep)
|
|
self.show_dialog(PathDialog("Save as", default, self._do_saveas, self.close_dialog))
|
|
|
|
def _do_saveas(self, path):
|
|
self.close_dialog()
|
|
target = self._confine(path)
|
|
if target is None:
|
|
self.set_status("path must be inside pages/ or files/"); return
|
|
|
|
def proceed():
|
|
self.close_dialog()
|
|
try:
|
|
os.makedirs(os.path.dirname(target), exist_ok=True)
|
|
except Exception:
|
|
pass
|
|
self.editable = True
|
|
self.save_file(target)
|
|
|
|
already = self.current_path is None or os.path.realpath(target) != os.path.realpath(self.current_path)
|
|
if os.path.exists(target) and already:
|
|
self._confirm_overwrite(target, proceed)
|
|
else:
|
|
proceed()
|
|
|
|
def show_dialog(self, dialog):
|
|
self.dialog = dialog
|
|
overlay = urwid.Overlay(
|
|
dialog, self.columns,
|
|
align=urwid.CENTER, width=(urwid.RELATIVE, 60),
|
|
valign=urwid.MIDDLE, height=(urwid.RELATIVE, 40),
|
|
min_width=30, min_height=8,
|
|
)
|
|
self.body.original_widget = overlay
|
|
|
|
def close_dialog(self):
|
|
self.dialog = None
|
|
self.body.original_widget = self.columns
|
|
|
|
def on_edit_change(self, widget, old_text):
|
|
if self._loading or (self.current_path is not None and not self.editable):
|
|
return
|
|
self.dirty = True
|
|
self.set_modified(True)
|
|
if self.is_exec:
|
|
return
|
|
if self.preview_alarm is not None:
|
|
try: self.app.ui.loop.remove_alarm(self.preview_alarm)
|
|
except Exception: pass
|
|
self.preview_alarm = self.app.ui.loop.set_alarm_in(PREVIEW_DELAY, self._render_preview_alarm)
|
|
|
|
def _render_preview_alarm(self, loop, user_data):
|
|
self.preview_alarm = None
|
|
self.render_preview()
|
|
|
|
def render_preview(self):
|
|
if len(self.editor.edit_text) > MAX_PREVIEW_CHARS:
|
|
self.set_preview_note("File too large to preview (%d KB). Editing still works." % (len(self.editor.edit_text) // 1000))
|
|
return
|
|
pos = self._preview_scrollpos()
|
|
try:
|
|
body = self.preview_browser.render_markup_buffer(self.editor.edit_text)
|
|
except Exception as e:
|
|
body = urwid.Filler(urwid.Text("preview error: %s" % e), urwid.TOP)
|
|
self.pane_widgets["preview"].original_widget = body
|
|
self._set_preview_scrollpos(pos)
|
|
|
|
def _preview_scrollable(self):
|
|
try:
|
|
return self.preview_browser.browser_body.original_widget.original_widget
|
|
except Exception:
|
|
return None
|
|
|
|
def _preview_scrollpos(self):
|
|
s = self._preview_scrollable()
|
|
try: return s.get_scrollpos()
|
|
except Exception: return None
|
|
|
|
def _set_preview_scrollpos(self, pos):
|
|
if pos is None: return
|
|
s = self._preview_scrollable()
|
|
try: s.set_scrollpos(pos)
|
|
except Exception: pass
|
|
|
|
def set_modified(self, modified):
|
|
self.update_title(modified)
|
|
|
|
def set_status(self, msg):
|
|
self.shortcuts_display.set_status(msg)
|