Skip to content
This repository has been archived by the owner on Oct 13, 2024. It is now read-only.

Commit

Permalink
fix: generate kodi_id for cases where a list item has no DBID (#22)
Browse files Browse the repository at this point in the history
  • Loading branch information
@ReenigneArcher
ReenigneArcher authored Dec 30, 2023
1 parent 4d576bd commit ee0881f
Show file tree
Hide file tree
Showing 3 changed files with 124 additions and 71 deletions.
129 changes: 77 additions & 52 deletions src/themerr/gui.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -43,20 +43,22 @@ class Window:
The current selected item ID.
last_selected_item_id : Optional[int]
The last selected item ID.
kodi_id_mapping : dict
A mapping of Kodi IDs to YouTube URLs. This is used to cache the YouTube URLs for faster lookups.
uuid_mapping : dict
A mapping of uuids to YouTube URLs.
The UUID will be the database type and the database ID, separated by an underscore. e.g. `tmdb_1`
This is used to cache the YouTube URLs for faster lookups.
Methods
-------
window_watcher()
The main method that watches for changes to the Kodi window.
pre_checks()
Perform pre-checks before starting/stopping the theme.
process_kodi_id(kodi_id: int)
process_kodi_id(kodi_id: str)
Process the Kodi ID and return a YouTube URL.
process_movie(kodi_id: int)
Process the Kodi ID and return a dictionary of IDs.
find_youtube_url_from_ids(ids: dict, db_type: str)
find_youtube_url(kodi_id: str, db_type: str)
Find the YouTube URL from the IDs.
any_true(check: Optional[bool] = None, checks: Optional[Union[List[bool], Set[bool]]] = ())
Determine if the check is True or if any of the checks are True.
Expand Down Expand Up @@ -92,7 +94,24 @@ def __init__(self, player_instance=None):
self.playing_item_not_selected_for = 0
self.current_selected_item_id = None
self.last_selected_item_id = None
self.kodi_id_mapping = {}
self.uuid_mapping = {}

self._kodi_db_map = {
'tmdb': 'themoviedb',
'imdb': 'imdb',
}
self._supported_dbs = {
'games': ['igdb'],
'game_collections': ['igdb'],
'game_franchises': ['igdb'],
'movies': ['themoviedb', 'imdb'],
'movie_collections': ['themoviedb'],
}
self._dbs = (
'tmdb',
'imdb',
# 'igdb', # placeholder for video game support
)

def window_watcher(self):
"""
Expand All @@ -114,14 +133,20 @@ def window_watcher(self):
timeout_factor = settings.settings.theme_timeout()
timeout = timeout_factor * (1000 / sleep_time)

selected_title = xbmc.getInfoLabel("ListItem.Label")
kodi_id = xbmc.getInfoLabel("ListItem.DBID")
kodi_id = int(kodi_id) if kodi_id else None
selected_title = xbmc.getInfoLabel("ListItem.Label") # this is only used for logging

kodi_id = None

for db in self._dbs:
db_id = xbmc.getInfoLabel(f'ListItem.UniqueID({db})')
if db_id:
kodi_id = f"{db}_{db_id}"
break # break on the first supported db

# prefetch the YouTube url (if not already cached or cache is greater than 1 hour)
if kodi_id and (kodi_id not in list(self.kodi_id_mapping.keys())
or (datetime.now().timestamp() - self.kodi_id_mapping[kodi_id]['timestamp']) > 3600):
self.kodi_id_mapping[kodi_id] = {
if kodi_id and (kodi_id not in list(self.uuid_mapping.keys())
or (datetime.now().timestamp() - self.uuid_mapping[kodi_id]['timestamp']) > 3600):
self.uuid_mapping[kodi_id] = {
'timestamp': datetime.now().timestamp(),
'youtube_url': self.process_kodi_id(kodi_id=kodi_id)
}
Expand Down Expand Up @@ -149,13 +174,13 @@ def window_watcher(self):
else:
self.playing_item_not_selected_for = 0
if not self.player.theme_is_playing and self.item_selected_for >= timeout:
if not self.kodi_id_mapping.get(kodi_id):
if not self.uuid_mapping.get(kodi_id):
continue
if not self.kodi_id_mapping[kodi_id].get('youtube_url'):
if not self.uuid_mapping[kodi_id].get('youtube_url'):
continue
self.log.debug(f"Playing theme for {selected_title}, ID: {kodi_id}")
self.player.play_url(
url=self.kodi_id_mapping[kodi_id]['youtube_url'],
url=self.uuid_mapping[kodi_id]['youtube_url'],
kodi_id=kodi_id,
)

Expand Down Expand Up @@ -199,15 +224,15 @@ def pre_checks(self) -> bool:
self.log.debug("pre-checks passed")
return True

def process_kodi_id(self, kodi_id: int) -> Optional[str]:
def process_kodi_id(self, kodi_id: str) -> Optional[str]:
"""
Generate YouTube URL from a given Kodi ID.
This method takes a Kodi ID and returns a YouTube URL.
Parameters
----------
kodi_id : int
kodi_id : str
The Kodi ID to process.
Returns
Expand All @@ -218,34 +243,33 @@ def process_kodi_id(self, kodi_id: int) -> Optional[str]:
Examples
--------
>>> window = Window()
>>> window.process_kodi_id(kodi_id=1)
>>> window.process_kodi_id(kodi_id='tmdb_1')
"""
ids = None
database_type = None
if self.is_movies():
ids = self.process_movie(kodi_id=kodi_id)
database_type = 'movies'
elif self.is_movie_set():
database_type = 'movie_sets'
database_type = 'movie_collections'

if ids and database_type:
youtube_url = self.find_youtube_url_from_ids(
ids=ids,
if database_type:
youtube_url = self.find_youtube_url(
kodi_id=kodi_id,
db_type=database_type,
)

return youtube_url

def process_movie(self, kodi_id: int) -> Dict[str, Optional[Union[str, int]]]:
def _process_movie(self, dbid: int) -> Dict[str, Optional[Union[str, int]]]:
"""
Generate a dictionary of IDs from a given Kodi ID, for a movie.
This method takes a Kodi ID and returns a dictionary of IDs.
This method is no longer used, and may be removed in the future.
Parameters
----------
kodi_id : int
The Kodi ID to process.
dbid : int
The Kodi DBID to process.
Returns
-------
Expand All @@ -255,15 +279,15 @@ def process_movie(self, kodi_id: int) -> Dict[str, Optional[Union[str, int]]]:
Examples
--------
>>> window = Window()
>>> window.process_movie(kodi_id=1)
>>> window._process_movie(kodi_id=1)
{'themoviedb': ..., 'imdb': ...}
"""
# query the kodi database to get tmdb and imdb unique ids
rpc_query = {
"jsonrpc": "2.0",
"method": "VideoLibrary.GetMovieDetails",
"params": {
"movieid": int(kodi_id),
"movieid": int(dbid),
"properties": [
"imdbnumber",
"uniqueid",
Expand All @@ -275,24 +299,24 @@ def process_movie(self, kodi_id: int) -> Dict[str, Optional[Union[str, int]]]:
json_response = json.loads(rpc_response)
self.log.debug(f"JSON response: {json_response}")

# get the supported:
# get the supported ids
ids = {
'themoviedb': json_response['result']['moviedetails']['uniqueid'].get('tmdb'),
'imdb': json_response['result']['moviedetails']['uniqueid'].get('imdb'),
}
self.log.debug(f"IDs: {ids}")
return ids

def find_youtube_url_from_ids(self, ids: Dict[str, Optional[Union[str, int]]], db_type: str) -> Optional[str]:
def find_youtube_url(self, kodi_id: str, db_type: str) -> Optional[str]:
"""
Find YouTube URL from the Dictionary of IDs.
Given a dictionary of IDs, this method will query the Themerr DB to find the YouTube URL.
Parameters
----------
ids : Dict[str, Optional[Union[str, int]]]
The dictionary of IDs.
kodi_id : str
The Kodi ID to process.
db_type : str
The database type.
Expand All @@ -304,28 +328,29 @@ def find_youtube_url_from_ids(self, ids: Dict[str, Optional[Union[str, int]]], d
Examples
--------
>>> window = Window()
>>> window.find_youtube_url_from_ids(ids={'themoviedb': 10378}, db_type='movies')
>>> window.find_youtube_url(kodi_id='tmdb_1', db_type='movies')
"""
for key, value in list(ids.items()):
if value is None:
continue
self.log.debug(f"{key.upper()}_ID: {value}")
themerr_db_url = f"https://app.lizardbyte.dev/ThemerrDB/{db_type}/{key}/{value}.json"
self.log.debug(f"Themerr DB URL: {themerr_db_url}")

try:
response_data = requests.get(
url=themerr_db_url,
).json()
except requests.exceptions.RequestException as e:
self.log.debug(f"Exception getting data from {themerr_db_url}: {e}")
except json.decoder.JSONDecodeError:
self.log.debug(f"Exception decoding JSON from {themerr_db_url}")
else:
youtube_theme_url = response_data['youtube_theme_url']
self.log.debug(f"Youtube theme URL: {youtube_theme_url}")
split_id = kodi_id.split('_')
db = self._kodi_db_map[split_id[0]]
db_id = split_id[1]

return youtube_theme_url
self.log.debug(f"{db.upper()}_ID: {db_id}")
themerr_db_url = f"https://app.lizardbyte.dev/ThemerrDB/{db_type}/{db}/{db_id}.json"
self.log.debug(f"Themerr DB URL: {themerr_db_url}")

try:
response_data = requests.get(
url=themerr_db_url,
).json()
except requests.exceptions.RequestException as e:
self.log.debug(f"Exception getting data from {themerr_db_url}: {e}")
except json.decoder.JSONDecodeError:
self.log.debug(f"Exception decoding JSON from {themerr_db_url}")
else:
youtube_theme_url = response_data['youtube_theme_url']
self.log.debug(f"Youtube theme URL: {youtube_theme_url}")

return youtube_theme_url

@staticmethod
def any_true(check: Optional[bool] = None, checks: Optional[Union[List[bool], Set[bool]]] = ()):
Expand Down
10 changes: 5 additions & 5 deletions src/themerr/player.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ class Player(xbmc.Player):
True if a theme is currently playing, False otherwise.
theme_is_playing_for : int
The number of seconds the theme has been playing for.
theme_playing_kodi_id : Optional[int]
theme_playing_kodi_id : Optional[str]
The Kodi ID of the theme currently playing.
theme_playing_url : Optional[str]
The URL of the theme currently playing.
Expand All @@ -32,7 +32,7 @@ class Player(xbmc.Player):
-------
ytdl_extract_url(url: str) -> Optional[str]
Extract the audio URL from a YouTube URL.
play_url(url: str, kodi_id: int, windowed: bool = False)
play_url(url: str, kodi_id: str, windowed: bool = False)
Play a YouTube URL.
stop()
Stop playback.
Expand All @@ -59,7 +59,7 @@ def ytdl_extract_url(url: str) -> Optional[str]:
def play_url(
self,
url: str,
kodi_id: int,
kodi_id: str,
windowed: bool = False,
):
"""
Expand All @@ -71,15 +71,15 @@ def play_url(
----------
url : str
The url to play.
kodi_id : int
kodi_id : str
The Kodi ID of the item.
windowed : bool
True to play in a window, False otherwise.
Examples
--------
>>> player = Player()
>>> player.play_url(url="https://www.youtube.com/watch?v=dQw4w9WgXcQ", kodi_id=1)
>>> player.play_url(url="https://www.youtube.com/watch?v=dQw4w9WgXcQ", kodi_id='tmdb_1')
"""
playable_url = self.ytdl_extract_url(url=url)
if playable_url:
Expand Down
56 changes: 42 additions & 14 deletions tests/unit/test_gui.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,16 @@
from src.themerr import gui


@pytest.fixture(
scope='function',
params=[
'tmdb_10378',
],
)
def kodi_id(request):
return request.param


@pytest.fixture(scope='function')
def window_obj(mock_xbmc_player):
"""Return the Window object with a mocked player"""
Expand All @@ -28,7 +38,7 @@ def test_window_init(window_obj):
assert window_obj.playing_item_not_selected_for == 0
assert window_obj.current_selected_item_id is None
assert window_obj.last_selected_item_id is None
assert window_obj.kodi_id_mapping == {}
assert window_obj.uuid_mapping == {}


def test_pre_checks_no_item_playing(window_obj):
Expand Down Expand Up @@ -67,27 +77,45 @@ def test_pre_checks_all_passing(window_obj):
assert window_obj.pre_checks() is True


def test_find_youtube_url_from_ids(window_obj):
test_ids = {
'themoviedb': 10378, # Big Buck Bunny
}
def test_process_kodi_id_movies(kodi_id, mock_xbmc_get_cond_visibility, window_obj):
condition = 'Container.Content(movies)'
env_var = f'_KODI_GET_COND_VISIBILITY_{condition}'
os.environ[env_var] = '1'

youtube_url = window_obj.find_youtube_url_from_ids(
ids=test_ids,
youtube_url = window_obj.process_kodi_id(kodi_id=kodi_id)
assert youtube_url

del os.environ[env_var]


def test_process_kodi_id_movie_collection(mock_xbmc_get_cond_visibility, window_obj):
_kodi_id = 'tmdb_645'
condition = 'ListItem.IsCollection'
env_var = f'_KODI_GET_COND_VISIBILITY_{condition}'
os.environ[env_var] = '1'

youtube_url = window_obj.process_kodi_id(kodi_id=_kodi_id)
assert youtube_url

del os.environ[env_var]


def test_find_youtube_url(kodi_id, window_obj):
youtube_url = window_obj.find_youtube_url(
kodi_id=kodi_id,
db_type='movies',
)

assert youtube_url
assert youtube_url.startswith('https://')


def test_find_youtube_url_from_ids_exception(window_obj):
test_ids = {
'foo': 0,
}

youtube_url = window_obj.find_youtube_url_from_ids(
ids=test_ids,
@pytest.mark.parametrize('kodi_id_invalid', [
'tmdb_0',
])
def test_find_youtube_url_exception(window_obj, kodi_id_invalid):
youtube_url = window_obj.find_youtube_url(
kodi_id=kodi_id_invalid,
db_type='bar',
)

Expand Down

0 comments on commit ee0881f

Please sign in to comment.