dandri/NomadNet
1
0
Fork 0
mirror of https://github.com/markqvist/NomadNet.git synced 2026-05-13 18:43:09 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
master
NomadNet/nomadnet/ui/textui/Conversations.py
T
Mark Qvist 29a2c9f7c6 Added better name sanitization options
2026-05-06 18:44:17 +02:00

2022 lines
79 KiB
Python
Raw Permalink Blame History

import RNS
import os
import shutil
import time
import nomadnet
import LXMF
import urwid
from datetime import datetime, timedelta
from nomadnet.Directory import DirectoryEntry
from nomadnet.Conversation import ConversationMessage
from nomadnet.util import strip_modifiers
from nomadnet.util import sanitize_name
def relative_time(timestamp):
now = time.time()
delta = now - timestamp
if delta < 0:
return "just now"
elif delta < 60:
return "just now"
elif delta < 3600:
m = int(delta / 60)
return str(m)+"m ago"
elif delta < 86400:
h = int(delta / 3600)
return str(h)+"h ago"
elif delta < 172800:
return "yesterday"
elif delta < 604800:
d = int(delta / 86400)
return str(d)+"d ago"
elif delta < 2592000:
w = int(delta / 604800)
return str(w)+"w ago"
else:
return datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d")
def _format_size(size):
if size < 1024:
return str(size)+" B"
elif size < 1048576:
return str(round(size/1024, 1))+" KB"
else:
return str(round(size/1048576, 1))+" MB"
from nomadnet.vendor.additional_urwid_widgets import IndicativeListBox
class ConversationListDisplayShortcuts():
def __init__(self, app):
self.app = app
self.widget = urwid.AttrMap(urwid.Text("[C-e] Peer Info [C-x] Delete [C-r] Sync [C-n] New [C-u] Ingest URI [C-o] Sort [C-g] Fullscreen"), "shortcutbar")
class ConversationDisplayShortcuts():
def __init__(self, app):
self.app = app
self.widget = urwid.AttrMap(urwid.Text("[C-d] Send [C-p] Paper Msg [C-t] Title [C-a] Attach [C-s] Save [C-k] Clear [C-w] Close [C-u] Purge [C-x] Clear History [C-o] Sort"), "shortcutbar")
class ConversationsArea(urwid.LineBox):
def keypress(self, size, key):
if key == "ctrl e":
self.delegate.edit_selected_in_directory()
elif key == "ctrl x":
self.delegate.delete_selected_conversation()
elif key == "ctrl n":
self.delegate.new_conversation()
elif key == "ctrl u":
self.delegate.ingest_lxm_uri()
elif key == "ctrl r":
self.delegate.sync_conversations()
elif key == "ctrl g":
self.delegate.toggle_fullscreen()
elif key == "ctrl o":
self.delegate.toggle_list_sort()
elif key == "tab":
self.delegate.app.ui.main_display.frame.focus_position = "header"
elif key == "up" and (self.delegate.ilb.first_item_is_selected() or self.delegate.ilb.body_is_empty()):
self.delegate.app.ui.main_display.frame.focus_position = "header"
else:
return super(ConversationsArea, self).keypress(size, key)
class DialogLineBox(urwid.LineBox):
def keypress(self, size, key):
if key == "esc":
if hasattr(self.delegate, "update_conversation_list"):
self.delegate.update_conversation_list()
elif hasattr(self.delegate, "dialog_active"):
self.delegate.dialog_active = False
self.delegate.conversation_changed(None)
else:
return super(DialogLineBox, self).keypress(size, key)
class ConversationsDisplay():
list_width = 0.33
given_list_width = 52
cached_conversation_widgets = {}
SORT_RECENT = 0
SORT_NAME = 1
def __init__(self, app):
self.app = app
self.dialog_open = False
self.sync_dialog = None
self.currently_displayed_conversation = None
self.list_sort_mode = ConversationsDisplay.SORT_RECENT
def disp_list_shortcuts(sender, arg1, arg2):
self.shortcuts_display = self.list_shortcuts
self.app.ui.main_display.update_active_shortcuts()
self.update_listbox()
self.columns_widget = urwid.Columns(
[
# (urwid.WEIGHT, ConversationsDisplay.list_width, self.listbox),
# (urwid.WEIGHT, 1-ConversationsDisplay.list_width, self.make_conversation_widget(None))
(ConversationsDisplay.given_list_width, self.listbox),
(urwid.WEIGHT, 1, self.make_conversation_widget(None))
],
dividechars=0, focus_column=0, box_columns=[0]
)
self.list_shortcuts = ConversationListDisplayShortcuts(self.app)
self.editor_shortcuts = ConversationDisplayShortcuts(self.app)
self.shortcuts_display = self.list_shortcuts
self.widget = self.columns_widget
nomadnet.Conversation.created_callback = self.update_conversation_list
def focus_change_event(self):
if not self.dialog_open:
self.update_conversation_list()
def toggle_list_sort(self):
if self.list_sort_mode == ConversationsDisplay.SORT_RECENT:
self.list_sort_mode = ConversationsDisplay.SORT_NAME
else:
self.list_sort_mode = ConversationsDisplay.SORT_RECENT
self.update_conversation_list()
def update_listbox(self):
conversations = self.app.conversations()
if self.list_sort_mode == ConversationsDisplay.SORT_NAME:
conversations.sort(key=lambda e: (e[3].lower(), e[0]))
conversation_list_widgets = []
for conversation in conversations:
conversation_list_widgets.append(self.conversation_list_widget(conversation))
self.list_widgets = conversation_list_widgets
self.ilb = IndicativeListBox(
self.list_widgets,
on_selection_change=self.conversation_list_selection,
initialization_is_selection_change=False,
highlight_offFocus="list_off_focus"
)
self.listbox = ConversationsArea(urwid.Filler(self.ilb, height=urwid.RELATIVE_100), title="Conversations")
self.listbox.delegate = self
def delete_selected_conversation(self):
self.dialog_open = True
item = self.ilb.get_selected_item()
if item == None:
return
source_hash = item.source_hash
def dismiss_dialog(sender):
self.update_conversation_list()
self.dialog_open = False
def confirmed(sender):
self.dialog_open = False
self.delete_conversation(source_hash)
nomadnet.Conversation.delete_conversation(source_hash, self.app)
self.update_conversation_list()
dialog = DialogLineBox(
urwid.Pile([
urwid.Text(
"Delete conversation with\n"+self.app.directory.simplest_display_str(bytes.fromhex(source_hash))+"\n",
align=urwid.CENTER,
),
urwid.Columns([
(urwid.WEIGHT, 0.45, urwid.Button("Yes", on_press=confirmed)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, urwid.Button("No", on_press=dismiss_dialog)),
])
]), title="?"
)
dialog.delegate = self
bottom = self.listbox
overlay = urwid.Overlay(
dialog,
bottom,
align=urwid.CENTER,
width=urwid.RELATIVE_100,
valign=urwid.MIDDLE,
height=urwid.PACK,
left=2,
right=2,
)
# options = self.columns_widget.options(urwid.WEIGHT, ConversationsDisplay.list_width)
options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width)
self.columns_widget.contents[0] = (overlay, options)
def edit_selected_in_directory(self):
g = self.app.ui.glyphs
self.dialog_open = True
item = self.ilb.get_selected_item()
if item == None:
return
source_hash_text = item.source_hash
display_name = self.ilb.get_selected_item().display_name
if display_name == None:
display_name = ""
e_id = urwid.Edit(caption="Addr : ",edit_text=source_hash_text)
t_id = urwid.Text("Addr : "+source_hash_text)
e_name = urwid.Edit(caption="Name : ",edit_text=display_name)
selected_id_widget = t_id
untrusted_selected = False
unknown_selected = True
trusted_selected = False
direct_selected = True
propagated_selected = False
try:
if self.app.directory.find(bytes.fromhex(source_hash_text)):
trust_level = self.app.directory.trust_level(bytes.fromhex(source_hash_text))
if trust_level == DirectoryEntry.UNTRUSTED:
untrusted_selected = True
unknown_selected = False
trusted_selected = False
elif trust_level == DirectoryEntry.UNKNOWN:
untrusted_selected = False
unknown_selected = True
trusted_selected = False
elif trust_level == DirectoryEntry.TRUSTED:
untrusted_selected = False
unknown_selected = False
trusted_selected = True
if self.app.directory.preferred_delivery(bytes.fromhex(source_hash_text)) == DirectoryEntry.PROPAGATED:
direct_selected = False
propagated_selected = True
except Exception as e:
pass
trust_button_group = []
r_untrusted = urwid.RadioButton(trust_button_group, "Untrusted", state=untrusted_selected)
r_unknown = urwid.RadioButton(trust_button_group, "Unknown", state=unknown_selected)
r_trusted = urwid.RadioButton(trust_button_group, "Trusted", state=trusted_selected)
method_button_group = []
r_direct = urwid.RadioButton(method_button_group, "Deliver directly", state=direct_selected)
r_propagated = urwid.RadioButton(method_button_group, "Use propagation nodes", state=propagated_selected)
def dismiss_dialog(sender):
self.update_conversation_list()
self.dialog_open = False
def confirmed(sender):
try:
display_name = e_name.get_edit_text()
source_hash = bytes.fromhex(e_id.get_edit_text())
trust_level = DirectoryEntry.UNTRUSTED
if r_unknown.state == True:
trust_level = DirectoryEntry.UNKNOWN
elif r_trusted.state == True:
trust_level = DirectoryEntry.TRUSTED
delivery = DirectoryEntry.DIRECT
if r_propagated.state == True:
delivery = DirectoryEntry.PROPAGATED
entry = DirectoryEntry(source_hash, display_name, trust_level, preferred_delivery=delivery)
self.app.directory.remember(entry)
self.update_conversation_list()
self.dialog_open = False
self.app.ui.main_display.sub_displays.network_display.directory_change_callback()
except Exception as e:
RNS.log("Could not save directory entry. The contained exception was: "+str(e), RNS.LOG_VERBOSE)
if not dialog_pile.error_display:
dialog_pile.error_display = True
options = dialog_pile.options(height_type=urwid.PACK)
dialog_pile.contents.append((urwid.Text(""), options))
dialog_pile.contents.append((
urwid.Text(("error_text", "Could not save entry. Check your input."), align=urwid.CENTER),
options,)
)
source_is_known = self.app.directory.is_known(bytes.fromhex(source_hash_text))
if source_is_known:
known_section = urwid.Divider(g["divider1"])
else:
def query_action(sender, user_data):
self.close_conversation_by_hash(user_data)
nomadnet.Conversation.query_for_peer(user_data)
options = dialog_pile.options(height_type=urwid.PACK)
dialog_pile.contents = [
(urwid.Text("Query sent"), options),
(urwid.Button("OK", on_press=dismiss_dialog), options)
]
query_button = urwid.Button("Query network for keys", on_press=query_action, user_data=source_hash_text)
known_section = urwid.Pile([
urwid.Divider(g["divider1"]),
urwid.Text(g["info"]+"\n", align=urwid.CENTER),
urwid.Text(
"The identity of this peer is not known, and you cannot currently send messages to it. "
"You can query the network to obtain the identity.\n",
align=urwid.CENTER,
),
query_button,
urwid.Divider(g["divider1"]),
])
dialog_pile = urwid.Pile([
selected_id_widget,
e_name,
urwid.Divider(g["divider1"]),
r_untrusted,
r_unknown,
r_trusted,
urwid.Divider(g["divider1"]),
r_direct,
r_propagated,
known_section,
urwid.Columns([
(urwid.WEIGHT, 0.45, urwid.Button("Save", on_press=confirmed)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, urwid.Button("Back", on_press=dismiss_dialog)),
])
])
dialog_pile.error_display = False
dialog = DialogLineBox(dialog_pile, title="Peer Info")
dialog.delegate = self
bottom = self.listbox
overlay = urwid.Overlay(
dialog,
bottom,
align=urwid.CENTER,
width=urwid.RELATIVE_100,
valign=urwid.MIDDLE,
height=urwid.PACK,
left=2,
right=2,
)
# options = self.columns_widget.options(urwid.WEIGHT, ConversationsDisplay.list_width)
options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width)
self.columns_widget.contents[0] = (overlay, options)
def new_conversation(self):
self.dialog_open = True
source_hash = ""
display_name = ""
e_id = urwid.Edit(caption="Addr : ",edit_text=source_hash)
e_name = urwid.Edit(caption="Name : ",edit_text=display_name)
trust_button_group = []
r_untrusted = urwid.RadioButton(trust_button_group, "Untrusted")
r_unknown = urwid.RadioButton(trust_button_group, "Unknown", state=True)
r_trusted = urwid.RadioButton(trust_button_group, "Trusted")
def dismiss_dialog(sender):
self.update_conversation_list()
self.dialog_open = False
def confirmed(sender):
try:
existing_conversations = nomadnet.Conversation.conversation_list(self.app)
display_name = e_name.get_edit_text()
source_hash_text = e_id.get_edit_text().strip()
source_hash = bytes.fromhex(source_hash_text)
trust_level = DirectoryEntry.UNTRUSTED
if r_unknown.state == True:
trust_level = DirectoryEntry.UNKNOWN
elif r_trusted.state == True:
trust_level = DirectoryEntry.TRUSTED
if not source_hash in [c[0] for c in existing_conversations]:
entry = DirectoryEntry(source_hash, display_name, trust_level)
self.app.directory.remember(entry)
new_conversation = nomadnet.Conversation(source_hash_text, nomadnet.NomadNetworkApp.get_shared_instance(), initiator=True)
self.update_conversation_list()
self.display_conversation(source_hash_text)
self.dialog_open = False
except Exception as e:
RNS.log("Could not start conversation. The contained exception was: "+str(e), RNS.LOG_VERBOSE)
if not dialog_pile.error_display:
dialog_pile.error_display = True
options = dialog_pile.options(height_type=urwid.PACK)
dialog_pile.contents.append((urwid.Text(""), options))
dialog_pile.contents.append((
urwid.Text(
("error_text", "Could not start conversation. Check your input."),
align=urwid.CENTER,
),
options,
))
dialog_pile = urwid.Pile([
e_id,
e_name,
urwid.Text(""),
r_untrusted,
r_unknown,
r_trusted,
urwid.Text(""),
urwid.Columns([
(urwid.WEIGHT, 0.45, urwid.Button("Create", on_press=confirmed)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, urwid.Button("Back", on_press=dismiss_dialog)),
])
])
dialog_pile.error_display = False
dialog = DialogLineBox(dialog_pile, title="New Conversation")
dialog.delegate = self
bottom = self.listbox
overlay = urwid.Overlay(
dialog,
bottom,
align=urwid.CENTER,
width=urwid.RELATIVE_100,
valign=urwid.MIDDLE,
height=urwid.PACK,
left=2,
right=2,
)
# options = self.columns_widget.options(urwid.WEIGHT, ConversationsDisplay.list_width)
options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width)
self.columns_widget.contents[0] = (overlay, options)
def ingest_lxm_uri(self):
self.dialog_open = True
lxm_uri = ""
e_uri = urwid.Edit(caption="URI : ",edit_text=lxm_uri)
def dismiss_dialog(sender):
self.update_conversation_list()
self.dialog_open = False
def confirmed(sender):
try:
local_delivery_signal = "local_delivery_occurred"
duplicate_signal = "duplicate_lxm"
lxm_uri = e_uri.get_edit_text().strip()
ingest_result = self.app.message_router.ingest_lxm_uri(
lxm_uri,
signal_local_delivery=local_delivery_signal,
signal_duplicate=duplicate_signal
)
if ingest_result == False:
raise ValueError("The URI contained no decodable messages")
elif ingest_result == local_delivery_signal:
rdialog_pile = urwid.Pile([
urwid.Text("Message was decoded, decrypted successfully, and added to your conversation list."),
urwid.Text(""),
urwid.Columns([
(urwid.WEIGHT, 0.6, urwid.Text("")),
(urwid.WEIGHT, 0.4, urwid.Button("OK", on_press=dismiss_dialog)),
])
])
rdialog_pile.error_display = False
rdialog = DialogLineBox(rdialog_pile, title="Ingest message URI")
rdialog.delegate = self
bottom = self.listbox
roverlay = urwid.Overlay(
rdialog,
bottom,
align=urwid.CENTER,
width=urwid.RELATIVE_100,
valign=urwid.MIDDLE,
height=urwid.PACK,
left=2,
right=2,
)
options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width)
self.columns_widget.contents[0] = (roverlay, options)
elif ingest_result == duplicate_signal:
rdialog_pile = urwid.Pile([
urwid.Text("The decoded message has already been processed by the LXMF Router, and will not be ingested again."),
urwid.Text(""),
urwid.Columns([
(urwid.WEIGHT, 0.6, urwid.Text("")),
(urwid.WEIGHT, 0.4, urwid.Button("OK", on_press=dismiss_dialog)),
])
])
rdialog_pile.error_display = False
rdialog = DialogLineBox(rdialog_pile, title="Ingest message URI")
rdialog.delegate = self
bottom = self.listbox
roverlay = urwid.Overlay(
rdialog,
bottom,
align=urwid.CENTER,
width=urwid.RELATIVE_100,
valign=urwid.MIDDLE,
height=urwid.PACK,
left=2,
right=2,
)
options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width)
self.columns_widget.contents[0] = (roverlay, options)
else:
if self.app.enable_node:
propagation_text = "The decoded message was not addressed to this LXMF address, but has been added to the propagation node queues, and will be distributed on the propagation network."
else:
propagation_text = "The decoded message was not addressed to this LXMF address, and has been discarded."
rdialog_pile = urwid.Pile([
urwid.Text(propagation_text),
urwid.Text(""),
urwid.Columns([
(urwid.WEIGHT, 0.6, urwid.Text("")),
(urwid.WEIGHT, 0.4, urwid.Button("OK", on_press=dismiss_dialog)),
])
])
rdialog_pile.error_display = False
rdialog = DialogLineBox(rdialog_pile, title="Ingest message URI")
rdialog.delegate = self
bottom = self.listbox
roverlay = urwid.Overlay(
rdialog,
bottom,
align=urwid.CENTER,
width=urwid.RELATIVE_100,
valign=urwid.MIDDLE,
height=urwid.PACK,
left=2,
right=2,
)
options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width)
self.columns_widget.contents[0] = (roverlay, options)
except Exception as e:
RNS.log("Could not ingest LXM URI. The contained exception was: "+str(e), RNS.LOG_VERBOSE)
if not dialog_pile.error_display:
dialog_pile.error_display = True
options = dialog_pile.options(height_type=urwid.PACK)
dialog_pile.contents.append((urwid.Text(""), options))
dialog_pile.contents.append((urwid.Text(("error_text", "Could ingest LXM from URI data. Check your input."), align=urwid.CENTER), options))
dialog_pile = urwid.Pile([
e_uri,
urwid.Text(""),
urwid.Columns([
(urwid.WEIGHT, 0.45, urwid.Button("Ingest", on_press=confirmed)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, urwid.Button("Back", on_press=dismiss_dialog)),
])
])
dialog_pile.error_display = False
dialog = DialogLineBox(dialog_pile, title="Ingest message URI")
dialog.delegate = self
bottom = self.listbox
overlay = urwid.Overlay(
dialog,
bottom,
align=urwid.CENTER,
width=urwid.RELATIVE_100,
valign=urwid.MIDDLE,
height=urwid.PACK,
left=2,
right=2,
)
options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width)
self.columns_widget.contents[0] = (overlay, options)
def delete_conversation(self, source_hash):
if source_hash in ConversationsDisplay.cached_conversation_widgets:
conversation = ConversationsDisplay.cached_conversation_widgets[source_hash]
self.close_conversation(conversation)
def toggle_fullscreen(self):
if ConversationsDisplay.given_list_width != 0:
self.saved_list_width = ConversationsDisplay.given_list_width
ConversationsDisplay.given_list_width = 0
else:
ConversationsDisplay.given_list_width = self.saved_list_width
self.update_conversation_list()
def sync_conversations(self):
g = self.app.ui.glyphs
self.dialog_open = True
def dismiss_dialog(sender):
self.dialog_open = False
self.sync_dialog = None
self.update_conversation_list()
if self.app.message_router.propagation_transfer_state >= LXMF.LXMRouter.PR_COMPLETE:
self.app.cancel_lxmf_sync()
max_messages_group = []
r_mall = urwid.RadioButton(max_messages_group, "Download all", state=True)
r_mlim = urwid.RadioButton(max_messages_group, "Limit to", state=False)
ie_lim = urwid.IntEdit("", 5)
rbs = urwid.GridFlow([r_mlim, ie_lim], 12, 1, 0, align=urwid.LEFT)
def sync_now(sender):
limit = None
if r_mlim.get_state():
limit = ie_lim.value()
self.app.request_lxmf_sync(limit)
self.update_sync_dialog()
def cancel_sync(sender):
self.app.cancel_lxmf_sync()
self.update_sync_dialog()
cancel_button = urwid.Button("Close", on_press=dismiss_dialog)
sync_progress = SyncProgressBar("progress_empty" , "progress_full", current=self.app.get_sync_progress(), done=1.0, satt=None)
real_sync_button = urwid.Button("Sync Now", on_press=sync_now)
hidden_sync_button = urwid.Button("Cancel Sync", on_press=cancel_sync)
if self.app.get_sync_status() == "Idle" or self.app.message_router.propagation_transfer_state >= LXMF.LXMRouter.PR_COMPLETE:
sync_button = real_sync_button
else:
sync_button = hidden_sync_button
button_columns = urwid.Columns([
(urwid.WEIGHT, 0.45, sync_button),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, cancel_button),
])
real_sync_button.bc = button_columns
pn_ident = None
if self.app.get_default_propagation_node() != None:
pn_hash = self.app.get_default_propagation_node()
pn_ident = RNS.Identity.recall(pn_hash)
if pn_ident == None:
RNS.log("Propagation node identity is unknown, requesting from network...", RNS.LOG_DEBUG)
RNS.Transport.request_path(pn_hash)
if pn_ident != None:
node_hash = RNS.Destination.hash_from_name_and_identity("nomadnetwork.node", pn_ident)
pn_entry = self.app.directory.find(node_hash)
pn_display_str = " "
if pn_entry != None:
pn_display_str += " "+str(pn_entry.display_name)
else:
pn_display_str += " "+RNS.prettyhexrep(pn_hash)
dialog = DialogLineBox(
urwid.Pile([
urwid.Text(""+g["node"]+pn_display_str, align=urwid.CENTER),
urwid.Divider(g["divider1"]),
sync_progress,
urwid.Divider(g["divider1"]),
r_mall,
rbs,
urwid.Text(""),
button_columns
]), title="Message Sync"
)
else:
button_columns = urwid.Columns([
(urwid.WEIGHT, 0.45, urwid.Text("" )),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, cancel_button),
])
dialog = DialogLineBox(
urwid.Pile([
urwid.Text(""),
urwid.Text("No trusted nodes found, cannot sync!\n", align=urwid.CENTER),
urwid.Text(
"To synchronise messages from the network, "
"one or more nodes must be marked as trusted in the Known Nodes list, "
"or a node must manually be selected as the default propagation node. "
"Nomad Network will then automatically sync from the nearest trusted node, "
"or the manually selected one.",
align=urwid.LEFT,
),
urwid.Text(""),
button_columns
]), title="Message Sync"
)
dialog.delegate = self
dialog.sync_progress = sync_progress
dialog.cancel_button = cancel_button
dialog.real_sync_button = real_sync_button
dialog.hidden_sync_button = hidden_sync_button
dialog.bc = button_columns
self.sync_dialog = dialog
bottom = self.listbox
overlay = urwid.Overlay(
dialog,
bottom,
align=urwid.CENTER,
width=urwid.RELATIVE_100,
valign=urwid.MIDDLE,
height=urwid.PACK,
left=2,
right=2,
)
# options = self.columns_widget.options(urwid.WEIGHT, ConversationsDisplay.list_width)
options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width)
self.columns_widget.contents[0] = (overlay, options)
def update_sync_dialog(self, loop = None, sender = None):
if self.dialog_open and self.sync_dialog != None:
self.sync_dialog.sync_progress.set_completion(self.app.get_sync_progress())
if self.app.get_sync_status() == "Idle" or self.app.message_router.propagation_transfer_state >= LXMF.LXMRouter.PR_COMPLETE:
self.sync_dialog.bc.contents[0] = (self.sync_dialog.real_sync_button, self.sync_dialog.bc.options(urwid.WEIGHT, 0.45))
else:
self.sync_dialog.bc.contents[0] = (self.sync_dialog.hidden_sync_button, self.sync_dialog.bc.options(urwid.WEIGHT, 0.45))
self.app.ui.loop.set_alarm_in(0.2, self.update_sync_dialog)
def conversation_list_selection(self, arg1, arg2):
pass
def update_conversation_list(self):
selected_hash = None
selected_item = self.ilb.get_selected_item()
if selected_item is not None:
if hasattr(selected_item, "source_hash"):
selected_hash = selected_item.source_hash
self.update_listbox()
options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width)
if not (self.dialog_open and self.sync_dialog != None):
self.columns_widget.contents[0] = (self.listbox, options)
else:
bottom = self.listbox
overlay = urwid.Overlay(
self.sync_dialog,
bottom,
align=urwid.CENTER,
width=urwid.RELATIVE_100,
valign=urwid.MIDDLE,
height=urwid.PACK,
left=2,
right=2,
)
self.columns_widget.contents[0] = (overlay, options)
if selected_hash is not None:
for idx, widget in enumerate(self.list_widgets):
if widget.source_hash == selected_hash:
self.ilb.select_item(idx)
break
nomadnet.NomadNetworkApp.get_shared_instance().ui.loop.draw_screen()
if self.app.ui.main_display.sub_displays.active_display == self.app.ui.main_display.sub_displays.conversations_display:
if self.currently_displayed_conversation != None:
if self.app.conversation_is_unread(self.currently_displayed_conversation):
self.app.mark_conversation_read(self.currently_displayed_conversation)
try:
if os.path.isfile(self.app.conversationpath + "/" + self.currently_displayed_conversation + "/unread"):
os.unlink(self.app.conversationpath + "/" + self.currently_displayed_conversation + "/unread")
except Exception as e:
raise e
def display_conversation(self, sender=None, source_hash=None):
if self.currently_displayed_conversation != None:
if self.app.conversation_is_unread(self.currently_displayed_conversation):
self.app.mark_conversation_read(self.currently_displayed_conversation)
self.currently_displayed_conversation = source_hash
# options = self.widget.options(urwid.WEIGHT, 1-ConversationsDisplay.list_width)
options = self.widget.options(urwid.WEIGHT, 1)
self.widget.contents[1] = (self.make_conversation_widget(source_hash), options)
if source_hash == None:
self.widget.focus_position = 0
else:
if self.app.conversation_is_unread(source_hash):
self.app.mark_conversation_read(source_hash)
self.update_conversation_list()
self.widget.focus_position = 1
conversation_position = None
index = 0
for widget in self.list_widgets:
if widget.source_hash == source_hash:
conversation_position = index
index += 1
if conversation_position != None:
self.ilb.select_item(conversation_position)
def make_conversation_widget(self, source_hash):
if source_hash in ConversationsDisplay.cached_conversation_widgets:
conversation_widget = ConversationsDisplay.cached_conversation_widgets[source_hash]
if source_hash != None:
conversation_widget.update_message_widgets(replace=True)
conversation_widget.check_editor_allowed()
return conversation_widget
else:
widget = ConversationWidget(source_hash)
widget.delegate = self
ConversationsDisplay.cached_conversation_widgets[source_hash] = widget
widget.check_editor_allowed()
return widget
def close_conversation_by_hash(self, conversation_hash):
if conversation_hash in ConversationsDisplay.cached_conversation_widgets:
ConversationsDisplay.cached_conversation_widgets.pop(conversation_hash)
if self.currently_displayed_conversation == conversation_hash:
self.display_conversation(sender=None, source_hash=None)
def close_conversation(self, conversation):
if conversation.source_hash in ConversationsDisplay.cached_conversation_widgets:
ConversationsDisplay.cached_conversation_widgets.pop(conversation.source_hash)
if self.currently_displayed_conversation == conversation.source_hash:
self.display_conversation(sender=None, source_hash=None)
def conversation_list_widget(self, conversation):
trust_level = conversation[2]
display_name = conversation[1]
source_hash = conversation[0]
unread = conversation[4]
last_activity = conversation[5]
g = self.app.ui.glyphs
if trust_level == DirectoryEntry.UNTRUSTED:
symbol = g["cross"]
style = "list_untrusted"
focus_style = "list_focus_untrusted"
elif trust_level == DirectoryEntry.UNKNOWN:
symbol = "?"
style = "list_unknown"
focus_style = "list_focus"
elif trust_level == DirectoryEntry.TRUSTED:
symbol = g["check"]
style = "list_trusted"
focus_style = "list_focus_trusted"
elif trust_level == DirectoryEntry.WARNING:
symbol = g["warning"]
style = "list_warning"
focus_style = "list_focus"
else:
symbol = g["warning"]
style = "list_untrusted"
focus_style = "list_focus_untrusted"
display_text = symbol
if display_name != None and display_name != "":
display_text += " "+display_name
if trust_level != DirectoryEntry.TRUSTED:
display_text += " <"+source_hash+">"
if trust_level != DirectoryEntry.UNTRUSTED:
if unread:
if source_hash != self.currently_displayed_conversation:
if unread > 1:
display_text += " "+g["unread"]+" ("+str(unread)+")"
else:
display_text += " "+g["unread"]
if last_activity > 0:
display_text += "\n "+relative_time(last_activity)
widget = ListEntry(display_text)
urwid.connect_signal(widget, "click", self.display_conversation, conversation[0])
display_widget = urwid.AttrMap(widget, style, focus_style)
display_widget.source_hash = source_hash
display_widget.display_name = display_name
return display_widget
def shortcuts(self):
focus_path = self.widget.get_focus_path()
if focus_path[0] == 0:
return self.list_shortcuts
elif focus_path[0] == 1:
return self.editor_shortcuts
else:
return self.list_shortcuts
class ListEntry(urwid.Text):
_selectable = True
signals = ["click"]
def keypress(self, size, key):
"""
Send 'click' signal on 'activate' command.
"""
if self._command_map[key] != urwid.ACTIVATE:
return key
self._emit('click')
def mouse_event(self, size, event, button, x, y, focus):
"""
Send 'click' signal on button 1 press.
"""
if button != 1 or not urwid.util.is_mouse_press(event):
return False
self._emit('click')
return True
class MessageEdit(urwid.Edit):
def keypress(self, size, key):
if key == "ctrl d":
self.delegate.send_message()
elif key == "ctrl p":
self.delegate.paper_message()
elif key == "ctrl a":
self.delegate.attach_file()
elif key == "ctrl s":
self.delegate.save_focused_attachments()
elif key == "ctrl k":
self.delegate.clear_editor()
elif key == "up":
y = self.get_cursor_coords(size)[1]
if y == 0:
if self.delegate.full_editor_active and self.name == "title_editor":
self.delegate.frame.focus_position = "body"
elif not self.delegate.full_editor_active and self.name == "content_editor":
self.delegate.frame.focus_position = "body"
else:
return super(MessageEdit, self).keypress(size, key)
else:
return super(MessageEdit, self).keypress(size, key)
else:
return super(MessageEdit, self).keypress(size, key)
class ConversationFrame(urwid.Frame):
def keypress(self, size, key):
if self.focus_position == "body":
if getattr(self.delegate, "dialog_active", False) or getattr(self.delegate, "dialog_open", False):
return super(ConversationFrame, self).keypress(size, key)
elif key == "up" and self.delegate.messagelist.top_is_visible:
nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.frame.focus_position = "header"
elif key == "down" and self.delegate.messagelist.bottom_is_visible:
self.focus_position = "footer"
else:
return super(ConversationFrame, self).keypress(size, key)
elif key == "ctrl k":
self.delegate.clear_editor()
else:
return super(ConversationFrame, self).keypress(size, key)
class ConversationWidget(urwid.WidgetWrap):
def __init__(self, source_hash):
self.app = nomadnet.NomadNetworkApp.get_shared_instance()
g = self.app.ui.glyphs
if source_hash == None:
self.frame = None
display_widget = urwid.LineBox(urwid.Filler(urwid.Text("\n No conversation selected"), "top"))
super().__init__(display_widget)
else:
if source_hash in ConversationsDisplay.cached_conversation_widgets:
return ConversationsDisplay.cached_conversation_widgets[source_hash]
else:
self.source_hash = source_hash
self.conversation = nomadnet.Conversation(source_hash, nomadnet.NomadNetworkApp.get_shared_instance())
self.message_widgets = []
self.sort_by_timestamp = False
self.updating_message_widgets = False
self.pending_attachments = []
self.dialog_active = False
self.update_message_widgets()
self.conversation.register_changed_callback(self.conversation_changed)
#title_editor = MessageEdit(caption="\u270E", edit_text="", multiline=False)
title_editor = MessageEdit(caption="", edit_text="", multiline=False)
title_editor.delegate = self
title_editor.name = "title_editor"
#msg_editor = MessageEdit(caption="\u270E", edit_text="", multiline=True)
msg_editor = MessageEdit(caption="", edit_text="", multiline=True)
msg_editor.delegate = self
msg_editor.name = "content_editor"
self.peer_info_widget = urwid.AttrMap(urwid.Text(""), "msg_header_sent")
self._update_peer_info()
header_widgets = [self.peer_info_widget]
if self.conversation.trust_level == DirectoryEntry.UNTRUSTED:
header_widgets.append(urwid.AttrMap(
urwid.Padding(
urwid.Text(g["warning"]+" Warning: Conversation with untrusted peer "+g["warning"], align=urwid.CENTER)),
"msg_warning_untrusted",
))
header = urwid.Pile(header_widgets)
self.minimal_editor = urwid.AttrMap(msg_editor, "msg_editor")
self.minimal_editor.name = "minimal_editor"
title_columns = urwid.Columns([
(8, urwid.Text("Title")),
urwid.AttrMap(title_editor, "msg_editor"),
])
content_columns = urwid.Columns([
(8, urwid.Text("Content")),
urwid.AttrMap(msg_editor, "msg_editor")
])
self.full_editor = urwid.Pile([
title_columns,
content_columns
])
self.full_editor.name = "full_editor"
self.content_editor = msg_editor
self.title_editor = title_editor
self.full_editor_active = False
self.frame = ConversationFrame(
self.messagelist,
header=header,
footer=self.minimal_editor,
focus_part="footer"
)
self.frame.delegate = self
self.display_widget = urwid.LineBox(
self.frame
)
super().__init__(self.display_widget)
def _update_peer_info(self):
def san(name):
if self.app.config["textui"]["sanitize_names"]: return sanitize_name(name)
else: return strip_modifiers(name)
g = self.app.ui.glyphs
source_hash_bytes = bytes.fromhex(self.source_hash)
display_name = self.app.directory.display_name(source_hash_bytes)
app_data = None
if display_name is None or self.app.message_router.get_outbound_stamp_cost(source_hash_bytes) is None:
app_data = RNS.Identity.recall_app_data(source_hash_bytes)
if display_name is None:
if app_data:
display_name = san(LXMF.display_name_from_app_data(app_data))
if display_name is None:
display_name = RNS.prettyhexrep(source_hash_bytes)
stamp_cost = self.app.message_router.get_outbound_stamp_cost(source_hash_bytes)
if stamp_cost is None and app_data:
stamp_cost = LXMF.stamp_cost_from_app_data(app_data)
hops = RNS.Transport.hops_to(source_hash_bytes)
if hops >= RNS.Transport.PATHFINDER_M:
hops_str = "unknown"
else:
hops_str = str(hops)+" hop" + ("s" if hops != 1 else "")
right_parts = []
if stamp_cost is not None:
right_parts.append("Stamp: "+str(stamp_cost))
right_parts.append(g["speed"]+hops_str)
left = " "+display_name
right = " ".join(right_parts)+" "
self.peer_info_widget.original_widget.set_text(left+" | "+right)
def clear_history_dialog(self):
def dismiss_dialog(sender):
self.dialog_open = False
self.conversation_changed(None)
def confirmed(sender):
self.dialog_open = False
self.conversation.clear_history()
self.conversation_changed(None)
dialog = DialogLineBox(
urwid.Pile([
urwid.Text("Clear conversation history\n", align=urwid.CENTER),
urwid.Columns([
(urwid.WEIGHT, 0.45, urwid.Button("Yes", on_press=confirmed)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, urwid.Button("No", on_press=dismiss_dialog)),
])
]), title="?"
)
dialog.delegate = self
bottom = self.messagelist
overlay = urwid.Overlay(
dialog,
bottom,
align=urwid.CENTER,
width=34,
valign=urwid.MIDDLE,
height=urwid.PACK,
left=2,
right=2,
)
self.frame.contents["body"] = (overlay, self.frame.options())
self.frame.focus_position = "body"
def _build_footer(self):
g = self.app.ui.glyphs
if self.full_editor_active:
editor = self.full_editor
else:
editor = self.minimal_editor
if self.pending_attachments:
attachment_texts = []
for path in self.pending_attachments:
attachment_texts.append(os.path.basename(path))
indicator = urwid.AttrMap(
urwid.Text(g["file"]+" "+str(len(self.pending_attachments))+" file(s): "+", ".join(attachment_texts)),
"msg_header_sent",
)
return urwid.Pile([indicator, editor])
else:
return editor
def toggle_editor(self):
if self.full_editor_active:
self.full_editor_active = False
else:
self.full_editor_active = True
self.frame.contents["footer"] = (self._build_footer(), None)
def check_editor_allowed(self):
g = self.app.ui.glyphs
if self.frame:
allowed = nomadnet.NomadNetworkApp.get_shared_instance().directory.is_known(bytes.fromhex(self.source_hash))
if allowed:
self.frame.contents["footer"] = (self._build_footer(), None)
else:
warning = urwid.AttrMap(
urwid.Padding(urwid.Text(
"\n"+g["info"]+"\n\nYou cannot currently message this peer, since its identity keys are not known. "
"The keys have been requested from the network and should arrive shortly, if available. "
"Close this conversation and reopen it to try again.\n\n"
"To query the network manually, select this conversation in the conversation list, "
"press Ctrl-E, and use the query button.\n",
align=urwid.CENTER,
)),
"msg_header_caution",
)
self.frame.contents["footer"] = (warning, None)
def toggle_focus_area(self):
name = ""
try:
name = self.frame.get_focus_widgets()[0].name
except Exception as e:
pass
if name == "messagelist":
self.frame.focus_position = "footer"
elif name == "minimal_editor" or name == "full_editor":
self.frame.focus_position = "body"
def keypress(self, size, key):
if key == "tab":
self.toggle_focus_area()
elif key == "ctrl w":
self.close()
elif key == "ctrl u":
self.conversation.purge_failed()
self.conversation_changed(None)
elif key == "ctrl t":
self.toggle_editor()
elif key == "ctrl x":
self.clear_history_dialog()
elif key == "ctrl g":
nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.sub_displays.conversations_display.toggle_fullscreen()
elif key == "ctrl o":
self.sort_by_timestamp ^= True
self.conversation_changed(None)
elif key == "ctrl a":
self.attach_file()
elif key == "ctrl s":
self.save_focused_attachments()
else:
return super(ConversationWidget, self).keypress(size, key)
def conversation_changed(self, conversation):
if hasattr(self, "peer_info_widget"):
self._update_peer_info()
self.update_message_widgets(replace = True)
def update_message_widgets(self, replace = False):
while self.updating_message_widgets:
time.sleep(0.5)
self.updating_message_widgets = True
self.message_widgets = []
added_hashes = []
needs_index = []
for message in self.conversation.messages:
message_hash = message.get_hash()
if not message_hash in added_hashes:
added_hashes.append(message_hash)
was_loaded = message.loaded
message_widget = LXMessageWidget(message)
self.message_widgets.append(message_widget)
if not was_loaded and message.loaded:
needs_index.append(message)
message.unload()
if needs_index:
try:
ConversationMessage.write_index(
self.conversation.messages_path, needs_index)
except Exception:
pass
if self.sort_by_timestamp:
self.message_widgets.sort(key=lambda m: m.timestamp, reverse=False)
else:
self.message_widgets.sort(key=lambda m: m.sort_timestamp, reverse=False)
from nomadnet.vendor.additional_urwid_widgets import IndicativeListBox
self.messagelist = IndicativeListBox(self.message_widgets, position = len(self.message_widgets)-1)
self.messagelist.name = "messagelist"
if replace:
self.frame.contents["body"] = (self.messagelist, None)
nomadnet.NomadNetworkApp.get_shared_instance().ui.loop.draw_screen()
self.updating_message_widgets = False
def clear_editor(self):
self.content_editor.set_edit_text("")
self.title_editor.set_edit_text("")
self.pending_attachments = []
self.frame.contents["footer"] = (self._build_footer(), None)
def _collect_attachment_refs(self):
g = self.app.ui.glyphs
refs = []
sorted_messages = sorted(self.conversation.messages, key=lambda m: m.sort_timestamp, reverse=True)
for conv_message in sorted_messages:
if not conv_message.has_attachments():
continue
cached_names = conv_message._cached_attachment_names or []
att_file_idx = 0
for atype, aname, *arest in cached_names:
asize = arest[0] if arest else 0
glyph = g["file"] if atype == "file" else g[atype]
label = glyph+" "+aname
if asize > 0:
label += " ("+_format_size(asize)+")"
if atype == "file":
refs.append((label, aname, conv_message, "file", att_file_idx))
att_file_idx += 1
else:
refs.append((label, aname, conv_message, atype, 0))
return refs
def save_focused_attachments(self):
g = self.app.ui.glyphs
self.dialog_active = True
try:
attachment_items = self._collect_attachment_refs()
except Exception as e:
RNS.log("Error collecting attachments: "+str(e), RNS.LOG_ERROR)
attachment_items = []
save_dir = self.app.attachment_save_path if self.app.attachment_save_path else self.app.downloads_path
def dismiss_dialog(sender):
self.dialog_active = False
self.conversation_changed(None)
if not attachment_items:
dialog = DialogLineBox(
urwid.Pile([
urwid.Text("No attachments in this conversation.\n"),
urwid.Columns([
(urwid.WEIGHT, 0.6, urwid.Text("")),
(urwid.WEIGHT, 0.4, urwid.Button("OK", on_press=dismiss_dialog)),
])
]), title="Attachments"
)
dialog.delegate = self
bottom = self.messagelist
overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=45, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2)
self.frame.contents["body"] = (overlay, self.frame.options())
self.frame.focus_position = "body"
return
checkboxes = []
for label, filename, conv_msg, field_type, field_index in attachment_items:
cb = urwid.CheckBox(label, state=False)
cb._attachment_filename = filename
cb._conv_message = conv_msg
cb._field_type = field_type
cb._field_index = field_index
checkboxes.append(cb)
status_text = urwid.Text("")
def do_save(sender):
saved = []
errors = []
for cb in checkboxes:
if cb.get_state():
try:
src_path = cb._conv_message.get_attachment_file_path(cb._field_type, cb._field_index)
if src_path and os.path.isfile(src_path):
path = _copy_attachment_to_dest(cb._attachment_filename, src_path)
saved.append(path)
except Exception as e:
errors.append(str(e))
if saved:
lines = [g["check"]+" Copied "+str(len(saved))+" file(s) to "+save_dir+":"]
for p in saved:
lines.append(" "+os.path.basename(p))
if errors:
lines.append(g["cross"]+" "+str(len(errors))+" failed")
status_text.set_text("\n".join(lines))
elif errors:
status_text.set_text(g["cross"]+" Failed: "+errors[0])
else:
status_text.set_text("No files selected")
dialog_widgets = list(checkboxes)
dialog_widgets.append(urwid.Divider(g["divider1"]))
dialog_widgets.append(urwid.Text("Copy to: "+save_dir))
dialog_widgets.append(status_text)
dialog_widgets.append(urwid.Text(""))
dialog_widgets.append(urwid.Columns([
(urwid.WEIGHT, 0.45, urwid.Button("Copy to Downloads", on_press=do_save)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, urwid.Button("Close", on_press=dismiss_dialog)),
]))
dialog = DialogLineBox(urwid.ListBox(urwid.SimpleFocusListWalker(dialog_widgets)), title="Attachments")
dialog.delegate = self
bottom = self.messagelist
overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=("relative", 80), valign=urwid.MIDDLE, height=("relative", 80), left=2, right=2)
self.frame.contents["body"] = (overlay, self.frame.options())
self.frame.focus_position = "body"
def send_message(self):
content = self.content_editor.get_edit_text()
title = self.title_editor.get_edit_text()
if not content == "":
fields = None
if self.pending_attachments:
file_attachments = []
for file_path in self.pending_attachments:
try:
with open(file_path, "rb") as af:
file_data = af.read()
file_name = os.path.basename(file_path)
file_attachments.append([file_name, file_data])
except Exception as e:
RNS.log("Error reading attachment "+str(file_path)+": "+str(e), RNS.LOG_ERROR)
if file_attachments:
fields = {LXMF.FIELD_FILE_ATTACHMENTS: file_attachments}
if self.conversation.send(content, title, fields=fields):
self.clear_editor()
def attach_file(self):
self.dialog_active = True
browser = FileBrowserDialog(self)
bottom = self.messagelist
overlay = urwid.Overlay(browser, bottom, align=urwid.CENTER, width=("relative", 90), valign=urwid.MIDDLE, height=("relative", 80), left=2, right=2)
self.frame.contents["body"] = (overlay, self.frame.options())
self.frame.focus_position = "body"
def file_browser_closed(self):
self.dialog_active = False
self.frame.contents["footer"] = (self._build_footer(), None)
self.conversation_changed(None)
def paper_message_saved(self, path):
g = self.app.ui.glyphs
def dismiss_dialog(sender):
self.dialog_open = False
self.conversation_changed(None)
dialog = DialogLineBox(
urwid.Pile([
urwid.Text("The paper message was saved to:\n\n"+str(path)+"\n", align=urwid.CENTER),
urwid.Columns([
(urwid.WEIGHT, 0.6, urwid.Text("")),
(urwid.WEIGHT, 0.4, urwid.Button("OK", on_press=dismiss_dialog)),
])
]), title=g["papermsg"].replace(" ", "")
)
dialog.delegate = self
bottom = self.messagelist
overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=60, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2)
self.frame.contents["body"] = (overlay, self.frame.options())
self.frame.focus_position = "body"
def print_paper_message_qr(self):
content = self.content_editor.get_edit_text()
title = self.title_editor.get_edit_text()
if not content == "":
if self.conversation.paper_output(content, title):
self.clear_editor()
else:
self.paper_message_failed()
def save_paper_message_qr(self):
content = self.content_editor.get_edit_text()
title = self.title_editor.get_edit_text()
if not content == "":
output_result = self.conversation.paper_output(content, title, mode="save_qr")
if output_result != False:
self.clear_editor()
self.paper_message_saved(output_result)
else:
self.paper_message_failed()
def save_paper_message_uri(self):
content = self.content_editor.get_edit_text()
title = self.title_editor.get_edit_text()
if not content == "":
output_result = self.conversation.paper_output(content, title, mode="save_uri")
if output_result != False:
self.clear_editor()
self.paper_message_saved(output_result)
else:
self.paper_message_failed()
def paper_message(self):
def dismiss_dialog(sender):
self.dialog_open = False
self.conversation_changed(None)
def print_qr(sender):
dismiss_dialog(self)
self.print_paper_message_qr()
def save_qr(sender):
dismiss_dialog(self)
self.save_paper_message_qr()
def save_uri(sender):
dismiss_dialog(self)
self.save_paper_message_uri()
dialog = DialogLineBox(
urwid.Pile([
urwid.Text(
"Select the desired paper message output method.\nSaved files will be written to:\n\n"+str(self.app.downloads_path)+"\n",
align=urwid.CENTER,
),
urwid.Columns([
(urwid.WEIGHT, 0.5, urwid.Button("Print QR", on_press=print_qr)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.5, urwid.Button("Save QR", on_press=save_qr)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.5, urwid.Button("Save URI", on_press=save_uri)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.5, urwid.Button("Cancel", on_press=dismiss_dialog))
])
]), title="Create Paper Message"
)
dialog.delegate = self
bottom = self.messagelist
overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=60, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2)
self.frame.contents["body"] = (overlay, self.frame.options())
self.frame.focus_position = "body"
def paper_message_failed(self):
def dismiss_dialog(sender):
self.dialog_open = False
self.conversation_changed(None)
dialog = DialogLineBox(
urwid.Pile([
urwid.Text(
"Could not output paper message,\ncheck your settings. See the log\nfile for any error messages.\n",
align=urwid.CENTER,
),
urwid.Columns([
(urwid.WEIGHT, 0.6, urwid.Text("")),
(urwid.WEIGHT, 0.4, urwid.Button("OK", on_press=dismiss_dialog)),
])
]), title="!"
)
dialog.delegate = self
bottom = self.messagelist
overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=34, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2)
self.frame.contents["body"] = (overlay, self.frame.options())
self.frame.focus_position = "body"
def close(self):
self.delegate.close_conversation(self)
class LXMessageWidget(urwid.WidgetWrap):
def __init__(self, message):
app = nomadnet.NomadNetworkApp.get_shared_instance()
g = app.ui.glyphs
self.timestamp = message.get_timestamp()
self.sort_timestamp = message.sort_timestamp
self.transfer_done = False
self._live_lxm = None
msg_hash = message.get_hash()
msg_state = message.get_state()
msg_source_hash = message._cached_source_hash
msg_method = message._cached_method
time_format = app.time_format
message_time = datetime.fromtimestamp(self.timestamp)
encryption_string = ""
if message.get_transport_encrypted():
encryption_string = " "+g["encrypted"]
else:
encryption_string = " "+g["plaintext"]
title_string = relative_time(self.timestamp)+" | "+message_time.strftime(time_format)+encryption_string
is_outbound = False
if msg_source_hash is None:
header_style = "msg_header_failed"
title_string = g["warning"]+" "+title_string
elif app.lxmf_destination.hash == msg_source_hash:
is_outbound = True
if msg_state == LXMF.LXMessage.DELIVERED:
header_style = "msg_header_delivered"
title_string = g["check"]+" "+g["arrow_r"]+" "+title_string
elif msg_state == LXMF.LXMessage.FAILED:
header_style = "msg_header_failed"
title_string = g["cross"]+" "+g["arrow_r"]+" "+title_string
elif msg_state == LXMF.LXMessage.REJECTED:
header_style = "msg_header_failed"
title_string = g["cross"]+" "+g["arrow_r"]+" Rejected "+title_string
elif msg_method == LXMF.LXMessage.PROPAGATED and msg_state == LXMF.LXMessage.SENT:
header_style = "msg_header_propagated"
title_string = g["sent"]+" "+g["arrow_r"]+" "+title_string
elif msg_method == LXMF.LXMessage.PAPER and msg_state == LXMF.LXMessage.PAPER:
header_style = "msg_header_propagated"
title_string = g["papermsg"]+" "+g["arrow_r"]+" "+title_string
elif msg_state == LXMF.LXMessage.SENT:
header_style = "msg_header_sent"
title_string = g["sent"]+" "+g["arrow_r"]+" "+title_string
else:
header_style = "msg_header_sent"
title_string = g["arrow_r"]+" "+title_string
else:
if message.signature_validated():
header_style = "msg_header_ok"
title_string = g["check"]+" "+g["arrow_l"]+" "+title_string
else:
header_style = "msg_header_caution"
title_string = g["warning"]+" "+g["arrow_l"]+" "+message.get_signature_description() + "\n " + title_string
if message.get_title() != "":
title_string += " | " + message.get_title()
has_attachments = message.has_attachments()
cached_names = message._cached_attachment_names or []
if has_attachments and cached_names:
attachment_strings = []
for atype, aname, *arest in cached_names:
attachment_strings.append(g[atype if atype != "file" else "file"]+" "+aname)
title_string += " | " + " ".join(attachment_strings)
title = urwid.AttrMap(urwid.Text(title_string), header_style)
self.progress_widget = urwid.Text("")
self.progress_attr = urwid.AttrMap(self.progress_widget, "progress_full")
content_text = message.get_content()
content_lines = content_text.split("\n")
indented = "\n".join(" "+line for line in content_lines)
pile_widgets = [title]
if is_outbound and msg_state is not None and msg_state < LXMF.LXMessage.SENT and msg_hash is not None:
try:
for pending in app.message_router.pending_outbound:
if pending.hash == msg_hash:
if pending.representation == LXMF.LXMessage.RESOURCE:
self._live_lxm = pending
break
except Exception:
pass
if self._live_lxm is not None:
pct = int(self._live_lxm.progress * 100)
bar_width = 20
filled = int(bar_width * self._live_lxm.progress)
if app.ui.colormode >= 256:
bar = "\u2588" * filled + "\u2591" * (bar_width - filled)
else:
bar = "#" * filled + "-" * (bar_width - filled)
self.progress_widget.set_text(" ["+bar+"] "+str(pct)+"%")
pile_widgets.append(self.progress_attr)
self._start_progress_poll()
pile_widgets.append(urwid.Text(indented))
if has_attachments and cached_names:
att_file_idx = 0
for atype, aname, *arest in cached_names:
glyph = g["file"] if atype == "file" else g[atype]
asize = arest[0] if arest else 0
label = " "+glyph+" "+aname
if asize > 0:
label += " ("+_format_size(asize)+")"
if atype == "file":
pile_widgets.append(ClickableAttachment(label, aname, message, "file", att_file_idx))
att_file_idx += 1
else:
pile_widgets.append(ClickableAttachment(label, aname, message, atype))
pile_widgets.append(urwid.Text(""))
super().__init__(urwid.Pile(pile_widgets))
def _start_progress_poll(self):
try:
loop = nomadnet.NomadNetworkApp.get_shared_instance().ui.loop
if loop:
loop.set_alarm_in(0.3, self._poll_progress)
except Exception:
pass
def _poll_progress(self, loop=None, user_data=None):
if self.transfer_done:
return
if self._live_lxm is None:
self.transfer_done = True
return
app = nomadnet.NomadNetworkApp.get_shared_instance()
g = app.ui.glyphs
progress = self._live_lxm.progress
state = self._live_lxm.state
pct = int(progress * 100)
if state == LXMF.LXMessage.FAILED:
self.progress_widget.set_text(" "+g["cross"]+" Transfer failed")
self.transfer_done = True
self._live_lxm = None
elif state == LXMF.LXMessage.REJECTED:
self.progress_widget.set_text(" "+g["cross"]+" Rejected: too large or not accepted")
self.transfer_done = True
self._live_lxm = None
elif state >= LXMF.LXMessage.SENT:
self.progress_widget.set_text("")
self.transfer_done = True
self._live_lxm = None
else:
bar_width = 20
filled = int(bar_width * progress)
if app.ui.colormode >= 256:
bar = "\u2588" * filled + "\u2591" * (bar_width - filled)
else:
bar = "#" * filled + "-" * (bar_width - filled)
self.progress_widget.set_text(" ["+bar+"] "+str(pct)+"%")
if not self.transfer_done:
try:
ui_loop = app.ui.loop
if ui_loop:
ui_loop.set_alarm_in(0.3, self._poll_progress)
ui_loop.draw_screen()
except Exception:
pass
class ClickableAttachment(urwid.Text):
def __init__(self, label, filename, conv_message, field_type, field_index=0):
self.filename = filename
self.conv_message = conv_message
self.field_type = field_type
self.field_index = field_index
self.saved = False
super().__init__(label)
def mouse_event(self, size, event, button, x, y, focus):
if button == 1 and urwid.util.is_mouse_press(event):
self._save()
return True
return False
def _save(self):
if self.saved:
return
app = nomadnet.NomadNetworkApp.get_shared_instance()
g = app.ui.glyphs
try:
src_path = self.conv_message.get_attachment_file_path(self.field_type, self.field_index)
if src_path and os.path.isfile(src_path):
save_path = _copy_attachment_to_dest(self.filename, src_path)
else:
if self.field_type == "file":
attachments = self.conv_message.get_file_attachments()
if self.field_index < len(attachments):
att = attachments[self.field_index]
if isinstance(att, list) and len(att) >= 2:
data = att[1] if isinstance(att[1], bytes) else b""
else:
data = b""
else:
data = b""
elif self.field_type == "image":
data = self.conv_message.get_image()
data = data if isinstance(data, bytes) else b""
elif self.field_type == "audio":
data = self.conv_message.get_audio()
data = data if isinstance(data, bytes) else b""
else:
data = b""
self.conv_message.unload()
if not data:
return
save_path = _save_attachment_to_disk(self.filename, data)
self.saved = True
self.set_text(" "+g["check"]+" Copied to: "+save_path)
except Exception as e:
RNS.log("Error saving attachment: "+str(e), RNS.LOG_ERROR)
self.set_text(" "+g["cross"]+" Save failed: "+str(e))
def _copy_attachment_to_dest(filename, src_path):
app = nomadnet.NomadNetworkApp.get_shared_instance()
save_dir = app.attachment_save_path if app.attachment_save_path else app.downloads_path
if not os.path.isdir(save_dir):
os.makedirs(save_dir)
save_path = os.path.join(save_dir, filename)
counter = 0
base, ext = os.path.splitext(filename)
while os.path.isfile(save_path):
counter += 1
save_path = os.path.join(save_dir, base+"_"+str(counter)+ext)
shutil.copy2(src_path, save_path)
return save_path
def _save_attachment_to_disk(filename, data):
app = nomadnet.NomadNetworkApp.get_shared_instance()
save_dir = app.attachment_save_path if app.attachment_save_path else app.downloads_path
if not os.path.isdir(save_dir):
os.makedirs(save_dir)
save_path = os.path.join(save_dir, filename)
counter = 0
base, ext = os.path.splitext(filename)
while os.path.isfile(save_path):
counter += 1
save_path = os.path.join(save_dir, base+"_"+str(counter)+ext)
with open(save_path, "wb") as f:
f.write(data)
return save_path
class FileBrowserEntry(urwid.WidgetWrap):
signals = ["click"]
def __init__(self, name, full_path, is_dir=False, is_parent=False, selected=False):
self.full_path = full_path
self.name = name
self.is_dir = is_dir
self.is_parent = is_parent
self.selected = selected
g = nomadnet.NomadNetworkApp.get_shared_instance().ui.glyphs
if is_parent:
display = g["arrow_l"]+" .."
elif is_dir:
display = g["arrow_r"]+" "+name+"/"
elif selected:
display = g["check"]+" "+name
else:
display = " "+name
self.text_widget = urwid.SelectableIcon(display, 0)
if is_dir or is_parent:
style = "list_trusted"
focus_style = "list_focus"
elif selected:
style = "list_trusted"
focus_style = "list_focus_trusted"
else:
style = "list_unknown"
focus_style = "list_focus"
display_widget = urwid.AttrMap(self.text_widget, style, focus_style)
super().__init__(display_widget)
def keypress(self, size, key):
if key == "enter":
self._emit("click")
else:
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 FileBrowserDialog(urwid.WidgetWrap):
def __init__(self, delegate):
self.delegate = delegate
app = nomadnet.NomadNetworkApp.get_shared_instance()
self.g = app.ui.glyphs
self.current_path = os.path.expanduser("~")
self.path_label = urwid.Text("")
self.status_label = urwid.Text("")
self.file_walker = urwid.SimpleFocusListWalker([])
self.file_listbox = urwid.ListBox(self.file_walker)
self.button_columns = urwid.Columns([
(urwid.WEIGHT, 0.45, urwid.Button("Done", on_press=self._dismiss)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, urwid.Button("Cancel", on_press=self._cancel)),
])
header_pile = urwid.Pile([
self.path_label,
self.status_label,
urwid.Divider(self.g["divider1"]),
])
footer_pile = urwid.Pile([
urwid.Divider(self.g["divider1"]),
self.button_columns,
])
self._populate()
self.browser_frame = urwid.Frame(
self.file_listbox,
header=header_pile,
footer=footer_pile,
)
linebox = urwid.LineBox(self.browser_frame, title="Attach File")
super().__init__(linebox)
def _update_status(self):
pending = self.delegate.pending_attachments
if pending:
names = [os.path.basename(p) for p in pending]
self.status_label.set_text(" "+self.g["file"]+" "+str(len(pending))+" selected: "+", ".join(names))
else:
self.status_label.set_text(" No files selected")
def _populate(self):
self.path_label.set_text(" "+self.current_path)
self._update_status()
focus_pos = None
try:
focus_pos = self.file_listbox.focus_position
except Exception:
pass
entries = []
parent = os.path.dirname(self.current_path)
if parent != self.current_path:
entry = FileBrowserEntry("..", parent, is_parent=True)
urwid.connect_signal(entry, "click", self._entry_clicked, entry)
entries.append(entry)
try:
items = sorted(os.listdir(self.current_path))
except PermissionError:
entries.append(urwid.Text(("error_text", " Permission denied")))
self.file_walker[:] = entries
return
dirs = []
files = []
for item in items:
if item.startswith("."):
continue
full = os.path.join(self.current_path, item)
if os.path.isdir(full):
dirs.append((item, full))
elif os.path.isfile(full):
files.append((item, full))
for name, full in dirs:
entry = FileBrowserEntry(name, full, is_dir=True)
urwid.connect_signal(entry, "click", self._entry_clicked, entry)
entries.append(entry)
for name, full in files:
is_selected = full in self.delegate.pending_attachments
entry = FileBrowserEntry(name, full, selected=is_selected)
urwid.connect_signal(entry, "click", self._entry_clicked, entry)
entries.append(entry)
if not dirs and not files:
entries.append(urwid.Text(("inactive_text", " (empty)")))
self.file_walker[:] = entries
if focus_pos is not None and focus_pos < len(entries):
self.file_listbox.set_focus(focus_pos)
elif entries:
self.file_listbox.set_focus(0)
def _entry_clicked(self, entry_widget, user_data=None):
entry = user_data if user_data else entry_widget
if entry.is_dir or entry.is_parent:
self.current_path = entry.full_path
self._populate()
else:
if entry.full_path in self.delegate.pending_attachments:
self.delegate.pending_attachments.remove(entry.full_path)
else:
self.delegate.pending_attachments.append(entry.full_path)
self.delegate.frame.contents["footer"] = (self.delegate._build_footer(), None)
self._populate()
def _dismiss(self, sender):
self.delegate.file_browser_closed()
def _cancel(self, sender):
self.delegate.pending_attachments.clear()
self.delegate.frame.contents["footer"] = (self.delegate._build_footer(), None)
self.delegate.file_browser_closed()
def keypress(self, size, key):
if key == "esc":
self.delegate.file_browser_closed()
return
result = super().keypress(size, key)
if result == "down" and self.browser_frame.focus_position == "body":
self.browser_frame.focus_position = "footer"
return
elif result == "up" and self.browser_frame.focus_position == "footer":
self.browser_frame.focus_position = "body"
return
return result
class SyncProgressBar(urwid.ProgressBar):
def get_text(self):
status = nomadnet.NomadNetworkApp.get_shared_instance().get_sync_status()
show_percent = nomadnet.NomadNetworkApp.get_shared_instance().sync_status_show_percent()
if show_percent:
return status+" "+super().get_text()
else:
return status
Reference in New Issue View Git Blame Copy Permalink