Nemcio/DemonEditor
1
0
Fork 0
mirror of https://github.com/DYefremov/DemonEditor.git synced 2025-12-21 16:09:41 +01:00
Code Issues Packages Projects Releases Activity

http api refactoring

Browse Source
This commit is contained in:
DYefremov
2020-01-08 21:33:24 +03:00
parent 024d48b464
commit 2e3ec1c99d
3 changed files with 83 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
app/connections.py
View File

@@ -1,15 +1,17 @@
import json
import os import os
import re
import socket import socket
import time import time
import urllib import urllib
import xml.etree.ElementTree as ETree
from enum import Enum from enum import Enum
from ftplib import FTP, error_perm from ftplib import FTP, error_perm
from http.client import RemoteDisconnected from http.client import RemoteDisconnected
from telnetlib import Telnet from telnetlib import Telnet
from urllib.error import HTTPError, URLError from urllib.error import HTTPError, URLError
from urllib.parse import urlencode from urllib.parse import urlencode
from urllib.request import urlopen, HTTPPasswordMgrWithDefaultRealm, HTTPBasicAuthHandler, build_opener, install_opener from urllib.request import urlopen, HTTPPasswordMgrWithDefaultRealm, HTTPBasicAuthHandler, \
build_opener, install_opener, Request
from app.commons import log from app.commons import log
from app.settings import SettingsType from app.settings import SettingsType
@@ -22,6 +24,8 @@ _DATA_FILES_LIST = ("lamedb", "lamedb5", "services.xml", "blacklist", "whitelist
_SAT_XML_FILE = "satellites.xml" _SAT_XML_FILE = "satellites.xml"
_WEBTV_XML_FILE = "webtv.xml" _WEBTV_XML_FILE = "webtv.xml"
_HEADERS = {"User-Agent": "Mozilla/5.0 (X11; Linux i586; rv:31.0) Gecko/20100101 Firefox/71.0"}
class DownloadType(Enum): class DownloadType(Enum):
ALL = 0 ALL = 0
@@ -37,8 +41,9 @@ class HttpRequestType(Enum):
INFO = "about" INFO = "about"
SIGNAL = "tunersignal" SIGNAL = "tunersignal"
STREAM = "streamcurrentm3u" STREAM = "streamcurrentm3u"
STATUS = "statusinfo" CURRENT = "getcurrent"
PLAY = "mediaplayerplay?file=4097:0:1:0:0:0:0:0:0:0:" PLAY = "mediaplayerplay?file=4097:0:1:0:0:0:0:0:0:0:"
TEST = None
class TestException(Exception): class TestException(Exception):
@@ -101,7 +106,7 @@ def upload_data(*, settings, download_type=DownloadType.ALL, remove_unused=False
s_type = settings.setting_type s_type = settings.setting_type
data_path = settings.data_local_path data_path = settings.data_local_path
host = settings.host host = settings.host
base_url = "http{}://{}:{}/api/".format("s" if settings.http_use_ssl else "", host, settings.http_port) base_url = "http{}://{}:{}/web/".format("s" if settings.http_use_ssl else "", host, settings.http_port)
tn, ht = None, None # telnet, http tn, ht = None, None # telnet, http
try: try:
@@ -122,7 +127,7 @@ def upload_data(*, settings, download_type=DownloadType.ALL, remove_unused=False
if download_type is DownloadType.ALL: if download_type is DownloadType.ALL:
time.sleep(5) time.sleep(5)
ht.send(base_url + "/powerstate?newstate=0") ht.send(base_url + "powerstate?newstate=0")
time.sleep(2) time.sleep(2)
else: else:
# telnet # telnet
@@ -167,10 +172,10 @@ def upload_data(*, settings, download_type=DownloadType.ALL, remove_unused=False
tn.send("init 3" if s_type is SettingsType.ENIGMA_2 else "init 6") tn.send("init 3" if s_type is SettingsType.ENIGMA_2 else "init 6")
elif ht and use_http: elif ht and use_http:
if download_type is DownloadType.BOUQUETS: if download_type is DownloadType.BOUQUETS:
ht.send(base_url + "/servicelistreload?mode=2") ht.send(base_url + "servicelistreload?mode=2")
elif download_type is DownloadType.ALL: elif download_type is DownloadType.ALL:
ht.send(base_url + "/servicelistreload?mode=0") ht.send(base_url + "servicelistreload?mode=0")
ht.send(base_url + "/powerstate?newstate=4") ht.send(base_url + "powerstate?newstate=4")
if done_callback is not None: if done_callback is not None:
done_callback() done_callback()
@@ -246,10 +251,9 @@ def http(user, password, url, callback):
init_auth(user, password, url) init_auth(user, password, url)
while True: while True:
url = yield url = yield
with urlopen(url, timeout=5) as f: msg = get_response(HttpRequestType.TEST, url).get("e2statetext", None)
msg = json.loads(f.read().decode("utf-8")).get("message", None) if msg:
if msg: callback("HTTP: {}\n".format(msg))
callback("HTTP: {}\n".format(msg))
def telnet(host, port=23, user="", password="", timeout=5): def telnet(host, port=23, user="", password="", timeout=5):
@@ -298,28 +302,36 @@ class HttpAPI:
elif req_type is HttpRequestType.PLAY: elif req_type is HttpRequestType.PLAY:
url += urllib.parse.quote(ref).replace("%3A", "%253A") url += urllib.parse.quote(ref).replace("%3A", "%253A")
future = self._executor.submit(get_json, req_type, url) future = self._executor.submit(get_response, req_type, url)
future.add_done_callback(lambda f: callback(f.result())) future.add_done_callback(lambda f: callback(f.result()))
def init(self): def init(self):
use_ssl = self._settings.http_use_ssl use_ssl = self._settings.http_use_ssl
url = "http{}://{}:{}".format("s" if use_ssl else "", self._settings.host, self._settings.http_port) url = "http{}://{}:{}".format("s" if use_ssl else "", self._settings.host, self._settings.http_port)
self._base_url = "{}/api/".format(url) self._base_url = "{}/web/".format(url)
init_auth(self._settings.http_user, self._settings.http_password, url, use_ssl) init_auth(self._settings.http_user, self._settings.http_password, url, use_ssl)
def close(self): def close(self):
self._executor.shutdown(False) self._executor.shutdown(False)
def get_json(req_type, url): def get_response(req_type, url):
try: try:
with urlopen(url, timeout=10) as f: with urlopen(Request(url, headers=_HEADERS), timeout=10) as f:
if req_type is HttpRequestType.STREAM: if req_type is HttpRequestType.STREAM:
return f.read().decode("utf-8") return f.read().decode("utf-8")
elif req_type is HttpRequestType.CURRENT:
for e in ETree.fromstring(f.read().decode("utf-8")).iter("e2event"):
return {e.tag: e.text for e in e.iter()} # return first[current] event from the list
else: else:
return json.loads(f.read().decode("utf-8")) return {e.tag: e.text for e in ETree.fromstring(f.read().decode("utf-8")).iter()}
except (URLError, HTTPError, RemoteDisconnected): except (URLError, HTTPError, RemoteDisconnected, ConnectionResetError) as e:
pass if req_type is HttpRequestType.TEST:
raise e
except ETree.ParseError as e:
log("Parsing response error: {}".format(e))
return {}
# ***************** Connections testing *******************# # ***************** Connections testing *******************#
@@ -339,24 +351,9 @@ def test_http(host, port, user, password, timeout=5, use_ssl=False, skip_message
# authentication # authentication
init_auth(user, password, base_url, use_ssl) init_auth(user, password, base_url, use_ssl)
try: try:
with urlopen("{}/api/{}".format(base_url, params), timeout=5) as f: return get_response(HttpRequestType.TEST, "{}/web/{}".format(base_url, params)).get("e2statetext", "")
return json.loads(f.read().decode("utf-8")).get("message", "")
except HTTPError as e:
if e.code == 404:
return test_api("{}/web/{}".format(base_url, params))
raise TestException(e)
except (RemoteDisconnected, URLError) as e:
raise TestException(e)
def test_api(url):
""" Additional HTTP API compatibility test. """
try:
with urlopen(url, timeout=5) as f:
pass # NOP
except (RemoteDisconnected, URLError, HTTPError) as e: except (RemoteDisconnected, URLError, HTTPError) as e:
raise TestException(e) raise TestException(e)
raise HttpApiException("HTTP API is not supported yet for this receiver!")
def init_auth(user, password, url, use_ssl=False): def init_auth(user, password, url, use_ssl=False):
@@ -381,9 +378,12 @@ def test_telnet(host, port, user, password, timeout=5):
try: try:
gen = telnet_test(host, port, user, password, timeout) gen = telnet_test(host, port, user, password, timeout)
res = next(gen) res = next(gen)
print(res) msg = str(res, encoding="utf8").strip()
res = next(gen) log(msg)
return res next(gen)
if re.search("password", msg, re.IGNORECASE):
raise TestException(msg)
return msg
except (socket.timeout, OSError) as e: except (socket.timeout, OSError) as e:
raise TestException(e) raise TestException(e)
@@ -392,14 +392,14 @@ def telnet_test(host, port, user, password, timeout):
tn = Telnet(host=host, port=port, timeout=timeout) tn = Telnet(host=host, port=port, timeout=timeout)
time.sleep(1) time.sleep(1)
tn.read_until(b"login: ", timeout=2) tn.read_until(b"login: ", timeout=2)
tn.write(user.encode("utf-8") + b"\n") tn.write(user.encode("utf-8") + b"\r")
time.sleep(timeout) time.sleep(timeout)
tn.read_until(b"Password: ", timeout=2) tn.read_until(b"Password: ", timeout=2)
tn.write(password.encode("utf-8") + b"\n") tn.write(password.encode("utf-8") + b"\r")
time.sleep(timeout) time.sleep(timeout)
yield tn.read_very_eager() yield tn.read_very_eager()
tn.close() tn.close()
yield "Done!" yield
if __name__ == "__main__": if __name__ == "__main__":

83
app/ui/main_app_window.py
View File

@@ -1,6 +1,7 @@
import os import os
import sys import sys
from contextlib import suppress from contextlib import suppress
from datetime import datetime
from functools import lru_cache from functools import lru_cache
from itertools import chain from itertools import chain
@@ -232,6 +233,7 @@ class Application(Gtk.Application):
self._save_header_button.bind_property("sensitive", builder.get_object("save_menu_button"), "sensitive") self._save_header_button.bind_property("sensitive", builder.get_object("save_menu_button"), "sensitive")
self._signal_level_bar.bind_property("visible", builder.get_object("play_current_service_button"), "visible") self._signal_level_bar.bind_property("visible", builder.get_object("play_current_service_button"), "visible")
self._receiver_info_box.bind_property("visible", self._http_status_image, "visible", 4) self._receiver_info_box.bind_property("visible", self._http_status_image, "visible", 4)
self._receiver_info_box.bind_property("visible", builder.get_object("signal_box"), "visible")
# Force ctrl press event for view. Multiple selections in lists only with Space key(as in file managers)!!! # Force ctrl press event for view. Multiple selections in lists only with Space key(as in file managers)!!!
self._services_view.connect("key-press-event", self.force_ctrl) self._services_view.connect("key-press-event", self.force_ctrl)
self._fav_view.connect("key-press-event", self.force_ctrl) self._fav_view.connect("key-press-event", self.force_ctrl)
@@ -267,6 +269,7 @@ class Application(Gtk.Application):
self._player_box.bind_property("visible", builder.get_object("main_popover_menu_box"), "visible", 4) self._player_box.bind_property("visible", builder.get_object("main_popover_menu_box"), "visible", 4)
self._player_box.bind_property("visible", builder.get_object("download_header_button"), "visible", 4) self._player_box.bind_property("visible", builder.get_object("download_header_button"), "visible", 4)
self._player_box.bind_property("visible", builder.get_object("left_header_separator"), "visible", 4) self._player_box.bind_property("visible", builder.get_object("left_header_separator"), "visible", 4)
self._player_box.bind_property("visible", self._profile_combo_box, "sensitive", 4)
# Enabling events for the drawing area # Enabling events for the drawing area
self._player_drawing_area.set_events(Gdk.ModifierType.BUTTON1_MASK) self._player_drawing_area.set_events(Gdk.ModifierType.BUTTON1_MASK)
self._player_frame = builder.get_object("player_frame") self._player_frame = builder.get_object("player_frame")
@@ -1203,11 +1206,9 @@ class Application(Gtk.Application):
self._s_type = self._settings.setting_type self._s_type = self._settings.setting_type
self._profile_combo_box.set_tooltip_text(self._profile_combo_box.get_tooltip_text() + self._settings.host) self._profile_combo_box.set_tooltip_text(self._profile_combo_box.get_tooltip_text() + self._settings.host)
self.update_profile_label() self.update_profile_label()
gen = self.init_http_api()
if self._http_api and self._settings.http_api_support: GLib.idle_add(lambda: next(gen, False), priority=GLib.PRIORITY_LOW)
self._http_api.init() self.open_data()
self.open_data()
def update_profiles(self): def update_profiles(self):
self._profile_combo_box.remove_all() self._profile_combo_box.remove_all()
@@ -1600,11 +1601,11 @@ class Application(Gtk.Application):
def on_player_previous(self, item): def on_player_previous(self, item):
if self._fav_view.do_move_cursor(self._fav_view, Gtk.MovementStep.DISPLAY_LINES, -1): if self._fav_view.do_move_cursor(self._fav_view, Gtk.MovementStep.DISPLAY_LINES, -1):
self.on_play_stream() self.on_zap(self.on_watch) if self._fav_click_mode is FavClickMode.PLAY else self.on_play_stream()
def on_player_next(self, item): def on_player_next(self, item):
if self._fav_view.do_move_cursor(self._fav_view, Gtk.MovementStep.DISPLAY_LINES, 1): if self._fav_view.do_move_cursor(self._fav_view, Gtk.MovementStep.DISPLAY_LINES, 1):
self.on_play_stream() self.on_zap(self.on_watch) if self._fav_click_mode is FavClickMode.PLAY else self.on_play_stream()
def on_player_rewind(self, scale, scroll_type, value): def on_player_rewind(self, scale, scroll_type, value):
self._player.set_time(int(value)) self._player.set_time(int(value))
@@ -1680,17 +1681,16 @@ class Application(Gtk.Application):
if not state & Gdk.WindowState.ICONIFIED and self._links_transmitter: if not state & Gdk.WindowState.ICONIFIED and self._links_transmitter:
self._links_transmitter.hide() self._links_transmitter.hide()
# ************************ HTTP API ****************************# # ************************ HTTP API **************************** #
def init_http_api(self): def init_http_api(self):
self._fav_click_mode = FavClickMode(self._settings.fav_click_mode) self._fav_click_mode = FavClickMode(self._settings.fav_click_mode)
http_api_enable = self._settings.http_api_support http_api_enable = self._settings.http_api_support
status = all( st = all((http_api_enable, self._s_type is SettingsType.ENIGMA_2, not self._receiver_info_box.get_visible()))
(http_api_enable, self._s_type is SettingsType.ENIGMA_2, not self._receiver_info_box.get_visible())) GLib.idle_add(self._http_status_image.set_visible, st)
GLib.idle_add(self._http_status_image.set_visible, status)
if self._s_type is SettingsType.NEUTRINO_MP or not http_api_enable: if self._s_type is SettingsType.NEUTRINO_MP or not http_api_enable:
GLib.idle_add(self.update_info_boxes_visible, False) GLib.idle_add(self._receiver_info_box.set_visible, False)
if self._http_api: if self._http_api:
self._http_api.close() self._http_api.close()
yield True yield True
@@ -1701,6 +1701,8 @@ class Application(Gtk.Application):
if not self._http_api: if not self._http_api:
self._http_api = HttpAPI(self._settings) self._http_api = HttpAPI(self._settings)
GLib.timeout_add_seconds(3, self.update_info, priority=GLib.PRIORITY_LOW) GLib.timeout_add_seconds(3, self.update_info, priority=GLib.PRIORITY_LOW)
else:
self._http_api.init()
self.init_send_to(http_api_enable and self._settings.enable_send_to) self.init_send_to(http_api_enable and self._settings.enable_send_to)
yield True yield True
@@ -1738,7 +1740,7 @@ class Application(Gtk.Application):
ref = srv.picon_id.rstrip(".png").replace("_", ":") ref = srv.picon_id.rstrip(".png").replace("_", ":")
def zap(rq): def zap(rq):
if rq and rq.get("result", False): if rq and rq.get("e2state", False):
GLib.idle_add(scroll_to, path, self._fav_view) GLib.idle_add(scroll_to, path, self._fav_view)
if callback is not None: if callback is not None:
callback() callback()
@@ -1747,49 +1749,51 @@ class Application(Gtk.Application):
def update_info(self): def update_info(self):
""" Updating current info over HTTP API """ """ Updating current info over HTTP API """
if not self._http_api: if not self._http_api or self._s_type is SettingsType.NEUTRINO_MP:
GLib.idle_add(self._http_status_image.set_visible, False) GLib.idle_add(self._http_status_image.set_visible, False)
GLib.idle_add(self._receiver_info_box.set_visible, False)
return False return False
self._http_api.send(HttpRequestType.INFO, None, self.update_receiver_info) self._http_api.send(HttpRequestType.INFO, None, self.update_receiver_info)
self._http_api.send(HttpRequestType.INFO, None, self.update_service_info)
return True return True
def update_receiver_info(self, info): def update_receiver_info(self, info):
res_info = info.get("info", None) if info else None res_info = info.get("e2about", None) if info else None
if res_info: if res_info:
image = res_info.get("friendlyimagedistro", "") image = info.get("e2distroversion", "")
image_ver = res_info.get("imagever", "") image_ver = info.get("e2imageversion", "")
brand = res_info.get("brand", "") model = info.get("e2model", "")
model = res_info.get("model", "") info_text = "{} Image: {} {}".format(model, image, image_ver)
info_text = "{} {} Image: {} {}".format(brand, model, image, image_ver)
GLib.idle_add(self._receiver_info_label.set_text, info_text) GLib.idle_add(self._receiver_info_label.set_text, info_text)
GLib.idle_add(self._service_name_label.set_text, info.get("e2servicename", None) or "")
self.update_service_info(info)
GLib.idle_add(self._receiver_info_box.set_visible, bool(res_info)) GLib.idle_add(self._receiver_info_box.set_visible, bool(res_info))
@run_idle
def update_service_info(self, info): def update_service_info(self, info):
service_info = info.get("service", None) if info else None has_onid = info.get("e2onid", "N/A") != "N/A"
if service_info: if has_onid and self._http_api:
GLib.idle_add(self._service_name_label.set_text, service_info.get("name", "")) self._http_api.send(HttpRequestType.SIGNAL, None, self.update_signal)
if service_info.get("onid", None) and self._http_api: self._http_api.send(HttpRequestType.CURRENT, None, self.update_status)
self._http_api.send(HttpRequestType.SIGNAL, None, self.update_signal) self._signal_level_bar.set_visible(has_onid)
self._http_api.send(HttpRequestType.STATUS, None, self.update_status)
GLib.idle_add(self._signal_box.set_visible, bool(service_info))
def update_signal(self, sig): def update_signal(self, sig):
self.set_signal(sig.get("snr", 0) if sig else 0) self.set_signal(sig.get("e2snr", "0 %") if sig else "0 %")
@lru_cache(maxsize=2) @lru_cache(maxsize=2)
def set_signal(self, val): def set_signal(self, val):
self._signal_level_bar.set_value(val if isinstance(val, int) else 0) self._signal_level_bar.set_value(int(val.rstrip("%").strip() or 0))
self._signal_level_bar.set_visible(val)
def update_status(self, status): def update_status(self, evn):
if status: if evn:
dsc = "{} {} - {}".format(status.get("currservice_name", ""), s_duration = int(evn.get("e2eventstart", "0"))
status.get("currservice_begin", ""), s_time = datetime.fromtimestamp(s_duration)
status.get("currservice_end", "")) end_time = datetime.fromtimestamp(s_duration + int(evn.get("e2eventduration", "0")))
dsc = "{} {}:{} - {}:{}".format(evn.get("e2eventtitle", ""),
s_time.hour, s_time.minute,
end_time.hour, end_time.minute)
self._service_epg_label.set_text(dsc) self._service_epg_label.set_text(dsc)
self._service_epg_label.set_tooltip_text(status.get("currservice_description", "")) self._service_epg_label.set_tooltip_text(evn.get("e2eventdescription", ""))
# ***************** Filter and search *********************# # ***************** Filter and search *********************#
@@ -2108,11 +2112,6 @@ class Application(Gtk.Application):
def get_format_version(self): def get_format_version(self):
return 5 if self._settings.v5_support else 4 return 5 if self._settings.v5_support else 4
@run_idle
def update_info_boxes_visible(self, visible):
self._signal_box.set_visible(visible)
self._receiver_info_box.set_visible(visible)
@run_idle @run_idle
def show_error_dialog(self, message): def show_error_dialog(self, message):
show_dialog(DialogType.ERROR, self._main_window, message) show_dialog(DialogType.ERROR, self._main_window, message)

2
app/ui/transmitter.py
View File

@@ -88,7 +88,7 @@ class LinksTransmitter:
def on_play(self, res): def on_play(self, res):
""" Play callback """ """ Play callback """
GLib.idle_add(self._url_entry.set_sensitive, True) GLib.idle_add(self._url_entry.set_sensitive, True)
res = res.get("result", None) if res else res res = res.get("e2state", None) if res else res
self._url_entry.set_name("GtkEntry" if res else "digit-entry") self._url_entry.set_name("GtkEntry" if res else "digit-entry")
def on_exit(self, item=None): def on_exit(self, item=None):