diff --git a/fastanime/cli/auth/__init__.py b/fastanime/cli/auth/__init__.py new file mode 100644 index 0000000..5270aa7 --- /dev/null +++ b/fastanime/cli/auth/__init__.py @@ -0,0 +1,79 @@ +# In fastanime/cli/auth/manager.py +from __future__ import annotations + +import json +import logging +from typing import TYPE_CHECKING, Optional + +from ...core.exceptions import ConfigError +from ..constants import USER_DATA_PATH + +if TYPE_CHECKING: + from ...libs.api.types import UserProfile + +logger = logging.getLogger(__name__) + + +class CredentialsManager: + """ + Handles loading and saving of user credentials and profile data. + + This class abstracts the storage mechanism (currently a JSON file), + allowing for future changes (e.g., to a system keyring) without + affecting the rest of the application. + """ + + def __init__(self): + """Initializes the manager with the path to the user data file.""" + self.path = USER_DATA_PATH + + def load_user_profile(self) -> Optional[dict]: + """ + Loads the user profile data from the JSON file. + + Returns: + A dictionary containing user data, or None if the file doesn't exist + or is invalid. + """ + if not self.path.exists(): + return None + try: + with self.path.open("r", encoding="utf-8") as f: + return json.load(f) + except (json.JSONDecodeError, IOError) as e: + logger.error(f"Failed to load user credentials from {self.path}: {e}") + return None + + def save_user_profile(self, profile: UserProfile, token: str) -> None: + """ + Saves the user profile and token to the JSON file. + + Args: + profile: The generic UserProfile dataclass. + token: The authentication token string. + """ + user_data = { + "id": profile.id, + "name": profile.name, + "bannerImage": profile.banner_url, + "avatar": {"large": profile.avatar_url}, + "token": token, + } + try: + self.path.parent.mkdir(parents=True, exist_ok=True) + with self.path.open("w", encoding="utf-8") as f: + json.dump(user_data, f, indent=2) + logger.info(f"Successfully saved user credentials to {self.path}") + except IOError as e: + raise ConfigError(f"Could not save user credentials to {self.path}: {e}") + + def clear_user_profile(self) -> None: + """Deletes the user credentials file.""" + if self.path.exists(): + try: + self.path.unlink() + logger.info("Cleared user credentials.") + except IOError as e: + raise ConfigError( + f"Could not clear user credentials at {self.path}: {e}" + ) diff --git a/fastanime/cli/commands/anilist/subcommands/login.py b/fastanime/cli/commands/anilist/subcommands/login.py index 0cc6b8b..80e3ecc 100644 --- a/fastanime/cli/commands/anilist/subcommands/login.py +++ b/fastanime/cli/commands/anilist/subcommands/login.py @@ -1,76 +1,57 @@ -from typing import TYPE_CHECKING +from __future__ import annotations import click +from rich import print +from rich.prompt import Confirm, Prompt -if TYPE_CHECKING: - from ...config import Config +from .....cli.auth.manager import CredentialsManager -@click.command(help="Login to your anilist account") -@click.option("--status", "-s", help="Whether you are logged in or not", is_flag=True) -@click.option("--erase", "-e", help="Erase your login details", is_flag=True) -@click.pass_obj -def login(config: "Config", status, erase): - from os import path - from sys import exit - - from rich import print - from rich.prompt import Confirm, Prompt - - from ....constants import S_PLATFORM +@click.command(help="Login to your AniList account to enable progress tracking.") +@click.option("--status", "-s", is_flag=True, help="Check current login status.") +@click.option("--logout", "-l", is_flag=True, help="Log out and erase credentials.") +@click.pass_context +def login(ctx: click.Context, status: bool, logout: bool): + """Handles user authentication and credential management.""" + manager = CredentialsManager() if status: - is_logged_in = True if config.user else False - message = ( - "You are logged in :smile:" - if is_logged_in - else "You aren't logged in :cry:" - ) - print(message) - print(config.user) - exit(0) - elif erase: + user_data = manager.load_user_profile() + if user_data: + print(f"[bold green]Logged in as:[/] {user_data.get('name')}") + print(f"User ID: {user_data.get('id')}") + else: + print("[bold yellow]Not logged in.[/]") + return + + if logout: if Confirm.ask( - "Are you sure you want to erase your login status", default=False + "[bold red]Are you sure you want to log out and erase your token?[/]" ): - config.update_user({}) - print("Success") - exit(0) - else: - exit(1) + manager.clear_user_profile() + print("You have been logged out.") + return + + # --- Start Login Flow --- + from ....libs.api.factory import create_api_client + + api_client = create_api_client("anilist", ctx.obj) + + click.launch( + "https://anilist.co/api/v2/oauth/authorize?client_id=20148&response_type=token" + ) + print("Your browser has been opened to obtain an AniList token.") + print("After authorizing, copy the token from the address bar and paste it below.") + + token = Prompt.ask("Enter your AniList Access Token") + if not token.strip(): + print("[bold red]Login cancelled.[/]") + return + + profile = api_client.authenticate(token.strip()) + + if profile: + manager.save_user_profile(profile, token) + print(f"[bold green]Successfully logged in as {profile.name}! ✨[/]") else: - from click import launch - - from ....anilist import AniList - - if config.user: - print("Already logged in :confused:") - if not Confirm.ask("or would you like to reloggin", default=True): - exit(0) - # ---- new loggin ----- - print( - f"A browser session will be opened ( [link]{config.fastanime_anilist_app_login_url}[/link] )", - ) - token = "" - if S_PLATFORM.startswith("darwin"): - anilist_key_file_path = path.expanduser("~") + "/Downloads/anilist_key.txt" - launch(config.fastanime_anilist_app_login_url, wait=False) - Prompt.ask( - "MacOS detected.\nPress any key once the token provided has been pasted into " - + anilist_key_file_path - ) - with open(anilist_key_file_path) as key_file: - token = key_file.read().strip() - else: - launch(config.fastanime_anilist_app_login_url, wait=False) - token = Prompt.ask("Enter token") - user = AniList.login_user(token) - if not user: - print("Sth went wrong", user) - exit(1) - return - user["token"] = token - config.update_user(user) - print("Successfully saved credentials") - print(user) - exit(0) + print("[bold red]Login failed. The token may be invalid or expired.[/]") diff --git a/fastanime/cli/interactive/anilist/states/menu_states.py b/fastanime/cli/interactive/anilist/states/menu_states.py index fb71db1..8a2e1b4 100644 --- a/fastanime/cli/interactive/anilist/states/menu_states.py +++ b/fastanime/cli/interactive/anilist/states/menu_states.py @@ -1,17 +1,14 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING, Callable, Optional, Tuple +from .....libs.api.base import ApiSearchParams from .base import GoBack, State -from .task_states import ( - AnimeActionsState, - EpisodeSelectionState, - ProviderSearchState, - StreamPlaybackState, -) +from .task_states import AnimeActionsState if TYPE_CHECKING: + from .....libs.api.types import MediaSearchResult from ...session import Session from .. import ui @@ -24,46 +21,72 @@ class MainMenuState(State): def run(self, session: Session) -> Optional[State | type[GoBack]]: from .. import ui - menu_actions = { - "πŸ”₯ Trending": (session.anilist.get_trending, ResultsState()), - "πŸ”Ž Search": ( - lambda: session.anilist.search(query=ui.prompt_for_search(session)), + # Define actions as tuples: (Display Name, SearchParams, Next State) + # This centralizes the "business logic" of what each menu item means. + menu_actions: List[ + Tuple[str, Callable[[], Optional[ApiSearchParams]], Optional[State]] + ] = [ + ( + "πŸ”₯ Trending", + lambda: ApiSearchParams(sort="TRENDING_DESC"), ResultsState(), ), - "πŸ“Ί Watching": ( - lambda: session.anilist.get_anime_list("CURRENT"), + ( + "🌟 Most Popular", + lambda: ApiSearchParams(sort="POPULARITY_DESC"), ResultsState(), ), - "🌟 Most Popular": (session.anilist.get_most_popular, ResultsState()), - "πŸ’– Most Favourite": (session.anilist.get_most_favourite, ResultsState()), - "❌ Exit": (lambda: (True, None), None), - } + ( + "πŸ’– Most Favourite", + lambda: ApiSearchParams(sort="FAVOURITES_DESC"), + ResultsState(), + ), + ( + "πŸ”Ž Search", + lambda: ApiSearchParams(query=ui.prompt_for_search(session)), + ResultsState(), + ), + ( + "πŸ“Ί Watching", + lambda: session.api_client.fetch_user_list, + ResultsState(), + ), # Direct method call + ("❌ Exit", lambda: None, None), + ] - choice = ui.prompt_main_menu(session, list(menu_actions.keys())) + display_choices = [action[0] for action in menu_actions] + choice_str = ui.prompt_main_menu(session, display_choices) - if not choice: + if not choice_str: return None - data_loader, next_state = menu_actions[choice] - if not next_state: + # Find the chosen action + chosen_action = next( + (action for action in menu_actions if action[0] == choice_str), None + ) + if not chosen_action: + return self # Should not happen + + _, param_creator, next_state = chosen_action + + if not next_state: # Exit case return None - with ui.progress_spinner(f"Fetching {choice.strip('πŸ”₯πŸ”ŽπŸ“ΊπŸŒŸπŸ’–βŒ ')}..."): - success, data = data_loader() + # Execute the data fetch + with ui.progress_spinner(f"Fetching {choice_str.strip('πŸ”₯πŸ”ŽπŸ“ΊπŸŒŸπŸ’–βŒ ')}..."): + if choice_str == "πŸ“Ί Watching": # Special case for user list + result_data = param_creator(status="CURRENT") + else: + search_params = param_creator() + if search_params is None: # User cancelled search prompt + return self + result_data = session.api_client.search_media(search_params) - if not success or not data: - ui.display_error(f"Failed to fetch data. Reason: {data}") + if not result_data: + ui.display_error(f"Failed to fetch data for '{choice_str}'.") return self - if "mediaList" in data.get("data", {}).get("Page", {}): - data["data"]["Page"]["media"] = [ - item["media"] for item in data["data"]["Page"]["mediaList"] - ] - - session.state.anilist.results_data = data - session.state.navigation.current_page = 1 - # Store the data loader for pagination - session.current_data_loader = data_loader + session.state.anilist.results_data = result_data # Store the generic dataclass return next_state @@ -73,59 +96,20 @@ class ResultsState(State): def run(self, session: Session) -> Optional[State | type[GoBack]]: from .. import ui - if not session.state.anilist.results_data: + search_result = session.state.anilist.results_data + if not search_result or not isinstance(search_result, MediaSearchResult): ui.display_error("No results to display.") return GoBack - media_list = ( - session.state.anilist.results_data.get("data", {}) - .get("Page", {}) - .get("media", []) - ) - selection = ui.prompt_anime_selection(session, media_list) + selection = ui.prompt_anime_selection(session, search_result.media) if selection == "Back": return GoBack if selection is None: - return None # User cancelled prompt + return None - if selection == "Next Page": - page_info = ( - session.state.anilist.results_data.get("data", {}) - .get("Page", {}) - .get("pageInfo", {}) - ) - if page_info.get("hasNextPage"): - session.state.navigation.current_page += 1 - with ui.progress_spinner("Fetching next page..."): - success, data = session.current_data_loader( - page=session.state.navigation.current_page - ) - if success: - session.state.anilist.results_data = data - else: - ui.display_error("Failed to fetch next page.") - session.state.navigation.current_page -= 1 - else: - ui.display_error("Already on the last page.") - return self # Return to the same results state + # TODO: Implement pagination logic here by checking selection for "Next Page" etc. + # and re-calling the search_media method with an updated page number. - if selection == "Previous Page": - if session.state.navigation.current_page > 1: - session.state.navigation.current_page -= 1 - with ui.progress_spinner("Fetching previous page..."): - success, data = session.current_data_loader( - page=session.state.navigation.current_page - ) - if success: - session.state.anilist.results_data = data - else: - ui.display_error("Failed to fetch previous page.") - session.state.navigation.current_page += 1 - else: - ui.display_error("Already on the first page.") - return self - - # If it's a valid anime object session.state.anilist.selected_anime = selection return AnimeActionsState() diff --git a/fastanime/cli/interactive/session.py b/fastanime/cli/interactive/session.py index 47aad9e..662af00 100644 --- a/fastanime/cli/interactive/session.py +++ b/fastanime/cli/interactive/session.py @@ -7,53 +7,42 @@ from pydantic import BaseModel, Field if TYPE_CHECKING: from ...core.config import AppConfig - from ...libs.anilist.api import AniListApi - from ...libs.anilist.types import AnilistBaseMediaDataSchema - from ...libs.anime.provider import AnimeProvider - - # Import the dataclasses for type hinting - from ...libs.anime.types import Anime, SearchResult, SearchResults, Server + from ...libs.api.base import BaseApiClient + from ...libs.api.types import Anime, SearchResult, Server, UserProfile from ...libs.players.base import BasePlayer from ...libs.selector.base import BaseSelector logger = logging.getLogger(__name__) -# --- Nested State Models --- +# --- Nested State Models (Unchanged) --- class AnilistState(BaseModel): - """Holds state related to AniList data and selections.""" - - results_data: dict | None = None - selected_anime: Optional[AnilistBaseMediaDataSchema] = None + results_data: Optional[dict] = None + selected_anime: Optional[dict] = ( + None # Using dict for AnilistBaseMediaDataSchema for now + ) class ProviderState(BaseModel): - """Holds state related to the current anime provider, using specific dataclasses.""" - - search_results: Optional[SearchResults] = None selected_search_result: Optional[SearchResult] = None anime_details: Optional[Anime] = None current_episode: Optional[str] = None current_server: Optional[Server] = None + class Config: + arbitrary_types_allowed = True + class NavigationState(BaseModel): - """Holds state related to the UI navigation stack.""" - current_page: int = 1 history_stack_class_names: list[str] = Field(default_factory=list) class TrackingState(BaseModel): - """Holds state for user progress tracking preferences.""" - progress_mode: str = "prompt" -# --- Top-Level SessionState --- class SessionState(BaseModel): - """The root model for all serializable runtime state.""" - anilist: AnilistState = Field(default_factory=AnilistState) provider: ProviderState = Field(default_factory=ProviderState) navigation: NavigationState = Field(default_factory=NavigationState) @@ -64,41 +53,48 @@ class SessionState(BaseModel): class Session: - """ - Manages the entire runtime session for the interactive anilist command. - """ - - def __init__(self, config: AppConfig, anilist_client: AniListApi) -> None: + def __init__(self, config: AppConfig) -> None: self.config: AppConfig = config self.state: SessionState = SessionState() self.is_running: bool = True - self.anilist: AniListApi = anilist_client + self.user_profile: Optional[UserProfile] = None self._initialize_components() def _initialize_components(self) -> None: - """Creates instances of core components based on the current config.""" - from ...libs.anime.provider import create_provider + from ...cli.auth.manager import CredentialsManager + from ...libs.api.factory import create_api_client from ...libs.players import create_player from ...libs.selector import create_selector - logger.debug("Initializing session components from configuration...") + logger.debug("Initializing session components...") self.selector: BaseSelector = create_selector(self.config) self.provider: AnimeProvider = create_provider(self.config.general.provider) self.player: BasePlayer = create_player(self.config.stream.player, self.config) + # Instantiate and use the API factory + self.api_client: BaseApiClient = create_api_client("anilist", self.config) + + # Load credentials and authenticate the API client + manager = CredentialsManager() + user_data = manager.load_user_profile() + if user_data and (token := user_data.get("token")): + self.user_profile = self.api_client.authenticate(token) + if not self.user_profile: + logger.warning( + "Loaded token is invalid or expired. User is not logged in." + ) + def change_provider(self, provider_name: str) -> None: from ...libs.anime.provider import create_provider self.config.general.provider = provider_name self.provider = create_provider(provider_name) - logger.info(f"Provider changed to: {self.provider.__class__.__name__}") def change_player(self, player_name: str) -> None: from ...libs.players import create_player self.config.stream.player = player_name self.player = create_player(player_name, self.config) - logger.info(f"Player changed to: {self.player.__class__.__name__}") def stop(self) -> None: self.is_running = False diff --git a/fastanime/core/utils/graphql.py b/fastanime/core/utils/graphql.py index 2251f97..2fc07bc 100644 --- a/fastanime/core/utils/graphql.py +++ b/fastanime/core/utils/graphql.py @@ -1,26 +1,84 @@ -import json -from pathlib import Path +from __future__ import annotations -from httpx import AsyncClient, Client, Response -from typing_extensions import Counter +import json +import logging +from pathlib import Path +from typing import TYPE_CHECKING from .networking import TIMEOUT +if TYPE_CHECKING: + from httpx import Client + +logger = logging.getLogger(__name__) + + +def load_graphql_from_file(file: Path) -> str: + """ + Reads and returns the content of a .gql file. + + Args: + file: The Path object pointing to the .gql file. + + Returns: + The string content of the file. + """ + try: + return file.read_text(encoding="utf-8") + except FileNotFoundError: + logger.error(f"GraphQL file not found at: {file}") + raise + def execute_graphql_query( url: str, httpx_client: Client, graphql_file: Path, variables: dict -): - response = httpx_client.get( - url, - params={ - "variables": json.dumps(variables), - "query": load_graphql_from_file(graphql_file), - }, - timeout=TIMEOUT, - ) - return response +) -> dict | None: + """ + Executes a GraphQL query using a GET request with query parameters. + Suitable for read-only operations. + + Args: + url: The base GraphQL endpoint URL. + httpx_client: The httpx.Client instance to use. + graphql_file: Path to the .gql file containing the query. + variables: A dictionary of variables for the query. + + Returns: + The JSON response as a dictionary, or None on failure. + """ + query = load_graphql_from_file(graphql_file) + params = {"query": query, "variables": json.dumps(variables)} + try: + response = httpx_client.get(url, params=params, timeout=TIMEOUT) + response.raise_for_status() + return response.json() + except Exception as e: + logger.error(f"GraphQL GET request failed for {graphql_file.name}: {e}") + return None -def load_graphql_from_file(file: Path) -> str: - query = file.read_text(encoding="utf-8") - return query +def execute_graphql_mutation( + url: str, httpx_client: Client, graphql_file: Path, variables: dict +) -> dict | None: + """ + Executes a GraphQL mutation using a POST request with a JSON body. + Suitable for write/update operations. + + Args: + url: The GraphQL endpoint URL. + httpx_client: The httpx.Client instance to use. + graphql_file: Path to the .gql file containing the mutation. + variables: A dictionary of variables for the mutation. + + Returns: + The JSON response as a dictionary, or None on failure. + """ + query = load_graphql_from_file(graphql_file) + json_body = {"query": query, "variables": variables} + try: + response = httpx_client.post(url, json=json_body, timeout=TIMEOUT) + response.raise_for_status() + return response.json() + except Exception as e: + logger.error(f"GraphQL POST request failed for {graphql_file.name}: {e}") + return None diff --git a/fastanime/libs/anilist/api.py b/fastanime/libs/anilist/api.py deleted file mode 100644 index 62318e7..0000000 --- a/fastanime/libs/anilist/api.py +++ /dev/null @@ -1,472 +0,0 @@ -""" -This is the core module availing all the abstractions of the anilist api -""" - -import logging -import os -from typing import TYPE_CHECKING - -import requests - -from .queries_graphql import ( - airing_schedule_query, - anime_characters_query, - anime_query, - anime_relations_query, - delete_list_entry_query, - get_logged_in_user_query, - get_medialist_item_query, - get_user_info, - media_list_mutation, - media_list_query, - most_favourite_query, - most_popular_query, - most_recently_updated_query, - most_scored_query, - notification_query, - recommended_query, - search_query, - trending_query, - upcoming_anime_query, -) - -if TYPE_CHECKING: - from .types import ( - AnilistDataSchema, - AnilistMediaLists, - AnilistMediaListStatus, - AnilistNotifications, - AnilistUser_, - AnilistUserData, - AnilistViewerData, - ) -logger = logging.getLogger(__name__) -ANILIST_ENDPOINT = "https://graphql.anilist.co" - - -class AniListApi: - """An abstraction over the anilist api offering an easy and simple interface - - Attributes: - session: [TODO:attribute] - session: [TODO:attribute] - token: [TODO:attribute] - headers: [TODO:attribute] - user_id: [TODO:attribute] - token: [TODO:attribute] - headers: [TODO:attribute] - user_id: [TODO:attribute] - """ - - session: requests.Session - - def __init__(self) -> None: - self.session = requests.session() - - def login_user(self, token: str): - """method used to login a new user enabling authenticated requests - - Args: - token: anilist app token - - Returns: - the logged in user - """ - self.token = token - self.headers = {"Authorization": f"Bearer {self.token}"} - self.session.headers.update(self.headers) - success, user = self.get_logged_in_user() - if not user: - return - if not success or not user: - return - user_info: AnilistUser_ = user["data"]["Viewer"] - self.user_id = user_info["id"] - return user_info - - def get_notification( - self, - ) -> tuple[bool, "AnilistNotifications"] | tuple[bool, None]: - """get the top five latest notifications for anime thats airing - - Returns: - airing notifications - """ - return self._make_authenticated_request(notification_query) - - def update_login_info(self, user: "AnilistUser_", token: str): - """method used to login a user enabling authenticated requests - - Args: - user: an anilist user object - token: the login token - """ - self.token = token - self.headers = {"Authorization": f"Bearer {self.token}"} - self.session.headers.update(self.headers) - self.user_id = user["id"] - - def get_user_info(self) -> tuple[bool, "AnilistUserData"] | tuple[bool, None]: - """get the details of the user who is currently logged in - - Returns: - an anilist user - """ - - return self._make_authenticated_request(get_user_info, {"userId": self.user_id}) - - def get_logged_in_user( - self, - ) -> tuple[bool, "AnilistViewerData"] | tuple[bool, None]: - """get the details of the user who is currently logged in - - Returns: - an anilist user - """ - if not self.headers: - return (False, None) - return self._make_authenticated_request(get_logged_in_user_query) - - def update_anime_list(self, values_to_update: dict): - """a powerful method for managing mediaLists giving full power to the user - - Args: - values_to_update: a dict containing valid media list options - - Returns: - an anilist object indicating success - """ - variables = {"userId": self.user_id, **values_to_update} - return self._make_authenticated_request(media_list_mutation, variables) - - def get_anime_list( - self, - status: "AnilistMediaListStatus", - type="ANIME", - page=1, - perPage=os.environ.get("FASTANIME_PER_PAGE", 15), - **kwargs, - ) -> tuple[bool, "AnilistMediaLists"] | tuple[bool, None]: - """gets an anime list from your media list given the list status - - Args: - status: the mediaListStatus of the anime list - - Returns: - a media list - """ - variables = { - "status": status, - "userId": self.user_id, - "type": type, - "page": page, - "perPage": int(perPage), - } - return self._make_authenticated_request(media_list_query, variables) - - def get_medialist_entry( - self, mediaId: int - ) -> tuple[bool, dict] | tuple[bool, None]: - """Get the id entry of the items in an Anilist MediaList - - Args: - mediaId: The mediaList item entry mediaId - - Returns: - a boolean indicating whether the request succeeded and either a dict object containing the id of the media list entry - """ - variables = {"mediaId": mediaId} - return self._make_authenticated_request(get_medialist_item_query, variables) - - def delete_medialist_entry(self, mediaId: int): - """Deletes a mediaList item given its mediaId - - Args: - mediaId: the media id of the anime - - Returns: - a tuple containing a boolean whether the operation was successful and either an anilist object or none depending on success - """ - result = self.get_medialist_entry(mediaId) - data = result[1] - if not result[0] or not data: - return result - id = data["data"]["MediaList"]["id"] - variables = {"id": id} - return self._make_authenticated_request(delete_list_entry_query, variables) - - # TODO: unify the _make_authenticated_request with original since sessions are now in use - def _make_authenticated_request(self, query: str, variables: dict = {}): - """the abstraction over all authenticated requests - - Args: - query: the anilist query to make - variables: the anilist variables to use - - Returns: - an anilist object containing the queried data or none and a boolean indicating whether the request was successful - """ - try: - response = self.session.post( - ANILIST_ENDPOINT, - json={"query": query, "variables": variables}, - timeout=10, - headers=self.headers, - ) - anilist_data = response.json() - - # ensuring you dont get blocked - if ( - int(response.headers.get("X-RateLimit-Remaining", 0)) < 30 - and not response.status_code == 500 - ): - print( - "Warning you are exceeding the allowed number of calls per minute" - ) - logger.warning( - "You are exceeding the allowed number of calls per minute for the AniList api enforcing timeout" - ) - print("Forced timeout will now be initiated") - import time - - print("sleeping...") - time.sleep(1 * 60) - if response.status_code == 200: - return (True, anilist_data) - else: - return (False, anilist_data) - except requests.exceptions.Timeout: - logger.warning( - "Timeout has been exceeded this could mean anilist is down or you have lost internet connection" - ) - return (False, None) - except requests.exceptions.ConnectionError: - logger.warning( - "ConnectionError this could mean anilist is down or you have lost internet connection" - ) - return (False, None) - - except Exception as e: - logger.error(f"Something unexpected occurred {e}") - return (False, None) # type: ignore - - def get_data( - self, query: str, variables: dict = {} - ) -> tuple[bool, "AnilistDataSchema"]: - """the abstraction over all none authenticated requests and that returns data of a similar type - - Args: - query: the anilist query - variables: the anilist api variables - - Returns: - a boolean indicating success and none or an anilist object depending on success - """ - try: - response = self.session.post( - ANILIST_ENDPOINT, - json={"query": query, "variables": variables}, - timeout=10, - ) - anilist_data: AnilistDataSchema = response.json() - - # ensuring you dont get blocked - if ( - int(response.headers.get("X-RateLimit-Remaining", 0)) < 30 - and not response.status_code == 500 - ): - print( - "Warning you are exceeding the allowed number of calls per minute" - ) - logger.warning( - "You are exceeding the allowed number of calls per minute for the AniList api enforcing timeout" - ) - print("Forced timeout will now be initiated") - import time - - print("sleeping...") - time.sleep(1 * 60) - if response.status_code == 200: - return (True, anilist_data) - else: - return (False, anilist_data) - except requests.exceptions.Timeout: - logger.warning( - "Timeout has been exceeded this could mean anilist is down or you have lost internet connection" - ) - return ( - False, - { - "Error": "Timeout Exceeded for connection there might be a problem with your internet or anilist is down." - }, - ) # type: ignore - except requests.exceptions.ConnectionError: - logger.warning( - "ConnectionError this could mean anilist is down or you have lost internet connection" - ) - return ( - False, - { - "Error": "There might be a problem with your internet or anilist is down." - }, - ) # type: ignore - except Exception as e: - logger.error(f"Something unexpected occurred {e}") - return (False, {"Error": f"{e}"}) # type: ignore - - def search( - self, - max_results=50, - query: str | None = None, - sort: str | None = None, - genre_in: list[str] | None = None, - id_in: list[int] | None = None, - genre_not_in: list[str] = ["hentai"], - popularity_greater: int | None = None, - popularity_lesser: int | None = None, - averageScore_greater: int | None = None, - averageScore_lesser: int | None = None, - tag_in: list[str] | None = None, - tag_not_in: list[str] | None = None, - status: str | None = None, - status_in: list[str] | None = None, - status_not_in: list[str] | None = None, - endDate_greater: int | None = None, - endDate_lesser: int | None = None, - startDate_greater: int | None = None, - startDate_lesser: int | None = None, - startDate: str | None = None, - seasonYear: str | None = None, - page: int | None = None, - season: str | None = None, - format_in: list[str] | None = None, - on_list: bool | None = None, - type="ANIME", - **kwargs, - ): - """ - A powerful method abstracting all of anilist media queries - """ - variables = {} - for key, val in list(locals().items())[1:]: - if (val or val is False) and key not in ["variables"]: - variables[key] = val - search_results = self.get_data(search_query, variables=variables) - return search_results - - def get_anime(self, id: int): - """ - Gets a single anime by a valid anilist anime id - """ - variables = {"id": id} - return self.get_data(anime_query, variables) - - def get_trending( - self, - type="ANIME", - page=1, - perPage=os.environ.get("FASTANIME_PER_PAGE", 15), - *_, - **kwargs, - ): - """ - Gets the currently trending anime - """ - variables = {"type": type, "page": page, "perPage": int(perPage)} - trending = self.get_data(trending_query, variables) - return trending - - def get_most_favourite( - self, - type="ANIME", - page=1, - perPage=os.environ.get("FASTANIME_PER_PAGE", 15), - *_, - **kwargs, - ): - """ - Gets the most favoured anime on anilist - """ - variables = {"type": type, "page": page, "perPage": int(perPage)} - most_favourite = self.get_data(most_favourite_query, variables) - return most_favourite - - def get_most_scored( - self, - type="ANIME", - page=1, - perPage=os.environ.get("FASTANIME_PER_PAGE", 15), - *_, - **kwargs, - ): - """ - Gets most scored anime on anilist - """ - variables = {"type": type, "page": page, "perPage": int(perPage)} - most_scored = self.get_data(most_scored_query, variables) - return most_scored - - def get_most_recently_updated( - self, - type="ANIME", - page=1, - perPage=os.environ.get("FASTANIME_PER_PAGE", 15), - *_, - **kwargs, - ): - """ - Gets most recently updated anime from anilist - """ - variables = {"type": type, "page": page, "perPage": int(perPage)} - most_recently_updated = self.get_data(most_recently_updated_query, variables) - return most_recently_updated - - def get_most_popular( - self, - type="ANIME", - page=1, - perPage=os.environ.get("FASTANIME_PER_PAGE", 15), - **kwargs, - ): - """ - Gets most popular anime on anilist - """ - variables = {"type": type, "page": page, "perPage": int(perPage)} - most_popular = self.get_data(most_popular_query, variables) - return most_popular - - def get_upcoming_anime( - self, - type="ANIME", - page: int = 1, - perPage=os.environ.get("FASTANIME_PER_PAGE", 15), - *_, - **kwargs, - ): - """ - Gets upcoming anime from anilist - """ - variables = {"page": page, "type": type, "perPage": int(perPage)} - upcoming_anime = self.get_data(upcoming_anime_query, variables) - return upcoming_anime - - # NOTE: THe following methods will probably be scraped soon - def get_recommended_anime_for(self, mediaRecommendationId, page=1, *_, **kwargs): - variables = {"mediaRecommendationId": mediaRecommendationId, "page": page} - recommended_anime = self.get_data(recommended_query, variables) - return recommended_anime - - def get_characters_of(self, id: int, type="ANIME", *_, **kwargs): - variables = {"id": id} - characters = self.get_data(anime_characters_query, variables) - return characters - - def get_related_anime_for(self, id: int, *_, **kwargs): - variables = {"id": id} - related_anime = self.get_data(anime_relations_query, variables) - return related_anime - - def get_airing_schedule_for(self, id: int, type="ANIME", *_, **kwargs): - variables = {"id": id} - airing_schedule = self.get_data(airing_schedule_query, variables) - return airing_schedule diff --git a/fastanime/libs/anilist/gql.py b/fastanime/libs/anilist/gql.py deleted file mode 100644 index d53f34a..0000000 --- a/fastanime/libs/anilist/gql.py +++ /dev/null @@ -1,1134 +0,0 @@ -""" -This module contains all the preset queries for the sake of neatness and convenience -Mostly for internal usage -""" - -mark_as_read_mutation = """ -mutation{ - UpdateUser{ - unreadNotificationCount - } -} -""" -reviews_query = """ -query($id:Int){ - Page{ - pageInfo{ - total - } - reviews(mediaId:$id){ - summary - user{ - name - avatar { - large - medium - } - } - body - - } - } -} - -""" -notification_query = """ -query { - Page(perPage: 5) { - pageInfo { - total - } - notifications(resetNotificationCount: true, type: AIRING) { - ... on AiringNotification { - id - type - episode - contexts - createdAt - media { - id - idMal - title { - romaji - english - } - coverImage { - medium - } - } - } - } - } -} -""" - -get_medialist_item_query = """ -query ($mediaId: Int) { - MediaList(mediaId: $mediaId) { - id - } -} -""" - -delete_list_entry_query = """ -mutation ($id: Int) { - DeleteMediaListEntry(id: $id) { - deleted - } -} -""" - -get_logged_in_user_query = """ -query{ - Viewer{ - id - name - bannerImage - avatar { - large - medium - } - - } -} -""" - -get_user_info = """ -query ($userId: Int) { - User(id: $userId) { - name - about - avatar { - large - medium - } - bannerImage - statistics { - anime { - count - minutesWatched - episodesWatched - genres { - count - meanScore - genre - } - tags { - tag { - id - } - count - meanScore - } - } - manga { - count - meanScore - chaptersRead - volumesRead - tags { - count - meanScore - } - genres { - count - meanScore - } - } - } - favourites { - anime { - nodes { - title { - romaji - english - } - } - } - manga { - nodes { - title { - romaji - english - } - } - } - } - } -} -""" -media_list_mutation = """ -mutation ( - $mediaId: Int - $scoreRaw: Int - $repeat: Int - $progress: Int - $status: MediaListStatus -) { - SaveMediaListEntry( - mediaId: $mediaId - scoreRaw: $scoreRaw - progress: $progress - repeat: $repeat - status: $status - ) { - id - status - mediaId - score - progress - repeat - startedAt { - year - month - day - } - completedAt { - year - month - day - } - } -} -""" - -media_list_query = """ -query ($userId: Int, $status: MediaListStatus, $type: MediaType, $page: Int, $perPage: Int) { - Page(perPage: $perPage, page: $page) { - pageInfo { - currentPage - total - } - mediaList(userId: $userId, status: $status, type: $type) { - mediaId - media { - id - idMal - title { - romaji - english - } - coverImage { - medium - large - } - trailer { - site - id - } - popularity - streamingEpisodes { - title - thumbnail - } - favourites - averageScore - episodes - genres - synonyms - studios { - nodes { - name - isAnimationStudio - } - } - tags { - name - } - startDate { - year - month - day - } - endDate { - year - month - day - } - status - description - mediaListEntry { - status - id - progress - } - nextAiringEpisode { - timeUntilAiring - airingAt - episode - } - } - status - progress - score - repeat - notes - startedAt { - year - month - day - } - completedAt { - year - month - day - } - createdAt - } - } -} -""" - - -optional_variables = "\ -$max_results:Int,\ -$page:Int,\ -$sort:[MediaSort],\ -$id_in:[Int],\ -$genre_in:[String],\ -$genre_not_in:[String],\ -$tag_in:[String],\ -$tag_not_in:[String],\ -$status_in:[MediaStatus],\ -$status:MediaStatus,\ -$status_not_in:[MediaStatus],\ -$popularity_greater:Int,\ -$popularity_lesser:Int,\ -$averageScore_greater:Int,\ -$averageScore_lesser:Int,\ -$seasonYear:Int,\ -$startDate_greater:FuzzyDateInt,\ -$startDate_lesser:FuzzyDateInt,\ -$startDate:FuzzyDateInt,\ -$endDate_greater:FuzzyDateInt,\ -$endDate_lesser:FuzzyDateInt,\ -$format_in:[MediaFormat],\ -$type:MediaType\ -$season:MediaSeason\ -$on_list:Boolean\ -" - -search_query = ( - """ -query($query:String,%s){ - Page(perPage: $max_results, page: $page) { - pageInfo { - total - currentPage - hasNextPage - } - media( - search: $query - id_in: $id_in - genre_in: $genre_in - genre_not_in: $genre_not_in - tag_in: $tag_in - tag_not_in: $tag_not_in - status_in: $status_in - status: $status - startDate: $startDate - status_not_in: $status_not_in - popularity_greater: $popularity_greater - popularity_lesser: $popularity_lesser - averageScore_greater: $averageScore_greater - averageScore_lesser: $averageScore_lesser - startDate_greater: $startDate_greater - startDate_lesser: $startDate_lesser - endDate_greater: $endDate_greater - endDate_lesser: $endDate_lesser - format_in: $format_in - sort: $sort - season: $season - seasonYear: $seasonYear - type: $type - onList:$on_list - ) { - id - idMal - title { - romaji - english - } - coverImage { - medium - large - } - trailer { - site - id - } - mediaListEntry { - status - id - progress - } - popularity - streamingEpisodes { - title - thumbnail - } - favourites - averageScore - episodes - genres - synonyms - studios { - nodes { - name - isAnimationStudio - } - } - tags { - name - } - startDate { - year - month - day - } - endDate { - year - month - day - } - status - description - nextAiringEpisode { - timeUntilAiring - airingAt - episode - } - } - } -} -""" - % optional_variables -) - -trending_query = """ -query ($type: MediaType, $page: Int,$perPage:Int) { - Page(perPage: $perPage, page: $page) { - media(sort: TRENDING_DESC, type: $type, genre_not_in: ["hentai"]) { - id - idMal - title { - romaji - english - } - coverImage { - medium - large - } - trailer { - site - id - } - popularity - streamingEpisodes { - title - thumbnail - } - favourites - averageScore - genres - synonyms - episodes - description - studios { - nodes { - name - isAnimationStudio - } - } - tags { - name - } - startDate { - year - month - day - } - mediaListEntry { - status - id - progress - } - endDate { - year - month - day - } - status - nextAiringEpisode { - timeUntilAiring - airingAt - episode - } - } - } -} -""" - -# mosts -most_favourite_query = """ -query ($type: MediaType, $page: Int,$perPage:Int) { - Page(perPage: $perPage, page: $page) { - media(sort: FAVOURITES_DESC, type: $type, genre_not_in: ["hentai"]) { - id - idMal - title { - romaji - english - } - coverImage { - medium - large - } - trailer { - site - id - } - mediaListEntry { - status - id - progress - } - popularity - streamingEpisodes { - title - thumbnail - } - streamingEpisodes { - title - thumbnail - } - favourites - averageScore - episodes - description - genres - synonyms - studios { - nodes { - name - isAnimationStudio - } - } - tags { - name - } - startDate { - year - month - day - } - endDate { - year - month - day - } - status - nextAiringEpisode { - timeUntilAiring - airingAt - episode - } - } - } -} -""" - -most_scored_query = """ -query ($type: MediaType, $page: Int,$perPage:Int) { - Page(perPage: $perPage, page: $page) { - media(sort: SCORE_DESC, type: $type, genre_not_in: ["hentai"]) { - id - idMal - title { - romaji - english - } - coverImage { - medium - large - } - trailer { - site - id - } - mediaListEntry { - status - id - progress - } - popularity - streamingEpisodes { - title - thumbnail - } - episodes - favourites - averageScore - description - genres - synonyms - studios { - nodes { - name - isAnimationStudio - } - } - tags { - name - } - startDate { - year - month - day - } - endDate { - year - month - day - } - status - nextAiringEpisode { - timeUntilAiring - airingAt - episode - } - } - } -} -""" - -most_popular_query = """ -query ($type: MediaType, $page: Int,$perPage:Int) { - Page(perPage: $perPage, page: $page) { - media(sort: POPULARITY_DESC, type: $type, genre_not_in: ["hentai"]) { - id - idMal - title { - romaji - english - } - coverImage { - medium - large - } - trailer { - site - id - } - popularity - streamingEpisodes { - title - thumbnail - } - favourites - averageScore - description - episodes - genres - synonyms - mediaListEntry { - status - id - progress - } - studios { - nodes { - name - isAnimationStudio - } - } - tags { - name - } - startDate { - year - month - day - } - endDate { - year - month - day - } - status - nextAiringEpisode { - timeUntilAiring - airingAt - episode - } - } - } -} -""" - -most_recently_updated_query = """ -query ($type: MediaType, $page: Int,$perPage:Int) { - Page(perPage: $perPage, page: $page) { - media( - sort: UPDATED_AT_DESC - type: $type - averageScore_greater: 50 - genre_not_in: ["hentai"] - status: RELEASING - ) { - id - idMal - title { - romaji - english - } - coverImage { - medium - large - } - trailer { - site - id - } - mediaListEntry { - status - id - progress - } - popularity - streamingEpisodes { - title - thumbnail - } - - favourites - averageScore - description - genres - synonyms - episodes - studios { - nodes { - name - isAnimationStudio - } - } - tags { - name - } - startDate { - year - month - day - } - endDate { - year - month - day - } - status - nextAiringEpisode { - timeUntilAiring - airingAt - episode - } - } - } -} -""" - -recommended_query = """ -query ($mediaRecommendationId: Int, $page: Int) { - Page(perPage: 50, page: $page) { - recommendations(mediaRecommendationId: $mediaRecommendationId) { - media { - id - idMal - mediaListEntry { - status - id - progress - } - title { - english - romaji - native - } - coverImage { - medium - large - } - mediaListEntry { - status - id - progress - } - description - episodes - trailer { - site - id - } - genres - synonyms - averageScore - popularity - streamingEpisodes { - title - thumbnail - } - favourites - tags { - name - } - startDate { - year - month - day - } - endDate { - year - month - day - } - status - nextAiringEpisode { - timeUntilAiring - airingAt - episode - } - } - } - } -} -""" - - -anime_characters_query = """ -query ($id: Int, $type: MediaType) { - Page { - media(id: $id, type: $type) { - characters { - nodes { - name { - first - middle - last - full - native - } - image { - medium - large - } - description - gender - dateOfBirth { - year - month - day - } - age - bloodType - favourites - } - } - } - } -} -""" - - -anime_relations_query = """ -query ($id: Int) { - Media(id: $id) { - relations { - nodes { - id - idMal - type - title { - english - romaji - native - } - coverImage { - medium - large - } - mediaListEntry { - status - id - progress - } - description - episodes - trailer { - site - id - } - genres - synonyms - averageScore - popularity - streamingEpisodes { - title - thumbnail - } - favourites - tags { - name - } - startDate { - year - month - day - } - endDate { - year - month - day - } - status - nextAiringEpisode { - timeUntilAiring - airingAt - episode - } - } - } - } -} -""" - -airing_schedule_query = """ -query ($id: Int,$type:MediaType) { - Page { - media(id: $id, sort: POPULARITY_DESC, type: $type) { - airingSchedule(notYetAired:true){ - nodes{ - airingAt - timeUntilAiring - episode - - } - } - } - } -} -""" - -upcoming_anime_query = """ -query ($page: Int, $type: MediaType,$perPage:Int) { - Page(perPage: $perPage, page: $page) { - pageInfo { - total - perPage - currentPage - hasNextPage - } - media( - type: $type - status: NOT_YET_RELEASED - sort: POPULARITY_DESC - genre_not_in: ["hentai"] - ) { - id - idMal - title { - romaji - english - } - coverImage { - medium - large - } - trailer { - site - id - } - mediaListEntry { - status - id - progress - } - popularity - streamingEpisodes { - title - thumbnail - } - favourites - averageScore - genres - synonyms - episodes - description - studios { - nodes { - name - isAnimationStudio - } - } - tags { - name - } - startDate { - year - month - day - } - endDate { - year - month - day - } - status - nextAiringEpisode { - timeUntilAiring - airingAt - episode - } - } - } -} -""" - -anime_query = """ -query ($id: Int) { - Page { - media(id: $id) { - id - idMal - title { - romaji - english - } - mediaListEntry { - status - id - progress - } - nextAiringEpisode { - timeUntilAiring - airingAt - episode - } - coverImage { - extraLarge - } - characters(perPage: 5, sort: FAVOURITES_DESC) { - edges { - node { - name { - full - } - gender - dateOfBirth { - year - month - day - } - age - image { - medium - large - } - description - } - voiceActors { - name { - full - } - image { - medium - large - } - } - } - } - studios { - nodes { - name - isAnimationStudio - } - } - season - format - status - seasonYear - description - genres - synonyms - startDate { - year - month - day - } - endDate { - year - month - day - } - duration - countryOfOrigin - averageScore - popularity - streamingEpisodes { - title - thumbnail - } - - favourites - source - hashtag - siteUrl - tags { - name - rank - } - reviews(sort: SCORE_DESC, perPage: 3) { - nodes { - summary - user { - name - avatar { - medium - large - } - } - } - } - recommendations(sort: RATING_DESC, perPage: 10) { - nodes { - mediaRecommendation { - title { - romaji - english - } - } - } - } - relations { - nodes { - title { - romaji - english - native - } - } - } - externalLinks { - url - site - icon - } - rankings { - rank - context - } - bannerImage - episodes - } - } -} -""" diff --git a/fastanime/libs/anilist/mutations/delete-list-entry.gql b/fastanime/libs/api/__init__.py similarity index 100% rename from fastanime/libs/anilist/mutations/delete-list-entry.gql rename to fastanime/libs/api/__init__.py diff --git a/fastanime/libs/anilist/__init__.py b/fastanime/libs/api/anilist/__init__.py similarity index 100% rename from fastanime/libs/anilist/__init__.py rename to fastanime/libs/api/anilist/__init__.py diff --git a/fastanime/libs/api/anilist/api.py b/fastanime/libs/api/anilist/api.py new file mode 100644 index 0000000..1a0ea3c --- /dev/null +++ b/fastanime/libs/api/anilist/api.py @@ -0,0 +1,112 @@ +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Any, List, Optional + +from ....core.utils.graphql import execute_graphql_mutation, execute_graphql_query +from ..base import ApiSearchParams, BaseApiClient, UpdateListEntryParams, UserListParams +from ..types import MediaSearchResult, UserProfile +from . import gql, mapper + +if TYPE_CHECKING: + from httpx import Client + + from ....core.config import AnilistConfig + +logger = logging.getLogger(__name__) +ANILIST_ENDPOINT = "https://graphql.anilist.co" + + +class AniListApi(BaseApiClient): + """AniList API implementation of the BaseApiClient contract.""" + + def __init__(self, config: AnilistConfig, client: Client): + super().__init__(config, client) + self.token: Optional[str] = None + self.user_profile: Optional[UserProfile] = None + + def authenticate(self, token: str) -> Optional[UserProfile]: + self.token = token + self.http_client.headers["Authorization"] = f"Bearer {token}" + self.user_profile = self.get_viewer_profile() + if not self.user_profile: + self.token = None + self.http_client.headers.pop("Authorization", None) + return self.user_profile + + def get_viewer_profile(self) -> Optional[UserProfile]: + if not self.token: + return None + raw_data = execute_graphql_query( + ANILIST_ENDPOINT, self.http_client, gql.GET_LOGGED_IN_USER, {} + ) + return mapper.to_generic_user_profile(raw_data) if raw_data else None + + def search_media(self, params: ApiSearchParams) -> Optional[MediaSearchResult]: + variables = {k: v for k, v in params.__dict__.items() if v is not None} + variables["perPage"] = params.per_page + raw_data = execute_graphql_query( + ANILIST_ENDPOINT, self.http_client, gql.SEARCH_MEDIA, variables + ) + return mapper.to_generic_search_result(raw_data) if raw_data else None + + def fetch_user_list(self, params: UserListParams) -> Optional[MediaSearchResult]: + if not self.user_profile: + logger.error("Cannot fetch user list: user is not authenticated.") + return None + variables = { + "userId": self.user_profile.id, + "status": params.status, + "page": params.page, + "perPage": params.per_page, + } + raw_data = execute_graphql_query( + ANILIST_ENDPOINT, self.http_client, gql.GET_USER_LIST, variables + ) + return mapper.to_generic_user_list_result(raw_data) if raw_data else None + + def update_list_entry(self, params: UpdateListEntryParams) -> bool: + if not self.token: + return False + score_raw = int(params.score * 10) if params.score is not None else None + variables = { + "mediaId": params.media_id, + "status": params.status, + "progress": params.progress, + "scoreRaw": score_raw, + } + variables = {k: v for k, v in variables.items() if v is not None} + response = execute_graphql_mutation( + ANILIST_ENDPOINT, self.http_client, gql.SAVE_MEDIA_LIST_ENTRY, variables + ) + return response is not None and "errors" not in response + + def delete_list_entry(self, media_id: int) -> bool: + if not self.token: + return False + entry_data = execute_graphql_query( + ANILIST_ENDPOINT, + self.http_client, + gql.GET_MEDIA_LIST_ITEM, + {"mediaId": media_id}, + ) + list_id = ( + entry_data.get("data", {}).get("MediaList", {}).get("id") + if entry_data + else None + ) + if not list_id: + return False + response = execute_graphql_mutation( + ANILIST_ENDPOINT, + self.http_client, + gql.DELETE_MEDIA_LIST_ENTRY, + {"id": list_id}, + ) + return ( + response.get("data", {}) + .get("DeleteMediaListEntry", {}) + .get("deleted", False) + if response + else False + ) diff --git a/fastanime/libs/anilist/constants.py b/fastanime/libs/api/anilist/constants.py similarity index 100% rename from fastanime/libs/anilist/constants.py rename to fastanime/libs/api/anilist/constants.py diff --git a/fastanime/libs/api/anilist/gql.py b/fastanime/libs/api/anilist/gql.py new file mode 100644 index 0000000..9741108 --- /dev/null +++ b/fastanime/libs/api/anilist/gql.py @@ -0,0 +1,51 @@ +# -*- coding: utf-8 -*- +""" +GraphQL Path Registry for the AniList API Client. + +This module uses `importlib.resources` to create robust, cross-platform +`pathlib.Path` objects for every .gql file in the `queries` and `mutations` +directories. This provides a single, type-safe source of truth for all +GraphQL operations, making the codebase easier to maintain and validate. + +Constants are named to reflect the action they perform, e.g., +`SEARCH_MEDIA` points to the `search.gql` file. +""" + +from __future__ import annotations + +from importlib import resources +from pathlib import Path + +# --- Base Paths --- +# Safely access package data directories using the standard library. +_QUERIES_PATH = resources.files("fastanime.libs.api.anilist") / "queries" +_MUTATIONS_PATH = resources.files("fastanime.libs.api.anilist") / "mutations" + + +# --- Queries --- +# Each constant is a Path object pointing to a specific .gql query file. +GET_AIRING_SCHEDULE: Path = _QUERIES_PATH / "airing.gql" +GET_ANIME_DETAILS: Path = _QUERIES_PATH / "anime.gql" +GET_CHARACTERS: Path = _QUERIES_PATH / "character.gql" +GET_FAVOURITES: Path = _QUERIES_PATH / "favourite.gql" +GET_MEDIA_LIST_ITEM: Path = _QUERIES_PATH / "get-medialist-item.gql" +GET_LOGGED_IN_USER: Path = _QUERIES_PATH / "logged-in-user.gql" +GET_MEDIA_LIST: Path = _QUERIES_PATH / "media-list.gql" +GET_MEDIA_RELATIONS: Path = _QUERIES_PATH / "media-relations.gql" +GET_NOTIFICATIONS: Path = _QUERIES_PATH / "notifications.gql" +GET_POPULAR: Path = _QUERIES_PATH / "popular.gql" +GET_RECENTLY_UPDATED: Path = _QUERIES_PATH / "recently-updated.gql" +GET_RECOMMENDATIONS: Path = _QUERIES_PATH / "recommended.gql" +GET_REVIEWS: Path = _QUERIES_PATH / "reviews.gql" +GET_SCORES: Path = _QUERIES_PATH / "score.gql" +SEARCH_MEDIA: Path = _QUERIES_PATH / "search.gql" +GET_TRENDING: Path = _QUERIES_PATH / "trending.gql" +GET_UPCOMING: Path = _QUERIES_PATH / "upcoming.gql" +GET_USER_INFO: Path = _QUERIES_PATH / "user-info.gql" + + +# --- Mutations --- +# Each constant is a Path object pointing to a specific .gql mutation file. +DELETE_MEDIA_LIST_ENTRY: Path = _MUTATIONS_PATH / "delete-list-entry.gql" +MARK_NOTIFICATIONS_AS_READ: Path = _MUTATIONS_PATH / "mark-read.gql" +SAVE_MEDIA_LIST_ENTRY: Path = _MUTATIONS_PATH / "media-list.gql" diff --git a/fastanime/libs/api/anilist/mapper.py b/fastanime/libs/api/anilist/mapper.py new file mode 100644 index 0000000..0e0adee --- /dev/null +++ b/fastanime/libs/api/anilist/mapper.py @@ -0,0 +1,239 @@ +from __future__ import annotations + +import logging +from datetime import datetime +from typing import TYPE_CHECKING, List, Optional + +from ..types import ( + AiringSchedule, + MediaImage, + MediaItem, + MediaSearchResult, + MediaTag, + MediaTitle, + MediaTrailer, + PageInfo, + Studio, + UserListStatus, + UserProfile, +) + +if TYPE_CHECKING: + from .types import AnilistBaseMediaDataSchema, AnilistPageInfo, AnilistUser_ + +logger = logging.getLogger(__name__) + + +def _to_generic_media_title(anilist_title: Optional[dict]) -> MediaTitle: + """Maps an AniList title object to a generic MediaTitle.""" + if not anilist_title: + return MediaTitle() + return MediaTitle( + romaji=anilist_title.get("romaji"), + english=anilist_title.get("english"), + native=anilist_title.get("native"), + ) + + +def _to_generic_media_image(anilist_image: Optional[dict]) -> MediaImage: + """Maps an AniList image object to a generic MediaImage.""" + if not anilist_image: + return MediaImage() + return MediaImage( + medium=anilist_image.get("medium"), + large=anilist_image.get("large"), + extra_large=anilist_image.get("extraLarge"), + ) + + +def _to_generic_media_trailer( + anilist_trailer: Optional[dict], +) -> Optional[MediaTrailer]: + """Maps an AniList trailer object to a generic MediaTrailer.""" + if not anilist_trailer or not anilist_trailer.get("id"): + return None + return MediaTrailer( + id=anilist_trailer["id"], + site=anilist_trailer.get("site"), + thumbnail_url=anilist_trailer.get("thumbnail"), + ) + + +def _to_generic_airing_schedule( + anilist_schedule: Optional[dict], +) -> Optional[AiringSchedule]: + """Maps an AniList nextAiringEpisode object to a generic AiringSchedule.""" + if not anilist_schedule or not anilist_schedule.get("airingAt"): + return None + return AiringSchedule( + airing_at=datetime.fromtimestamp(anilist_schedule["airingAt"]), + episode=anilist_schedule.get("episode", 0), + ) + + +def _to_generic_studios(anilist_studios: Optional[dict]) -> List[Studio]: + """Maps AniList studio nodes to a list of generic Studio objects.""" + if not anilist_studios or not anilist_studios.get("nodes"): + return [] + return [ + Studio(id=s["id"], name=s["name"]) + for s in anilist_studios["nodes"] + if s.get("id") and s.get("name") + ] + + +def _to_generic_tags(anilist_tags: Optional[list[dict]]) -> List[MediaTag]: + """Maps a list of AniList tags to generic MediaTag objects.""" + if not anilist_tags: + return [] + return [ + MediaTag(name=t["name"], rank=t.get("rank")) + for t in anilist_tags + if t.get("name") + ] + + +def _to_generic_user_status( + anilist_list_entry: Optional[dict], +) -> Optional[UserListStatus]: + """Maps an AniList mediaListEntry to a generic UserListStatus.""" + if not anilist_list_entry: + return None + + score = anilist_list_entry.get("score") + + return UserListStatus( + status=anilist_list_entry.get("status"), + progress=anilist_list_entry.get("progress"), + score=score + if score is not None + else None, # AniList score is 0-10, matches our generic model + ) + + +def _to_generic_media_item(data: AnilistBaseMediaDataSchema) -> MediaItem: + """Maps a single AniList media schema to a generic MediaItem.""" + return MediaItem( + id=data["id"], + id_mal=data.get("idMal"), + type=data.get("type", "ANIME"), + title=_to_generic_media_title(data.get("title")), + status=data.get("status"), + format=data.get("format"), + cover_image=_to_generic_media_image(data.get("coverImage")), + banner_image=data.get("bannerImage"), + trailer=_to_generic_media_trailer(data.get("trailer")), + description=data.get("description"), + episodes=data.get("episodes"), + duration=data.get("duration"), + genres=data.get("genres", []), + tags=_to_generic_tags(data.get("tags")), + studios=_to_generic_studios(data.get("studios")), + synonyms=data.get("synonyms", []), + average_score=data.get("averageScore"), + popularity=data.get("popularity"), + favourites=data.get("favourites"), + next_airing=_to_generic_airing_schedule(data.get("nextAiringEpisode")), + user_list_status=_to_generic_user_status(data.get("mediaListEntry")), + ) + + +def _to_generic_page_info(data: AnilistPageInfo) -> PageInfo: + """Maps an AniList page info object to a generic PageInfo.""" + return PageInfo( + total=data.get("total", 0), + current_page=data.get("currentPage", 1), + has_next_page=data.get("hasNextPage", False), + per_page=data.get("perPage", 0), + ) + + +def to_generic_search_result(api_response: dict) -> Optional[MediaSearchResult]: + """ + Top-level mapper to convert a raw AniList search/list API response + into a generic MediaSearchResult object. + """ + if not api_response or "data" not in api_response: + logger.warning("Mapping failed: API response is missing 'data' key.") + return None + + page_data = api_response["data"].get("Page") + if not page_data: + logger.warning("Mapping failed: API response 'data' is missing 'Page' key.") + return None + + raw_media_list = page_data.get("media", []) + media_items: List[MediaItem] = [ + _to_generic_media_item(item) for item in raw_media_list if item + ] + page_info = _to_generic_page_info(page_data.get("pageInfo", {})) + + return MediaSearchResult(page_info=page_info, media=media_items) + + +def to_generic_user_list_result(api_response: dict) -> Optional[MediaSearchResult]: + """ + Mapper for user list queries where media data is nested inside a 'mediaList' key. + """ + if not api_response or "data" not in api_response: + return None + page_data = api_response["data"].get("Page") + if not page_data: + return None + + # Extract media objects from the 'mediaList' array + media_list_items = page_data.get("mediaList", []) + raw_media_list = [ + item.get("media") for item in media_list_items if item.get("media") + ] + + # Now that we have a standard list of media, we can reuse the main search result mapper + page_data["media"] = raw_media_list + return to_generic_search_result({"data": {"Page": page_data}}) + + +def to_generic_user_profile(api_response: dict) -> Optional[UserProfile]: + """Maps a raw AniList viewer response to a generic UserProfile.""" + if not api_response or "data" not in api_response: + return None + + viewer_data: Optional[AnilistUser_] = api_response["data"].get("Viewer") + if not viewer_data: + return None + + return UserProfile( + id=viewer_data["id"], + name=viewer_data["name"], + avatar_url=viewer_data.get("avatar", {}).get("large"), + banner_url=viewer_data.get("bannerImage"), + ) + + +def to_generic_relations(api_response: dict) -> Optional[List[MediaItem]]: + """Maps the 'relations' part of an API response.""" + if not api_response or "data" not in api_response: + return None + nodes = ( + api_response.get("data", {}) + .get("Media", {}) + .get("relations", {}) + .get("nodes", []) + ) + return [_to_generic_media_item(node) for node in nodes if node] + + +def to_generic_recommendations(api_response: dict) -> Optional[List[MediaItem]]: + """Maps the 'recommendations' part of an API response.""" + if not api_response or "data" not in api_response: + return None + recs = ( + api_response.get("data", {}) + .get("Media", {}) + .get("recommendations", {}) + .get("nodes", []) + ) + return [ + _to_generic_media_item(rec.get("mediaRecommendation")) + for rec in recs + if rec.get("mediaRecommendation") + ] diff --git a/fastanime/libs/anilist/mutations/mark-read.gql b/fastanime/libs/api/anilist/mutations/delete-list-entry.gql similarity index 100% rename from fastanime/libs/anilist/mutations/mark-read.gql rename to fastanime/libs/api/anilist/mutations/delete-list-entry.gql diff --git a/fastanime/libs/anilist/mutations/media-list.gql b/fastanime/libs/api/anilist/mutations/mark-read.gql similarity index 100% rename from fastanime/libs/anilist/mutations/media-list.gql rename to fastanime/libs/api/anilist/mutations/mark-read.gql diff --git a/fastanime/libs/anilist/queries/media-list.gql b/fastanime/libs/api/anilist/mutations/media-list.gql similarity index 100% rename from fastanime/libs/anilist/queries/media-list.gql rename to fastanime/libs/api/anilist/mutations/media-list.gql diff --git a/fastanime/libs/anilist/queries/airing.gql b/fastanime/libs/api/anilist/queries/airing.gql similarity index 100% rename from fastanime/libs/anilist/queries/airing.gql rename to fastanime/libs/api/anilist/queries/airing.gql diff --git a/fastanime/libs/anilist/queries/anime.gql b/fastanime/libs/api/anilist/queries/anime.gql similarity index 100% rename from fastanime/libs/anilist/queries/anime.gql rename to fastanime/libs/api/anilist/queries/anime.gql diff --git a/fastanime/libs/anilist/queries/character.gql b/fastanime/libs/api/anilist/queries/character.gql similarity index 100% rename from fastanime/libs/anilist/queries/character.gql rename to fastanime/libs/api/anilist/queries/character.gql diff --git a/fastanime/libs/anilist/queries/favourite.gql b/fastanime/libs/api/anilist/queries/favourite.gql similarity index 100% rename from fastanime/libs/anilist/queries/favourite.gql rename to fastanime/libs/api/anilist/queries/favourite.gql diff --git a/fastanime/libs/anilist/queries/get-medialist-item.gql b/fastanime/libs/api/anilist/queries/get-medialist-item.gql similarity index 100% rename from fastanime/libs/anilist/queries/get-medialist-item.gql rename to fastanime/libs/api/anilist/queries/get-medialist-item.gql diff --git a/fastanime/libs/anilist/queries/logged-in-user.gql b/fastanime/libs/api/anilist/queries/logged-in-user.gql similarity index 100% rename from fastanime/libs/anilist/queries/logged-in-user.gql rename to fastanime/libs/api/anilist/queries/logged-in-user.gql diff --git a/fastanime/libs/anilist/queries/media-relations.gql b/fastanime/libs/api/anilist/queries/media-list.gql similarity index 100% rename from fastanime/libs/anilist/queries/media-relations.gql rename to fastanime/libs/api/anilist/queries/media-list.gql diff --git a/fastanime/libs/anilist/queries/notifications.gql b/fastanime/libs/api/anilist/queries/media-relations.gql similarity index 100% rename from fastanime/libs/anilist/queries/notifications.gql rename to fastanime/libs/api/anilist/queries/media-relations.gql diff --git a/fastanime/libs/anilist/queries/popular.gql b/fastanime/libs/api/anilist/queries/notifications.gql similarity index 100% rename from fastanime/libs/anilist/queries/popular.gql rename to fastanime/libs/api/anilist/queries/notifications.gql diff --git a/fastanime/libs/anilist/queries/recently-updated.gql b/fastanime/libs/api/anilist/queries/popular.gql similarity index 100% rename from fastanime/libs/anilist/queries/recently-updated.gql rename to fastanime/libs/api/anilist/queries/popular.gql diff --git a/fastanime/libs/anilist/queries/recommended.gql b/fastanime/libs/api/anilist/queries/recently-updated.gql similarity index 100% rename from fastanime/libs/anilist/queries/recommended.gql rename to fastanime/libs/api/anilist/queries/recently-updated.gql diff --git a/fastanime/libs/anilist/queries/reviews.gql b/fastanime/libs/api/anilist/queries/recommended.gql similarity index 100% rename from fastanime/libs/anilist/queries/reviews.gql rename to fastanime/libs/api/anilist/queries/recommended.gql diff --git a/fastanime/libs/anilist/queries/score.gql b/fastanime/libs/api/anilist/queries/reviews.gql similarity index 100% rename from fastanime/libs/anilist/queries/score.gql rename to fastanime/libs/api/anilist/queries/reviews.gql diff --git a/fastanime/libs/anilist/queries/search.gql b/fastanime/libs/api/anilist/queries/score.gql similarity index 100% rename from fastanime/libs/anilist/queries/search.gql rename to fastanime/libs/api/anilist/queries/score.gql diff --git a/fastanime/libs/anilist/queries/trending.gql b/fastanime/libs/api/anilist/queries/search.gql similarity index 100% rename from fastanime/libs/anilist/queries/trending.gql rename to fastanime/libs/api/anilist/queries/search.gql diff --git a/fastanime/libs/anilist/queries/upcoming.gql b/fastanime/libs/api/anilist/queries/trending.gql similarity index 100% rename from fastanime/libs/anilist/queries/upcoming.gql rename to fastanime/libs/api/anilist/queries/trending.gql diff --git a/fastanime/libs/anilist/queries/user-info.gql b/fastanime/libs/api/anilist/queries/upcoming.gql similarity index 100% rename from fastanime/libs/anilist/queries/user-info.gql rename to fastanime/libs/api/anilist/queries/upcoming.gql diff --git a/fastanime/libs/api/anilist/queries/user-info.gql b/fastanime/libs/api/anilist/queries/user-info.gql new file mode 100644 index 0000000..e69de29 diff --git a/fastanime/libs/anilist/types.py b/fastanime/libs/api/anilist/types.py similarity index 100% rename from fastanime/libs/anilist/types.py rename to fastanime/libs/api/anilist/types.py diff --git a/fastanime/libs/api/base.py b/fastanime/libs/api/base.py new file mode 100644 index 0000000..69d6ae8 --- /dev/null +++ b/fastanime/libs/api/base.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import abc +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Literal, Optional + +from .types import MediaSearchResult, UserProfile + +if TYPE_CHECKING: + from httpx import Client + + from ...core.config import AnilistConfig # Import the specific config part + + +# --- Parameter Dataclasses (Unchanged) --- + + +@dataclass(frozen=True) +class ApiSearchParams: + query: Optional[str] = None + page: int = 1 + per_page: int = 20 + sort: Optional[str] = None + + +@dataclass(frozen=True) +class UserListParams: + status: Literal[ + "CURRENT", "PLANNING", "COMPLETED", "DROPPED", "PAUSED", "REPEATING" + ] + page: int = 1 + per_page: int = 20 + + +@dataclass(frozen=True) +class UpdateListEntryParams: + media_id: int + status: Optional[ + Literal["CURRENT", "PLANNING", "COMPLETED", "DROPPED", "PAUSED", "REPEATING"] + ] = None + progress: Optional[int] = None + score: Optional[float] = None + + +# --- Abstract Base Class (Simplified) --- + + +class BaseApiClient(abc.ABC): + """ + Abstract Base Class defining a generic contract for media database APIs. + """ + + # The constructor now expects a specific config model, not the whole AppConfig. + def __init__(self, config: AnilistConfig | Any, client: Client): + self.config = config + self.http_client = client + + # --- Authentication & User --- + @abc.abstractmethod + def authenticate(self, token: str) -> Optional[UserProfile]: + pass + + @abc.abstractmethod + def get_viewer_profile(self) -> Optional[UserProfile]: + pass + + # --- Media Browsing & Search --- + @abc.abstractmethod + def search_media(self, params: ApiSearchParams) -> Optional[MediaSearchResult]: + """Searches for media based on a query and other filters.""" + pass + + # Redundant fetch methods are REMOVED. + + # --- User List Management --- + @abc.abstractmethod + def fetch_user_list(self, params: UserListParams) -> Optional[MediaSearchResult]: + pass + + @abc.abstractmethod + def update_list_entry(self, params: UpdateListEntryParams) -> bool: + pass + + @abc.abstractmethod + def delete_list_entry(self, media_id: int) -> bool: + pass diff --git a/fastanime/libs/api/factory.py b/fastanime/libs/api/factory.py new file mode 100644 index 0000000..100488f --- /dev/null +++ b/fastanime/libs/api/factory.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import importlib +import logging +from typing import TYPE_CHECKING + +from httpx import Client +from yt_dlp.utils.networking import random_user_agent + +if TYPE_CHECKING: + from ...core.config import AppConfig + from .base import BaseApiClient + +logger = logging.getLogger(__name__) + +# Map the client name to its import path AND the config section it needs. +API_CLIENTS = { + "anilist": ("fastanime.libs.api.anilist.api.AniListApi", "anilist"), + # "jikan": ("fastanime.libs.jikan.api.JikanApi", "jikan"), # For the future +} + + +def create_api_client(client_name: str, config: AppConfig) -> BaseApiClient: + """ + Factory to create an instance of a specific API client, injecting only + the relevant section of the application configuration. + """ + if client_name not in API_CLIENTS: + raise ValueError(f"Unsupported API client: '{client_name}'") + + import_path, config_section_name = API_CLIENTS[client_name] + module_name, class_name = import_path.rsplit(".", 1) + + try: + module = importlib.import_module(module_name) + client_class = getattr(module, class_name) + except (ImportError, AttributeError) as e: + raise ImportError(f"Could not load API client '{client_name}': {e}") from e + + # Create a shared httpx client for the API + http_client = Client(headers={"User-Agent": random_user_agent()}) + + # Retrieve the specific config section from the main AppConfig + scoped_config = getattr(config, config_section_name) + + # Inject the scoped config into the client's constructor + return client_class(scoped_config, http_client) diff --git a/fastanime/libs/api/types.py b/fastanime/libs/api/types.py new file mode 100644 index 0000000..686cc3a --- /dev/null +++ b/fastanime/libs/api/types.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from typing import List, Literal, Optional + +# --- Generic Enums and Type Aliases --- + +MediaType = Literal["ANIME", "MANGA"] +MediaStatus = Literal[ + "FINISHED", "RELEASING", "NOT_YET_RELEASED", "CANCELLED", "HIATUS" +] +UserListStatusType = Literal[ + "CURRENT", "PLANNING", "COMPLETED", "DROPPED", "PAUSED", "REPEATING" +] + +# --- Generic Data Models --- + + +@dataclass(frozen=True) +class MediaImage: + """A generic representation of media imagery URLs.""" + + medium: Optional[str] = None + large: Optional[str] = None + extra_large: Optional[str] = None + + +@dataclass(frozen=True) +class MediaTitle: + """A generic representation of media titles.""" + + romaji: Optional[str] = None + english: Optional[str] = None + native: Optional[str] = None + + +@dataclass(frozen=True) +class MediaTrailer: + """A generic representation of a media trailer.""" + + id: str + site: str # e.g., "youtube" + thumbnail_url: Optional[str] = None + + +@dataclass(frozen=True) +class AiringSchedule: + """A generic representation of the next airing episode.""" + + airing_at: datetime + episode: int + + +@dataclass(frozen=True) +class Studio: + """A generic representation of an animation studio.""" + + id: int + name: str + + +@dataclass(frozen=True) +class MediaTag: + """A generic representation of a descriptive tag.""" + + name: str + rank: Optional[int] = None # Percentage relevance from 0-100 + + +@dataclass(frozen=True) +class UserListStatus: + """Generic representation of a user's list status for a media item.""" + + status: Optional[UserListStatusType] = None + progress: Optional[int] = None + score: Optional[float] = None # Standardized to a 0-10 scale + + +@dataclass(frozen=True) +class MediaItem: + """ + The definitive, backend-agnostic representation of a single media item. + This is the primary data model the application will interact with. + """ + + id: int + id_mal: Optional[int] = None + type: MediaType = "ANIME" + title: MediaTitle = field(default_factory=MediaTitle) + status: Optional[MediaStatus] = None + format: Optional[str] = None # e.g., TV, MOVIE, OVA + + cover_image: MediaImage = field(default_factory=MediaImage) + banner_image: Optional[str] = None + trailer: Optional[MediaTrailer] = None + + description: Optional[str] = None + episodes: Optional[int] = None + duration: Optional[int] = None # In minutes + genres: List[str] = field(default_factory=list) + tags: List[MediaTag] = field(default_factory=list) + studios: List[Studio] = field(default_factory=list) + synonyms: List[str] = field(default_factory=list) + + average_score: Optional[float] = None # Standardized to a 0-10 scale + popularity: Optional[int] = None + favourites: Optional[int] = None + + next_airing: Optional[AiringSchedule] = None + user_list_status: Optional[UserListStatus] = None + + +@dataclass(frozen=True) +class PageInfo: + """Generic pagination information.""" + + total: int + current_page: int + has_next_page: bool + per_page: int + + +@dataclass(frozen=True) +class MediaSearchResult: + """A generic representation of a page of media search results.""" + + page_info: PageInfo + media: List[MediaItem] = field(default_factory=list) + + +@dataclass(frozen=True) +class UserProfile: + """A generic representation of a user's profile.""" + + id: int + name: str + avatar_url: Optional[str] = None + banner_url: Optional[str] = None