Mirrors/Plex-Meta-Manager
2
0
Fork 0
mirror of https://github.com/meisnate12/Plex-Meta-Manager synced 2024-11-10 06:54:21 +00:00
Code Issues Projects Releases Packages Wiki Activity

[75] #557, #770 add sync_to_trakt_list to Collections/Playlists

Browse source
This commit is contained in:
meisnate12 2022-05-09 02:58:09 -04:00
parent 6c44ea858b
commit b9a0f87c40
4 changed files with 115 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
VERSION
View file

@ -1 +1 @@
1.16.5-develop74
1.16.5-develop75

31
modules/builder.py
View file

@ -418,6 +418,7 @@ class CollectionBuilder:
self.current_year = self.current_time.year
self.url_theme = None
self.file_theme = None
self.sync_to_trakt_list = None
self.collection_poster = None
self.collection_background = None
self.exists = False
@ -752,7 +753,7 @@ class CollectionBuilder:
self._tautulli(method_name, method_data)
elif method_name in tmdb.builders:
self._tmdb(method_name, method_data)
elif method_name in trakt.builders:
elif method_name in trakt.builders or method_name == "sync_to_trakt_list":
self._trakt(method_name, method_data)
elif method_name in tvdb.builders:
self._tvdb(method_name, method_data)
@ -1460,6 +1461,10 @@ class CollectionBuilder:
raise Failed(f"{self.Type} Error: {method_name} must be set to true")
elif method_name == "trakt_recommendations":
self.builders.append((method_name, util.parse(self.Type, method_name, method_data, datatype="int", default=10, maximum=100)))
elif method_name == "sync_to_trakt_list":
if method_data not in self.config.Trakt.slugs:
raise Failed(f"{self.Type} Error: {method_data} invalid. Options {', '.join(self.config.Trakt.slugs)}")
self.sync_to_trakt_list = method_data
elif method_name in trakt.builders:
if method_name in ["trakt_chart", "trakt_userlist"]:
trakt_dicts = method_data
@ -2687,6 +2692,30 @@ class CollectionBuilder:
self.library.moveItem(self.obj, item, previous)
previous = item
def sync_trakt_list(self):
logger.info("")
logger.separator(f"Syncing {self.name} {self.Type} to Trakt List {self.sync_to_trakt_list}", space=False, border=False)
logger.info("")
if self.obj:
self.obj.reload()
self.load_collection_items()
current_ids = []
for item in self.items:
for pl_library in self.libraries:
new_id = None
if isinstance(item, Movie) and item.ratingKey in pl_library.movie_rating_key_map:
new_id = (pl_library.movie_rating_key_map[item.ratingKey], "tmdb")
elif isinstance(item, Show) and item.ratingKey in pl_library.show_rating_key_map:
new_id = (pl_library.show_rating_key_map[item.ratingKey], "tvdb")
elif isinstance(item, Season) and item.parentRatingKey in pl_library.show_rating_key_map:
new_id = (f"{pl_library.show_rating_key_map[item.parentRatingKey]}_{item.seasonNumber}", "tvdb_season")
elif isinstance(item, Episode) and item.grandparentRatingKey in pl_library.show_rating_key_map:
new_id = (f"{pl_library.show_rating_key_map[item.grandparentRatingKey]}_{item.seasonNumber}_{item.episodeNumber}", "tvdb_episode")
if new_id:
current_ids.append(new_id)
break
self.config.Trakt.sync_list(self.sync_to_trakt_list, current_ids)
def delete(self):
output = ""
if self.obj:

101
modules/trakt.py
View file

@ -1,6 +1,7 @@
import requests, webbrowser
import requests, time, webbrowser
from modules import util
from modules.util import Failed, TimeoutExpired
from retrying import retry
from ruamel import yaml
logger = util.logger
@ -46,6 +47,7 @@ class Trakt:
if not self._save(self.authorization):
if not self._refresh():
self._authorization()
self._slugs = None
self._movie_genres = None
self._show_genres = None
self._movie_languages = None
@ -55,6 +57,17 @@ class Trakt:
self._movie_certifications = None
self._show_certifications = None
@property
def slugs(self):
if self._slugs is None:
items = []
try:
items = [i["ids"]["slug"] for i in self._request(f"/users/me/lists")]
except Failed:
pass
self._slugs = items
return self._slugs
@property
def movie_genres(self):
if not self._movie_genres:
@ -175,7 +188,8 @@ class Trakt:
return True
return False
def _request(self, url, params=None):
@retry(stop_max_attempt_number=6, wait_fixed=10000, retry_on_exception=util.retry_if_not_failed)
def _request(self, url, params=None, json=None):
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.authorization['access_token']}",
@ -189,24 +203,28 @@ class Trakt:
current = 1
if self.config.trace_mode:
logger.debug(f"URL: {base_url}{url}")
if params:
logger.debug(f"Params: {params}")
if json:
logger.debug(f"JSON: {json}")
while current <= pages:
if pages == 1:
response = self.config.get(f"{base_url}{url}", headers=headers, params=params)
if "X-Pagination-Page-Count" in response.headers and not params:
pages = int(response.headers["X-Pagination-Page-Count"])
else:
if pages > 1:
params["page"] = current
response = self.config.get(f"{base_url}{url}", headers=headers, params=params)
if response.status_code == 200:
json_data = response.json()
if self.config.trace_mode:
logger.debug(f"Response: {json_data}")
if isinstance(json_data, dict):
return json_data
else:
output_json.extend(response.json())
if json is not None:
response = self.config.post(f"{base_url}{url}", json=json, headers=headers)
else:
response = self.config.get(f"{base_url}{url}", headers=headers, params=params)
if pages == 1 and "X-Pagination-Page-Count" in response.headers:
pages = int(response.headers["X-Pagination-Page-Count"])
if response.status_code >= 400:
raise Failed(f"({response.status_code}) {response.reason}")
json_data = response.json()
if self.config.trace_mode:
logger.debug(f"Response: {json_data}")
if isinstance(json_data, dict):
return json_data
else:
output_json.extend(json_data)
current += 1
return output_json
@ -229,7 +247,7 @@ class Trakt:
except Failed:
raise Failed(f"Trakt Error: List {data} not found")
def _parse(self, items, typeless=False, item_type=None):
def _parse(self, items, typeless=False, item_type=None, trakt_ids=False):
ids = []
for item in items:
if typeless:
@ -253,12 +271,55 @@ class Trakt:
if current_type in ["person", "list"]:
final_id = (final_id, data["name"])
final_type = f"{id_type}_{current_type}" if current_type in ["episode", "season", "person"] else id_type
ids.append((final_id, final_type))
ids.append((int(item["id"]), final_id, final_type) if trakt_ids else (final_id, final_type))
else:
name = data["name"] if current_type in ["person", "list"] else f"{data['title']} ({data['year']})"
logger.error(f"Trakt Error: No {id_display} found for {name}")
return ids
def _build_item_json(self, ids):
data = {}
for input_id, id_type in ids:
movies = id_type in ["imdb", "tmdb"]
shows = id_type in ["imdb", "tvdb", "tmdb_show", "tvdb_season", "tvdb_episode"]
if not movies and not shows:
continue
type_set = str(id_type).split("_")
id_set = str(input_id).split("_")
item = {"ids": {type_set[0]: id_set[0] if type_set[0] == "imdb" else int(id_set[0])}}
if id_type in ["tvdb_season", "tvdb_episode"]:
season_data = {"number": int(id_set[1])}
if id_type == "tvdb_episode":
season_data["episodes"] = [{"number": int(id_set[2])}]
item["seasons"] = [season_data]
if movies:
if "movies" not in data:
data["movies"] = []
data["movies"].append(item)
if shows:
if "shows" not in data:
data["shows"] = []
data["shows"].append(item)
return data
def sync_list(self, slug, ids):
current_ids = self._list(slug, urlparse=False)
add_ids = [id_set for id_set in ids if id_set not in current_ids]
if add_ids:
self._request(f"/users/me/lists/{slug}/items", json=self._build_item_json(add_ids))
time.sleep(1)
remove_ids = [id_set for id_set in current_ids if id_set not in ids]
if remove_ids:
self._request(f"/users/me/lists/{slug}/items/remove", json=self._build_item_json(remove_ids))
time.sleep(1)
trakt_ids = self._list(slug, urlparse=False, trakt_ids=True)
trakt_lookup = {f"{ty}_{i_id}": t_id for t_id, i_id, ty in trakt_ids}
rank_ids = [trakt_lookup[f"{ty}_{i_id}"] for i_id, ty in ids if f"{ty}_{i_id}" in trakt_lookup]
self._request(f"/users/me/lists/{slug}/items/reorder", json={"rank": rank_ids})
def all_user_lists(self, user="me"):
try:
items = self._request(f"/users/{user}/lists")
@ -277,7 +338,7 @@ class Trakt:
def build_user_url(self, user, name):
return f"{base_url.replace('api.', '')}/users/{user}/lists/{name}"
def _list(self, data, urlparse=True):
def _list(self, data, urlparse=True, trakt_ids=False):
try:
url = requests.utils.urlparse(data).path if urlparse else f"/users/me/lists/{data}"
items = self._request(f"{url}/items")
@ -285,7 +346,7 @@ class Trakt:
raise Failed(f"Trakt Error: List {data} not found")
if len(items) == 0:
raise Failed(f"Trakt Error: List {data} is empty")
return self._parse(items)
return self._parse(items, trakt_ids=trakt_ids)
def _userlist(self, list_type, user, is_movie, sort_by=None):
try:

4
plex_meta_manager.py
View file

@ -601,7 +601,7 @@ def run_collection(config, library, metadata, requested_collections):
logger.info("")
logger.info(f"Plex Server Movie pre-roll video updated to {builder.server_preroll}")
if valid and run_item_details and builder.builders and (builder.item_details or builder.custom_sort):
if valid and run_item_details and builder.builders and (builder.item_details or builder.custom_sort or builder.sync_to_trakt_list):
try:
builder.load_collection_items()
except Failed:
@ -612,6 +612,8 @@ def run_collection(config, library, metadata, requested_collections):
builder.update_item_details()
if builder.custom_sort:
builder.sort_collection()
if builder.sync_to_trakt_list:
builder.sync_trakt_list()
builder.send_notifications()