diff --git a/app/ui/app_menu_bar.ui b/app/ui/app_menu_bar.ui index 7f9aa73f..efe51dd4 100644 --- a/app/ui/app_menu_bar.ui +++ b/app/ui/app_menu_bar.ui @@ -87,18 +87,22 @@ Tools - - Satellites editor - app.on_satellite_editor_show - - - Picons downloader - app.on_picons_loader_show - - - Backups - app.on_backup_tool_show - +
+ + Satellites editor + app.on_satellite_editor_show + + + Picons downloader + app.on_picons_loader_show + + + Backups + app.on_backup_tool_show + +
+
+
IPTV diff --git a/app/ui/main_app_window.py b/app/ui/main_app_window.py index 55e8ee05..14eb2153 100644 --- a/app/ui/main_app_window.py +++ b/app/ui/main_app_window.py @@ -71,6 +71,7 @@ class Application(Gtk.Application): # Adding command line options self.add_main_option("log", ord("l"), GLib.OptionFlags.NONE, GLib.OptionArg.NONE, "", None) self.add_main_option("record", ord("r"), GLib.OptionFlags.NONE, GLib.OptionArg.NONE, "", None) + self.add_main_option("telnet", ord("t"), GLib.OptionFlags.NONE, GLib.OptionArg.STRING, "", None) self._handlers = {"on_close_app": self.on_close_app, "on_resize": self.on_resize, @@ -305,9 +306,13 @@ class Application(Gtk.Application): self.set_menubar(builder.get_object("menu_bar")) self.set_app_menu(builder.get_object("app-menu")) + if self._settings.get("telnet"): + self.init_telnet(builder) + self.update_profile_label() self.init_drag_and_drop() self.init_colors() + if self._settings.load_last_config: config = self._settings.get("last_config") or {} self.init_profiles(config.get("last_profile", None)) @@ -318,6 +323,14 @@ class Application(Gtk.Application): gen = self.init_http_api() GLib.idle_add(lambda: next(gen, False), priority=GLib.PRIORITY_LOW) + def init_telnet(self, builder): + t_section = builder.get_object("telnet_section") + t_section.append_item(Gio.MenuItem.new("Telnet", "app.on_telnet_client_show")) + ac = Gio.SimpleAction.new("on_telnet_client_show", None) + ac.connect("activate", self.on_telnet_client_show) + self.add_action(ac) + self.set_accels_for_action("app.on_telnet_client_show", ["t"]) + def init_keys(self): main_handlers = ("on_new_configuration", "on_data_open", "on_download", "on_settings", "on_close_app", "on_import_bouquet", "on_import_bouquets", "on_satellite_editor_show", @@ -411,6 +424,17 @@ class Application(Gtk.Application): log("Starting record of current stream...") log("Not implemented yet!") + if "telnet" in options: + t_op = options.get("telnet", "off") + if t_op == "on": + self._settings.add("telnet", True) + elif t_op == "off": + self._settings.add("telnet", False) + else: + log("No valid [on, off] arguments for -t found!") + return 1 + log("Telnet support is {}. Restart the program to apply the settings!".format(t_op)) + self.activate() return 0 @@ -1707,6 +1731,12 @@ class Application(Gtk.Application): """ Shows backup tool dialog """ BackupDialog(self._main_window, self._settings, self.open_data).show() + # ***************** Telnet *********************# + + def on_telnet_client_show(self, action, value=None): + from app.ui.telnet import TelnetDialog + TelnetDialog(self._main_window, self._settings).show() + # ***************** Player *********************# def on_play_stream(self, item=None): diff --git a/app/ui/main_window.glade b/app/ui/main_window.glade index 14f3e820..f90cde02 100644 --- a/app/ui/main_window.glade +++ b/app/ui/main_window.glade @@ -2597,6 +2597,7 @@ Author: Dmitriy Yefremov status-bar-button + 28 False True Record @@ -2618,6 +2619,7 @@ Author: Dmitriy Yefremov status-bar-button + 28 False False True diff --git a/app/ui/telnet.glade b/app/ui/telnet.glade new file mode 100755 index 00000000..a38edcca --- /dev/null +++ b/app/ui/telnet.glade @@ -0,0 +1,249 @@ + + + + + + + + + + + + Normal + False + + + + + tag_table + + + False + DemonEditor [Telnet client] + True + terminal + True + True + + + + + + + 560 + 320 + True + False + vertical + 2 + + + True + True + in + + + True + True + char + 5 + 5 + text_buffer + True + GTK_INPUT_HINT_WORD_COMPLETION | GTK_INPUT_HINT_NONE + True + + + + + + + True + True + 0 + + + + + True + False + 5 + 5 + 2 + 2 + + + True + False + 0 + False + + + + False + True + 0 + + + + + True + True + True + Connect + + + + True + False + gtk-connect + + + + + False + True + 1 + + + + + True + True + Disconnect + + + + True + False + gtk-disconnect + + + + + False + True + 4 + + + + + + + + True + True + True + Clear + center + center + + + + True + False + gtk-clear + + + + + False + True + end + 1 + + + + + False + True + 1 + + + + + True + False + 2 + True + + + + False + end + + + + + + False + False + 0 + + + + + False + + + True + False + Info + center + True + word-char + 2 + + + False + False + 1 + + + + + False + False + 0 + + + + + + + + False + True + 2 + + + + + + diff --git a/app/ui/telnet.py b/app/ui/telnet.py new file mode 100755 index 00000000..99dc3bc6 --- /dev/null +++ b/app/ui/telnet.py @@ -0,0 +1,257 @@ +import re +import selectors +import socket +from collections import deque +from telnetlib import Telnet + +from gi.repository import GLib + +from app.commons import run_task, run_idle, log +from app.settings import Settings +from app.ui.uicommons import Gtk, Gdk, UI_RESOURCES_PATH, KeyboardKey, MOD_MASK + + +class ExtTelnet(Telnet): + + def __init__(self, output_callback, **kwargs): + super().__init__(**kwargs) + self._output_callback = output_callback + + def interact(self): + """Interaction function, emulates a very dumb telnet client.""" + with selectors.DefaultSelector() as selector: + selector.register(self, selectors.EVENT_READ) + + while True: + for key, events in selector.select(): + if key.fileobj is self: + try: + text = self.read_very_eager() + except EOFError as e: + msg = "\n*** Connection closed by remote host ***\n" + self._output_callback(msg) + log(msg) + raise e + else: + if text: + self._output_callback(text) + + +class TelnetDialog: + """ Dialog of very simple telnet client. """ + _COLOR_PATTERN = re.compile("\x1b.*?m") # Color info + _ERASING_PATTERN = re.compile("\x1b.*?K") # Erase to right + _APP_MODE_PATTERN = re.compile("\x1b.*?(1h)|(1l)") # h - on, l - off + _ALL_PATTERN = re.compile(r'(\x1b\[|\x9b)[0-?]*[@-~]') + _NOT_SUPPORTED = {"mc", "mcedit", "vi", "nano"} + + def __init__(self, transient, settings): + self._handlers = {"on_profile_changed": self.on_profile_changed, + "on_clear": self.on_clear, + "on_text_view_realize": self.on_text_view_realize, + "on_view_key_press": self.on_view_key_press, + "on_info_bar_close": self.on_info_bar_close, + "on_connect": self.on_connect, + "on_disconnect": self.on_disconnect, + "on_close": self.on_close} + + builder = Gtk.Builder() + builder.add_from_file(UI_RESOURCES_PATH + "telnet.glade") + builder.connect_signals(self._handlers) + self._dialog_window = builder.get_object("dialog_window") + self._dialog_window.set_transient_for(transient) + self._profile_combo_box = builder.get_object("profile_combo_box") + self._info_bar = builder.get_object("info_bar") + self._info_message_label = builder.get_object("info_bar_message_label") + self._text_view = builder.get_object("text_view") + self._buf = builder.get_object("text_buffer") + self._end_tag = builder.get_object("end_tag") + self._connect_button = builder.get_object("connect_button") + self._connect_button.bind_property("visible", builder.get_object("disconnect_button"), "visible", 4) + # Font style + provider = Gtk.CssProvider() + provider.load_from_data(b"textview {font-size: 14px;}") + self._text_view.get_style_context().add_provider(provider, Gtk.STYLE_PROVIDER_PRIORITY_USER) + + window_size = settings.get("telnet_dialog_window_size") + if window_size: + self._dialog_window.resize(*window_size) + + self._ext_settings = settings + self._settings = Settings(settings.settings) + self._tn = None + self._app_mode = False + self._commands = deque(maxlen=10) + + def show(self): + self._dialog_window.show() + + def on_close(self, window, event): + """ Performs shutdown tasks """ + self._ext_settings.add("telnet_dialog_window_size", window.get_size()) + self.on_disconnect() + + def on_info_bar_close(self, bar=None, resp=None): + self._info_bar.set_visible(False) + + @run_idle + def show_info_message(self, text, message_type): + self._info_bar.set_visible(True) + self._info_bar.set_message_type(message_type) + self._info_message_label.set_text(text) + + def on_text_view_realize(self, view): + self.init_profiles() + self.on_connect() + + @run_idle + def init_profiles(self): + for p in self._settings.profiles: + self._profile_combo_box.append(p, p) + self._profile_combo_box.set_active_id(self._settings.current_profile) + + @run_task + def on_connect(self, item=None): + try: + GLib.idle_add(self._connect_button.set_visible, False) + GLib.idle_add(self.on_info_bar_close) + user, password = self._settings.telnet_user, self._settings.telnet_password + timeout = self._settings.telnet_timeout + + self._tn = ExtTelnet(self.append_output, + host=self._settings.host, + port=self._settings.telnet_port, + timeout=timeout) + + if user != "": + self._tn.read_until(b"login: ") + self._tn.write(user.encode("utf-8") + b"\n") + if password != "": + self._tn.read_until(b"Password: ") + self._tn.write(password.encode("utf-8") + b"\n") + + self._tn.interact() + except (OSError, EOFError, socket.timeout, ConnectionRefusedError) as e: + log("{}: {}".format(self.__class__.__name__, e)) + self.show_info_message(str(e), Gtk.MessageType.ERROR) + finally: + GLib.idle_add(self._connect_button.set_visible, True) + + @run_task + def on_disconnect(self, item=None): + if self._tn: + GLib.idle_add(self._connect_button.set_visible, True) + self._tn.close() + + def on_profile_changed(self, button): + self._settings.current_profile = button.get_active_id() + + def on_command_done(self, entry): + command = entry.get_text() + entry.set_text("") + if command and self._tn: + self._tn.write(command.encode("ascii") + b"\r") + + def on_clear(self, item=None): + self._buf.delete(self._buf.get_start_iter(), self._buf.get_end_iter()) + + def on_view_key_press(self, view, event): + """ Handling keystrokes on press """ + if event.keyval == Gdk.KEY_Return: + self.do_command() + return True + + key_code = event.hardware_keycode + if not KeyboardKey.value_exist(key_code): + return + + key = KeyboardKey(key_code) + ctrl = event.state & MOD_MASK + if ctrl and key is KeyboardKey.C: + if self._tn and self._tn.sock: + self._tn.write(b"\x03") # interrupt + + # last commands navigation + if key is KeyboardKey.UP: + self.delete_last_command() + if self._commands: + cmd = self._commands.pop() + self._commands.appendleft(cmd) + self._buf.insert_at_cursor(cmd, -1) + return True + elif key is KeyboardKey.DOWN: + self.delete_last_command() + if self._commands: + cmd = self._commands.popleft() + self._commands.append(cmd) + self._buf.insert_at_cursor(cmd, -1) + return True + + def delete_last_command(self): + end = self._buf.get_end_iter() + if end.ends_tag(self._end_tag): + return + + if end.backward_to_tag_toggle(self._end_tag): + self._buf.delete(self._buf.get_end_iter(), end) + + def do_command(self): + count = self._buf.get_line_count() + begin = self._buf.get_iter_at_line(count) + end = self._buf.get_end_iter() + command = [] + + while end.backward_to_tag_toggle(self._end_tag): + command.append(self._buf.get_text(end, begin, False)) + break + else: # if buf is empty + command.append(self._buf.get_text(begin, end, False)) + + # to preventing duplication of the command in the buf + self._buf.delete(end, begin) + + if command and self._tn.sock: + cmd = command[0] + if cmd in self._NOT_SUPPORTED: + self.show_info_message("'{}' is not supported by this client.".format(cmd), Gtk.MessageType.ERROR) + else: + self._tn.write(cmd.encode("ascii") + b"\r") + self._commands.append(cmd) + + @run_idle + def append_output(self, txt): + t = txt.decode("ascii", errors="ignore") + + ap = re.search(self._APP_MODE_PATTERN, t) + if ap: + on, of = ap.group(1), ap.group(2) + if on: + self._app_mode = True + elif of: + self._app_mode = False + self.on_clear() + + t = re.sub(self._ALL_PATTERN, "", t) # removing [replacing] ascii escape sequences + + if self._app_mode: + start, end = self._buf.get_start_iter(), self._buf.get_end_iter() + count = self._buf.get_line_count() + new_lines = t.split("\r\n") + ext_lines = self._buf.get_text(start, end, True).split("\r\n") + if count < len(new_lines): + self._buf.set_text(re.sub(self._ERASING_PATTERN, "", t)) + else: + for i, line in enumerate(new_lines): + if line: + ext_lines[i] = re.sub(self._ERASING_PATTERN, "", line) + self._buf.set_text("\r\n".join(ext_lines)) + else: + self._buf.insert_at_cursor(t, -1) + + insert = self._buf.get_insert() + self._text_view.scroll_to_mark(insert, 0.0, True, 0.0, 1.0) + self._buf.apply_tag(self._end_tag, self._buf.get_start_iter(), self._buf.get_end_iter()) + + +if __name__ == "__main__": + pass