# -*- coding: utf-8 -*-
# standard imports
import hashlib
import json
import os
import shutil
# plex debugging
try:
import plexhints # noqa: F401
except ImportError:
pass
else: # the code is running outside of Plex
from plexhints.core_kit import Core # core kit
from plexhints.log_kit import Log # log kit
from plexhints.parse_kit import XML # parse kit
from plexhints.prefs_kit import Prefs # prefs kit
# local imports
from constants import (
contributes_to,
metadata_base_directory,
metadata_type_map,
plex_section_type_settings_map,
plex_url,
themerr_data_directory
)
# constants
legacy_keys = [
'downloaded_timestamp'
]
[docs]def agent_enabled(item_agent, item_type):
# type: (str, str) -> bool
"""
Check if the specified agent is enabled.
Parameters
----------
item_agent : str
The agent to check.
item_type : str
The type of the item to check.
Returns
-------
py:class:`bool`
True if the agent is enabled, False otherwise.
Examples
--------
>>> agent_enabled(item_agent='com.plexapp.agents.imdb', item_type='movie')
True
>>> agent_enabled(item_agent='com.plexapp.agents.themoviedb', item_type='movie')
True
>>> agent_enabled(item_agent='com.plexapp.agents.themoviedb', item_type='show')
True
>>> agent_enabled(item_agent='com.plexapp.agents.thetvdb', item_type='show')
True
>>> agent_enabled(item_agent='dev.lizardbyte.retroarcher-plex', item_type='movie')
True
"""
# get the settings for this agent
settings_url = '{}/system/agents/{}/config/{}'.format(
plex_url, item_agent, plex_section_type_settings_map[item_type])
settings_data = XML.ElementFromURL(
url=settings_url,
cacheTime=0
)
Log.Debug('settings data: {}'.format(settings_data))
themerr_plex_element = settings_data.find(".//Agent[@name='Themerr-plex']")
if themerr_plex_element.get('enabled') == '1': # Plex is using a string
return True
else:
return False
[docs]def continue_update(item_agent, item_type):
# type: (str, str) -> bool
"""
Check if the specified agent should continue updating.
Parameters
----------
item_agent : str
The agent to check.
item_type : str
The type of the item to check.
Returns
-------
py:class:`bool`
True if the agent should continue updating, False otherwise.
Examples
--------
>>> continue_update(item_agent='tv.plex.agents.movie', item_type='movie')
True
>>> continue_update(item_agent='com.plexapp.agents.imdb', item_type='movie')
True
>>> continue_update(item_agent='com.plexapp.agents.themoviedb', item_type='movie')
True
>>> continue_update(item_agent='com.plexapp.agents.themoviedb', item_type='show')
True
>>> continue_update(item_agent='com.plexapp.agents.thetvdb', item_type='show')
True
>>> continue_update(item_agent='dev.lizardbyte.retroarcher-plex', item_type='movie')
True
"""
if item_agent == 'tv.plex.agents.movie':
return Prefs['bool_plex_movie_support']
elif item_agent in contributes_to:
return agent_enabled(item_agent=item_agent, item_type=item_type)
else:
return False
[docs]def get_themerr_json_path(item):
# type: (any) -> str
"""
Get the path to the Themerr data file.
Get the path to the Themerr data file for the item specified by the ``item``.
Parameters
----------
item : any
The item to get the Themerr data file path for.
Returns
-------
str
The path to the Themerr data file.
Examples
--------
>>> get_themerr_json_path(item=...)
'.../Plex Media Server/Plug-in Support/Data/dev.lizardbyte.themerr-plex/DataItems/...'
"""
themerr_json_path = os.path.join(themerr_data_directory, metadata_type_map[item.type],
'{}.json'.format(item.ratingKey))
return themerr_json_path
[docs]def get_themerr_json_data(item):
# type: (any) -> dict
"""
Get the Themerr data for the specified item.
Themerr data is stored as a JSON file in the Themerr data directory, and is used to ensure that we don't
unnecessarily re-upload media to the Plex server.
Parameters
----------
item : any
The item to get the Themerr data for.
Returns
-------
dict
The Themerr data for the specified item, or empty dict if no Themerr data exists.
"""
themerr_json_path = get_themerr_json_path(item=item)
if os.path.isfile(themerr_json_path):
themerr_data = json.loads(s=str(Core.storage.load(filename=themerr_json_path, binary=False)))
else:
themerr_data = dict()
return themerr_data
[docs]def get_themerr_settings_hash():
# type: () -> str
"""
Get a hash of the current Themerr settings.
Returns
-------
str
Hash of the current Themerr settings.
Examples
--------
>>> get_themerr_settings_hash()
'...'
"""
# use to compare previous settings to new settings
themerr_settings = dict(
bool_prefer_mp4a_codec=Prefs['bool_prefer_mp4a_codec'],
int_plexapi_plexapi_timeout=Prefs['int_plexapi_plexapi_timeout'],
)
settings_hash = hashlib.sha256(json.dumps(themerr_settings)).hexdigest()
return settings_hash
[docs]def update_themerr_data_file(item, new_themerr_data):
# type: (any, dict) -> None
"""
Update the Themerr data file for the specified item.
This updates the themerr data file after uploading media to the Plex server.
Parameters
----------
item : any
The item to update the Themerr data file for.
new_themerr_data : dict
The Themerr data to update the Themerr data file with.
"""
# get the old themerr data
themerr_data = get_themerr_json_data(item=item)
# remove legacy keys
for key in legacy_keys:
try:
del themerr_data[key]
except KeyError:
pass
# update the old themerr data with the new themerr data
themerr_data.update(new_themerr_data)
# get path
themerr_json_path = get_themerr_json_path(item=item)
# create directory if it doesn't exist
if not os.path.isdir(os.path.dirname(themerr_json_path)):
os.makedirs(os.path.dirname(themerr_json_path))
# write themerr json
Core.storage.save(filename=themerr_json_path, data=json.dumps(themerr_data), binary=False)