gitea-mirror/FastAnime
1
0
Fork 0
mirror of https://github.com/Benexl/FastAnime.git synced 2026-01-17 23:25:40 -08:00
Code Issues Packages Projects Releases Wiki Activity

feat: update interactive session logic

Browse Source
This commit is contained in:
Benexl
2025-07-21 22:28:09 +03:00
parent 452c2cf764
commit 0e6aeeea18
16 changed files with 739 additions and 706 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
fastanime/cli/interactive/menus/auth.py
View File

@@ -75,7 +75,9 @@ def auth(ctx: Context, state: State) -> State | ControlFlow:
return ControlFlow.BACK
def _display_auth_status(console: Console, user_profile: Optional[UserProfile], icons: bool):
def _display_auth_status(
console: Console, user_profile: Optional[UserProfile], icons: bool
):
"""Display current authentication status in a nice panel."""
if user_profile:
status_icon = "🟢" if icons else "[green]●[/green]"
@@ -95,37 +97,49 @@ def _display_auth_status(console: Console, user_profile: Optional[UserProfile],
console.print()
def _handle_login(ctx: Context, auth_manager: AuthManager, feedback, icons: bool) -> State | ControlFlow:
def _handle_login(
ctx: Context, auth_manager: AuthManager, feedback, icons: bool
) -> State | ControlFlow:
"""Handle the interactive login process."""
def perform_login():
# Open browser to AniList OAuth page
oauth_url = "https://anilist.co/api/v2/oauth/authorize?client_id=20148&response_type=token"
if feedback.confirm("Open AniList authorization page in browser?", default=True):
if feedback.confirm(
"Open AniList authorization page in browser?", default=True
):
try:
webbrowser.open(oauth_url)
feedback.info("Browser opened", "Complete the authorization process in your browser")
feedback.info(
"Browser opened",
"Complete the authorization process in your browser",
)
except Exception as e:
feedback.warning("Could not open browser automatically", f"Please manually visit: {oauth_url}")
feedback.warning(
"Could not open browser automatically",
f"Please manually visit: {oauth_url}",
)
else:
feedback.info("Manual authorization", f"Please visit: {oauth_url}")
# Get token from user
feedback.info("Token Input", "Paste the token from the browser URL after '#access_token='")
token = ctx.selector.ask(
"Enter your AniList Access Token"
feedback.info(
"Token Input", "Paste the token from the browser URL after '#access_token='"
)
token = ctx.selector.ask("Enter your AniList Access Token")
if not token or not token.strip():
feedback.error("Login cancelled", "No token provided")
return None
# Authenticate with the API
profile = ctx.media_api.authenticate(token.strip())
if not profile:
feedback.error("Authentication failed", "The token may be invalid or expired")
feedback.error(
"Authentication failed", "The token may be invalid or expired"
)
return None
# Save credentials using the auth manager
@@ -137,40 +151,46 @@ def _handle_login(ctx: Context, auth_manager: AuthManager, feedback, icons: bool
feedback,
"authenticate",
loading_msg="Validating token with AniList",
success_msg=f"Successfully logged in! 🎉" if icons else f"Successfully logged in!",
success_msg=f"Successfully logged in! 🎉"
if icons
else f"Successfully logged in!",
error_msg="Login failed",
show_loading=True
show_loading=True,
)
if success and profile:
feedback.success(f"Logged in as {profile.name}" if profile else "Successfully logged in")
feedback.success(
f"Logged in as {profile.name}" if profile else "Successfully logged in"
)
feedback.pause_for_user("Press Enter to continue")
return ControlFlow.CONTINUE
def _handle_logout(ctx: Context, auth_manager: AuthManager, feedback, icons: bool) -> State | ControlFlow:
def _handle_logout(
ctx: Context, auth_manager: AuthManager, feedback, icons: bool
) -> State | ControlFlow:
"""Handle the logout process with confirmation."""
if not feedback.confirm(
"Are you sure you want to logout?",
"Are you sure you want to logout?",
"This will remove your saved AniList token and log you out",
default=False
default=False,
):
return ControlFlow.CONTINUE
def perform_logout():
# Clear from auth manager
if hasattr(auth_manager, 'logout'):
if hasattr(auth_manager, "logout"):
auth_manager.logout()
else:
auth_manager.clear_user_profile()
# Clear from API client
ctx.media_api.token = None
ctx.media_api.user_profile = None
if hasattr(ctx.media_api, 'http_client'):
if hasattr(ctx.media_api, "http_client"):
ctx.media_api.http_client.headers.pop("Authorization", None)
return True
success, _ = execute_with_feedback(
@@ -178,18 +198,22 @@ def _handle_logout(ctx: Context, auth_manager: AuthManager, feedback, icons: boo
feedback,
"logout",
loading_msg="Logging out",
success_msg="Successfully logged out 👋" if icons else "Successfully logged out",
success_msg="Successfully logged out 👋"
if icons
else "Successfully logged out",
error_msg="Logout failed",
show_loading=False
show_loading=False,
)
if success:
feedback.pause_for_user("Press Enter to continue")
return ControlFlow.RELOAD_CONFIG
return ControlFlow.CONFIG_EDIT
def _display_user_profile_details(console: Console, user_profile: UserProfile, icons: bool):
def _display_user_profile_details(
console: Console, user_profile: UserProfile, icons: bool
):
"""Display detailed user profile information."""
if not user_profile:
console.print("[red]No user profile available[/red]")
@@ -202,10 +226,10 @@ def _display_user_profile_details(console: Console, user_profile: UserProfile, i
table.add_row("Name", user_profile.name)
table.add_row("User ID", str(user_profile.id))
if user_profile.avatar_url:
table.add_row("Avatar URL", user_profile.avatar_url)
if user_profile.banner_url:
table.add_row("Banner URL", user_profile.banner_url)
@@ -222,7 +246,7 @@ def _display_user_profile_details(console: Console, user_profile: UserProfile, i
f"{'🔄 ' if icons else '• '}Sync progress with AniList\n"
f"{'🔔 ' if icons else '• '}Access AniList notifications",
title="Available with Authentication",
border_style="green"
border_style="green",
)
console.print(features_panel)
@@ -254,7 +278,7 @@ list management and does not access sensitive account information.
panel = Panel(
help_text,
title=f"{'❓ ' if icons else ''}AniList Token Help",
border_style="blue"
border_style="blue",
)
console.print()
console.print(panel)

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

@@ -5,12 +5,15 @@ from rich.console import Console
from ....libs.api.params import ApiSearchParams, UserListParams
from ....libs.api.types import MediaSearchResult, MediaStatus, UserListStatusType
from ...utils.auth.utils import check_authentication_required, format_auth_menu_header
from ...utils.feedback import create_feedback_manager, execute_with_feedback
from ...utils.auth.utils import format_auth_menu_header, check_authentication_required
from ..session import Context, session
from ..state import ControlFlow, MediaApiState, State
MenuAction = Callable[[], Tuple[str, MediaSearchResult | None, ApiSearchParams | None, UserListParams | None]]
MenuAction = Callable[
[],
Tuple[str, MediaSearchResult | None, ApiSearchParams | None, UserListParams | None],
]
@session.menu
@@ -59,13 +62,33 @@ def main(ctx: Context, state: State) -> State | ControlFlow:
ctx, "REPEATING"
),
# --- List Management ---
f"{'📚 ' if icons else ''}AniList Lists Manager": lambda: ("ANILIST_LISTS", None, None, None),
f"{'📖 ' if icons else ''}Local Watch History": lambda: ("WATCH_HISTORY", None, None, None),
f"{'📚 ' if icons else ''}AniList Lists Manager": lambda: (
"ANILIST_LISTS",
None,
None,
None,
),
f"{'📖 ' if icons else ''}Local Watch History": lambda: (
"WATCH_HISTORY",
None,
None,
None,
),
# --- Authentication and Account Management ---
f"{'🔐 ' if icons else ''}Authentication": lambda: ("AUTH", None, None, None),
# --- Control Flow and Utility Options ---
f"{'🔧 ' if icons else ''}Session Management": lambda: ("SESSION_MANAGEMENT", None, None, None),
f"{'📝 ' if icons else ''}Edit Config": lambda: ("RELOAD_CONFIG", None, None, None),
f"{'🔧 ' if icons else ''}Session Management": lambda: (
"SESSION_MANAGEMENT",
None,
None,
None,
),
f"{'📝 ' if icons else ''}Edit Config": lambda: (
"RELOAD_CONFIG",
None,
None,
None,
),
f"{'❌ ' if icons else ''}Exit": lambda: ("EXIT", None, None, None),
}
@@ -86,7 +109,7 @@ def main(ctx: Context, state: State) -> State | ControlFlow:
if next_menu_name == "EXIT":
return ControlFlow.EXIT
if next_menu_name == "RELOAD_CONFIG":
return ControlFlow.RELOAD_CONFIG
return ControlFlow.CONFIG_EDIT
if next_menu_name == "SESSION_MANAGEMENT":
return State(menu_name="SESSION_MANAGEMENT")
if next_menu_name == "AUTH":
@@ -141,7 +164,11 @@ def _create_media_list_action(
)
# Return the search parameters along with the result for pagination
return ("RESULTS", result, search_params, None) if success else ("CONTINUE", None, None, None)
return (
("RESULTS", result, search_params, None)
if success
else ("CONTINUE", None, None, None)
)
return action
@@ -168,7 +195,11 @@ def _create_random_media_list(ctx: Context) -> MenuAction:
)
# Return the search parameters along with the result for pagination
return ("RESULTS", result, search_params, None) if success else ("CONTINUE", None, None, None)
return (
("RESULTS", result, search_params, None)
if success
else ("CONTINUE", None, None, None)
)
return action
@@ -196,7 +227,11 @@ def _create_search_media_list(ctx: Context) -> MenuAction:
)
# Return the search parameters along with the result for pagination
return ("RESULTS", result, search_params, None) if success else ("CONTINUE", None, None, None)
return (
("RESULTS", result, search_params, None)
if success
else ("CONTINUE", None, None, None)
)
return action
@@ -214,7 +249,9 @@ def _create_user_list_action(ctx: Context, status: UserListStatusType) -> MenuAc
return "CONTINUE", None, None, None
# Create the user list parameters
user_list_params = UserListParams(status=status, per_page=ctx.config.anilist.per_page)
user_list_params = UserListParams(
status=status, per_page=ctx.config.anilist.per_page
)
def fetch_data():
return ctx.media_api.fetch_user_list(user_list_params)
@@ -228,6 +265,10 @@ def _create_user_list_action(ctx: Context, status: UserListStatusType) -> MenuAc
)
# Return the user list parameters along with the result for pagination
return ("RESULTS", result, None, user_list_params) if success else ("CONTINUE", None, None, None)
return (
("RESULTS", result, None, user_list_params)
if success
else ("CONTINUE", None, None, None)
)
return action

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

@@ -2,20 +2,27 @@ import importlib.util
import logging
import os
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Callable, List
from typing import Callable, List, Optional
import click
from ...core.config import AppConfig
from ...core.constants import APP_DIR, USER_CONFIG_PATH
from ...libs.api.base import BaseApiClient
from ...libs.api.factory import create_api_client
from ...libs.players import create_player
from ...libs.players.base import BasePlayer
from ...libs.providers.anime.base import BaseAnimeProvider
from ...libs.providers.anime.provider import create_provider
from ...libs.selectors import create_selector
from ...libs.selectors.base import BaseSelector
from ..config import ConfigLoader
from ..utils.session.manager import SessionManager
from ..services.auth import AuthService
from ..services.feedback import FeedbackService
from ..services.registry import MediaRegistryService
from ..services.session import SessionsService
from ..services.watch_history import WatchHistoryService
from .state import ControlFlow, State
logger = logging.getLogger(__name__)
@@ -27,55 +34,53 @@ MENUS_DIR = APP_DIR / "cli" / "interactive" / "menus"
@dataclass(frozen=True)
class Context:
"""
A mutable container for long-lived, shared services and configurations.
This object is passed to every menu state, providing access to essential
application components like API clients and UI selectors.
"""
class Services:
feedback: FeedbackService
media_registry: MediaRegistryService
watch_history: WatchHistoryService
session: SessionsService
auth: AuthService
@dataclass(frozen=True)
class Context:
config: AppConfig
provider: BaseAnimeProvider
selector: BaseSelector
player: BasePlayer
media_api: BaseApiClient
services: Services
@dataclass(frozen=True)
class Menu:
"""Represents a registered menu, linking a name to an executable function."""
name: str
execute: MenuFunction
class Session:
"""
The orchestrator for the interactive UI state machine.
This class manages the state history, holds the application context,
runs the main event loop, and provides the decorator for registering menus.
"""
def __init__(self):
self._context: Context | None = None
self._history: List[State] = []
self._menus: dict[str, Menu] = {}
self._session_manager = SessionManager()
self._auto_save_enabled = True
_context: Context
_history: List[State] = []
_menus: dict[str, Menu] = {}
def _load_context(self, config: AppConfig):
"""Initializes all shared services based on the provided configuration."""
from ...libs.api.factory import create_api_client
from ...libs.players import create_player
from ...libs.providers.anime.provider import create_provider
from ...libs.selectors import create_selector
media_registry = MediaRegistryService(
media_api=config.general.media_api, config=config.media_registry
)
auth = AuthService(config.general.media_api)
services = Services(
feedback=FeedbackService(config.general.icons),
media_registry=media_registry,
watch_history=WatchHistoryService(config, media_registry),
session=SessionsService(config.sessions),
auth=auth,
)
# Create API client
media_api = create_api_client(config.general.api_client, config)
media_api = create_api_client(config.general.media_api, config)
# Attempt to load saved user authentication
self._load_saved_authentication(media_api)
if auth_profile := auth.get_auth():
media_api.authenticate(auth_profile.token)
self._context = Context(
config=config,
@@ -83,267 +88,66 @@ class Session:
selector=create_selector(config),
player=create_player(config),
media_api=media_api,
services=services,
)
logger.info("Application context reloaded.")
def _load_saved_authentication(self, media_api):
"""Attempt to load saved user authentication."""
try:
from ..auth.manager import AuthManager
auth_manager = AuthManager()
user_data = auth_manager.load_user_profile()
if user_data and user_data.get("token"):
# Try to authenticate with the saved token
profile = media_api.authenticate(user_data["token"])
if profile:
logger.info(f"Successfully authenticated as {profile.name}")
else:
logger.warning("Saved authentication token is invalid or expired")
else:
logger.debug("No saved authentication found")
except Exception as e:
logger.error(f"Failed to load saved authentication: {e}")
# Continue without authentication rather than failing completely
def _edit_config(self):
"""Handles the logic for editing the config file and reloading the context."""
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):
"""
Starts and manages the main interactive session loop.
Args:
config: The initial application configuration.
resume_path: Optional path to a saved session file to resume from.
"""
from ..utils.feedback import create_feedback_manager
feedback = create_feedback_manager(True) # Always use icons for session messages
click.edit(filename=str(USER_CONFIG_PATH))
logger.debug(f"Config changed; Reloading context")
loader = ConfigLoader()
config = loader.load()
self._load_context(config)
# Handle session recovery
if resume_path:
self.resume(resume_path, feedback)
elif self._session_manager.has_crash_backup():
# Offer to resume from crash backup
if feedback.confirm(
"Found a crash backup from a previous session. Would you like to resume?",
default=True
def run(
self,
config: AppConfig,
resume: bool = False,
history: Optional[List[State]] = None,
):
self._load_context(config)
if resume:
if (
history
:= self._context.services.session.get_most_recent_session_history()
):
crash_history = self._session_manager.load_crash_backup(feedback)
if crash_history:
self._history = crash_history
feedback.info("Session restored from crash backup")
# Clear the crash backup after successful recovery
self._session_manager.clear_crash_backup()
elif self._session_manager.has_auto_save():
# Offer to resume from auto-save
if feedback.confirm(
"Found an auto-saved session. Would you like to resume?",
default=False
):
auto_history = self._session_manager.load_auto_save(feedback)
if auto_history:
self._history = auto_history
feedback.info("Session restored from auto-save")
self._history = history
else:
logger.warning("Failed to continue from history. No sessions found")
# Start with main menu if no history
if not self._history:
self._history.append(State(menu_name="MAIN"))
# Create crash backup before starting
if self._auto_save_enabled:
self._session_manager.create_crash_backup(self._history)
try:
self._run_main_loop()
except KeyboardInterrupt:
feedback.warning("Session interrupted by user")
self._handle_session_exit(feedback, interrupted=True)
except Exception as e:
feedback.error("Session crashed unexpectedly", str(e))
self._handle_session_exit(feedback, crashed=True)
self._context.services.session.save_session(self._history)
raise
else:
self._handle_session_exit(feedback, normal_exit=True)
self._context.services.session.save_session(self._history)
def _run_main_loop(self):
"""Run the main session loop."""
while self._history:
current_state = self._history[-1]
menu_to_run = self._menus.get(current_state.menu_name)
if not menu_to_run or not self._context:
logger.error(
f"Menu '{current_state.menu_name}' not found or context not loaded."
)
break
# Auto-save periodically (every 5 state changes)
if self._auto_save_enabled and len(self._history) % 5 == 0:
self._session_manager.auto_save_session(self._history)
# Execute the menu function, which returns the next step.
next_step = menu_to_run.execute(self._context, current_state)
next_step = self._menus[current_state.menu_name].execute(
self._context, current_state
)
if isinstance(next_step, ControlFlow):
# A control command was issued.
if next_step == ControlFlow.EXIT:
break # Exit the loop
break
elif next_step == ControlFlow.BACK:
if len(self._history) > 1:
self._history.pop() # Go back one state
elif next_step == ControlFlow.RELOAD_CONFIG:
self._history.pop()
elif next_step == ControlFlow.CONFIG_EDIT:
self._edit_config()
# For CONTINUE, we do nothing, allowing the loop to re-run the current state.
elif isinstance(next_step, State):
else:
# if the state is main menu we should reset the history
if next_step.menu_name == "MAIN":
self._history = [next_step]
else:
# A new state was returned, push it to history for the next loop.
self._history.append(next_step)
else:
logger.error(
f"Menu '{current_state.menu_name}' returned invalid type: {type(next_step)}"
)
break
def _handle_session_exit(self, feedback, normal_exit=False, interrupted=False, crashed=False):
"""Handle session cleanup on exit."""
if self._auto_save_enabled and self._history:
if normal_exit:
# Clear auto-save on normal exit
self._session_manager.clear_auto_save()
self._session_manager.clear_crash_backup()
feedback.info("Session completed normally")
elif interrupted:
# Save session on interruption
self._session_manager.auto_save_session(self._history)
feedback.info("Session auto-saved due to interruption")
elif crashed:
# Keep crash backup on crash
feedback.error("Session backup maintained for recovery")
click.echo("Exiting interactive session.")
def save(self, file_path: Path, session_name: str = None, description: str = None):
"""
Save session history to a file with comprehensive metadata and error handling.
Args:
file_path: Path to save the session
session_name: Optional name for the session
description: Optional description for the session
"""
from ..utils.feedback import create_feedback_manager
feedback = create_feedback_manager(True)
return self._session_manager.save_session(
self._history,
file_path,
session_name=session_name,
description=description,
feedback=feedback
)
def resume(self, file_path: Path, feedback=None):
"""
Load session history from a file with comprehensive error handling.
Args:
file_path: Path to the session file
feedback: Optional feedback manager for user notifications
"""
if not feedback:
from ..utils.feedback import create_feedback_manager
feedback = create_feedback_manager(True)
history = self._session_manager.load_session(file_path, feedback)
if history:
self._history = history
return True
return False
def list_saved_sessions(self):
"""List all saved sessions with their metadata."""
return self._session_manager.list_saved_sessions()
def cleanup_old_sessions(self, max_sessions: int = 10):
"""Clean up old session files, keeping only the most recent ones."""
return self._session_manager.cleanup_old_sessions(max_sessions)
def enable_auto_save(self, enabled: bool = True):
"""Enable or disable auto-save functionality."""
self._auto_save_enabled = enabled
def get_session_stats(self) -> dict:
"""Get statistics about the current session."""
return {
"current_states": len(self._history),
"current_menu": self._history[-1].menu_name if self._history else None,
"auto_save_enabled": self._auto_save_enabled,
"has_auto_save": self._session_manager.has_auto_save(),
"has_crash_backup": self._session_manager.has_crash_backup()
}
def create_manual_backup(self, backup_name: str = None):
"""Create a manual backup of the current session."""
from ..utils.feedback import create_feedback_manager
from ...core.constants import APP_DIR
feedback = create_feedback_manager(True)
backup_name = backup_name or f"manual_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
backup_path = APP_DIR / "sessions" / f"{backup_name}.json"
return self._session_manager.save_session(
self._history,
backup_path,
session_name=backup_name,
description="Manual backup created by user",
feedback=feedback
)
@property
def menu(self) -> Callable[[MenuFunction], MenuFunction]:

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

@@ -3,13 +3,13 @@ from typing import Iterator, List, Literal, Optional
from pydantic import BaseModel, ConfigDict
from ...libs.api.params import ApiSearchParams, UserListParams # Add this import
from ...libs.api.types import (
MediaItem,
MediaSearchResult,
MediaStatus,
UserListStatusType,
)
from ...libs.api.params import ApiSearchParams, UserListParams # Add this import
from ...libs.players.types import PlayerResult
from ...libs.providers.anime.types import Anime, SearchResults, Server
@@ -27,7 +27,7 @@ class ControlFlow(Enum):
EXIT = auto()
"""Terminate the interactive session gracefully."""
RELOAD_CONFIG = auto()
CONFIG_EDIT = auto()
"""Reload the application configuration and re-initialize the context."""
CONTINUE = auto()
@@ -77,7 +77,7 @@ class MediaApiState(BaseModel):
user_media_status: Optional[UserListStatusType] = None
media_status: Optional[MediaStatus] = None
anime: Optional[MediaItem] = None
# Add pagination support: store original search parameters to enable page navigation
original_api_params: Optional[ApiSearchParams] = None
original_user_list_params: Optional[UserListParams] = None