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

move arms to its own file

Browse source
This commit is contained in:
meisnate12 2021-04-22 19:51:03 -04:00
parent f3d38ebc30
commit 1a70b7d80e
8 changed files with 194 additions and 217 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
modules/anidb.py
View file

@ -64,7 +64,7 @@ class AniDBAPI:
if method == "anidb_id": anidb_ids.append(data)
elif method == "anidb_relation": anidb_ids.extend(self.get_anidb_relations(data, language))
else: raise Failed(f"AniDB Error: Method {method} not supported")
movie_ids, show_ids = self.config.convert_anidb_list(anidb_ids, language)
movie_ids, show_ids = self.config.Arms.anidb_to_ids(anidb_ids, language)
if status_message:
logger.debug(f"AniDB IDs Found: {anidb_ids}")
logger.debug(f"TMDb IDs Found: {movie_ids}")

2
modules/anilist.py
View file

@ -254,7 +254,7 @@ class AniListAPI:
logger.info(f"Processing {pretty}: ({data}) {name} ({len(anilist_ids)} Anime)")
else:
raise Failed(f"AniList Error: Method {method} not supported")
movie_ids, show_ids = self.config.convert_anilist_list(anilist_ids, language)
movie_ids, show_ids = self.config.Arms.anilist_to_ids(anilist_ids, language)
if status_message:
logger.debug(f"AniList IDs Found: {anilist_ids}")
logger.debug(f"Shows Found: {show_ids}")

168
modules/arms.py Normal file
View file

@ -0,0 +1,168 @@
import logging, requests
from lxml import html
from modules import util
from modules.util import Failed
from retrying import retry
logger = logging.getLogger("Plex Meta Manager")
class ArmsAPI:
def __init__(self, config):
self.config = config
self.arms_url = "https://relations.yuna.moe/api/ids"
self.anidb_url = "https://raw.githubusercontent.com/Anime-Lists/anime-lists/master/anime-list-master.xml"
self.AniDBIDs = self._get_anidb()
@retry(stop_max_attempt_number=6, wait_fixed=10000)
def _get_anidb(self):
return html.fromstring(requests.get(self.anidb_url).content)
def anidb_to_tvdb(self, anidb_id): return self._anidb(anidb_id, "tvdbid")
def anidb_to_imdb(self, anidb_id): return self._anidb(anidb_id, "imdbid")
def _anidb(self, input_id, to_id):
ids = self.AniDBIDs.xpath(f"//anime[contains(@anidbid, '{input_id}')]/@{to_id}")
if len(ids) > 0:
try:
if len(ids[0]) > 0:
return ids[0].split(",") if to_id == "imdbid" else int(ids[0])
raise ValueError
except ValueError:
raise Failed(f"AniDB Error: No {util.pretty_ids[to_id]} ID found for AniDB ID: {input_id}")
else:
raise Failed(f"AniDB Error: AniDB ID: {input_id} not found")
@retry(stop_max_attempt_number=6, wait_fixed=10000)
def send_request(self, ids):
return requests.post(self.arms_url, json=ids).json()
def mal_to_anidb(self, mal_id):
anime_ids = self._arms_ids(mal_ids=mal_id)
if anime_ids[0] is None:
raise Failed(f"Arms Server Error: MyAnimeList ID: {mal_id} does not exist")
if anime_ids[0]["anidb"] is None:
raise Failed(f"Arms Server Error: No AniDB ID for MyAnimeList ID: {mal_id}")
return anime_ids[0]["anidb"]
def anidb_to_ids(self, anidb_list, language):
show_ids = []
movie_ids = []
for anidb_id in anidb_list:
try:
for imdb_id in self.anidb_to_imdb(anidb_id):
tmdb_id, _ = self.imdb_to_ids(imdb_id, language)
if tmdb_id:
movie_ids.append(tmdb_id)
break
else:
raise Failed
except Failed:
try:
tvdb_id = self.anidb_to_tvdb(anidb_id)
if tvdb_id:
show_ids.append(tvdb_id)
except Failed:
logger.error(f"AniDB Error: No TVDb ID or IMDb ID found for AniDB ID: {anidb_id}")
return movie_ids, show_ids
def anilist_to_ids(self, anilist_ids, language):
anidb_ids = []
for id_set in self._arms_ids(anilist_ids=anilist_ids):
if id_set["anidb"] is not None:
anidb_ids.append(id_set["anidb"])
else:
logger.error(f"Convert Error: AniDB ID not found for AniList ID: {id_set['anilist']}")
return self.anidb_to_ids(anidb_ids, language)
def myanimelist_to_ids(self, mal_ids, language):
anidb_ids = []
for id_set in self._arms_ids(mal_ids=mal_ids):
if id_set["anidb"] is not None:
anidb_ids.append(id_set["anidb"])
else:
logger.error(f"Convert Error: AniDB ID not found for MyAnimeList ID: {id_set['myanimelist']}")
return self.anidb_to_ids(anidb_ids, language)
def _arms_ids(self, anilist_ids=None, anidb_ids=None, mal_ids=None):
all_ids = []
def collect_ids(ids, id_name):
if ids:
if isinstance(ids, list):
all_ids.extend([{id_name: a_id} for a_id in ids])
else:
all_ids.append({id_name: ids})
collect_ids(anilist_ids, "anilist")
collect_ids(anidb_ids, "anidb")
collect_ids(mal_ids, "myanimelist")
converted_ids = []
if self.config.Cache:
unconverted_ids = []
for anime_dict in all_ids:
for id_type, anime_id in anime_dict.items():
query_ids, update = self.config.Cache.query_anime_map(anime_id, id_type)
if not update and query_ids:
converted_ids.append(query_ids)
else:
unconverted_ids.append({id_type: anime_id})
else:
unconverted_ids = all_ids
for anime_ids in self.send_request(unconverted_ids):
if anime_ids:
if self.config.Cache:
self.config.Cache.update_anime(False, anime_ids)
converted_ids.append(anime_ids)
return converted_ids
def imdb_to_ids(self, imdb_id, language):
update_tmdb = False
update_tvdb = False
if self.config.Cache:
tmdb_id, tvdb_id = self.config.Cache.get_ids_from_imdb(imdb_id)
update_tmdb = False
if not tmdb_id:
tmdb_id, update_tmdb = self.config.Cache.get_tmdb_from_imdb(imdb_id)
if update_tmdb:
tmdb_id = None
update_tvdb = False
if not tvdb_id:
tvdb_id, update_tvdb = self.config.Cache.get_tvdb_from_imdb(imdb_id)
if update_tvdb:
tvdb_id = None
else:
tmdb_id = None
tvdb_id = None
from_cache = tmdb_id is not None or tvdb_id is not None
if not tmdb_id and not tvdb_id and self.config.TMDb:
try:
tmdb_id = self.config.TMDb.convert_imdb_to_tmdb(imdb_id)
except Failed:
pass
if not tmdb_id and not tvdb_id and self.config.TMDb:
try:
tvdb_id = self.config.TMDb.convert_imdb_to_tvdb(imdb_id)
except Failed:
pass
if not tmdb_id and not tvdb_id and self.config.Trakt:
try:
tmdb_id = self.config.Trakt.convert_imdb_to_tmdb(imdb_id)
except Failed:
pass
if not tmdb_id and not tvdb_id and self.config.Trakt:
try:
tvdb_id = self.config.Trakt.convert_imdb_to_tvdb(imdb_id)
except Failed:
pass
try:
if tmdb_id and not from_cache: self.config.TMDb.get_movie(tmdb_id)
except Failed: tmdb_id = None
try:
if tvdb_id and not from_cache: self.config.TVDb.get_series(language, tvdb_id)
except Failed: tvdb_id = None
if not tmdb_id and not tvdb_id: raise Failed(f"IMDb Error: No TMDb ID or TVDb ID found for IMDb: {imdb_id}")
if self.config.Cache:
if tmdb_id and update_tmdb is not False:
self.config.Cache.update_imdb("movie", update_tmdb, imdb_id, tmdb_id)
if tvdb_id and update_tvdb is not False:
self.config.Cache.update_imdb("show", update_tvdb, imdb_id, tvdb_id)
return tmdb_id, tvdb_id

181
modules/config.py
View file

@ -2,6 +2,7 @@ import glob, logging, os, re, requests, time
from modules import util
from modules.anidb import AniDBAPI
from modules.anilist import AniListAPI
from modules.arms import ArmsAPI
from modules.builder import CollectionBuilder
from modules.cache import Cache
from modules.imdb import IMDbAPI
@ -264,6 +265,7 @@ class Config:
self.TVDb = TVDbAPI(self)
self.IMDb = IMDbAPI(self)
self.AniDB = AniDBAPI(self)
self.Arms = ArmsAPI(self)
self.AniDBIDs = self.AniDB.get_AniDB_IDs()
self.AniList = AniListAPI(self)
self.Letterboxd = LetterboxdAPI(self)
@ -643,179 +645,6 @@ class Config:
continue
builder.run_collections_again(collection_obj, movie_map, show_map)
def covert_anilist_to_id(self, anilist_id, language):
return self.convert_anidb_to_id(self.anilist_to_anidb(anilist_id), language)
def convert_mal_to_id(self, mal_id, language):
return self.convert_anidb_to_id(self.mal_to_anidb(mal_id), language)
def convert_anidb_to_id(self, anidb_id, language):
try:
for imdb_id in self.convert_anidb_to_imdb(anidb_id):
tmdb_id, _ = self.convert_from_imdb(imdb_id, language)
if tmdb_id: return tmdb_id, None
else: raise Failed
except Failed:
try: return None, self.convert_anidb_to_tvdb(anidb_id)
except Failed: raise Failed(f"AniDB Error: No TVDb ID or IMDb ID found for AniDB ID: {anidb_id}")
def convert_anidb_list(self, anidb_list, language):
show_ids = []
movie_ids = []
for anidb_id in anidb_list:
try:
tmdb_id, tvdb_id = self.convert_anidb_to_id(anidb_id, language)
if tmdb_id:
movie_ids.append(tmdb_id)
if tvdb_id:
show_ids.append(tvdb_id)
except Failed as e:
logger.error(e)
return movie_ids, show_ids
def convert_anilist_list(self, anilist_list, language):
return self.convert_anidb_list(self.convert_anilist_to_anidb(anilist_list), language)
def convert_myanimelist_list(self, myanimelist_list, language):
return self.convert_anidb_list(self.convert_myanimelist_to_anidb(myanimelist_list), language)
def convert_anidb_to_tvdb(self, anidb_id): return self.convert_anidb(anidb_id, "anidbid", "tvdbid")
def convert_anidb_to_imdb(self, anidb_id): return self.convert_anidb(anidb_id, "anidbid", "imdbid")
def convert_tvdb_to_anidb(self, tvdb_id): return self.convert_anidb(tvdb_id, "tvdbid", "anidbid")
def convert_imdb_to_anidb(self, imdb_id): return self.convert_anidb(imdb_id, "imdbid", "anidbid")
def convert_anidb(self, input_id, from_id, to_id):
ids = self.AniDBIDs.xpath(f"//anime[contains(@{from_id}, '{input_id}')]/@{to_id}")
if len(ids) > 0:
if from_id == "tvdbid": return [int(i) for i in ids]
if len(ids[0]) > 0:
try: return ids[0].split(",") if to_id == "imdbid" else int(ids[0])
except ValueError: raise Failed(f"AniDB Error: No {util.pretty_ids[to_id]} ID found for {util.pretty_ids[from_id]} ID: {input_id}")
else: raise Failed(f"AniDB Error: No {util.pretty_ids[to_id]} ID found for {util.pretty_ids[from_id]} ID: {input_id}")
else: raise Failed(f"AniDB Error: {util.pretty_ids[from_id]} ID: {input_id} not found")
def mal_to_anidb(self, mal_id):
if mal_id is None:
raise Failed(f"Convert Error: MyAnimeList ID is None")
anime_ids = self.convert_anime_ids(mal_ids=mal_id)
if anime_ids[0] is None:
raise Failed(f"Convert Error: MyAnimeList ID: {mal_id} does not exist")
return anime_ids[0]["anidb"]
def anilist_to_anidb(self, anilist_id):
if anilist_id is None:
raise Failed(f"Convert Error: AniList ID is None")
anime_ids = self.convert_anime_ids(anilist_ids=anilist_id)
if anime_ids[0] is None:
raise Failed(f"Convert Error: AniList ID: {anilist_id} does not exist")
return anime_ids[0]["anidb"]
def convert_anilist_to_anidb(self, anilist_ids):
anidb_ids = []
for id_set in self.convert_anime_ids(anilist_ids=anilist_ids):
if id_set["anidb"] is not None:
anidb_ids.append(id_set["anidb"])
else:
logger.error(f"Convert Error: AniDB ID not found for AniList ID: {id_set['anilist']}")
return anidb_ids
def convert_myanimelist_to_anidb(self, mal_ids):
anidb_ids = []
for id_set in self.convert_anime_ids(mal_ids=mal_ids):
if id_set["anidb"] is not None:
anidb_ids.append(id_set["anidb"])
else:
logger.error(f"Convert Error: AniDB ID not found for MyAnimeList ID: {id_set['myanimelist']}")
return anidb_ids
def convert_anime_ids(self, anilist_ids=None, anidb_ids=None, mal_ids=None):
all_ids = []
def collect_ids(ids, id_name):
if ids:
if isinstance(ids, list):
all_ids.extend([{id_name: a_id} for a_id in ids])
else:
all_ids.append({id_name: ids})
collect_ids(anilist_ids, "anilist")
collect_ids(anidb_ids, "anidb")
collect_ids(mal_ids, "myanimelist")
converted_ids = []
if self.Cache:
unconverted_ids = []
for anime_dict in all_ids:
for id_type, anime_id in anime_dict.items():
query_ids, update = self.Cache.query_anime_map(anime_id, id_type)
if not update and query_ids:
converted_ids.append(query_ids)
else:
unconverted_ids.append({id_type: anime_id})
else:
unconverted_ids = all_ids
for anime_ids in self.call_arm_server(unconverted_ids):
if anime_ids:
if self.Cache:
self.Cache.update_anime(False, anime_ids)
converted_ids.append(anime_ids)
return converted_ids
@retry(stop_max_attempt_number=6, wait_fixed=10000)
def call_arm_server(self, ids):
return requests.post("https://relations.yuna.moe/api/ids", json=ids).json()
def convert_from_imdb(self, imdb_id, language):
update_tmdb = False
update_tvdb = False
if self.Cache:
tmdb_id, tvdb_id = self.Cache.get_ids_from_imdb(imdb_id)
update_tmdb = False
if not tmdb_id:
tmdb_id, update_tmdb = self.Cache.get_tmdb_from_imdb(imdb_id)
if update_tmdb:
tmdb_id = None
update_tvdb = False
if not tvdb_id:
tvdb_id, update_tvdb = self.Cache.get_tvdb_from_imdb(imdb_id)
if update_tvdb:
tvdb_id = None
else:
tmdb_id = None
tvdb_id = None
from_cache = tmdb_id is not None or tvdb_id is not None
if not tmdb_id and not tvdb_id and self.TMDb:
try:
tmdb_id = self.TMDb.convert_imdb_to_tmdb(imdb_id)
except Failed:
pass
if not tmdb_id and not tvdb_id and self.TMDb:
try:
tvdb_id = self.TMDb.convert_imdb_to_tvdb(imdb_id)
except Failed:
pass
if not tmdb_id and not tvdb_id and self.Trakt:
try:
tmdb_id = self.Trakt.convert_imdb_to_tmdb(imdb_id)
except Failed:
pass
if not tmdb_id and not tvdb_id and self.Trakt:
try:
tvdb_id = self.Trakt.convert_imdb_to_tvdb(imdb_id)
except Failed:
pass
try:
if tmdb_id and not from_cache: self.TMDb.get_movie(tmdb_id)
except Failed: tmdb_id = None
try:
if tvdb_id and not from_cache: self.TVDb.get_series(language, tvdb_id)
except Failed: tvdb_id = None
if not tmdb_id and not tvdb_id: raise Failed(f"IMDb Error: No TMDb ID or TVDb ID found for IMDb: {imdb_id}")
if self.Cache:
if tmdb_id and update_tmdb is not False:
self.Cache.update_imdb("movie", update_tmdb, imdb_id, tmdb_id)
if tvdb_id and update_tvdb is not False:
self.Cache.update_imdb("show", update_tvdb, imdb_id, tvdb_id)
return tmdb_id, tvdb_id
def mass_metadata(self, library, movie_map, show_map):
length = 0
logger.info("")
@ -972,13 +801,13 @@ class Config:
if not error_message:
if mal_id and not anidb_id:
try: anidb_id = self.mal_to_anidb(mal_id)
try: anidb_id = self.Arms.mal_to_anidb(mal_id)
except Failed: pass
if anidb_id and not tvdb_id:
try: tvdb_id = self.convert_anidb_to_tvdb(anidb_id)
try: tvdb_id = self.Arms.anidb_to_tvdb(anidb_id)
except Failed: pass
if anidb_id and not imdb_id:
try: imdb_id = self.convert_anidb_to_imdb(anidb_id)
try: imdb_id = self.Arms.anidb_to_imdb(anidb_id)
except Failed: pass
if not tmdb_id and imdb_id:
if isinstance(imdb_id, list):

4
modules/imdb.py
View file

@ -100,7 +100,7 @@ class IMDbAPI:
if method == "imdb_id":
if status_message:
logger.info(f"Processing {pretty}: {data}")
tmdb_id, tvdb_id = self.config.convert_from_imdb(data, language)
tmdb_id, tvdb_id = self.config.Arms.imdb_to_ids(data, language)
if tmdb_id: movie_ids.append(tmdb_id)
if tvdb_id: show_ids.append(tvdb_id)
elif method == "imdb_list":
@ -113,7 +113,7 @@ class IMDbAPI:
for i, imdb_id in enumerate(imdb_ids, 1):
length = util.print_return(length, f"Converting IMDb ID {i}/{total_ids}")
try:
tmdb_id, tvdb_id = self.config.convert_from_imdb(imdb_id, language)
tmdb_id, tvdb_id = self.config.Arms.imdb_to_ids(imdb_id, language)
if tmdb_id: movie_ids.append(tmdb_id)
if tvdb_id: show_ids.append(tvdb_id)
except Failed as e: logger.warning(e)

2
modules/mal.py
View file

@ -219,7 +219,7 @@ class MyAnimeListAPI:
logger.info(f"Processing {pretty}: {data['limit']} Anime from {self.get_username() if data['username'] == '@me' else data['username']}'s {pretty_names[data['status']]} list sorted by {pretty_names[data['sort_by']]}")
else:
raise Failed(f"MyAnimeList Error: Method {method} not supported")
movie_ids, show_ids = self.config.convert_myanimelist_list(mal_ids, language)
movie_ids, show_ids = self.config.Arms.myanimelist_to_ids(mal_ids, language)
if status_message:
logger.debug(f"MyAnimeList IDs Found: {mal_ids}")
logger.debug(f"Shows Found: {show_ids}")

28
modules/tests.py
View file

@ -27,34 +27,6 @@ def anidb_tests(config):
if config.AniDB:
util.separator("AniDB Tests")
try:
config.convert_anidb_to_tvdb(69)
logger.info("Success | Convert AniDB to TVDb")
except Failed as e:
util.print_stacktrace()
logger.error(f"Failure | Convert AniDB to TVDb: {e}")
try:
config.convert_anidb_to_imdb(112)
logger.info("Success | Convert AniDB to IMDb")
except Failed as e:
util.print_stacktrace()
logger.error(f"Failure | Convert AniDB to IMDb: {e}")
try:
config.convert_tvdb_to_anidb(81797)
logger.info("Success | Convert TVDb to AniDB")
except Failed as e:
util.print_stacktrace()
logger.error(f"Failure | Convert TVDb to AniDB: {e}")
try:
config.convert_imdb_to_anidb("tt0245429")
logger.info("Success | Convert IMDb to AniDB")
except Failed as e:
util.print_stacktrace()
logger.error(f"Failure | Convert IMDb to AniDB: {e}")
try:
config.AniDB.get_items("anidb_id", 69, "en", status_message=False)
logger.info("Success | Get AniDB ID")

24
modules/tvdb.py
View file

@ -55,13 +55,17 @@ class TVDbObj:
if is_movie:
results = response.xpath("//*[text()='TheMovieDB.com']/@href")
if len(results) > 0:
try: tmdb_id = util.regex_first_int(results[0], "TMDb ID")
except Failed as e: logger.error(e)
try:
tmdb_id = util.regex_first_int(results[0], "TMDb ID")
except Failed as e:
logger.error(e)
if not tmdb_id:
results = response.xpath("//*[text()='IMDB']/@href")
if len(results) > 0:
try: tmdb_id, _ = TVDb.config.convert_from_imdb(util.get_id_from_imdb_url(results[0]), language)
except Failed as e: logger.error(e)
try:
tmdb_id, _ = TVDb.config.Arms.imdb_to_ids(util.get_id_from_imdb_url(results[0]), language)
except Failed as e:
logger.error(e)
self.tmdb_id = tmdb_id
self.tvdb_url = tvdb_url
self.language = language
@ -114,13 +118,17 @@ class TVDbAPI:
title = item.xpath(".//div[@class='col-xs-12 col-sm-9 mt-2']//a/text()")[0]
item_url = item.xpath(".//div[@class='col-xs-12 col-sm-9 mt-2']//a/@href")[0]
if item_url.startswith("/series/"):
try: show_ids.append(self.get_series(language, f"{self.site_url}{item_url}").id)
except Failed as e: logger.error(f"{e} for series {title}")
try:
show_ids.append(self.get_series(language, f"{self.site_url}{item_url}").id)
except Failed as e:
logger.error(f"{e} for series {title}")
elif item_url.startswith("/movies/"):
try:
tmdb_id = self.get_movie(language, f"{self.site_url}{item_url}").tmdb_id
if tmdb_id: movie_ids.append(tmdb_id)
else: raise Failed(f"TVDb Error: TMDb ID not found from TVDb URL: {tvdb_url}")
if tmdb_id:
movie_ids.append(tmdb_id)
else:
raise Failed(f"TVDb Error: TMDb ID not found from TVDb URL: {tvdb_url}")
except Failed as e:
logger.error(f"{e} for series {title}")
else: