diff --git a/usr/lib/hypnotix/hypnotix.py b/usr/lib/hypnotix/hypnotix.py
index 9612aa94..03d14d2a 100755
--- a/usr/lib/hypnotix/hypnotix.py
+++ b/usr/lib/hypnotix/hypnotix.py
@@ -27,12 +27,14 @@
import mpv
import requests
import setproctitle
-from imdb import IMDb
+from imdb import IMDb, IMDbError
from unidecode import unidecode
from common import Manager, Provider, Channel, MOVIES_GROUP, PROVIDERS_PATH, SERIES_GROUP, TV_GROUP,\
async_function, idle_function
+# Load xtream class
+from xtream import XTream
setproctitle.setproctitle("hypnotix")
@@ -250,6 +252,8 @@ def __init__(self, application):
"referer_entry",
"mpv_entry",
"mpv_link",
+ "adult_switch",
+ "empty_groups_switch",
"ytdlp_local_switch",
"ytdlp_system_version_label",
"ytdlp_local_version_label",
@@ -287,6 +291,7 @@ def __init__(self, application):
# Widget signals
self.window.connect("key-press-event", self.on_key_press_event)
+ self.window.connect("realize", self.on_window_realize)
self.mpv_drawing_area.connect("realize", self.on_mpv_drawing_area_realize)
self.mpv_drawing_area.connect("draw", self.on_mpv_drawing_area_draw)
self.fullscreen_button.connect("clicked", self.on_fullscreen_button_clicked)
@@ -359,6 +364,18 @@ def __init__(self, application):
except Exception:
pass
+ # hide adult stream
+ self.prefer_hide_adult = self.adult_switch.get_active()
+ self.prefer_hide_adult = self.settings.get_boolean("prefer-hide-adult")
+ self.adult_switch.set_active(self.prefer_hide_adult)
+ self.adult_switch.connect("notify::active", self.on_hide_adult_switch_toggled)
+
+ # hide empty groups stream
+ self.prefer_hide_empty_groups = self.empty_groups_switch.get_active()
+ self.prefer_hide_empty_groups = self.settings.get_boolean("prefer-hide-empty-groups")
+ self.empty_groups_switch.set_active(self.prefer_hide_empty_groups)
+ self.empty_groups_switch.connect("notify::active", self.on_hide_empty_groups_switch_toggled)
+
# Menubar
accel_group = Gtk.AccelGroup()
self.window.add_accel_group(accel_group)
@@ -413,12 +430,11 @@ def __init__(self, application):
self.movies_logo.set_from_surface(self.get_surface_for_file("/usr/share/hypnotix/pictures/movies.svg", 258, 258))
self.series_logo.set_from_surface(self.get_surface_for_file("/usr/share/hypnotix/pictures/series.svg", 258, 258))
- self.reload(page="landing_page")
-
# Redownload playlists by default
# This is going to get readjusted
self._timerid = GLib.timeout_add_seconds(self.reload_timeout_sec, self.force_reload)
+ self.current_cursor = None
self.window.show()
self.playback_bar.hide()
self.search_bar.hide()
@@ -476,8 +492,21 @@ def show_groups(self, widget, content_type):
self.active_group = None
found_groups = False
for group in self.active_provider.groups:
- if group.group_type != self.content_type:
+ # Skip if the group is not from the current displayed content type
+ if (group.group_type != self.content_type):
continue
+ if self.prefer_hide_empty_groups:
+ # Skip group with empty channels
+ if (self.content_type != SERIES_GROUP) and (len(group.channels) == 0):
+ continue
+ # Skip group with empty series
+ if (self.content_type == SERIES_GROUP) and (len(group.series) == 0):
+ continue
+ # Check if need to skip channels marked as adult in TV and Movies groups
+ if self.prefer_hide_adult:
+ if (self.content_type != SERIES_GROUP) and (hasattr(group.channels[0], "is_adult")):
+ if (group.channels[0].is_adult == 1):
+ continue
found_groups = True
button = Gtk.Button()
button.connect("clicked", self.on_category_button_clicked, group)
@@ -604,7 +633,7 @@ def show_episodes(self, serie):
# If we are using xtream provider
# Load every Episodes of every Season for this Series
if self.active_provider.type_id == "xtream":
- self.x.get_series_info_by_id(self.active_serie)
+ serie.xtream.get_series_info_by_id(self.active_serie)
self.navigate_to("episodes_page")
for child in self.episodes_box.get_children():
@@ -659,6 +688,13 @@ def bind_setting_widget(self, key, widget):
def on_entry_changed(self, widget, key):
self.settings.set_string(key, widget.get_text())
+ def on_hide_adult_switch_toggled(self, widget, key):
+ self.prefer_hide_adult = widget.get_active()
+ self.settings.set_boolean("prefer-hide-adult", self.prefer_hide_adult)
+
+ def on_hide_empty_groups_switch_toggled(self, widget, key):
+ self.prefer_hide_empty_groups = widget.get_active()
+ self.settings.set_boolean("prefer-hide-empty-groups", self.prefer_hide_empty_groups)
def on_ytdlp_local_switch_activated(self, widget, data=None):
self.settings.set_boolean("use-local-ytdlp", widget.get_active())
if widget.get_active():
@@ -900,7 +936,6 @@ def on_next_channel(self):
self.channels_listbox.do_move_cursor(self.channels_listbox, Gtk.MovementStep.DISPLAY_LINES, 1)
self.channels_listbox.do_activate_cursor_row(self.channels_listbox)
- @async_function
def play_async(self, channel):
print("CHANNEL: '%s' (%s)" % (channel.name, channel.url))
if channel is not None and channel.url is not None:
@@ -910,8 +945,15 @@ def play_async(self, channel):
self.before_play(channel)
self.reinit_mpv()
self.mpv.play(channel.url)
+ self.wait_for_mpv_playing(channel)
+
+ @async_function
+ def wait_for_mpv_playing(self, channel):
+ try:
self.mpv.wait_until_playing()
- self.after_play(channel)
+ except mpv.ShutdownError:
+ pass
+ self.after_play(channel)
@idle_function
def before_play(self, channel):
@@ -1043,7 +1085,11 @@ def on_audio_codec(self, property, codec):
@async_function
def get_imdb_details(self, name):
- movies = self.ia.search_movie(name)
+ movies = []
+ try:
+ movies = self.ia.search_movie(name)
+ except IMDbError:
+ print("IMDB Redirect Error will be fixed in IMDbPY latest version")
match = None
for movie in movies:
self.ia.update(movie)
@@ -1507,6 +1553,7 @@ def close(w, res):
dlg.show()
def on_menu_quit(self, widget):
+ self.mpv.terminate()
self.application.quit()
def on_key_press_event(self, widget, event):
@@ -1534,24 +1581,31 @@ def on_key_press_event(self, widget, event):
self.on_prev_channel()
elif event.keyval == Gdk.KEY_Right:
self.on_next_channel()
+ elif event.keyval == Gdk.KEY_Escape:
+ # Go back one level
+ self.on_go_back_button(widget)
# elif event.keyval == Gdk.KEY_Up:
- # # Up of in the list
- # pass
+ # Up of in the list
+ # print("UP")
+ # pass
# elif event.keyval == Gdk.KEY_Down:
- # # Down of in the list
- # pass
- # elif event.keyval == Gdk.KEY_Escape:
- # # Go back one level
+ # Down of in the list
+ # print("DOWN")
+ # pass
+ #elif event.keyval == Gdk.KEY_Return:
+ # Same as click
# pass
- # #elif event.keyval == Gdk.KEY_Return:
- # # Same as click
- # # pass
@async_function
def reload(self, page=None, refresh=False):
self.favorite_data = self.manager.load_favorites()
self.status(_("Loading providers..."))
+ self.start_loading_cursor()
self.providers = []
+ headers = {
+ 'User-Agent': self.settings.get_string("user-agent"),
+ 'Referer': self.settings.get_string("http-referer")
+ }
for provider_info in self.settings.get_strv("providers"):
try:
provider = Provider(name=None, provider_info=provider_info)
@@ -1581,43 +1635,40 @@ def reload(self, page=None, refresh=False):
self.status(_("Failed to download playlist from %s") % provider.name, provider)
else:
- # Load xtream class
- from xtream import XTream
-
# Download via Xtream
- self.x = XTream(
+ x = XTream(
+ self.status,
provider.name,
provider.username,
provider.password,
provider.url,
- hide_adult_content=False,
+ headers=headers,
+ hide_adult_content=self.prefer_hide_adult,
cache_path=PROVIDERS_PATH,
)
- if self.x.auth_data != {}:
- print("XTREAM `{}` Loading Channels".format(provider.name))
- # Save default cursor
- current_cursor = self.window.get_window().get_cursor()
- # Set waiting cursor
- self.window.get_window().set_cursor(Gdk.Cursor.new_from_name(Gdk.Display.get_default(), "wait"))
+ if x.auth_data != {}:
+ self.status("Loading Channels...", provider)
# Load data
- self.x.load_iptv()
- # Restore default cursor
- self.window.get_window().set_cursor(current_cursor)
- # Inform Provider of data
- provider.channels = self.x.channels
- provider.movies = self.x.movies
- provider.series = self.x.series
- provider.groups = self.x.groups
-
- # Change redownload timeout
- self.reload_timeout_sec = 60 * 60 * 2 # 2 hours
- if self._timerid:
- GLib.source_remove(self._timerid)
- self._timerid = GLib.timeout_add_seconds(self.reload_timeout_sec, self.force_reload)
-
- # If no errors, approve provider
- if provider.name == self.settings.get_string("active-provider"):
- self.active_provider = provider
+ x.load_iptv()
+ # If there are no stream to show, pass this provider.
+ if (len(x.channels) == 0) and (len(x.movies) == 0) and (len(x.series) == 0) and (len(x.groups) == 0):
+ pass
+ else:
+ # Inform Provider of data
+ provider.channels = x.channels
+ provider.movies = x.movies
+ provider.series = x.series
+ provider.groups = x.groups
+
+ # Change redownload timeout
+ self.reload_timeout_sec = 60 * 60 * 4 # 4 hours
+ if self._timerid:
+ GLib.source_remove(self._timerid)
+ self._timerid = GLib.timeout_add_seconds(self.reload_timeout_sec, self.force_reload)
+
+ # If no errors, approve provider
+ if provider.name == self.settings.get_string("active-provider"):
+ self.active_provider = provider
self.status(None)
else:
print("XTREAM Authentication Failed")
@@ -1638,12 +1689,26 @@ def reload(self, page=None, refresh=False):
self.status(None)
self.latest_search_bar_text = None
+ self.end_loading_cursor()
+
+ @idle_function
+ def start_loading_cursor(self):
+ # Restore default cursor
+ self.current_cursor = self.window.get_window().get_cursor()
+ self.window.get_window().set_cursor(Gdk.Cursor.new_from_name(Gdk.Display.get_default(), "wait"))
+
+ @idle_function
+ def end_loading_cursor(self):
+ # Restore default cursor
+ self.window.get_window().set_cursor(self.current_cursor)
+ self.current_cursor = None
+
def force_reload(self):
self.reload(page=None, refresh=True)
return False
@idle_function
- def status(self, string, provider=None):
+ def status(self, string, provider=None, guiOnly=False):
if string is None:
self.status_label.set_text("")
self.status_label.hide()
@@ -1651,17 +1716,22 @@ def status(self, string, provider=None):
self.status_label.show()
if provider is not None:
self.status_label.set_text("%s: %s" % (provider.name, string))
- print("%s: %s" % (provider.name, string))
+ if not guiOnly:
+ print("%s: %s" % (provider.name, string))
else:
self.status_label.set_text(string)
- print(string)
+ if not guiOnly:
+ print(string)
def on_mpv_drawing_area_realize(self, widget):
self.reinit_mpv()
+ def on_window_realize(self, widget):
+ self.reload(page="landing_page")
+
def reinit_mpv(self):
if self.mpv is not None:
- self.mpv.stop()
+ self.mpv.quit()
options = {}
try:
mpv_options = self.settings.get_string("mpv-options")
@@ -1677,14 +1747,17 @@ def reinit_mpv(self):
options["user_agent"] = self.settings.get_string("user-agent")
options["referrer"] = self.settings.get_string("http-referer")
- while not self.mpv_drawing_area.get_window() and not Gtk.events_pending():
- time.sleep(0.1)
-
osc = True
if "osc" in options:
# To prevent 'multiple values for keyword argument'!
osc = options.pop("osc") != "no"
+ # This is a race - depending on whether you do tv shows or movies, the drawing area may or may not
+ # already be realized - just make sure it's done by this point so there's a window to give to the mpv
+ # initializer.
+ if not self.mpv_drawing_area.get_realized():
+ self.mpv_drawing_area.realize()
+
self.mpv = mpv.MPV(
**options,
script_opts="osc-layout=box,osc-seekbarstyle=bar,osc-deadzonesize=0,osc-minmousemove=3",
diff --git a/usr/lib/hypnotix/xtream.py b/usr/lib/hypnotix/xtream.py
index 9da1532c..9a317f9e 100644
--- a/usr/lib/hypnotix/xtream.py
+++ b/usr/lib/hypnotix/xtream.py
@@ -16,22 +16,26 @@
"""
-__version__ = "0.5.0"
+__version__ = "0.6.0"
__author__ = "Claudio Olmi"
import json
import re # used for URL validation
import time
-from os import path as osp
+import sys
from os import makedirs
+from os import path as osp
+from sys import stdout
from timeit import default_timer as timer # Timing xtream json downloads
-from typing import List, Tuple
+from typing import List, Protocol, Tuple, Optional
+from datetime import datetime, timedelta
import requests
class Channel:
# Required by Hypnotix
+ info = ""
id = ""
name = "" # What is the difference between the below name and title?
logo = ""
@@ -54,13 +58,11 @@ class Channel:
def __init__(self, xtream: object, group_title, stream_info):
stream_type = stream_info["stream_type"]
# Adjust the odd "created_live" type
- if stream_type == "created_live" or stream_type == "radio_streams":
+ if stream_type in ("created_live", "radio_streams"):
stream_type = "live"
- if stream_type != "live" and stream_type != "movie":
- print("Error the channel has unknown stream type `{}`\n`{}`".format(
- stream_type, stream_info
- ))
+ if stream_type not in ("live", "movie"):
+ print(f"Error the channel has unknown stream type `{stream_type}`\n`{stream_info}`")
else:
# Raw JSON Channel
self.raw = stream_info
@@ -98,18 +100,15 @@ def __init__(self, xtream: object, group_title, stream_info):
stream_extension = stream_info["container_extension"]
# Required by Hypnotix
- self.url = "{}/{}/{}/{}/{}.{}".format(
- xtream.server,
- stream_info["stream_type"],
- xtream.authorization["username"],
- xtream.authorization["password"],
- stream_info["stream_id"],
- stream_extension,
- )
+ self.url = f"{xtream.server}/{stream_type}/{xtream.authorization['username']}/" \
+ f"{xtream.authorization['password']}/{stream_info['stream_id']}.{stream_extension}"
# Check that the constructed URL is valid
if not xtream._validate_url(self.url):
- print("{} - Bad URL? `{}`".format(self.name, self.url))
+ print(f"{self.name} - Bad URL? `{self.url}`")
+
+ # Add Channel info in M3U8 format to support Favorite Channel
+ self.info = f'#EXTINF:-1 tvg-name="{self.name}" tvg-logo="{self.logo}" group-title="{self.group_title}",{self.name}'
def export_json(self):
jsondata = {}
@@ -145,12 +144,10 @@ def __init__(self, group_info: dict, stream_type: str):
self.group_type = MOVIES_GROUP
elif "Series" == stream_type:
self.group_type = SERIES_GROUP
- elif "Live":
+ elif "Live" == stream_type:
self.group_type = TV_GROUP
else:
- print("Unrecognized stream type `{}` for `{}`".format(
- stream_type, group_info
- ))
+ print(f"Unrecognized stream type `{stream_type}` for `{group_info}`")
self.name = group_info["category_name"]
@@ -163,6 +160,7 @@ class Episode:
# Required by Hypnotix
title = ""
name = ""
+ info = ""
# XTream
@@ -184,17 +182,13 @@ def __init__(self, xtream: object, series_info, group_title, episode_info) -> No
self.logo = series_info["cover"]
self.logo_path = xtream._get_logo_local_path(self.logo)
- self.url = "{}/series/{}/{}/{}.{}".format(
- xtream.server,
- xtream.authorization["username"],
- xtream.authorization["password"],
- self.id,
- self.container_extension,
- )
+ self.url = f"{xtream.server}/series/" \
+ f"{xtream.authorization['username']}/" \
+ f"{xtream.authorization['password']}/{self.id}.{self.container_extension}"
# Check that the constructed URL is valid
if not xtream._validate_url(self.url):
- print("{} - Bad URL? `{}`".format(self.name, self.url))
+ print(f"{self.name} - Bad URL? `{self.url}`")
class Serie:
@@ -215,6 +209,7 @@ class Serie:
def __init__(self, xtream: object, series_info):
# Raw JSON Series
self.raw = series_info
+ self.xtream = xtream
# Required by Hypnotix
self.name = series_info["name"]
@@ -249,13 +244,22 @@ def __init__(self, name):
self.name = name
self.episodes = {}
+class MyStatus(Protocol):
+ def __call__(self, string: str, guiOnly: bool) -> None: ...
class XTream:
name = ""
server = ""
+ secure_server = ""
username = ""
password = ""
+ base_url = ""
+ base_url_ssl = ""
+
+ cache_path = ""
+
+ account_expiration: timedelta
live_type = "Live"
vod_type = "VOD"
@@ -269,17 +273,20 @@ class XTream:
series = []
movies = []
- state = {"authenticated": False, "loaded": False}
+ connection_headers = {}
+
+ state = {'authenticated': False, 'loaded': False}
hide_adult_content = False
- catch_all_group = Group(
- {
- "category_id": "9999",
- "category_name":"xEverythingElse",
- "parent_id":0
- },
- ""
+ live_catch_all_group = Group(
+ {"category_id": "9999", "category_name":"xEverythingElse", "parent_id":0}, live_type
+ )
+ vod_catch_all_group = Group(
+ {"category_id": "9999", "category_name":"xEverythingElse", "parent_id":0}, vod_type
+ )
+ series_catch_all_group = Group(
+ {"category_id": "9999", "category_name":"xEverythingElse", "parent_id":0}, series_type
)
# If the cached JSON file is older than threshold_time_sec then load a new
# JSON dictionary from the provider
@@ -287,12 +294,14 @@ class XTream:
def __init__(
self,
+ update_status: MyStatus,
provider_name: str,
provider_username: str,
provider_password: str,
provider_url: str,
+ headers: dict = None,
hide_adult_content: bool = False,
- cache_path: str = "",
+ cache_path: str = ""
):
"""Initialize Xtream Class
@@ -301,8 +310,9 @@ def __init__(
provider_username (str): User name of the IPTV provider
provider_password (str): Password of the IPTV provider
provider_url (str): URL of the IPTV provider
- hide_adult_content(bool): When `True` hide stream that are marked for adult
- cache_path (str, optional): Location where to save loaded files. Defaults to empty string.
+ headers (dict): Requests Headers
+ hide_adult_content(bool, optional): When `True` hide stream that are marked for adult
+ cache_path (str, optional): Location where to save loaded files. Defaults to empty string
Returns: XTream Class Instance
@@ -316,13 +326,14 @@ def __init__(
self.name = provider_name
self.cache_path = cache_path
self.hide_adult_content = hide_adult_content
+ self.update_status = update_status
# if the cache_path is specified, test that it is a directory
if self.cache_path != "":
# If the cache_path is not a directory, clear it
if not osp.isdir(self.cache_path):
print(" - Cache Path is not a directory, using default '~/.xtream-cache/'")
- self.cache_path == ""
+ self.cache_path = ""
# If the cache_path is still empty, use default
if self.cache_path == "":
@@ -330,6 +341,11 @@ def __init__(
if not osp.isdir(self.cache_path):
makedirs(self.cache_path, exist_ok=True)
+ if headers is not None:
+ self.connection_headers = headers
+ else:
+ self.connection_headers = {'User-Agent':"Wget/1.20.3 (linux-gnu)"}
+
self.authenticate()
def search_stream(self, keyword: str, ignore_case: bool = True, return_type: str = "LIST") -> List:
@@ -351,24 +367,24 @@ def search_stream(self, keyword: str, ignore_case: bool = True, return_type: str
else:
regex = re.compile(keyword)
- print("Checking {} movies".format(len(self.movies)))
+ print(f"Checking {len(self.movies)} movies")
for stream in self.movies:
if re.match(regex, stream.name) is not None:
search_result.append(stream.export_json())
- print("Checking {} channels".format(len(self.channels)))
+ print(f"Checking {len(self.channels)} channels")
for stream in self.channels:
if re.match(regex, stream.name) is not None:
search_result.append(stream.export_json())
- print("Checking {} series".format(len(self.series)))
+ print(f"Checking {len(self.series)} series")
for stream in self.series:
if re.match(regex, stream.name) is not None:
search_result.append(stream.export_json())
if return_type == "JSON":
if search_result is not None:
- print("Found {} results `{}`".format(len(search_result), keyword))
+ print(f"Found {len(search_result)} results `{keyword}`")
return json.dumps(search_result, ensure_ascii=False)
else:
return search_result
@@ -414,10 +430,9 @@ def _get_logo_local_path(self, logo_url: str) -> str:
if not self._validate_url(logo_url):
logo_url = None
else:
- local_logo_path = osp.join(self.cache_path, "{}-{}".format(
- self._slugify(self.name),
- self._slugify(osp.split(logo_url)[-1])
- )
+ local_logo_path = osp.join(
+ self.cache_path,
+ f"{self._slugify(self.name)}-{self._slugify(osp.split(logo_url)[-1])}"
)
return local_logo_path
@@ -427,22 +442,50 @@ def authenticate(self):
if self.state["authenticated"] is False:
# Erase any previous data
self.auth_data = {}
- try:
- # Request authentication, wait 4 seconds maximum
- r = requests.get(self.get_authenticate_URL(), timeout=(4))
+ # Loop through 30 seconds
+ i = 0
+ r = None
+ # Prepare the authentication url
+ url = f"{self.server}/player_api.php?username={self.username}&password={self.password}"
+ print("Attempting connection... ", end='')
+ while i < 30:
+ try:
+ # Request authentication, wait 4 seconds maximum
+ r = requests.get(url, timeout=(4), headers=self.connection_headers)
+ i = 31
+ except requests.exceptions.ConnectionError:
+ time.sleep(1)
+ print(f"{i} ", end='',flush=True)
+ i += 1
+
+ if r is not None:
# If the answer is ok, process data and change state
if r.ok:
+ print("Connected")
self.auth_data = r.json()
self.authorization = {
"username": self.auth_data["user_info"]["username"],
- "password": self.auth_data["user_info"]["password"],
+ "password": self.auth_data["user_info"]["password"]
}
+ # Account expiration date
+ self.account_expiration = timedelta(
+ seconds=(
+ int(self.auth_data["user_info"]["exp_date"])-datetime.now().timestamp()
+ )
+ )
+ # Mark connection authorized
self.state["authenticated"] = True
+ # Construct the base url for all requests
+ self.base_url = f"{self.server}/player_api.php?username={self.username}&password={self.password}"
+ # If there is a secure server connection, construct the base url SSL for all requests
+ if "https_port" in self.auth_data["server_info"]:
+ self.base_url_ssl = f"https://{self.auth_data['server_info']['url']}:{self.auth_data['server_info']['https_port']}" \
+ f"/player_api.php?username={self.username}&password={self.password}"
+ print(f"Account expires in {str(self.account_expiration)}")
else:
- print("Provider `{}` could not be loaded. Reason: `{} {}`".format(self.name, r.status_code, r.reason))
- except requests.exceptions.ConnectionError:
- # If connection refused
- print("{} - Connection refused URL: {}".format(self.name, self.server))
+ self.update_status(f"{self.name}: Provider could not be loaded. Reason: `{r.status_code} {r.reason}`")
+ else:
+ self.update_status(f"{self.name}: Provider refused the connection")
def _load_from_file(self, filename) -> dict:
"""Try to load the dictionary from file
@@ -454,21 +497,19 @@ def _load_from_file(self, filename) -> dict:
dict: Dictionary if found and no errors, None if file does not exists
"""
# Build the full path
- full_filename = osp.join(self.cache_path, "{}-{}".format(
- self._slugify(self.name),
- filename
- ))
+ full_filename = osp.join(self.cache_path, f"{self._slugify(self.name)}-{filename}")
+ # If the cached file exists, attempt to load it
if osp.isfile(full_filename):
my_data = None
# Get the enlapsed seconds since last file update
- diff_time = time.time() - osp.getmtime(full_filename)
+ file_age_sec = time.time() - osp.getmtime(full_filename)
# If the file was updated less than the threshold time,
# it means that the file is still fresh, we can load it.
# Otherwise skip and return None to force a re-download
- if self.threshold_time_sec > diff_time:
+ if self.threshold_time_sec > file_age_sec:
# Load the JSON data
try:
with open(full_filename, mode="r", encoding="utf-8") as myfile:
@@ -476,12 +517,10 @@ def _load_from_file(self, filename) -> dict:
if len(my_data) == 0:
my_data = None
except Exception as e:
- print(" - Could not load from file `{}`: e=`{}`".format(
- full_filename, e
- ))
+ print(f" - Could not load from file `{full_filename}`: e=`{e}`")
return my_data
- else:
- return None
+
+ return None
def _save_to_file(self, data_list: dict, filename: str) -> bool:
"""Save a dictionary to file
@@ -498,26 +537,21 @@ def _save_to_file(self, data_list: dict, filename: str) -> bool:
if data_list is not None:
#Build the full path
- full_filename = osp.join(self.cache_path, "{}-{}".format(
- self._slugify(self.name),
- filename
- ))
+ full_filename = osp.join(self.cache_path, f"{self._slugify(self.name)}-{filename}")
# If the path makes sense, save the file
json_data = json.dumps(data_list, ensure_ascii=False)
try:
with open(full_filename, mode="wt", encoding="utf-8") as myfile:
myfile.write(json_data)
except Exception as e:
- print(" - Could not save to file `{}`: e=`{}`".format(
- full_filename, e
- ))
+ print(f" - Could not save to file `{full_filename}`: e=`{e}`")
return False
return True
else:
return False
- def load_iptv(self):
+ def load_iptv(self) -> bool:
"""Load XTream IPTV
- Add all Live TV to XTream.channels
@@ -528,169 +562,213 @@ def load_iptv(self):
- Add all groups to XTream.groups
Groups are for all three channel types, Live TV, VOD, and Series
+ Returns:
+ bool: True if successfull, False if error
"""
- # If pyxtream has already authenticated the connection and not loaded the data, start loading
- if self.state["authenticated"] is True:
- if self.state["loaded"] is False:
-
- for loading_stream_type in (self.live_type, self.vod_type, self.series_type):
- ## Get GROUPS
-
- # Try loading local file
- dt = 0
- all_cat = self._load_from_file("all_groups_{}.json".format(
- loading_stream_type
- ))
- # If file empty or does not exists, download it from remote
- if all_cat is None:
- # Load all Groups and save file locally
- start = timer()
- all_cat = self._load_categories_from_provider(loading_stream_type)
- self._save_to_file(all_cat,"all_groups_{}.json".format(
- loading_stream_type
- ))
- dt = timer() - start
-
- # If we got the GROUPS data, show the statistics and load GROUPS
- if all_cat is not None:
- print("Loaded {} {} Groups in {:.3f} seconds".format(
- len(all_cat), loading_stream_type, dt
- ))
- ## Add GROUPS to dictionaries
-
- # Add the catch-all-errors group
- self.groups.append(self.catch_all_group)
-
- for cat_obj in all_cat:
- # Create Group (Category)
- new_group = Group(cat_obj, loading_stream_type)
- # Add to xtream class
- self.groups.append(new_group)
-
- # Add the catch-all-errors group
- self.groups.append(Group({"category_id": "9999", "category_name": "xEverythingElse", "parent_id": 0}, loading_stream_type))
-
- # Sort Categories
- self.groups.sort(key=lambda x: x.name)
- else:
- print(" - Could not load {} Groups".format(loading_stream_type))
- break
-
- ## Get Streams
-
- # Try loading local file
- dt = 0
- all_streams = self._load_from_file("all_stream_{}.json".format(
- loading_stream_type
- ))
- # If file empty or does not exists, download it from remote
- if all_streams is None:
- # Load all Streams and save file locally
- start = timer()
- all_streams = self._load_streams_from_provider(loading_stream_type)
- self._save_to_file(all_streams,"all_stream_{}.json".format(
- loading_stream_type
- ))
- dt = timer() - start
-
- # If we got the STREAMS data, show the statistics and load Streams
- if all_streams is not None:
- print("Loaded {} {} Streams in {:.3f} seconds".format(
- len(all_streams), loading_stream_type, dt
- ))
- ## Add Streams to dictionaries
-
- skipped_adult_content = 0
- skipped_no_name_content = 0
-
- for stream_channel in all_streams:
- skip_stream = False
-
- # Skip if the name of the stream is empty
- if stream_channel["name"] == "":
- skip_stream = True
- skipped_no_name_content = skipped_no_name_content + 1
- self._save_to_file_skipped_streams(stream_channel)
+ # If pyxtream has not authenticated the connection, return empty
+ if self.state["authenticated"] is False:
+ print("Warning, cannot load steams since authorization failed")
+ return False
- # Skip if the user chose to hide adult streams
- if self.hide_adult_content and loading_stream_type == self.live_type:
- try:
- if stream_channel["is_adult"] == "1":
- skip_stream = True
- skipped_adult_content = skipped_adult_content + 1
- self._save_to_file_skipped_streams(stream_channel)
- except Exception:
- print(" - Stream does not have `is_adult` key:\n\t`{}`".format(json.dumps(stream_channel)))
- pass
-
- if not skip_stream:
- # Some channels have no group,
- # so let's add them to the catch all group
- if not stream_channel["category_id"]:
- stream_channel["category_id"] = "9999"
-
- # Find the first occurence of the group that the
- # Channel or Stream is pointing to
- the_group = next(
- (x for x in self.groups if x.group_id == int(stream_channel["category_id"])),
- None
+ # If pyxtream has already loaded the data, skip and return success
+ if self.state["loaded"] is True:
+ print("Warning, data has already been loaded.")
+ return True
+
+ # Delete skipped channels from cache
+ full_filename = osp.join(self.cache_path, "skipped_streams.json")
+ try:
+ f = open(full_filename, mode="r+", encoding="utf-8")
+ f.truncate(0)
+ f.close()
+ except FileNotFoundError:
+ pass
+
+ for loading_stream_type in (self.live_type, self.vod_type, self.series_type):
+ ## Get GROUPS
+
+ # Try loading local file
+ dt = 0
+ start = timer()
+ all_cat = self._load_from_file(f"all_groups_{loading_stream_type}.json")
+ # If file empty or does not exists, download it from remote
+ if all_cat is None:
+ # Load all Groups and save file locally
+ all_cat = self._load_categories_from_provider(loading_stream_type)
+ if all_cat is not None:
+ self._save_to_file(all_cat,f"all_groups_{loading_stream_type}.json")
+ dt = timer() - start
+
+ # If we got the GROUPS data, show the statistics and load GROUPS
+ if all_cat is not None:
+ self.update_status(
+ f"{self.name}: Loaded {len(all_cat)} {loading_stream_type} Groups in {dt:.3f} seconds"
+ )
+
+ ## Add GROUPS to dictionaries
+
+ # Add the catch-all-errors group
+ if loading_stream_type == self.live_type:
+ self.groups.append(self.live_catch_all_group)
+ elif loading_stream_type == self.vod_type:
+ self.groups.append(self.vod_catch_all_group)
+ elif loading_stream_type == self.series_type:
+ self.groups.append(self.series_catch_all_group)
+
+ for cat_obj in all_cat:
+ # Create Group (Category)
+ new_group = Group(cat_obj, loading_stream_type)
+ # Add to xtream class
+ self.groups.append(new_group)
+
+ # Add the catch-all-errors group
+ self.groups.append(Group({"category_id": "9999", "category_name": "xEverythingElse", "parent_id": 0}, loading_stream_type))
+
+ # Sort Categories
+ self.groups.sort(key=lambda x: x.name)
+ else:
+ print(f" - Could not load {loading_stream_type} Groups")
+ break
+
+ ## Get Streams
+
+ # Try loading local file
+ dt = 0
+ start = timer()
+ all_streams = self._load_from_file(f"all_stream_{loading_stream_type}.json")
+ # If file empty or does not exists, download it from remote
+ if all_streams is None:
+ # Load all Streams and save file locally
+ all_streams = self._load_streams_from_provider(loading_stream_type)
+ self._save_to_file(all_streams,f"all_stream_{loading_stream_type}.json")
+ dt = timer() - start
+
+ # If we got the STREAMS data, show the statistics and load Streams
+ if all_streams is not None:
+ print(
+ f"{self.name}: Loaded {len(all_streams)} {loading_stream_type} Streams " \
+ f"in {dt:.3f} seconds"
+ )
+ ## Add Streams to dictionaries
+
+ skipped_adult_content = 0
+ skipped_no_name_content = 0
+
+ number_of_streams = len(all_streams)
+ current_stream_number = 0
+ # Calculate 1% of total number of streams
+ # This is used to slow down the progress bar
+ one_percent_number_of_streams = number_of_streams/100
+ start = timer()
+ for stream_channel in all_streams:
+ skip_stream = False
+ current_stream_number += 1
+
+ # Show download progress every 1% of total number of streams
+ if current_stream_number < one_percent_number_of_streams:
+ percent = progress(
+ current_stream_number,
+ number_of_streams,
+ f"Processing {loading_stream_type} Streams"
)
+ one_percent_number_of_streams *= 2
+
+ # Inform the user
+ self.update_status(
+ f"{self.name}: Processing {number_of_streams} {loading_stream_type} Streams {percent:.0f}%",
+ None,
+ True
+ )
+
+ # Skip if the name of the stream is empty
+ if stream_channel["name"] == "":
+ skip_stream = True
+ skipped_no_name_content = skipped_no_name_content + 1
+ self._save_to_file_skipped_streams(stream_channel)
+
+ # Skip if the user chose to hide adult streams
+ if self.hide_adult_content and loading_stream_type == self.live_type:
+ if "is_adult" in stream_channel:
+ if stream_channel["is_adult"] == "1":
+ skip_stream = True
+ skipped_adult_content = skipped_adult_content + 1
+ self._save_to_file_skipped_streams(stream_channel)
- # Set group title
- if the_group is not None:
- group_title = the_group.name
- else:
- group_title = self.catch_all_group.name
- the_group = self.catch_all_group
-
- if loading_stream_type == self.series_type:
- # Load all Series
- new_series = Serie(self, stream_channel)
- # To get all the Episodes for every Season of each
- # Series is very time consuming, we will only
- # populate the Series once the user click on the
- # Series, the Seasons and Episodes will be loaded
- # using x.getSeriesInfoByID() function
-
- else:
- new_channel = Channel(
- self, group_title, stream_channel
- )
-
- if new_channel.group_id == "9999":
- print(" - xEverythingElse Channel -> {} - {}".format(new_channel.name,new_channel.stream_type))
-
- # Save the new channel to the local list of channels
- if loading_stream_type == self.live_type:
- self.channels.append(new_channel)
- elif loading_stream_type == self.vod_type:
- self.movies.append(new_channel)
- else:
- self.series.append(new_series)
-
- # Add stream to the specific Group
- if the_group is not None:
- if loading_stream_type != self.series_type:
- the_group.channels.append(new_channel)
- else:
- the_group.series.append(new_series)
- else:
- print(" - Group not found `{}`".format(stream_channel["name"]))
-
- # Print information of which streams have been skipped
- if self.hide_adult_content:
- print(" - Skipped {} adult {} streams".format(skipped_adult_content, loading_stream_type))
- if skipped_no_name_content > 0:
- print(" - Skipped {} unprintable {} streams".format(skipped_no_name_content, loading_stream_type))
- else:
- print(" - Could not load {} Streams".format(loading_stream_type))
-
- self.state["loaded"] = True
+ if not skip_stream:
+ # Some channels have no group,
+ # so let's add them to the catch all group
+ if stream_channel["category_id"] is None:
+ stream_channel["category_id"] = "9999"
+ elif stream_channel["category_id"] != "1":
+ pass
+
+ # Find the first occurence of the group that the
+ # Channel or Stream is pointing to
+ the_group = next(
+ (x for x in self.groups if x.group_id == int(stream_channel["category_id"])),
+ None
+ )
+ # Set group title
+ if the_group is not None:
+ group_title = the_group.name
+ else:
+ if loading_stream_type == self.live_type:
+ group_title = self.live_catch_all_group.name
+ the_group = self.live_catch_all_group
+ elif loading_stream_type == self.vod_type:
+ group_title = self.vod_catch_all_group.name
+ the_group = self.vod_catch_all_group
+ elif loading_stream_type == self.series_type:
+ group_title = self.series_catch_all_group.name
+ the_group = self.series_catch_all_group
+
+
+ if loading_stream_type == self.series_type:
+ # Load all Series
+ new_series = Serie(self, stream_channel)
+ # To get all the Episodes for every Season of each
+ # Series is very time consuming, we will only
+ # populate the Series once the user click on the
+ # Series, the Seasons and Episodes will be loaded
+ # using x.getSeriesInfoByID() function
+
+ else:
+ new_channel = Channel(
+ self, group_title, stream_channel
+ )
+
+ if new_channel.group_id == "9999":
+ print(f" - xEverythingElse Channel -> {new_channel.name} - {new_channel.stream_type}")
+
+ # Save the new channel to the local list of channels
+ if loading_stream_type == self.live_type:
+ self.channels.append(new_channel)
+ elif loading_stream_type == self.vod_type:
+ self.movies.append(new_channel)
+ else:
+ self.series.append(new_series)
+
+ # Add stream to the specific Group
+ if the_group is not None:
+ if loading_stream_type != self.series_type:
+ the_group.channels.append(new_channel)
+ else:
+ the_group.series.append(new_series)
+ else:
+ print(f" - Group not found `{stream_channel['name']}`")
+ print("\n")
+ dt = timer() - start
+ # Print information of which streams have been skipped
+ if self.hide_adult_content:
+ print(f" - Skipped {skipped_adult_content} adult {loading_stream_type} streams")
+ if skipped_no_name_content > 0:
+ print(f" - Skipped {skipped_no_name_content} "
+ "unprintable {loading_stream_type} streams")
else:
- print("Warning, data has already been loaded.")
- else:
- print("Warning, cannot load steams since authorization failed")
+ print(f" - Could not load {loading_stream_type} Streams")
+
+ self.state["loaded"] = True
def _save_to_file_skipped_streams(self, stream_channel: Channel):
@@ -702,25 +780,28 @@ def _save_to_file_skipped_streams(self, stream_channel: Channel):
try:
with open(full_filename, mode="a", encoding="utf-8") as myfile:
myfile.writelines(json_data)
+ myfile.write('\n')
+ return True
except Exception as e:
- print(" - Could not save to skipped stream file `{}`: e=`{}`".format(
- full_filename, e
- ))
- return False
+ print(f" - Could not save to skipped stream file `{full_filename}`: e=`{e}`")
+ return False
def get_series_info_by_id(self, get_series: dict):
- """Get Seasons and Episodes for a Serie
+ """Get Seasons and Episodes for a Series
Args:
- get_series (dict): Serie dictionary
+ get_series (dict): Series dictionary
"""
- start = timer()
+
series_seasons = self._load_series_info_by_id_from_provider(get_series.series_id)
- dt = timer() - start
- # print("Loaded in {:.3f} sec".format(dt))
+
+ if series_seasons["seasons"] is None:
+ series_seasons["seasons"] = [
+ {"name": "Season 1", "cover": series_seasons["info"]["cover"]}
+ ]
+
for series_info in series_seasons["seasons"]:
season_name = series_info["name"]
- season_key = series_info["season_number"]
season = Season(season_name)
get_series.seasons[season_name] = season
if "episodes" in series_seasons.keys():
@@ -731,32 +812,77 @@ def get_series_info_by_id(self, get_series: dict):
)
season.episodes[episode_info["title"]] = new_episode_channel
- def _get_request(self, URL: str, timeout: Tuple = (2, 15)):
+ def _handle_request_exception(self, exception: requests.exceptions.RequestException):
+ """Handle different types of request exceptions."""
+ if isinstance(exception, requests.exceptions.ConnectionError):
+ print(" - Connection Error: Possible network problem \
+ (e.g. DNS failure, refused connection, etc)")
+ elif isinstance(exception, requests.exceptions.HTTPError):
+ print(" - HTTP Error")
+ elif isinstance(exception, requests.exceptions.TooManyRedirects):
+ print(" - TooManyRedirects")
+ elif isinstance(exception, requests.exceptions.ReadTimeout):
+ print(" - Timeout while loading data")
+ else:
+ print(f" - An unexpected error occurred: {exception}")
+
+ def _get_request(self, url: str, timeout: Tuple[int, int] = (2, 15)) -> Optional[dict]:
"""Generic GET Request with Error handling
Args:
URL (str): The URL where to GET content
- timeout (Tuple, optional): Connection and Downloading Timeout. Defaults to (2,15).
+ timeout (Tuple[int, int], optional): Connection and Downloading Timeout.
+ Defaults to (2,15).
Returns:
- [type]: JSON dictionary of the loaded data, or None
+ Optional[dict]: JSON dictionary of the loaded data, or None
"""
- try:
- r = requests.get(URL, timeout=timeout)
- if r.status_code == 200:
- return r.json()
-
- except requests.exceptions.ConnectionError:
- print(" - Connection Error")
- except requests.exceptions.HTTPError:
- print(" - HTTP Error")
+ kb_size = 1024
+ all_data = []
+ down_stats = {"bytes": 0, "kbytes": 0, "mbytes": 0, "start": 0.0, "delta_sec": 0.0}
- except requests.exceptions.TooManyRedirects:
- print(" - TooManyRedirects")
-
- except requests.exceptions.ReadTimeout:
- print(" - Timeout while loading data")
+ for attempt in range(10):
+ try:
+ response = requests.get(
+ url,
+ stream=True,
+ timeout=timeout,
+ headers=self.connection_headers
+ )
+ response.raise_for_status() # Raise an HTTPError for bad responses (4xx and 5xx)
+ break
+ except requests.exceptions.RequestException as e:
+ self._handle_request_exception(e)
+ return None
+
+ # If there is an answer from the remote server
+ if response.status_code in (200, 206):
+ down_stats["start"] = time.perf_counter()
+
+ # Set downloaded size
+ down_stats["bytes"] = 0
+
+ # Set stream blocks
+ block_bytes = int(1*kb_size*kb_size) # 4 MB
+
+ # Grab data by block_bytes
+ for data in response.iter_content(block_bytes, decode_unicode=False):
+ down_stats["bytes"] += len(data)
+ down_stats["kbytes"] = down_stats["bytes"]/kb_size
+ down_stats["mbytes"] = down_stats["bytes"]/kb_size/kb_size
+ down_stats["delta_sec"] = time.perf_counter() - down_stats["start"]
+ download_speed_average = down_stats["kbytes"]//down_stats["delta_sec"]
+ msg = f'\rDownloading {down_stats["kbytes"]:.1f} MB at {download_speed_average:.0f} kB/s'
+ sys.stdout.write(msg)
+ sys.stdout.flush()
+ self.update_status(msg, None, True)
+ all_data.append(data)
+ print(" - Done")
+ full_content = b''.join(all_data)
+ return json.loads(full_content)
+
+ print(f"HTTP error {response.status_code} while retrieving from {url}")
return None
@@ -770,17 +896,17 @@ def _load_categories_from_provider(self, stream_type: str):
Returns:
[type]: JSON if successfull, otherwise None
"""
- theURL = ""
+ url = ""
if stream_type == self.live_type:
- theURL = self.get_live_categories_URL()
+ url = self.get_live_categories_URL()
elif stream_type == self.vod_type:
- theURL = self.get_vod_cat_URL()
+ url = self.get_vod_cat_URL()
elif stream_type == self.series_type:
- theURL = self.get_series_cat_URL()
+ url = self.get_series_cat_URL()
else:
- theURL = ""
+ url = ""
- return self._get_request(theURL)
+ return self._get_request(url)
# GET Streams
def _load_streams_from_provider(self, stream_type: str):
@@ -792,17 +918,17 @@ def _load_streams_from_provider(self, stream_type: str):
Returns:
[type]: JSON if successfull, otherwise None
"""
- theURL = ""
+ url = ""
if stream_type == self.live_type:
- theURL = self.get_live_streams_URL()
+ url = self.get_live_streams_URL()
elif stream_type == self.vod_type:
- theURL = self.get_vod_streams_URL()
+ url = self.get_vod_streams_URL()
elif stream_type == self.series_type:
- theURL = self.get_series_URL()
+ url = self.get_series_URL()
else:
- theURL = ""
+ url = ""
- return self._get_request(theURL)
+ return self._get_request(url)
# GET Streams by Category
def _load_streams_by_category_from_provider(self, stream_type: str, category_id):
@@ -815,18 +941,18 @@ def _load_streams_by_category_from_provider(self, stream_type: str, category_id)
Returns:
[type]: JSON if successfull, otherwise None
"""
- theURL = ""
+ url = ""
if stream_type == self.live_type:
- theURL = self.get_live_streams_URL_by_category(category_id)
+ url = self.get_live_streams_URL_by_category(category_id)
elif stream_type == self.vod_type:
- theURL = self.get_vod_streams_URL_by_category(category_id)
+ url = self.get_vod_streams_URL_by_category(category_id)
elif stream_type == self.series_type:
- theURL = self.get_series_URL_by_category(category_id)
+ url = self.get_series_URL_by_category(category_id)
else:
- theURL = ""
+ url = ""
- return self._get_request(theURL)
+ return self._get_request(url)
# GET SERIES Info
def _load_series_info_by_id_from_provider(self, series_id: str):
@@ -868,66 +994,79 @@ def allEpg(self):
return self._get_request(self.get_all_epg_URL())
## URL-builder methods
- def get_authenticate_URL(self):
- URL = "%s/player_api.php?username=%s&password=%s" % (self.server, self.username, self.password)
- return URL
-
- def get_live_categories_URL(self):
- URL = "%s/player_api.php?username=%s&password=%s&action=%s" % (self.server, self.username, self.password, "get_live_categories")
- return URL
-
- def get_live_streams_URL(self):
- URL = "%s/player_api.php?username=%s&password=%s&action=%s" % (self.server, self.username, self.password, "get_live_streams")
- return URL
-
- def get_live_streams_URL_by_category(self, category_id):
- URL = "%s/player_api.php?username=%s&password=%s&action=%s&category_id=%s" % (self.server, self.username, self.password, "get_live_streams", category_id)
- return URL
-
- def get_vod_cat_URL(self):
- URL = "%s/player_api.php?username=%s&password=%s&action=%s" % (self.server, self.username, self.password, "get_vod_categories")
- return URL
-
- def get_vod_streams_URL(self):
- URL = "%s/player_api.php?username=%s&password=%s&action=%s" % (self.server, self.username, self.password, "get_vod_streams")
- return URL
-
- def get_vod_streams_URL_by_category(self, category_id):
- URL = "%s/player_api.php?username=%s&password=%s&action=%s&category_id=%s" % (self.server, self.username, self.password, "get_vod_streams", category_id)
- return URL
-
- def get_series_cat_URL(self):
- URL = "%s/player_api.php?username=%s&password=%s&action=%s" % (self.server, self.username, self.password, "get_series_categories")
- return URL
-
- def get_series_URL(self):
- URL = "%s/player_api.php?username=%s&password=%s&action=%s" % (self.server, self.username, self.password, "get_series")
- return URL
-
- def get_series_URL_by_category(self, category_id):
- URL = "%s/player_api.php?username=%s&password=%s&action=%s&category_id=%s" % (self.server, self.username, self.password, "get_series", category_id)
- return URL
-
- def get_series_info_URL_by_ID(self, series_id):
- URL = "%s/player_api.php?username=%s&password=%s&action=%s&series_id=%s" % (self.server, self.username, self.password, "get_series_info", series_id)
- return URL
-
- def get_VOD_info_URL_by_ID(self, vod_id):
- URL = "%s/player_api.php?username=%s&password=%s&action=%s&vod_id=%s" % (self.server, self.username, self.password, "get_vod_info", vod_id)
- return URL
-
- def get_live_epg_URL_by_stream(self, stream_id):
- URL = "%s/player_api.php?username=%s&password=%s&action=%s&stream_id=%s" % (self.server, self.username, self.password, "get_short_epg", stream_id)
- return URL
-
- def get_live_epg_URL_by_stream_and_limit(self, stream_id, limit):
- URL = "%s/player_api.php?username=%s&password=%s&action=%s&stream_id=%s&limit=%s" % (self.server, self.username, self.password, "get_short_epg", stream_id, limit)
- return URL
-
- def get_all_live_epg_URL_by_stream(self, stream_id):
- URL = "%s/player_api.php?username=%s&password=%s&action=%s&stream_id=%s" % (self.server, self.username, self.password, "get_simple_data_table", stream_id)
- return URL
-
- def get_all_epg_URL(self):
- URL = "%s/xmltv.php?username=%s&password=%s" % (self.server, self.username, self.password)
- return URL
+ def get_live_categories_URL(self) -> str:
+ return f"{self.base_url}&action=get_live_categories"
+
+ def get_live_streams_URL(self) -> str:
+ return f"{self.base_url}&action=get_live_streams"
+
+ def get_live_streams_URL_by_category(self, category_id) -> str:
+ return f"{self.base_url}&action=get_live_streams&category_id={category_id}"
+
+ def get_vod_cat_URL(self) -> str:
+ return f"{self.base_url}&action=get_vod_categories"
+
+ def get_vod_streams_URL(self) -> str:
+ return f"{self.base_url}&action=get_vod_streams"
+
+ def get_vod_streams_URL_by_category(self, category_id) -> str:
+ return f"{self.base_url}&action=get_vod_streams&category_id={category_id}"
+
+ def get_series_cat_URL(self) -> str:
+ return f"{self.base_url}&action=get_series_categories"
+
+ def get_series_URL(self) -> str:
+ return f"{self.base_url}&action=get_series"
+
+ def get_series_URL_by_category(self, category_id) -> str:
+ return f"{self.base_url}&action=get_series&category_id={category_id}"
+
+ def get_series_info_URL_by_ID(self, series_id) -> str:
+ return f"{self.base_url}&action=get_series_info&series_id={series_id}"
+
+ def get_VOD_info_URL_by_ID(self, vod_id) -> str:
+ return f"{self.base_url}&action=get_vod_info&vod_id={vod_id}"
+
+ def get_live_epg_URL_by_stream(self, stream_id) -> str:
+ return f"{self.base_url}&action=get_short_epg&stream_id={stream_id}"
+
+ def get_live_epg_URL_by_stream_and_limit(self, stream_id, limit) -> str:
+ return f"{self.base_url}&action=get_short_epg&stream_id={stream_id}&limit={limit}"
+
+ def get_all_live_epg_URL_by_stream(self, stream_id) -> str:
+ return f"{self.base_url}&action=get_simple_data_table&stream_id={stream_id}"
+
+ def get_all_epg_URL(self) -> str:
+ return f"{self.server}/xmltv.php?username={self.username}&password={self.password}"
+
+# The MIT License (MIT)
+# Copyright (c) 2016 Vladimir Ignatev
+#
+# Permission is hereby granted, free of charge, to any person obtaining
+# a copy of this software and associated documentation files (the "Software"),
+# to deal in the Software without restriction, including without limitation
+# the rights to use, copy, modify, merge, publish, distribute, sublicense,
+# and/or sell copies of the Software, and to permit persons to whom the Software
+# is furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included
+# in all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
+# INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
+# PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE
+# FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT
+# OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE
+# OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+def progress(count, total, status='') -> float:
+ bar_len = 60
+ filled_len = int(round(bar_len * count / float(total)))
+
+ percents = round(100.0 * count / float(total), 1)
+ bar_value = '=' * filled_len + '-' * (bar_len - filled_len)
+
+ #stdout.write('[%s] %s%s ...%s\r' % (bar_value, percents, '%', status))
+ stdout.write(f"[{bar_value}] {percents:.0f}% ...{status}\r")
+ stdout.flush() # As suggested by Rom Ruben (see: http://stackoverflow.com/questions/3173320/text-progress-bar-in-the-console/27871113#comment50529068_27871113)
+ return percents
diff --git a/usr/share/glib-2.0/schemas/org.x.hypnotix.gschema.xml b/usr/share/glib-2.0/schemas/org.x.hypnotix.gschema.xml
index 0ffd7b3d..00be8a3a 100644
--- a/usr/share/glib-2.0/schemas/org.x.hypnotix.gschema.xml
+++ b/usr/share/glib-2.0/schemas/org.x.hypnotix.gschema.xml
@@ -16,6 +16,16 @@
+
+ true
+
+
+
+
+ true
+
+
+
"Free-TV"
Provider selected by default
diff --git a/usr/share/hypnotix/hypnotix.ui b/usr/share/hypnotix/hypnotix.ui
index 0308f1bb..d7eef450 100644
--- a/usr/share/hypnotix/hypnotix.ui
+++ b/usr/share/hypnotix/hypnotix.ui
@@ -1002,7 +1002,7 @@
False
12
-
+
-
-
+
+
1
@@ -1073,10 +1082,40 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
- Playback
- video-x-generic-symbolic
+ General
+ preferences-system-network-symbolic
@@ -1337,6 +1376,85 @@
2
+
+
+
+
+ Playback
+ video-x-generic-symbolic
+ 3
+
+
True