feat: mass refactor

This commit is contained in:
Benexl
2025-07-06 18:52:14 +03:00
parent 2f2ffc0a84
commit 355f10dd9e
24 changed files with 150 additions and 575 deletions

View File

@@ -2,8 +2,9 @@ import click
from click.core import ParameterSource
from .. import __version__
from ..core.config import AppConfig
from ..core.constants import APP_NAME
from .config import AppConfig, ConfigLoader
from .config import ConfigLoader
from .constants import USER_CONFIG_PATH
from .options import options_from_model
from .utils.lazyloader import LazyGroup

View File

@@ -1,6 +1,6 @@
import click
from ..config.model import AppConfig
from ...core.config import AppConfig
@click.command(
@@ -47,7 +47,6 @@ def config(user_config: AppConfig, path, view, desktop_entry, update):
from ..config.generate import generate_config_ini_from_app_model
from ..constants import USER_CONFIG_PATH
print(user_config.mpv.args)
if path:
print(USER_CONFIG_PATH)
elif view:

View File

@@ -1,4 +1,4 @@
from .generate import generate_config_ini_from_app_model
from .loader import ConfigLoader
from .model import AppConfig
__all__ = ["AppConfig", "ConfigLoader"]
__all__ = ["ConfigLoader", "generate_config_ini_from_app_model"]

View File

@@ -1,8 +1,8 @@
import textwrap
from pathlib import Path
from ...core.config import AppConfig
from ..constants import APP_ASCII_ART
from .model import AppConfig
# The header for the config file.
config_asci = "\n".join([f"# {line}" for line in APP_ASCII_ART.split()])

View File

@@ -4,10 +4,10 @@ from pathlib import Path
import click
from pydantic import ValidationError
from ...core.config import AppConfig
from ...core.exceptions import ConfigError
from ..constants import USER_CONFIG_PATH
from .generate import generate_config_ini_from_app_model
from .model import AppConfig
class ConfigLoader:

View File

@@ -1,299 +0,0 @@
import os
from pathlib import Path
from typing import Literal
from pydantic import BaseModel, Field, PrivateAttr, computed_field, field_validator
from ...core.constants import (
FZF_DEFAULT_OPTS,
ROFI_THEME_CONFIRM,
ROFI_THEME_INPUT,
ROFI_THEME_MAIN,
ROFI_THEME_PREVIEW,
)
from ...libs.anilist.constants import SORTS_AVAILABLE
from ...libs.providers.anime import PROVIDERS_AVAILABLE, SERVERS_AVAILABLE
from ..constants import APP_ASCII_ART, USER_VIDEOS_DIR
class OtherConfig(BaseModel):
pass
class FzfConfig(OtherConfig):
"""Configuration specific to the FZF selector."""
_opts: str = PrivateAttr(default=FZF_DEFAULT_OPTS.read_text(encoding="utf-8"))
header_color: str = Field(
default="95,135,175", description="RGB color for the main TUI header."
)
_header_ascii_art: str = PrivateAttr(default=APP_ASCII_ART)
preview_header_color: str = Field(
default="215,0,95", description="RGB color for preview pane headers."
)
preview_separator_color: str = Field(
default="208,208,208", description="RGB color for preview pane separators."
)
def __init__(self, **kwargs):
opts = kwargs.pop("opts", None)
header_ascii_art = kwargs.pop("header_ascii_art", None)
super().__init__(**kwargs)
if opts:
self._opts = opts
if header_ascii_art:
self._header_ascii_art = header_ascii_art
@computed_field(
description="The FZF options, formatted with leading tabs for the config file."
)
@property
def opts(self) -> str:
return "\n" + "\n".join([f"\t{line}" for line in self._opts.split()])
@computed_field(
description="The ASCII art to display as a header in the FZF interface."
)
@property
def header_ascii_art(self) -> str:
return "\n" + "\n".join(
[f"\t{line}" for line in self._header_ascii_art.split()]
)
class RofiConfig(OtherConfig):
"""Configuration specific to the Rofi selector."""
theme_main: Path = Field(
default=Path(str(ROFI_THEME_MAIN)),
description="Path to the main Rofi theme file.",
)
theme_preview: Path = Field(
default=Path(str(ROFI_THEME_PREVIEW)),
description="Path to the Rofi theme file for previews.",
)
theme_confirm: Path = Field(
default=Path(str(ROFI_THEME_CONFIRM)),
description="Path to the Rofi theme file for confirmation prompts.",
)
theme_input: Path = Field(
default=Path(str(ROFI_THEME_INPUT)),
description="Path to the Rofi theme file for user input prompts.",
)
class MpvConfig(OtherConfig):
"""Configuration specific to the MPV player integration."""
args: str = Field(
default="", description="Comma-separated arguments to pass to the MPV player."
)
pre_args: str = Field(
default="",
description="Comma-separated arguments to prepend before the MPV command.",
)
disable_popen: bool = Field(
default=True,
description="Disable using subprocess.Popen for MPV, which can be unstable on some systems.",
)
force_window: str = Field(
default="immediate", description="Value for MPV's --force-window option."
)
use_python_mpv: bool = Field(
default=False,
description="Use the python-mpv library for enhanced player control.",
)
class AnilistConfig(OtherConfig):
"""Configuration for interacting with the AniList API."""
per_page: int = Field(
default=15,
gt=0,
le=50,
description="Number of items to fetch per page from AniList.",
)
sort_by: str = Field(
default="SEARCH_MATCH",
description="Default sort order for AniList search results.",
examples=SORTS_AVAILABLE,
)
preferred_language: Literal["english", "romaji"] = Field(
default="english",
description="Preferred language for anime titles from AniList.",
)
@field_validator("sort_by")
@classmethod
def validate_sort_by(cls, v: str) -> str:
if v not in SORTS_AVAILABLE:
raise ValueError(
f"'{v}' is not a valid sort option. See documentation for available options."
)
return v
class GeneralConfig(BaseModel):
"""Configuration for general application behavior and integrations."""
provider: str = Field(
default="allanime",
description="The default anime provider to use for scraping.",
examples=list(PROVIDERS_AVAILABLE.keys()),
)
selector: Literal["default", "fzf", "rofi"] = Field(
default="default", description="The interactive selector tool to use for menus."
)
auto_select_anime_result: bool = Field(
default=True,
description="Automatically select the best-matching search result from a provider.",
)
icons: bool = Field(
default=False, description="Display emoji icons in the user interface."
)
preview: Literal["full", "text", "image", "none"] = Field(
default="none", description="Type of preview to display in selectors."
)
image_renderer: Literal["icat", "chafa", "imgcat"] = Field(
default="icat" if os.environ.get("KITTY_WINDOW_ID") else "chafa",
description="The command-line tool to use for rendering images in the terminal.",
)
manga_viewer: Literal["feh", "icat"] = Field(
default="feh",
description="The external application to use for viewing manga pages.",
)
downloads_dir: Path = Field(
default_factory=lambda: USER_VIDEOS_DIR,
description="The default directory to save downloaded anime.",
)
check_for_updates: bool = Field(
default=True,
description="Automatically check for new versions of FastAnime on startup.",
)
cache_requests: bool = Field(
default=True,
description="Enable caching of network requests to speed up subsequent operations.",
)
max_cache_lifetime: str = Field(
default="03:00:00",
description="Maximum lifetime for a cached request in DD:HH:MM format.",
)
normalize_titles: bool = Field(
default=True,
description="Attempt to normalize provider titles to match AniList titles.",
)
discord: bool = Field(
default=False,
description="Enable Discord Rich Presence to show your current activity.",
)
recent: int = Field(
default=50,
ge=0,
description="Number of recently watched anime to keep in history.",
)
@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="mpv", description="The media player to use for streaming."
)
quality: Literal["360", "480", "720", "1080"] = Field(
default="1080", description="Preferred stream quality."
)
translation_type: Literal["sub", "dub"] = Field(
default="sub", description="Preferred audio/subtitle language type."
)
server: str = Field(
default="TOP",
description="The default server to use from a provider. 'top' uses the first available.",
examples=SERVERS_AVAILABLE,
)
auto_next: bool = Field(
default=False,
description="Automatically play the next episode when the current one finishes.",
)
continue_from_watch_history: bool = Field(
default=True,
description="Automatically resume playback from the last known episode and position.",
)
preferred_watch_history: Literal["local", "remote"] = Field(
default="local",
description="Which watch history to prioritize: local file or remote AniList progress.",
)
auto_skip: bool = Field(
default=False,
description="Automatically skip openings/endings if skip data is available.",
)
episode_complete_at: int = Field(
default=80,
ge=0,
le=100,
description="Percentage of an episode to watch before it's marked as complete.",
)
ytdlp_format: str = Field(
default="best[height<=1080]/bestvideo[height<=1080]+bestaudio/best",
description="The format selection string for yt-dlp.",
)
force_forward_tracking: bool = Field(
default=True,
description="Prevent updating AniList progress to a lower episode number.",
)
default_media_list_tracking: Literal["track", "disabled", "prompt"] = Field(
default="prompt",
description="Default behavior for tracking progress on AniList.",
)
sub_lang: str = Field(
default="eng",
description="Preferred language code for subtitles (e.g., 'en', 'es').",
)
@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 AppConfig(BaseModel):
"""The root configuration model for the FastAnime application."""
general: GeneralConfig = Field(
default_factory=GeneralConfig,
description="General configuration settings for application behavior.",
)
stream: StreamConfig = Field(
default_factory=StreamConfig,
description="Settings related to video streaming and playback.",
)
anilist: AnilistConfig = Field(
default_factory=AnilistConfig,
description="Configuration for AniList API integration.",
)
fzf: FzfConfig = Field(
default_factory=FzfConfig,
description="Settings for the FZF selector interface.",
)
rofi: RofiConfig = Field(
default_factory=RofiConfig,
description="Settings for the Rofi selector interface.",
)
mpv: MpvConfig = Field(
default_factory=MpvConfig, description="Configuration for the MPV media player."
)

View File

@@ -1,10 +1,9 @@
import os
import sys
from pathlib import Path
import click
from ..core.constants import APP_NAME, ICONS_DIR
from ..core.constants import APP_NAME, ICONS_DIR, PLATFORM
APP_ASCII_ART = """\
███████╗░█████╗░░██████╗████████╗░█████╗░███╗░░██╗██╗███╗░░░███╗███████╗
@@ -14,7 +13,6 @@ APP_ASCII_ART = """\
██║░░░░░██║░░██║██████╔╝░░░██║░░░██║░░██║██║░╚███║██║██║░╚═╝░██║███████╗
╚═╝░░░░░╚═╝░░╚═╝╚═════╝░░░░╚═╝░░░╚═╝░░╚═╝╚═╝░░╚══╝╚═╝╚═╝░░░░░╚═╝╚══════╝
"""
PLATFORM = sys.platform
USER_NAME = os.environ.get("USERNAME", "Anime Fan")

View File

@@ -1 +0,0 @@

View File

@@ -1,74 +0,0 @@
fetch_anime_for_fzf() {
local search_term="$1"
if [ -z "$search_term" ]; then exit 0; fi
local query='
query ($search: String) {
Page(page: 1, perPage: 25) {
media(search: $search, type: ANIME, sort: [SEARCH_MATCH]) {
id
title { romaji english }
meanScore
format
status
}
}
}
'
local json_payload
json_payload=$(jq -n --arg query "$query" --arg search "$search_term" \
'{query: $query, variables: {search: $search}}')
curl --silent \
--header "Content-Type: application/json" \
--header "Accept: application/json" \
--request POST \
--data "$json_payload" \
https://graphql.anilist.co | \
jq -r '.data.Page.media[]? | select(.title.romaji) |
"\(.title.english // .title.romaji) | Score: \(.meanScore // "N/A") | ID: \(.id)"'
}
fetch_anime_details() {
local anime_id
anime_id=$(echo "$1" | sed -n 's/.*ID: \([0-9]*\).*/\1/p')
if [ -z "$anime_id" ]; then echo "Select an item to see details..."; return; fi
local query='
query ($id: Int) {
Media(id: $id, type: ANIME) {
title { romaji english }
description(asHtml: false)
genres
meanScore
episodes
status
format
season
seasonYear
studios(isMain: true) { nodes { name } }
}
}
'
local json_payload
json_payload=$(jq -n --arg query "$query" --argjson id "$anime_id" \
'{query: $query, variables: {id: $id}}')
# Fetch and format details for the preview window
curl --silent \
--header "Content-Type: application/json" \
--header "Accept: application/json" \
--request POST \
--data "$json_payload" \
https://graphql.anilist.co | \
jq -r '
.data.Media |
"Title: \(.title.english // .title.romaji)\n" +
"Score: \(.meanScore // "N/A") | Episodes: \(.episodes // "N/A")\n" +
"Status: \(.status // "N/A") | Format: \(.format // "N/A")\n" +
"Season: \(.season // "N/A") \(.seasonYear // "")\n" +
"Genres: \(.genres | join(", "))\n" +
"Studio: \(.studios.nodes[0].name // "N/A")\n\n" +
"\(.description | gsub("<br><br>"; "\n\n") | gsub("<[^>]*>"; "") | gsub("&quot;"; "\""))"
'
}

View File

@@ -1,244 +0,0 @@
import logging
import os
import shutil
import subprocess
import sys
from collections.abc import Callable
from typing import List
from click import clear
from rich import print
from ...cli.utils.tools import exit_app
from .scripts import FETCH_ANIME_SCRIPT
logger = logging.getLogger(__name__)
FZF_DEFAULT_OPTS = """
--color=fg:#d0d0d0,fg+:#d0d0d0,bg:#121212,bg+:#262626
--color=hl:#5f87af,hl+:#5fd7ff,info:#afaf87,marker:#87ff00
--color=prompt:#d7005f,spinner:#af5fff,pointer:#af5fff,header:#87afaf
--color=border:#262626,label:#aeaeae,query:#d9d9d9
--border="rounded" --border-label="" --preview-window="border-rounded" --prompt="> "
--marker=">" --pointer="" --separator="" --scrollbar=""
"""
HEADER = """
███████╗░█████╗░░██████╗████████╗░█████╗░███╗░░██╗██╗███╗░░░███╗███████╗
██╔════╝██╔══██╗██╔════╝╚══██╔══╝██╔══██╗████╗░██║██║████╗░████║██╔════╝
█████╗░░███████║╚█████╗░░░░██║░░░███████║██╔██╗██║██║██╔████╔██║█████╗░░
██╔══╝░░██╔══██║░╚═══██╗░░░██║░░░██╔══██║██║╚████║██║██║╚██╔╝██║██╔══╝░░
██║░░░░░██║░░██║██████╔╝░░░██║░░░██║░░██║██║░╚███║██║██║░╚═╝░██║███████╗
╚═╝░░░░░╚═╝░░╚═╝╚═════╝░░░░╚═╝░░░╚═╝░░╚═╝╚═╝░░╚══╝╚═╝╚═╝░░░░░╚═╝╚══════╝
"""
class FZF:
"""an abstraction over the fzf commandline utility
Attributes:
FZF_EXECUTABLE: [TODO:attribute]
default_options: [TODO:attribute]
stdout: [TODO:attribute]
"""
# if not os.getenv("FZF_DEFAULT_OPTS"):
# os.environ["FZF_DEFAULT_OPTS"] = FZF_DEFAULT_OPTS
FZF_EXECUTABLE = shutil.which("fzf")
default_options = [
"--cycle",
"--info=hidden",
"--layout=reverse",
"--height=100%",
"--bind=right:accept,ctrl-/:toggle-preview,ctrl-space:toggle-wrap+toggle-preview-wrap",
"--no-margin",
"+m",
"-i",
"--exact",
"--tabstop=1",
"--preview-window=left,35%,wrap",
"--wrap",
]
def _with_filter(self, command: str, work: Callable) -> list[str]:
"""ported from the fzf docs demo
Args:
command: [TODO:description]
work: [TODO:description]
Returns:
[TODO:return]
"""
try:
process = subprocess.Popen(
command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
text=True,
shell=True,
)
except subprocess.SubprocessError as e:
print(f"Failed to start subprocess: {e}", file=sys.stderr)
return []
original_stdout = sys.stdout
sys.stdout = process.stdin
try:
work()
if process.stdin:
process.stdin.close()
except Exception as e:
print(f"Exception during work execution: {e}", file=sys.stderr)
finally:
sys.stdout = original_stdout
output = []
if process.stdout:
output = process.stdout.read().splitlines()
process.stdout.close()
return output
def _run_fzf(self, commands: list[str], _fzf_input) -> str:
"""core abstraction
Args:
_fzf_input ([TODO:parameter]): [TODO:description]
commands: [TODO:description]
Raises:
Exception: [TODO:throw]
Returns:
[TODO:return]
"""
fzf_input = "\n".join(_fzf_input)
if not self.FZF_EXECUTABLE:
raise Exception("fzf executable not found")
result = subprocess.run(
[self.FZF_EXECUTABLE, *commands],
input=fzf_input,
stdout=subprocess.PIPE,
text=True,
encoding="utf-8",
check=False,
)
if not result or result.returncode != 0 or not result.stdout:
if result.returncode == 130: # fzf terminated by ctrl-c
exit_app()
print("sth went wrong :confused:")
input("press enter to try again...")
clear()
return self._run_fzf(commands, _fzf_input)
clear()
return result.stdout.strip()
def run(
self,
fzf_input: list[str],
prompt: str,
header: str = HEADER,
preview: str | None = None,
expect: str | None = None,
validator: Callable | None = None,
) -> str:
"""a helper method that wraps common use cases over the fzf utility
Args:
fzf_input: [TODO:description]
prompt: [TODO:description]
header: [TODO:description]
preview: [TODO:description]
expect: [TODO:description]
validator: [TODO:description]
Returns:
[TODO:return]
"""
_HEADER_COLOR = os.environ.get("FASTANIME_HEADER_COLOR", "215,0,95").split(",")
header = os.environ.get("FASTANIME_HEADER_ASCII_ART", HEADER)
header = "\n".join(
[
f"\033[38;2;{_HEADER_COLOR[0]};{_HEADER_COLOR[1]};{_HEADER_COLOR[2]};m{line}\033[0m"
for line in header.split("\n")
]
)
_commands = [
*self.default_options,
"--header",
header,
"--header-first",
"--prompt",
f"{prompt.title()}: ",
] # pyright:ignore
if preview:
_commands.append(f"--preview={preview}")
if expect:
_commands.append(f"--expect={expect}")
result = self._run_fzf(_commands, fzf_input) # pyright:ignore
if not result:
print("Please enter a value")
input("Enter to do it right")
return self.run(fzf_input, prompt, header, preview, expect, validator)
elif validator:
success, info = validator(result)
if not success:
print(info)
input("Enter to try again")
return self.run(fzf_input, prompt, header, preview, expect, validator)
# os.environ["FZF_DEFAULT_OPTS"] = ""
return result
def search_for_anime(self):
commands = [
"--preview",
f"{FETCH_ANIME_SCRIPT}fetch_anime_details {{}}",
"--prompt",
"Search For Anime: ",
"--header",
"Type to search, results are dynamically loaded, enter to select",
"--bind",
f"change:reload({FETCH_ANIME_SCRIPT}fetch_anime_for_fzf {{q}})",
"--preview-window",
"wrap",
# "--bind",
# f"enter:become(echo {{}})",
"--reverse",
]
if not self.FZF_EXECUTABLE:
raise Exception("fzf executable not found")
os.environ["SHELL"] = "bash"
result = subprocess.run(
[self.FZF_EXECUTABLE, *commands],
input="",
stdout=subprocess.PIPE,
text=True,
encoding="utf-8",
check=False,
)
if not result or result.returncode != 0 or not result.stdout:
if result.returncode == 130: # fzf terminated by ctrl-c
exit_app()
return ""
return result.stdout.strip().split("|")[0].strip()
fzf = FZF()
if __name__ == "__main__":
print(fzf.search_for_anime())
exit()

View File

@@ -1 +0,0 @@
from .rofi import Rofi

View File

@@ -1,169 +0,0 @@
import subprocess
from shutil import which
from sys import exit
from fastanime import APP_NAME
from ...constants import ICON_PATH
class RofiApi:
ROFI_EXECUTABLE = which("rofi")
rofi_theme = ""
rofi_theme_preview = ""
rofi_theme_confirm = ""
rofi_theme_input = ""
def run_with_icons(self, options: list[str], prompt_text: str) -> str:
rofi_input = "\n".join(options)
if not self.ROFI_EXECUTABLE:
raise Exception("Rofi not found")
args = [self.ROFI_EXECUTABLE]
if self.rofi_theme_preview:
args.extend(["-no-config", "-theme", self.rofi_theme_preview])
args.extend(["-p", f"{prompt_text.title()}", "-i", "-show-icons", "-dmenu"])
result = subprocess.run(
args,
input=rofi_input,
stdout=subprocess.PIPE,
text=True,
check=False,
)
choice = result.stdout.strip()
if not choice:
try:
from plyer import notification
except ImportError:
print(
"Plyer is not installed; install it for desktop notifications to be enabled"
)
exit(1)
notification.notify(
app_name=APP_NAME,
app_icon=ICON_PATH,
message="FastAnime is shutting down",
title="No Valid Input Provided",
) # pyright:ignore
exit(1)
return choice
def run(self, options: list[str], prompt_text: str) -> str:
rofi_input = "\n".join(options)
if not self.ROFI_EXECUTABLE:
raise Exception("Rofi not found")
args = [self.ROFI_EXECUTABLE]
if self.rofi_theme:
args.extend(["-no-config", "-theme", self.rofi_theme])
args.extend(["-p", prompt_text, "-i", "-dmenu"])
result = subprocess.run(
args,
input=rofi_input,
stdout=subprocess.PIPE,
text=True,
check=False,
)
choice = result.stdout.strip()
if not choice or choice not in options:
try:
from plyer import notification
except ImportError:
print(
"Plyer is not installed; install it for desktop notifications to be enabled"
)
exit(1)
notification.notify(
app_name=APP_NAME,
app_icon=ICON_PATH,
message="FastAnime is shutting down",
title="No Valid Input Provided",
) # pyright:ignore
exit(1)
return choice
def confirm(self, prompt_text: str) -> bool:
rofi_choices = "Yes\nNo"
if not self.ROFI_EXECUTABLE:
raise Exception("Rofi not found")
args = [self.ROFI_EXECUTABLE]
if self.rofi_theme_confirm:
args.extend(["-no-config", "-theme", self.rofi_theme_confirm])
args.extend(["-p", prompt_text, "-i", "", "-no-fixed-num-lines", "-dmenu"])
result = subprocess.run(
args,
input=rofi_choices,
stdout=subprocess.PIPE,
text=True,
check=False,
)
choice = result.stdout.strip()
if not choice:
try:
from plyer import notification
except ImportError:
print(
"Plyer is not installed; install it for desktop notifications to be enabled"
)
exit(1)
notification.notify(
app_name=APP_NAME,
app_icon=ICON_PATH,
message="FastAnime is shutting down",
title="No Valid Input Provided",
) # pyright:ignore
exit(1)
if choice == "Yes":
return True
else:
return False
def ask(
self, prompt_text: str, is_int: bool = False, is_float: bool = False
) -> str | float | int:
if not self.ROFI_EXECUTABLE:
raise Exception("Rofi not found")
args = [self.ROFI_EXECUTABLE]
if self.rofi_theme_input:
args.extend(["-no-config", "-theme", self.rofi_theme_input])
args.extend(["-p", prompt_text, "-i", "-no-fixed-num-lines", "-dmenu"])
result = subprocess.run(
args,
stdout=subprocess.PIPE,
text=True,
check=False,
)
user_input = result.stdout.strip()
if not user_input:
try:
from plyer import notification
except ImportError:
print(
"Plyer is not installed; install it for desktop notifications to be enabled"
)
exit(1)
notification.notify(
app_name=APP_NAME,
app_icon=ICON_PATH,
message="FastAnime is shutting down",
title="No Valid Input Provided",
) # pyright:ignore
exit(1)
if is_float:
user_input = float(user_input)
elif is_int:
user_input = int(user_input)
return user_input
Rofi = RofiApi()