gitea-mirror/FastAnime
1
0
Fork 0
mirror of https://github.com/Benexl/FastAnime.git synced 2025-12-12 15:50:01 -08:00
Code Issues Packages Projects Releases Wiki Activity

feat: stabilize the interactive workflow

Browse Source
This commit is contained in:
Benexl
2025-07-14 20:09:57 +03:00
parent e8491e3723
commit d1dfddf290
24 changed files with 617 additions and 406 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
fastanime/cli/cli.py
View File

@@ -30,6 +30,7 @@ commands = {
"config": ".config",
"search": ".search",
"download": ".download",
"anilist": ".anilist",
}

3
fastanime/cli/commands/__init__.py
View File

@@ -1,5 +1,6 @@
from .anilist import anilist
from .config import config
from .download import download
from .search import search
__all__ = ["config", "search", "download"]
__all__ = ["config", "search", "download", "anilist"]

55
fastanime/cli/commands/anilist/cmd.py
View File

@@ -1,36 +1,15 @@
import click
from ...interactive.anilist.controller import InteractiveController
from ...interactive.session import session
# Import the new interactive components
from ...interactive.session import Session
from ...utils.lazyloader import LazyGroup
# Define your subcommands (this part remains the same)
commands = {
"trending": "trending.trending",
"recent": "recent.recent",
"search": "search.search",
# ... add all your other subcommands
}
@click.group(
lazy_subcommands=commands,
cls=LazyGroup(root="fastanime.cli.commands.anilist.subcommands"),
invoke_without_command=True,
help="A beautiful interface that gives you access to a complete streaming experience",
short_help="Access all streaming options",
epilog="""
\b
\b\bExamples:
# Launch the interactive TUI
fastanime anilist
\b
# Run a specific subcommand
fastanime anilist trending --dump-json
""",
)
@click.command(name="anilist")
@click.option(
"--resume", is_flag=True, help="Resume from the last session (Not yet implemented)."
)
@@ -40,35 +19,9 @@ def anilist(ctx: click.Context, resume: bool):
The entry point for the 'anilist' command. If no subcommand is invoked,
it launches the interactive TUI mode.
"""
from ....libs.anilist.api import AniListApi
config = ctx.obj
# Initialize the AniList API client.
anilist_client = AniListApi()
if user := getattr(config, "user", None): # Safely access user attribute
anilist_client.update_login_info(user, user["token"])
if ctx.invoked_subcommand is None:
# ---- LAUNCH INTERACTIVE MODE ----
# 1. Create the session object.
session = Session(config, anilist_client)
# 2. Handle resume logic (placeholder for now).
if resume:
click.echo(
"Resume functionality is not yet implemented in the new architecture.",
err=True,
)
# You would load session.state from a file here.
# 3. Initialize and run the controller.
controller = InteractiveController(session)
# Clear the screen for a clean TUI experience.
click.clear()
controller.run()
# Print a goodbye message on exit.
click.echo("Exiting FastAnime. Have a great day!")
session.load_menus_from_folder()
session.run(config)

18
fastanime/cli/interactive/menus/episodes.py
View File

@@ -1,13 +1,11 @@
from typing import TYPE_CHECKING
import click
from rich.console import Console
from ..session import Context, session
from ..state import ControlFlow, ProviderState, State
if TYPE_CHECKING:
pass
@session.menu
def episodes(ctx: Context, state: State) -> State | ControlFlow:
@@ -18,9 +16,11 @@ def episodes(ctx: Context, state: State) -> State | ControlFlow:
provider_anime = state.provider.anime
anilist_anime = state.media_api.anime
config = ctx.config
console = Console()
console.clear()
if not provider_anime or not anilist_anime:
click.echo("[bold red]Error: Anime details are missing.[/bold red]")
console.print("[bold red]Error: Anime details are missing.[/bold red]")
return ControlFlow.BACK
# Get the list of episode strings based on the configured translation type
@@ -28,15 +28,14 @@ def episodes(ctx: Context, state: State) -> State | ControlFlow:
provider_anime.episodes, config.stream.translation_type, []
)
if not available_episodes:
click.echo(
console.print(
f"[bold yellow]No '{config.stream.translation_type}' episodes found for this anime.[/bold yellow]"
)
return ControlFlow.BACK
chosen_episode: str | None = None
# --- "Continue from History" Logic ---
if config.stream.continue_from_watch_history:
if config.stream.continue_from_watch_history and False:
progress = (
anilist_anime.user_status.progress
if anilist_anime.user_status and anilist_anime.user_status.progress
@@ -64,7 +63,6 @@ def episodes(ctx: Context, state: State) -> State | ControlFlow:
f"[yellow]Could not find episode based on your watch history. Please select manually.[/yellow]"
)
# --- Manual Selection Logic ---
if not chosen_episode:
choices = [*sorted(available_episodes, key=float), "Back"]
@@ -72,7 +70,7 @@ def episodes(ctx: Context, state: State) -> State | ControlFlow:
# preview_command = get_episode_preview(...)
chosen_episode_str = ctx.selector.choose(
prompt="Select Episode", choices=choices, header=provider_anime.title
prompt="Select Episode", choices=choices
)
if not chosen_episode_str or chosen_episode_str == "Back":
@@ -80,8 +78,6 @@ def episodes(ctx: Context, state: State) -> State | ControlFlow:
chosen_episode = chosen_episode_str
# --- Transition to Servers Menu ---
# Create a new state, updating the provider state with the chosen episode.
return State(
menu_name="SERVERS",
media_api=state.media_api,

155
fastanime/cli/interactive/menus/main.py
View File

@@ -1,23 +1,14 @@
# fastanime/cli/interactive/menus/main.py
from __future__ import annotations
import random
from typing import TYPE_CHECKING, Callable, Dict, Tuple
from typing import Callable, Dict, Tuple
import click
from rich.console import Console
from rich.progress import Progress
from ....libs.api.params import ApiSearchParams, UserListParams
from ....libs.api.types import MediaSearchResult, MediaStatus, UserListStatusType
from ..session import Context, session
from ..state import ControlFlow, MediaApiState, State
if TYPE_CHECKING:
from ....libs.api.types import MediaSearchResult
# A type alias for the actions this menu can perform.
# It returns a tuple: (NextMenuNameOrControlFlow, Optional[DataPayload])
MenuAction = Callable[[], Tuple[str, MediaSearchResult | None]]
@@ -28,66 +19,32 @@ def main(ctx: Context, state: State) -> State | ControlFlow:
Displays top-level categories for the user to browse and select.
"""
icons = ctx.config.general.icons
api_client = ctx.media_api
per_page = ctx.config.anilist.per_page
console = Console()
console.clear()
# The lambdas now correctly use the versatile search_media for most actions.
options: Dict[str, MenuAction] = {
# --- Search-based Actions ---
f"{'🔥 ' if icons else ''}Trending": lambda: (
"RESULTS",
api_client.search_media(
ApiSearchParams(sort="TRENDING_DESC", per_page=per_page)
),
f"{'🔥 ' if icons else ''}Trending": _create_media_list_action(
ctx, "TRENDING_DESC"
),
f"{'✨ ' if icons else ''}Popular": lambda: (
"RESULTS",
api_client.search_media(
ApiSearchParams(sort="POPULARITY_DESC", per_page=per_page)
),
f"{'✨ ' if icons else ''}Popular": _create_media_list_action(
ctx, "POPULARITY_DESC"
),
f"{'💖 ' if icons else ''}Favourites": lambda: (
"RESULTS",
api_client.search_media(
ApiSearchParams(sort="FAVOURITES_DESC", per_page=per_page)
),
f"{'💖 ' if icons else ''}Favourites": _create_media_list_action(
ctx, "FAVOURITES_DESC"
),
f"{'💯 ' if icons else ''}Top Scored": lambda: (
"RESULTS",
api_client.search_media(
ApiSearchParams(sort="SCORE_DESC", per_page=per_page)
),
f"{'💯 ' if icons else ''}Top Scored": _create_media_list_action(
ctx, "SCORE_DESC"
),
f"{'🎬 ' if icons else ''}Upcoming": lambda: (
"RESULTS",
api_client.search_media(
ApiSearchParams(
status="NOT_YET_RELEASED", sort="POPULARITY_DESC", per_page=per_page
)
),
f"{'🎬 ' if icons else ''}Upcoming": _create_media_list_action(
ctx, "POPULARITY_DESC", "NOT_YET_RELEASED"
),
f"{'🔔 ' if icons else ''}Recently Updated": lambda: (
"RESULTS",
api_client.search_media(
ApiSearchParams(
status="RELEASING", sort="UPDATED_AT_DESC", per_page=per_page
)
),
),
f"{'🎲 ' if icons else ''}Random": lambda: (
"RESULTS",
api_client.search_media(
ApiSearchParams(
id_in=random.sample(range(1, 160000), k=50), per_page=per_page
)
),
),
f"{'🔎 ' if icons else ''}Search": lambda: (
"RESULTS",
api_client.search_media(
ApiSearchParams(query=ctx.selector.ask("Search for Anime"))
),
f"{'🔔 ' if icons else ''}Recently Updated": _create_media_list_action(
ctx, "UPDATED_AT_DESC"
),
# --- special case media list --
f"{'🎲 ' if icons else ''}Random": _create_random_media_list(ctx),
f"{'🔎 ' if icons else ''}Search": _create_search_media_list(ctx),
# --- Authenticated User List Actions ---
f"{'📺 ' if icons else ''}Watching": _create_user_list_action(ctx, "CURRENT"),
f"{'📑 ' if icons else ''}Planned": _create_user_list_action(ctx, "PLANNING"),
@@ -116,10 +73,7 @@ def main(ctx: Context, state: State) -> State | ControlFlow:
# --- Action Handling ---
selected_action = options[choice_str]
with Progress(transient=True) as progress:
task = progress.add_task(f"[cyan]Fetching {choice_str.strip()}...", total=None)
next_menu_name, result_data = selected_action()
progress.update(task, completed=True)
next_menu_name, result_data = selected_action()
if next_menu_name == "EXIT":
return ControlFlow.EXIT
@@ -129,7 +83,7 @@ def main(ctx: Context, state: State) -> State | ControlFlow:
return ControlFlow.CONTINUE
if not result_data:
click.echo(
console.print(
f"[bold red]Error:[/bold red] Failed to fetch data for '{choice_str.strip()}'."
)
return ControlFlow.CONTINUE
@@ -141,17 +95,62 @@ def main(ctx: Context, state: State) -> State | ControlFlow:
)
def _create_user_list_action(ctx: Context, status: str) -> MenuAction:
"""A factory to create menu actions for fetching user lists, handling authentication."""
def _create_media_list_action(
ctx: Context, sort, status: MediaStatus | None = None
) -> MenuAction:
"""A factory to create menu actions for fetching media lists"""
def action() -> Tuple[str, MediaSearchResult | None]:
if not ctx.media_api.user_profile:
click.echo(
f"[bold yellow]Please log in to view your '{status.title()}' list.[/]"
def action():
with Progress(transient=True) as progress:
progress.add_task(f"[cyan]Fetching anime...", total=None)
return "RESULTS", ctx.media_api.search_media(
ApiSearchParams(
sort=sort, per_page=ctx.config.anilist.per_page, status=status
)
)
return action
def _create_random_media_list(ctx: Context) -> MenuAction:
def action():
with Progress(transient=True) as progress:
progress.add_task(f"[cyan]Fetching random anime...", total=None)
return "RESULTS", ctx.media_api.search_media(
ApiSearchParams(
id_in=random.sample(range(1, 160000), k=50),
per_page=ctx.config.anilist.per_page,
)
)
return action
def _create_search_media_list(ctx: Context) -> MenuAction:
def action():
query = ctx.selector.ask("Search for Anime")
if not query:
return "CONTINUE", None
with Progress(transient=True) as progress:
progress.add_task(f"[cyan]Searching for {query}...", total=None)
return "RESULTS", ctx.media_api.search_media(ApiSearchParams(query=query))
return action
def _create_user_list_action(ctx: Context, status: UserListStatusType) -> MenuAction:
"""A factory to create menu actions for fetching user lists, handling authentication."""
def action():
# if not ctx.media_api.user_profile:
# click.echo(
# f"[bold yellow]Please log in to view your '{status.title()}' list.[/]"
# )
# return "CONTINUE", None
with Progress(transient=True) as progress:
progress.add_task(f"[cyan]Fetching random anime...", total=None)
return "RESULTS", ctx.media_api.fetch_user_list(
UserListParams(status=status, per_page=ctx.config.anilist.per_page)
)
return "CONTINUE", None
return "RESULTS", ctx.media_api.fetch_user_list(
UserListParams(status=status, per_page=ctx.config.anilist.per_page)
)
return action

216
fastanime/cli/interactive/menus/media_actions.py
View File

@@ -1,16 +1,15 @@
from typing import TYPE_CHECKING, Callable, Dict, Tuple
from typing import Callable, Dict
import click
from InquirerPy.validator import EmptyInputValidator, NumberValidator
from rich.console import Console
from ....libs.api.params import UpdateListEntryParams
from ....libs.api.types import UserListStatusType
from ...utils.anilist import anilist_data_helper
from ....libs.api.types import MediaItem
from ....libs.players.params import PlayerParams
from ..session import Context, session
from ..state import ControlFlow, MediaApiState, ProviderState, State
from ..state import ControlFlow, ProviderState, State
if TYPE_CHECKING:
from ....libs.api.types import MediaItem
MenuAction = Callable[[], State | ControlFlow]
@session.menu
@@ -19,101 +18,21 @@ def media_actions(ctx: Context, state: State) -> State | ControlFlow:
Displays actions for a single, selected anime, such as streaming,
viewing details, or managing its status on the user's list.
"""
anime = state.media_api.anime
if not anime:
click.echo("[bold red]Error: No anime selected.[/bold red]")
return ControlFlow.BACK
icons = ctx.config.general.icons
selector = ctx.selector
player = ctx.player
# --- Action Implementations ---
def stream() -> State | ControlFlow:
# This is the key transition to the provider-focused part of the app.
# We create a new state for the next menu, carrying over the selected
# anime's details for the provider to use.
return State(
menu_name="PROVIDER_SEARCH",
media_api=state.media_api, # Carry over the existing api state
provider=ProviderState(), # Initialize a fresh provider state
)
def watch_trailer() -> State | ControlFlow:
if not anime.trailer or not anime.trailer.id:
click.echo(
"[bold yellow]No trailer available for this anime.[/bold yellow]"
)
else:
trailer_url = f"https://www.youtube.com/watch?v={anime.trailer.id}"
click.echo(
f"Playing trailer for '{anime.title.english or anime.title.romaji}'..."
)
player.play(url=trailer_url, title=f"Trailer: {anime.title.english}")
return ControlFlow.CONTINUE
def add_to_list() -> State | ControlFlow:
choices = ["CURRENT", "PLANNING", "COMPLETED", "DROPPED", "PAUSED", "REPEATING"]
status = selector.choose("Select list status:", choices=choices)
if status:
_update_user_list(
ctx,
anime,
UpdateListEntryParams(media_id=anime.id, status=status),
)
return ControlFlow.CONTINUE
def score_anime() -> State | ControlFlow:
score_str = selector.ask(
"Enter score (0.0 - 10.0):",
)
try:
score = float(score_str) if score_str else 0.0
if not 0.0 <= score <= 10.0:
raise ValueError("Score out of range.")
_update_user_list(
ctx, anime, UpdateListEntryParams(media_id=anime.id, score=score)
)
except (ValueError, TypeError):
click.echo(
"[bold red]Invalid score. Please enter a number between 0 and 10.[/bold red]"
)
return ControlFlow.CONTINUE
def view_info() -> State | ControlFlow:
# Placeholder for a more detailed info screen if needed.
# For now, we'll just print key details.
from rich import box
from rich.panel import Panel
from rich.text import Text
title = Text(anime.title.english or anime.title.romaji, style="bold cyan")
description = anilist_data_helper.clean_html(
anime.description or "No description."
)
genres = f"[bold]Genres:[/bold] {', '.join(anime.genres)}"
panel_content = f"{genres}\n\n{description}"
click.echo(Panel(panel_content, title=title, box=box.ROUNDED, expand=False))
selector.ask("Press Enter to continue...") # Pause to allow reading
return ControlFlow.CONTINUE
# --- Build Menu Options ---
options: Dict[str, Callable[[], State | ControlFlow]] = {
f"{'▶️ ' if icons else ''}Stream": stream,
f"{'📼 ' if icons else ''}Watch Trailer": watch_trailer,
f"{'➕ ' if icons else ''}Add/Update List": add_to_list,
f"{'⭐ ' if icons else ''}Score Anime": score_anime,
f"{'ℹ️ ' if icons else ''}View Info": view_info,
# TODO: Add 'Recommendations' and 'Relations' here later.
# TODO: Add 'Recommendations' and 'Relations' here later.
options: Dict[str, MenuAction] = {
f"{'▶️ ' if icons else ''}Stream": _stream(ctx, state),
f"{'📼 ' if icons else ''}Watch Trailer": _watch_trailer(ctx, state),
f"{'➕ ' if icons else ''}Add/Update List": _add_to_list(ctx, state),
f"{'⭐ ' if icons else ''}Score Anime": _score_anime(ctx, state),
f"{'ℹ️ ' if icons else ''}View Info": _view_info(ctx, state),
f"{'🔙 ' if icons else ''}Back to Results": lambda: ControlFlow.BACK,
}
# --- Prompt and Execute ---
header = f"Actions for: {anime.title.english or anime.title.romaji}"
choice_str = ctx.selector.choose(
prompt="Select Action", choices=list(options.keys()), header=header
prompt="Select Action", choices=list(options.keys())
)
if choice_str and choice_str in options:
@@ -122,11 +41,112 @@ def media_actions(ctx: Context, state: State) -> State | ControlFlow:
return ControlFlow.BACK
# --- Action Implementations ---
def _stream(ctx: Context, state: State) -> MenuAction:
def action():
return State(
menu_name="PROVIDER_SEARCH",
media_api=state.media_api, # Carry over the existing api state
provider=ProviderState(), # Initialize a fresh provider state
)
return action
def _watch_trailer(ctx: Context, state: State) -> MenuAction:
def action():
anime = state.media_api.anime
if not anime:
return ControlFlow.CONTINUE
if not anime.trailer or not anime.trailer.id:
print("[bold yellow]No trailer available for this anime.[/bold yellow]")
else:
trailer_url = f"https://www.youtube.com/watch?v={anime.trailer.id}"
print(
f"Playing trailer for '{anime.title.english or anime.title.romaji}'..."
)
ctx.player.play(PlayerParams(url=trailer_url, title=""))
return ControlFlow.CONTINUE
return action
def _add_to_list(ctx: Context, state: State) -> MenuAction:
def action():
anime = state.media_api.anime
if not anime:
return ControlFlow.CONTINUE
choices = ["CURRENT", "PLANNING", "COMPLETED", "DROPPED", "PAUSED", "REPEATING"]
status = ctx.selector.choose("Select list status:", choices=choices)
if status:
_update_user_list(
ctx,
anime,
UpdateListEntryParams(media_id=anime.id, status=status),
)
return ControlFlow.CONTINUE
return action
def _score_anime(ctx: Context, state: State) -> MenuAction:
def action():
anime = state.media_api.anime
if not anime:
return ControlFlow.CONTINUE
score_str = ctx.selector.ask("Enter score (0.0 - 10.0):")
try:
score = float(score_str) if score_str else 0.0
if not 0.0 <= score <= 10.0:
raise ValueError("Score out of range.")
_update_user_list(
ctx, anime, UpdateListEntryParams(media_id=anime.id, score=score)
)
except (ValueError, TypeError):
print(
"[bold red]Invalid score. Please enter a number between 0 and 10.[/bold red]"
)
return ControlFlow.CONTINUE
return action
def _view_info(ctx: Context, state: State) -> MenuAction:
def action():
anime = state.media_api.anime
if not anime:
return ControlFlow.CONTINUE
# Placeholder for a more detailed info screen if needed.
# For now, we'll just print key details.
from rich import box
from rich.panel import Panel
from rich.text import Text
from ...utils import image
console = Console()
title = Text(anime.title.english or anime.title.romaji or "", style="bold cyan")
description = Text(anime.description or "NO description")
genres = Text(f"Genres: {', '.join(anime.genres)}")
panel_content = f"{genres}\n\n{description}"
console.clear()
if cover_image := anime.cover_image:
image.render_image(cover_image.large)
console.print(Panel(panel_content, title=title, box=box.ROUNDED, expand=True))
ctx.selector.ask("Press Enter to continue...")
return ControlFlow.CONTINUE
return action
def _update_user_list(ctx: Context, anime: MediaItem, params: UpdateListEntryParams):
"""Helper to call the API to update a user's list and show feedback."""
if not ctx.media_api.user_profile:
click.echo("[bold yellow]You must be logged in to modify your list.[/]")
return
# if not ctx.media_api.user_profile:
# click.echo("[bold yellow]You must be logged in to modify your list.[/]")
# return
success = ctx.media_api.update_list_entry(params)
if success:

17
fastanime/cli/interactive/menus/player_controls.py
View File

@@ -2,10 +2,11 @@ import threading
from typing import TYPE_CHECKING, Callable, Dict
import click
from rich.console import Console
from ....libs.api.params import UpdateListEntryParams
from ..session import Context, session
from ..state import ControlFlow, ProviderState, State
from ..state import ControlFlow, State
if TYPE_CHECKING:
from ....libs.providers.anime.types import Server
@@ -27,8 +28,8 @@ def _update_progress_in_background(ctx: Context, anime_id: int, progress: int):
"""Fires off a non-blocking request to update AniList progress."""
def task():
if not ctx.media_api.user_profile:
return
# if not ctx.media_api.user_profile:
# return
params = UpdateListEntryParams(media_id=anime_id, progress=progress)
ctx.media_api.update_list_entry(params)
# We don't need to show feedback here, it's a background task.
@@ -46,6 +47,8 @@ def player_controls(ctx: Context, state: State) -> State | ControlFlow:
config = ctx.config
player = ctx.player
selector = ctx.selector
console = Console()
console.clear()
provider_anime = state.provider.anime
anilist_anime = state.media_api.anime
@@ -63,7 +66,9 @@ def player_controls(ctx: Context, state: State) -> State | ControlFlow:
all_servers,
)
):
click.echo("[bold red]Error: Player state is incomplete. Returning.[/bold red]")
console.print(
"[bold red]Error: Player state is incomplete. Returning.[/bold red]"
)
return ControlFlow.BACK
# --- Post-Playback Logic ---
@@ -86,7 +91,7 @@ def player_controls(ctx: Context, state: State) -> State | ControlFlow:
current_index = available_episodes.index(current_episode_num)
if config.stream.auto_next and current_index < len(available_episodes) - 1:
click.echo("[cyan]Auto-playing next episode...[/cyan]")
console.print("[cyan]Auto-playing next episode...[/cyan]")
next_episode_num = available_episodes[current_index + 1]
return State(
menu_name="SERVERS",
@@ -108,7 +113,7 @@ def player_controls(ctx: Context, state: State) -> State | ControlFlow:
update={"episode_number": next_episode_num}
),
)
click.echo("[bold yellow]This is the last available episode.[/bold yellow]")
console.print("[bold yellow]This is the last available episode.[/bold yellow]")
return ControlFlow.CONTINUE
def replay() -> State | ControlFlow:

23
fastanime/cli/interactive/menus/provider_search.py
View File

@@ -1,6 +1,7 @@
from typing import TYPE_CHECKING
import click
from rich.console import Console
from rich.progress import Progress
from thefuzz import fuzz
@@ -27,10 +28,12 @@ def provider_search(ctx: Context, state: State) -> State | ControlFlow:
provider = ctx.provider
selector = ctx.selector
config = ctx.config
console = Console()
console.clear()
anilist_title = anilist_anime.title.english or anilist_anime.title.romaji
if not anilist_title:
click.echo(
console.print(
"[bold red]Error: Selected anime has no searchable title.[/bold red]"
)
return ControlFlow.BACK
@@ -48,10 +51,10 @@ def provider_search(ctx: Context, state: State) -> State | ControlFlow:
)
if not provider_search_results or not provider_search_results.results:
click.echo(
console.print(
f"[bold yellow]Could not find '{anilist_title}' on {provider.__class__.__name__}.[/bold yellow]"
)
click.echo("Try another provider from the config or go back.")
console.print("Try another provider from the config or go back.")
return ControlFlow.BACK
# --- Map results for selection ---
@@ -68,16 +71,14 @@ def provider_search(ctx: Context, state: State) -> State | ControlFlow:
provider_results_map.keys(),
key=lambda p_title: fuzz.ratio(p_title.lower(), anilist_title.lower()),
)
click.echo(f"[cyan]Auto-selecting best match:[/] {best_match_title}")
console.print(f"[cyan]Auto-selecting best match:[/] {best_match_title}")
selected_provider_anime = provider_results_map[best_match_title]
else:
choices = list(provider_results_map.keys())
choices.append("Back")
chosen_title = selector.choose(
prompt=f"Confirm match for '{anilist_title}'",
choices=choices,
header="Provider Search Results",
prompt=f"Confirm match for '{anilist_title}'", choices=choices
)
if not chosen_title or chosen_title == "Back":
@@ -85,9 +86,6 @@ def provider_search(ctx: Context, state: State) -> State | ControlFlow:
selected_provider_anime = provider_results_map[chosen_title]
if not selected_provider_anime:
return ControlFlow.BACK
# --- Fetch Full Anime Details from Provider ---
with Progress(transient=True) as progress:
progress.add_task(
@@ -99,14 +97,11 @@ def provider_search(ctx: Context, state: State) -> State | ControlFlow:
full_provider_anime = provider.get(AnimeParams(id=selected_provider_anime.id))
if not full_provider_anime:
click.echo(
console.print(
f"[bold red]Failed to fetch details for '{selected_provider_anime.title}'.[/bold red]"
)
return ControlFlow.BACK
# --- Transition to Episodes Menu ---
# Create the next state, populating the 'provider' field for the first time
# while carrying over the 'media_api' state.
return State(
menu_name="EPISODES",
media_api=state.media_api,

25
fastanime/cli/interactive/menus/results.py
View File

@@ -1,19 +1,10 @@
from typing import TYPE_CHECKING, List
import click
from rich.progress import Progress
from yt_dlp.utils import sanitize_filename
from rich.console import Console
from ...utils.anilist import (
anilist_data_helper, # Assuming this is the new location
)
from ...utils.previews import get_anime_preview
from ....libs.api.types import MediaItem
from ..session import Context, session
from ..state import ControlFlow, MediaApiState, State
if TYPE_CHECKING:
from ....libs.api.types import MediaItem
@session.menu
def results(ctx: Context, state: State) -> State | ControlFlow:
@@ -22,8 +13,12 @@ def results(ctx: Context, state: State) -> State | ControlFlow:
Allows the user to select an anime to view its actions or navigate pages.
"""
search_results = state.media_api.search_results
console = Console()
console.clear()
if not search_results or not search_results.media:
click.echo("[bold yellow]No anime found for the given criteria.[/bold yellow]")
console.print(
"[bold yellow]No anime found for the given criteria.[/bold yellow]"
)
return ControlFlow.BACK
# --- Prepare choices and previews ---
@@ -38,6 +33,8 @@ def results(ctx: Context, state: State) -> State | ControlFlow:
preview_command = None
if ctx.config.general.preview != "none":
# This function will start background jobs to cache preview data
from ...utils.previews import get_anime_preview
preview_command = get_anime_preview(anime_items, formatted_titles, ctx.config)
# --- Build Navigation and Final Choice List ---
@@ -55,7 +52,6 @@ def results(ctx: Context, state: State) -> State | ControlFlow:
choice_str = ctx.selector.choose(
prompt="Select Anime",
choices=choices,
header="AniList Results",
preview=preview_command,
)
@@ -119,5 +115,4 @@ def _format_anime_choice(anime: MediaItem, config) -> str:
icon = "🔹" if config.general.icons else "!"
display_title += f" {icon}{unwatched} new{icon}"
# Sanitize for use as a potential filename/cache key
return sanitize_filename(display_title, restricted=True)
return display_title

40
fastanime/cli/interactive/menus/servers.py
View File

@@ -1,18 +1,14 @@
from typing import TYPE_CHECKING, Dict, List
from typing import Dict, List
import click
from rich.console import Console
from rich.progress import Progress
from ....libs.players.params import PlayerParams
from ....libs.providers.anime.params import EpisodeStreamsParams
from ....libs.providers.anime.types import Server
from ..session import Context, session
from ..state import ControlFlow, ProviderState, State
if TYPE_CHECKING:
from ....cli.utils.utils import (
filter_by_quality, # You may need to create this helper
)
from ....libs.providers.anime.types import Server
from ..state import ControlFlow, State
def _filter_by_quality(links, quality):
@@ -34,9 +30,14 @@ def servers(ctx: Context, state: State) -> State | ControlFlow:
config = ctx.config
provider = ctx.provider
selector = ctx.selector
console = Console()
console.clear()
if not provider_anime or not episode_number:
click.echo("[bold red]Error: Anime or episode details are missing.[/bold red]")
console.print(
"[bold red]Error: Anime or episode details are missing.[/bold red]"
)
selector.ask("Enter to continue...")
return ControlFlow.BACK
# --- Fetch Server Streams ---
@@ -55,7 +56,7 @@ def servers(ctx: Context, state: State) -> State | ControlFlow:
all_servers: List[Server] = list(server_iterator) if server_iterator else []
if not all_servers:
click.echo(
console.print(
f"[bold yellow]No streaming servers found for this episode.[/bold yellow]"
)
return ControlFlow.BACK
@@ -67,10 +68,12 @@ def servers(ctx: Context, state: State) -> State | ControlFlow:
preferred_server = config.stream.server.lower()
if preferred_server == "top":
selected_server = all_servers[0]
click.echo(f"[cyan]Auto-selecting top server:[/] {selected_server.name}")
console.print(f"[cyan]Auto-selecting top server:[/] {selected_server.name}")
elif preferred_server in server_map:
selected_server = server_map[preferred_server]
click.echo(f"[cyan]Auto-selecting preferred server:[/] {selected_server.name}")
console.print(
f"[cyan]Auto-selecting preferred server:[/] {selected_server.name}"
)
else:
choices = [*server_map.keys(), "Back"]
chosen_name = selector.choose("Select Server", choices)
@@ -78,20 +81,16 @@ def servers(ctx: Context, state: State) -> State | ControlFlow:
return ControlFlow.BACK
selected_server = server_map[chosen_name]
if not selected_server:
return ControlFlow.BACK
# --- Select Stream Quality ---
stream_link_obj = _filter_by_quality(selected_server.links, config.stream.quality)
if not stream_link_obj:
click.echo(
console.print(
f"[bold red]No stream of quality '{config.stream.quality}' found on server '{selected_server.name}'.[/bold red]"
)
return ControlFlow.CONTINUE
# --- Launch Player ---
final_title = f"{provider_anime.title} - Ep {episode_number}"
click.echo(f"[bold green]Launching player for:[/] {final_title}")
console.print(f"[bold green]Launching player for:[/] {final_title}")
player_result = ctx.player.play(
PlayerParams(
@@ -99,12 +98,9 @@ def servers(ctx: Context, state: State) -> State | ControlFlow:
title=final_title,
subtitles=[sub.url for sub in selected_server.subtitles],
headers=selected_server.headers,
# start_time logic will be added in player_controls
)
)
# --- Transition to Player Controls ---
# We now have all the data for post-playback actions.
return State(
menu_name="PLAYER_CONTROLS",
media_api=state.media_api,
@@ -112,7 +108,7 @@ def servers(ctx: Context, state: State) -> State | ControlFlow:
update={
"servers": all_servers,
"selected_server": selected_server,
"last_player_result": player_result, # We should add this to ProviderState
"last_player_result": player_result,
}
),
)

27
fastanime/cli/interactive/session.py
View File

@@ -8,21 +8,21 @@ from typing import TYPE_CHECKING, Callable, List
import click
from ...core.config import AppConfig
from ...core.constants import USER_CONFIG_PATH
from ...core.constants import APP_DIR, USER_CONFIG_PATH
from ...libs.api.base import BaseApiClient
from ...libs.players.base import BasePlayer
from ...libs.providers.anime.base import BaseAnimeProvider
from ...libs.selectors.base import BaseSelector
from ..config import ConfigLoader
from .state import ControlFlow, State
if TYPE_CHECKING:
from ...libs.api.base import BaseApiClient
from ...libs.players.base import BasePlayer
from ...libs.providers.anime.base import BaseAnimeProvider
from ...libs.selectors.base import BaseSelector
logger = logging.getLogger(__name__)
# A type alias for the signature all menu functions must follow.
MenuFunction = Callable[["Context", State], "State | ControlFlow"]
MENUS_DIR = APP_DIR / "cli" / "interactive" / "menus"
@dataclass(frozen=True)
class Context:
@@ -113,10 +113,7 @@ class Session:
# Execute the menu function, which returns the next step.
next_step = menu_to_run.execute(self._context, current_state)
if isinstance(next_step, State):
# A new state was returned, push it to history for the next loop.
self._history.append(next_step)
elif isinstance(next_step, ControlFlow):
if isinstance(next_step, ControlFlow):
# A control command was issued.
if next_step == ControlFlow.EXIT:
break # Exit the loop
@@ -126,6 +123,12 @@ class Session:
elif next_step == ControlFlow.RELOAD_CONFIG:
self._edit_config()
# For CONTINUE, we do nothing, allowing the loop to re-run the current state.
elif isinstance(next_step, State):
# if the state is main menu we should reset the history
if next_step.menu_name == "MAIN":
self._history = [next_step]
# A new state was returned, push it to history for the next loop.
self._history.append(next_step)
else:
logger.error(
f"Menu '{current_state.menu_name}' returned invalid type: {type(next_step)}"
@@ -169,7 +172,7 @@ class Session:
return decorator
def load_menus_from_folder(self, package_path: Path):
def load_menus_from_folder(self, package_path: Path = MENUS_DIR):
"""
Dynamically imports all Python modules from a folder to register their menus.

23
fastanime/cli/interactive/state.py
View File

@@ -1,12 +1,16 @@
from enum import Enum, auto
from typing import Iterator, Optional
from typing import Iterator, List, Literal, Optional
from pydantic import BaseModel, ConfigDict
# Import the actual data models from your libs.
# These will be the data types held within our state models.
from ....libs.api.types import MediaItem, MediaSearchResult
from ....libs.providers.anime.types import Anime, SearchResults, Server
from ...libs.api.types import (
MediaItem,
MediaSearchResult,
MediaStatus,
UserListStatusType,
)
from ...libs.players.types import PlayerResult
from ...libs.providers.anime.types import Anime, SearchResults, Server
class ControlFlow(Enum):
@@ -47,6 +51,10 @@ class ProviderState(BaseModel):
search_results: Optional[SearchResults] = None
anime: Optional[Anime] = None
episode_streams: Optional[Iterator[Server]] = None
episode_number: Optional[str] = None
last_player_result: Optional[PlayerResult] = None
servers: Optional[List[Server]] = None
selected_server: Optional[Server] = None
model_config = ConfigDict(
frozen=True,
@@ -62,6 +70,11 @@ class MediaApiState(BaseModel):
"""
search_results: Optional[MediaSearchResult] = None
search_results_type: Optional[Literal["MEDIA_LIST", "USER_MEDIA_LIST"]] = None
sort: Optional[str] = None
query: Optional[str] = None
user_media_status: Optional[UserListStatusType] = None
media_status: Optional[MediaStatus] = None
anime: Optional[MediaItem] = None
model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)

29
fastanime/cli/utils/ansi.py Normal file
View File

@@ -0,0 +1,29 @@
# Define ANSI escape codes as constants
RESET = "\033[0m"
BOLD = "\033[1m"
INVISIBLE_CURSOR = "\033[?25l"
VISIBLE_CURSOR = "\033[?25h"
UNDERLINE = "\033[4m"
def get_true_fg(color: list[str], bold: bool = True) -> str:
"""Custom helper function that enables colored text in the terminal
Args:
bold: whether to bolden the text
string: string to color
r: red
g: green
b: blue
Returns:
colored string
"""
# NOTE: Currently only supports terminals that support true color
r = color[0]
g = color[1]
b = color[2]
if bold:
return f"{BOLD}\033[38;2;{r};{g};{b};m"
else:
return f"\033[38;2;{r};{g};{b};m"

63
fastanime/cli/utils/formatters.py Normal file
View File

@@ -0,0 +1,63 @@
import re
from typing import TYPE_CHECKING, List, Optional
from yt_dlp.utils import clean_html as ytdlp_clean_html
from ...libs.api.types import AiringSchedule, MediaItem
COMMA_REGEX = re.compile(r"([0-9]{3})(?=\d)")
def clean_html(raw_html: str) -> str:
"""A wrapper around yt-dlp's clean_html to handle None inputs."""
return ytdlp_clean_html(raw_html) if raw_html else ""
def format_number_with_commas(number: Optional[int]) -> str:
"""Formats an integer with commas for thousands separation."""
if number is None:
return "N/A"
return COMMA_REGEX.sub(r"\1,", str(number)[::-1])[::-1]
def format_airing_schedule(airing: Optional[AiringSchedule]) -> str:
"""Formats the next airing episode information into a readable string."""
if not airing or not airing.airing_at:
return "N/A"
# Get a human-readable date and time
air_date = airing.airing_at.strftime("%a, %b %d at %I:%M %p")
return f"Ep {airing.episode} on {air_date}"
def format_genres(genres: List[str]) -> str:
"""Joins a list of genres into a single, comma-separated string."""
return ", ".join(genres) if genres else "N/A"
def format_score_stars_full(score: Optional[float]) -> str:
"""Formats an AniList score (0-100) to a 0-10 scale using full stars."""
if score is None:
return "N/A"
# Convert 0-100 to 0-10, then to a whole number of stars
num_stars = min(round(score * 6 / 100), 6)
return "⭐" * num_stars
def format_score(score: Optional[float]) -> str:
"""Formats an AniList score (0-100) to a 0-10 scale."""
if score is None:
return "N/A"
return f"{score / 10.0:.1f} / 10"
def shell_safe(text: Optional[str]) -> str:
"""
Escapes a string for safe inclusion in a shell script,
specifically for use within double quotes. It escapes backticks,
double quotes, and dollar signs.
"""
if not text:
return ""
return text.replace("`", "\\`").replace('"', '\\"').replace("$", "\\$")

87
fastanime/cli/utils/image.py Normal file
View File

@@ -0,0 +1,87 @@
# fastanime/cli/utils/image.py
from __future__ import annotations
import logging
import shutil
import subprocess
from typing import Optional
import click
import httpx
logger = logging.getLogger(__name__)
def render_image(url: str, capture: bool = False, size: str = "30x30") -> Optional[str]:
"""
Renders an image from a URL in the terminal using icat or chafa.
This function automatically detects the best available tool.
Args:
url: The URL of the image to render.
capture: If True, returns the terminal-formatted image as a string
instead of printing it. Defaults to False.
size: The size parameter to pass to the rendering tool (e.g., "WxH").
Returns:
If capture is True, returns the image data as a string.
If capture is False, prints directly to the terminal and returns None.
Returns None on any failure.
"""
# --- Common subprocess arguments ---
subprocess_kwargs = {
"check": False, # We will handle errors manually
"capture_output": capture,
"text": capture, # Decode stdout/stderr as text if capturing
}
# --- Try icat (Kitty terminal) first ---
if icat_executable := shutil.which("icat"):
process = subprocess.run(
[icat_executable, "--align", "left", url], **subprocess_kwargs
)
if process.returncode == 0:
return process.stdout if capture else None
logger.warning(f"icat failed for URL {url} with code {process.returncode}")
# --- Fallback to chafa ---
if chafa_executable := shutil.which("chafa"):
try:
# Chafa requires downloading the image data first
with httpx.Client() as client:
response = client.get(url, follow_redirects=True, timeout=20)
response.raise_for_status()
img_bytes = response.content
# Add stdin input to the subprocess arguments
subprocess_kwargs["input"] = img_bytes
process = subprocess.run(
[chafa_executable, f"--size={size}", "-"], **subprocess_kwargs
)
if process.returncode == 0:
return process.stdout if capture else None
logger.warning(f"chafa failed for URL {url} with code {process.returncode}")
except httpx.HTTPStatusError as e:
logger.error(
f"HTTP error fetching image for chafa: {e.response.status_code}"
)
click.echo(
f"[dim]Error fetching image: {e.response.status_code}[/dim]", err=True
)
except Exception as e:
logger.error(f"An exception occurred while running chafa: {e}")
return None
# --- Final fallback if no tool is found ---
if not capture:
# Only show this message if the user expected to see something.
click.echo(
"[dim](Image preview skipped: icat or chafa not found)[/dim]", err=True
)
return None

101
fastanime/cli/utils/previews.py
View File

@@ -1,11 +1,11 @@
import concurrent.futures
import logging
import textwrap
import os
import shutil
from hashlib import sha256
from io import StringIO
from pathlib import Path
from threading import Thread
from typing import TYPE_CHECKING, List
from typing import List
import httpx
from rich.console import Console
@@ -13,11 +13,9 @@ from rich.panel import Panel
from rich.text import Text
from ...core.config import AppConfig
from ...core.constants import APP_DIR, PLATFORM
from .scripts import bash_functions
if TYPE_CHECKING:
from ...libs.api.types import MediaItem
from ...core.constants import APP_CACHE_DIR, APP_DIR, PLATFORM
from ...libs.api.types import MediaItem
from . import ansi, formatters
logger = logging.getLogger(__name__)
@@ -27,15 +25,7 @@ IMAGES_CACHE_DIR = PREVIEWS_CACHE_DIR / "images"
INFO_CACHE_DIR = PREVIEWS_CACHE_DIR / "info"
FZF_SCRIPTS_DIR = APP_DIR / "libs" / "selectors" / "fzf" / "scripts"
PREVIEW_SCRIPT_TEMPLATE_PATH = FZF_SCRIPTS_DIR / "preview.sh"
# Ensure cache directories exist on startup
IMAGES_CACHE_DIR.mkdir(parents=True, exist_ok=True)
INFO_CACHE_DIR.mkdir(parents=True, exist_ok=True)
# The helper functions (_get_cache_hash, _save_image_from_url, _save_info_text,
# _format_info_text, and _cache_worker) remain exactly the same as before.
# I am including them here for completeness.
INFO_SCRIPT_TEMPLATE_PATH = FZF_SCRIPTS_DIR / "info.sh"
def _get_cache_hash(text: str) -> str:
@@ -45,9 +35,9 @@ def _get_cache_hash(text: str) -> str:
def _save_image_from_url(url: str, hash_id: str):
"""Downloads an image using httpx and saves it to the cache."""
temp_image_path = IMAGES_CACHE_DIR / f"{hash_id}.png.tmp"
image_path = IMAGES_CACHE_DIR / f"{hash_id}.png"
try:
temp_image_path = IMAGES_CACHE_DIR / f"{hash_id}.png.tmp"
image_path = IMAGES_CACHE_DIR / f"{hash_id}.png"
with httpx.stream("GET", url, follow_redirects=True, timeout=20) as response:
response.raise_for_status()
with temp_image_path.open("wb") as f:
@@ -69,25 +59,40 @@ def _save_info_text(info_text: str, hash_id: str):
logger.error(f"Failed to write info cache for {hash_id}: {e}")
def _format_info_text(item: MediaItem) -> str:
"""Uses Rich to format a media item's details into a string."""
from .anilist import anilist_data_helper
def _populate_info_template(item: MediaItem, config: AppConfig) -> str:
"""
Takes the info.sh template and injects formatted, shell-safe data.
"""
template = INFO_SCRIPT_TEMPLATE_PATH.read_text(encoding="utf-8")
description = formatters.clean_html(item.description or "No description available.")
io_buffer = StringIO()
console = Console(file=io_buffer, force_terminal=True, color_system="truecolor")
title = Text(
item.title.english or item.title.romaji or "Unknown Title", style="bold cyan"
)
description = anilist_data_helper.clean_html(
item.description or "No description available."
)
description = (description[:350] + "...") if len(description) > 350 else description
genres = f"[bold]Genres:[/bold] {', '.join(item.genres)}"
status = f"[bold]Status:[/bold] {item.status}"
score = f"[bold]Score:[/bold] {item.average_score / 10 if item.average_score else 'N/A'}"
panel_content = f"{genres}\n{status}\n{score}\n\n{description}"
console.print(Panel(panel_content, title=title, border_style="dim"))
return io_buffer.getvalue()
HEADER_COLOR = config.fzf.preview_header_color.split(",")
SEPARATOR_COLOR = config.fzf.preview_separator_color.split(",")
# Escape all variables before injecting them into the script
replacements = {
"TITLE": formatters.shell_safe(item.title.english or item.title.romaji),
"SCORE": formatters.shell_safe(
formatters.format_score_stars_full(item.average_score)
),
"STATUS": formatters.shell_safe(item.status),
"FAVOURITES": formatters.shell_safe(
formatters.format_number_with_commas(item.favourites)
),
"GENRES": formatters.shell_safe(formatters.format_genres(item.genres)),
"SYNOPSIS": formatters.shell_safe(description),
# Color codes
"C_TITLE": ansi.get_true_fg(HEADER_COLOR, bold=True),
"C_KEY": ansi.get_true_fg(HEADER_COLOR, bold=True),
"C_VALUE": ansi.get_true_fg(HEADER_COLOR, bold=True),
"C_RULE": ansi.get_true_fg(SEPARATOR_COLOR, bold=True),
"RESET": ansi.RESET,
}
for key, value in replacements.items():
template = template.replace(f"{{{key}}}", value)
return template
def _cache_worker(items: List[MediaItem], titles: List[str], config: AppConfig):
@@ -102,7 +107,7 @@ def _cache_worker(items: List[MediaItem], titles: List[str], config: AppConfig):
)
if config.general.preview in ("full", "text"):
if not (INFO_CACHE_DIR / hash_id).exists():
info_text = _format_info_text(item)
info_text = _populate_info_template(item, config)
executor.submit(_save_info_text, info_text, hash_id)
@@ -114,6 +119,10 @@ def get_anime_preview(
Starts a background task to cache preview data and returns the fzf preview command
by formatting a shell script template.
"""
# Ensure cache directories exist on startup
IMAGES_CACHE_DIR.mkdir(parents=True, exist_ok=True)
INFO_CACHE_DIR.mkdir(parents=True, exist_ok=True)
# Start the non-blocking background Caching
Thread(target=_cache_worker, args=(items, titles, config), daemon=True).start()
@@ -130,15 +139,17 @@ def get_anime_preview(
path_sep = "\\" if PLATFORM == "win32" else "/"
# Format the template with the dynamic values
final_script = template.format(
bash_functions=bash_functions,
preview_mode=config.general.preview,
image_cache_path=str(IMAGES_CACHE_DIR),
info_cache_path=str(INFO_CACHE_DIR),
path_sep=path_sep,
final_script = (
template.replace("{preview_mode}", config.general.preview)
.replace("{image_cache_path}", str(IMAGES_CACHE_DIR))
.replace("{info_cache_path}", str(INFO_CACHE_DIR))
.replace("{path_sep}", path_sep)
.replace("{image_renderer}", config.general.image_renderer)
)
# )
# Return the command for fzf to execute. `sh -c` is used to run the script string.
# The -- "{}" ensures that the selected item is passed as the first argument ($1)
# to the script, even if it contains spaces or special characters.
return f'sh -c {final_script!r} -- "{{}}"'
os.environ["SHELL"] = "bash"
return final_script

33
fastanime/cli/utils/print_img.py
View File

@@ -1,33 +0,0 @@
import shutil
import subprocess
import requests
def print_img(url: str):
"""helper function to print an image given its url
Args:
url: [TODO:description]
"""
if EXECUTABLE := shutil.which("icat"):
subprocess.run([EXECUTABLE, url], check=False)
else:
EXECUTABLE = shutil.which("chafa")
if EXECUTABLE is None:
print("chafanot found")
return
res = requests.get(url)
if res.status_code != 200:
print("Error fetching image")
return
img_bytes = res.content
"""
Change made in call to chafa. Chafa dev dropped ability
to pull from urls. Keeping old line here just in case.
subprocess.run([EXECUTABLE, url, "--size=15x15"], input=img_bytes)
"""
subprocess.run([EXECUTABLE, "--size=15x15"], input=img_bytes, check=False)

4
fastanime/libs/api/anilist/api.py
View File

@@ -43,7 +43,7 @@ class AniListApi(BaseApiClient):
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
variables["perPage"] = self.config.per_page or params.per_page
response = execute_graphql(
ANILIST_ENDPOINT, self.http_client, gql.SEARCH_MEDIA, variables
)
@@ -57,7 +57,7 @@ class AniListApi(BaseApiClient):
"userId": self.user_profile.id,
"status": params.status,
"page": params.page,
"perPage": params.per_page,
"perPage": self.config.per_page or params.per_page,
}
response = execute_graphql(
ANILIST_ENDPOINT, self.http_client, gql.GET_USER_MEDIA_LIST, variables

2
fastanime/libs/providers/anime/allanime/extractors/extractor.py
View File

@@ -18,7 +18,7 @@ from .yt_mp4 import YtExtractor
AVAILABLE_SOURCES = {
"Sak": SakExtractor,
"S-mp4": Smp4Extractor,
"Luf-mp4": Lufmp4Extractor,
"Luf-Mp4": Lufmp4Extractor,
"Default": DefaultExtractor,
"Yt-mp4": YtExtractor,
"Kir": KirExtractor,

9
fastanime/libs/providers/anime/allanime/extractors/gogoanime.py
View File

@@ -1,6 +1,6 @@
from ...types import EpisodeStream, Server
from ..constants import API_BASE_URL
from ..types import AllAnimeEpisode, AllAnimeSource
from ..types import AllAnimeEpisode, AllAnimeEpisodeStreams, AllAnimeSource
from .base import BaseExtractor
@@ -19,12 +19,15 @@ class Lufmp4Extractor(BaseExtractor):
timeout=10,
)
response.raise_for_status()
streams = response.json()
streams: AllAnimeEpisodeStreams = response.json()
return Server(
name="gogoanime",
links=[
EpisodeStream(link=link, quality="1080") for link in streams["links"]
EpisodeStream(
link=stream["link"], quality="1080", format=stream["resolutionStr"]
)
for stream in streams["links"]
],
episode_title=episode["notes"],
headers={"Referer": f"https://{API_BASE_URL}/"},

2
fastanime/libs/providers/anime/types.py
View File

@@ -55,7 +55,7 @@ class Anime(BaseAnimeProviderModel):
class EpisodeStream(BaseAnimeProviderModel):
episode: str
# episode: str
link: str
title: str | None = None
quality: Literal["360", "480", "720", "1080"] = "720"

71
fastanime/libs/selectors/fzf/scripts/info.sh Normal file
View File

@@ -0,0 +1,71 @@
#!/bin/sh
#
# FastAnime Preview Info Script Template
# This script formats and displays the textual information in the FZF preview pane.
# Some values are injected by python those with '{name}' syntax using .replace()
# --- Terminal Dimensions ---
WIDTH=${FZF_PREVIEW_COLUMNS:-80} # Set a fallback width of 80
# --- Helper function for printing a key-value pair, aligning the value to the right ---
print_kv() {
local key="$1"
local value="$2"
local key_len=${#key}
local value_len=${#value}
local multiplier="${3:-1}"
# Correctly calculate padding by accounting for the key, the ": ", and the value.
local padding_len=$((WIDTH - key_len - 2 - value_len * multiplier))
# If the text is too long to fit, just add a single space for separation.
if [ "$padding_len" -lt 1 ]; then
padding_len=1
value=$(echo $value| fold -s -w "$((WIDTH - key_len - 3))")
printf "{C_KEY}%s:{RESET}%*s%s\\n" "$key" "$padding_len" "" " $value"
else
printf "{C_KEY}%s:{RESET}%*s%s\\n" "$key" "$padding_len" "" " $value"
fi
}
# --- Draw a rule across the screen ---
draw_rule() {
local rule
# Generate the line of '─' characters, removing the trailing newline `tr` adds.
rule=$(printf '%*s' "$WIDTH" | tr ' ' '─' | tr -d '\n')
# Print the rule with colors and a single, clean newline.
printf "{C_RULE}%s{RESET}\\n" "$rule"
}
draw_rule(){
ll=2
while [ $ll -le $FZF_PREVIEW_COLUMNS ];do
echo -n -e "{C_RULE}─{RESET}"
((ll++))
done
echo
}
# --- Display Content ---
draw_rule
print_kv "Title" "{TITLE}"
draw_rule
# Key-Value Stats Section
score_multiplier=1
if ! [ "{SCORE}" = "N/A" ];then
score_multiplier=2
fi
print_kv "Score" "{SCORE}" $score_multiplier
print_kv "Status" "{STATUS}"
print_kv "Favourites" "{FAVOURITES}"
draw_rule
print_kv "Genres" "{GENRES}"
draw_rule
# Synopsis
echo "{SYNOPSIS}" | fold -s -w "$WIDTH"

13
fastanime/libs/selectors/fzf/scripts/preview.sh
View File

@@ -3,11 +3,11 @@
# FastAnime FZF Preview Script Template
#
# This script is a template. The placeholders in curly braces, like
# {placeholder}, are filled in by the Python application at runtime.
# placeholder, are filled in by the Python application at runtime.
# It is executed by `sh -c "..."` for each item fzf previews.
# The first argument ($1) is the item string from fzf (the sanitized title).
IMAGE_RENDERER="{image_renderer}"
generate_sha256() {
local input
@@ -37,11 +37,11 @@ fzf_preview() {
if [ "$dim" = x ]; then
dim=$(stty size </dev/tty | awk "{print \$2 \"x\" \$1}")
fi
if ! [ "$FASTANIME_IMAGE_RENDERER" = "icat" ] && [ -z "$KITTY_WINDOW_ID" ] && [ "$((FZF_PREVIEW_TOP + FZF_PREVIEW_LINES))" -eq "$(stty size </dev/tty | awk "{print \$1}")" ]; then
if ! [ "$IMAGE_RENDERER" = "icat" ] && [ -z "$KITTY_WINDOW_ID" ] && [ "$((FZF_PREVIEW_TOP + FZF_PREVIEW_LINES))" -eq "$(stty size </dev/tty | awk "{print \$1}")" ]; then
dim=${FZF_PREVIEW_COLUMNS}x$((FZF_PREVIEW_LINES - 1))
fi
if [ "$FASTANIME_IMAGE_RENDERER" = "icat" ] && [ -z "$GHOSTTY_BIN_DIR" ]; then
if [ "$IMAGE_RENDERER" = "icat" ] && [ -z "$GHOSTTY_BIN_DIR" ]; then
if command -v kitten >/dev/null 2>&1; then
kitten icat --clear --transfer-mode=memory --unicode-placeholder --stdin=no --place="$dim@0x0" "$file" | sed "\$d" | sed "$(printf "\$s/\$/\033[m/")"
elif command -v icat >/dev/null 2>&1; then
@@ -75,7 +75,7 @@ fzf_preview() {
fi
}
# Generate the same cache key that the Python worker uses
hash=$(_get_cache_hash "$1")
hash=$(generate_sha256 {})
# Display image if configured and the cached file exists
if [ "{preview_mode}" = "full" ] || [ "{preview_mode}" = "image" ]; then
@@ -87,12 +87,11 @@ if [ "{preview_mode}" = "full" ] || [ "{preview_mode}" = "image" ]; then
fi
echo # Add a newline for spacing
fi
# Display text info if configured and the cached file exists
if [ "{preview_mode}" = "full" ] || [ "{preview_mode}" = "text" ]; then
info_file="{info_cache_path}{path_sep}$hash"
if [ -f "$info_file" ]; then
cat "$info_file"
source "$info_file"
else
echo "📝 Loading details..."
fi

6
fastanime/libs/selectors/fzf/selector.py
View File

@@ -3,6 +3,8 @@ import os
import shutil
import subprocess
from rich.prompt import Prompt
from ....core.config import FzfConfig
from ....core.exceptions import FastAnimeError
from ..base import BaseSelector
@@ -58,7 +60,9 @@ class FzfSelector(BaseSelector):
return result == "Yes"
def ask(self, prompt, *, default=None):
# Use FZF's --print-query to capture user input
# cleaner to use rich
return Prompt.ask(prompt, default=default)
# -- not going to be used --
commands = [
self.executable,
"--prompt",