Plex-Meta-Manager/modules/mal.py

244 lines
12 KiB
Python
Raw Normal View History

import math, re, secrets, time, webbrowser
2021-01-20 21:37:59 +00:00
from modules import util
from modules.util import Failed, TimeoutExpired
from ruamel import yaml
logger = util.logger
2021-01-20 21:37:59 +00:00
2021-03-30 05:50:53 +00:00
builders = [
"mal_id", "mal_all", "mal_airing", "mal_upcoming", "mal_tv", "mal_ova", "mal_movie", "mal_special",
2021-08-17 00:07:35 +00:00
"mal_popular", "mal_favorite", "mal_season", "mal_suggested", "mal_userlist", "mal_genre", "mal_studio"
2021-03-30 05:50:53 +00:00
]
mal_ranked_name = {
2021-07-23 19:44:21 +00:00
"mal_all": "all", "mal_airing": "airing", "mal_upcoming": "upcoming", "mal_tv": "tv", "mal_ova": "ova",
"mal_movie": "movie", "mal_special": "special", "mal_popular": "bypopularity", "mal_favorite": "favorite"
2021-03-30 05:50:53 +00:00
}
2021-08-01 04:35:42 +00:00
mal_ranked_pretty = {
"mal_all": "MyAnimeList All", "mal_airing": "MyAnimeList Airing",
"mal_upcoming": "MyAnimeList Upcoming", "mal_tv": "MyAnimeList TV", "mal_ova": "MyAnimeList OVA",
"mal_movie": "MyAnimeList Movie", "mal_special": "MyAnimeList Special", "mal_popular": "MyAnimeList Popular",
2021-08-17 00:07:35 +00:00
"mal_favorite": "MyAnimeList Favorite", "mal_genre": "MyAnimeList Genre", "mal_studio": "MyAnimeList Studio"
2021-08-01 04:35:42 +00:00
}
2021-07-23 19:44:21 +00:00
season_sort_translation = {"score": "anime_score", "anime_score": "anime_score", "members": "anime_num_list_users", "anime_num_list_users": "anime_num_list_users"}
2021-07-23 18:45:49 +00:00
season_sort_options = ["score", "members"]
2021-03-30 05:50:53 +00:00
pretty_names = {
2021-07-23 19:44:21 +00:00
"anime_score": "Score", "list_score": "Score", "anime_num_list_users": "Members", "list_updated_at": "Last Updated",
"anime_title": "Title", "anime_start_date": "Start Date", "all": "All Anime", "watching": "Currently Watching",
"completed": "Completed", "on_hold": "On Hold", "dropped": "Dropped", "plan_to_watch": "Plan to Watch"
2021-03-30 05:50:53 +00:00
}
2021-07-23 18:45:49 +00:00
userlist_sort_translation = {
2021-07-23 19:44:21 +00:00
"score": "list_score", "list_score": "list_score",
"last_updated": "list_updated_at", "list_updated": "list_updated_at", "list_updated_at": "list_updated_at",
"title": "anime_title", "anime_title": "anime_title",
"start_date": "anime_start_date", "anime_start_date": "anime_start_date"
2021-03-30 05:50:53 +00:00
}
2021-07-23 18:45:49 +00:00
userlist_sort_options = ["score", "last_updated", "title", "start_date"]
2021-07-23 19:44:21 +00:00
userlist_status = ["all", "watching", "completed", "on_hold", "dropped", "plan_to_watch"]
2021-07-26 19:17:33 +00:00
base_url = "https://api.myanimelist.net"
jiken_base_url = "https://api.jikan.moe/v3"
2021-07-14 14:47:20 +00:00
urls = {
2021-07-26 19:17:33 +00:00
"oauth_token": f"https://myanimelist.net/v1/oauth2/token",
"oauth_authorize": f"https://myanimelist.net/v1/oauth2/authorize",
2021-07-23 19:44:21 +00:00
"ranking": f"{base_url}/v2/anime/ranking",
"season": f"{base_url}/v2/anime/season",
"suggestions": f"{base_url}/v2/anime/suggestions",
"user": f"{base_url}/v2/users"
2021-07-14 14:47:20 +00:00
}
2021-07-15 17:42:28 +00:00
2021-06-14 15:24:11 +00:00
class MyAnimeList:
2021-07-26 19:03:17 +00:00
def __init__(self, config, params):
2021-04-15 20:45:35 +00:00
self.config = config
2021-01-20 21:37:59 +00:00
self.client_id = params["client_id"]
self.client_secret = params["client_secret"]
self.config_path = params["config_path"]
2021-07-26 19:03:17 +00:00
self.authorization = params["authorization"]
logger.secret(self.client_secret)
2021-05-07 19:53:54 +00:00
if not self._save(self.authorization):
if not self._refresh():
self._authorization()
2021-01-20 21:37:59 +00:00
2021-05-07 19:53:54 +00:00
def _authorization(self):
2021-01-20 21:37:59 +00:00
code_verifier = secrets.token_urlsafe(100)[:128]
2021-07-14 14:47:20 +00:00
url = f"{urls['oauth_authorize']}?response_type=code&client_id={self.client_id}&code_challenge={code_verifier}"
2021-01-20 21:37:59 +00:00
logger.info("")
2021-02-24 06:44:06 +00:00
logger.info(f"Navigate to: {url}")
2021-01-20 21:37:59 +00:00
logger.info("")
logger.info("Login and click the Allow option. You will then be redirected to a localhost")
logger.info("url that most likely won't load, which is fine. Copy the URL and paste it below")
webbrowser.open(url, new=2)
try: url = util.logger_input("URL").strip()
except TimeoutExpired: raise Failed("Input Timeout: URL required.")
if not url: raise Failed("MyAnimeList Error: No input MyAnimeList code required.")
match = re.search("code=([^&]+)", str(url))
if not match:
raise Failed("MyAnimeList Error: Invalid URL")
code = match.group(1)
data = {
"client_id": self.client_id,
"client_secret": self.client_secret,
"code": code,
"code_verifier": code_verifier,
"grant_type": "authorization_code"
}
2021-05-07 19:53:54 +00:00
new_authorization = self._oauth(data)
2021-01-20 21:37:59 +00:00
if "error" in new_authorization:
raise Failed("MyAnimeList Error: Invalid code")
2021-05-07 19:53:54 +00:00
if not self._save(new_authorization):
2021-01-20 21:37:59 +00:00
raise Failed("MyAnimeList Error: New Authorization Failed")
2021-05-07 19:53:54 +00:00
def _check(self, authorization):
2021-01-20 21:37:59 +00:00
try:
2021-07-14 14:47:20 +00:00
self._request(urls["suggestions"], authorization=authorization)
2021-01-20 21:37:59 +00:00
return True
except Failed as e:
logger.debug(e)
return False
2021-05-07 19:53:54 +00:00
def _refresh(self):
2021-01-20 21:37:59 +00:00
if self.authorization and "refresh_token" in self.authorization and self.authorization["refresh_token"]:
logger.info("Refreshing Access Token...")
data = {
"client_id": self.client_id,
"client_secret": self.client_secret,
"refresh_token": self.authorization["refresh_token"],
"grant_type": "refresh_token"
}
2021-05-07 19:53:54 +00:00
refreshed_authorization = self._oauth(data)
return self._save(refreshed_authorization)
2021-01-20 21:37:59 +00:00
return False
2021-05-07 19:53:54 +00:00
def _save(self, authorization):
if authorization is not None and "access_token" in authorization and authorization["access_token"] and self._check(authorization):
2021-12-28 00:01:43 +00:00
if self.authorization != authorization and not self.config.read_only:
2021-01-20 21:37:59 +00:00
yaml.YAML().allow_duplicate_keys = True
config, ind, bsi = yaml.util.load_yaml_guess_indent(open(self.config_path, encoding="utf-8"))
2021-01-20 21:37:59 +00:00
config["mal"]["authorization"] = {
"access_token": authorization["access_token"],
"token_type": authorization["token_type"],
"expires_in": authorization["expires_in"],
"refresh_token": authorization["refresh_token"]
}
2021-02-24 06:44:06 +00:00
logger.info(f"Saving authorization information to {self.config_path}")
2021-01-20 21:37:59 +00:00
yaml.round_trip_dump(config, open(self.config_path, "w"), indent=ind, block_seq_indent=bsi)
self.authorization = authorization
return True
return False
2021-05-07 19:53:54 +00:00
def _oauth(self, data):
2021-07-14 14:47:20 +00:00
return self.config.post_json(urls["oauth_token"], data=data)
2021-01-20 21:37:59 +00:00
2021-05-07 19:53:54 +00:00
def _request(self, url, authorization=None):
token = authorization["access_token"] if authorization else self.authorization["access_token"]
logger.secret(token)
2021-11-03 14:38:43 +00:00
if self.config.trace_mode:
logger.debug(f"URL: {url}")
response = self.config.get_json(url, headers={"Authorization": f"Bearer {token}"})
2021-11-03 14:38:43 +00:00
if self.config.trace_mode:
logger.debug(f"Response: {response}")
2021-02-24 06:44:06 +00:00
if "error" in response: raise Failed(f"MyAnimeList Error: {response['error']}")
2021-01-20 21:37:59 +00:00
else: return response
def _jiken_request(self, url):
data = self.config.get_json(f"{jiken_base_url}{url}")
time.sleep(2)
return data
2021-05-07 19:53:54 +00:00
def _parse_request(self, url):
data = self._request(url)
2021-03-10 16:51:27 +00:00
return [d["node"]["id"] for d in data["data"]] if "data" in data else []
2021-01-20 21:37:59 +00:00
2021-05-07 19:53:54 +00:00
def _username(self):
2021-07-14 14:47:20 +00:00
return self._request(f"{urls['user']}/@me")["name"]
2021-01-20 21:37:59 +00:00
2021-05-07 19:53:54 +00:00
def _ranked(self, ranking_type, limit):
2021-07-14 14:47:20 +00:00
url = f"{urls['ranking']}?ranking_type={ranking_type}&limit={limit}"
2021-05-07 19:53:54 +00:00
return self._parse_request(url)
2021-01-20 21:37:59 +00:00
2021-05-07 19:53:54 +00:00
def _season(self, season, year, sort_by, limit):
2021-07-14 14:47:20 +00:00
url = f"{urls['season']}/{year}/{season}?sort={sort_by}&limit={limit}"
2021-05-07 19:53:54 +00:00
return self._parse_request(url)
2021-01-20 21:37:59 +00:00
2021-05-07 19:53:54 +00:00
def _suggestions(self, limit):
2021-07-14 14:47:20 +00:00
url = f"{urls['suggestions']}?limit={limit}"
2021-05-07 19:53:54 +00:00
return self._parse_request(url)
2021-01-20 21:37:59 +00:00
2021-05-07 19:53:54 +00:00
def _userlist(self, username, status, sort_by, limit):
2021-02-24 06:44:06 +00:00
final_status = "" if status == "all" else f"status={status}&"
2021-07-14 14:47:20 +00:00
url = f"{urls['user']}/{username}/animelist?{final_status}sort={sort_by}&limit={limit}"
2021-05-07 19:53:54 +00:00
return self._parse_request(url)
2021-01-20 21:37:59 +00:00
def _genre(self, genre_id, limit):
data = self._jiken_request(f"/genre/anime/{genre_id}")
if "item_count" not in data:
raise Failed(f"MyAnimeList Error: No MyAnimeList IDs for Genre ID: {genre_id}")
total_items = data["item_count"]
if total_items < limit or limit <= 0:
limit = total_items
mal_ids = []
2021-08-16 05:41:04 +00:00
num_of_pages = math.ceil(int(limit) / 100)
current_page = 1
chances = 0
while current_page <= num_of_pages:
if chances > 6:
logger.debug(data)
raise Failed("AniList Error: Connection Failed")
start_num = (current_page - 1) * 100 + 1
logger.ghost(f"Parsing Page {current_page}/{num_of_pages} {start_num}-{limit if current_page == num_of_pages else current_page * 100}")
2021-08-16 05:41:04 +00:00
if current_page > 1:
data = self._jiken_request(f"/genre/anime/{genre_id}/{current_page}")
if "anime" in data:
chances = 0
mal_ids.extend([anime["mal_id"] for anime in data["anime"]])
if len(mal_ids) > limit:
return mal_ids[:limit]
current_page += 1
else:
chances += 1
logger.exorcise()
return mal_ids
2021-08-17 00:07:35 +00:00
def _studio(self, studio_id, limit):
data = self._jiken_request(f"/producer/{studio_id}")
if "anime" not in data:
2021-08-17 00:07:35 +00:00
raise Failed(f"MyAnimeList Error: No MyAnimeList IDs for Studio ID: {studio_id}")
mal_ids = []
count = 1
while True:
if count > 1:
2021-08-17 00:07:35 +00:00
data = self._jiken_request(f"/producer/{studio_id}/{count}")
if "anime" not in data:
break
mal_ids.extend([anime["mal_id"] for anime in data["anime"]])
if len(mal_ids) > limit > 0:
return mal_ids[:limit]
count += 1
return mal_ids
2021-08-07 06:01:21 +00:00
def get_mal_ids(self, method, data):
2021-01-20 21:37:59 +00:00
if method == "mal_id":
2021-08-01 04:35:42 +00:00
logger.info(f"Processing MyAnimeList ID: {data}")
2021-01-20 21:37:59 +00:00
mal_ids = [data]
2021-03-30 05:50:53 +00:00
elif method in mal_ranked_name:
2021-08-01 04:35:42 +00:00
logger.info(f"Processing {mal_ranked_pretty[method]}: {data} Anime")
2021-05-07 19:53:54 +00:00
mal_ids = self._ranked(mal_ranked_name[method], data)
elif method == "mal_genre":
logger.info(f"Processing {mal_ranked_pretty[method]} ID: {data['genre_id']}")
mal_ids = self._genre(data["genre_id"], data["limit"])
2021-08-17 00:07:35 +00:00
elif method == "mal_studio":
logger.info(f"Processing {mal_ranked_pretty[method]} ID: {data['studio_id']}")
mal_ids = self._studio(data["studio_id"], data["limit"])
2021-01-20 21:37:59 +00:00
elif method == "mal_season":
2021-08-14 22:59:35 +00:00
logger.info(f"Processing MyAnimeList Season: {data['limit']} Anime from {data['season'].title()} {data['year']} sorted by {pretty_names[data['sort_by']]}")
2021-05-07 19:53:54 +00:00
mal_ids = self._season(data["season"], data["year"], data["sort_by"], data["limit"])
2021-01-20 21:37:59 +00:00
elif method == "mal_suggested":
2021-08-01 04:35:42 +00:00
logger.info(f"Processing MyAnimeList Suggested: {data} Anime")
2021-05-07 19:53:54 +00:00
mal_ids = self._suggestions(data)
2021-01-20 21:37:59 +00:00
elif method == "mal_userlist":
2021-08-01 04:35:42 +00:00
logger.info(f"Processing MyAnimeList Userlist: {data['limit']} Anime from {self._username() if data['username'] == '@me' else data['username']}'s {pretty_names[data['status']]} list sorted by {pretty_names[data['sort_by']]}")
2021-05-07 19:53:54 +00:00
mal_ids = self._userlist(data["username"], data["status"], data["sort_by"], data["limit"])
2021-01-20 21:37:59 +00:00
else:
2021-02-24 06:44:06 +00:00
raise Failed(f"MyAnimeList Error: Method {method} not supported")
2021-05-24 03:38:46 +00:00
logger.debug("")
2021-07-03 01:47:09 +00:00
logger.debug(f"{len(mal_ids)} MyAnimeList IDs Found: {mal_ids}")
2021-08-07 06:01:21 +00:00
return mal_ids