feat: improve provider types

This commit is contained in:
Benexl
2025-07-23 20:02:25 +03:00
parent d78b62fcee
commit 2067467134
14 changed files with 152 additions and 108 deletions

View File

@@ -106,7 +106,6 @@ def download(config: AppConfig, **options: "Unpack[Options]"):
from rich.progress import Progress
from ...core.exceptions import FastAnimeError
from ...libs.players.player import create_player
from ...libs.providers.anime.params import (
AnimeParams,
SearchParams,
@@ -248,7 +247,7 @@ def download_anime(
servers = {server.name: server for server in streams}
servers_names = list(servers.keys())
if config.stream.server in servers_names:
server = servers[config.stream.server]
server = servers[config.stream.server.value]
else:
server_name = selector.choose("Select Server", servers_names)
if not server_name:

View File

@@ -177,7 +177,7 @@ def stream_anime(
servers = {server.name: server for server in streams}
servers_names = list(servers.keys())
if config.stream.server in servers_names:
server = servers[config.stream.server]
server = servers[config.stream.server.value]
else:
server_name = selector.choose("Select Server", servers_names)
if not server_name:

View File

@@ -28,7 +28,7 @@ def servers(ctx: Context, state: State) -> State | ControlFlow:
if not state.media_api.anime:
return ControlFlow.BACK
anime_title = (
state.media_api.anime.title.romaji or state.media_api.anime.title.romaji
state.media_api.anime.title.romaji or state.media_api.anime.title.english
)
episode_number = state.provider.episode_number
config = ctx.config
@@ -70,7 +70,7 @@ def servers(ctx: Context, state: State) -> State | ControlFlow:
server_map: Dict[str, Server] = {s.name: s for s in all_servers}
selected_server: Server | None = None
preferred_server = config.stream.server.lower()
preferred_server = config.stream.server.value.lower()
if preferred_server == "top":
selected_server = all_servers[0]
console.print(f"[cyan]Auto-selecting top server:[/] {selected_server.name}")

View File

@@ -1,7 +1,7 @@
from collections.abc import Callable
from enum import Enum
from pathlib import Path
from typing import Any, Literal, get_args, get_origin
from typing import Any, Literal, Optional, get_args, get_origin
import click
from pydantic import BaseModel
@@ -25,8 +25,8 @@ class ConfigOption(click.Option):
This is used to ensure that options can be generated dynamically from Pydantic models.
"""
model_name: str | None
field_name: str | None
model_name: Optional[str]
field_name: Optional[str]
def __init__(self, *args, **kwargs):
self.model_name = kwargs.pop("model_name", None)
@@ -71,7 +71,13 @@ def options_from_model(model: type[BaseModel], parent_name: str = "") -> Callabl
"help": field_info.description or "",
}
if field_info.annotation is bool:
if (
field_info.annotation is not None
and isinstance(field_info.annotation, type)
and issubclass(field_info.annotation, Enum)
):
kwargs["default"] = field_info.default.value
elif field_info.annotation is bool:
if field_info.default is not PydanticUndefined:
kwargs["default"] = field_info.default
kwargs["show_default"] = True

View File

@@ -2,7 +2,7 @@ import os
from pathlib import Path
from typing import Literal
from pydantic import BaseModel, Field, PrivateAttr, computed_field, field_validator
from pydantic import BaseModel, Field, PrivateAttr, computed_field
from ...core.constants import (
FZF_DEFAULT_OPTS,
@@ -12,7 +12,7 @@ from ...core.constants import (
ROFI_THEME_PREVIEW,
)
from ...libs.api.types import MediaSort, UserMediaListSort
from ...libs.providers.anime import PROVIDERS_AVAILABLE, SERVERS_AVAILABLE
from ...libs.providers.anime.types import ProviderName, ProviderServer
from ..constants import APP_ASCII_ART
from . import defaults
from . import descriptions as desc
@@ -28,13 +28,13 @@ class GeneralConfig(BaseModel):
default=defaults.GENERAL_API_CLIENT,
description=desc.GENERAL_API_CLIENT,
)
provider: str = Field(
default=defaults.GENERAL_PROVIDER,
provider: ProviderName = Field(
default=ProviderName.ALLANIME,
description=desc.GENERAL_PROVIDER,
examples=list(PROVIDERS_AVAILABLE.keys()),
)
selector: Literal["default", "fzf", "rofi"] = Field(
default=defaults.GENERAL_SELECTOR, description=desc.GENERAL_SELECTOR
default=defaults.GENERAL_SELECTOR,
description=desc.GENERAL_SELECTOR,
)
auto_select_anime_result: bool = Field(
default=defaults.GENERAL_AUTO_SELECT_ANIME_RESULT,
@@ -42,7 +42,8 @@ class GeneralConfig(BaseModel):
)
icons: bool = Field(default=defaults.GENERAL_ICONS, description=desc.GENERAL_ICONS)
preview: Literal["full", "text", "image", "none"] = Field(
default=defaults.GENERAL_PREVIEW, description=desc.GENERAL_PREVIEW
default=defaults.GENERAL_PREVIEW,
description=desc.GENERAL_PREVIEW,
)
image_renderer: Literal["icat", "chafa", "imgcat"] = Field(
default="icat"
@@ -80,33 +81,25 @@ class GeneralConfig(BaseModel):
description=desc.GENERAL_RECENT,
)
@field_validator("provider")
@classmethod
def validate_provider(cls, v: str) -> str:
if v not in PROVIDERS_AVAILABLE:
raise ValueError(
f"'{v}' is not a valid provider. Must be one of: {PROVIDERS_AVAILABLE}"
)
return v
class StreamConfig(BaseModel):
"""Configuration specific to video streaming and playback."""
player: Literal["mpv", "vlc"] = Field(
default=defaults.STREAM_PLAYER, description=desc.STREAM_PLAYER
default=defaults.STREAM_PLAYER,
description=desc.STREAM_PLAYER,
)
quality: Literal["360", "480", "720", "1080"] = Field(
default=defaults.STREAM_QUALITY, description=desc.STREAM_QUALITY
default=defaults.STREAM_QUALITY,
description=desc.STREAM_QUALITY,
)
translation_type: Literal["sub", "dub"] = Field(
default=defaults.STREAM_TRANSLATION_TYPE,
description=desc.STREAM_TRANSLATION_TYPE,
)
server: str = Field(
default=defaults.STREAM_SERVER,
server: ProviderServer = Field(
default=ProviderServer.TOP,
description=desc.STREAM_SERVER,
examples=SERVERS_AVAILABLE,
)
auto_next: bool = Field(
default=defaults.STREAM_AUTO_NEXT,
@@ -147,15 +140,6 @@ class StreamConfig(BaseModel):
description=desc.STREAM_SUB_LANG,
)
@field_validator("server")
@classmethod
def validate_server(cls, v: str) -> str:
if v.lower() != "top" and v not in SERVERS_AVAILABLE:
raise ValueError(
f"'{v}' is not a valid server. Must be 'top' or one of: {SERVERS_AVAILABLE}"
)
return v
class ServiceConfig(BaseModel):
"""Configuration for the background download service."""

View File

@@ -1,3 +0,0 @@
from .anime import BaseAnimeProvider
__all__ = ["BaseAnimeProvider"]

View File

@@ -1,3 +0,0 @@
from .provider import PROVIDERS_AVAILABLE, SERVERS_AVAILABLE, BaseAnimeProvider
__all__ = ["SERVERS_AVAILABLE", "PROVIDERS_AVAILABLE", "BaseAnimeProvider"]

View File

@@ -1,11 +1,27 @@
from typing import Union
from httpx import Response
from ..types import Anime, AnimeEpisodes, PageInfo, SearchResult, SearchResults
from ..types import (
Anime,
AnimeEpisodes,
MediaTranslationType,
PageInfo,
SearchResult,
SearchResults,
)
from .types import AllAnimeSearchResults, AllAnimeShow
def generate_list(count: int) -> list[str]:
return list(map(str, range(count)))
def generate_list(count: Union[int, str]) -> list[str]:
return list(map(str, range(int(count))))
translation_type_map = {
"sub": MediaTranslationType.SUB,
"dub": MediaTranslationType.DUB,
"raw": MediaTranslationType.RAW,
}
def map_to_search_results(response: Response) -> SearchResults:

View File

@@ -1,6 +1,17 @@
from enum import Enum
from typing import Literal, TypedDict
class Server(Enum):
SHAREPOINT = "sharepoint"
DROPBOX = "dropbox"
GOGOANIME = "gogoanime"
WETRANSFER = "weTransfer"
WIXMP = "wixmp"
YT = "Yt"
MP4_UPLOAD = "mp4-upload"
class AllAnimeEpisodesDetail(TypedDict):
dub: list[str]
sub: list[str]
@@ -27,7 +38,7 @@ class AllAnimeShow(TypedDict):
class AllAnimeSearchResult(TypedDict):
_id: str
name: str
availableEpisodes: AllAnimeEpisodesDetail
availableEpisodes: AllAnimeEpisodes
__typename: str | None

View File

@@ -5,22 +5,23 @@ from ..types import (
AnimeEpisodeInfo,
AnimeEpisodes,
EpisodeStream,
MediaTranslationType,
PageInfo,
SearchResult,
SearchResults,
Server,
Subtitle,
)
from .types import (
AnimePaheAnime,
AnimePaheAnimePage,
AnimePaheEpisodeInfo,
AnimePaheSearchPage,
AnimePaheSearchResult,
AnimePaheServer,
AnimePaheStreamLink,
)
translation_type_map = {
"sub": MediaTranslationType.SUB,
"dub": MediaTranslationType.DUB,
"raw": MediaTranslationType.RAW,
}
def map_to_search_results(data: AnimePaheSearchPage) -> SearchResults:
results = []
@@ -91,7 +92,7 @@ def map_to_server(
EpisodeStream(
link=stream_link,
quality=quality,
translation_type=translation_type,
translation_type=translation_type_map[translation_type],
)
]
return Server(name="kwik", links=links, episode_title=episode.title)

View File

@@ -1,6 +1,11 @@
from enum import Enum
from typing import Literal, TypedDict
class Server(Enum):
KWIK = "Kwik"
class AnimePaheSearchResult(TypedDict):
id: str
title: str

View File

@@ -1,5 +1,5 @@
from dataclasses import dataclass
from typing import Literal
from typing import Literal, Optional
@dataclass(frozen=True)
@@ -16,12 +16,12 @@ class SearchParams:
# filters
translation_type: Literal["sub", "dub"] = "sub"
genre: str | None = None
year: int | None = None
status: str | None = None
genre: Optional[str] = None
year: Optional[int] = None
status: Optional[str] = None
allow_nsfw: bool = True
allow_unknown: bool = True
country_of_origin: str | None = None
country_of_origin: Optional[str] = None
@dataclass(frozen=True)
@@ -32,7 +32,7 @@ class EpisodeStreamsParams:
anime_id: str
episode: str
translation_type: Literal["sub", "dub"] = "sub"
server: str | None = None
server: Optional[str] = None
quality: Literal["1080", "720", "480", "360"] = "720"
subtitles: bool = True

View File

@@ -4,9 +4,8 @@ import logging
from httpx import Client
from yt_dlp.utils.networking import random_user_agent
from .allanime.constants import SERVERS_AVAILABLE as ALLANIME_SERVERS
from .animepahe.constants import SERVERS_AVAILABLE as ANIMEPAHE_SERVERS
from .base import BaseAnimeProvider
from .types import ProviderName
logger = logging.getLogger(__name__)
@@ -17,14 +16,13 @@ PROVIDERS_AVAILABLE = {
"nyaa": "provider.Nyaa",
"yugen": "provider.Yugen",
}
SERVERS_AVAILABLE = ["TOP", *ALLANIME_SERVERS, *ANIMEPAHE_SERVERS]
class AnimeProviderFactory:
"""Factory for creating anime provider instances."""
@staticmethod
def create(provider_name: str) -> BaseAnimeProvider:
def create(provider_name: ProviderName) -> BaseAnimeProvider:
"""
Dynamically creates an instance of the specified anime provider.
@@ -41,26 +39,23 @@ class AnimeProviderFactory:
ValueError: If the provider_name is not supported.
ImportError: If the provider module or class cannot be found.
"""
if provider_name not in PROVIDERS_AVAILABLE:
raise ValueError(
f"Unsupported provider: '{provider_name}'. Supported providers are: "
f"{list(PROVIDERS_AVAILABLE.keys())}"
)
# Correctly determine module and class name from the map
import_path = PROVIDERS_AVAILABLE[provider_name]
import_path = PROVIDERS_AVAILABLE[provider_name.value.lower()]
module_name, class_name = import_path.split(".", 1)
# Construct the full package path for dynamic import
package_path = f"fastanime.libs.providers.anime.{provider_name}"
package_path = f"fastanime.libs.providers.anime.{provider_name.value.lower()}"
try:
provider_module = importlib.import_module(f".{module_name}", package_path)
provider_class = getattr(provider_module, class_name)
except (ImportError, AttributeError) as e:
logger.error(f"Failed to load provider '{provider_name}': {e}")
logger.error(
f"Failed to load provider '{provider_name.value.lower()}': {e}"
)
raise ImportError(
f"Could not load provider '{provider_name}'. "
f"Could not load provider '{provider_name.value.lower()}'. "
"Check the module path and class name in PROVIDERS_AVAILABLE."
) from e

View File

@@ -1,82 +1,115 @@
from typing import Literal, Optional
from enum import Enum
from typing import List, Literal, Optional
from pydantic import BaseModel, ConfigDict
# from .allanime.types import Server as AllAnimeServer
# from .animepahe.types import Server as AnimePaheServer
# ENUMS
class ProviderName(Enum):
ALLANIME = "allanime"
ANIMEPAHE = "animepahe"
class ProviderServer(Enum):
TOP = "TOP"
# AllAnimeServer values
SHAREPOINT = "sharepoint"
DROPBOX = "dropbox"
GOGOANIME = "gogoanime"
WETRANSFER = "weTransfer"
WIXMP = "wixmp"
YT = "Yt"
MP4_UPLOAD = "mp4-upload"
# AnimePaheServer values
KWIK = "kwik"
class MediaTranslationType(Enum):
SUB = "sub"
DUB = "dub"
RAW = "raw"
# MODELS
class BaseAnimeProviderModel(BaseModel):
model_config = ConfigDict(frozen=True)
class PageInfo(BaseAnimeProviderModel):
total: int | None = None
per_page: int | None = None
current_page: int | None = None
total: Optional[int] = None
per_page: Optional[int] = None
current_page: Optional[int] = None
class AnimeEpisodes(BaseAnimeProviderModel):
sub: list[str]
dub: list[str] = []
raw: list[str] = []
sub: List[str]
dub: List[str] = []
raw: List[str] = []
class SearchResult(BaseAnimeProviderModel):
id: str
title: str
episodes: AnimeEpisodes
other_titles: list[str] = []
media_type: str | None = None
score: float | None = None
status: str | None = None
season: str | None = None
poster: str | None = None
year: str | None = None
other_titles: List[str] = []
media_type: Optional[str] = None
score: Optional[float] = None
status: Optional[str] = None
season: Optional[str] = None
poster: Optional[str] = None
year: Optional[str] = None
class SearchResults(BaseAnimeProviderModel):
page_info: PageInfo
results: list[SearchResult]
results: List[SearchResult]
class AnimeEpisodeInfo(BaseAnimeProviderModel):
id: str
episode: str
session_id: Optional[str] = None
title: str | None = None
poster: str | None = None
duration: str | None = None
title: Optional[str] = None
poster: Optional[str] = None
duration: Optional[str] = None
class Anime(BaseAnimeProviderModel):
id: str
title: str
episodes: AnimeEpisodes
type: str | None = None
episodes_info: list[AnimeEpisodeInfo] | None = None
poster: str | None = None
year: str | None = None
type: Optional[str] = None
episodes_info: List[AnimeEpisodeInfo] | None = None
poster: Optional[str] = None
year: Optional[str] = None
class EpisodeStream(BaseAnimeProviderModel):
# episode: str
link: str
title: str | None = None
title: Optional[str] = None
quality: Literal["360", "480", "720", "1080"] = "720"
translation_type: Literal["dub", "sub"] = "sub"
format: str | None = None
hls: bool | None = None
mp4: bool | None = None
priority: int | None = None
translation_type: MediaTranslationType = MediaTranslationType.SUB
format: Optional[str] = None
hls: Optional[bool] = None
mp4: Optional[bool] = None
priority: Optional[int] = None
class Subtitle(BaseAnimeProviderModel):
url: str
language: str | None = None
language: Optional[str] = None
class Server(BaseAnimeProviderModel):
name: str
links: list[EpisodeStream]
episode_title: str | None = None
links: List[EpisodeStream]
episode_title: Optional[str] = None
headers: dict[str, str] = dict()
subtitles: list[Subtitle] = []
audio: list[str] = []
subtitles: List[Subtitle] = []
audio: List[str] = []