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

fix: generate kodi_id for cases where a list item has no DBID #22

Merged
merged 1 commit into from
Dec 30, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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
56 changes: 42 additions & 14 deletions tests/unit/test_gui.py
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