gitea-mirror/FastAnime
1
0
Fork 0
mirror of https://github.com/Benexl/FastAnime.git synced 2026-02-04 11:07:48 -08:00
Code Issues Packages Projects Releases Wiki Activity

feat(interactive-state): create accessors that ensure values exist

Browse Source
This commit is contained in:
Benexl
2025-08-17 19:38:55 +03:00
parent ab4734b79d
commit 449f6c1e59
8 changed files with 116 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
viu_cli/cli/interactive/menu/media/episodes.py
View File

@@ -72,6 +72,6 @@ def episodes(ctx: Context, state: State) -> State | InternalDirective:
menu_name=MenuName.SERVERS,
media_api=state.media_api,
provider=state.provider.model_copy(
update={"episode": chosen_episode, "start_time": start_time}
update={"episode_": chosen_episode, "start_time_": start_time}
),
)

13
viu_cli/cli/interactive/menu/media/play_downloads.py
View File

@@ -15,9 +15,6 @@ def play_downloads(ctx: Context, state: State) -> State | InternalDirective:
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:
@@ -65,9 +62,7 @@ def play_downloads(ctx: Context, state: State) -> State | InternalDirective:
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
)
@@ -84,7 +79,7 @@ def play_downloads(ctx: Context, state: State) -> State | InternalDirective:
menu_name=MenuName.DOWNLOADS_PLAYER_CONTROLS,
media_api=state.media_api,
provider=state.provider.model_copy(
update={"episode": chosen_episode, "start_time": start_time}
update={"episode_": chosen_episode, "start_time_": start_time}
),
)
@@ -155,7 +150,7 @@ def downloads_player_controls(
menu_name=MenuName.DOWNLOADS_PLAYER_CONTROLS,
media_api=state.media_api,
provider=state.provider.model_copy(
update={"episode": next_episode_num, "start_time": None}
update={"episode_": next_episode_num, "start_time_": None}
),
)
@@ -230,7 +225,7 @@ def _next_episode(ctx: Context, state: State) -> MenuAction:
menu_name=MenuName.DOWNLOADS_PLAYER_CONTROLS,
media_api=state.media_api,
provider=state.provider.model_copy(
update={"episode": next_episode_num, "start_time": None}
update={"episode_": next_episode_num, "start_time_": None}
),
)
feedback.warning("This is the last available episode.")
@@ -279,7 +274,7 @@ def _previous_episode(ctx: Context, state: State) -> MenuAction:
menu_name=MenuName.DOWNLOADS_PLAYER_CONTROLS,
media_api=state.media_api,
provider=state.provider.model_copy(
update={"episode": prev_episode_num, "start_time": None}
update={"episode_": prev_episode_num, "start_time_": None}
),
)
feedback.warning("This is the last available episode.")

6
viu_cli/cli/interactive/menu/media/player_controls.py
View File

@@ -42,7 +42,7 @@ def player_controls(ctx: Context, state: State) -> Union[State, InternalDirectiv
return State(
menu_name=MenuName.SERVERS,
media_api=state.media_api,
provider=state.provider.model_copy(update={"episode": next_episode_num}),
provider=state.provider.model_copy(update={"episode_": next_episode_num}),
)
# --- Menu Options ---
@@ -116,7 +116,7 @@ def _next_episode(ctx: Context, state: State) -> MenuAction:
menu_name=MenuName.SERVERS,
media_api=state.media_api,
provider=state.provider.model_copy(
update={"episode": next_episode_num}
update={"episode_": next_episode_num}
),
)
feedback.warning("This is the last available episode.")
@@ -150,7 +150,7 @@ def _previous_episode(ctx: Context, state: State) -> MenuAction:
menu_name=MenuName.SERVERS,
media_api=state.media_api,
provider=state.provider.model_copy(
update={"episode": prev_episode_num}
update={"episode_": prev_episode_num}
),
)
feedback.warning("This is the last available episode.")

4
viu_cli/cli/interactive/menu/media/provider_search.py
View File

@@ -7,13 +7,11 @@ from ...state import InternalDirective, MenuName, ProviderState, State
@session.menu
def provider_search(ctx: Context, state: State) -> State | InternalDirective:
from viu_cli.cli.utils.search import find_best_match_title
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

6
viu_cli/cli/interactive/menu/media/results.py
View File

@@ -13,11 +13,7 @@ def results(ctx: Context, state: State) -> State | InternalDirective:
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
page_info = state.media_api.page_info_
search_result_dict = {
_format_title(ctx, media_item): media_item

6
viu_cli/cli/interactive/menu/media/servers.py
View File

@@ -18,8 +18,6 @@ def servers(ctx: Context, state: State) -> State | InternalDirective:
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
@@ -106,8 +104,8 @@ def servers(ctx: Context, state: State) -> State | InternalDirective:
media_api=state.media_api,
provider=state.provider.model_copy(
update={
"servers": server_map,
"server_name": selected_server.name,
"servers_": server_map,
"server_name_": selected_server.name,
}
),
)

141
viu_cli/cli/interactive/state.py
View File

@@ -1,32 +1,15 @@
from enum import Enum
from typing import Dict, Optional, Union
from __future__ import annotations
from pydantic import BaseModel, ConfigDict, Field
from enum import Enum
from typing import Any, Dict, Mapping, Optional, Union
from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, computed_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"
@@ -49,34 +32,116 @@ class MenuName(Enum):
DOWNLOAD_EPISODES = "DOWNLOAD_EPISODES"
class InternalDirective(Enum):
MAIN = "MAIN"
BACK = "BACK"
BACKX2 = "BACKX2"
BACKX3 = "BACKX3"
BACKX4 = "BACKX4"
EXIT = "EXIT"
CONFIG_EDIT = "CONFIG_EDIT"
RELOAD = "RELOAD"
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
search_result_: Optional[Dict[int, MediaItem]] = Field(
default=None, alias="search_result"
)
search_params_: Optional[Union[MediaSearchParams, UserMediaListSearchParams]] = (
Field(default=None, alias="search_params")
)
page_info_: Optional[PageInfo] = Field(default=None, alias="page_info")
media_id_: Optional[int] = Field(default=None, alias="media_id")
@property
def media_item(self) -> Optional[MediaItem]:
if self.search_result and self.media_id:
return self.search_result[self.media_id]
def search_result(self) -> dict[int, MediaItem]:
if not self.search_result_:
raise RuntimeError("Malformed state, please report")
return self.search_result_
@property
def search_params(self) -> Union[MediaSearchParams, UserMediaListSearchParams]:
if not self.search_params_:
raise RuntimeError("Malformed state, please report")
return self.search_params_
@property
def page_info(self) -> PageInfo | None:
# if not self._page_info:
# raise RuntimeError("Malformed state, please report")
return self.page_info_
@property
def media_id(self) -> int:
if not self.media_id_:
raise RuntimeError("Malformed state, please report")
return self.media_id_
@property
def media_item(self) -> MediaItem:
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
search_results_: Optional[SearchResults] = Field(
default=None, alias="search_results"
)
anime_: Optional[Anime] = Field(default=None, alias="anime")
episode_: Optional[str] = Field(default=None, alias="episode")
servers_: Optional[Dict[str, Server]] = Field(default=None, alias="servers")
server_name_: Optional[str] = Field(default=None, alias="server_name")
start_time_: Optional[str] = Field(default=None, alias="start_time")
@property
def server(self) -> Optional[Server]:
if self.servers and self.server_name:
return self.servers[self.server_name]
def search_results(self) -> SearchResults:
if not self.search_results_:
raise RuntimeError("Malformed state, please report")
return self.search_results_
@property
def anime(self) -> Anime:
if not self.anime_:
raise RuntimeError("Malformed state, please report")
return self.anime_
@property
def episode(self) -> str | None:
# if not self._episode:
# raise RuntimeError("Malformed state, please report")
return self.episode_
@property
def servers(self) -> Dict[str, Server]:
if not self.servers_:
raise RuntimeError("Malformed state, please report")
return self.servers_
@property
def server_name(self) -> str:
if not self.server_name_:
raise RuntimeError("Malformed state, please report")
return self.server_name_
@property
def start_time(self) -> str | None:
# if not self._start_time:
# raise RuntimeError("Malformed state, please report")
return self.start_time_
@property
def server(self) -> Server:
return self.servers[self.server_name]
class State(StateModel):