FastAnime/fastanime/cli/interactive/session.py

import importlib.util
import logging
import os
from dataclasses import dataclass
from typing import Callable, List, Optional, Union

import click

from ...core.config import AppConfig
from ...core.constants import APP_DIR, USER_CONFIG_PATH
from ...libs.media_api.base import BaseApiClient
from ...libs.player.base import BasePlayer
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.registry import MediaRegistryService
from ..service.session import SessionsService
from ..service.watch_history import WatchHistoryService
from .state import InternalDirective, MenuName, State

logger = logging.getLogger(__name__)

# A type alias for the signature all menu functions must follow.

MENUS_DIR = APP_DIR / "cli" / "interactive" / "menu"


@dataclass(frozen=True)
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
    service: Services


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):
        """Initializes all shared service based on the provided configuration."""
        from ...libs.media_api.api import create_api_client
        from ...libs.player import create_player
        from ...libs.provider.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,
        )

        media_api = create_api_client(config.general.media_api, config)

        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._context = Context(
            config=config,
            provider=create_provider(config.general.provider),
            selector=create_selector(config),
            player=create_player(config),
            media_api=media_api,
            service=services,
        )
        logger.info("Application context reloaded.")

    def _edit_config(self):
        from ..config import ConfigLoader

        click.edit(filename=str(USER_CONFIG_PATH))
        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.service.session.get_most_recent_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.service.session.create_crash_backup(self._history)
            raise
        self._context.service.session.save_session(self._history)

    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]]
                if 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.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"fastanime.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()