mirror of
https://github.com/Benexl/FastAnime.git
synced 2026-01-02 07:50:04 -08:00
chore: correct package issues
This commit is contained in:
Benexl
92
viu_cli/cli/interactive/menu/media/download_episodes.py
Normal file
92
viu_cli/cli/interactive/menu/media/download_episodes.py
Normal file
@@ -0,0 +1,92 @@
|
||||
from .....libs.provider.anime.params import AnimeParams, SearchParams
|
||||
from ...session import Context, session
|
||||
from ...state import InternalDirective, State
|
||||
|
||||
|
||||
@session.menu
|
||||
def download_episodes(ctx: Context, state: State) -> State | InternalDirective:
|
||||
"""Menu to select and download episodes synchronously."""
|
||||
from .....core.utils.fuzzy import fuzz
|
||||
from .....core.utils.normalizer import normalize_title
|
||||
from ....service.download.service import DownloadService
|
||||
|
||||
feedback = ctx.feedback
|
||||
selector = ctx.selector
|
||||
media_item = state.media_api.media_item
|
||||
config = ctx.config
|
||||
provider = ctx.provider
|
||||
|
||||
if not media_item:
|
||||
feedback.error("No media item selected for download.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
media_title = media_item.title.english or media_item.title.romaji
|
||||
if not media_title:
|
||||
feedback.error("Cannot download: Media item has no title.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
# Step 1: Find the anime on the provider to get a full episode list
|
||||
with feedback.progress(
|
||||
f"Searching for '{media_title}' on {provider.__class__.__name__}..."
|
||||
):
|
||||
provider_search_results = provider.search(
|
||||
SearchParams(
|
||||
query=normalize_title(media_title, config.general.provider.value, True)
|
||||
)
|
||||
)
|
||||
|
||||
if not provider_search_results or not provider_search_results.results:
|
||||
feedback.warning(f"Could not find '{media_title}' on provider.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
provider_results_map = {res.title: res for res in provider_search_results.results}
|
||||
best_match_title = max(
|
||||
provider_results_map.keys(),
|
||||
key=lambda p_title: fuzz.ratio(
|
||||
normalize_title(p_title, config.general.provider.value).lower(),
|
||||
media_title.lower(),
|
||||
),
|
||||
)
|
||||
selected_provider_anime_ref = provider_results_map[best_match_title]
|
||||
|
||||
with feedback.progress(f"Fetching episode list for '{best_match_title}'..."):
|
||||
full_provider_anime = provider.get(
|
||||
AnimeParams(id=selected_provider_anime_ref.id, query=media_title)
|
||||
)
|
||||
|
||||
if not full_provider_anime:
|
||||
feedback.warning(f"Failed to fetch details for '{best_match_title}'.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
available_episodes = getattr(
|
||||
full_provider_anime.episodes, config.stream.translation_type, []
|
||||
)
|
||||
if not available_episodes:
|
||||
feedback.warning("No episodes found for download.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
# Step 2: Let user select episodes
|
||||
selected_episodes = selector.choose_multiple(
|
||||
"Select episodes to download (TAB to select, ENTER to confirm)",
|
||||
choices=available_episodes,
|
||||
)
|
||||
|
||||
if not selected_episodes:
|
||||
feedback.info("No episodes selected for download.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
# Step 3: Download episodes synchronously
|
||||
# TODO: move to main ctx
|
||||
download_service = DownloadService(
|
||||
config, ctx.media_registry, ctx.media_api, ctx.provider
|
||||
)
|
||||
|
||||
feedback.info(
|
||||
f"Starting download of {len(selected_episodes)} episodes. This may take a while..."
|
||||
)
|
||||
download_service.download_episodes_sync(media_item, selected_episodes)
|
||||
|
||||
feedback.success(f"Finished downloading {len(selected_episodes)} episodes.")
|
||||
|
||||
# After downloading, return to the media actions menu
|
||||
return InternalDirective.BACK
|
||||
243
viu_cli/cli/interactive/menu/media/downloads.py
Normal file
243
viu_cli/cli/interactive/menu/media/downloads.py
Normal file
@@ -0,0 +1,243 @@
|
||||
import logging
|
||||
import random
|
||||
from typing import Callable, Dict
|
||||
|
||||
from .....libs.media_api.params import MediaSearchParams
|
||||
from .....libs.media_api.types import (
|
||||
MediaSort,
|
||||
MediaStatus,
|
||||
UserMediaListStatus,
|
||||
)
|
||||
from ...session import Context, session
|
||||
from ...state import InternalDirective, MediaApiState, MenuName, State
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
MenuAction = Callable[[], State | InternalDirective]
|
||||
|
||||
|
||||
@session.menu
|
||||
def downloads(ctx: Context, state: State) -> State | InternalDirective:
|
||||
"""Downloads menu showing locally stored media from registry."""
|
||||
icons = ctx.config.general.icons
|
||||
feedback = ctx.feedback
|
||||
feedback.clear_console()
|
||||
|
||||
options: Dict[str, MenuAction] = {
|
||||
f"{'🔥 ' if icons else ''}Trending (Local)": _create_local_media_list_action(
|
||||
ctx, state, MediaSort.TRENDING_DESC
|
||||
),
|
||||
f"{'🎞️ ' if icons else ''}Recent (Local)": _create_local_recent_media_action(
|
||||
ctx, state
|
||||
),
|
||||
f"{'📺 ' if icons else ''}Watching (Local)": _create_local_status_action(
|
||||
ctx, state, UserMediaListStatus.WATCHING
|
||||
),
|
||||
f"{'🔁 ' if icons else ''}Rewatching (Local)": _create_local_status_action(
|
||||
ctx, state, UserMediaListStatus.REPEATING
|
||||
),
|
||||
f"{'⏸️ ' if icons else ''}Paused (Local)": _create_local_status_action(
|
||||
ctx, state, UserMediaListStatus.PAUSED
|
||||
),
|
||||
f"{'📑 ' if icons else ''}Planned (Local)": _create_local_status_action(
|
||||
ctx, state, UserMediaListStatus.PLANNING
|
||||
),
|
||||
f"{'🔎 ' if icons else ''}Search (Local)": _create_local_search_media_list(
|
||||
ctx, state
|
||||
),
|
||||
f"{'🔔 ' if icons else ''}Recently Updated (Local)": _create_local_media_list_action(
|
||||
ctx, state, MediaSort.UPDATED_AT_DESC
|
||||
),
|
||||
f"{'✨ ' if icons else ''}Popular (Local)": _create_local_media_list_action(
|
||||
ctx, state, MediaSort.POPULARITY_DESC
|
||||
),
|
||||
f"{'💯 ' if icons else ''}Top Scored (Local)": _create_local_media_list_action(
|
||||
ctx, state, MediaSort.SCORE_DESC
|
||||
),
|
||||
f"{'💖 ' if icons else ''}Favourites (Local)": _create_local_media_list_action(
|
||||
ctx, state, MediaSort.FAVOURITES_DESC
|
||||
),
|
||||
f"{'🎲 ' if icons else ''}Random (Local)": _create_local_random_media_list(
|
||||
ctx, state
|
||||
),
|
||||
f"{'🎬 ' if icons else ''}Upcoming (Local)": _create_local_media_list_action(
|
||||
ctx, state, MediaSort.POPULARITY_DESC, MediaStatus.NOT_YET_RELEASED
|
||||
),
|
||||
f"{'✅ ' if icons else ''}Completed (Local)": _create_local_status_action(
|
||||
ctx, state, UserMediaListStatus.COMPLETED
|
||||
),
|
||||
f"{'🚮 ' if icons else ''}Dropped (Local)": _create_local_status_action(
|
||||
ctx, state, UserMediaListStatus.DROPPED
|
||||
),
|
||||
f"{'↩️ ' if icons else ''}Back to Main": lambda: InternalDirective.BACK,
|
||||
f"{'❌ ' if icons else ''}Exit": lambda: InternalDirective.EXIT,
|
||||
}
|
||||
|
||||
choice = ctx.selector.choose(
|
||||
prompt="Select Downloads Category",
|
||||
choices=list(options.keys()),
|
||||
)
|
||||
if not choice:
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
selected_action = options[choice]
|
||||
next_step = selected_action()
|
||||
return next_step
|
||||
|
||||
|
||||
def _create_local_media_list_action(
|
||||
ctx: Context, state: State, sort: MediaSort, status: MediaStatus | None = None
|
||||
) -> MenuAction:
|
||||
"""Create action for searching local media with sorting and optional status filter."""
|
||||
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
search_params = MediaSearchParams(sort=sort, status=status)
|
||||
|
||||
loading_message = "Searching local media registry"
|
||||
result = None
|
||||
with feedback.progress(loading_message):
|
||||
result = ctx.media_registry.search_for_media(search_params)
|
||||
|
||||
if result and result.media:
|
||||
return State(
|
||||
menu_name=MenuName.RESULTS,
|
||||
media_api=MediaApiState(
|
||||
search_result={
|
||||
media_item.id: media_item for media_item in result.media
|
||||
},
|
||||
search_params=search_params,
|
||||
page_info=result.page_info,
|
||||
),
|
||||
)
|
||||
else:
|
||||
feedback.info("No media found in local registry")
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _create_local_random_media_list(ctx: Context, state: State) -> MenuAction:
|
||||
"""Create action for getting random local media."""
|
||||
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
|
||||
loading_message = "Getting random local media"
|
||||
with feedback.progress(loading_message):
|
||||
# Get all records and pick random ones
|
||||
all_records = list(ctx.media_registry.get_all_media_records())
|
||||
|
||||
if not all_records:
|
||||
feedback.info("No media found in local registry")
|
||||
return InternalDirective.BACK
|
||||
|
||||
# Get up to 50 random records
|
||||
random_records = random.sample(all_records, min(50, len(all_records)))
|
||||
random_ids = [record.media_item.id for record in random_records]
|
||||
|
||||
search_params = MediaSearchParams(id_in=random_ids)
|
||||
result = ctx.media_registry.search_for_media(search_params)
|
||||
|
||||
if result and result.media:
|
||||
return State(
|
||||
menu_name=MenuName.RESULTS,
|
||||
media_api=MediaApiState(
|
||||
search_result={
|
||||
media_item.id: media_item for media_item in result.media
|
||||
},
|
||||
search_params=search_params,
|
||||
page_info=result.page_info,
|
||||
),
|
||||
)
|
||||
else:
|
||||
feedback.info("No media found in local registry")
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _create_local_search_media_list(ctx: Context, state: State) -> MenuAction:
|
||||
"""Create action for searching local media by query."""
|
||||
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
|
||||
query = ctx.selector.ask("Search Local Anime")
|
||||
if not query:
|
||||
return InternalDirective.BACK
|
||||
|
||||
search_params = MediaSearchParams(query=query)
|
||||
|
||||
loading_message = "Searching local media registry"
|
||||
result = None
|
||||
with feedback.progress(loading_message):
|
||||
result = ctx.media_registry.search_for_media(search_params)
|
||||
|
||||
if result and result.media:
|
||||
return State(
|
||||
menu_name=MenuName.RESULTS,
|
||||
media_api=MediaApiState(
|
||||
search_result={
|
||||
media_item.id: media_item for media_item in result.media
|
||||
},
|
||||
search_params=search_params,
|
||||
page_info=result.page_info,
|
||||
),
|
||||
)
|
||||
else:
|
||||
feedback.info("No media found in local registry")
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _create_local_status_action(
|
||||
ctx: Context, state: State, status: UserMediaListStatus
|
||||
) -> MenuAction:
|
||||
"""Create action for getting local media by user status."""
|
||||
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
|
||||
loading_message = f"Getting {status.value} media from local registry"
|
||||
result = None
|
||||
with feedback.progress(loading_message):
|
||||
result = ctx.media_registry.get_media_by_status(status)
|
||||
|
||||
if result and result.media:
|
||||
return State(
|
||||
menu_name=MenuName.RESULTS,
|
||||
media_api=MediaApiState(
|
||||
search_result={
|
||||
media_item.id: media_item for media_item in result.media
|
||||
},
|
||||
page_info=result.page_info,
|
||||
),
|
||||
)
|
||||
else:
|
||||
feedback.info(f"No {status.value} media found in local registry")
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _create_local_recent_media_action(ctx: Context, state: State) -> MenuAction:
|
||||
"""Create action for getting recently watched local media."""
|
||||
|
||||
def action():
|
||||
result = ctx.media_registry.get_recently_watched()
|
||||
if result and result.media:
|
||||
return State(
|
||||
menu_name=MenuName.RESULTS,
|
||||
media_api=MediaApiState(
|
||||
search_result={
|
||||
media_item.id: media_item for media_item in result.media
|
||||
},
|
||||
page_info=result.page_info,
|
||||
),
|
||||
)
|
||||
else:
|
||||
ctx.feedback.info("No recently watched media found in local registry")
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
119
viu_cli/cli/interactive/menu/media/dynamic_search.py
Normal file
119
viu_cli/cli/interactive/menu/media/dynamic_search.py
Normal file
@@ -0,0 +1,119 @@
|
||||
import json
|
||||
import logging
|
||||
|
||||
from .....core.constants import APP_CACHE_DIR, SCRIPTS_DIR
|
||||
from .....libs.media_api.params import MediaSearchParams
|
||||
from ...session import Context, session
|
||||
from ...state import InternalDirective, MediaApiState, MenuName, State
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SEARCH_CACHE_DIR = APP_CACHE_DIR / "search"
|
||||
SEARCH_RESULTS_FILE = SEARCH_CACHE_DIR / "current_search_results.json"
|
||||
FZF_SCRIPTS_DIR = SCRIPTS_DIR / "fzf"
|
||||
SEARCH_TEMPLATE_SCRIPT = (FZF_SCRIPTS_DIR / "search.template.sh").read_text(
|
||||
encoding="utf-8"
|
||||
)
|
||||
|
||||
|
||||
@session.menu
|
||||
def dynamic_search(ctx: Context, state: State) -> State | InternalDirective:
|
||||
"""Dynamic search menu that provides real-time search results."""
|
||||
feedback = ctx.feedback
|
||||
feedback.clear_console()
|
||||
|
||||
# Ensure cache directory exists
|
||||
SEARCH_CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Read the GraphQL search query
|
||||
from .....libs.media_api.anilist import gql
|
||||
|
||||
search_query = gql.SEARCH_MEDIA.read_text(encoding="utf-8")
|
||||
# Properly escape the GraphQL query for JSON
|
||||
search_query_escaped = json.dumps(search_query)
|
||||
|
||||
# Prepare the search script
|
||||
auth_header = ""
|
||||
profile = ctx.auth.get_auth()
|
||||
if ctx.media_api.is_authenticated() and profile:
|
||||
auth_header = f"Bearer {profile.token}"
|
||||
|
||||
search_command = SEARCH_TEMPLATE_SCRIPT
|
||||
|
||||
replacements = {
|
||||
"GRAPHQL_ENDPOINT": "https://graphql.anilist.co",
|
||||
"GRAPHQL_QUERY": search_query_escaped,
|
||||
"CACHE_DIR": str(SEARCH_CACHE_DIR),
|
||||
"SEARCH_RESULTS_FILE": str(SEARCH_RESULTS_FILE),
|
||||
"AUTH_HEADER": auth_header,
|
||||
}
|
||||
|
||||
for key, value in replacements.items():
|
||||
search_command = search_command.replace(f"{{{key}}}", str(value))
|
||||
|
||||
try:
|
||||
# Prepare preview functionality
|
||||
preview_command = None
|
||||
if ctx.config.general.preview != "none":
|
||||
from ....utils.preview import create_preview_context
|
||||
|
||||
with create_preview_context() as preview_ctx:
|
||||
preview_command = preview_ctx.get_dynamic_anime_preview(ctx.config)
|
||||
|
||||
choice = ctx.selector.search(
|
||||
prompt="Search Anime",
|
||||
search_command=search_command,
|
||||
preview=preview_command,
|
||||
)
|
||||
else:
|
||||
choice = ctx.selector.search(
|
||||
prompt="Search Anime",
|
||||
search_command=search_command,
|
||||
)
|
||||
except NotImplementedError:
|
||||
feedback.error("Dynamic search is not supported by your current selector")
|
||||
feedback.info("Please use the regular search option or switch to fzf selector")
|
||||
return InternalDirective.MAIN
|
||||
|
||||
if not choice:
|
||||
return InternalDirective.MAIN
|
||||
|
||||
# Read the cached search results
|
||||
if not SEARCH_RESULTS_FILE.exists():
|
||||
logger.error("Search results file not found")
|
||||
return InternalDirective.MAIN
|
||||
|
||||
with open(SEARCH_RESULTS_FILE, "r", encoding="utf-8") as f:
|
||||
raw_data = json.load(f)
|
||||
|
||||
# Transform the raw data into MediaSearchResult
|
||||
search_result = ctx.media_api.transform_raw_search_data(raw_data)
|
||||
|
||||
if not search_result or not search_result.media:
|
||||
feedback.info("No results found")
|
||||
return InternalDirective.MAIN
|
||||
|
||||
# Find the selected media item by matching the choice with the displayed format
|
||||
selected_media = None
|
||||
for media_item in search_result.media:
|
||||
if (
|
||||
media_item.title.english == choice.strip()
|
||||
or media_item.title.romaji == choice.strip()
|
||||
):
|
||||
selected_media = media_item
|
||||
break
|
||||
|
||||
if not selected_media:
|
||||
logger.error(f"Could not find selected media for choice: {choice}")
|
||||
return InternalDirective.MAIN
|
||||
|
||||
# Navigate to media actions with the selected item
|
||||
return State(
|
||||
menu_name=MenuName.MEDIA_ACTIONS,
|
||||
media_api=MediaApiState(
|
||||
search_result={media.id: media for media in search_result.media},
|
||||
media_id=selected_media.id,
|
||||
search_params=MediaSearchParams(),
|
||||
page_info=search_result.page_info,
|
||||
),
|
||||
)
|
||||
77
viu_cli/cli/interactive/menu/media/episodes.py
Normal file
77
viu_cli/cli/interactive/menu/media/episodes.py
Normal file
@@ -0,0 +1,77 @@
|
||||
from ...session import Context, session
|
||||
from ...state import InternalDirective, MenuName, State
|
||||
|
||||
|
||||
@session.menu
|
||||
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.
|
||||
"""
|
||||
config = ctx.config
|
||||
feedback = ctx.feedback
|
||||
feedback.clear_console()
|
||||
|
||||
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
|
||||
|
||||
available_episodes = getattr(
|
||||
provider_anime.episodes, config.stream.translation_type, []
|
||||
)
|
||||
if not available_episodes:
|
||||
feedback.warning(
|
||||
f"No '{config.stream.translation_type}' episodes found for this anime."
|
||||
)
|
||||
return InternalDirective.BACKX2
|
||||
|
||||
chosen_episode: str | None = None
|
||||
start_time: str | None = None
|
||||
|
||||
if config.stream.continue_from_watch_history:
|
||||
chosen_episode, start_time = ctx.watch_history.get_episode(media_item)
|
||||
|
||||
if not chosen_episode or ctx.switch.show_episodes_menu:
|
||||
choices = [*available_episodes, "Back"]
|
||||
|
||||
preview_command = None
|
||||
if ctx.config.general.preview != "none":
|
||||
from ....utils.preview import create_preview_context
|
||||
|
||||
with create_preview_context() as preview_ctx:
|
||||
preview_command = preview_ctx.get_episode_preview(
|
||||
available_episodes, media_item, ctx.config
|
||||
)
|
||||
|
||||
chosen_episode_str = ctx.selector.choose(
|
||||
prompt="Select Episode", choices=choices, preview=preview_command
|
||||
)
|
||||
|
||||
if not chosen_episode_str or chosen_episode_str == "Back":
|
||||
# TODO: should improve the back logic for menus that can be pass through
|
||||
return InternalDirective.BACKX2
|
||||
|
||||
chosen_episode = chosen_episode_str
|
||||
# Workers are automatically cleaned up when exiting the context
|
||||
else:
|
||||
# No preview mode
|
||||
chosen_episode_str = ctx.selector.choose(
|
||||
prompt="Select Episode", choices=choices, preview=None
|
||||
)
|
||||
|
||||
if not chosen_episode_str or chosen_episode_str == "Back":
|
||||
# TODO: should improve the back logic for menus that can be pass through
|
||||
return InternalDirective.BACKX2
|
||||
|
||||
chosen_episode = chosen_episode_str
|
||||
|
||||
return State(
|
||||
menu_name=MenuName.SERVERS,
|
||||
media_api=state.media_api,
|
||||
provider=state.provider.model_copy(
|
||||
update={"episode": chosen_episode, "start_time": start_time}
|
||||
),
|
||||
)
|
||||
242
viu_cli/cli/interactive/menu/media/main.py
Normal file
242
viu_cli/cli/interactive/menu/media/main.py
Normal file
@@ -0,0 +1,242 @@
|
||||
import logging
|
||||
import random
|
||||
from typing import Callable, Dict
|
||||
|
||||
from .....libs.media_api.params import MediaSearchParams, UserMediaListSearchParams
|
||||
from .....libs.media_api.types import (
|
||||
MediaSort,
|
||||
MediaStatus,
|
||||
UserMediaListStatus,
|
||||
)
|
||||
from ...session import Context, session
|
||||
from ...state import InternalDirective, MediaApiState, MenuName, State
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
MenuAction = Callable[[], State | InternalDirective]
|
||||
|
||||
|
||||
@session.menu
|
||||
def main(ctx: Context, state: State) -> State | InternalDirective:
|
||||
icons = ctx.config.general.icons
|
||||
feedback = ctx.feedback
|
||||
feedback.clear_console()
|
||||
|
||||
options: Dict[str, MenuAction] = {
|
||||
f"{'🔥 ' if icons else ''}Trending": _create_media_list_action(
|
||||
ctx, state, MediaSort.TRENDING_DESC
|
||||
),
|
||||
f"{'🎞️ ' if icons else ''}Recent": _create_recent_media_action(ctx, state),
|
||||
f"{'📺 ' if icons else ''}Watching": _create_user_list_action(
|
||||
ctx, state, UserMediaListStatus.WATCHING
|
||||
),
|
||||
f"{'🔁 ' if icons else ''}Rewatching": _create_user_list_action(
|
||||
ctx, state, UserMediaListStatus.REPEATING
|
||||
),
|
||||
f"{'⏸️ ' if icons else ''}Paused": _create_user_list_action(
|
||||
ctx, state, UserMediaListStatus.PAUSED
|
||||
),
|
||||
f"{'📑 ' if icons else ''}Planned": _create_user_list_action(
|
||||
ctx, state, UserMediaListStatus.PLANNING
|
||||
),
|
||||
f"{'🔎 ' if icons else ''}Search": _create_search_media_list(ctx, state),
|
||||
f"{'🔍 ' if icons else ''}Dynamic Search": _create_dynamic_search_action(
|
||||
ctx, state
|
||||
),
|
||||
f"{'🏠 ' if icons else ''}Downloads": _create_downloads_action(ctx, state),
|
||||
f"{'🔔 ' if icons else ''}Recently Updated": _create_media_list_action(
|
||||
ctx, state, MediaSort.UPDATED_AT_DESC
|
||||
),
|
||||
f"{'✨ ' if icons else ''}Popular": _create_media_list_action(
|
||||
ctx, state, MediaSort.POPULARITY_DESC
|
||||
),
|
||||
f"{'💯 ' if icons else ''}Top Scored": _create_media_list_action(
|
||||
ctx, state, MediaSort.SCORE_DESC
|
||||
),
|
||||
f"{'💖 ' if icons else ''}Favourites": _create_media_list_action(
|
||||
ctx, state, MediaSort.FAVOURITES_DESC
|
||||
),
|
||||
f"{'🎲 ' if icons else ''}Random": _create_random_media_list(ctx, state),
|
||||
f"{'🎬 ' if icons else ''}Upcoming": _create_media_list_action(
|
||||
ctx, state, MediaSort.POPULARITY_DESC, MediaStatus.NOT_YET_RELEASED
|
||||
),
|
||||
f"{'✅ ' if icons else ''}Completed": _create_user_list_action(
|
||||
ctx, state, UserMediaListStatus.COMPLETED
|
||||
),
|
||||
f"{'🚮 ' if icons else ''}Dropped": _create_user_list_action(
|
||||
ctx, state, UserMediaListStatus.DROPPED
|
||||
),
|
||||
f"{'📝 ' if icons else ''}Edit Config": lambda: InternalDirective.CONFIG_EDIT,
|
||||
f"{'❌ ' if icons else ''}Exit": lambda: InternalDirective.EXIT,
|
||||
}
|
||||
|
||||
choice = ctx.selector.choose(
|
||||
prompt="Select Category",
|
||||
choices=list(options.keys()),
|
||||
)
|
||||
if not choice:
|
||||
return InternalDirective.MAIN
|
||||
|
||||
selected_action = options[choice]
|
||||
|
||||
next_step = selected_action()
|
||||
return next_step
|
||||
|
||||
|
||||
def _create_media_list_action(
|
||||
ctx: Context, state: State, sort: MediaSort, status: MediaStatus | None = None
|
||||
) -> MenuAction:
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
search_params = MediaSearchParams(sort=sort, status=status)
|
||||
|
||||
loading_message = "Fetching media list"
|
||||
result = None
|
||||
with feedback.progress(loading_message):
|
||||
result = ctx.media_api.search_media(search_params)
|
||||
|
||||
if result:
|
||||
return State(
|
||||
menu_name=MenuName.RESULTS,
|
||||
media_api=MediaApiState(
|
||||
search_result={
|
||||
media_item.id: media_item for media_item in result.media
|
||||
},
|
||||
search_params=search_params,
|
||||
page_info=result.page_info,
|
||||
),
|
||||
)
|
||||
else:
|
||||
return InternalDirective.MAIN
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _create_random_media_list(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
search_params = MediaSearchParams(id_in=random.sample(range(1, 15000), k=50))
|
||||
|
||||
loading_message = "Fetching media list"
|
||||
result = None
|
||||
with feedback.progress(loading_message):
|
||||
result = ctx.media_api.search_media(search_params)
|
||||
|
||||
if result:
|
||||
return State(
|
||||
menu_name=MenuName.RESULTS,
|
||||
media_api=MediaApiState(
|
||||
search_result={
|
||||
media_item.id: media_item for media_item in result.media
|
||||
},
|
||||
search_params=search_params,
|
||||
page_info=result.page_info,
|
||||
),
|
||||
)
|
||||
else:
|
||||
return InternalDirective.MAIN
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _create_search_media_list(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
|
||||
query = ctx.selector.ask("Search for Anime")
|
||||
if not query:
|
||||
return InternalDirective.MAIN
|
||||
|
||||
search_params = MediaSearchParams(query=query)
|
||||
|
||||
loading_message = "Fetching media list"
|
||||
result = None
|
||||
with feedback.progress(loading_message):
|
||||
result = ctx.media_api.search_media(search_params)
|
||||
|
||||
if result:
|
||||
return State(
|
||||
menu_name=MenuName.RESULTS,
|
||||
media_api=MediaApiState(
|
||||
search_result={
|
||||
media_item.id: media_item for media_item in result.media
|
||||
},
|
||||
search_params=search_params,
|
||||
page_info=result.page_info,
|
||||
),
|
||||
)
|
||||
else:
|
||||
return InternalDirective.MAIN
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _create_user_list_action(
|
||||
ctx: Context, state: State, status: UserMediaListStatus
|
||||
) -> MenuAction:
|
||||
"""A factory to create menu actions for fetching user lists, handling authentication."""
|
||||
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
if not ctx.media_api.is_authenticated():
|
||||
feedback.error("You haven't logged in")
|
||||
return InternalDirective.MAIN
|
||||
|
||||
search_params = UserMediaListSearchParams(status=status)
|
||||
|
||||
loading_message = "Fetching media list"
|
||||
result = None
|
||||
with feedback.progress(loading_message):
|
||||
result = ctx.media_api.search_media_list(search_params)
|
||||
|
||||
if result:
|
||||
return State(
|
||||
menu_name=MenuName.RESULTS,
|
||||
media_api=MediaApiState(
|
||||
search_result={
|
||||
media_item.id: media_item for media_item in result.media
|
||||
},
|
||||
search_params=search_params,
|
||||
page_info=result.page_info,
|
||||
),
|
||||
)
|
||||
else:
|
||||
return InternalDirective.MAIN
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _create_recent_media_action(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
result = ctx.media_registry.get_recently_watched()
|
||||
if result:
|
||||
return State(
|
||||
menu_name=MenuName.RESULTS,
|
||||
media_api=MediaApiState(
|
||||
search_result={
|
||||
media_item.id: media_item for media_item in result.media
|
||||
},
|
||||
page_info=result.page_info,
|
||||
),
|
||||
)
|
||||
else:
|
||||
return InternalDirective.MAIN
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _create_downloads_action(ctx: Context, state: State) -> MenuAction:
|
||||
"""Create action to navigate to the downloads menu."""
|
||||
|
||||
def action():
|
||||
return State(menu_name=MenuName.DOWNLOADS)
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _create_dynamic_search_action(ctx: Context, state: State) -> MenuAction:
|
||||
"""Create action to navigate to the dynamic search menu."""
|
||||
|
||||
def action():
|
||||
return State(menu_name=MenuName.DYNAMIC_SEARCH)
|
||||
|
||||
return action
|
||||
749
viu_cli/cli/interactive/menu/media/media_actions.py
Normal file
749
viu_cli/cli/interactive/menu/media/media_actions.py
Normal file
@@ -0,0 +1,749 @@
|
||||
from typing import Callable, Dict, Literal, Optional
|
||||
|
||||
from .....libs.media_api.params import (
|
||||
MediaRecommendationParams,
|
||||
MediaRelationsParams,
|
||||
UpdateUserMediaListEntryParams,
|
||||
)
|
||||
from .....libs.media_api.types import (
|
||||
MediaItem,
|
||||
MediaStatus,
|
||||
UserMediaListStatus,
|
||||
)
|
||||
from .....libs.player.params import PlayerParams
|
||||
from ...session import Context, session
|
||||
from ...state import InternalDirective, MediaApiState, MenuName, State
|
||||
|
||||
MenuAction = Callable[[], State | InternalDirective]
|
||||
|
||||
|
||||
@session.menu
|
||||
def media_actions(ctx: Context, state: State) -> State | InternalDirective:
|
||||
from ....service.registry.service import DownloadStatus
|
||||
|
||||
feedback = ctx.feedback
|
||||
|
||||
icons = ctx.config.general.icons
|
||||
|
||||
media_item = state.media_api.media_item
|
||||
|
||||
if not media_item:
|
||||
feedback.error("Media item is not in state")
|
||||
return InternalDirective.BACK
|
||||
progress = _get_progress_string(ctx, state.media_api.media_item)
|
||||
|
||||
# Check for downloaded episodes to conditionally show options
|
||||
record = ctx.media_registry.get_media_record(media_item.id)
|
||||
has_downloads = False
|
||||
if record:
|
||||
has_downloads = any(
|
||||
ep.download_status == DownloadStatus.COMPLETED
|
||||
and ep.file_path
|
||||
and ep.file_path.exists()
|
||||
for ep in record.media_episodes
|
||||
)
|
||||
|
||||
options: Dict[str, MenuAction] = {
|
||||
f"{'▶️ ' if icons else ''}Stream {progress}": _stream(ctx, state),
|
||||
f"{'📽️ ' if icons else ''}Episodes": _stream(
|
||||
ctx, state, force_episodes_menu=True
|
||||
),
|
||||
}
|
||||
|
||||
if has_downloads:
|
||||
options[f"{'💾 ' if icons else ''}Stream (Downloads)"] = _stream_downloads(
|
||||
ctx, state
|
||||
)
|
||||
options[f"{'💿 ' if icons else ''}Episodes (Downloads)"] = _stream_downloads(
|
||||
ctx, state, force_episodes_menu=True
|
||||
)
|
||||
|
||||
options.update(
|
||||
{
|
||||
f"{'📥 ' if icons else ''}Download": _download_episodes(ctx, state),
|
||||
f"{'📼 ' if icons else ''}Watch Trailer": _watch_trailer(ctx, state),
|
||||
f"{'🔗 ' if icons else ''}Recommendations": _view_recommendations(
|
||||
ctx, state
|
||||
),
|
||||
f"{'🔄 ' if icons else ''}Related Anime": _view_relations(ctx, state),
|
||||
f"{'👥 ' if icons else ''}Characters": _view_characters(ctx, state),
|
||||
f"{'📅 ' if icons else ''}Airing Schedule": _view_airing_schedule(
|
||||
ctx, state
|
||||
),
|
||||
f"{'📝 ' if icons else ''}View Reviews": _view_reviews(ctx, state),
|
||||
f"{'➕ ' if icons else ''}Add/Update List": _manage_user_media_list(
|
||||
ctx, state
|
||||
),
|
||||
f"{'➕ ' if icons else ''}Add/Update List (Bulk)": _manage_user_media_list_in_bulk(
|
||||
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 ''}Change Provider (Current: {ctx.config.general.provider.value.upper()})": _change_provider(
|
||||
ctx, state
|
||||
),
|
||||
f"{'🔘 ' if icons else ''}Toggle Auto Select Anime (Current: {ctx.config.general.auto_select_anime_result})": _toggle_config_state(
|
||||
ctx, state, "AUTO_ANIME"
|
||||
),
|
||||
f"{'🔘 ' if icons else ''}Toggle Auto Next Episode (Current: {ctx.config.stream.auto_next})": _toggle_config_state(
|
||||
ctx, state, "AUTO_EPISODE"
|
||||
),
|
||||
f"{'🔘 ' if icons else ''}Toggle Continue From History (Current: {ctx.config.stream.continue_from_watch_history})": _toggle_config_state(
|
||||
ctx, state, "CONTINUE_FROM_HISTORY"
|
||||
),
|
||||
f"{'🔘 ' if icons else ''}Toggle Translation Type (Current: {ctx.config.stream.translation_type.upper()})": _toggle_config_state(
|
||||
ctx, state, "TRANSLATION_TYPE"
|
||||
),
|
||||
f"{'🔙 ' if icons else ''}Back to Results": lambda: InternalDirective.BACK,
|
||||
f"{'❌ ' if icons else ''}Exit": lambda: InternalDirective.EXIT,
|
||||
}
|
||||
)
|
||||
|
||||
choice = ctx.selector.choose(
|
||||
prompt="Select Action",
|
||||
choices=list(options.keys()),
|
||||
)
|
||||
|
||||
if choice and choice in options:
|
||||
return options[choice]()
|
||||
|
||||
return InternalDirective.BACK
|
||||
|
||||
|
||||
def _get_progress_string(ctx: Context, media_item: Optional[MediaItem]) -> str:
|
||||
if not media_item:
|
||||
return ""
|
||||
config = ctx.config
|
||||
|
||||
progress = "0"
|
||||
|
||||
if media_item.user_status:
|
||||
progress = str(media_item.user_status.progress or 0)
|
||||
|
||||
episodes_total = str(media_item.episodes or "??")
|
||||
display_title = f"({progress} of {episodes_total})"
|
||||
|
||||
# Add a visual indicator for new episodes if applicable
|
||||
if (
|
||||
media_item.status == MediaStatus.RELEASING
|
||||
and media_item.next_airing
|
||||
and media_item.user_status
|
||||
and media_item.user_status.status == UserMediaListStatus.WATCHING
|
||||
):
|
||||
last_aired = media_item.next_airing.episode - 1
|
||||
unwatched = last_aired - (media_item.user_status.progress or 0)
|
||||
if unwatched > 0:
|
||||
icon = "🔹" if config.general.icons else "!"
|
||||
display_title += f" {icon}{unwatched} new{icon}"
|
||||
|
||||
return display_title
|
||||
|
||||
|
||||
def _stream(ctx: Context, state: State, force_episodes_menu=False) -> MenuAction:
|
||||
def action():
|
||||
if force_episodes_menu:
|
||||
ctx.switch.force_episodes_menu()
|
||||
return State(menu_name=MenuName.PROVIDER_SEARCH, media_api=state.media_api)
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _stream_downloads(
|
||||
ctx: Context, state: State, force_episodes_menu=False
|
||||
) -> MenuAction:
|
||||
def action():
|
||||
if force_episodes_menu:
|
||||
ctx.switch.force_episodes_menu()
|
||||
return State(menu_name=MenuName.PLAY_DOWNLOADS, media_api=state.media_api)
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _download_episodes(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
return State(menu_name=MenuName.DOWNLOAD_EPISODES, media_api=state.media_api)
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _watch_trailer(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
media_item = state.media_api.media_item
|
||||
|
||||
if not media_item:
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
if not media_item.trailer or not media_item.trailer.id:
|
||||
feedback.warning(
|
||||
"No trailer available for this anime",
|
||||
"This anime doesn't have a trailer link in the database",
|
||||
)
|
||||
else:
|
||||
trailer_url = f"https://www.youtube.com/watch?v={media_item.trailer.id}"
|
||||
|
||||
ctx.player.play(
|
||||
PlayerParams(url=trailer_url, query="", episode="", title="")
|
||||
)
|
||||
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _manage_user_media_list(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
media_item = state.media_api.media_item
|
||||
|
||||
if not media_item:
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
if not ctx.media_api.is_authenticated():
|
||||
feedback.warning(
|
||||
"You are not authenticated",
|
||||
)
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
status = ctx.selector.choose(
|
||||
"Select list status", choices=[t.value for t in UserMediaListStatus]
|
||||
)
|
||||
if status:
|
||||
# local
|
||||
ctx.media_registry.update_media_index_entry(
|
||||
media_id=media_item.id,
|
||||
media_item=media_item,
|
||||
status=UserMediaListStatus(status),
|
||||
)
|
||||
# remote
|
||||
if not ctx.media_api.update_list_entry(
|
||||
UpdateUserMediaListEntryParams(
|
||||
media_item.id, status=UserMediaListStatus(status)
|
||||
)
|
||||
):
|
||||
print(f"Failed to update {media_item.title.english}")
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _manage_user_media_list_in_bulk(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
search_result = state.media_api.search_result
|
||||
|
||||
if not search_result:
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
if not ctx.media_api.is_authenticated():
|
||||
feedback.warning(
|
||||
"You are not authenticated",
|
||||
)
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
choice_map: Dict[str, MediaItem] = {
|
||||
item.title.english: item for item in search_result.values()
|
||||
}
|
||||
preview_command = None
|
||||
if ctx.config.general.preview != "none":
|
||||
from ....utils.preview import create_preview_context
|
||||
|
||||
with create_preview_context() as preview_ctx:
|
||||
preview_command = preview_ctx.get_anime_preview(
|
||||
list(choice_map.values()),
|
||||
list(choice_map.keys()),
|
||||
ctx.config,
|
||||
)
|
||||
selected_titles = ctx.selector.choose_multiple(
|
||||
"Select anime to manage",
|
||||
list(choice_map.keys()),
|
||||
preview=preview_command,
|
||||
)
|
||||
else:
|
||||
selected_titles = ctx.selector.choose_multiple(
|
||||
"Select anime to download",
|
||||
list(choice_map.keys()),
|
||||
)
|
||||
if not selected_titles:
|
||||
feedback.warning("No anime selected. Aborting download.")
|
||||
return InternalDirective.RELOAD
|
||||
anime_to_update_status = [choice_map[title] for title in selected_titles]
|
||||
|
||||
status = ctx.selector.choose(
|
||||
"Select list status", choices=[t.value for t in UserMediaListStatus]
|
||||
)
|
||||
if not status:
|
||||
feedback.warning("No status selected. Aborting bulk action.")
|
||||
return InternalDirective.RELOAD
|
||||
with feedback.progress(
|
||||
"Updating media list...", total=len(anime_to_update_status)
|
||||
) as (task_id, progress):
|
||||
for media_item in anime_to_update_status:
|
||||
feedback.info(f"Updating media status for {media_item.title.english}")
|
||||
ctx.media_registry.update_media_index_entry(
|
||||
media_id=media_item.id,
|
||||
media_item=media_item,
|
||||
status=UserMediaListStatus(status),
|
||||
)
|
||||
# remote
|
||||
|
||||
if not ctx.media_api.update_list_entry(
|
||||
UpdateUserMediaListEntryParams(
|
||||
media_item.id, status=UserMediaListStatus(status)
|
||||
)
|
||||
):
|
||||
print(f"Failed to update {media_item.title.english}")
|
||||
|
||||
progress.update(task_id, advance=1) # type: ignore
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _change_provider(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
from .....libs.provider.anime.types import ProviderName
|
||||
|
||||
new_provider = ctx.selector.choose(
|
||||
"Select Provider", [provider.value for provider in ProviderName]
|
||||
)
|
||||
ctx.config.general.provider = ProviderName(new_provider)
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _toggle_config_state(
|
||||
ctx: Context,
|
||||
state: State,
|
||||
config_state: Literal[
|
||||
"AUTO_ANIME", "AUTO_EPISODE", "CONTINUE_FROM_HISTORY", "TRANSLATION_TYPE"
|
||||
],
|
||||
) -> MenuAction:
|
||||
def action():
|
||||
match config_state:
|
||||
case "AUTO_ANIME":
|
||||
ctx.config.general.auto_select_anime_result = (
|
||||
not ctx.config.general.auto_select_anime_result
|
||||
)
|
||||
case "AUTO_EPISODE":
|
||||
ctx.config.stream.auto_next = not ctx.config.stream.auto_next
|
||||
case "CONTINUE_FROM_HISTORY":
|
||||
ctx.config.stream.continue_from_watch_history = (
|
||||
not ctx.config.stream.continue_from_watch_history
|
||||
)
|
||||
case "TRANSLATION_TYPE":
|
||||
ctx.config.stream.translation_type = (
|
||||
"sub" if ctx.config.stream.translation_type == "dub" else "dub"
|
||||
)
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _score_anime(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
media_item = state.media_api.media_item
|
||||
|
||||
if not media_item:
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
if not ctx.media_api.is_authenticated():
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
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.")
|
||||
# local
|
||||
ctx.media_registry.update_media_index_entry(
|
||||
media_id=media_item.id, media_item=media_item, score=score
|
||||
)
|
||||
# remote
|
||||
ctx.media_api.update_list_entry(
|
||||
UpdateUserMediaListEntryParams(media_id=media_item.id, score=score)
|
||||
)
|
||||
except (ValueError, TypeError):
|
||||
feedback.error(
|
||||
"Invalid score entered", "Please enter a number between 0.0 and 10.0"
|
||||
)
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _view_info(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
media_item = state.media_api.media_item
|
||||
|
||||
if not media_item:
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
import re
|
||||
|
||||
from rich import box
|
||||
from rich.columns import Columns
|
||||
from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
from rich.text import Text
|
||||
|
||||
from ....utils import image
|
||||
|
||||
console = Console()
|
||||
console.clear()
|
||||
|
||||
# Display cover image if available
|
||||
if cover_image := media_item.cover_image:
|
||||
image.render(cover_image.large)
|
||||
|
||||
# Create main title
|
||||
main_title = (
|
||||
media_item.title.english or media_item.title.romaji or "Unknown Title"
|
||||
)
|
||||
title_text = Text(main_title, style="bold cyan")
|
||||
|
||||
# Create info table
|
||||
info_table = Table(show_header=False, box=box.SIMPLE, pad_edge=False)
|
||||
info_table.add_column("Field", style="bold yellow", min_width=15)
|
||||
info_table.add_column("Value", style="white")
|
||||
|
||||
# Add basic information
|
||||
info_table.add_row("English Title", media_item.title.english or "N/A")
|
||||
info_table.add_row("Romaji Title", media_item.title.romaji or "N/A")
|
||||
info_table.add_row("Native Title", media_item.title.native or "N/A")
|
||||
|
||||
if media_item.synonymns:
|
||||
synonyms = ", ".join(media_item.synonymns[:3]) # Show first 3 synonyms
|
||||
if len(media_item.synonymns) > 3:
|
||||
synonyms += f" (+{len(media_item.synonymns) - 3} more)"
|
||||
info_table.add_row("Synonyms", synonyms)
|
||||
|
||||
info_table.add_row("Type", media_item.type.value if media_item.type else "N/A")
|
||||
info_table.add_row(
|
||||
"Format", media_item.format.value if media_item.format else "N/A"
|
||||
)
|
||||
info_table.add_row(
|
||||
"Status", media_item.status.value if media_item.status else "N/A"
|
||||
)
|
||||
info_table.add_row(
|
||||
"Episodes", str(media_item.episodes) if media_item.episodes else "Unknown"
|
||||
)
|
||||
info_table.add_row(
|
||||
"Duration",
|
||||
f"{media_item.duration} min" if media_item.duration else "Unknown",
|
||||
)
|
||||
|
||||
# Add dates
|
||||
if media_item.start_date:
|
||||
start_date = media_item.start_date.strftime("%Y-%m-%d")
|
||||
info_table.add_row("Start Date", start_date)
|
||||
if media_item.end_date:
|
||||
end_date = media_item.end_date.strftime("%Y-%m-%d")
|
||||
info_table.add_row("End Date", end_date)
|
||||
|
||||
# Add scores and popularity
|
||||
if media_item.average_score:
|
||||
info_table.add_row("Average Score", f"{media_item.average_score}/100")
|
||||
if media_item.popularity:
|
||||
info_table.add_row("Popularity", f"#{media_item.popularity:,}")
|
||||
if media_item.favourites:
|
||||
info_table.add_row("Favorites", f"{media_item.favourites:,}")
|
||||
|
||||
# Add MAL ID if available
|
||||
if media_item.id_mal:
|
||||
info_table.add_row("MyAnimeList ID", str(media_item.id_mal))
|
||||
|
||||
# Create genres panel
|
||||
if media_item.genres:
|
||||
genres_text = ", ".join([genre.value for genre in media_item.genres])
|
||||
genres_panel = Panel(
|
||||
Text(genres_text, style="green"),
|
||||
title="[bold]Genres[/bold]",
|
||||
border_style="green",
|
||||
box=box.ROUNDED,
|
||||
)
|
||||
else:
|
||||
genres_panel = Panel(
|
||||
Text("No genres available", style="dim"),
|
||||
title="[bold]Genres[/bold]",
|
||||
border_style="green",
|
||||
box=box.ROUNDED,
|
||||
)
|
||||
|
||||
# Create tags panel (show top tags)
|
||||
if media_item.tags:
|
||||
top_tags = sorted(media_item.tags, key=lambda x: x.rank or 0, reverse=True)[
|
||||
:10
|
||||
]
|
||||
tags_text = ", ".join([tag.name.value for tag in top_tags])
|
||||
tags_panel = Panel(
|
||||
Text(tags_text, style="yellow"),
|
||||
title="[bold]Tags[/bold]",
|
||||
border_style="yellow",
|
||||
box=box.ROUNDED,
|
||||
)
|
||||
else:
|
||||
tags_panel = Panel(
|
||||
Text("No tags available", style="dim"),
|
||||
title="[bold]Tags[/bold]",
|
||||
border_style="yellow",
|
||||
box=box.ROUNDED,
|
||||
)
|
||||
|
||||
# Create studios panel
|
||||
if media_item.studios:
|
||||
studios_text = ", ".join(
|
||||
[studio.name for studio in media_item.studios if studio.name]
|
||||
)
|
||||
studios_panel = Panel(
|
||||
Text(studios_text, style="blue"),
|
||||
title="[bold]Studios[/bold]",
|
||||
border_style="blue",
|
||||
box=box.ROUNDED,
|
||||
)
|
||||
else:
|
||||
studios_panel = Panel(
|
||||
Text("No studio information", style="dim"),
|
||||
title="[bold]Studios[/bold]",
|
||||
border_style="blue",
|
||||
box=box.ROUNDED,
|
||||
)
|
||||
|
||||
# Create description panel
|
||||
description = media_item.description or "No description available"
|
||||
# Clean HTML tags from description
|
||||
clean_description = re.sub(r"<[^>]+>", "", description)
|
||||
# Replace common HTML entities
|
||||
clean_description = (
|
||||
clean_description.replace(""", '"')
|
||||
.replace("&", "&")
|
||||
.replace("<", "<")
|
||||
.replace(">", ">")
|
||||
)
|
||||
|
||||
description_panel = Panel(
|
||||
Text(clean_description, style="white"),
|
||||
title="[bold]Description[/bold]",
|
||||
border_style="cyan",
|
||||
box=box.ROUNDED,
|
||||
)
|
||||
|
||||
# Create user status panel if available
|
||||
if media_item.user_status:
|
||||
user_info_table = Table(show_header=False, box=box.SIMPLE)
|
||||
user_info_table.add_column("Field", style="bold magenta")
|
||||
user_info_table.add_column("Value", style="white")
|
||||
|
||||
if media_item.user_status.status:
|
||||
user_info_table.add_row(
|
||||
"Status", media_item.user_status.status.value.title()
|
||||
)
|
||||
if media_item.user_status.progress is not None:
|
||||
progress = (
|
||||
f"{media_item.user_status.progress}/{media_item.episodes or '?'}"
|
||||
)
|
||||
user_info_table.add_row("Progress", progress)
|
||||
if media_item.user_status.score:
|
||||
user_info_table.add_row(
|
||||
"Your Score", f"{media_item.user_status.score}/10"
|
||||
)
|
||||
if media_item.user_status.repeat:
|
||||
user_info_table.add_row(
|
||||
"Rewatched", f"{media_item.user_status.repeat} times"
|
||||
)
|
||||
|
||||
user_panel = Panel(
|
||||
user_info_table,
|
||||
title="[bold]Your List Status[/bold]",
|
||||
border_style="magenta",
|
||||
box=box.ROUNDED,
|
||||
)
|
||||
else:
|
||||
user_panel = None
|
||||
|
||||
# Create next airing panel if available
|
||||
if media_item.next_airing:
|
||||
airing_info_table = Table(show_header=False, box=box.SIMPLE)
|
||||
airing_info_table.add_column("Field", style="bold red")
|
||||
airing_info_table.add_column("Value", style="white")
|
||||
|
||||
airing_info_table.add_row(
|
||||
"Next Episode", str(media_item.next_airing.episode)
|
||||
)
|
||||
|
||||
if media_item.next_airing.airing_at:
|
||||
air_date = media_item.next_airing.airing_at.strftime("%Y-%m-%d %H:%M")
|
||||
airing_info_table.add_row("Air Date", air_date)
|
||||
|
||||
airing_panel = Panel(
|
||||
airing_info_table,
|
||||
title="[bold]Next Airing[/bold]",
|
||||
border_style="red",
|
||||
box=box.ROUNDED,
|
||||
)
|
||||
else:
|
||||
airing_panel = None
|
||||
|
||||
# Create main info panel
|
||||
info_panel = Panel(
|
||||
info_table,
|
||||
title="[bold]Basic Information[/bold]",
|
||||
border_style="cyan",
|
||||
box=box.ROUNDED,
|
||||
)
|
||||
|
||||
# Display everything
|
||||
console.print(Panel(title_text, box=box.DOUBLE, border_style="bright_cyan"))
|
||||
console.print()
|
||||
|
||||
# Create columns for better layout
|
||||
panels_row1 = [info_panel, genres_panel]
|
||||
if user_panel:
|
||||
panels_row1.append(user_panel)
|
||||
|
||||
console.print(Columns(panels_row1, equal=True, expand=True))
|
||||
console.print()
|
||||
|
||||
panels_row2 = [tags_panel, studios_panel]
|
||||
if airing_panel:
|
||||
panels_row2.append(airing_panel)
|
||||
|
||||
console.print(Columns(panels_row2, equal=True, expand=True))
|
||||
console.print()
|
||||
|
||||
console.print(description_panel)
|
||||
|
||||
ctx.selector.ask("Press Enter to continue...")
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _view_recommendations(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
media_item = state.media_api.media_item
|
||||
|
||||
if not media_item:
|
||||
feedback.error("Media item is not in state")
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
loading_message = "Fetching recommendations..."
|
||||
recommendations = None
|
||||
|
||||
with feedback.progress(loading_message):
|
||||
recommendations = ctx.media_api.get_recommendation_for(
|
||||
MediaRecommendationParams(id=media_item.id, page=1)
|
||||
)
|
||||
|
||||
if not recommendations:
|
||||
feedback.warning(
|
||||
"No recommendations found",
|
||||
"This anime doesn't have any recommendations available",
|
||||
)
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
# Convert list of MediaItem to search result format
|
||||
search_result = {item.id: item for item in recommendations}
|
||||
|
||||
# Create a fake page info since recommendations don't have pagination
|
||||
from .....libs.media_api.types import PageInfo
|
||||
|
||||
page_info = PageInfo(
|
||||
total=len(recommendations),
|
||||
current_page=1,
|
||||
has_next_page=False,
|
||||
per_page=len(recommendations),
|
||||
)
|
||||
|
||||
return State(
|
||||
menu_name=MenuName.RESULTS,
|
||||
media_api=MediaApiState(
|
||||
search_result=search_result,
|
||||
page_info=page_info,
|
||||
search_params=None, # No search params for recommendations
|
||||
),
|
||||
)
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _view_relations(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
media_item = state.media_api.media_item
|
||||
|
||||
if not media_item:
|
||||
feedback.error("Media item is not in state")
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
loading_message = "Fetching related anime..."
|
||||
relations = None
|
||||
|
||||
with feedback.progress(loading_message):
|
||||
relations = ctx.media_api.get_related_anime_for(
|
||||
MediaRelationsParams(id=media_item.id)
|
||||
)
|
||||
|
||||
if not relations:
|
||||
feedback.warning(
|
||||
"No related anime found",
|
||||
"This anime doesn't have any related anime available",
|
||||
)
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
# Convert list of MediaItem to search result format
|
||||
search_result = {item.id: item for item in relations}
|
||||
|
||||
# Create a fake page info since relations don't have pagination
|
||||
from .....libs.media_api.types import PageInfo
|
||||
|
||||
page_info = PageInfo(
|
||||
total=len(relations),
|
||||
current_page=1,
|
||||
has_next_page=False,
|
||||
per_page=len(relations),
|
||||
)
|
||||
|
||||
return State(
|
||||
menu_name=MenuName.RESULTS,
|
||||
media_api=MediaApiState(
|
||||
search_result=search_result,
|
||||
page_info=page_info,
|
||||
search_params=None, # No search params for relations
|
||||
),
|
||||
)
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _view_characters(ctx: Context, state: State) -> MenuAction:
|
||||
"""Action to transition to the character selection menu."""
|
||||
|
||||
def action() -> State | InternalDirective:
|
||||
return State(menu_name=MenuName.MEDIA_CHARACTERS, media_api=state.media_api)
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _view_airing_schedule(ctx: Context, state: State) -> MenuAction:
|
||||
"""Action to transition to the airing schedule menu."""
|
||||
|
||||
def action() -> State | InternalDirective:
|
||||
return State(
|
||||
menu_name=MenuName.MEDIA_AIRING_SCHEDULE, media_api=state.media_api
|
||||
)
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _view_reviews(ctx: Context, state: State) -> MenuAction:
|
||||
"""Action to transition to the review selection menu."""
|
||||
|
||||
def action() -> State | InternalDirective:
|
||||
return State(menu_name=MenuName.MEDIA_REVIEW, media_api=state.media_api)
|
||||
|
||||
return action
|
||||
207
viu_cli/cli/interactive/menu/media/media_airing_schedule.py
Normal file
207
viu_cli/cli/interactive/menu/media/media_airing_schedule.py
Normal file
@@ -0,0 +1,207 @@
|
||||
from typing import Dict, Optional, Union
|
||||
|
||||
from .....libs.media_api.types import AiringScheduleItem, AiringScheduleResult
|
||||
from ...session import Context, session
|
||||
from ...state import InternalDirective, State
|
||||
|
||||
|
||||
@session.menu
|
||||
def media_airing_schedule(
|
||||
ctx: Context, state: State
|
||||
) -> Union[State, InternalDirective]:
|
||||
"""
|
||||
Fetches and displays the airing schedule for an anime.
|
||||
Shows upcoming episodes with air dates and countdown timers.
|
||||
"""
|
||||
from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
|
||||
feedback = ctx.feedback
|
||||
selector = ctx.selector
|
||||
console = Console()
|
||||
media_item = state.media_api.media_item
|
||||
|
||||
if not media_item:
|
||||
feedback.error("Media item is not in state.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
from .....libs.media_api.params import MediaAiringScheduleParams
|
||||
|
||||
loading_message = f"Fetching airing schedule for {media_item.title.english or media_item.title.romaji}..."
|
||||
schedule_result: Optional[AiringScheduleResult] = None
|
||||
|
||||
with feedback.progress(loading_message):
|
||||
schedule_result = ctx.media_api.get_airing_schedule_for(
|
||||
MediaAiringScheduleParams(id=media_item.id)
|
||||
)
|
||||
|
||||
if not schedule_result or not schedule_result.schedule_items:
|
||||
feedback.warning(
|
||||
"No airing schedule found",
|
||||
"This anime doesn't have upcoming episodes or airing data",
|
||||
)
|
||||
return InternalDirective.BACK
|
||||
|
||||
# Create choices for each episode in the schedule
|
||||
choice_map: Dict[str, AiringScheduleItem] = {}
|
||||
for item in schedule_result.schedule_items:
|
||||
display_name = f"Episode {item.episode}"
|
||||
if item.airing_at:
|
||||
airing_time = item.airing_at
|
||||
display_name += f" - {airing_time.strftime('%Y-%m-%d %H:%M')}"
|
||||
if item.time_until_airing:
|
||||
display_name += f" (in {item.time_until_airing})"
|
||||
|
||||
choice_map[display_name] = item
|
||||
|
||||
choices = list(choice_map.keys()) + ["View Full Schedule", "Back"]
|
||||
|
||||
preview_command = None
|
||||
if ctx.config.general.preview != "none":
|
||||
from ....utils.preview import create_preview_context
|
||||
|
||||
anime_title = media_item.title.english or media_item.title.romaji or "Unknown"
|
||||
with create_preview_context() as preview_ctx:
|
||||
preview_command = preview_ctx.get_airing_schedule_preview(
|
||||
schedule_result, ctx.config, anime_title
|
||||
)
|
||||
|
||||
while True:
|
||||
chosen_title = selector.choose(
|
||||
prompt="Select an episode or view full schedule",
|
||||
choices=choices,
|
||||
preview=preview_command,
|
||||
)
|
||||
|
||||
if not chosen_title or chosen_title == "Back":
|
||||
return InternalDirective.BACK
|
||||
|
||||
if chosen_title == "View Full Schedule":
|
||||
console.clear()
|
||||
# Display airing schedule
|
||||
anime_title = (
|
||||
media_item.title.english or media_item.title.romaji or "Unknown"
|
||||
)
|
||||
_display_airing_schedule(console, schedule_result, anime_title)
|
||||
selector.ask("\nPress Enter to return...")
|
||||
continue
|
||||
|
||||
# Show individual episode details
|
||||
selected_item = choice_map[chosen_title]
|
||||
console.clear()
|
||||
|
||||
episode_info = []
|
||||
episode_info.append(f"[bold cyan]Episode {selected_item.episode}[/bold cyan]")
|
||||
|
||||
if selected_item.airing_at:
|
||||
airing_time = selected_item.airing_at
|
||||
episode_info.append(
|
||||
f"[green]Airs at:[/green] {airing_time.strftime('%Y-%m-%d %H:%M:%S')}"
|
||||
)
|
||||
|
||||
if selected_item.time_until_airing:
|
||||
episode_info.append(
|
||||
f"[yellow]Time until airing:[/yellow] {selected_item.time_until_airing}"
|
||||
)
|
||||
|
||||
episode_content = "\n".join(episode_info)
|
||||
|
||||
console.print(
|
||||
Panel(
|
||||
episode_content,
|
||||
title=f"Episode Details - {media_item.title.english or media_item.title.romaji}",
|
||||
border_style="blue",
|
||||
expand=True,
|
||||
)
|
||||
)
|
||||
|
||||
selector.ask("\nPress Enter to return to the schedule list...")
|
||||
|
||||
return InternalDirective.BACK
|
||||
|
||||
|
||||
def _display_airing_schedule(
|
||||
console, schedule_result: AiringScheduleResult, anime_title: str
|
||||
):
|
||||
"""Display the airing schedule in a formatted table."""
|
||||
from datetime import datetime
|
||||
from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
from rich.text import Text
|
||||
|
||||
# Create title
|
||||
title = Text(f"Airing Schedule for {anime_title}", style="bold cyan")
|
||||
|
||||
# Create table for episodes
|
||||
table = Table(show_header=True, header_style="bold magenta", expand=True)
|
||||
table.add_column("Episode", style="cyan", justify="center", min_width=8)
|
||||
table.add_column("Air Date", style="green", min_width=20)
|
||||
table.add_column("Time Until Airing", style="yellow", min_width=15)
|
||||
table.add_column("Status", style="white", min_width=10)
|
||||
|
||||
# Sort episodes by episode number
|
||||
sorted_episodes = sorted(schedule_result.schedule_items, key=lambda x: x.episode)
|
||||
|
||||
for episode in sorted_episodes[:15]: # Show next 15 episodes
|
||||
ep_num = str(episode.episode)
|
||||
|
||||
# Format air date
|
||||
if episode.airing_at:
|
||||
formatted_date = episode.airing_at.strftime("%Y-%m-%d %H:%M")
|
||||
|
||||
# Check if episode has already aired
|
||||
now = datetime.now()
|
||||
if episode.airing_at < now:
|
||||
status = "[dim]Aired[/dim]"
|
||||
else:
|
||||
status = "[green]Upcoming[/green]"
|
||||
else:
|
||||
formatted_date = "[dim]Unknown[/dim]"
|
||||
status = "[dim]TBA[/dim]"
|
||||
|
||||
# Format time until airing
|
||||
if episode.time_until_airing and episode.time_until_airing > 0:
|
||||
time_until = episode.time_until_airing
|
||||
days = time_until // 86400
|
||||
hours = (time_until % 86400) // 3600
|
||||
minutes = (time_until % 3600) // 60
|
||||
|
||||
if days > 0:
|
||||
time_str = f"{days}d {hours}h"
|
||||
elif hours > 0:
|
||||
time_str = f"{hours}h {minutes}m"
|
||||
else:
|
||||
time_str = f"{minutes}m"
|
||||
elif episode.airing_at and episode.airing_at < datetime.now():
|
||||
time_str = "[dim]Aired[/dim]"
|
||||
else:
|
||||
time_str = "[dim]Unknown[/dim]"
|
||||
|
||||
table.add_row(ep_num, formatted_date, time_str, status)
|
||||
|
||||
# Display in a panel
|
||||
panel = Panel(table, title=title, border_style="blue", expand=True)
|
||||
console.print(panel)
|
||||
|
||||
# Add summary information
|
||||
total_episodes = len(schedule_result.schedule_items)
|
||||
upcoming_episodes = sum(
|
||||
1
|
||||
for ep in schedule_result.schedule_items
|
||||
if ep.airing_at and ep.airing_at > datetime.now()
|
||||
)
|
||||
|
||||
summary_text = Text()
|
||||
summary_text.append("Total episodes in schedule: ", style="bold")
|
||||
summary_text.append(f"{total_episodes}", style="cyan")
|
||||
summary_text.append("\nUpcoming episodes: ", style="bold")
|
||||
summary_text.append(f"{upcoming_episodes}", style="green")
|
||||
|
||||
summary_panel = Panel(
|
||||
summary_text,
|
||||
title="[bold]Summary[/bold]",
|
||||
border_style="yellow",
|
||||
expand=False,
|
||||
)
|
||||
console.print()
|
||||
console.print(summary_panel)
|
||||
166
viu_cli/cli/interactive/menu/media/media_characters.py
Normal file
166
viu_cli/cli/interactive/menu/media/media_characters.py
Normal file
@@ -0,0 +1,166 @@
|
||||
import re
|
||||
from typing import Dict, Optional, Union
|
||||
|
||||
from .....libs.media_api.types import Character, CharacterSearchResult
|
||||
from ...session import Context, session
|
||||
from ...state import InternalDirective, State
|
||||
|
||||
|
||||
@session.menu
|
||||
def media_characters(ctx: Context, state: State) -> Union[State, InternalDirective]:
|
||||
"""
|
||||
Fetches and displays a list of characters for the user to select from.
|
||||
Shows character details upon selection or in the preview pane.
|
||||
"""
|
||||
from rich.console import Console
|
||||
|
||||
feedback = ctx.feedback
|
||||
selector = ctx.selector
|
||||
console = Console()
|
||||
config = ctx.config
|
||||
media_item = state.media_api.media_item
|
||||
|
||||
if not media_item:
|
||||
feedback.error("Media item is not in state.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
from .....libs.media_api.params import MediaCharactersParams
|
||||
|
||||
loading_message = f"Fetching characters for {media_item.title.english or media_item.title.romaji}..."
|
||||
characters_result: Optional[CharacterSearchResult] = None
|
||||
|
||||
with feedback.progress(loading_message):
|
||||
characters_result = ctx.media_api.get_characters_of(
|
||||
MediaCharactersParams(id=media_item.id)
|
||||
)
|
||||
|
||||
if not characters_result or not characters_result.characters:
|
||||
feedback.error("No characters found for this anime.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
characters = characters_result.characters
|
||||
choice_map: Dict[str, Character] = {}
|
||||
|
||||
# Create display names for characters
|
||||
for character in characters:
|
||||
display_name = character.name.full or character.name.first or "Unknown"
|
||||
if character.gender:
|
||||
display_name += f" ({character.gender})"
|
||||
if character.age:
|
||||
display_name += f" - Age {character.age}"
|
||||
|
||||
choice_map[display_name] = character
|
||||
|
||||
choices = list(choice_map.keys()) + ["Back"]
|
||||
|
||||
preview_command = None
|
||||
if config.general.preview != "none":
|
||||
from ....utils.preview import create_preview_context
|
||||
|
||||
with create_preview_context() as preview_ctx:
|
||||
preview_command = preview_ctx.get_character_preview(choice_map, ctx.config)
|
||||
|
||||
while True:
|
||||
chosen_title = selector.choose(
|
||||
prompt="Select a character to view details",
|
||||
choices=choices,
|
||||
preview=preview_command,
|
||||
)
|
||||
|
||||
if not chosen_title or chosen_title == "Back":
|
||||
return InternalDirective.BACK
|
||||
|
||||
selected_character = choice_map[chosen_title]
|
||||
console.clear()
|
||||
|
||||
# Display character details
|
||||
anime_title = media_item.title.english or media_item.title.romaji or "Unknown"
|
||||
_display_character_details(console, selected_character, anime_title)
|
||||
|
||||
selector.ask("\nPress Enter to return to the character list...")
|
||||
|
||||
|
||||
def _display_character_details(console, character: Character, anime_title: str):
|
||||
"""Display detailed character information in a formatted panel."""
|
||||
from rich.columns import Columns
|
||||
from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
from rich.text import Text
|
||||
|
||||
# Character name panel
|
||||
name_text = Text()
|
||||
if character.name.full:
|
||||
name_text.append(character.name.full, style="bold cyan")
|
||||
elif character.name.first:
|
||||
full_name = character.name.first
|
||||
if character.name.last:
|
||||
full_name += f" {character.name.last}"
|
||||
name_text.append(full_name, style="bold cyan")
|
||||
else:
|
||||
name_text.append("Unknown Character", style="bold dim")
|
||||
|
||||
if character.name.native:
|
||||
name_text.append(f"\n{character.name.native}", style="green")
|
||||
|
||||
name_panel = Panel(
|
||||
name_text,
|
||||
title=f"[bold]Character from {anime_title}[/bold]",
|
||||
border_style="cyan",
|
||||
expand=False,
|
||||
)
|
||||
|
||||
# Basic info table
|
||||
info_table = Table(show_header=False, box=None, padding=(0, 1))
|
||||
info_table.add_column("Field", style="bold yellow", min_width=12)
|
||||
info_table.add_column("Value", style="white")
|
||||
|
||||
if character.gender:
|
||||
info_table.add_row("Gender", character.gender)
|
||||
if character.age:
|
||||
info_table.add_row("Age", str(character.age))
|
||||
if character.blood_type:
|
||||
info_table.add_row("Blood Type", character.blood_type)
|
||||
if character.favourites:
|
||||
info_table.add_row("Favorites", f"{character.favourites:,}")
|
||||
if character.date_of_birth:
|
||||
birth_date = character.date_of_birth.strftime("%B %d, %Y")
|
||||
info_table.add_row("Birthday", birth_date)
|
||||
|
||||
info_panel = Panel(
|
||||
info_table,
|
||||
title="[bold]Basic Information[/bold]",
|
||||
border_style="blue",
|
||||
)
|
||||
|
||||
# Description panel
|
||||
description = character.description or "No description available"
|
||||
# Clean HTML tags from description
|
||||
clean_description = re.sub(r"<[^>]+>", "", description)
|
||||
# Replace common HTML entities
|
||||
clean_description = (
|
||||
clean_description.replace(""", '"')
|
||||
.replace("&", "&")
|
||||
.replace("<", "<")
|
||||
.replace(">", ">")
|
||||
.replace("'", "'")
|
||||
.replace(" ", " ")
|
||||
)
|
||||
# Limit length for display
|
||||
if len(clean_description) > 500:
|
||||
clean_description = clean_description[:497] + "..."
|
||||
|
||||
description_panel = Panel(
|
||||
Text(clean_description, style="white"),
|
||||
title="[bold]Description[/bold]",
|
||||
border_style="green",
|
||||
)
|
||||
|
||||
# Display everything
|
||||
console.print(name_panel)
|
||||
console.print()
|
||||
|
||||
# Show panels side by side if there's basic info
|
||||
if info_table.rows:
|
||||
console.print(Columns([info_panel, description_panel], equal=True, expand=True))
|
||||
else:
|
||||
console.print(description_panel)
|
||||
82
viu_cli/cli/interactive/menu/media/media_review.py
Normal file
82
viu_cli/cli/interactive/menu/media/media_review.py
Normal file
@@ -0,0 +1,82 @@
|
||||
from typing import Dict, List, Optional, Union
|
||||
|
||||
from .....libs.media_api.types import MediaReview
|
||||
from ...session import Context, session
|
||||
from ...state import InternalDirective, State
|
||||
|
||||
|
||||
@session.menu
|
||||
def media_review(ctx: Context, state: State) -> Union[State, InternalDirective]:
|
||||
"""
|
||||
Fetches and displays a list of reviews for the user to select from.
|
||||
Shows the full review body upon selection or in the preview pane.
|
||||
"""
|
||||
from rich.console import Console
|
||||
from rich.markdown import Markdown
|
||||
from rich.panel import Panel
|
||||
|
||||
feedback = ctx.feedback
|
||||
selector = ctx.selector
|
||||
console = Console()
|
||||
config = ctx.config
|
||||
media_item = state.media_api.media_item
|
||||
|
||||
if not media_item:
|
||||
feedback.error("Media item is not in state.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
from .....libs.media_api.params import MediaReviewsParams
|
||||
|
||||
loading_message = (
|
||||
f"Fetching reviews for {media_item.title.english or media_item.title.romaji}..."
|
||||
)
|
||||
reviews: Optional[List[MediaReview]] = None
|
||||
|
||||
with feedback.progress(loading_message):
|
||||
reviews = ctx.media_api.get_reviews_for(
|
||||
MediaReviewsParams(id=media_item.id, per_page=15)
|
||||
)
|
||||
|
||||
if not reviews:
|
||||
feedback.error("No reviews found for this anime.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
choice_map: Dict[str, MediaReview] = {
|
||||
f"By {review.user.name}: {(review.summary or 'No summary')[:80]}": review
|
||||
for review in reviews
|
||||
}
|
||||
choices = list(choice_map.keys()) + ["Back"]
|
||||
|
||||
preview_command = None
|
||||
if config.general.preview != "none":
|
||||
from ....utils.preview import create_preview_context
|
||||
|
||||
with create_preview_context() as preview_ctx:
|
||||
preview_command = preview_ctx.get_review_preview(choice_map, ctx.config)
|
||||
|
||||
while True:
|
||||
chosen_title = selector.choose(
|
||||
prompt="Select a review to read",
|
||||
choices=choices,
|
||||
preview=preview_command,
|
||||
)
|
||||
|
||||
if not chosen_title or chosen_title == "Back":
|
||||
return InternalDirective.BACK
|
||||
|
||||
selected_review = choice_map[chosen_title]
|
||||
console.clear()
|
||||
|
||||
reviewer_name = f"[bold magenta]{selected_review.user.name}[/bold magenta]"
|
||||
review_summary = (
|
||||
f"[italic green]'{selected_review.summary}'[/italic green]"
|
||||
if selected_review.summary
|
||||
else ""
|
||||
)
|
||||
panel_title = f"Review by {reviewer_name} - {review_summary}"
|
||||
review_body = Markdown(selected_review.body)
|
||||
|
||||
console.print(
|
||||
Panel(review_body, title=panel_title, border_style="blue", expand=True)
|
||||
)
|
||||
selector.ask("\nPress Enter to return to the review list...")
|
||||
332
viu_cli/cli/interactive/menu/media/play_downloads.py
Normal file
332
viu_cli/cli/interactive/menu/media/play_downloads.py
Normal file
@@ -0,0 +1,332 @@
|
||||
from typing import Callable, Dict, Literal, Union
|
||||
|
||||
from .....libs.player.params import PlayerParams
|
||||
from ...session import Context, session
|
||||
from ...state import InternalDirective, MenuName, State
|
||||
|
||||
MenuAction = Callable[[], Union[State, InternalDirective]]
|
||||
|
||||
|
||||
@session.menu
|
||||
def play_downloads(ctx: Context, state: State) -> State | InternalDirective:
|
||||
"""Menu to select and play locally downloaded episodes."""
|
||||
from ....service.registry.models import DownloadStatus
|
||||
|
||||
feedback = ctx.feedback
|
||||
media_item = state.media_api.media_item
|
||||
current_episode_num = state.provider.episode
|
||||
if not media_item:
|
||||
feedback.error("No media item selected.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
record = ctx.media_registry.get_media_record(media_item.id)
|
||||
if not record or not record.media_episodes:
|
||||
feedback.warning("No downloaded episodes found for this anime.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
downloaded_episodes = {
|
||||
ep.episode_number: ep.file_path
|
||||
for ep in record.media_episodes
|
||||
if ep.download_status == DownloadStatus.COMPLETED
|
||||
and ep.file_path
|
||||
and ep.file_path.exists()
|
||||
}
|
||||
|
||||
if not downloaded_episodes:
|
||||
feedback.warning("No complete downloaded episodes found.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
chosen_episode: str | None = current_episode_num
|
||||
start_time: str | None = None
|
||||
|
||||
if not chosen_episode and ctx.config.stream.continue_from_watch_history:
|
||||
_chosen_episode, _start_time = ctx.watch_history.get_episode(media_item)
|
||||
if _chosen_episode in downloaded_episodes:
|
||||
chosen_episode = _chosen_episode
|
||||
start_time = _start_time
|
||||
|
||||
if not chosen_episode or ctx.switch.show_episodes_menu:
|
||||
choices = [*list(sorted(downloaded_episodes.keys(), key=float)), "Back"]
|
||||
|
||||
preview_command = None
|
||||
if ctx.config.general.preview != "none":
|
||||
from ....utils.preview import create_preview_context
|
||||
|
||||
with create_preview_context() as preview_ctx:
|
||||
preview_command = preview_ctx.get_episode_preview(
|
||||
list(downloaded_episodes.keys()), media_item, ctx.config
|
||||
)
|
||||
|
||||
chosen_episode_str = ctx.selector.choose(
|
||||
prompt="Select Episode", choices=choices, preview=preview_command
|
||||
)
|
||||
|
||||
if not chosen_episode_str or chosen_episode_str == "Back":
|
||||
return InternalDirective.BACK
|
||||
|
||||
chosen_episode = chosen_episode_str
|
||||
# Workers are automatically cleaned up when exiting the context
|
||||
else:
|
||||
# No preview mode
|
||||
chosen_episode_str = ctx.selector.choose(
|
||||
prompt="Select Episode", choices=choices, preview=None
|
||||
)
|
||||
|
||||
if not chosen_episode_str or chosen_episode_str == "Back":
|
||||
return InternalDirective.BACK
|
||||
|
||||
chosen_episode = chosen_episode_str
|
||||
|
||||
if not chosen_episode or chosen_episode == "Back":
|
||||
return InternalDirective.BACK
|
||||
|
||||
return State(
|
||||
menu_name=MenuName.DOWNLOADS_PLAYER_CONTROLS,
|
||||
media_api=state.media_api,
|
||||
provider=state.provider.model_copy(
|
||||
update={"episode": chosen_episode, "start_time": start_time}
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
# TODO: figure out the best way to implement this logic for next episode ...
|
||||
@session.menu
|
||||
def downloads_player_controls(
|
||||
ctx: Context, state: State
|
||||
) -> Union[State, InternalDirective]:
|
||||
from ....service.registry.models import DownloadStatus
|
||||
|
||||
feedback = ctx.feedback
|
||||
feedback.clear_console()
|
||||
|
||||
config = ctx.config
|
||||
selector = ctx.selector
|
||||
|
||||
media_item = state.media_api.media_item
|
||||
current_episode_num = state.provider.episode
|
||||
current_start_time = state.provider.start_time
|
||||
|
||||
if not media_item or not current_episode_num:
|
||||
feedback.error("Player state is incomplete. Returning.")
|
||||
return InternalDirective.BACK
|
||||
record = ctx.media_registry.get_media_record(media_item.id)
|
||||
if not record or not record.media_episodes:
|
||||
feedback.warning("No downloaded episodes found for this anime.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
downloaded_episodes = {
|
||||
ep.episode_number: ep.file_path
|
||||
for ep in record.media_episodes
|
||||
if ep.download_status == DownloadStatus.COMPLETED
|
||||
and ep.file_path
|
||||
and ep.file_path.exists()
|
||||
}
|
||||
available_episodes = list(sorted(downloaded_episodes.keys(), key=float))
|
||||
current_index = available_episodes.index(current_episode_num)
|
||||
|
||||
if not ctx.switch.dont_play:
|
||||
file_path = downloaded_episodes[current_episode_num]
|
||||
|
||||
# Use the player service to play the local file
|
||||
title = f"{media_item.title.english or media_item.title.romaji}; Episode {current_episode_num}"
|
||||
if media_item.streaming_episodes:
|
||||
streaming_episode = media_item.streaming_episodes.get(current_episode_num)
|
||||
title = streaming_episode.title if streaming_episode else title
|
||||
player_result = ctx.player.play(
|
||||
PlayerParams(
|
||||
url=str(file_path),
|
||||
title=title,
|
||||
query=media_item.title.english or media_item.title.romaji or "",
|
||||
episode=current_episode_num,
|
||||
start_time=current_start_time,
|
||||
),
|
||||
media_item=media_item,
|
||||
local=True,
|
||||
)
|
||||
|
||||
# Track watch history after playing
|
||||
ctx.watch_history.track(media_item, player_result)
|
||||
|
||||
if config.stream.auto_next and current_index < len(available_episodes) - 1:
|
||||
feedback.info("Auto-playing next episode...")
|
||||
next_episode_num = available_episodes[current_index + 1]
|
||||
|
||||
return State(
|
||||
menu_name=MenuName.DOWNLOADS_PLAYER_CONTROLS,
|
||||
media_api=state.media_api,
|
||||
provider=state.provider.model_copy(
|
||||
update={"episode": next_episode_num, "start_time": None}
|
||||
),
|
||||
)
|
||||
|
||||
# --- 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)
|
||||
if current_index:
|
||||
options[f"{'⏪ ' if icons else ''}Previous Episode"] = _previous_episode(
|
||||
ctx, state
|
||||
)
|
||||
|
||||
options.update(
|
||||
{
|
||||
f"{'🔂 ' if icons else ''}Replay": _replay(ctx, state),
|
||||
f"{'🎞️ ' if icons else ''}Episode List": _episodes_list(ctx, state),
|
||||
f"{'🔘 ' if icons else ''}Toggle Auto Next Episode (Current: {ctx.config.stream.auto_next})": _toggle_config_state(
|
||||
ctx, state, "AUTO_EPISODE"
|
||||
),
|
||||
f"{'🎥 ' if icons else ''}Media Actions Menu": lambda: InternalDirective.BACKX2,
|
||||
f"{'🏠 ' if icons else ''}Main Menu": lambda: InternalDirective.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]()
|
||||
else:
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
|
||||
def _next_episode(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
from ....service.registry.models import DownloadStatus
|
||||
|
||||
feedback = ctx.feedback
|
||||
|
||||
media_item = state.media_api.media_item
|
||||
current_episode_num = state.provider.episode
|
||||
|
||||
if not media_item or not current_episode_num:
|
||||
feedback.error(
|
||||
"Player state is incomplete. not going to next episode. Returning."
|
||||
)
|
||||
ctx.switch.force_dont_play()
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
record = ctx.media_registry.get_media_record(media_item.id)
|
||||
if not record or not record.media_episodes:
|
||||
feedback.warning("No downloaded episodes found for this anime.")
|
||||
ctx.switch.force_dont_play()
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
downloaded_episodes = {
|
||||
ep.episode_number: ep.file_path
|
||||
for ep in record.media_episodes
|
||||
if ep.download_status == DownloadStatus.COMPLETED
|
||||
and ep.file_path
|
||||
and ep.file_path.exists()
|
||||
}
|
||||
available_episodes = list(sorted(downloaded_episodes.keys(), key=float))
|
||||
current_index = available_episodes.index(current_episode_num)
|
||||
|
||||
if current_index < len(available_episodes) - 1:
|
||||
next_episode_num = available_episodes[current_index + 1]
|
||||
|
||||
return State(
|
||||
menu_name=MenuName.DOWNLOADS_PLAYER_CONTROLS,
|
||||
media_api=state.media_api,
|
||||
provider=state.provider.model_copy(
|
||||
update={"episode": next_episode_num, "start_time": None}
|
||||
),
|
||||
)
|
||||
feedback.warning("This is the last available episode.")
|
||||
ctx.switch.force_dont_play()
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _previous_episode(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
from ....service.registry.models import DownloadStatus
|
||||
|
||||
feedback = ctx.feedback
|
||||
|
||||
media_item = state.media_api.media_item
|
||||
current_episode_num = state.provider.episode
|
||||
|
||||
if not media_item or not current_episode_num:
|
||||
feedback.error(
|
||||
"Player state is incomplete not going to previous episode. Returning."
|
||||
)
|
||||
ctx.switch.force_dont_play()
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
record = ctx.media_registry.get_media_record(media_item.id)
|
||||
if not record or not record.media_episodes:
|
||||
feedback.warning("No downloaded episodes found for this anime.")
|
||||
ctx.switch.force_dont_play()
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
downloaded_episodes = {
|
||||
ep.episode_number: ep.file_path
|
||||
for ep in record.media_episodes
|
||||
if ep.download_status == DownloadStatus.COMPLETED
|
||||
and ep.file_path
|
||||
and ep.file_path.exists()
|
||||
}
|
||||
available_episodes = list(sorted(downloaded_episodes.keys(), key=float))
|
||||
current_index = available_episodes.index(current_episode_num)
|
||||
|
||||
if current_index:
|
||||
prev_episode_num = available_episodes[current_index - 1]
|
||||
|
||||
return State(
|
||||
menu_name=MenuName.DOWNLOADS_PLAYER_CONTROLS,
|
||||
media_api=state.media_api,
|
||||
provider=state.provider.model_copy(
|
||||
update={"episode": prev_episode_num, "start_time": None}
|
||||
),
|
||||
)
|
||||
feedback.warning("This is the last available episode.")
|
||||
ctx.switch.force_dont_play()
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _replay(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _toggle_config_state(
|
||||
ctx: Context,
|
||||
state: State,
|
||||
config_state: Literal[
|
||||
"AUTO_ANIME", "AUTO_EPISODE", "CONTINUE_FROM_HISTORY", "TRANSLATION_TYPE"
|
||||
],
|
||||
) -> MenuAction:
|
||||
def action():
|
||||
match config_state:
|
||||
case "AUTO_ANIME":
|
||||
ctx.config.general.auto_select_anime_result = (
|
||||
not ctx.config.general.auto_select_anime_result
|
||||
)
|
||||
case "AUTO_EPISODE":
|
||||
ctx.config.stream.auto_next = not ctx.config.stream.auto_next
|
||||
case "CONTINUE_FROM_HISTORY":
|
||||
ctx.config.stream.continue_from_watch_history = (
|
||||
not ctx.config.stream.continue_from_watch_history
|
||||
)
|
||||
case "TRANSLATION_TYPE":
|
||||
ctx.config.stream.translation_type = (
|
||||
"sub" if ctx.config.stream.translation_type == "dub" else "dub"
|
||||
)
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _episodes_list(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
ctx.switch.force_episodes_menu()
|
||||
return InternalDirective.BACK
|
||||
|
||||
return action
|
||||
258
viu_cli/cli/interactive/menu/media/player_controls.py
Normal file
258
viu_cli/cli/interactive/menu/media/player_controls.py
Normal file
@@ -0,0 +1,258 @@
|
||||
from typing import Callable, Dict, Literal, Union
|
||||
|
||||
from ...session import Context, session
|
||||
from ...state import InternalDirective, MenuName, State
|
||||
|
||||
MenuAction = Callable[[], Union[State, InternalDirective]]
|
||||
|
||||
|
||||
@session.menu
|
||||
def player_controls(ctx: Context, state: State) -> Union[State, InternalDirective]:
|
||||
feedback = ctx.feedback
|
||||
feedback.clear_console()
|
||||
|
||||
config = ctx.config
|
||||
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
|
||||
|
||||
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:
|
||||
feedback.info("Auto-playing next episode...")
|
||||
next_episode_num = available_episodes[current_index + 1]
|
||||
|
||||
return State(
|
||||
menu_name=MenuName.SERVERS,
|
||||
media_api=state.media_api,
|
||||
provider=state.provider.model_copy(update={"episode": next_episode_num}),
|
||||
)
|
||||
|
||||
# --- 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)
|
||||
if current_index:
|
||||
options[f"{'⏪ ' if icons else ''}Previous Episode"] = _previous_episode(
|
||||
ctx, state
|
||||
)
|
||||
|
||||
options.update(
|
||||
{
|
||||
f"{'🔂 ' if icons else ''}Replay": _replay(ctx, state),
|
||||
f"{'💽 ' if icons else ''}Change Server": _change_server(ctx, state),
|
||||
f"{'📀 ' if icons else ''}Change Quality": _change_quality(ctx, state),
|
||||
f"{'🎞️ ' if icons else ''}Episode List": _episodes_list(ctx, state),
|
||||
f"{'🔘 ' if icons else ''}Toggle Auto Next Episode (Current: {ctx.config.stream.auto_next})": _toggle_config_state(
|
||||
ctx, state, "AUTO_EPISODE"
|
||||
),
|
||||
f"{'🔘 ' if icons else ''}Toggle Translation Type (Current: {ctx.config.stream.translation_type.upper()})": _toggle_config_state(
|
||||
ctx, state, "TRANSLATION_TYPE"
|
||||
),
|
||||
f"{'🎥 ' if icons else ''}Media Actions Menu": lambda: InternalDirective.BACKX4,
|
||||
f"{'🏠 ' if icons else ''}Main Menu": lambda: InternalDirective.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]()
|
||||
else:
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
|
||||
def _next_episode(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
|
||||
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]
|
||||
|
||||
return State(
|
||||
menu_name=MenuName.SERVERS,
|
||||
media_api=state.media_api,
|
||||
provider=state.provider.model_copy(
|
||||
update={"episode": next_episode_num}
|
||||
),
|
||||
)
|
||||
feedback.warning("This is the last available episode.")
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _previous_episode(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
|
||||
config = ctx.config
|
||||
|
||||
provider_anime = state.provider.anime
|
||||
current_episode_num = state.provider.episode
|
||||
|
||||
if not provider_anime or not current_episode_num:
|
||||
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:
|
||||
prev_episode_num = available_episodes[current_index - 1]
|
||||
|
||||
return State(
|
||||
menu_name=MenuName.SERVERS,
|
||||
media_api=state.media_api,
|
||||
provider=state.provider.model_copy(
|
||||
update={"episode": prev_episode_num}
|
||||
),
|
||||
)
|
||||
feedback.warning("This is the last available episode.")
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _replay(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
return InternalDirective.BACK
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _toggle_config_state(
|
||||
ctx: Context,
|
||||
state: State,
|
||||
config_state: Literal[
|
||||
"AUTO_ANIME", "AUTO_EPISODE", "CONTINUE_FROM_HISTORY", "TRANSLATION_TYPE"
|
||||
],
|
||||
) -> MenuAction:
|
||||
def action():
|
||||
match config_state:
|
||||
case "AUTO_ANIME":
|
||||
ctx.config.general.auto_select_anime_result = (
|
||||
not ctx.config.general.auto_select_anime_result
|
||||
)
|
||||
case "AUTO_EPISODE":
|
||||
ctx.config.stream.auto_next = not ctx.config.stream.auto_next
|
||||
case "CONTINUE_FROM_HISTORY":
|
||||
ctx.config.stream.continue_from_watch_history = (
|
||||
not ctx.config.stream.continue_from_watch_history
|
||||
)
|
||||
case "TRANSLATION_TYPE":
|
||||
ctx.config.stream.translation_type = (
|
||||
"sub" if ctx.config.stream.translation_type == "dub" else "dub"
|
||||
)
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _change_server(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
from .....libs.provider.anime.types import ProviderServer
|
||||
|
||||
feedback = ctx.feedback
|
||||
|
||||
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
|
||||
|
||||
new_server_name = selector.choose(
|
||||
"Select a different server:", list(server_map.keys())
|
||||
)
|
||||
if new_server_name:
|
||||
ctx.config.stream.server = ProviderServer(new_server_name)
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _episodes_list(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
ctx.switch.force_episodes_menu()
|
||||
return InternalDirective.BACKX2
|
||||
|
||||
return action
|
||||
|
||||
|
||||
def _change_quality(ctx: Context, state: State) -> MenuAction:
|
||||
def action():
|
||||
feedback = ctx.feedback
|
||||
|
||||
selector = ctx.selector
|
||||
|
||||
server_map = state.provider.servers
|
||||
|
||||
if not server_map:
|
||||
feedback.error("Player state is incomplete. Returning.")
|
||||
return InternalDirective.BACK
|
||||
|
||||
new_quality = selector.choose(
|
||||
"Select a different server:", list(["360", "480", "720", "1080"])
|
||||
)
|
||||
if new_quality:
|
||||
ctx.config.stream.quality = new_quality # type:ignore
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
return action
|
||||
102
viu_cli/cli/interactive/menu/media/provider_search.py
Normal file
102
viu_cli/cli/interactive/menu/media/provider_search.py
Normal file
@@ -0,0 +1,102 @@
|
||||
from .....libs.provider.anime.params import AnimeParams, SearchParams
|
||||
from .....libs.provider.anime.types import SearchResult
|
||||
from ...session import Context, session
|
||||
from ...state import InternalDirective, MenuName, ProviderState, State
|
||||
|
||||
|
||||
@session.menu
|
||||
def provider_search(ctx: Context, state: State) -> State | InternalDirective:
|
||||
from .....core.utils.fuzzy import fuzz
|
||||
from .....core.utils.normalizer import normalize_title, update_user_normalizer_json
|
||||
|
||||
feedback = ctx.feedback
|
||||
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
|
||||
|
||||
provider = ctx.provider
|
||||
selector = ctx.selector
|
||||
config = ctx.config
|
||||
feedback.clear_console()
|
||||
|
||||
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",
|
||||
)
|
||||
return InternalDirective.BACK
|
||||
|
||||
provider_search_results = provider.search(
|
||||
SearchParams(
|
||||
query=normalize_title(media_title, config.general.provider.value, True),
|
||||
translation_type=config.stream.translation_type,
|
||||
)
|
||||
)
|
||||
|
||||
if not provider_search_results or not provider_search_results.results:
|
||||
feedback.warning(
|
||||
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
|
||||
|
||||
provider_results_map: dict[str, SearchResult] = {
|
||||
result.title: result for result in provider_search_results.results
|
||||
}
|
||||
|
||||
selected_provider_anime: SearchResult | None = None
|
||||
|
||||
# --- Auto-Select or Prompt ---
|
||||
if config.general.auto_select_anime_result:
|
||||
# Use fuzzy matching to find the best title
|
||||
best_match_title = max(
|
||||
provider_results_map.keys(),
|
||||
key=lambda p_title: fuzz.ratio(
|
||||
normalize_title(p_title, config.general.provider.value).lower(),
|
||||
media_title.lower(),
|
||||
),
|
||||
)
|
||||
feedback.info(f"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 '{media_title}'", choices=choices
|
||||
)
|
||||
|
||||
if not chosen_title or chosen_title == "Back":
|
||||
return InternalDirective.BACK
|
||||
|
||||
if selector.confirm(
|
||||
f"Would you like to update your local normalizer json with: {chosen_title} for {media_title}"
|
||||
):
|
||||
update_user_normalizer_json(
|
||||
chosen_title, media_title, config.general.provider.value
|
||||
)
|
||||
selected_provider_anime = provider_results_map[chosen_title]
|
||||
|
||||
with feedback.progress(
|
||||
f"[cyan]Fetching full details for '{selected_provider_anime.title}'"
|
||||
):
|
||||
full_provider_anime = provider.get(
|
||||
AnimeParams(id=selected_provider_anime.id, query=media_title.lower())
|
||||
)
|
||||
|
||||
if not full_provider_anime:
|
||||
feedback.warning(
|
||||
f"Failed to fetch details for '{selected_provider_anime.title}'."
|
||||
)
|
||||
return InternalDirective.BACK
|
||||
|
||||
return State(
|
||||
menu_name=MenuName.EPISODES,
|
||||
media_api=state.media_api,
|
||||
provider=ProviderState(
|
||||
search_results=provider_search_results,
|
||||
anime=full_provider_anime,
|
||||
),
|
||||
)
|
||||
207
viu_cli/cli/interactive/menu/media/results.py
Normal file
207
viu_cli/cli/interactive/menu/media/results.py
Normal file
@@ -0,0 +1,207 @@
|
||||
from dataclasses import asdict
|
||||
from typing import Callable, Dict, Union
|
||||
|
||||
from .....libs.media_api.params import MediaSearchParams, UserMediaListSearchParams
|
||||
from .....libs.media_api.types import MediaItem, MediaStatus, UserMediaListStatus
|
||||
from ...session import Context, session
|
||||
from ...state import InternalDirective, MediaApiState, MenuName, State
|
||||
|
||||
|
||||
@session.menu
|
||||
def results(ctx: Context, state: State) -> State | InternalDirective:
|
||||
feedback = ctx.feedback
|
||||
feedback.clear_console()
|
||||
|
||||
search_result = state.media_api.search_result
|
||||
page_info = state.media_api.page_info
|
||||
|
||||
if not search_result:
|
||||
feedback.info("No anime found for the given criteria")
|
||||
return InternalDirective.BACK
|
||||
|
||||
search_result_dict = {
|
||||
_format_title(ctx, media_item): media_item
|
||||
for media_item in search_result.values()
|
||||
}
|
||||
choices: Dict[str, Callable[[], Union[int, State, InternalDirective]]] = {
|
||||
title: lambda media_id=item.id: media_id
|
||||
for title, item in search_result_dict.items()
|
||||
}
|
||||
if page_info:
|
||||
if page_info.has_next_page:
|
||||
choices.update(
|
||||
{
|
||||
f"Next Page (Page {page_info.current_page + 1})": lambda: _handle_pagination(
|
||||
ctx, state, 1
|
||||
)
|
||||
}
|
||||
)
|
||||
if page_info.current_page > 1:
|
||||
choices.update(
|
||||
{
|
||||
f"Previous Page (Page {page_info.current_page - 1})": lambda: _handle_pagination(
|
||||
ctx, state, -1
|
||||
)
|
||||
}
|
||||
)
|
||||
choices.update(
|
||||
{
|
||||
"Back": lambda: InternalDirective.BACK
|
||||
if page_info and page_info.current_page == 1
|
||||
else InternalDirective.MAIN,
|
||||
"Exit": lambda: InternalDirective.EXIT,
|
||||
}
|
||||
)
|
||||
|
||||
preview_command = None
|
||||
if ctx.config.general.preview != "none":
|
||||
from ....utils.preview import create_preview_context
|
||||
|
||||
with create_preview_context() as preview_ctx:
|
||||
preview_command = preview_ctx.get_anime_preview(
|
||||
list(search_result_dict.values()),
|
||||
list(search_result_dict.keys()),
|
||||
ctx.config,
|
||||
)
|
||||
|
||||
choice = ctx.selector.choose(
|
||||
prompt="Select Anime",
|
||||
choices=list(choices),
|
||||
preview=preview_command,
|
||||
)
|
||||
|
||||
else:
|
||||
# No preview mode
|
||||
|
||||
choice = ctx.selector.choose(
|
||||
prompt="Select Anime",
|
||||
choices=list(choices),
|
||||
preview=None,
|
||||
)
|
||||
|
||||
if not choice:
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
next_step = choices[choice]()
|
||||
if isinstance(next_step, State) or isinstance(next_step, InternalDirective):
|
||||
return next_step
|
||||
else:
|
||||
return State(
|
||||
menu_name=MenuName.MEDIA_ACTIONS,
|
||||
media_api=MediaApiState(
|
||||
media_id=next_step,
|
||||
search_result=state.media_api.search_result,
|
||||
page_info=state.media_api.page_info,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _format_title(ctx: Context, media_item: MediaItem) -> str:
|
||||
config = ctx.config
|
||||
|
||||
title = media_item.title.english or media_item.title.romaji
|
||||
progress = "0"
|
||||
|
||||
if media_item.user_status:
|
||||
progress = str(media_item.user_status.progress or 0)
|
||||
|
||||
episodes_total = str(media_item.episodes or "??")
|
||||
display_title = f"{title} ({progress} of {episodes_total})"
|
||||
|
||||
# Add a visual indicator for new episodes if applicable
|
||||
if (
|
||||
media_item.status == MediaStatus.RELEASING
|
||||
and media_item.next_airing
|
||||
and media_item.user_status
|
||||
and media_item.user_status.status == UserMediaListStatus.WATCHING
|
||||
):
|
||||
last_aired = media_item.next_airing.episode - 1
|
||||
unwatched = last_aired - (media_item.user_status.progress or 0)
|
||||
if unwatched > 0:
|
||||
icon = "🔹" if config.general.icons else "!"
|
||||
display_title += f" {icon}{unwatched} new{icon}"
|
||||
|
||||
return display_title
|
||||
|
||||
|
||||
def _handle_pagination(
|
||||
ctx: Context, state: State, page_delta: int
|
||||
) -> State | InternalDirective:
|
||||
feedback = ctx.feedback
|
||||
|
||||
search_params = state.media_api.search_params
|
||||
|
||||
if (
|
||||
not state.media_api.search_result
|
||||
or not state.media_api.page_info
|
||||
or not search_params
|
||||
):
|
||||
feedback.error("No search results available for pagination")
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
current_page = state.media_api.page_info.current_page
|
||||
new_page = current_page + page_delta
|
||||
|
||||
# Validate page bounds
|
||||
if new_page < 1:
|
||||
feedback.warning("Already at the first page")
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
if page_delta == -1:
|
||||
return InternalDirective.BACK
|
||||
if page_delta > 0 and not state.media_api.page_info.has_next_page:
|
||||
feedback.warning("No more pages available")
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
# Determine which type of search to perform based on stored parameters
|
||||
if isinstance(search_params, UserMediaListSearchParams):
|
||||
if not ctx.media_api.is_authenticated():
|
||||
feedback.error("You haven't logged in")
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
search_params_dict = asdict(search_params)
|
||||
search_params_dict.pop("page")
|
||||
|
||||
loading_message = "Fetching media list"
|
||||
result = None
|
||||
new_search_params = UserMediaListSearchParams(
|
||||
**search_params_dict, page=new_page
|
||||
)
|
||||
with feedback.progress(loading_message):
|
||||
result = ctx.media_api.search_media_list(new_search_params)
|
||||
|
||||
if result:
|
||||
return State(
|
||||
menu_name=MenuName.RESULTS,
|
||||
media_api=MediaApiState(
|
||||
search_result={
|
||||
media_item.id: media_item for media_item in result.media
|
||||
},
|
||||
search_params=new_search_params,
|
||||
page_info=result.page_info,
|
||||
),
|
||||
)
|
||||
else:
|
||||
search_params_dict = asdict(search_params)
|
||||
search_params_dict.pop("page")
|
||||
|
||||
loading_message = "Fetching media list"
|
||||
result = None
|
||||
new_search_params = MediaSearchParams(**search_params_dict, page=new_page)
|
||||
with feedback.progress(loading_message):
|
||||
result = ctx.media_api.search_media(new_search_params)
|
||||
|
||||
if result:
|
||||
return State(
|
||||
menu_name=MenuName.RESULTS,
|
||||
media_api=MediaApiState(
|
||||
search_result={
|
||||
media_item.id: media_item for media_item in result.media
|
||||
},
|
||||
search_params=new_search_params,
|
||||
page_info=result.page_info,
|
||||
),
|
||||
)
|
||||
|
||||
feedback.warning("Failed to load page")
|
||||
return InternalDirective.RELOAD
|
||||
121
viu_cli/cli/interactive/menu/media/servers.py
Normal file
121
viu_cli/cli/interactive/menu/media/servers.py
Normal file
@@ -0,0 +1,121 @@
|
||||
from typing import Dict, List
|
||||
|
||||
from .....libs.player.params import PlayerParams
|
||||
from .....libs.provider.anime.params import EpisodeStreamsParams
|
||||
from .....libs.provider.anime.types import ProviderServer, Server
|
||||
from ...session import Context, session
|
||||
from ...state import InternalDirective, MenuName, State
|
||||
|
||||
|
||||
@session.menu
|
||||
def servers(ctx: Context, state: State) -> State | InternalDirective:
|
||||
feedback = ctx.feedback
|
||||
|
||||
config = ctx.config
|
||||
provider = ctx.provider
|
||||
selector = ctx.selector
|
||||
|
||||
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:
|
||||
feedback.error("Anime or episode details are missing")
|
||||
return InternalDirective.BACK
|
||||
|
||||
with feedback.progress("Fetching Servers"):
|
||||
server_iterator = provider.episode_streams(
|
||||
EpisodeStreamsParams(
|
||||
anime_id=provider_anime.id,
|
||||
query=anime_title,
|
||||
episode=episode_number,
|
||||
translation_type=config.stream.translation_type,
|
||||
)
|
||||
)
|
||||
# Consume the iterator to get a list of all servers
|
||||
if config.stream.server == ProviderServer.TOP and server_iterator:
|
||||
try:
|
||||
all_servers = [next(server_iterator)]
|
||||
except Exception:
|
||||
all_servers = []
|
||||
else:
|
||||
all_servers: List[Server] = list(server_iterator) if server_iterator else []
|
||||
|
||||
if not all_servers:
|
||||
feedback.error(f"o streaming servers found for episode {episode_number}")
|
||||
return InternalDirective.BACKX3
|
||||
|
||||
server_map: Dict[str, Server] = {s.name: s for s in all_servers}
|
||||
selected_server: Server | None = None
|
||||
|
||||
preferred_server = config.stream.server.value
|
||||
if preferred_server == "TOP":
|
||||
selected_server = all_servers[0]
|
||||
feedback.info(f"Auto-selecting top server: {selected_server.name}")
|
||||
elif preferred_server in server_map:
|
||||
selected_server = server_map[preferred_server]
|
||||
feedback.info(f"Auto-selecting preferred server: {selected_server.name}")
|
||||
else:
|
||||
choices = [*server_map.keys(), "Back"]
|
||||
chosen_name = selector.choose("Select Server", choices)
|
||||
if not chosen_name or chosen_name == "Back":
|
||||
return InternalDirective.BACK
|
||||
selected_server = server_map[chosen_name]
|
||||
|
||||
stream_link_obj = _filter_by_quality(selected_server.links, config.stream.quality)
|
||||
if not stream_link_obj:
|
||||
feedback.error(
|
||||
f"No stream of quality '{config.stream.quality}' found on server '{selected_server.name}'."
|
||||
)
|
||||
return InternalDirective.RELOAD
|
||||
|
||||
final_title = (
|
||||
media_item.streaming_episodes[episode_number].title
|
||||
if media_item.streaming_episodes.get(episode_number)
|
||||
else f"{media_item.title.english}; Episode {episode_number}"
|
||||
)
|
||||
feedback.info(f"[bold green]Launching player for:[/] {final_title}")
|
||||
|
||||
if not state.media_api.media_item or not state.provider.anime:
|
||||
return InternalDirective.BACKX3
|
||||
player_result = ctx.player.play(
|
||||
PlayerParams(
|
||||
url=stream_link_obj.link,
|
||||
title=final_title,
|
||||
query=(
|
||||
state.media_api.media_item.title.romaji
|
||||
or state.media_api.media_item.title.english
|
||||
),
|
||||
episode=episode_number,
|
||||
subtitles=[sub.url for sub in selected_server.subtitles],
|
||||
headers=selected_server.headers,
|
||||
start_time=state.provider.start_time,
|
||||
),
|
||||
state.provider.anime,
|
||||
state.media_api.media_item,
|
||||
)
|
||||
if media_item and episode_number:
|
||||
ctx.watch_history.track(media_item, player_result)
|
||||
|
||||
return State(
|
||||
menu_name=MenuName.PLAYER_CONTROLS,
|
||||
media_api=state.media_api,
|
||||
provider=state.provider.model_copy(
|
||||
update={
|
||||
"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
|
||||
331
viu_cli/cli/interactive/session.py
Normal file
331
viu_cli/cli/interactive/session.py
Normal file
@@ -0,0 +1,331 @@
|
||||
import importlib.util
|
||||
import logging
|
||||
import os
|
||||
from dataclasses import dataclass, field
|
||||
from typing import TYPE_CHECKING, Callable, List, Optional, Union
|
||||
|
||||
import click
|
||||
|
||||
from ...core.config import AppConfig
|
||||
from ...core.constants import APP_DIR, USER_CONFIG
|
||||
from .state import InternalDirective, MenuName, State
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ...libs.media_api.base import BaseApiClient
|
||||
from ...libs.provider.anime.base import BaseAnimeProvider
|
||||
from ...libs.selectors.base import BaseSelector
|
||||
from ..service.auth import AuthService
|
||||
from ..service.feedback import FeedbackService
|
||||
from ..service.player import PlayerService
|
||||
from ..service.registry import MediaRegistryService
|
||||
from ..service.session import SessionsService
|
||||
from ..service.watch_history import WatchHistoryService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
MENUS_DIR = APP_DIR / "cli" / "interactive" / "menu"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Switch:
|
||||
"Forces menus to show selector and not just pass through,once viewed it auto sets back to false"
|
||||
|
||||
_provider_results: bool = False
|
||||
_episodes: bool = False
|
||||
_servers: bool = False
|
||||
_dont_play: bool = False
|
||||
|
||||
@property
|
||||
def show_provider_results_menu(self):
|
||||
if self._provider_results:
|
||||
self._provider_results = False
|
||||
return True
|
||||
return False
|
||||
|
||||
def force_provider_results_menu(self):
|
||||
self._provider_results = True
|
||||
|
||||
@property
|
||||
def dont_play(self):
|
||||
if self._dont_play:
|
||||
self._dont_play = False
|
||||
return True
|
||||
return False
|
||||
|
||||
def force_dont_play(self):
|
||||
self._dont_play = True
|
||||
|
||||
@property
|
||||
def show_episodes_menu(self):
|
||||
if self._episodes:
|
||||
self._episodes = False
|
||||
return True
|
||||
return False
|
||||
|
||||
def force_episodes_menu(self):
|
||||
self._episodes = True
|
||||
|
||||
@property
|
||||
def servers(self):
|
||||
if self._servers:
|
||||
self._servers = False
|
||||
return True
|
||||
return False
|
||||
|
||||
def force_servers_menu(self):
|
||||
self._servers = True
|
||||
|
||||
|
||||
@dataclass
|
||||
class Context:
|
||||
config: "AppConfig"
|
||||
switch: Switch = field(default_factory=Switch)
|
||||
_provider: Optional["BaseAnimeProvider"] = None
|
||||
_selector: Optional["BaseSelector"] = None
|
||||
_media_api: Optional["BaseApiClient"] = None
|
||||
|
||||
_feedback: Optional["FeedbackService"] = None
|
||||
_media_registry: Optional["MediaRegistryService"] = None
|
||||
_watch_history: Optional["WatchHistoryService"] = None
|
||||
_session: Optional["SessionsService"] = None
|
||||
_auth: Optional["AuthService"] = None
|
||||
_player: Optional["PlayerService"] = None
|
||||
|
||||
@property
|
||||
def provider(self) -> "BaseAnimeProvider":
|
||||
if not self._provider:
|
||||
from ...libs.provider.anime.provider import create_provider
|
||||
|
||||
self._provider = create_provider(self.config.general.provider)
|
||||
return self._provider
|
||||
|
||||
@property
|
||||
def selector(self) -> "BaseSelector":
|
||||
if not self._selector:
|
||||
from ...libs.selectors.selector import create_selector
|
||||
|
||||
self._selector = create_selector(self.config)
|
||||
return self._selector
|
||||
|
||||
@property
|
||||
def media_api(self) -> "BaseApiClient":
|
||||
if not self._media_api:
|
||||
from ...libs.media_api.api import create_api_client
|
||||
|
||||
media_api = create_api_client(self.config.general.media_api, self.config)
|
||||
|
||||
auth = self.auth
|
||||
if auth_profile := auth.get_auth():
|
||||
p = media_api.authenticate(auth_profile.token)
|
||||
if p:
|
||||
logger.debug(f"Authenticated as {p.name}")
|
||||
else:
|
||||
logger.warning(f"Failed to authenticate with {auth_profile.token}")
|
||||
else:
|
||||
logger.debug("Not authenticated")
|
||||
self._media_api = media_api
|
||||
|
||||
return self._media_api
|
||||
|
||||
@property
|
||||
def player(self) -> "PlayerService":
|
||||
if not self._player:
|
||||
from ..service.player import PlayerService
|
||||
|
||||
self._player = PlayerService(
|
||||
self.config, self.provider, self.media_registry
|
||||
)
|
||||
return self._player
|
||||
|
||||
@property
|
||||
def feedback(self) -> "FeedbackService":
|
||||
if not self._feedback:
|
||||
from ..service.feedback.service import FeedbackService
|
||||
|
||||
self._feedback = FeedbackService(self.config)
|
||||
return self._feedback
|
||||
|
||||
@property
|
||||
def media_registry(self) -> "MediaRegistryService":
|
||||
if not self._media_registry:
|
||||
from ..service.registry.service import MediaRegistryService
|
||||
|
||||
self._media_registry = MediaRegistryService(
|
||||
self.config.general.media_api, self.config.media_registry
|
||||
)
|
||||
return self._media_registry
|
||||
|
||||
@property
|
||||
def watch_history(self) -> "WatchHistoryService":
|
||||
if not self._watch_history:
|
||||
from ..service.watch_history.service import WatchHistoryService
|
||||
|
||||
self._watch_history = WatchHistoryService(
|
||||
self.config, self.media_registry, self.media_api
|
||||
)
|
||||
return self._watch_history
|
||||
|
||||
@property
|
||||
def session(self) -> "SessionsService":
|
||||
if not self._session:
|
||||
from ..service.session.service import SessionsService
|
||||
|
||||
self._session = SessionsService(self.config.sessions)
|
||||
return self._session
|
||||
|
||||
@property
|
||||
def auth(self) -> "AuthService":
|
||||
if not self._auth:
|
||||
from ..service.auth.service import AuthService
|
||||
|
||||
self._auth = AuthService(self.config.general.media_api)
|
||||
return self._auth
|
||||
|
||||
|
||||
MenuFunction = Callable[[Context, State], Union[State, InternalDirective]]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Menu:
|
||||
name: MenuName
|
||||
execute: MenuFunction
|
||||
|
||||
|
||||
class Session:
|
||||
_context: Context
|
||||
_history: List[State] = []
|
||||
_menus: dict[MenuName, Menu] = {}
|
||||
|
||||
def _load_context(self, config: AppConfig):
|
||||
self._context = Context(config)
|
||||
logger.info("Application context reloaded.")
|
||||
|
||||
def _edit_config(self):
|
||||
from ..config import ConfigLoader
|
||||
|
||||
click.edit(filename=str(USER_CONFIG))
|
||||
logger.debug("Config changed; Reloading context")
|
||||
loader = ConfigLoader()
|
||||
config = loader.load()
|
||||
self._load_context(config)
|
||||
|
||||
def run(
|
||||
self,
|
||||
config: AppConfig,
|
||||
resume: bool = False,
|
||||
history: Optional[List[State]] = None,
|
||||
):
|
||||
self._load_context(config)
|
||||
if resume:
|
||||
if history := self._context.session.get_default_session_history():
|
||||
self._history = history
|
||||
else:
|
||||
logger.warning("Failed to continue from history. No sessions found")
|
||||
|
||||
if history:
|
||||
self._history = history
|
||||
else:
|
||||
self._history.append(State(menu_name=MenuName.MAIN))
|
||||
|
||||
try:
|
||||
self._run_main_loop()
|
||||
except Exception:
|
||||
self._context.session.create_crash_backup(self._history)
|
||||
raise
|
||||
finally:
|
||||
# Clean up preview workers when session ends
|
||||
self._cleanup_preview_workers()
|
||||
self._context.session.save_session(self._history)
|
||||
|
||||
def _cleanup_preview_workers(self):
|
||||
"""Clean up preview workers when session ends."""
|
||||
try:
|
||||
from ..utils.preview import shutdown_preview_workers
|
||||
|
||||
shutdown_preview_workers(wait=False, timeout=5.0)
|
||||
logger.debug("Preview workers cleaned up successfully")
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to cleanup preview workers: {e}")
|
||||
|
||||
def _run_main_loop(self):
|
||||
"""Run the main session loop."""
|
||||
while self._history:
|
||||
current_state = self._history[-1]
|
||||
|
||||
next_step = self._menus[current_state.menu_name].execute(
|
||||
self._context, current_state
|
||||
)
|
||||
|
||||
if isinstance(next_step, InternalDirective):
|
||||
if next_step == InternalDirective.MAIN:
|
||||
self._history = [self._history[0]]
|
||||
elif next_step == InternalDirective.RELOAD:
|
||||
continue
|
||||
elif next_step == InternalDirective.CONFIG_EDIT:
|
||||
self._edit_config()
|
||||
elif next_step == InternalDirective.BACK:
|
||||
if len(self._history) > 1:
|
||||
self._history.pop()
|
||||
elif next_step == InternalDirective.BACKX2:
|
||||
if len(self._history) > 2:
|
||||
self._history.pop()
|
||||
self._history.pop()
|
||||
elif next_step == InternalDirective.BACKX3:
|
||||
if len(self._history) > 3:
|
||||
self._history.pop()
|
||||
self._history.pop()
|
||||
self._history.pop()
|
||||
elif next_step == InternalDirective.BACKX4:
|
||||
if len(self._history) > 4:
|
||||
self._history.pop()
|
||||
self._history.pop()
|
||||
self._history.pop()
|
||||
self._history.pop()
|
||||
elif next_step == InternalDirective.EXIT:
|
||||
break
|
||||
else:
|
||||
self._history.append(next_step)
|
||||
|
||||
@property
|
||||
def menu(self) -> Callable[[MenuFunction], MenuFunction]:
|
||||
"""A decorator to register a function as a menu."""
|
||||
|
||||
def decorator(func: MenuFunction) -> MenuFunction:
|
||||
menu_name = MenuName(func.__name__.upper())
|
||||
if menu_name in self._menus:
|
||||
logger.warning(f"Menu '{menu_name}' is being redefined.")
|
||||
self._menus[menu_name] = Menu(name=menu_name, execute=func)
|
||||
return func
|
||||
|
||||
return decorator
|
||||
|
||||
def load_menus_from_folder(self, package: str):
|
||||
package_path = MENUS_DIR / package
|
||||
package_name = package_path.name
|
||||
logger.debug(f"Loading menus from '{package_path}'...")
|
||||
|
||||
for filename in os.listdir(package_path):
|
||||
if filename.endswith(".py") and not filename.startswith("__"):
|
||||
module_name = filename[:-3]
|
||||
full_module_name = (
|
||||
f"viu_cli.cli.interactive.menu.{package_name}.{module_name}"
|
||||
)
|
||||
file_path = package_path / filename
|
||||
|
||||
try:
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
full_module_name, file_path
|
||||
)
|
||||
if spec and spec.loader:
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
# The act of executing the module runs the @session.menu decorators
|
||||
spec.loader.exec_module(module)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Failed to load menu module '{full_module_name}': {e}"
|
||||
)
|
||||
|
||||
|
||||
# Create a single, global instance of the Session to be imported by menu modules.
|
||||
session = Session()
|
||||
85
viu_cli/cli/interactive/state.py
Normal file
85
viu_cli/cli/interactive/state.py
Normal file
@@ -0,0 +1,85 @@
|
||||
from enum import Enum
|
||||
from typing import Dict, Optional, Union
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
from ...libs.media_api.params import MediaSearchParams, UserMediaListSearchParams
|
||||
from ...libs.media_api.types import MediaItem, PageInfo
|
||||
from ...libs.provider.anime.types import Anime, SearchResults, Server
|
||||
|
||||
|
||||
# TODO: is internal directive a good name
|
||||
class InternalDirective(Enum):
|
||||
MAIN = "MAIN"
|
||||
|
||||
BACK = "BACK"
|
||||
|
||||
BACKX2 = "BACKX2"
|
||||
|
||||
BACKX3 = "BACKX3"
|
||||
|
||||
BACKX4 = "BACKX4"
|
||||
|
||||
EXIT = "EXIT"
|
||||
|
||||
CONFIG_EDIT = "CONFIG_EDIT"
|
||||
|
||||
RELOAD = "RELOAD"
|
||||
|
||||
|
||||
class MenuName(Enum):
|
||||
MAIN = "MAIN"
|
||||
AUTH = "AUTH"
|
||||
EPISODES = "EPISODES"
|
||||
RESULTS = "RESULTS"
|
||||
SERVERS = "SERVERS"
|
||||
WATCH_HISTORY = "WATCH_HISTORY"
|
||||
PROVIDER_SEARCH = "PROVIDER_SEARCH"
|
||||
PLAYER_CONTROLS = "PLAYER_CONTROLS"
|
||||
USER_MEDIA_LIST = "USER_MEDIA_LIST"
|
||||
SESSION_MANAGEMENT = "SESSION_MANAGEMENT"
|
||||
MEDIA_ACTIONS = "MEDIA_ACTIONS"
|
||||
DOWNLOADS = "DOWNLOADS"
|
||||
DYNAMIC_SEARCH = "DYNAMIC_SEARCH"
|
||||
MEDIA_REVIEW = "MEDIA_REVIEW"
|
||||
MEDIA_CHARACTERS = "MEDIA_CHARACTERS"
|
||||
MEDIA_AIRING_SCHEDULE = "MEDIA_AIRING_SCHEDULE"
|
||||
PLAY_DOWNLOADS = "PLAY_DOWNLOADS"
|
||||
DOWNLOADS_PLAYER_CONTROLS = "DOWNLOADS_PLAYER_CONTROLS"
|
||||
DOWNLOAD_EPISODES = "DOWNLOAD_EPISODES"
|
||||
|
||||
|
||||
class StateModel(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
|
||||
class MediaApiState(StateModel):
|
||||
search_result: Optional[Dict[int, MediaItem]] = None
|
||||
search_params: Optional[Union[MediaSearchParams, UserMediaListSearchParams]] = None
|
||||
page_info: Optional[PageInfo] = None
|
||||
media_id: Optional[int] = None
|
||||
|
||||
@property
|
||||
def media_item(self) -> Optional[MediaItem]:
|
||||
if self.search_result and self.media_id:
|
||||
return self.search_result[self.media_id]
|
||||
|
||||
|
||||
class ProviderState(StateModel):
|
||||
search_results: Optional[SearchResults] = None
|
||||
anime: Optional[Anime] = None
|
||||
episode: Optional[str] = None
|
||||
servers: Optional[Dict[str, Server]] = None
|
||||
server_name: Optional[str] = None
|
||||
start_time: Optional[str] = None
|
||||
|
||||
@property
|
||||
def server(self) -> Optional[Server]:
|
||||
if self.servers and self.server_name:
|
||||
return self.servers[self.server_name]
|
||||
|
||||
|
||||
class State(StateModel):
|
||||
menu_name: MenuName
|
||||
provider: ProviderState = Field(default_factory=ProviderState)
|
||||
media_api: MediaApiState = Field(default_factory=MediaApiState)
|
||||
Reference in New Issue
Block a user