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
Browse files Browse the repository at this point in the history
  • Loading branch information
ReenigneArcher committed Dec 30, 2023
1 parent 4d576bd commit bc21405
Show file tree
Hide file tree
Showing 3 changed files with 95 additions and 72 deletions.
129 changes: 77 additions & 52 deletions src/themerr/gui.py
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

Check warning on line 136 in src/themerr/gui.py

View check run for this annotation

Codecov / codecov/patch

src/themerr/gui.py#L136

Added line #L136 was not covered by tests

kodi_id = None

Check warning on line 138 in src/themerr/gui.py

View check run for this annotation

Codecov / codecov/patch

src/themerr/gui.py#L138

Added line #L138 was not covered by tests

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

Check warning on line 144 in src/themerr/gui.py

View check run for this annotation

Codecov / codecov/patch

src/themerr/gui.py#L140-L144

Added lines #L140 - L144 were not covered by tests

# 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())

Check warning on line 147 in src/themerr/gui.py

View check run for this annotation

Codecov / codecov/patch

src/themerr/gui.py#L147

Added line #L147 was not covered by tests
or (datetime.now().timestamp() - self.uuid_mapping[kodi_id]['timestamp']) > 3600):
self.uuid_mapping[kodi_id] = {

Check warning on line 149 in src/themerr/gui.py

View check run for this annotation

Codecov / codecov/patch

src/themerr/gui.py#L149

Added line #L149 was not covered by tests
'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):

Check warning on line 177 in src/themerr/gui.py

View check run for this annotation

Codecov / codecov/patch

src/themerr/gui.py#L177

Added line #L177 was not covered by tests
continue
if not self.kodi_id_mapping[kodi_id].get('youtube_url'):
if not self.uuid_mapping[kodi_id].get('youtube_url'):

Check warning on line 179 in src/themerr/gui.py

View check run for this annotation

Codecov / codecov/patch

src/themerr/gui.py#L179

Added line #L179 was not covered by tests
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'

Check warning on line 252 in src/themerr/gui.py

View check run for this annotation

Codecov / codecov/patch

src/themerr/gui.py#L252

Added line #L252 was not covered by tests

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

Check warning on line 255 in src/themerr/gui.py

View check run for this annotation

Codecov / codecov/patch

src/themerr/gui.py#L254-L255

Added lines #L254 - L255 were not covered by tests
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}")

Check warning on line 348 in src/themerr/gui.py

View check run for this annotation

Codecov / codecov/patch

src/themerr/gui.py#L347-L348

Added lines #L347 - L348 were not covered by tests
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
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
28 changes: 13 additions & 15 deletions tests/unit/test_gui.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,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 +67,25 @@ 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
}

youtube_url = window_obj.find_youtube_url_from_ids(
ids=test_ids,
@pytest.mark.parametrize('kodi_id', [
'tmdb_10378',
])
def test_find_youtube_url(window_obj, kodi_id):
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', [
'tmdb_0',
])
def test_find_youtube_url_exception(window_obj, kodi_id):
youtube_url = window_obj.find_youtube_url(
kodi_id=kodi_id,
db_type='bar',
)

Expand Down

0 comments on commit bc21405

Please sign in to comment.