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: media actions

Browse Source
This commit is contained in:
Benexl
2025-07-24 01:54:59 +03:00
parent 83933f7a63
commit 9efe9f9949
5 changed files with 176 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,10 +1,5 @@
from typing import TYPE_CHECKING
import click
from rich.console import Console
from ..session import Context, session
from ..state import InternalDirective, ProviderState, State
from ..state import InternalDirective, MenuName, State
@session.menu
@@ -13,13 +8,14 @@ def episodes(ctx: Context, state: State) -> State | InternalDirective:
Displays available episodes for a selected provider anime and handles
the logic for continuing from watch history or manual selection.
"""
provider_anime = state.provider.anime
anilist_anime = state.media_api.anime
config = ctx.config
feedback = ctx.services.feedback
feedback.clear_console()
if not provider_anime or not anilist_anime:
provider_anime = state.provider.anime
media_item = state.media_api.media_item
if not provider_anime or not media_item:
feedback.error("Error: Anime details are missing.")
return InternalDirective.BACK
@@ -46,7 +42,7 @@ def episodes(ctx: Context, state: State) -> State | InternalDirective:
from ...utils.previews import get_episode_preview
preview_command = get_episode_preview(
available_episodes, anilist_anime, ctx.config
available_episodes, media_item, ctx.config
)
chosen_episode_str = ctx.selector.choose(
@@ -68,7 +64,7 @@ def episodes(ctx: Context, state: State) -> State | InternalDirective:
pass
return State(
menu_name="SERVERS",
menu_name=MenuName.SERVERS,
media_api=state.media_api,
provider=state.provider.model_copy(update={"episode_number": chosen_episode}),
provider=state.provider.model_copy(update={"episode": chosen_episode}),
)

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

@@ -1,101 +1,163 @@
import threading
from typing import TYPE_CHECKING, Callable, Dict
import click
from rich.console import Console
from typing import Callable, Dict, Union
from ..session import Context, session
from ..state import InternalDirective, State
from ..state import InternalDirective, MenuName, State
if TYPE_CHECKING:
from ....libs.providers.anime.types import Server
MenuAction = Callable[[], Union[State, InternalDirective]]
@session.menu
def player_controls(ctx: Context, state: State) -> State | InternalDirective:
"""
Handles post-playback options like playing the next episode,
replaying, or changing streaming options.
"""
# --- State and Context Extraction ---
def player_controls(ctx: Context, state: State) -> Union[State, InternalDirective]:
feedback = ctx.services.feedback
feedback.clear_console()
config = ctx.config
player = ctx.player
selector = ctx.selector
console = Console()
console.clear()
provider_anime = state.provider.anime
anilist_anime = state.media_api.anime
current_episode_num = state.provider.episode_number
selected_server = state.provider.selected_server
all_servers = state.provider.servers
player_result = state.provider.last_player_result
media_item = state.media_api.media_item
current_episode_num = state.provider.episode
selected_server = state.provider.server
server_map = state.provider.servers
if not all(
(
provider_anime,
anilist_anime,
current_episode_num,
selected_server,
all_servers,
)
if (
not provider_anime
or not media_item
or not current_episode_num
or not selected_server
or not server_map
):
console.print(
"[bold red]Error: Player state is incomplete. Returning.[/bold red]"
)
feedback.error("Player state is incomplete. Returning.")
return InternalDirective.BACK
# --- Auto-Next Logic ---
available_episodes = getattr(
provider_anime.episodes, config.stream.translation_type, []
)
current_index = available_episodes.index(current_episode_num)
if config.stream.auto_next and current_index < len(available_episodes) - 1:
console.print("[cyan]Auto-playing next episode...[/cyan]")
feedback.info("Auto-playing next episode...")
next_episode_num = available_episodes[current_index + 1]
# Track next episode in unified media registry
return State(
menu_name="SERVERS",
menu_name=MenuName.SERVERS,
media_api=state.media_api,
provider=state.provider.model_copy(
update={"episode_number": next_episode_num}
),
)
# --- Action Definitions ---
def next_episode() -> State | InternalDirective:
# --- Menu Options ---
icons = config.general.icons
options: Dict[str, Callable[[], Union[State, InternalDirective]]] = {}
if current_index < len(available_episodes) - 1:
options[f"{'⏭️ ' if icons else ''}Next Episode"] = _next_episode(ctx, state)
options.update(
{
f"{'🔄 ' if icons else ''}Replay Episode": _replay(ctx, state),
f"{'💻 ' if icons else ''}Change Server": _change_server(ctx, state),
f"{'🎞️ ' if icons else ''}Back to Episode List": lambda: State(
menu_name=MenuName.EPISODES,
media_api=state.media_api,
provider=state.provider,
),
f"{'🏠 ' if icons else ''}Main Menu": lambda: State(
menu_name=MenuName.MAIN
),
f"{'❌ ' if icons else ''}Exit": lambda: InternalDirective.EXIT,
}
)
choice = selector.choose(prompt="What's next?", choices=list(options.keys()))
if choice and choice in options:
return options[choice]()
return InternalDirective.BACK
def _next_episode(ctx: Context, state: State) -> MenuAction:
def action():
feedback = ctx.services.feedback
feedback.clear_console()
config = ctx.config
provider_anime = state.provider.anime
media_item = state.media_api.media_item
current_episode_num = state.provider.episode
selected_server = state.provider.server
server_map = state.provider.servers
if (
not provider_anime
or not media_item
or not current_episode_num
or not selected_server
or not server_map
):
feedback.error("Player state is incomplete. Returning.")
return InternalDirective.BACK
available_episodes = getattr(
provider_anime.episodes, config.stream.translation_type, []
)
current_index = available_episodes.index(current_episode_num)
if current_index < len(available_episodes) - 1:
next_episode_num = available_episodes[current_index + 1]
# Transition back to the SERVERS menu with the new episode number.
return State(
menu_name="SERVERS",
menu_name=MenuName.SERVERS,
media_api=state.media_api,
provider=state.provider.model_copy(
update={"episode_number": next_episode_num}
),
)
console.print("[bold yellow]This is the last available episode.[/bold yellow]")
feedback.warning("This is the last available episode.")
return InternalDirective.RELOAD
def replay() -> State | InternalDirective:
# We don't need to change state, just re-trigger the SERVERS menu's logic.
return State(
menu_name="SERVERS", media_api=state.media_api, provider=state.provider
)
return action
def _replay(ctx: Context, state: State) -> MenuAction:
def action():
return InternalDirective.BACK
return action
def _change_server(ctx: Context, state: State) -> MenuAction:
def action():
feedback = ctx.services.feedback
feedback.clear_console()
selector = ctx.selector
provider_anime = state.provider.anime
media_item = state.media_api.media_item
current_episode_num = state.provider.episode
selected_server = state.provider.server
server_map = state.provider.servers
if (
not provider_anime
or not media_item
or not current_episode_num
or not selected_server
or not server_map
):
feedback.error("Player state is incomplete. Returning.")
return InternalDirective.BACK
def change_server() -> State | InternalDirective:
server_map: Dict[str, Server] = {s.name: s for s in all_servers}
new_server_name = selector.choose(
"Select a different server:", list(server_map.keys())
)
if new_server_name:
# Update the selected server and re-run the SERVERS logic.
return State(
menu_name="SERVERS",
menu_name=MenuName.SERVERS,
media_api=state.media_api,
provider=state.provider.model_copy(
update={"selected_server": server_map[new_server_name]}
@@ -103,32 +165,4 @@ def player_controls(ctx: Context, state: State) -> State | InternalDirective:
)
return InternalDirective.RELOAD
# --- Menu Options ---
icons = config.general.icons
options: Dict[str, Callable[[], State | InternalDirective]] = {}
if current_index < len(available_episodes) - 1:
options[f"{'⏭️ ' if icons else ''}Next Episode"] = next_episode
options.update(
{
f"{'🔄 ' if icons else ''}Replay Episode": replay,
f"{'💻 ' if icons else ''}Change Server": change_server,
f"{'🎞️ ' if icons else ''}Back to Episode List": lambda: State(
menu_name="EPISODES", media_api=state.media_api, provider=state.provider
),
f"{'🏠 ' if icons else ''}Main Menu": lambda: State(menu_name="MAIN"),
f"{'❌ ' if icons else ''}Exit": lambda: InternalDirective.EXIT,
}
)
# --- Prompt and Execute ---
header = f"Finished Episode {current_episode_num} of {provider_anime.title}"
choice_str = selector.choose(
prompt="What's next?", choices=list(options.keys()), header=header
)
if choice_str and choice_str in options:
return options[choice_str]()
return InternalDirective.BACK
return action

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

@@ -1,20 +1,17 @@
from typing import TYPE_CHECKING
from rich.console import Console
from rich.progress import Progress
from thefuzz import fuzz
from ....libs.providers.anime.params import SearchParams
from ....libs.providers.anime.types import SearchResult
from ..session import Context, session
from ..state import InternalDirective, ProviderState, State
from ..state import InternalDirective, MenuName, ProviderState, State
@session.menu
def provider_search(ctx: Context, state: State) -> State | InternalDirective:
feedback = ctx.services.feedback
anilist_anime = state.media_api.anime
if not anilist_anime:
media_item = state.media_api.media_item
if not media_item:
feedback.error("No AniList anime to search for", "Please select an anime first")
return InternalDirective.BACK
@@ -23,8 +20,8 @@ def provider_search(ctx: Context, state: State) -> State | InternalDirective:
config = ctx.config
feedback.clear_console()
anilist_title = anilist_anime.title.english or anilist_anime.title.romaji
if not anilist_title:
media_title = media_item.title.english or media_item.title.romaji
if not media_title:
feedback.error(
"Selected anime has no searchable title",
"This anime entry is missing required title information",
@@ -32,14 +29,12 @@ def provider_search(ctx: Context, state: State) -> State | InternalDirective:
return InternalDirective.BACK
provider_search_results = provider.search(
SearchParams(
query=anilist_title, translation_type=config.stream.translation_type
)
SearchParams(query=media_title, translation_type=config.stream.translation_type)
)
if not provider_search_results or not provider_search_results.results:
feedback.warning(
f"Could not find '{anilist_title}' on {provider.__class__.__name__}",
f"Could not find '{media_title}' on {provider.__class__.__name__}",
"Try another provider from the config or go back to search again",
)
return InternalDirective.BACK
@@ -55,7 +50,7 @@ def provider_search(ctx: Context, state: State) -> State | InternalDirective:
# Use fuzzy matching to find the best title
best_match_title = max(
provider_results_map.keys(),
key=lambda p_title: fuzz.ratio(p_title.lower(), anilist_title.lower()),
key=lambda p_title: fuzz.ratio(p_title.lower(), media_title.lower()),
)
feedback.info("Auto-selecting best match: {best_match_title}")
selected_provider_anime = provider_results_map[best_match_title]
@@ -64,7 +59,7 @@ def provider_search(ctx: Context, state: State) -> State | InternalDirective:
choices.append("Back")
chosen_title = selector.choose(
prompt=f"Confirm match for '{anilist_title}'", choices=choices
prompt=f"Confirm match for '{media_title}'", choices=choices
)
if not chosen_title or chosen_title == "Back":
@@ -81,7 +76,7 @@ def provider_search(ctx: Context, state: State) -> State | InternalDirective:
from ....libs.providers.anime.params import AnimeParams
full_provider_anime = provider.get(
AnimeParams(id=selected_provider_anime.id, query=anilist_title.lower())
AnimeParams(id=selected_provider_anime.id, query=media_title.lower())
)
if not full_provider_anime:
@@ -91,7 +86,7 @@ def provider_search(ctx: Context, state: State) -> State | InternalDirective:
return InternalDirective.BACK
return State(
menu_name="EPISODES",
menu_name=MenuName.EPISODES,
media_api=state.media_api,
provider=ProviderState(
search_results=provider_search_results,

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

@@ -1,54 +1,33 @@
from typing import Dict, List
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 ....libs.providers.anime.types import ProviderServer, Server
from ..session import Context, session
from ..state import InternalDirective, State
def _filter_by_quality(links, quality):
# Simplified version of your filter_by_quality for brevity
for link in links:
if str(link.quality) == quality:
return link
return links[0] if links else None
from ..state import InternalDirective, MenuName, State
@session.menu
def servers(ctx: Context, state: State) -> State | InternalDirective:
"""
Fetches and displays available streaming servers for a chosen episode,
then launches the media player and transitions to post-playback controls.
"""
provider_anime = state.provider.anime
if not state.media_api.anime:
return InternalDirective.BACK
anime_title = (
state.media_api.anime.title.romaji or state.media_api.anime.title.english
)
episode_number = state.provider.episode_number
feedback = ctx.services.feedback
config = ctx.config
provider = ctx.provider
selector = ctx.selector
console = Console()
console.clear()
provider_anime = state.provider.anime
media_item = state.media_api.media_item
if not media_item:
return InternalDirective.BACK
anime_title = media_item.title.romaji or media_item.title.english
episode_number = state.provider.episode
if not provider_anime or not episode_number:
console.print(
"[bold red]Error: Anime or episode details are missing.[/bold red]"
)
selector.ask("Enter to continue...")
feedback.error("Anime or episode details are missing")
return InternalDirective.BACK
# --- Fetch Server Streams ---
with Progress(transient=True) as progress:
progress.add_task(
f"[cyan]Fetching servers for episode {episode_number}...", total=None
)
with feedback.progress("Fetching Servers"):
server_iterator = provider.episode_streams(
EpisodeStreamsParams(
anime_id=provider_anime.id,
@@ -58,27 +37,28 @@ def servers(ctx: Context, state: State) -> State | InternalDirective:
)
)
# Consume the iterator to get a list of all servers
all_servers: List[Server] = list(server_iterator) if server_iterator else []
if config.stream.server == ProviderServer.TOP and server_iterator:
try:
all_servers = [next(server_iterator)]
except Exception as e:
all_servers = []
else:
all_servers: List[Server] = list(server_iterator) if server_iterator else []
if not all_servers:
console.print(
f"[bold yellow]No streaming servers found for this episode.[/bold yellow]"
)
feedback.error(f"o streaming servers found for this episode")
return InternalDirective.BACK
# --- Auto-Select or Prompt for Server ---
server_map: Dict[str, Server] = {s.name: s for s in all_servers}
selected_server: Server | None = None
preferred_server = config.stream.server.value.lower()
if preferred_server == "top":
selected_server = all_servers[0]
console.print(f"[cyan]Auto-selecting top server:[/] {selected_server.name}")
feedback.info(f"Auto-selecting top server: {selected_server.name}")
elif preferred_server in server_map:
selected_server = server_map[preferred_server]
console.print(
f"[cyan]Auto-selecting preferred server:[/] {selected_server.name}"
)
feedback.info(f"Auto-selecting preferred server: {selected_server.name}")
else:
choices = [*server_map.keys(), "Back"]
chosen_name = selector.choose("Select Server", choices)
@@ -88,14 +68,13 @@ def servers(ctx: Context, state: State) -> State | InternalDirective:
stream_link_obj = _filter_by_quality(selected_server.links, config.stream.quality)
if not stream_link_obj:
console.print(
f"[bold red]No stream of quality '{config.stream.quality}' found on server '{selected_server.name}'.[/bold red]"
feedback.error(
f"No stream of quality '{config.stream.quality}' found on server '{selected_server.name}'."
)
return InternalDirective.RELOAD
# --- Launch Player ---
final_title = f"{provider_anime.title} - Ep {episode_number}"
console.print(f"[bold green]Launching player for:[/] {final_title}")
feedback.info(f"[bold green]Launching player for:[/] {final_title}")
player_result = ctx.player.play(
PlayerParams(
@@ -105,19 +84,24 @@ def servers(ctx: Context, state: State) -> State | InternalDirective:
headers=selected_server.headers,
)
)
if state.media_api.anime and state.provider.episode_number:
ctx.services.watch_history.track(
state.media_api.anime, state.provider.episode_number, player_result
)
if media_item and episode_number:
ctx.services.watch_history.track(media_item, episode_number, player_result)
return State(
menu_name="PLAYER_CONTROLS",
menu_name=MenuName.PLAYER_CONTROLS,
media_api=state.media_api,
provider=state.provider.model_copy(
update={
"servers": all_servers,
"selected_server": selected_server,
"last_player_result": player_result,
"servers": server_map,
"server_name": selected_server.name,
}
),
)
def _filter_by_quality(links, quality):
# Simplified version of your filter_by_quality for brevity
for link in links:
if str(link.quality) == quality:
return link
return links[0] if links else None

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

@@ -14,6 +14,8 @@ class InternalDirective(Enum):
BACK = auto()
BACK_FORCE = auto()
BACKX2 = auto()
BACKX3 = auto()