mirror of
https://github.com/Benexl/FastAnime.git
synced 2025-12-24 20:10:11 -08:00
395 lines
15 KiB
Python
395 lines
15 KiB
Python
import importlib.util
|
|
import logging
|
|
import os
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Callable, List
|
|
|
|
import click
|
|
|
|
from ...core.config import AppConfig
|
|
from ...core.constants import APP_DIR, USER_CONFIG_PATH
|
|
from ...libs.api.base import BaseApiClient
|
|
from ...libs.players.base import BasePlayer
|
|
from ...libs.providers.anime.base import BaseAnimeProvider
|
|
from ...libs.selectors.base import BaseSelector
|
|
from ..config import ConfigLoader
|
|
from ..utils.session_manager import SessionManager
|
|
from .state import ControlFlow, State
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# A type alias for the signature all menu functions must follow.
|
|
MenuFunction = Callable[["Context", State], "State | ControlFlow"]
|
|
|
|
MENUS_DIR = APP_DIR / "cli" / "interactive" / "menus"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Context:
|
|
"""
|
|
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.
|
|
"""
|
|
|
|
config: AppConfig
|
|
provider: BaseAnimeProvider
|
|
selector: BaseSelector
|
|
player: BasePlayer
|
|
media_api: BaseApiClient
|
|
|
|
|
|
@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
|
|
|
|
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
|
|
|
|
# Create API client
|
|
media_api = create_api_client(config.general.api_client, config)
|
|
|
|
# Attempt to load saved user authentication
|
|
self._load_saved_authentication(media_api)
|
|
|
|
self._context = Context(
|
|
config=config,
|
|
provider=create_provider(config.general.provider),
|
|
selector=create_selector(config),
|
|
player=create_player(config),
|
|
media_api=media_api,
|
|
)
|
|
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
|
|
|
|
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
|
|
):
|
|
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")
|
|
|
|
# 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)
|
|
raise
|
|
else:
|
|
self._handle_session_exit(feedback, normal_exit=True)
|
|
|
|
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)
|
|
|
|
if isinstance(next_step, ControlFlow):
|
|
# A control command was issued.
|
|
if next_step == ControlFlow.EXIT:
|
|
break # Exit the loop
|
|
elif next_step == ControlFlow.BACK:
|
|
if len(self._history) > 1:
|
|
self._history.pop() # Go back one state
|
|
elif next_step == ControlFlow.RELOAD_CONFIG:
|
|
self._edit_config()
|
|
# For CONTINUE, we do nothing, allowing the loop to re-run the current state.
|
|
elif isinstance(next_step, State):
|
|
# if the state is main menu we should reset the history
|
|
if next_step.menu_name == "MAIN":
|
|
self._history = [next_step]
|
|
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]:
|
|
"""A decorator to register a function as a menu."""
|
|
|
|
def decorator(func: MenuFunction) -> MenuFunction:
|
|
menu_name = 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_path: Path = MENUS_DIR):
|
|
"""
|
|
Dynamically imports all Python modules from a folder to register their menus.
|
|
|
|
Args:
|
|
package_path: The filesystem path to the 'menus' package directory.
|
|
"""
|
|
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"fastanime.cli.interactive.{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()
|