mirror of
https://github.com/Benexl/FastAnime.git
synced 2025-12-12 15:50:01 -08:00
refactor: anilist subcommands
This commit is contained in:
171
fastanime/cli/commands/anilist/helpers.py
Normal file
171
fastanime/cli/commands/anilist/helpers.py
Normal file
@@ -0,0 +1,171 @@
|
||||
"""
|
||||
Common helper functions for anilist subcommands.
|
||||
"""
|
||||
|
||||
import json
|
||||
from typing import TYPE_CHECKING, Optional
|
||||
|
||||
import click
|
||||
from rich.progress import Progress
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastanime.core.config import AppConfig
|
||||
from fastanime.libs.api.base import BaseApiClient
|
||||
from fastanime.libs.api.types import MediaSearchResult
|
||||
|
||||
|
||||
def get_authenticated_api_client(config: "AppConfig") -> "BaseApiClient":
|
||||
"""
|
||||
Get an authenticated API client or raise an error if not authenticated.
|
||||
|
||||
Args:
|
||||
config: Application configuration
|
||||
|
||||
Returns:
|
||||
Authenticated API client
|
||||
|
||||
Raises:
|
||||
click.Abort: If user is not authenticated
|
||||
"""
|
||||
from fastanime.libs.api.factory import create_api_client
|
||||
from fastanime.cli.utils.feedback import create_feedback_manager
|
||||
|
||||
feedback = create_feedback_manager(config.general.icons)
|
||||
api_client = create_api_client(config.general.api_client, config)
|
||||
|
||||
# Check if user is authenticated by trying to get viewer profile
|
||||
try:
|
||||
user_profile = api_client.get_viewer_profile()
|
||||
if not user_profile:
|
||||
feedback.error(
|
||||
"Not authenticated",
|
||||
"Please run: fastanime anilist login"
|
||||
)
|
||||
raise click.Abort()
|
||||
except Exception:
|
||||
feedback.error(
|
||||
"Authentication check failed",
|
||||
"Please run: fastanime anilist login"
|
||||
)
|
||||
raise click.Abort()
|
||||
|
||||
return api_client
|
||||
|
||||
|
||||
def handle_media_search_command(
|
||||
config: "AppConfig",
|
||||
dump_json: bool,
|
||||
task_name: str,
|
||||
search_params_factory,
|
||||
empty_message: str
|
||||
):
|
||||
"""
|
||||
Generic handler for media search commands (trending, popular, recent, etc).
|
||||
|
||||
Args:
|
||||
config: Application configuration
|
||||
dump_json: Whether to output JSON instead of launching interactive mode
|
||||
task_name: Name to display in progress indicator
|
||||
search_params_factory: Function that returns ApiSearchParams
|
||||
empty_message: Message to show when no results found
|
||||
"""
|
||||
from fastanime.core.exceptions import FastAnimeError
|
||||
from fastanime.libs.api.factory import create_api_client
|
||||
from fastanime.cli.utils.feedback import create_feedback_manager
|
||||
|
||||
feedback = create_feedback_manager(config.general.icons)
|
||||
|
||||
try:
|
||||
# Create API client
|
||||
api_client = create_api_client(config.general.api_client, config)
|
||||
|
||||
# Fetch media
|
||||
with Progress() as progress:
|
||||
progress.add_task(task_name, total=None)
|
||||
search_params = search_params_factory(config)
|
||||
search_result = api_client.search_media(search_params)
|
||||
|
||||
if not search_result or not search_result.media:
|
||||
raise FastAnimeError(empty_message)
|
||||
|
||||
if dump_json:
|
||||
# Use Pydantic's built-in serialization
|
||||
print(json.dumps(search_result.model_dump(), indent=2))
|
||||
else:
|
||||
# Launch interactive session for browsing results
|
||||
from fastanime.cli.interactive.session import session
|
||||
|
||||
feedback.info(f"Found {len(search_result.media)} anime. Launching interactive mode...")
|
||||
session.load_menus_from_folder()
|
||||
session.run(config)
|
||||
|
||||
except FastAnimeError as e:
|
||||
feedback.error(f"Failed to fetch {task_name.lower()}", str(e))
|
||||
raise click.Abort()
|
||||
except Exception as e:
|
||||
feedback.error("Unexpected error occurred", str(e))
|
||||
raise click.Abort()
|
||||
|
||||
|
||||
def handle_user_list_command(
|
||||
config: "AppConfig",
|
||||
dump_json: bool,
|
||||
status: str,
|
||||
list_name: str
|
||||
):
|
||||
"""
|
||||
Generic handler for user list commands (watching, completed, planning, etc).
|
||||
|
||||
Args:
|
||||
config: Application configuration
|
||||
dump_json: Whether to output JSON instead of launching interactive mode
|
||||
status: The list status to fetch (CURRENT, COMPLETED, PLANNING, etc)
|
||||
list_name: Human-readable name for the list (e.g., "watching", "completed")
|
||||
"""
|
||||
from fastanime.core.exceptions import FastAnimeError
|
||||
from fastanime.libs.api.params import UserListParams
|
||||
from fastanime.cli.utils.feedback import create_feedback_manager
|
||||
|
||||
feedback = create_feedback_manager(config.general.icons)
|
||||
|
||||
# Validate status parameter
|
||||
valid_statuses = ["CURRENT", "PLANNING", "COMPLETED", "DROPPED", "PAUSED", "REPEATING"]
|
||||
if status not in valid_statuses:
|
||||
feedback.error(f"Invalid status: {status}", f"Valid statuses are: {valid_statuses}")
|
||||
raise click.Abort()
|
||||
|
||||
try:
|
||||
# Get authenticated API client
|
||||
api_client = get_authenticated_api_client(config)
|
||||
|
||||
# Fetch user's anime list
|
||||
with Progress() as progress:
|
||||
progress.add_task(f"Fetching your {list_name} list...", total=None)
|
||||
list_params = UserListParams(
|
||||
status=status, # type: ignore # We validated it above
|
||||
page=1,
|
||||
per_page=config.anilist.per_page or 50
|
||||
)
|
||||
user_list = api_client.fetch_user_list(list_params)
|
||||
|
||||
if not user_list or not user_list.media:
|
||||
feedback.info(f"You have no anime in your {list_name} list")
|
||||
return
|
||||
|
||||
if dump_json:
|
||||
# Use Pydantic's built-in serialization
|
||||
print(json.dumps(user_list.model_dump(), indent=2))
|
||||
else:
|
||||
# Launch interactive session for browsing results
|
||||
from fastanime.cli.interactive.session import session
|
||||
|
||||
feedback.info(f"Found {len(user_list.media)} anime in your {list_name} list. Launching interactive mode...")
|
||||
session.load_menus_from_folder()
|
||||
session.run(config)
|
||||
|
||||
except FastAnimeError as e:
|
||||
feedback.error(f"Failed to fetch {list_name} list", str(e))
|
||||
raise click.Abort()
|
||||
except Exception as e:
|
||||
feedback.error("Unexpected error occurred", str(e))
|
||||
raise click.Abort()
|
||||
@@ -3,7 +3,7 @@ from typing import TYPE_CHECKING
|
||||
import click
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ...config import Config
|
||||
from fastanime.core.config import AppConfig
|
||||
|
||||
|
||||
@click.command(help="View anime you completed")
|
||||
@@ -14,40 +14,12 @@ if TYPE_CHECKING:
|
||||
help="Only print out the results dont open anilist menu",
|
||||
)
|
||||
@click.pass_obj
|
||||
def completed(config: "Config", dump_json):
|
||||
from sys import exit
|
||||
def completed(config: "AppConfig", dump_json: bool):
|
||||
from ..helpers import handle_user_list_command
|
||||
|
||||
from ....anilist import AniList
|
||||
from ...utils.tools import FastAnimeRuntimeState
|
||||
|
||||
if not config.user:
|
||||
print("Not authenticated")
|
||||
print("Please run: fastanime anilist loggin")
|
||||
exit(1)
|
||||
anime_list = AniList.get_anime_list("COMPLETED")
|
||||
if not anime_list or not anime_list[1]:
|
||||
return
|
||||
if not anime_list[0]:
|
||||
return
|
||||
media = [
|
||||
mediaListItem["media"]
|
||||
for mediaListItem in anime_list[1]["data"]["Page"]["mediaList"]
|
||||
] # pyright:ignore
|
||||
anime_list[1]["data"]["Page"]["media"] = media # pyright:ignore
|
||||
if dump_json:
|
||||
import json
|
||||
|
||||
print(json.dumps(anime_list))
|
||||
else:
|
||||
from ...interfaces import anilist_interfaces
|
||||
|
||||
fastanime_runtime_state = FastAnimeRuntimeState()
|
||||
|
||||
fastanime_runtime_state.current_page = 1
|
||||
fastanime_runtime_state.current_data_loader = (
|
||||
lambda config, **kwargs: anilist_interfaces._handle_animelist(
|
||||
config, fastanime_runtime_state, "Completed", **kwargs
|
||||
)
|
||||
)
|
||||
fastanime_runtime_state.anilist_results_data = anime_list[1]
|
||||
anilist_interfaces.anilist_results_menu(config, fastanime_runtime_state)
|
||||
handle_user_list_command(
|
||||
config=config,
|
||||
dump_json=dump_json,
|
||||
status="COMPLETED",
|
||||
list_name="completed"
|
||||
)
|
||||
|
||||
@@ -3,7 +3,7 @@ from typing import TYPE_CHECKING
|
||||
import click
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastanime.cli.config import Config
|
||||
from fastanime.core.config import AppConfig
|
||||
|
||||
|
||||
@click.command(help="View anime you dropped")
|
||||
@@ -14,15 +14,15 @@ if TYPE_CHECKING:
|
||||
help="Only print out the results dont open anilist menu",
|
||||
)
|
||||
@click.pass_obj
|
||||
def dropped(config: "Config", dump_json):
|
||||
from sys import exit
|
||||
def dropped(config: "AppConfig", dump_json: bool):
|
||||
from ..helpers import handle_user_list_command
|
||||
|
||||
from ....anilist import AniList
|
||||
|
||||
if not config.user:
|
||||
print("Not authenticated")
|
||||
print("Please run: fastanime anilist loggin")
|
||||
exit(1)
|
||||
handle_user_list_command(
|
||||
config=config,
|
||||
dump_json=dump_json,
|
||||
status="DROPPED",
|
||||
list_name="dropped"
|
||||
)
|
||||
anime_list = AniList.get_anime_list("DROPPED")
|
||||
if not anime_list:
|
||||
exit(1)
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import click
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastanime.core.config import AppConfig
|
||||
|
||||
|
||||
@click.command(
|
||||
help="Fetch the top 15 most favourited anime from anilist",
|
||||
@@ -12,26 +17,22 @@ import click
|
||||
help="Only print out the results dont open anilist menu",
|
||||
)
|
||||
@click.pass_obj
|
||||
def favourites(config, dump_json):
|
||||
from ....anilist import AniList
|
||||
def favourites(config: "AppConfig", dump_json: bool):
|
||||
from fastanime.libs.api.params import ApiSearchParams
|
||||
from ..helpers import handle_media_search_command
|
||||
|
||||
anime_data = AniList.get_most_favourite()
|
||||
if anime_data[0]:
|
||||
if dump_json:
|
||||
import json
|
||||
def create_search_params(config):
|
||||
return ApiSearchParams(
|
||||
per_page=config.anilist.per_page or 15,
|
||||
sort=["FAVOURITES_DESC"]
|
||||
)
|
||||
|
||||
print(json.dumps(anime_data[1]))
|
||||
else:
|
||||
from ...interfaces.anilist_interfaces import anilist_results_menu
|
||||
from ...utils.tools import FastAnimeRuntimeState
|
||||
|
||||
fastanime_runtime_state = FastAnimeRuntimeState()
|
||||
|
||||
fastanime_runtime_state.current_page = 1
|
||||
fastanime_runtime_state.current_data_loader = AniList.get_most_favourite
|
||||
fastanime_runtime_state.anilist_results_data = anime_data[1]
|
||||
anilist_results_menu(config, fastanime_runtime_state)
|
||||
else:
|
||||
from sys import exit
|
||||
handle_media_search_command(
|
||||
config=config,
|
||||
dump_json=dump_json,
|
||||
task_name="Fetching most favourited anime...",
|
||||
search_params_factory=create_search_params,
|
||||
empty_message="No favourited anime found"
|
||||
)
|
||||
|
||||
exit(1)
|
||||
|
||||
@@ -2,6 +2,30 @@ from typing import TYPE_CHECKING
|
||||
|
||||
import click
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastanime.core.config import AppConfig
|
||||
|
||||
|
||||
@click.command(help="View anime you paused")
|
||||
@click.option(
|
||||
"--dump-json",
|
||||
"-d",
|
||||
is_flag=True,
|
||||
help="Only print out the results dont open anilist menu",
|
||||
)
|
||||
@click.pass_obj
|
||||
def paused(config: "AppConfig", dump_json: bool):
|
||||
from ..helpers import handle_user_list_command
|
||||
|
||||
handle_user_list_command(
|
||||
config=config,
|
||||
dump_json=dump_json,
|
||||
status="PAUSED",
|
||||
list_name="paused"
|
||||
)t TYPE_CHECKING
|
||||
|
||||
import click
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ...config import Config
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@ from typing import TYPE_CHECKING
|
||||
import click
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ...config import Config
|
||||
from fastanime.core.config import AppConfig
|
||||
|
||||
|
||||
@click.command(help="View anime you are planning on watching")
|
||||
@@ -14,40 +14,12 @@ if TYPE_CHECKING:
|
||||
help="Only print out the results dont open anilist menu",
|
||||
)
|
||||
@click.pass_obj
|
||||
def planning(config: "Config", dump_json):
|
||||
from sys import exit
|
||||
def planning(config: "AppConfig", dump_json: bool):
|
||||
from ..helpers import handle_user_list_command
|
||||
|
||||
from ....anilist import AniList
|
||||
|
||||
if not config.user:
|
||||
print("Not authenticated")
|
||||
print("Please run: fastanime anilist loggin")
|
||||
exit(1)
|
||||
anime_list = AniList.get_anime_list("PLANNING")
|
||||
if not anime_list:
|
||||
exit(1)
|
||||
if not anime_list[0] or not anime_list[1]:
|
||||
exit(1)
|
||||
media = [
|
||||
mediaListItem["media"]
|
||||
for mediaListItem in anime_list[1]["data"]["Page"]["mediaList"]
|
||||
] # pyright:ignore
|
||||
anime_list[1]["data"]["Page"]["media"] = media # pyright:ignore
|
||||
if dump_json:
|
||||
import json
|
||||
|
||||
print(json.dumps(anime_list[1]))
|
||||
else:
|
||||
from ...interfaces import anilist_interfaces
|
||||
from ...utils.tools import FastAnimeRuntimeState
|
||||
|
||||
fastanime_runtime_state = FastAnimeRuntimeState()
|
||||
|
||||
fastanime_runtime_state.current_page = 1
|
||||
fastanime_runtime_state.current_data_loader = (
|
||||
lambda config, **kwargs: anilist_interfaces._handle_animelist(
|
||||
config, fastanime_runtime_state, "Planned", **kwargs
|
||||
)
|
||||
)
|
||||
fastanime_runtime_state.anilist_results_data = anime_list[1]
|
||||
anilist_interfaces.anilist_results_menu(config, fastanime_runtime_state)
|
||||
handle_user_list_command(
|
||||
config=config,
|
||||
dump_json=dump_json,
|
||||
status="PLANNING",
|
||||
list_name="planning"
|
||||
)
|
||||
|
||||
@@ -1,8 +1,14 @@
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import click
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastanime.core.config import AppConfig
|
||||
|
||||
|
||||
@click.command(
|
||||
help="Fetch the top 15 most popular anime", short_help="View most popular anime"
|
||||
help="Fetch the top 15 most popular anime",
|
||||
short_help="View most popular anime"
|
||||
)
|
||||
@click.option(
|
||||
"--dump-json",
|
||||
@@ -11,26 +17,22 @@ import click
|
||||
help="Only print out the results dont open anilist menu",
|
||||
)
|
||||
@click.pass_obj
|
||||
def popular(config, dump_json):
|
||||
from ....anilist import AniList
|
||||
def popular(config: "AppConfig", dump_json: bool):
|
||||
from fastanime.libs.api.params import ApiSearchParams
|
||||
from ..helpers import handle_media_search_command
|
||||
|
||||
anime_data = AniList.get_most_popular()
|
||||
if anime_data[0]:
|
||||
if dump_json:
|
||||
import json
|
||||
def create_search_params(config):
|
||||
return ApiSearchParams(
|
||||
per_page=config.anilist.per_page or 15,
|
||||
sort=["POPULARITY_DESC"]
|
||||
)
|
||||
|
||||
print(json.dumps(anime_data[1]))
|
||||
else:
|
||||
from ...interfaces.anilist_interfaces import anilist_results_menu
|
||||
from ...utils.tools import FastAnimeRuntimeState
|
||||
|
||||
fastanime_runtime_state = FastAnimeRuntimeState()
|
||||
|
||||
fastanime_runtime_state.current_page = 1
|
||||
fastanime_runtime_state.current_data_loader = AniList.get_most_popular
|
||||
fastanime_runtime_state.anilist_results_data = anime_data[1]
|
||||
anilist_results_menu(config, fastanime_runtime_state)
|
||||
else:
|
||||
from sys import exit
|
||||
handle_media_search_command(
|
||||
config=config,
|
||||
dump_json=dump_json,
|
||||
task_name="Fetching popular anime...",
|
||||
search_params_factory=create_search_params,
|
||||
empty_message="No popular anime found"
|
||||
)
|
||||
|
||||
exit(1)
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import click
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastanime.core.config import AppConfig
|
||||
|
||||
|
||||
@click.command(
|
||||
help="Get random anime from anilist based on a range of anilist anime ids that are selected at random",
|
||||
@@ -12,28 +17,51 @@ import click
|
||||
help="Only print out the results dont open anilist menu",
|
||||
)
|
||||
@click.pass_obj
|
||||
def random_anime(config, dump_json):
|
||||
def random_anime(config: "AppConfig", dump_json: bool):
|
||||
import json
|
||||
import random
|
||||
from rich.progress import Progress
|
||||
|
||||
from ....anilist import AniList
|
||||
from fastanime.core.exceptions import FastAnimeError
|
||||
from fastanime.libs.api.factory import create_api_client
|
||||
from fastanime.libs.api.params import ApiSearchParams
|
||||
from fastanime.cli.utils.feedback import create_feedback_manager
|
||||
|
||||
random_anime = range(1, 100000)
|
||||
|
||||
random_anime = random.sample(random_anime, k=50)
|
||||
|
||||
anime_data = AniList.search(id_in=list(random_anime))
|
||||
|
||||
if anime_data[0]:
|
||||
feedback = create_feedback_manager(config.general.icons)
|
||||
|
||||
try:
|
||||
# Create API client
|
||||
api_client = create_api_client(config.general.api_client, config)
|
||||
|
||||
# Generate random IDs
|
||||
random_ids = random.sample(range(1, 100000), k=50)
|
||||
|
||||
# Search for random anime
|
||||
with Progress() as progress:
|
||||
progress.add_task("Fetching random anime...", total=None)
|
||||
search_params = ApiSearchParams(
|
||||
id_in=random_ids,
|
||||
per_page=50
|
||||
)
|
||||
search_result = api_client.search_media(search_params)
|
||||
|
||||
if not search_result or not search_result.media:
|
||||
raise FastAnimeError("No random anime found")
|
||||
|
||||
if dump_json:
|
||||
import json
|
||||
|
||||
print(json.dumps(anime_data[1]))
|
||||
# Use Pydantic's built-in serialization
|
||||
print(json.dumps(search_result.model_dump(), indent=2))
|
||||
else:
|
||||
from ...interfaces.anilist_interfaces import anilist_results_menu
|
||||
from ...utils.tools import FastAnimeRuntimeState
|
||||
|
||||
fastanime_runtime_state = FastAnimeRuntimeState()
|
||||
fastanime_runtime_state.anilist_results_data = anime_data[1]
|
||||
anilist_results_menu(config, fastanime_runtime_state)
|
||||
else:
|
||||
exit(1)
|
||||
# Launch interactive session for browsing results
|
||||
from fastanime.cli.interactive.session import session
|
||||
|
||||
feedback.info(f"Found {len(search_result.media)} random anime. Launching interactive mode...")
|
||||
session.load_menus_from_folder()
|
||||
session.run(config)
|
||||
|
||||
except FastAnimeError as e:
|
||||
feedback.error("Failed to fetch random anime", str(e))
|
||||
raise click.Abort()
|
||||
except Exception as e:
|
||||
feedback.error("Unexpected error occurred", str(e))
|
||||
raise click.Abort()
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import click
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastanime.core.config import AppConfig
|
||||
|
||||
|
||||
@click.command(
|
||||
help="Fetch the 15 most recently updated anime from anilist that are currently releasing",
|
||||
@@ -12,27 +17,24 @@ import click
|
||||
help="Only print out the results dont open anilist menu",
|
||||
)
|
||||
@click.pass_obj
|
||||
def recent(config, dump_json):
|
||||
from ....anilist import AniList
|
||||
def recent(config: "AppConfig", dump_json: bool):
|
||||
from fastanime.libs.api.params import ApiSearchParams
|
||||
from ..helpers import handle_media_search_command
|
||||
|
||||
anime_data = AniList.get_most_recently_updated()
|
||||
if anime_data[0]:
|
||||
if dump_json:
|
||||
import json
|
||||
def create_search_params(config):
|
||||
return ApiSearchParams(
|
||||
per_page=config.anilist.per_page or 15,
|
||||
sort=["UPDATED_AT_DESC"],
|
||||
status_in=["RELEASING"]
|
||||
)
|
||||
|
||||
print(json.dumps(anime_data[1]))
|
||||
else:
|
||||
from ...interfaces.anilist_interfaces import anilist_results_menu
|
||||
from ...utils.tools import FastAnimeRuntimeState
|
||||
|
||||
fastanime_runtime_state = FastAnimeRuntimeState()
|
||||
|
||||
fastanime_runtime_state.current_page = 1
|
||||
fastanime_runtime_state.current_data_loader = (
|
||||
AniList.get_most_recently_updated
|
||||
)
|
||||
fastanime_runtime_state.anilist_results_data = anime_data[1]
|
||||
anilist_results_menu(config, fastanime_runtime_state)
|
||||
handle_media_search_command(
|
||||
config=config,
|
||||
dump_json=dump_json,
|
||||
task_name="Fetching recently updated anime...",
|
||||
search_params_factory=create_search_params,
|
||||
empty_message="No recently updated anime found"
|
||||
)
|
||||
else:
|
||||
from sys import exit
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@ from typing import TYPE_CHECKING
|
||||
import click
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ...config import Config
|
||||
from fastanime.core.config import AppConfig
|
||||
|
||||
|
||||
@click.command(help="View anime you are rewatching")
|
||||
@@ -14,7 +14,15 @@ if TYPE_CHECKING:
|
||||
help="Only print out the results dont open anilist menu",
|
||||
)
|
||||
@click.pass_obj
|
||||
def rewatching(config: "Config", dump_json):
|
||||
def rewatching(config: "AppConfig", dump_json: bool):
|
||||
from ..helpers import handle_user_list_command
|
||||
|
||||
handle_user_list_command(
|
||||
config=config,
|
||||
dump_json=dump_json,
|
||||
status="REPEATING",
|
||||
list_name="rewatching"
|
||||
)
|
||||
from sys import exit
|
||||
|
||||
from ....anilist import AniList
|
||||
|
||||
@@ -1,8 +1,14 @@
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import click
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastanime.core.config import AppConfig
|
||||
|
||||
|
||||
@click.command(
|
||||
help="Fetch the 15 most scored anime", short_help="View most scored anime"
|
||||
help="Fetch the 15 most scored anime",
|
||||
short_help="View most scored anime"
|
||||
)
|
||||
@click.option(
|
||||
"--dump-json",
|
||||
@@ -11,26 +17,22 @@ import click
|
||||
help="Only print out the results dont open anilist menu",
|
||||
)
|
||||
@click.pass_obj
|
||||
def scores(config, dump_json):
|
||||
from ....anilist import AniList
|
||||
def scores(config: "AppConfig", dump_json: bool):
|
||||
from fastanime.libs.api.params import ApiSearchParams
|
||||
from ..helpers import handle_media_search_command
|
||||
|
||||
anime_data = AniList.get_most_scored()
|
||||
if anime_data[0]:
|
||||
if dump_json:
|
||||
import json
|
||||
def create_search_params(config):
|
||||
return ApiSearchParams(
|
||||
per_page=config.anilist.per_page or 15,
|
||||
sort=["SCORE_DESC"]
|
||||
)
|
||||
|
||||
print(json.dumps(anime_data[1]))
|
||||
else:
|
||||
from ...interfaces.anilist_interfaces import anilist_results_menu
|
||||
from ...utils.tools import FastAnimeRuntimeState
|
||||
|
||||
fastanime_runtime_state = FastAnimeRuntimeState()
|
||||
|
||||
fastanime_runtime_state.current_page = 1
|
||||
fastanime_runtime_state.current_data_loader = AniList.get_most_scored
|
||||
fastanime_runtime_state.anilist_results_data = anime_data[1]
|
||||
anilist_results_menu(config, fastanime_runtime_state)
|
||||
else:
|
||||
from sys import exit
|
||||
handle_media_search_command(
|
||||
config=config,
|
||||
dump_json=dump_json,
|
||||
task_name="Fetching highest scored anime...",
|
||||
search_params_factory=create_search_params,
|
||||
empty_message="No scored anime found"
|
||||
)
|
||||
|
||||
exit(1)
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import click
|
||||
|
||||
from ...utils.completion_functions import anime_titles_shell_complete
|
||||
from fastanime.cli.utils.completion_functions import anime_titles_shell_complete
|
||||
from .data import (
|
||||
genres_available,
|
||||
media_formats_available,
|
||||
@@ -11,6 +13,9 @@ from .data import (
|
||||
years_available,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastanime.core.config import AppConfig
|
||||
|
||||
|
||||
@click.command(
|
||||
help="Search for anime using anilists api and get top ~50 results",
|
||||
@@ -76,60 +81,68 @@ from .data import (
|
||||
)
|
||||
@click.pass_obj
|
||||
def search(
|
||||
config,
|
||||
title,
|
||||
dump_json,
|
||||
season,
|
||||
status,
|
||||
sort,
|
||||
genres,
|
||||
tags,
|
||||
media_format,
|
||||
year,
|
||||
on_list,
|
||||
config: "AppConfig",
|
||||
title: str,
|
||||
dump_json: bool,
|
||||
season: str,
|
||||
status: tuple,
|
||||
sort: str,
|
||||
genres: tuple,
|
||||
tags: tuple,
|
||||
media_format: tuple,
|
||||
year: str,
|
||||
on_list: bool,
|
||||
):
|
||||
from ....anilist import AniList
|
||||
import json
|
||||
from rich.progress import Progress
|
||||
|
||||
success, search_results = AniList.search(
|
||||
query=title,
|
||||
sort=sort,
|
||||
status_in=list(status),
|
||||
genre_in=list(genres),
|
||||
season=season,
|
||||
tag_in=list(tags),
|
||||
seasonYear=year,
|
||||
format_in=list(media_format),
|
||||
on_list=on_list,
|
||||
)
|
||||
if success:
|
||||
from fastanime.core.exceptions import FastAnimeError
|
||||
from fastanime.libs.api.factory import create_api_client
|
||||
from fastanime.libs.api.params import ApiSearchParams
|
||||
from fastanime.cli.utils.feedback import create_feedback_manager
|
||||
|
||||
feedback = create_feedback_manager(config.general.icons)
|
||||
|
||||
try:
|
||||
# Create API client
|
||||
api_client = create_api_client(config.general.api_client, config)
|
||||
|
||||
# Build search parameters
|
||||
search_params = ApiSearchParams(
|
||||
query=title,
|
||||
per_page=config.anilist.per_page or 50,
|
||||
sort=[sort] if sort else None,
|
||||
status_in=list(status) if status else None,
|
||||
genre_in=list(genres) if genres else None,
|
||||
tag_in=list(tags) if tags else None,
|
||||
format_in=list(media_format) if media_format else None,
|
||||
season=season,
|
||||
seasonYear=int(year) if year else None,
|
||||
on_list=on_list
|
||||
)
|
||||
|
||||
# Search for anime
|
||||
with Progress() as progress:
|
||||
progress.add_task("Searching anime...", total=None)
|
||||
search_result = api_client.search_media(search_params)
|
||||
|
||||
if not search_result or not search_result.media:
|
||||
raise FastAnimeError("No anime found matching your search criteria")
|
||||
|
||||
if dump_json:
|
||||
import json
|
||||
|
||||
print(json.dumps(search_results))
|
||||
# Use Pydantic's built-in serialization
|
||||
print(json.dumps(search_result.model_dump(), indent=2))
|
||||
else:
|
||||
from ...interfaces.anilist_interfaces import anilist_results_menu
|
||||
from ...utils.tools import FastAnimeRuntimeState
|
||||
|
||||
fastanime_runtime_state = FastAnimeRuntimeState()
|
||||
|
||||
fastanime_runtime_state.current_page = 1
|
||||
fastanime_runtime_state.current_data_loader = (
|
||||
lambda page=1, **kwargs: AniList.search(
|
||||
query=title,
|
||||
sort=sort,
|
||||
status_in=list(status),
|
||||
genre_in=list(genres),
|
||||
season=season,
|
||||
tag_in=list(tags),
|
||||
seasonYear=year,
|
||||
format_in=list(media_format),
|
||||
on_list=on_list,
|
||||
page=page,
|
||||
)
|
||||
)
|
||||
fastanime_runtime_state.anilist_results_data = search_results
|
||||
anilist_results_menu(config, fastanime_runtime_state)
|
||||
else:
|
||||
from sys import exit
|
||||
|
||||
exit(1)
|
||||
# Launch interactive session for browsing results
|
||||
from fastanime.cli.interactive.session import session
|
||||
|
||||
feedback.info(f"Found {len(search_result.media)} anime matching your search. Launching interactive mode...")
|
||||
session.load_menus_from_folder()
|
||||
session.run(config)
|
||||
|
||||
except FastAnimeError as e:
|
||||
feedback.error("Search failed", str(e))
|
||||
raise click.Abort()
|
||||
except Exception as e:
|
||||
feedback.error("Unexpected error occurred", str(e))
|
||||
raise click.Abort()
|
||||
|
||||
@@ -3,62 +3,83 @@ from typing import TYPE_CHECKING
|
||||
import click
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ...config import Config
|
||||
from fastanime.core.config import AppConfig
|
||||
|
||||
|
||||
@click.command(help="Print out your anilist stats")
|
||||
@click.pass_obj
|
||||
def stats(
|
||||
config: "Config",
|
||||
):
|
||||
def stats(config: "AppConfig"):
|
||||
import shutil
|
||||
import subprocess
|
||||
from sys import exit
|
||||
|
||||
from rich.console import Console
|
||||
|
||||
console = Console()
|
||||
from rich.markdown import Markdown
|
||||
from rich.panel import Panel
|
||||
|
||||
from ....anilist import AniList
|
||||
from fastanime.core.exceptions import FastAnimeError
|
||||
from fastanime.libs.api.factory import create_api_client
|
||||
from fastanime.cli.utils.feedback import create_feedback_manager
|
||||
|
||||
user_data = AniList.get_user_info()
|
||||
if not user_data[0] or not user_data[1]:
|
||||
print("Failed to get user info")
|
||||
print(user_data[1])
|
||||
exit(1)
|
||||
feedback = create_feedback_manager(config.general.icons)
|
||||
console = Console()
|
||||
|
||||
KITTEN_EXECUTABLE = shutil.which("kitten")
|
||||
if not KITTEN_EXECUTABLE:
|
||||
print("Kitten not found")
|
||||
exit(1)
|
||||
try:
|
||||
# Create API client and ensure authentication
|
||||
api_client = create_api_client(config.general.api_client, config)
|
||||
|
||||
if not api_client.user_profile:
|
||||
feedback.error(
|
||||
"Not authenticated",
|
||||
"Please run: fastanime anilist login"
|
||||
)
|
||||
raise click.Abort()
|
||||
|
||||
image_url = user_data[1]["data"]["User"]["avatar"]["medium"]
|
||||
user_name = user_data[1]["data"]["User"]["name"]
|
||||
about = user_data[1]["data"]["User"]["about"] or ""
|
||||
console.clear()
|
||||
image_x = int(console.size.width * 0.1)
|
||||
image_y = int(console.size.height * 0.1)
|
||||
img_w = console.size.width // 3
|
||||
img_h = console.size.height // 3
|
||||
image_process = subprocess.run(
|
||||
[
|
||||
KITTEN_EXECUTABLE,
|
||||
"icat",
|
||||
"--clear",
|
||||
"--place",
|
||||
f"{img_w}x{img_h}@{image_x}x{image_y}",
|
||||
image_url,
|
||||
],
|
||||
check=False,
|
||||
)
|
||||
if not image_process.returncode == 0:
|
||||
print("failed to get image from icat")
|
||||
exit(1)
|
||||
console.print(
|
||||
Panel(
|
||||
Markdown(about),
|
||||
title=user_name,
|
||||
user_profile = api_client.user_profile
|
||||
|
||||
# Check if kitten is available for image display
|
||||
KITTEN_EXECUTABLE = shutil.which("kitten")
|
||||
if not KITTEN_EXECUTABLE:
|
||||
feedback.warning("Kitten not found - profile image will not be displayed")
|
||||
else:
|
||||
# Display profile image using kitten icat
|
||||
if user_profile.avatar_url:
|
||||
console.clear()
|
||||
image_x = int(console.size.width * 0.1)
|
||||
image_y = int(console.size.height * 0.1)
|
||||
img_w = console.size.width // 3
|
||||
img_h = console.size.height // 3
|
||||
|
||||
image_process = subprocess.run(
|
||||
[
|
||||
KITTEN_EXECUTABLE,
|
||||
"icat",
|
||||
"--clear",
|
||||
"--place",
|
||||
f"{img_w}x{img_h}@{image_x}x{image_y}",
|
||||
user_profile.avatar_url,
|
||||
],
|
||||
check=False,
|
||||
)
|
||||
|
||||
if image_process.returncode != 0:
|
||||
feedback.warning("Failed to display profile image")
|
||||
|
||||
# Display user information
|
||||
about_text = getattr(user_profile, 'about', '') or "No description available"
|
||||
|
||||
console.print(
|
||||
Panel(
|
||||
Markdown(about_text),
|
||||
title=f"📊 {user_profile.name}'s Profile",
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
# You can add more stats here if the API provides them
|
||||
feedback.success("User profile displayed successfully")
|
||||
|
||||
except FastAnimeError as e:
|
||||
feedback.error("Failed to fetch user stats", str(e))
|
||||
raise click.Abort()
|
||||
except Exception as e:
|
||||
feedback.error("Unexpected error occurred", str(e))
|
||||
raise click.Abort()
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import click
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastanime.core.config import AppConfig
|
||||
|
||||
|
||||
@click.command(
|
||||
help="Fetch the top 15 anime that are currently trending",
|
||||
@@ -12,26 +17,20 @@ import click
|
||||
help="Only print out the results dont open anilist menu",
|
||||
)
|
||||
@click.pass_obj
|
||||
def trending(config, dump_json):
|
||||
from ....anilist import AniList
|
||||
def trending(config: "AppConfig", dump_json: bool):
|
||||
from fastanime.libs.api.params import ApiSearchParams
|
||||
from ..helpers import handle_media_search_command
|
||||
|
||||
success, data = AniList.get_trending()
|
||||
if success:
|
||||
if dump_json:
|
||||
import json
|
||||
def create_search_params(config):
|
||||
return ApiSearchParams(
|
||||
per_page=config.anilist.per_page or 15,
|
||||
sort=["TRENDING_DESC"]
|
||||
)
|
||||
|
||||
print(json.dumps(data))
|
||||
else:
|
||||
from ...interfaces.anilist_interfaces import anilist_results_menu
|
||||
from ...utils.tools import FastAnimeRuntimeState
|
||||
|
||||
fastanime_runtime_state = FastAnimeRuntimeState()
|
||||
|
||||
fastanime_runtime_state.current_page = 1
|
||||
fastanime_runtime_state.current_data_loader = AniList.get_trending
|
||||
fastanime_runtime_state.anilist_results_data = data
|
||||
anilist_results_menu(config, fastanime_runtime_state)
|
||||
else:
|
||||
from sys import exit
|
||||
|
||||
exit(1)
|
||||
handle_media_search_command(
|
||||
config=config,
|
||||
dump_json=dump_json,
|
||||
task_name="Fetching trending anime...",
|
||||
search_params_factory=create_search_params,
|
||||
empty_message="No trending anime found"
|
||||
)
|
||||
|
||||
@@ -1,8 +1,14 @@
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import click
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastanime.core.config import AppConfig
|
||||
|
||||
|
||||
@click.command(
|
||||
help="Fetch the 15 most anticipated anime", short_help="View upcoming anime"
|
||||
help="Fetch the 15 most anticipated anime",
|
||||
short_help="View upcoming anime"
|
||||
)
|
||||
@click.option(
|
||||
"--dump-json",
|
||||
@@ -11,26 +17,23 @@ import click
|
||||
help="Only print out the results dont open anilist menu",
|
||||
)
|
||||
@click.pass_obj
|
||||
def upcoming(config, dump_json):
|
||||
from ....anilist import AniList
|
||||
def upcoming(config: "AppConfig", dump_json: bool):
|
||||
from fastanime.libs.api.params import ApiSearchParams
|
||||
from ..helpers import handle_media_search_command
|
||||
|
||||
success, data = AniList.get_upcoming_anime()
|
||||
if success:
|
||||
if dump_json:
|
||||
import json
|
||||
def create_search_params(config):
|
||||
return ApiSearchParams(
|
||||
per_page=config.anilist.per_page or 15,
|
||||
sort=["POPULARITY_DESC"],
|
||||
status_in=["NOT_YET_RELEASED"]
|
||||
)
|
||||
|
||||
print(json.dumps(data))
|
||||
else:
|
||||
from ...interfaces.anilist_interfaces import anilist_results_menu
|
||||
from ...utils.tools import FastAnimeRuntimeState
|
||||
|
||||
fastanime_runtime_state = FastAnimeRuntimeState()
|
||||
|
||||
fastanime_runtime_state.current_page = 1
|
||||
fastanime_runtime_state.current_data_loader = AniList.get_upcoming_anime
|
||||
fastanime_runtime_state.anilist_results_data = data
|
||||
anilist_results_menu(config, fastanime_runtime_state)
|
||||
else:
|
||||
from sys import exit
|
||||
handle_media_search_command(
|
||||
config=config,
|
||||
dump_json=dump_json,
|
||||
task_name="Fetching upcoming anime...",
|
||||
search_params_factory=create_search_params,
|
||||
empty_message="No upcoming anime found"
|
||||
)
|
||||
|
||||
exit(1)
|
||||
|
||||
@@ -3,7 +3,7 @@ from typing import TYPE_CHECKING
|
||||
import click
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ...config import Config
|
||||
from fastanime.core.config import AppConfig
|
||||
|
||||
|
||||
@click.command(help="View anime you are watching")
|
||||
@@ -14,40 +14,12 @@ if TYPE_CHECKING:
|
||||
help="Only print out the results dont open anilist menu",
|
||||
)
|
||||
@click.pass_obj
|
||||
def watching(config: "Config", dump_json):
|
||||
from sys import exit
|
||||
def watching(config: "AppConfig", dump_json: bool):
|
||||
from ..helpers import handle_user_list_command
|
||||
|
||||
from ....anilist import AniList
|
||||
|
||||
if not config.user:
|
||||
print("Not authenticated")
|
||||
print("Please run: fastanime anilist loggin")
|
||||
exit(1)
|
||||
anime_list = AniList.get_anime_list("CURRENT")
|
||||
if not anime_list:
|
||||
exit(1)
|
||||
if not anime_list[0] or not anime_list[1]:
|
||||
exit(1)
|
||||
media = [
|
||||
mediaListItem["media"]
|
||||
for mediaListItem in anime_list[1]["data"]["Page"]["mediaList"]
|
||||
] # pyright:ignore
|
||||
anime_list[1]["data"]["Page"]["media"] = media # pyright:ignore
|
||||
if dump_json:
|
||||
import json
|
||||
|
||||
print(json.dumps(anime_list[1]))
|
||||
else:
|
||||
from ...interfaces import anilist_interfaces
|
||||
from ...utils.tools import FastAnimeRuntimeState
|
||||
|
||||
fastanime_runtime_state = FastAnimeRuntimeState()
|
||||
|
||||
fastanime_runtime_state.current_page = 1
|
||||
fastanime_runtime_state.current_data_loader = (
|
||||
lambda config, **kwargs: anilist_interfaces._handle_animelist(
|
||||
config, fastanime_runtime_state, "Watching", **kwargs
|
||||
)
|
||||
)
|
||||
fastanime_runtime_state.anilist_results_data = anime_list[1]
|
||||
anilist_interfaces.anilist_results_menu(config, fastanime_runtime_state)
|
||||
handle_user_list_command(
|
||||
config=config,
|
||||
dump_json=dump_json,
|
||||
status="CURRENT",
|
||||
list_name="watching"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user