feat: implement enhanced feedback system for user interactions

This commit is contained in:
Benexl
2025-07-14 20:58:52 +03:00
parent a88df7f3ef
commit a079f9919c
6 changed files with 415 additions and 44 deletions

View File

@@ -2,10 +2,10 @@ import random
from typing import Callable, Dict, Tuple
from rich.console import Console
from rich.progress import Progress
from ....libs.api.params import ApiSearchParams, UserListParams
from ....libs.api.types import MediaSearchResult, MediaStatus, UserListStatusType
from ...utils.feedback import create_feedback_manager, execute_with_feedback
from ..session import Context, session
from ..state import ControlFlow, MediaApiState, State
@@ -19,6 +19,7 @@ def main(ctx: Context, state: State) -> State | ControlFlow:
Displays top-level categories for the user to browse and select.
"""
icons = ctx.config.general.icons
feedback = create_feedback_manager(icons)
console = Console()
console.clear()
@@ -83,8 +84,9 @@ def main(ctx: Context, state: State) -> State | ControlFlow:
return ControlFlow.CONTINUE
if not result_data:
console.print(
f"[bold red]Error:[/bold red] Failed to fetch data for '{choice_str.strip()}'."
feedback.error(
f"Failed to fetch data for '{choice_str.strip()}'",
"Please check your internet connection and try again.",
)
return ControlFlow.CONTINUE
@@ -101,39 +103,73 @@ def _create_media_list_action(
"""A factory to create menu actions for fetching media lists"""
def action():
with Progress(transient=True) as progress:
progress.add_task(f"[cyan]Fetching anime...", total=None)
return "RESULTS", ctx.media_api.search_media(
feedback = create_feedback_manager(ctx.config.general.icons)
def fetch_data():
return ctx.media_api.search_media(
ApiSearchParams(
sort=sort, per_page=ctx.config.anilist.per_page, status=status
)
)
success, result = execute_with_feedback(
fetch_data,
feedback,
"fetch anime list",
loading_msg="Fetching anime",
success_msg="Anime list loaded successfully",
)
return "RESULTS" if success else "CONTINUE", result
return action
def _create_random_media_list(ctx: Context) -> MenuAction:
def action():
with Progress(transient=True) as progress:
progress.add_task(f"[cyan]Fetching random anime...", total=None)
return "RESULTS", ctx.media_api.search_media(
feedback = create_feedback_manager(ctx.config.general.icons)
def fetch_data():
return ctx.media_api.search_media(
ApiSearchParams(
id_in=random.sample(range(1, 160000), k=50),
per_page=ctx.config.anilist.per_page,
)
)
success, result = execute_with_feedback(
fetch_data,
feedback,
"fetch random anime",
loading_msg="Fetching random anime",
success_msg="Random anime loaded successfully",
)
return "RESULTS" if success else "CONTINUE", result
return action
def _create_search_media_list(ctx: Context) -> MenuAction:
def action():
feedback = create_feedback_manager(ctx.config.general.icons)
query = ctx.selector.ask("Search for Anime")
if not query:
return "CONTINUE", None
with Progress(transient=True) as progress:
progress.add_task(f"[cyan]Searching for {query}...", total=None)
return "RESULTS", ctx.media_api.search_media(ApiSearchParams(query=query))
def fetch_data():
return ctx.media_api.search_media(ApiSearchParams(query=query))
success, result = execute_with_feedback(
fetch_data,
feedback,
"search anime",
loading_msg=f"Searching for '{query}'",
success_msg=f"Search results for '{query}' loaded successfully",
)
return "RESULTS" if success else "CONTINUE", result
return action
@@ -142,15 +178,29 @@ def _create_user_list_action(ctx: Context, status: UserListStatusType) -> MenuAc
"""A factory to create menu actions for fetching user lists, handling authentication."""
def action():
feedback = create_feedback_manager(ctx.config.general.icons)
# Check authentication (commented code from original)
# if not ctx.media_api.user_profile:
# click.echo(
# f"[bold yellow]Please log in to view your '{status.title()}' list.[/]"
# feedback.warning(
# f"Please log in to view your '{status.title()}' list",
# "You need to authenticate with AniList to access your personal lists"
# )
# return "CONTINUE", None
with Progress(transient=True) as progress:
progress.add_task(f"[cyan]Fetching random anime...", total=None)
return "RESULTS", ctx.media_api.fetch_user_list(
def fetch_data():
return ctx.media_api.fetch_user_list(
UserListParams(status=status, per_page=ctx.config.anilist.per_page)
)
success, result = execute_with_feedback(
fetch_data,
feedback,
f"fetch {status.lower()} list",
loading_msg=f"Fetching your {status.lower()} list",
success_msg=f"Your {status.lower()} list loaded successfully",
)
return "RESULTS" if success else "CONTINUE", result
return action

View File

@@ -6,6 +6,7 @@ from rich.console import Console
from ....libs.api.params import UpdateListEntryParams
from ....libs.api.types import MediaItem
from ....libs.players.params import PlayerParams
from ...utils.feedback import create_feedback_manager, execute_with_feedback
from ..session import Context, session
from ..state import ControlFlow, ProviderState, State
@@ -55,17 +56,29 @@ def _stream(ctx: Context, state: State) -> MenuAction:
def _watch_trailer(ctx: Context, state: State) -> MenuAction:
def action():
feedback = create_feedback_manager(ctx.config.general.icons)
anime = state.media_api.anime
if not anime:
return ControlFlow.CONTINUE
if not anime.trailer or not anime.trailer.id:
print("[bold yellow]No trailer available for this anime.[/bold yellow]")
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={anime.trailer.id}"
print(
f"Playing trailer for '{anime.title.english or anime.title.romaji}'..."
def play_trailer():
ctx.player.play(PlayerParams(url=trailer_url, title=""))
execute_with_feedback(
play_trailer,
feedback,
"play trailer",
loading_msg=f"Playing trailer for '{anime.title.english or anime.title.romaji}'",
success_msg="Trailer started successfully",
show_loading=False,
)
ctx.player.play(PlayerParams(url=trailer_url, title=""))
return ControlFlow.CONTINUE
return action
@@ -73,16 +86,18 @@ def _watch_trailer(ctx: Context, state: State) -> MenuAction:
def _add_to_list(ctx: Context, state: State) -> MenuAction:
def action():
feedback = create_feedback_manager(ctx.config.general.icons)
anime = state.media_api.anime
if not anime:
return ControlFlow.CONTINUE
choices = ["CURRENT", "PLANNING", "COMPLETED", "DROPPED", "PAUSED", "REPEATING"]
status = ctx.selector.choose("Select list status:", choices=choices)
if status:
_update_user_list(
_update_user_list_with_feedback(
ctx,
anime,
UpdateListEntryParams(media_id=anime.id, status=status),
feedback,
)
return ControlFlow.CONTINUE
@@ -91,6 +106,7 @@ def _add_to_list(ctx: Context, state: State) -> MenuAction:
def _score_anime(ctx: Context, state: State) -> MenuAction:
def action():
feedback = create_feedback_manager(ctx.config.general.icons)
anime = state.media_api.anime
if not anime:
return ControlFlow.CONTINUE
@@ -99,12 +115,15 @@ def _score_anime(ctx: Context, state: State) -> MenuAction:
score = float(score_str) if score_str else 0.0
if not 0.0 <= score <= 10.0:
raise ValueError("Score out of range.")
_update_user_list(
ctx, anime, UpdateListEntryParams(media_id=anime.id, score=score)
_update_user_list_with_feedback(
ctx,
anime,
UpdateListEntryParams(media_id=anime.id, score=score),
feedback,
)
except (ValueError, TypeError):
print(
"[bold red]Invalid score. Please enter a number between 0 and 10.[/bold red]"
feedback.error(
"Invalid score entered", "Please enter a number between 0.0 and 10.0"
)
return ControlFlow.CONTINUE
@@ -155,3 +174,30 @@ def _update_user_list(ctx: Context, anime: MediaItem, params: UpdateListEntryPar
)
else:
click.echo("[bold red]Failed to update list entry.[/bold red]")
def _update_user_list_with_feedback(
ctx: Context, anime: MediaItem, params: UpdateListEntryParams, feedback
):
"""Helper to call the API to update a user's list with comprehensive feedback."""
# Check authentication (commented code from original)
# if not ctx.media_api.user_profile:
# feedback.warning(
# "You must be logged in to modify your list",
# "Please authenticate with AniList to manage your anime lists"
# )
# return
def update_operation():
return ctx.media_api.update_list_entry(params)
anime_title = anime.title.english or anime.title.romaji
success, result = execute_with_feedback(
update_operation,
feedback,
"update anime list",
loading_msg=f"Updating '{anime_title}' on your list",
success_msg=f"Successfully updated '{anime_title}' on your list!",
error_msg="Failed to update list entry",
show_loading=False,
)

View File

@@ -6,6 +6,7 @@ from rich.progress import Progress
from thefuzz import fuzz
from ....libs.providers.anime.params import SearchParams
from ...utils.feedback import create_feedback_manager, execute_with_feedback
from ..session import Context, session
from ..state import ControlFlow, ProviderState, State
@@ -20,9 +21,10 @@ def provider_search(ctx: Context, state: State) -> State | ControlFlow:
This state allows the user to confirm the correct provider entry before
proceeding to list episodes.
"""
feedback = create_feedback_manager(ctx.config.general.icons)
anilist_anime = state.media_api.anime
if not anilist_anime:
click.echo("[bold red]Error: No AniList anime to search for.[/bold red]")
feedback.error("No AniList anime to search for", "Please select an anime first")
return ControlFlow.BACK
provider = ctx.provider
@@ -33,28 +35,37 @@ def provider_search(ctx: Context, state: State) -> State | ControlFlow:
anilist_title = anilist_anime.title.english or anilist_anime.title.romaji
if not anilist_title:
console.print(
"[bold red]Error: Selected anime has no searchable title.[/bold red]"
feedback.error(
"Selected anime has no searchable title",
"This anime entry is missing required title information",
)
return ControlFlow.BACK
# --- Perform Search on Provider ---
with Progress(transient=True) as progress:
progress.add_task(
f"[cyan]Searching for '{anilist_title}' on {provider.__class__.__name__}...",
total=None,
)
provider_search_results = provider.search(
def search_provider():
return provider.search(
SearchParams(
query=anilist_title, translation_type=config.stream.translation_type
)
)
if not provider_search_results or not provider_search_results.results:
console.print(
f"[bold yellow]Could not find '{anilist_title}' on {provider.__class__.__name__}.[/bold yellow]"
success, provider_search_results = execute_with_feedback(
search_provider,
feedback,
"search provider",
loading_msg=f"Searching for '{anilist_title}' on {provider.__class__.__name__}",
success_msg=f"Found results on {provider.__class__.__name__}",
)
if (
not success
or not provider_search_results
or not provider_search_results.results
):
feedback.warning(
f"Could not find '{anilist_title}' on {provider.__class__.__name__}",
"Try another provider from the config or go back to search again",
)
console.print("Try another provider from the config or go back.")
return ControlFlow.BACK
# --- Map results for selection ---

View File

@@ -78,11 +78,43 @@ class Session:
def _edit_config(self):
"""Handles the logic for editing the config file and reloading the context."""
click.edit(filename=str(USER_CONFIG_PATH))
loader = ConfigLoader()
new_config = loader.load()
self._load_context(new_config)
click.echo("[bold green]Configuration reloaded.[/bold green]")
from ..utils.feedback import create_feedback_manager
feedback = create_feedback_manager(
True
) # Always use icons for session feedback
# Confirm before opening editor
if not feedback.confirm("Open configuration file in editor?", default=True):
return
try:
click.edit(filename=str(USER_CONFIG_PATH))
def reload_config():
loader = ConfigLoader()
new_config = loader.load()
self._load_context(new_config)
return new_config
from ..utils.feedback import execute_with_feedback
success, _ = execute_with_feedback(
reload_config,
feedback,
"reload configuration",
loading_msg="Reloading configuration",
success_msg="Configuration reloaded successfully",
error_msg="Failed to reload configuration",
show_loading=False,
)
if success:
feedback.pause_for_user("Press Enter to continue")
except Exception as e:
feedback.error("Failed to edit configuration", str(e))
feedback.pause_for_user("Press Enter to continue")
def run(self, config: AppConfig, resume_path: Path | None = None):
"""

View File

@@ -0,0 +1,157 @@
"""
User feedback utilities for the interactive CLI.
Provides standardized success, error, warning, and confirmation dialogs.
"""
from contextlib import contextmanager
from typing import Any, Callable, Optional
import click
from rich.console import Console
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn
from rich.prompt import Confirm
console = Console()
class FeedbackManager:
"""Centralized manager for user feedback in interactive menus."""
def __init__(self, icons_enabled: bool = True):
self.icons_enabled = icons_enabled
def success(self, message: str, details: Optional[str] = None) -> None:
"""Show a success message with optional details."""
icon = "" if self.icons_enabled else ""
main_msg = f"[bold green]{icon}{message}[/bold green]"
if details:
console.print(f"{main_msg}\n[dim]{details}[/dim]")
else:
console.print(main_msg)
def error(self, message: str, details: Optional[str] = None) -> None:
"""Show an error message with optional details."""
icon = "" if self.icons_enabled else ""
main_msg = f"[bold red]{icon}Error: {message}[/bold red]"
if details:
console.print(f"{main_msg}\n[dim]{details}[/dim]")
else:
console.print(main_msg)
def warning(self, message: str, details: Optional[str] = None) -> None:
"""Show a warning message with optional details."""
icon = "⚠️ " if self.icons_enabled else ""
main_msg = f"[bold yellow]{icon}Warning: {message}[/bold yellow]"
if details:
console.print(f"{main_msg}\n[dim]{details}[/dim]")
else:
console.print(main_msg)
def info(self, message: str, details: Optional[str] = None) -> None:
"""Show an informational message with optional details."""
icon = " " if self.icons_enabled else ""
main_msg = f"[bold blue]{icon}{message}[/bold blue]"
if details:
console.print(f"{main_msg}\n[dim]{details}[/dim]")
else:
console.print(main_msg)
def confirm(self, message: str, default: bool = False) -> bool:
"""Show a confirmation dialog and return user's choice."""
icon = "" if self.icons_enabled else ""
return Confirm.ask(f"[bold]{icon}{message}[/bold]", default=default)
def notify_operation_result(
self,
operation_name: str,
success: bool,
success_msg: Optional[str] = None,
error_msg: Optional[str] = None,
) -> None:
"""Notify user of operation result with standardized messaging."""
if success:
msg = success_msg or f"{operation_name} completed successfully"
self.success(msg)
else:
msg = error_msg or f"{operation_name} failed"
self.error(msg)
@contextmanager
def loading_operation(
self,
message: str,
success_msg: Optional[str] = None,
error_msg: Optional[str] = None,
):
"""Context manager for operations with loading indicator and result feedback."""
with Progress(
SpinnerColumn(),
TextColumn(f"[cyan]{message}..."),
transient=True,
console=console,
) as progress:
progress.add_task("", total=None)
try:
yield
if success_msg:
self.success(success_msg)
except Exception as e:
error_details = str(e) if str(e) else None
final_error_msg = error_msg or "Operation failed"
self.error(final_error_msg, error_details)
raise
def pause_for_user(self, message: str = "Press Enter to continue") -> None:
"""Pause execution and wait for user input."""
icon = "⏸️ " if self.icons_enabled else ""
click.pause(f"{icon}{message}...")
def show_detailed_panel(
self, title: str, content: str, style: str = "blue"
) -> None:
"""Show detailed information in a styled panel."""
console.print(Panel(content, title=title, border_style=style, expand=True))
self.pause_for_user()
def execute_with_feedback(
operation: Callable[[], Any],
feedback: FeedbackManager,
operation_name: str,
loading_msg: Optional[str] = None,
success_msg: Optional[str] = None,
error_msg: Optional[str] = None,
show_loading: bool = True,
) -> tuple[bool, Any]:
"""
Execute an operation with comprehensive feedback handling.
Returns:
tuple of (success: bool, result: Any)
"""
loading_message = loading_msg or f"Executing {operation_name}"
try:
if show_loading:
with feedback.loading_operation(loading_message, success_msg, error_msg):
result = operation()
return True, result
else:
result = operation()
if success_msg:
feedback.success(success_msg)
return True, result
except Exception as e:
final_error_msg = error_msg or f"{operation_name} failed"
feedback.error(final_error_msg, str(e) if str(e) else None)
return False, None
def create_feedback_manager(icons_enabled: bool = True) -> FeedbackManager:
"""Factory function to create a FeedbackManager instance."""
return FeedbackManager(icons_enabled)

75
test_feedback.py Normal file
View File

@@ -0,0 +1,75 @@
"""
Test script to verify the feedback system works correctly.
Run this to see the feedback system in action.
"""
import sys
import time
from pathlib import Path
# Add the project root to the path so we can import fastanime modules
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))
from fastanime.cli.utils.feedback import create_feedback_manager, execute_with_feedback
def test_feedback_system():
"""Test all feedback system components."""
print("=== Testing FastAnime Enhanced Feedback System ===\n")
# Test with icons enabled
feedback = create_feedback_manager(icons_enabled=True)
print("1. Testing success message:")
feedback.success("Operation completed successfully", "All data has been processed")
time.sleep(1)
print("\n2. Testing error message:")
feedback.error("Failed to connect to server", "Network timeout after 30 seconds")
time.sleep(1)
print("\n3. Testing warning message:")
feedback.warning(
"Anime not found on provider", "Try searching with a different title"
)
time.sleep(1)
print("\n4. Testing info message:")
feedback.info("Loading anime data", "This may take a few moments")
time.sleep(1)
print("\n5. Testing loading operation:")
def mock_long_operation():
time.sleep(2)
return "Operation result"
success, result = execute_with_feedback(
mock_long_operation,
feedback,
"fetch anime data",
loading_msg="Fetching anime from AniList",
success_msg="Anime data loaded successfully",
)
print(f"Operation success: {success}, Result: {result}")
print("\n6. Testing confirmation dialog:")
if feedback.confirm("Do you want to continue with the test?", default=True):
feedback.success("User confirmed to continue")
else:
feedback.info("User chose to stop")
print("\n7. Testing detailed panel:")
feedback.show_detailed_panel(
"Anime Information",
"Title: Attack on Titan\nGenres: Action, Drama\nStatus: Completed\nEpisodes: 25",
"cyan",
)
print("\n=== Test completed! ===")
if __name__ == "__main__":
test_feedback_system()