feat: update config logic with new philosophy

This commit is contained in:
Benexl
2025-07-05 17:13:21 +03:00
parent 759889acd4
commit 3af31a2dfd
59 changed files with 981 additions and 1610 deletions

View File

@@ -5,10 +5,10 @@ import logging
import os
from typing import TYPE_CHECKING
from .libs.anime_provider import anime_sources
from .libs.anime_provider import PROVIDERS_AVAILABLE
if TYPE_CHECKING:
from typing import Iterator
from collections.abc import Iterator
from .libs.anime_provider.types import Anime, SearchResults, Server
@@ -27,7 +27,7 @@ class AnimeProvider:
anime_provider: [TODO:attribute]
"""
PROVIDERS = list(anime_sources.keys())
PROVIDERS = list(PROVIDERS_AVAILABLE.keys())
provider = PROVIDERS[0]
def __init__(
@@ -53,7 +53,7 @@ class AnimeProvider:
self.anime_provider.session.kill_connection_to_db()
except Exception:
pass
_, anime_provider_cls_name = anime_sources[provider].split(".", 1)
_, anime_provider_cls_name = PROVIDERS_AVAILABLE[provider].split(".", 1)
package = f"fastanime.libs.anime_provider.{provider}"
provider_api = importlib.import_module(".api", package)
anime_provider = getattr(provider_api, anime_provider_cls_name)

View File

@@ -76,7 +76,7 @@ class YtDLPDownloader:
"--out",
os.path.join(download_dir, anime_title, episode_title),
]
subprocess.run(cmd)
subprocess.run(cmd, check=False)
return
ydl_opts = {
# Specify the output path and template
@@ -106,21 +106,34 @@ class YtDLPDownloader:
if hls_use_mpegts:
options = options | {
"hls_use_mpegts": True,
"outtmpl": ".".join(options["outtmpl"].split(".")[:-1]) + ".ts", # force .ts extension
"outtmpl": ".".join(options["outtmpl"].split(".")[:-1])
+ ".ts", # force .ts extension
}
elif hls_use_h264:
options = options | {
"external_downloader_args": options["external_downloader_args"] | {
"ffmpeg_o1": [
"-c:v", "copy",
"-c:a", "aac",
"-bsf:a", "aac_adtstoasc",
"-q:a", "1",
"-ac", "2",
"-af", "loudnorm=I=-22:TP=-2.5:LRA=11,alimiter=limit=-1.5dB", # prevent clipping from HE-AAC to AAC convertion
],
options = (
options
| {
"external_downloader_args": options[
"external_downloader_args"
]
| {
"ffmpeg_o1": [
"-c:v",
"copy",
"-c:a",
"aac",
"-bsf:a",
"aac_adtstoasc",
"-q:a",
"1",
"-ac",
"2",
"-af",
"loudnorm=I=-22:TP=-2.5:LRA=11,alimiter=limit=-1.5dB", # prevent clipping from HE-AAC to AAC convertion
],
}
}
}
)
with yt_dlp.YoutubeDL(options) as ydl:
info = ydl.extract_info(url, download=True)

View File

@@ -1,12 +1,13 @@
import sys
import importlib.metadata
if sys.version_info < (3, 10):
raise ImportError(
"You are using an unsupported version of Python. Only Python versions 3.10 and above are supported by FastAnime"
) # noqa: F541
)
__version__ = "v2.8.8"
__version__ = importlib.metadata.version("FastAnime")
APP_NAME = "FastAnime"
AUTHOR = "Benexl"

View File

@@ -1,401 +1,3 @@
import signal
from .cli import cli as run_cli
import click
from .. import __version__
from ..libs.anime_provider import SERVERS_AVAILABLE, anime_sources
from .commands import LazyGroup
commands = {
"search": "search.search",
"download": "download.download",
"anilist": "anilist.anilist",
"config": "config.config",
"downloads": "downloads.downloads",
"cache": "cache.cache",
"completions": "completions.completions",
"update": "update.update",
"grab": "grab.grab",
"serve": "serve.serve",
}
# handle keyboard interupt
def handle_exit(signum, frame):
from click import clear
from .utils.tools import exit_app
clear()
exit_app()
signal.signal(signal.SIGINT, handle_exit)
@click.group(
lazy_subcommands=commands,
cls=LazyGroup,
help="A command line application for streaming anime that provides a complete and featureful interface",
short_help="Stream Anime",
epilog="""
\b
\b\bExamples:
# example of syncplay intergration
fastanime --sync-play --server sharepoint search -t <anime-title>
\b
# --- or ---
\b
# to watch with anilist intergration
fastanime --sync-play --server sharepoint anilist
\b
# downloading dubbed anime
fastanime --dub download -t <anime>
\b
# use icons and fzf for a more elegant ui with preview
fastanime --icons --preview --fzf anilist
\b
# use icons with default ui
fastanime --icons --default anilist
\b
# viewing manga
fastanime --manga search -t <manga-title>
""",
)
@click.version_option(__version__, "--version")
@click.option("--manga", "-m", help="Enable manga mode", is_flag=True)
@click.option("--log", help="Allow logging to stdout", is_flag=True)
@click.option("--log-file", help="Allow logging to a file", is_flag=True)
@click.option("--rich-traceback", help="Use rich to output tracebacks", is_flag=True)
@click.option(
"-p",
"--provider",
type=click.Choice(list(anime_sources.keys()), case_sensitive=False),
help="Provider of your choice",
)
@click.option(
"-s",
"--server",
type=click.Choice([*SERVERS_AVAILABLE, "top"]),
help="Server of choice",
)
@click.option(
"-f",
"--format",
type=str,
help="yt-dlp format to use",
)
@click.option(
"-c/-no-c",
"--continue/--no-continue",
"continue_",
type=bool,
help="Continue from last episode?",
)
@click.option(
"--local-history/--remote-history",
type=bool,
help="Whether to continue from local history or remote history",
)
@click.option(
"--skip/--no-skip",
type=bool,
help="Skip opening and ending theme songs?",
)
@click.option(
"-q",
"--quality",
type=click.Choice(
[
"360",
"480",
"720",
"1080",
]
),
help="set the quality of the stream",
)
@click.option(
"-t",
"--translation-type",
type=click.Choice(["dub", "sub"]),
help="Anime language[dub/sub]",
)
@click.option(
"-sl",
"--sub-lang",
help="Set the preferred language for subs",
)
@click.option(
"-A/-no-A",
"--auto-next/--no-auto-next",
type=bool,
help="Auto select next episode?",
)
@click.option(
"-a/-no-a",
"--auto-select/--no-auto-select",
type=bool,
help="Auto select anime title?",
)
@click.option(
"--normalize-titles/--no-normalize-titles",
type=bool,
help="whether to normalize anime and episode titles given by providers",
)
@click.option("-d", "--downloads-dir", type=click.Path(), help="Downloads location")
@click.option("--fzf", is_flag=True, help="Use fzf for the ui")
@click.option("--default", is_flag=True, help="Use the default interface")
@click.option("--preview", is_flag=True, help="Show preview when using fzf")
@click.option("--no-preview", is_flag=True, help="Dont show preview when using fzf")
@click.option(
"--icons/--no-icons",
type=bool,
help="Use icons in the interfaces",
)
@click.option("--dub", help="Set the translation type to dub", is_flag=True)
@click.option("--sub", help="Set the translation type to sub", is_flag=True)
@click.option("--rofi", help="Use rofi for the ui", is_flag=True)
@click.option("--rofi-theme", help="Rofi theme to use", type=click.Path())
@click.option(
"--rofi-theme-preview", help="Rofi theme to use for previews", type=click.Path()
)
@click.option(
"--rofi-theme-confirm",
help="Rofi theme to use for the confirm prompt",
type=click.Path(),
)
@click.option(
"--rofi-theme-input",
help="Rofi theme to use for the user input prompt",
type=click.Path(),
)
@click.option(
"--use-python-mpv/--use-default-player", help="Whether to use python-mpv", type=bool
)
@click.option("--sync-play", "-sp", help="Use sync play", is_flag=True)
@click.option(
"--player",
"-P",
help="the player to use when streaming",
type=click.Choice(["mpv", "vlc"]),
)
@click.option(
"--fresh-requests", is_flag=True, help="Force the requests cache to be updated"
)
@click.option("--no-config", is_flag=True, help="Don't load the user config")
@click.pass_context
def run_cli(
ctx: click.Context,
manga,
log,
log_file,
rich_traceback,
provider,
server,
format,
continue_,
local_history,
skip,
translation_type,
sub_lang,
quality,
auto_next,
auto_select,
normalize_titles,
downloads_dir,
fzf,
default,
preview,
no_preview,
icons,
dub,
sub,
rofi,
rofi_theme,
rofi_theme_preview,
rofi_theme_confirm,
rofi_theme_input,
use_python_mpv,
sync_play,
player,
fresh_requests,
no_config,
):
import os
import sys
from .config import Config
ctx.obj = Config(no_config)
if (
ctx.obj.check_for_updates
and ctx.invoked_subcommand != "completions"
and "notifier" not in sys.argv
):
import time
last_update = ctx.obj.user_data["meta"]["last_updated"]
now = time.time()
# checks after every 12 hours
if (now - last_update) > 43200:
ctx.obj.user_data["meta"]["last_updated"] = now
ctx.obj._update_user_data()
from .app_updater import check_for_updates
print("Checking for updates...", file=sys.stderr)
print("So you can enjoy the latest features and bug fixes", file=sys.stderr)
print(
"You can disable this by setting check_for_updates to False in the config",
file=sys.stderr,
)
is_latest, github_release_data = check_for_updates()
if not is_latest:
from rich.console import Console
from rich.markdown import Markdown
from rich.prompt import Confirm
from .app_updater import update_app
def _print_release(release_data):
console = Console()
body = Markdown(release_data["body"])
tag = github_release_data["tag_name"]
tag_title = release_data["name"]
github_page_url = release_data["html_url"]
console.print(f"Release Page: {github_page_url}")
console.print(f"Tag: {tag}")
console.print(f"Title: {tag_title}")
console.print(body)
if Confirm.ask(
"A new version of fastanime is available, would you like to update?"
):
_, release_json = update_app()
print("Successfully updated")
_print_release(release_json)
exit(0)
else:
print("You are using the latest version of fastanime", file=sys.stderr)
ctx.obj.manga = manga
if log:
import logging
from rich.logging import RichHandler
FORMAT = "%(message)s"
logging.basicConfig(
level=logging.DEBUG, format=FORMAT, datefmt="[%X]", handlers=[RichHandler()]
)
logger = logging.getLogger(__name__)
logger.info("logging has been initialized")
elif log_file:
import logging
from ..constants import LOG_FILE_PATH
format = "%(asctime)s%(levelname)s: %(message)s"
logging.basicConfig(
level=logging.DEBUG,
filename=LOG_FILE_PATH,
format=format,
datefmt="[%d/%m/%Y@%H:%M:%S]",
filemode="w",
)
else:
import logging
logging.basicConfig(level=logging.CRITICAL)
if rich_traceback:
from rich.traceback import install
install()
if fresh_requests:
os.environ["FASTANIME_FRESH_REQUESTS"] = "1"
if sync_play:
ctx.obj.sync_play = sync_play
if provider:
ctx.obj.provider = provider
if server:
ctx.obj.server = server
if format:
ctx.obj.format = format
if sub_lang:
ctx.obj.sub_lang = sub_lang
if ctx.get_parameter_source("continue_") == click.core.ParameterSource.COMMANDLINE:
ctx.obj.continue_from_history = continue_
if ctx.get_parameter_source("player") == click.core.ParameterSource.COMMANDLINE:
ctx.obj.player = player
if ctx.get_parameter_source("skip") == click.core.ParameterSource.COMMANDLINE:
ctx.obj.skip = skip
if (
ctx.get_parameter_source("normalize_titles")
== click.core.ParameterSource.COMMANDLINE
):
ctx.obj.normalize_titles = normalize_titles
if quality:
ctx.obj.quality = quality
if ctx.get_parameter_source("auto_next") == click.core.ParameterSource.COMMANDLINE:
ctx.obj.auto_next = auto_next
if ctx.get_parameter_source("icons") == click.core.ParameterSource.COMMANDLINE:
ctx.obj.icons = icons
if (
ctx.get_parameter_source("local_history")
== click.core.ParameterSource.COMMANDLINE
):
ctx.obj.preferred_history = "local" if local_history else "remote"
if (
ctx.get_parameter_source("auto_select")
== click.core.ParameterSource.COMMANDLINE
):
ctx.obj.auto_select = auto_select
if (
ctx.get_parameter_source("use_python_mpv")
== click.core.ParameterSource.COMMANDLINE
):
ctx.obj.use_python_mpv = use_python_mpv
if downloads_dir:
ctx.obj.downloads_dir = downloads_dir
if translation_type:
ctx.obj.translation_type = translation_type
if default:
ctx.obj.use_fzf = False
ctx.obj.use_rofi = False
if fzf:
ctx.obj.use_fzf = True
if preview:
ctx.obj.preview = True
if no_preview:
ctx.obj.preview = False
if dub:
ctx.obj.translation_type = "dub"
if sub:
ctx.obj.translation_type = "sub"
if rofi:
ctx.obj.use_fzf = False
ctx.obj.use_rofi = True
if rofi:
from ..libs.rofi import Rofi
if rofi_theme_preview:
ctx.obj.rofi_theme_preview = rofi_theme_preview
Rofi.rofi_theme_preview = rofi_theme_preview
if rofi_theme:
ctx.obj.rofi_theme = rofi_theme
Rofi.rofi_theme = rofi_theme
if rofi_theme_input:
ctx.obj.rofi_theme_input = rofi_theme_input
Rofi.rofi_theme_input = rofi_theme_input
if rofi_theme_confirm:
ctx.obj.rofi_theme_confirm = rofi_theme_confirm
Rofi.rofi_theme_confirm = rofi_theme_confirm
ctx.obj.set_fastanime_config_environs()
__all__ = ["run_cli"]

54
fastanime/cli/cli.py Normal file
View File

@@ -0,0 +1,54 @@
import click
from click.core import ParameterSource
from .. import __version__
from .utils.lazyloader import LazyGroup
from .utils.logging import setup_logging
from .config import AppConfig, ConfigLoader
from .constants import USER_CONFIG_PATH
from .options import options_from_model
commands = {
"config": ".config",
}
@click.version_option(__version__, "--version")
@click.option("--no-config", is_flag=True, help="Don't load the user config file.")
@click.group(cls=LazyGroup, root="fastanime.cli.commands", lazy_subcommands=commands)
@options_from_model(AppConfig)
@click.pass_context
def cli(ctx: click.Context, no_config: bool, **kwargs):
"""
The main entry point for the FastAnime CLI.
"""
setup_logging(
kwargs.get("log", False),
kwargs.get("log_file", False),
kwargs.get("rich_traceback", False),
)
loader = ConfigLoader(config_path=USER_CONFIG_PATH)
config = AppConfig.model_validate({}) if no_config else loader.load()
# update app config with command line parameters
for param_name, param_value in ctx.params.items():
source = ctx.get_parameter_source(param_name)
if source == ParameterSource.COMMANDLINE:
parameter = None
for param in ctx.command.params:
if param.name == param_name:
parameter = param
break
if (
parameter
and hasattr(parameter, "model_name")
and hasattr(parameter, "field_name")
):
model_name = getattr(parameter, "model_name")
field_name = getattr(parameter, "field_name")
if hasattr(config, model_name):
model_instance = getattr(config, model_name)
if hasattr(model_instance, field_name):
setattr(model_instance, field_name, param_value)
ctx.obj = config

View File

@@ -1,40 +1,3 @@
# in lazy_group.py
import importlib
from .config import config
import click
class LazyGroup(click.Group):
def __init__(self, *args, lazy_subcommands=None, **kwargs):
super().__init__(*args, **kwargs)
# lazy_subcommands is a map of the form:
#
# {command-name} -> {module-name}.{command-object-name}
#
self.lazy_subcommands = lazy_subcommands or {}
def list_commands(self, ctx):
base = super().list_commands(ctx)
lazy = sorted(self.lazy_subcommands.keys())
return base + lazy
def get_command(self, ctx, cmd_name): # pyright:ignore
if cmd_name in self.lazy_subcommands:
return self._lazy_load(cmd_name)
return super().get_command(ctx, cmd_name)
def _lazy_load(self, cmd_name: str):
# lazily loading a command, first get the module name and attribute name
import_path: str = self.lazy_subcommands[cmd_name]
modname, cmd_object_name = import_path.rsplit(".", 1)
# do the import
mod = importlib.import_module(f".{modname}", package="fastanime.cli.commands")
# get the Command object from that module
cmd_object = getattr(mod, cmd_object_name)
# check the result to make debugging easier
if not isinstance(cmd_object, click.BaseCommand):
raise ValueError(
f"Lazy loading of {import_path} failed by returning "
"a non-command object"
)
return cmd_object
__all__ = ["config"]

View File

@@ -1,7 +1,7 @@
import click
from ...utils.tools import FastAnimeRuntimeState
from .__lazyloader__ import LazyGroup
from ...utils.lazyloader import LazyGroup
commands = {
"trending": "trending.trending",

View File

@@ -1,6 +1,6 @@
import click
from ...completion_functions import anime_titles_shell_complete
from ...utils.completion_functions import anime_titles_shell_complete
from .data import (
genres_available,
media_formats_available,
@@ -155,7 +155,7 @@ def download(
from ....anilist import AniList
force_ffmpeg |= (hls_use_mpegts or hls_use_h264)
force_ffmpeg |= hls_use_mpegts or hls_use_h264
success, anilist_search_results = AniList.search(
query=title,
@@ -206,9 +206,7 @@ def download(
anime_title, translation_type=translation_type
)
if not search_results:
print(
"No search results found from provider for {}".format(anime_title)
)
print(f"No search results found from provider for {anime_title}")
continue
search_results = search_results["results"]
if not search_results:
@@ -246,7 +244,7 @@ def download(
search_results_[selected_anime_title]["id"]
)
if not anime:
print("Failed to fetch anime {}".format(selected_anime_title))
print(f"Failed to fetch anime {selected_anime_title}")
continue
episodes = sorted(
@@ -329,14 +327,13 @@ def download(
servers_names = list(servers.keys())
if config.server in servers_names:
server_name = config.server
elif config.use_fzf:
server_name = fzf.run(servers_names, "Select an link")
else:
if config.use_fzf:
server_name = fzf.run(servers_names, "Select an link")
else:
server_name = fuzzy_inquirer(
servers_names,
"Select link",
)
server_name = fuzzy_inquirer(
servers_names,
"Select link",
)
stream_link = filter_by_quality(
config.quality, servers[server_name]["links"]
)

View File

@@ -3,7 +3,7 @@ from typing import TYPE_CHECKING
import click
from ...completion_functions import downloaded_anime_titles
from ...utils.completion_functions import downloaded_anime_titles
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
@@ -110,6 +110,7 @@ def downloads(
],
stderr=subprocess.PIPE,
stdout=subprocess.PIPE,
check=False,
)
def get_previews_anime(workers=None, bg=True):
@@ -343,16 +344,15 @@ def downloads(
stream_episode(
playlist,
)
else:
if config.sync_play:
from ...utils.syncplay import SyncPlayer
elif config.sync_play:
from ...utils.syncplay import SyncPlayer
SyncPlayer(playlist)
else:
run_mpv(
playlist,
player=config.player,
)
SyncPlayer(playlist)
else:
run_mpv(
playlist,
player=config.player,
)
stream_anime()
stream_anime(title)

View File

@@ -59,7 +59,7 @@ def login(config: "Config", status, erase):
"MacOS detected.\nPress any key once the token provided has been pasted into "
+ anilist_key_file_path
)
with open(anilist_key_file_path, "r") as key_file:
with open(anilist_key_file_path) as key_file:
token = key_file.read().strip()
else:
launch(config.fastanime_anilist_app_login_url, wait=False)

View File

@@ -41,7 +41,7 @@ def notifier(config: "Config"):
# WARNING: Mess around with this value at your own risk
timeout = 2 # time is in minutes
if os.path.exists(notified):
with open(notified, "r") as f:
with open(notified) as f:
past_notifications = json.load(f)
else:
past_notifications = {}

View File

@@ -1,6 +1,6 @@
import click
from ...completion_functions import anime_titles_shell_complete
from ...utils.completion_functions import anime_titles_shell_complete
from .data import (
genres_available,
media_formats_available,

View File

@@ -51,6 +51,7 @@ def stats(
f"{img_w}x{img_h}@{image_x}x{image_y}",
image_url,
],
check=False,
)
if not image_process.returncode == 0:
print("failed to get image from icat")

View File

@@ -37,7 +37,7 @@ def completions(fish, zsh, bash):
current_shell = None
else:
current_shell = None
if fish or current_shell == "fish" and not zsh and not bash:
if fish or (current_shell == "fish" and not zsh and not bash):
print(
"""
function _fastanime_completion;
@@ -59,7 +59,7 @@ end;
complete --no-files --command fastanime --arguments "(_fastanime_completion)";
"""
)
elif zsh or current_shell == "zsh" and not bash:
elif zsh or (current_shell == "zsh" and not bash):
print(
"""
#compdef fastanime

View File

@@ -1,9 +1,6 @@
from typing import TYPE_CHECKING
import click
if TYPE_CHECKING:
from ..config import Config
from ..config.model import AppConfig
@click.command(
@@ -46,75 +43,81 @@ if TYPE_CHECKING:
is_flag=True,
)
@click.pass_obj
def config(user_config: "Config", path, view, desktop_entry, update):
import sys
from rich import print
from ... import __version__
from ...constants import APP_NAME, ICON_PATH, S_PLATFORM, USER_CONFIG_PATH
def config(user_config: AppConfig, path, view, desktop_entry, update):
from ..constants import USER_CONFIG_PATH
from ..config.generate import generate_config_ini_from_app_model
print(user_config.mpv.args)
if path:
print(USER_CONFIG_PATH)
elif view:
print(user_config)
print(generate_config_ini_from_app_model(user_config))
elif desktop_entry:
import os
import shutil
from pathlib import Path
from textwrap import dedent
from rich import print
from rich.prompt import Confirm
from ..utils.tools import exit_app
FASTANIME_EXECUTABLE = shutil.which("fastanime")
if FASTANIME_EXECUTABLE:
cmds = f"{FASTANIME_EXECUTABLE} --rofi anilist"
else:
cmds = f"{sys.executable} -m fastanime --rofi anilist"
# TODO: Get funs of the other platforms to complete this lol
if S_PLATFORM == "win32":
print(
"Not implemented; the author thinks its not straight forward so welcomes lovers of windows to try and implement it themselves or to switch to a proper os like arch linux or pray the author gets bored 😜"
)
elif S_PLATFORM == "darwin":
print(
"Not implemented; the author thinks its not straight forward so welcomes lovers of mac to try and implement it themselves or to switch to a proper os like arch linux or pray the author gets bored 😜"
)
else:
desktop_entry = dedent(
f"""
[Desktop Entry]
Name={APP_NAME}
Type=Application
version={__version__}
Path={Path().home()}
Comment=Watch anime from your terminal
Terminal=false
Icon={ICON_PATH}
Exec={cmds}
Categories=Entertainment
"""
)
base = os.path.expanduser("~/.local/share/applications")
desktop_entry_path = os.path.join(base, f"{APP_NAME}.desktop")
if os.path.exists(desktop_entry_path):
if not Confirm.ask(
f"The file already exists {desktop_entry_path}; or would you like to rewrite it",
default=False,
):
exit_app(1)
with open(desktop_entry_path, "w") as f:
f.write(desktop_entry)
with open(desktop_entry_path) as f:
print(f"Successfully wrote \n{f.read()}")
exit_app(0)
_generate_desktop_entry()
elif update:
with open(USER_CONFIG_PATH, "w", encoding="utf-8") as file:
file.write(user_config.__str__())
file.write(generate_config_ini_from_app_model(user_config))
print("update successfull")
else:
click.edit(filename=USER_CONFIG_PATH)
click.edit(filename=str(USER_CONFIG_PATH))
def _generate_desktop_entry():
"""
Generates a desktop entry for FastAnime.
"""
from ... import __version__
import sys
import os
import shutil
from pathlib import Path
from textwrap import dedent
from rich import print
from rich.prompt import Confirm
from ..constants import APP_NAME, ICON_PATH, PLATFORM
FASTANIME_EXECUTABLE = shutil.which("fastanime")
if FASTANIME_EXECUTABLE:
cmds = f"{FASTANIME_EXECUTABLE} --rofi anilist"
else:
cmds = f"{sys.executable} -m fastanime --rofi anilist"
# TODO: Get funs of the other platforms to complete this lol
if PLATFORM == "win32":
print(
"Not implemented; the author thinks its not straight forward so welcomes lovers of windows to try and implement it themselves or to switch to a proper os like arch linux or pray the author gets bored 😜"
)
elif PLATFORM == "darwin":
print(
"Not implemented; the author thinks its not straight forward so welcomes lovers of mac to try and implement it themselves or to switch to a proper os like arch linux or pray the author gets bored 😜"
)
else:
desktop_entry = dedent(
f"""
[Desktop Entry]
Name={APP_NAME}
Type=Application
version={__version__}
Path={Path().home()}
Comment=Watch anime from your terminal
Terminal=false
Icon={ICON_PATH}
Exec={cmds}
Categories=Entertainment
"""
)
base = os.path.expanduser("~/.local/share/applications")
desktop_entry_path = os.path.join(base, f"{APP_NAME}.desktop")
if os.path.exists(desktop_entry_path):
if not Confirm.ask(
f"The file already exists {desktop_entry_path}; or would you like to rewrite it",
default=False,
):
return
with open(desktop_entry_path, "w") as f:
f.write(desktop_entry)
with open(desktop_entry_path) as f:
print(f"Successfully wrote \n{f.read()}")

View File

@@ -2,7 +2,7 @@ from typing import TYPE_CHECKING
import click
from ..completion_functions import anime_titles_shell_complete
from ..utils.completion_functions import anime_titles_shell_complete
if TYPE_CHECKING:
from ..config import Config
@@ -344,14 +344,13 @@ def download(
servers_names = list(servers.keys())
if config.server in servers_names:
server_name = config.server
elif config.use_fzf:
server_name = fzf.run(servers_names, "Select an link")
else:
if config.use_fzf:
server_name = fzf.run(servers_names, "Select an link")
else:
server_name = fuzzy_inquirer(
servers_names,
"Select link",
)
server_name = fuzzy_inquirer(
servers_names,
"Select link",
)
stream_link = filter_by_quality(
config.quality, servers[server_name]["links"]
)

View File

@@ -3,7 +3,7 @@ from typing import TYPE_CHECKING
import click
from ..completion_functions import downloaded_anime_titles
from ..utils.completion_functions import downloaded_anime_titles
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
@@ -110,6 +110,7 @@ def downloads(
],
stderr=subprocess.PIPE,
stdout=subprocess.PIPE,
check=False,
)
def get_previews_anime(workers=None, bg=True):
@@ -343,16 +344,15 @@ def downloads(
stream_episode(
playlist,
)
else:
if config.sync_play:
from ..utils.syncplay import SyncPlayer
elif config.sync_play:
from ..utils.syncplay import SyncPlayer
SyncPlayer(playlist)
else:
run_mpv(
playlist,
player=config.player,
)
SyncPlayer(playlist)
else:
run_mpv(
playlist,
player=config.player,
)
stream_anime()
stream_anime(title)

View File

@@ -2,7 +2,7 @@ from typing import TYPE_CHECKING
import click
from ..completion_functions import anime_titles_shell_complete
from ..utils.completion_functions import anime_titles_shell_complete
if TYPE_CHECKING:
from ..config import Config

View File

@@ -2,7 +2,7 @@ from typing import TYPE_CHECKING
import click
from ..completion_functions import anime_titles_shell_complete
from ..utils.completion_functions import anime_titles_shell_complete
if TYPE_CHECKING:
from ...cli.config import Config
@@ -66,7 +66,6 @@ def search(config: "Config", anime_titles: str, episode_range: str):
from yt_dlp.utils import sanitize_filename
from ...MangaProvider import MangaProvider
from ..utils.feh import feh_manga_viewer
from ..utils.icat import icat_manga_viewer
@@ -325,16 +324,15 @@ def search(config: "Config", anime_titles: str, episode_range: str):
servers_names = list(servers.keys())
if config.server in servers_names:
server = config.server
elif config.use_fzf:
server = fzf.run(servers_names, "Select an link")
elif config.use_rofi:
server = Rofi.run(servers_names, "Select an link")
else:
if config.use_fzf:
server = fzf.run(servers_names, "Select an link")
elif config.use_rofi:
server = Rofi.run(servers_names, "Select an link")
else:
server = fuzzy_inquirer(
servers_names,
"Select link",
)
server = fuzzy_inquirer(
servers_names,
"Select link",
)
stream_link = filter_by_quality(
config.quality, servers[server]["links"]
)

View File

@@ -1,634 +0,0 @@
import json
import logging
import os
import sys
import time
from configparser import ConfigParser
from typing import TYPE_CHECKING
from rich import print
from ..constants import (
ASSETS_DIR,
S_PLATFORM,
USER_CONFIG_PATH,
USER_DATA_PATH,
USER_VIDEOS_DIR,
USER_WATCH_HISTORY_PATH,
)
from ..libs.fzf import FZF_DEFAULT_OPTS, HEADER
from ..libs.rofi import Rofi
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
from ..AnimeProvider import AnimeProvider
class Config(object):
manga = False
sync_play = False
anime_list: list
watch_history: dict = {}
fastanime_anilist_app_login_url = (
"https://anilist.co/api/v2/oauth/authorize?client_id=20148&response_type=token"
)
anime_provider: "AnimeProvider"
user_data = {
"recent_anime": [],
"animelist": [],
"user": {},
"meta": {"last_updated": 0},
}
default_config = {
"auto_next": "False",
"menu_order": "",
"manga_viewer": "feh",
"auto_select": "True",
"cache_requests": "true",
"check_for_updates": "True",
"continue_from_history": "True",
"default_media_list_tracking": "None",
"downloads_dir": USER_VIDEOS_DIR,
"disable_mpv_popen": "True",
"discord": "False",
"episode_complete_at": "80",
"use_experimental_fzf_anilist_search": "True",
"ffmpegthumbnailer_seek_time": "-1",
"force_forward_tracking": "true",
"force_window": "immediate",
"fzf_opts": FZF_DEFAULT_OPTS,
"header_color": "95,135,175",
"header_ascii_art": HEADER,
"format": "best[height<=1080]/bestvideo[height<=1080]+bestaudio/best",
"icons": "false",
"image_previews": "True" if S_PLATFORM != "win32" else "False",
"image_renderer": "icat" if os.environ.get("KITTY_WINDOW_ID") else "chafa",
"normalize_titles": "True",
"notification_duration": "120",
"max_cache_lifetime": "03:00:00",
"mpv_args": "",
"mpv_pre_args": "",
"per_page": "15",
"player": "mpv",
"preferred_history": "local",
"preferred_language": "english",
"preview": "False",
"preview_header_color": "215,0,95",
"preview_separator_color": "208,208,208",
"provider": "allanime",
"quality": "1080",
"recent": "50",
"rofi_theme": os.path.join(ASSETS_DIR, "rofi_theme.rasi"),
"rofi_theme_preview": os.path.join(ASSETS_DIR, "rofi_theme_preview.rasi"),
"rofi_theme_confirm": os.path.join(ASSETS_DIR, "rofi_theme_confirm.rasi"),
"rofi_theme_input": os.path.join(ASSETS_DIR, "rofi_theme_input.rasi"),
"server": "top",
"skip": "false",
"sort_by": "search match",
"sub_lang": "eng",
"translation_type": "sub",
"use_fzf": "False",
"use_persistent_provider_store": "false",
"use_python_mpv": "false",
"use_rofi": "false",
}
def __init__(self, no_config) -> None:
self.initialize_user_data_and_watch_history_recent_anime()
self.load_config(no_config)
def load_config(self, no_config=False):
self.configparser = ConfigParser(self.default_config)
self.configparser.add_section("stream")
self.configparser.add_section("general")
self.configparser.add_section("anilist")
# --- set config values from file or using defaults ---
try:
if os.path.exists(USER_CONFIG_PATH) and not no_config:
self.configparser.read(USER_CONFIG_PATH, encoding="utf-8")
except Exception as e:
print(
"[yellow]Warning[/]: Failed to read config file using default configuration",
file=sys.stderr,
)
logger.error(f"Failed to read config file: {e}")
time.sleep(5)
# get the configuration
self.auto_next = self.configparser.getboolean("stream", "auto_next")
self.auto_select = self.configparser.getboolean("stream", "auto_select")
self.cache_requests = self.configparser.getboolean("general", "cache_requests")
self.check_for_updates = self.configparser.getboolean(
"general", "check_for_updates"
)
self.continue_from_history = self.configparser.getboolean(
"stream", "continue_from_history"
)
self.default_media_list_tracking = self.configparser.get(
"general", "default_media_list_tracking"
)
self.disable_mpv_popen = self.configparser.getboolean(
"stream", "disable_mpv_popen"
)
self.discord = self.configparser.getboolean("general", "discord")
self.downloads_dir = self.configparser.get("general", "downloads_dir")
self.episode_complete_at = self.configparser.getint(
"stream", "episode_complete_at"
)
self.use_experimental_fzf_anilist_search = self.configparser.getboolean(
"general", "use_experimental_fzf_anilist_search"
)
self.ffmpegthumbnailer_seek_time = self.configparser.getint(
"general", "ffmpegthumbnailer_seek_time"
)
self.force_forward_tracking = self.configparser.getboolean(
"general", "force_forward_tracking"
)
self.force_window = self.configparser.get("stream", "force_window")
self.format = self.configparser.get("stream", "format")
self.fzf_opts = self.configparser.get("general", "fzf_opts")
self.header_color = self.configparser.get("general", "header_color")
self.header_ascii_art = self.configparser.get("general", "header_ascii_art")
self.icons = self.configparser.getboolean("general", "icons")
self.image_previews = self.configparser.getboolean("general", "image_previews")
self.image_renderer = self.configparser.get("general", "image_renderer")
self.normalize_titles = self.configparser.getboolean(
"general", "normalize_titles"
)
self.notification_duration = self.configparser.getint(
"general", "notification_duration"
)
self._max_cache_lifetime = self.configparser.get(
"general", "max_cache_lifetime"
)
max_cache_lifetime = list(map(int, self._max_cache_lifetime.split(":")))
self.max_cache_lifetime = (
max_cache_lifetime[0] * 86400
+ max_cache_lifetime[1] * 3600
+ max_cache_lifetime[2] * 60
)
self.mpv_args = self.configparser.get("general", "mpv_args")
self.mpv_pre_args = self.configparser.get("general", "mpv_pre_args")
self.per_page = self.configparser.get("anilist", "per_page")
self.player = self.configparser.get("stream", "player")
self.preferred_history = self.configparser.get("stream", "preferred_history")
self.preferred_language = self.configparser.get("general", "preferred_language")
self.preview = self.configparser.getboolean("general", "preview")
self.preview_separator_color = self.configparser.get(
"general", "preview_separator_color"
)
self.preview_header_color = self.configparser.get(
"general", "preview_header_color"
)
self.provider = self.configparser.get("general", "provider")
self.quality = self.configparser.get("stream", "quality")
self.recent = self.configparser.getint("general", "recent")
self.rofi_theme_confirm = self.configparser.get("general", "rofi_theme_confirm")
self.rofi_theme_input = self.configparser.get("general", "rofi_theme_input")
self.rofi_theme = self.configparser.get("general", "rofi_theme")
self.rofi_theme_preview = self.configparser.get("general", "rofi_theme_preview")
self.server = self.configparser.get("stream", "server")
self.skip = self.configparser.getboolean("stream", "skip")
self.sort_by = self.configparser.get("anilist", "sort_by")
self.menu_order = self.configparser.get("general", "menu_order")
self.manga_viewer = self.configparser.get("general", "manga_viewer")
self.sub_lang = self.configparser.get("general", "sub_lang")
self.translation_type = self.configparser.get("stream", "translation_type")
self.use_fzf = self.configparser.getboolean("general", "use_fzf")
self.use_python_mpv = self.configparser.getboolean("stream", "use_python_mpv")
self.use_rofi = self.configparser.getboolean("general", "use_rofi")
self.use_persistent_provider_store = self.configparser.getboolean(
"general", "use_persistent_provider_store"
)
Rofi.rofi_theme = self.rofi_theme
Rofi.rofi_theme_input = self.rofi_theme_input
Rofi.rofi_theme_confirm = self.rofi_theme_confirm
Rofi.rofi_theme_preview = self.rofi_theme_preview
os.environ["FZF_DEFAULT_OPTS"] = self.fzf_opts
# ---- setup user data ------
self.anime_list: list = self.user_data.get("animelist", [])
self.user: dict = self.user_data.get("user", {})
if not os.path.exists(USER_CONFIG_PATH):
with open(USER_CONFIG_PATH, "w", encoding="utf-8") as config:
config.write(self.__repr__())
def set_fastanime_config_environs(self):
current_config = []
for key in self.default_config:
if not os.environ.get(f"FASTANIME_{key.upper()}"):
current_config.append(
(f"FASTANIME_{key.upper()}", str(getattr(self, key)))
)
os.environ.update(current_config)
def update_user(self, user):
self.user = user
self.user_data["user"] = user
self._update_user_data()
def update_recent(self, recent_anime: list):
recent_anime_ids = []
_recent_anime = []
for anime in recent_anime:
if (
anime["id"] not in recent_anime_ids
and len(recent_anime_ids) <= self.recent
):
_recent_anime.append(anime)
recent_anime_ids.append(anime["id"])
self.user_data["recent_anime"] = _recent_anime
self._update_user_data()
def media_list_track(
self,
anime_id: int,
episode_no: str,
episode_stopped_at="0",
episode_total_length="0",
progress_tracking="prompt",
):
self.watch_history.update(
{
str(anime_id): {
"episode_no": episode_no,
"episode_stopped_at": episode_stopped_at,
"episode_total_length": episode_total_length,
"progress_tracking": progress_tracking,
}
}
)
with open(USER_WATCH_HISTORY_PATH, "w") as f:
json.dump(self.watch_history, f)
def initialize_user_data_and_watch_history_recent_anime(self):
try:
if os.path.isfile(USER_DATA_PATH):
with open(USER_DATA_PATH, "r") as f:
user_data = json.load(f)
self.user_data.update(user_data)
except Exception as e:
logger.error(e)
try:
if os.path.isfile(USER_WATCH_HISTORY_PATH):
with open(USER_WATCH_HISTORY_PATH, "r") as f:
watch_history = json.load(f)
self.watch_history.update(watch_history)
except Exception as e:
logger.error(e)
def _update_user_data(self):
"""method that updates the actual user data file"""
with open(USER_DATA_PATH, "w") as f:
json.dump(self.user_data, f)
def update_config(self, section: str, key: str, value: str):
self.configparser.set(section, key, value)
with open(USER_CONFIG_PATH, "w") as config:
self.configparser.write(config)
def __repr__(self):
new_line = "\n"
tab = "\t"
current_config_state = f"""\
#
# ███████╗░█████╗░░██████╗████████╗░█████╗░███╗░░██╗██╗███╗░░░███╗███████╗ ░█████╗░░█████╗░███╗░░██╗███████╗██╗░██████╗░
# ██╔════╝██╔══██╗██╔════╝╚══██╔══╝██╔══██╗████╗░██║██║████╗░████║██╔════╝ ██╔══██╗██╔══██╗████╗░██║██╔════╝██║██╔════╝░
# █████╗░░███████║╚█████╗░░░░██║░░░███████║██╔██╗██║██║██╔████╔██║█████╗░░ ██║░░╚═╝██║░░██║██╔██╗██║█████╗░░██║██║░░██╗░
# ██╔══╝░░██╔══██║░╚═══██╗░░░██║░░░██╔══██║██║╚████║██║██║╚██╔╝██║██╔══╝░░ ██║░░██╗██║░░██║██║╚████║██╔══╝░░██║██║░░╚██╗
# ██║░░░░░██║░░██║██████╔╝░░░██║░░░██║░░██║██║░╚███║██║██║░╚═╝░██║███████╗ ╚█████╔╝╚█████╔╝██║░╚███║██║░░░░░██║╚██████╔╝
# ╚═╝░░░░░╚═╝░░╚═╝╚═════╝░░░░╚═╝░░░╚═╝░░╚═╝╚═╝░░╚══╝╚═╝╚═╝░░░░░╚═╝╚══════╝ ░╚════╝░░╚════╝░╚═╝░░╚══╝╚═╝░░░░░╚═╝░╚═════╝░
#
[general]
# Can you rice it?
# For the preview pane
preview_separator_color = {self.preview_separator_color}
preview_header_color = {self.preview_header_color}
# For the header
# Be sure to indent
header_ascii_art = {new_line.join([tab + line for line in self.header_ascii_art.split(new_line)])}
header_color = {self.header_color}
# the image renderer to use [icat/chafa]
image_renderer = {self.image_renderer}
# To be passed to fzf
# Be sure to indent
fzf_opts = {new_line.join([tab + line for line in self.fzf_opts.split(new_line)])}
# Whether to show the icons in the TUI [True/False]
# More like emojis
# By the way, if you have any recommendations
# for which should be used where, please
# don't hesitate to share your opinion
# because it's a lot of work
# to look for the right one for each menu option
# Be sure to also give the replacement emoji
icons = {self.icons}
# Whether to normalize provider titles [True/False]
# Basically takes the provider titles and finds the corresponding Anilist title, then changes the title to that
# Useful for uniformity, especially when downloading from different providers
# This also applies to episode titles
normalize_titles = {self.normalize_titles}
# Whether to check for updates every time you run the script [True/False]
# This is useful for keeping your script up to date
# because there are always new features being added 😄
check_for_updates = {self.check_for_updates}
# Can be [allanime, animepahe, hianime, nyaa, yugen]
# Allanime is the most reliable
# Animepahe provides different links to streams of different quality, so a quality can be selected reliably with the --quality option
# Hianime usually provides subs in different languages, and its servers are generally faster
# NOTE: Currently, they are encrypting the video links
# though Im working on it
# However, you can still get the links to the subs
# with ```fastanime grab``` command
# Yugen meh
# Nyaa for those who prefer torrents, though not reliable due to auto-selection of results
# as most of the data in Nyaa is not structured
# though it works relatively well for new anime
# especially with SubsPlease and HorribleSubs
# Oh, and you should have webtorrent CLI to use this
provider = {self.provider}
# Display language [english, romaji]
# This is passed to Anilist directly and is used to set the language for anime titles
# when using the Anilist interface
preferred_language = {self.preferred_language}
# Download directory
# Where you will find your videos after downloading them with 'fastanime download' command
downloads_dir = {self.downloads_dir}
# Whether to show a preview window when using fzf or rofi [True/False]
# The preview requires you to have a command-line image viewer as documented in the README
# This is only when using fzf or rofi
# If you don't care about image and text previews, it doesnt matter
# though its awesome
# Try it, and you will see
preview = {self.preview}
# Whether to show images in the preview [True/False]
# Windows users: just switch to Linux 😄
# because even if you enable it
# it won't look pretty
# Just be satisfied with the text previews
# So forget it exists 🤣
image_previews = {self.image_previews}
# the time to seek when using ffmpegthumbnailer [-1 to 100]
# -1 means random and is the default
# ffmpegthumbnailer is used to generate previews,
# allowing you to select the time in the video to extract an image.
# Random makes things quite exciting because you never know at what time it will extract the image.
# Used by the `fastanime downloads` command.
ffmpegthumbnailer_seek_time = {self.ffmpegthumbnailer_seek_time}
# specify the order of menu items in a comma-separated list.
# Only include the base names of menu options (e.g., "Trending", "Recent").
# The default value is 'Trending,Recent,Watching,Paused,Dropped,Planned,Completed,Rewatching,Recently Updated Anime,Search,Watch History,Random Anime,Most Popular Anime,Most Favourite Anime,Most Scored Anime,Upcoming Anime,Edit Config,Exit'.
# Leave blank to use the default menu order.
# You can also omit some options by not including them in the list.
menu_order = {self.menu_order}
# whether to use fzf as the interface for the anilist command and others. [True/False]
use_fzf = {self.use_fzf}
# whether to use rofi for the UI [True/False]
# It's more useful if you want to create a desktop entry,
# which can be set up with 'fastanime config --desktop-entry'.
# If you want it to be your sole interface even when fastanime is run directly from the terminal, enable this.
use_rofi = {self.use_rofi}
# rofi themes to use <path>
# The value of this option is the path to the rofi config files to use.
# I chose to split it into 4 since it gives the best look and feel.
# You can refer to the rofi demo on GitHub to see for yourself.
# I need help designing the default rofi themes.
# If you fancy yourself a rofi ricer, please contribute to improving
# the default theme.
rofi_theme = {self.rofi_theme}
rofi_theme_preview = {self.rofi_theme_preview}
rofi_theme_input = {self.rofi_theme_input}
rofi_theme_confirm = {self.rofi_theme_confirm}
# the duration in minutes a notification will stay on the screen.
# Used by the notifier command.
notification_duration = {self.notification_duration}
# used when the provider offers subtitles in different languages.
# Currently, this is the case for:
# hianime.
# The values for this option are the short names for languages.
# Regex is used to determine what you selected.
sub_lang = {self.sub_lang}
# what is your default media list tracking [track/disabled/prompt]
# This only affects your anilist anime list.
# track - means your progress will always be reflected in your anilist anime list.
# disabled - means progress tracking will no longer be reflected in your anime list.
# prompt - means you will be prompted for each anime whether you want your progress to be tracked or not.
default_media_list_tracking = {self.default_media_list_tracking}
# whether media list tracking should only be updated when the next episode is greater than the previous.
# This only affects your anilist anime list.
force_forward_tracking = {self.force_forward_tracking}
# whether to cache requests [true/false]
# This improves the experience by making it faster,
# as data doesn't always need to be fetched from the web server
# and can instead be retrieved locally from the cached_requests_db.
cache_requests = {self.cache_requests}
# the max lifetime for a cached request <days:hours:minutes>
# Defaults to 3 days = 03:00:00.
# This is the time after which a cached request will be deleted (technically).
max_cache_lifetime = {self._max_cache_lifetime}
# whether to use a persistent store (basically an SQLite DB) for storing some data the provider requires
# to enable a seamless experience. [true/false]
# This option exists primarily to optimize FastAnime as a library in a website project.
# For now, it's not recommended to change it. Leave it as is.
use_persistent_provider_store = {self.use_persistent_provider_store}
# number of recent anime to keep [0-50].
# 0 will disable recent anime tracking.
recent = {self.recent}
# enable or disable Discord activity updater.
# If you want to enable it, please follow the link below to register the app with your Discord account:
# https://discord.com/oauth2/authorize?client_id=1292070065583165512
discord = {self.discord}
# comma separated list of args that will be passed to mpv
# example: --vo=kitty,--fullscreen,--volume=50
mpv_args = {self.mpv_args}
# command line options passed before the mpv command
# example: kitty
# useful incase of wanting to run sth like: kitty mpv --vo=kitty <url>
mpv_pre_args = {self.mpv_pre_args}
# choose manga viewer [feh/icat]
# feh is the default and requires feh to be installed
# icat is for kitty terminal users only
manga_viewer = {self.manga_viewer}
# a little little something i introduced
# remember how in a browser site when you search for an anime it dynamically reloads
# after every type
# well who says it cant be done in the terminal lol
# though its still experimental lol
# use this to disable it
use_experimental_fzf_anilist_search = {self.use_experimental_fzf_anilist_search}
[stream]
# the quality of the stream [1080,720,480,360]
# this option is usually only reliable when:
# provider=animepahe
# since it provides links that actually point to streams of different qualities
# while the rest just point to another link that can provide the anime from the same server
quality = {self.quality}
# Auto continue from watch history [True/False]
# this will make fastanime to choose the episode that you last watched to completion
# and increment it by one
# and use that to auto select the episode you want to watch
continue_from_history = {self.continue_from_history}
# which history to use [local/remote]
# local history means it will just use the watch history stored locally in your device
# the file that stores it is called watch_history.json
# and is stored next to your config file
# remote means it ignores the last episode stored locally
# and instead uses the one in your anilist anime list
# this config option is useful if you want to overwrite your local history
# or import history covered from another device or platform
# since remote history will take precendence over whats available locally
preferred_history = {self.preferred_history}
# Preferred language for anime [dub/sub]
translation_type = {self.translation_type}
# what server to use for a particular provider
# allanime: [dropbox, sharepoint, wetransfer, gogoanime, wixmp]
# animepahe: [kwik]
# hianime: [HD1, HD2, StreamSB, StreamTape] : only HD2 for now
# yugen: [gogoanime]
# 'top' can also be used as a value for this option
# 'top' will cause fastanime to auto select the first server it sees
# this saves on resources and is faster since not all servers are being fetched
server = {self.server}
# Auto select next episode [True/False]
# this makes fastanime increment the current episode number
# then after using that value to fetch the next episode instead of prompting
# this option is useful for binging
auto_next = {self.auto_next}
# Auto select the anime provider results with fuzzy find. [True/False]
# Note this won't always be correct
# this is because the providers sometime use non-standard names
# that are there own preference rather than the official names
# But 99% of the time will be accurate
# if this happens just turn off auto_select in the menus or from the commandline
# and manually select the correct anime title
# edit this file <https://github.com/Benexl/FastAnime/blob/master/fastanime/Utility/data.py>
# and to the dictionary of the provider
# the provider title (key) and their corresponding anilist names (value)
# and then please open a pr
# issues on the same will be ignored and then closed 😆
auto_select = {self.auto_select}
# whether to skip the opening and ending theme songs [True/False]
# NOTE: requires ani-skip to be in path
# for python-mpv users am planning to create this functionality n python without the use of an external script
# so its disabled for now
# and anyways Dan Da Dan
# taught as the importance of letting it flow 🙃
skip = {self.skip}
# at what percentage progress should the episode be considered as completed [0-100]
# this value is used to determine whether to increment the current episode number and save it to your local list
# so you can continue immediately to the next episode without select it the next time you decide to watch the anime
# it is also used to determine whether your anilist anime list should be updated or not
episode_complete_at = {self.episode_complete_at}
# whether to use python-mpv [True/False]
# to enable superior control over the player
# adding more options to it
# Enabling this option and you will ask yourself
# why you did not discover fastanime sooner 🙃
# Since you basically don't have to close the player window
# to go to the next or previous episode, switch servers,
# change translation type or change to a given episode x
# so try it if you haven't already
# if you have any issues setting it up
# don't be afraid to ask
# especially on windows
# honestly it can be a pain to set it up there
# personally it took me quite sometime to figure it out
# this is because of how windows handles shared libraries
# so just ask when you find yourself stuck
# or just switch to nixos 😄
use_python_mpv = {self.use_python_mpv}
# whether to use popen to get the timestamps for continue_from_history
# implemented because popen does not work for some reason in nixos and apparently on mac as well
# if you are on nixos or mac and you have a solution to this problem please share
# i will be glad to hear it 😄
# So for now ignore this option
# and anyways the new method of getting timestamps is better
disable_mpv_popen = {self.disable_mpv_popen}
# force mpv window
# the default 'immediate' just makes mpv to open the window even if the video has not yet loaded
# done for asthetics
# passed directly to mpv so values are same
force_window = immediate
# the format of downloaded anime and trailer
# based on yt-dlp format and passed directly to it
# learn more by looking it up on their site
# only works for downloaded anime if:
# provider=allanime, server=gogoanime
# provider=allanime, server=wixmp
# provider=hianime
# this is because they provider a m3u8 file that contans multiple quality streams
format = {self.format}
# set the player to use for streaming [mpv/vlc]
# while this option exists i will still recommend that you use mpv
# since you will miss out on some features if you use the others
player = {self.player}
[anilist]
per_page = {self.per_page}
#
# HOPE YOU ENJOY FASTANIME AND BE SURE TO STAR THE PROJECT ON GITHUB
# https://github.com/Benexl/FastAnime
#
# Also join the discord server
# where the anime tech community lives :)
# https://discord.gg/C4rhMA4mmK
#
"""
return current_config_state
def __str__(self):
return self.__repr__()

View File

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

View File

@@ -0,0 +1,60 @@
from .model import AppConfig
import textwrap
from pathlib import Path
from ..constants import APP_ASCII_ART
# The header for the config file.
config_asci = "\n".join([f"# {line}" for line in APP_ASCII_ART.split()])
CONFIG_HEADER = f"""
# ==============================================================================
#
{config_asci}
#
# ==============================================================================
# This file was auto-generated from the application's configuration model.
# You can modify these values to customize the behavior of FastAnime.
# For path-based options, you can use '~' for your home directory.
""".lstrip()
def generate_config_ini_from_app_model(app_model: AppConfig) -> str:
"""Generate a configuration file content from a Pydantic model."""
model_schema = AppConfig.model_json_schema()
config_ini_content = [CONFIG_HEADER]
for section_name, section_model in app_model:
section_class_name = model_schema["properties"][section_name]["$ref"].split(
"/"
)[-1]
section_comment = model_schema["$defs"][section_class_name]["description"]
config_ini_content.append(f"\n#\n# {section_comment}\n#")
config_ini_content.append(f"[{section_name}]")
for field_name, field_value in section_model:
description = model_schema["$defs"][section_class_name]["properties"][
field_name
].get("description", "")
if description:
# Wrap long comments for better readability in the .ini file
wrapped_comment = textwrap.fill(
description,
width=78,
initial_indent="# ",
subsequent_indent="# ",
)
config_ini_content.append(f"\n{wrapped_comment}")
if isinstance(field_value, bool):
value_str = str(field_value).lower()
elif isinstance(field_value, Path):
value_str = str(field_value)
elif field_value is None:
value_str = ""
else:
value_str = str(field_value)
config_ini_content.append(f"{field_name} = {value_str}")
return "\n".join(config_ini_content)

View File

@@ -1,30 +1,13 @@
import configparser
import textwrap
from pathlib import Path
import click
from pydantic import ValidationError
from ...core.exceptions import ConfigError
from ..constants import USER_CONFIG_PATH
from .model import AppConfig
from ...core.exceptions import ConfigError
from ..constants import ASCII_ART
# The header for the config file.
config_asci = "\n".join([f"# {line}" for line in ASCII_ART.split()])
CONFIG_HEADER = f"""
# ==============================================================================
#
{config_asci}
#
# ==============================================================================
# This file was auto-generated from the application's configuration model.
# You can modify these values to customize the behavior of FastAnime.
# For path-based options, you can use '~' for your home directory.
""".lstrip()
from .generate import generate_config_ini_from_app_model
class ConfigLoader:
@@ -58,55 +41,16 @@ class ConfigLoader:
This is the only time we write to the user's config directory.
"""
if not self.config_path.exists():
default_config = AppConfig.model_validate({})
model_schema = AppConfig.model_json_schema()
config_ini_content = [CONFIG_HEADER]
for section_name, section_model in default_config:
section_class_name = model_schema["properties"][section_name][
"$ref"
].split("/")[-1]
section_comment = model_schema["$defs"][section_class_name][
"description"
]
config_ini_content.append(f"\n#\n# {section_comment}\n#")
config_ini_content.append(f"[{section_name}]")
for field_name, field_value in section_model:
description = model_schema["$defs"][section_class_name][
"properties"
][field_name].get("description", "")
if description:
# Wrap long comments for better readability in the .ini file
wrapped_comment = textwrap.fill(
description,
width=78,
initial_indent="# ",
subsequent_indent="# ",
)
config_ini_content.append(f"\n{wrapped_comment}")
if isinstance(field_value, bool):
value_str = str(field_value).lower()
elif isinstance(field_value, Path):
value_str = str(field_value)
elif field_value is None:
value_str = ""
else:
value_str = str(field_value)
config_ini_content.append(f"{field_name} = {value_str}")
config_ini_content = generate_config_ini_from_app_model(
AppConfig().model_validate({})
)
try:
final_output = "\n".join(config_ini_content)
self.config_path.parent.mkdir(parents=True, exist_ok=True)
self.config_path.write_text(final_output, encoding="utf-8")
self.config_path.write_text(config_ini_content, encoding="utf-8")
click.echo(f"Created default configuration file at: {self.config_path}")
except Exception as e:
raise ConfigError(
f"Could not create default configuration file at {str(self.config_path)}. Please check permissions. Error: {e}",
f"Could not create default configuration file at {self.config_path!s}. Please check permissions. Error: {e}",
)
def load(self) -> AppConfig:

View File

@@ -1,7 +1,8 @@
from pathlib import Path
from typing import List, Literal
from pydantic import BaseModel, Field, ValidationError, field_validator, ConfigDict
from ..constants import USER_VIDEOS_DIR, ASCII_ART
from typing import Literal
import os
from pydantic import BaseModel, Field, field_validator
from ...core.constants import (
FZF_DEFAULT_OPTS,
ROFI_THEME_MAIN,
@@ -9,11 +10,16 @@ from ...core.constants import (
ROFI_THEME_CONFIRM,
ROFI_THEME_PREVIEW,
)
from ...libs.anime_provider import SERVERS_AVAILABLE
from ..constants import USER_VIDEOS_DIR, APP_ASCII_ART
from ...libs.anime_provider import SERVERS_AVAILABLE, PROVIDERS_AVAILABLE
from ...libs.anilist.constants import SORTS_AVAILABLE
class FzfConfig(BaseModel):
class External(BaseModel):
pass
class FzfConfig(External):
"""Configuration specific to the FZF selector."""
opts: str = Field(
@@ -26,21 +32,43 @@ class FzfConfig(BaseModel):
),
description="Command-line options to pass to FZF for theming and behavior.",
)
header_color: str = "95,135,175"
preview_header_color: str = "215,0,95"
preview_separator_color: str = "208,208,208"
header_color: str = Field(
default="95,135,175", description="RGB color for the main TUI header."
)
header_ascii_art: str = Field(
default="\n" + "\n".join([f"\t{line}" for line in APP_ASCII_ART.split("\n")]),
description="The ASCII art to display in TUI headers.",
)
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."
)
class RofiConfig(BaseModel):
class RofiConfig(External):
"""Configuration specific to the Rofi selector."""
theme_main: Path = Path(str(ROFI_THEME_MAIN))
theme_preview: Path = Path(str(ROFI_THEME_PREVIEW))
theme_confirm: Path = Path(str(ROFI_THEME_CONFIRM))
theme_input: Path = Path(str(ROFI_THEME_INPUT))
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(BaseModel):
class MpvConfig(External):
"""Configuration specific to the MPV player integration."""
args: str = Field(
@@ -63,90 +91,194 @@ class MpvConfig(BaseModel):
)
class GeneralConfig(BaseModel):
"""Configuration for general application behavior and integrations."""
provider: Literal["allanime", "animepahe", "hianime", "nyaa", "yugen"] = "allanime"
selector: Literal["default", "fzf", "rofi"] = "default"
auto_select_anime_result: bool = True
# UI/UX Settings
icons: bool = False
preview: Literal["full", "text", "image", "none"] = "none"
image_renderer: Literal["icat", "chafa", "imgcat"] = "chafa"
preferred_language: Literal["english", "romaji"] = "english"
sub_lang: str = "eng"
manga_viewer: Literal["feh", "icat"] = "feh"
# Paths & Files
downloads_dir: Path = USER_VIDEOS_DIR
# Theming & Appearance
header_ascii_art: str = Field(
default="\n" + "\n".join([f"\t{line}" for line in ASCII_ART.split()]),
description="ASCII art for TUI headers.",
)
# Advanced / Developer
check_for_updates: bool = True
cache_requests: bool = True
max_cache_lifetime: str = "03:00:00"
normalize_titles: bool = True
discord: bool = False
class StreamConfig(BaseModel):
"""Configuration specific to video streaming and playback."""
player: Literal["mpv", "vlc"] = "mpv"
quality: Literal["360", "480", "720", "1080"] = "1080"
translation_type: Literal["sub", "dub"] = "sub"
server: str = "top"
# Playback Behavior
auto_next: bool = False
continue_from_watch_history: bool = True
preferred_watch_history: Literal["local", "remote"] = "local"
auto_skip: bool = False
episode_complete_at: int = Field(default=80, ge=0, le=100)
# Technical/Downloader Settings
ytdlp_format: str = "best[height<=1080]/bestvideo[height<=1080]+bestaudio/best"
@field_validator("server")
@classmethod
def validate_server(cls, v: str) -> str:
if v not in SERVERS_AVAILABLE:
raise ValidationError(f"server must be one of {SERVERS_AVAILABLE}")
return v
class AnilistConfig(BaseModel):
class AnilistConfig(External):
"""Configuration for interacting with the AniList API."""
per_page: int = Field(default=15, gt=0, le=50)
sort_by: str = "SEARCH_MATCH"
default_media_list_tracking: Literal["track", "disabled", "prompt"] = "prompt"
force_forward_tracking: bool = True
recent: int = Field(default=50, ge=0)
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 ValidationError(f"sort_by must be one of {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_server(cls, v: str) -> str:
if v.lower() != "top" and v not in PROVIDERS_AVAILABLE:
raise ValueError(
f"'{v}' is not a valid server. Must be 'top' or 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)
stream: StreamConfig = Field(default_factory=StreamConfig)
anilist: AnilistConfig = Field(default_factory=AnilistConfig)
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.",
)
# Nested Tool-Specific Configs
fzf: FzfConfig = Field(default_factory=FzfConfig)
rofi: RofiConfig = Field(default_factory=RofiConfig)
mpv: MpvConfig = Field(default_factory=MpvConfig)
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,33 +1,30 @@
import os
import sys
from pathlib import Path
from platform import system
import click
from ..core.constants import APP_NAME, ICONS_DIR
ASCII_ART = """
APP_ASCII_ART = """\
███████╗░█████╗░░██████╗████████╗░█████╗░███╗░░██╗██╗███╗░░░███╗███████╗
██╔════╝██╔══██╗██╔════╝╚══██╔══╝██╔══██╗████╗░██║██║████╗░████║██╔════╝
█████╗░░███████║╚█████╗░░░░██║░░░███████║██╔██╗██║██║██╔████╔██║█████╗░░
██╔══╝░░██╔══██║░╚═══██╗░░░██║░░░██╔══██║██║╚████║██║██║╚██╔╝██║██╔══╝░░
██║░░░░░██║░░██║██████╔╝░░░██║░░░██║░░██║██║░╚███║██║██║░╚═╝░██║███████╗
╚═╝░░░░░╚═╝░░╚═╝╚═════╝░░░░╚═╝░░░╚═╝░░╚═╝╚═╝░░╚══╝╚═╝╚═╝░░░░░╚═╝╚══════╝
"""
PLATFORM = system()
PLATFORM = sys.platform
USER_NAME = os.environ.get("USERNAME", "Anime Fan")
APP_DATA_DIR = Path(click.get_app_dir(APP_NAME, roaming=False))
if sys.platform == "win32":
if PLATFORM == "win32":
APP_CACHE_DIR = APP_DATA_DIR / "cache"
USER_VIDEOS_DIR = Path.home() / "Videos" / APP_NAME
elif sys.platform == "darwin":
elif PLATFORM == "darwin":
APP_CACHE_DIR = Path.home() / "Library" / "Caches" / APP_NAME
USER_VIDEOS_DIR = Path.home() / "Movies" / APP_NAME

View File

@@ -63,7 +63,7 @@ def discord_updater(show, episode, switch):
def media_player_controls(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
config: Config, fastanime_runtime_state: FastAnimeRuntimeState
):
"""Menu that that offers media player controls
@@ -98,7 +98,7 @@ def media_player_controls(
def _replay():
"""replay the current media"""
selected_server: "Server" = fastanime_runtime_state.provider_current_server
selected_server: Server = fastanime_runtime_state.provider_current_server
print(
"[bold magenta]Now Replaying:[/]",
provider_anime_title,
@@ -223,13 +223,12 @@ def media_player_controls(
):
media_actions_menu(config, fastanime_runtime_state)
return
else:
if not Confirm.ask(
"Are you sure you wish to continue to the next episode you haven't completed the current episode?",
default=False,
):
media_actions_menu(config, fastanime_runtime_state)
return
elif not Confirm.ask(
"Are you sure you wish to continue to the next episode you haven't completed the current episode?",
default=False,
):
media_actions_menu(config, fastanime_runtime_state)
return
elif not config.use_rofi:
if not Confirm.ask(
"Are you sure you wish to continue to the next episode, your progress for the current episodes will be erased?",
@@ -269,8 +268,7 @@ def media_player_controls(
def _previous_episode():
"""Watch previous episode"""
prev_episode = available_episodes.index(current_episode_number) - 1
if prev_episode <= 0:
prev_episode = 0
prev_episode = max(0, prev_episode)
# fastanime_runtime_state.episode_title = episode["title"]
fastanime_runtime_state.provider_current_episode_number = available_episodes[
prev_episode
@@ -337,7 +335,7 @@ def media_player_controls(
options[f"{'' if icons else ''}Next Episode"] = _next_episode
def _toggle_auto_next(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
config: Config, fastanime_runtime_state: FastAnimeRuntimeState
):
"""helper function to toggle auto next
@@ -394,7 +392,7 @@ def media_player_controls(
def provider_anime_episode_servers_menu(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
config: Config, fastanime_runtime_state: FastAnimeRuntimeState
):
"""Menu that enables selection of a server either manually or automatically based on user config then plays the stream link of the quality the user prefers
@@ -416,7 +414,7 @@ def provider_anime_episode_servers_menu(
)
provider_anime_title: str = fastanime_runtime_state.provider_anime_title
anime_id_anilist: int = fastanime_runtime_state.selected_anime_id_anilist
provider_anime: "Anime" = fastanime_runtime_state.provider_anime
provider_anime: Anime = fastanime_runtime_state.provider_anime
server_name = ""
# get streams for episode from provider
@@ -431,9 +429,8 @@ def provider_anime_episode_servers_menu(
if not config.use_rofi:
print("Failed to fetch :cry:")
input("Enter to retry...")
else:
if not Rofi.confirm("Sth went wrong!!Enter to continue..."):
exit(1)
elif not Rofi.confirm("Sth went wrong!!Enter to continue..."):
exit(1)
media_actions_menu(config, fastanime_runtime_state)
return
@@ -701,7 +698,7 @@ def provider_anime_episode_servers_menu(
def provider_anime_episodes_menu(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
config: Config, fastanime_runtime_state: FastAnimeRuntimeState
):
"""A menu that handles selection of episode either manually or automatically based on either local episode progress or remote(anilist) progress
@@ -717,8 +714,8 @@ def provider_anime_episodes_menu(
# runtime configuration
anime_id_anilist: int = fastanime_runtime_state.selected_anime_id_anilist
anime_title: str = fastanime_runtime_state.provider_anime_title
provider_anime: "Anime" = fastanime_runtime_state.provider_anime
selected_anime_anilist: "AnilistBaseMediaDataSchema" = (
provider_anime: Anime = fastanime_runtime_state.provider_anime
selected_anime_anilist: AnilistBaseMediaDataSchema = (
fastanime_runtime_state.selected_anime_anilist
)
@@ -846,12 +843,8 @@ def provider_anime_episodes_menu(
provider_anime_episode_servers_menu(config, fastanime_runtime_state)
def fetch_anime_episode(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
):
selected_anime: "SearchResult" = (
fastanime_runtime_state.provider_anime_search_result
)
def fetch_anime_episode(config: Config, fastanime_runtime_state: FastAnimeRuntimeState):
selected_anime: SearchResult = fastanime_runtime_state.provider_anime_search_result
anime_provider = config.anime_provider
with Progress() as progress:
progress.add_task("Fetching Anime Info...", total=None)
@@ -864,9 +857,8 @@ def fetch_anime_episode(
)
if not config.use_rofi:
input("Enter to continue...")
else:
if not Rofi.confirm("Sth went wrong!!Enter to continue..."):
exit(1)
elif not Rofi.confirm("Sth went wrong!!Enter to continue..."):
exit(1)
return media_actions_menu(config, fastanime_runtime_state)
fastanime_runtime_state.provider_anime = provider_anime
@@ -879,7 +871,7 @@ def fetch_anime_episode(
def set_prefered_progress_tracking(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState", update=False
config: Config, fastanime_runtime_state: FastAnimeRuntimeState, update=False
):
if (
fastanime_runtime_state.progress_tracking == ""
@@ -910,7 +902,7 @@ def set_prefered_progress_tracking(
def anime_provider_search_results_menu(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
config: Config, fastanime_runtime_state: FastAnimeRuntimeState
):
"""A menu that handles searching and selecting provider results; either manually or through fuzzy matching
@@ -924,7 +916,7 @@ def anime_provider_search_results_menu(
# runtime data
selected_anime_title = fastanime_runtime_state.selected_anime_title_anilist
selected_anime_anilist: "AnilistBaseMediaDataSchema" = (
selected_anime_anilist: AnilistBaseMediaDataSchema = (
fastanime_runtime_state.selected_anime_anilist
)
anime_provider = config.anime_provider
@@ -942,9 +934,8 @@ def anime_provider_search_results_menu(
)
if not config.use_rofi:
input("Enter to continue...")
else:
if not Rofi.confirm("Sth went wrong!!Enter to continue..."):
exit(1)
elif not Rofi.confirm("Sth went wrong!!Enter to continue..."):
exit(1)
return media_actions_menu(config, fastanime_runtime_state)
provider_search_results = {
@@ -1004,7 +995,7 @@ def anime_provider_search_results_menu(
fetch_anime_episode(config, fastanime_runtime_state)
def download_anime(config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"):
def download_anime(config: Config, fastanime_runtime_state: FastAnimeRuntimeState):
import time
from rich.prompt import Confirm, Prompt
@@ -1164,14 +1155,13 @@ def download_anime(config: "Config", fastanime_runtime_state: "FastAnimeRuntimeS
servers_names = list(servers.keys())
if config.server in servers_names:
server_name = config.server
elif config.use_fzf:
server_name = fzf.run(servers_names, "Select an link")
else:
if config.use_fzf:
server_name = fzf.run(servers_names, "Select an link")
else:
server_name = fuzzy_inquirer(
servers_names,
"Select link",
)
server_name = fuzzy_inquirer(
servers_names,
"Select link",
)
stream_link = filter_by_quality(
config.quality, servers[server_name]["links"]
)
@@ -1228,16 +1218,14 @@ def download_anime(config: "Config", fastanime_runtime_state: "FastAnimeRuntimeS
#
# ---- ANILIST MEDIA ACTIONS MENU ----
#
def media_actions_menu(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
):
def media_actions_menu(config: Config, fastanime_runtime_state: FastAnimeRuntimeState):
"""The menu responsible for handling all media actions such as watching a trailer or streaming it
Args:
config: [TODO:description]
fastanime_runtime_state: [TODO:description]
"""
selected_anime_anilist: "AnilistBaseMediaDataSchema" = (
selected_anime_anilist: AnilistBaseMediaDataSchema = (
fastanime_runtime_state.selected_anime_anilist
)
selected_anime_title_anilist: str = (
@@ -1250,9 +1238,7 @@ def media_actions_menu(
)
episodes_total = selected_anime_anilist["episodes"] or "Inf"
def _watch_trailer(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
):
def _watch_trailer(config: Config, fastanime_runtime_state: FastAnimeRuntimeState):
"""Helper function to watch trailers with
Args:
@@ -1272,14 +1258,11 @@ def media_actions_menu(
if not config.use_rofi:
print("no trailer available :confused")
input("Enter to continue...")
else:
if not Rofi.confirm("No trailler found!!Enter to continue"):
exit(0)
elif not Rofi.confirm("No trailler found!!Enter to continue"):
exit(0)
media_actions_menu(config, fastanime_runtime_state)
def _add_to_list(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
):
def _add_to_list(config: Config, fastanime_runtime_state: FastAnimeRuntimeState):
"""Helper function to update an anime's media_list_type
Args:
@@ -1327,9 +1310,7 @@ def media_actions_menu(
input("Enter to continue...")
media_actions_menu(config, fastanime_runtime_state)
def _score_anime(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
):
def _score_anime(config: Config, fastanime_runtime_state: FastAnimeRuntimeState):
"""Helper function to score anime on anilist from terminal or rofi
Args:
@@ -1365,7 +1346,7 @@ def media_actions_menu(
# FIX: For some reason this fails to delete
def _remove_from_list(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
config: Config, fastanime_runtime_state: FastAnimeRuntimeState
):
"""Remove an anime from your media list
@@ -1391,7 +1372,7 @@ def media_actions_menu(
media_actions_menu(config, fastanime_runtime_state)
def _change_translation_type(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
config: Config, fastanime_runtime_state: FastAnimeRuntimeState
):
"""Change the translation type to use
@@ -1418,9 +1399,7 @@ def media_actions_menu(
media_actions_menu(config, fastanime_runtime_state)
def _change_player(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
):
def _change_player(config: Config, fastanime_runtime_state: FastAnimeRuntimeState):
"""Change the translation type to use
Args:
@@ -1454,7 +1433,7 @@ def media_actions_menu(
config.use_python_mpv = False
media_actions_menu(config, fastanime_runtime_state)
def _view_info(config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"):
def _view_info(config: Config, fastanime_runtime_state: FastAnimeRuntimeState):
"""helper function to view info of an anime from terminal
Args:
@@ -1520,10 +1499,9 @@ def media_actions_menu(
)
if Confirm.ask("Enter to continue...", default=True):
media_actions_menu(config, fastanime_runtime_state)
return
def _toggle_auto_select(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
config: Config, fastanime_runtime_state: FastAnimeRuntimeState
):
"""helper function to toggle auto select anime title using fuzzy matching
@@ -1535,7 +1513,7 @@ def media_actions_menu(
media_actions_menu(config, fastanime_runtime_state)
def _toggle_continue_from_history(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
config: Config, fastanime_runtime_state: FastAnimeRuntimeState
):
"""helper function to toggle continue from history
@@ -1547,7 +1525,7 @@ def media_actions_menu(
media_actions_menu(config, fastanime_runtime_state)
def _toggle_auto_next(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
config: Config, fastanime_runtime_state: FastAnimeRuntimeState
):
"""helper function to toggle auto next
@@ -1559,7 +1537,7 @@ def media_actions_menu(
media_actions_menu(config, fastanime_runtime_state)
def _change_provider(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
config: Config, fastanime_runtime_state: FastAnimeRuntimeState
):
"""Helper function to change provider to use
@@ -1567,9 +1545,9 @@ def media_actions_menu(
config: [TODO:description]
fastanime_runtime_state: [TODO:description]
"""
from ...libs.anime_provider import anime_sources
from ...libs.anime_provider import PROVIDERS_AVAILABLE
options = list(anime_sources.keys())
options = list(PROVIDERS_AVAILABLE.keys())
if config.use_fzf:
provider = fzf.run(
options, prompt="Select Translation Type", header="Language Options"
@@ -1588,9 +1566,7 @@ def media_actions_menu(
media_actions_menu(config, fastanime_runtime_state)
def _stream_anime(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
):
def _stream_anime(config: Config, fastanime_runtime_state: FastAnimeRuntimeState):
"""helper function to go to the next menu respecting your config
Args:
@@ -1600,7 +1576,7 @@ def media_actions_menu(
anime_provider_search_results_menu(config, fastanime_runtime_state)
def _select_episode_to_stream(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
config: Config, fastanime_runtime_state: FastAnimeRuntimeState
):
"""Convinience function to disable continue from history and show the episodes menu
@@ -1612,12 +1588,12 @@ def media_actions_menu(
anime_provider_search_results_menu(config, fastanime_runtime_state)
def _set_progress_tracking(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
config: Config, fastanime_runtime_state: FastAnimeRuntimeState
):
set_prefered_progress_tracking(config, fastanime_runtime_state, update=True)
media_actions_menu(config, fastanime_runtime_state)
def _relations(config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"):
def _relations(config: Config, fastanime_runtime_state: FastAnimeRuntimeState):
"""Helper function to get anime recommendations
Args:
config: [TODO:description]
@@ -1642,7 +1618,7 @@ def media_actions_menu(
anilist_results_menu(config, fastanime_runtime_state)
def _recommendations(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
config: Config, fastanime_runtime_state: FastAnimeRuntimeState
):
"""Helper function to get anime recommendations
Args:
@@ -1672,9 +1648,7 @@ def media_actions_menu(
}
anilist_results_menu(config, fastanime_runtime_state)
def _download_anime(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
):
def _download_anime(config: Config, fastanime_runtime_state: FastAnimeRuntimeState):
download_anime(config, fastanime_runtime_state)
media_actions_menu(config, fastanime_runtime_state)
@@ -1717,7 +1691,7 @@ def media_actions_menu(
# ---- ANILIST RESULTS MENU ----
#
def anilist_results_menu(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
config: Config, fastanime_runtime_state: FastAnimeRuntimeState
):
"""The menu that handles and displays the results of an anilist action enabling using to select anime of choice
@@ -1731,7 +1705,7 @@ def anilist_results_menu(
anime_data = {}
for anime in search_results:
anime: "AnilistBaseMediaDataSchema"
anime: AnilistBaseMediaDataSchema
# determine the progress of watching the anime based on whats in anilist data !! NOT LOCALLY
progress = (anime["mediaListEntry"] or {"progress": 0}).get("progress", 0)
@@ -1846,7 +1820,7 @@ def anilist_results_menu(
anilist_results_menu(config, fastanime_runtime_state)
return
selected_anime: "AnilistBaseMediaDataSchema" = anime_data[selected_anime_title]
selected_anime: AnilistBaseMediaDataSchema = anime_data[selected_anime_title]
fastanime_runtime_state.selected_anime_anilist = selected_anime
fastanime_runtime_state.selected_anime_title_anilist = (
selected_anime["title"]["romaji"] or selected_anime["title"]["english"]
@@ -1860,8 +1834,8 @@ def anilist_results_menu(
# ---- FASTANIME MAIN MENU ----
#
def _handle_animelist(
config: "Config",
fastanime_runtime_state: "FastAnimeRuntimeState",
config: Config,
fastanime_runtime_state: FastAnimeRuntimeState,
list_type: str,
page=1,
):
@@ -1879,9 +1853,8 @@ def _handle_animelist(
if not config.use_rofi:
print("You haven't logged in please run: fastanime anilist login")
input("Enter to continue...")
else:
if not Rofi.confirm("You haven't logged in!!Enter to continue"):
exit(1)
elif not Rofi.confirm("You haven't logged in!!Enter to continue"):
exit(1)
fastanime_main_menu(config, fastanime_runtime_state)
return
# determine the watch list to get
@@ -1908,9 +1881,8 @@ def _handle_animelist(
print("Sth went wrong", anime_list)
if not config.use_rofi:
input("Enter to continue")
else:
if not Rofi.confirm("Sth went wrong!!Enter to continue..."):
exit(1)
elif not Rofi.confirm("Sth went wrong!!Enter to continue..."):
exit(1)
fastanime_main_menu(config, fastanime_runtime_state)
return
# handle failure
@@ -1918,9 +1890,8 @@ def _handle_animelist(
print("Sth went wrong", anime_list)
if not config.use_rofi:
input("Enter to continue")
else:
if not Rofi.confirm("Sth went wrong!!Enter to continue..."):
exit(1)
elif not Rofi.confirm("Sth went wrong!!Enter to continue..."):
exit(1)
# recall anilist menu
fastanime_main_menu(config, fastanime_runtime_state)
return
@@ -1933,7 +1904,7 @@ def _handle_animelist(
return anime_list
def _anilist_search(config: "Config", page=1):
def _anilist_search(config: Config, page=1):
"""A function that enables seaching of an anime
Returns:
@@ -1954,7 +1925,7 @@ def _anilist_search(config: "Config", page=1):
return AniList.search(query=search_term, page=page)
def _anilist_random(config: "Config", page=1):
def _anilist_random(config: Config, page=1):
"""A function that generates random anilist ids enabling random discovery of anime
Returns:
@@ -1966,7 +1937,7 @@ def _anilist_random(config: "Config", page=1):
return AniList.search(id_in=list(random_anime))
def _watch_history(config: "Config", page=1):
def _watch_history(config: Config, page=1):
"""Function that lets you see all the anime that has locally been saved to your watch history
Returns:
@@ -1976,7 +1947,7 @@ def _watch_history(config: "Config", page=1):
return AniList.search(id_in=watch_history, sort="TRENDING_DESC", page=page)
def _recent(config: "Config", page=1):
def _recent(config: Config, page=1):
return (
True,
{"data": {"Page": {"media": config.user_data["recent_anime"]}}},
@@ -1984,14 +1955,12 @@ def _recent(config: "Config", page=1):
# WARNING: Will probably be depracated
def _anime_list(config: "Config", page=1):
def _anime_list(config: Config, page=1):
anime_list = config.anime_list
return AniList.search(id_in=anime_list, pages=page)
def fastanime_main_menu(
config: "Config", fastanime_runtime_state: "FastAnimeRuntimeState"
):
def fastanime_main_menu(config: Config, fastanime_runtime_state: FastAnimeRuntimeState):
"""The main entry point to the anilist command
Args:
@@ -2024,22 +1993,34 @@ def fastanime_main_menu(
options = {
f"{'🔥 ' if icons else ''}Trending": AniList.get_trending,
f"{'🎞️ ' if icons else ''}Recent": _recent,
f"{'📺 ' if icons else ''}Watching": lambda config, media_list_type="Watching", page=1: _handle_animelist(
f"{'📺 ' if icons else ''}Watching": lambda config,
media_list_type="Watching",
page=1: _handle_animelist(
config, fastanime_runtime_state, media_list_type, page=page
),
f"{'' if icons else ''}Paused": lambda config, media_list_type="Paused", page=1: _handle_animelist(
f"{'' if icons else ''}Paused": lambda config,
media_list_type="Paused",
page=1: _handle_animelist(
config, fastanime_runtime_state, media_list_type, page=page
),
f"{'🚮 ' if icons else ''}Dropped": lambda config, media_list_type="Dropped", page=1: _handle_animelist(
f"{'🚮 ' if icons else ''}Dropped": lambda config,
media_list_type="Dropped",
page=1: _handle_animelist(
config, fastanime_runtime_state, media_list_type, page=page
),
f"{'📑 ' if icons else ''}Planned": lambda config, media_list_type="Planned", page=1: _handle_animelist(
f"{'📑 ' if icons else ''}Planned": lambda config,
media_list_type="Planned",
page=1: _handle_animelist(
config, fastanime_runtime_state, media_list_type, page=page
),
f"{'' if icons else ''}Completed": lambda config, media_list_type="Completed", page=1: _handle_animelist(
f"{'' if icons else ''}Completed": lambda config,
media_list_type="Completed",
page=1: _handle_animelist(
config, fastanime_runtime_state, media_list_type, page=page
),
f"{'🔁 ' if icons else ''}Rewatching": lambda config, media_list_type="Rewatching", page=1: _handle_animelist(
f"{'🔁 ' if icons else ''}Rewatching": lambda config,
media_list_type="Rewatching",
page=1: _handle_animelist(
config, fastanime_runtime_state, media_list_type, page=page
),
f"{'🔔 ' if icons else ''}Recently Updated Anime": AniList.get_most_recently_updated,
@@ -2094,8 +2075,7 @@ def fastanime_main_menu(
print(anilist_data[1])
if not config.use_rofi:
input("Enter to continue...")
else:
if not Rofi.confirm("Sth went wrong!!Enter to continue..."):
exit(1)
elif not Rofi.confirm("Sth went wrong!!Enter to continue..."):
exit(1)
# recall the anilist function for the user to reattempt their choice
fastanime_main_menu(config, fastanime_runtime_state)

View File

@@ -4,11 +4,12 @@ import os
import shutil
import subprocess
import textwrap
from threading import Thread
from hashlib import sha256
from threading import Thread
import requests
from yt_dlp.utils import clean_html
from ...constants import APP_CACHE_DIR, S_PLATFORM
from ...libs.anilist.types import AnilistBaseMediaDataSchema
from ...Utility import anilist_data_helper
@@ -34,7 +35,9 @@ def aniskip(mal_id: int, episode: str):
print("Aniskip not found, please install and try again")
return
args = [ANISKIP, "-q", str(mal_id), "-e", str(episode)]
aniskip_result = subprocess.run(args, text=True, stdout=subprocess.PIPE)
aniskip_result = subprocess.run(
args, text=True, stdout=subprocess.PIPE, check=False
)
if aniskip_result.returncode != 0:
return
mpv_skip_args = aniskip_result.stdout.strip()
@@ -111,7 +114,7 @@ def write_search_results(
# use concurency to download and write as fast as possible
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
future_to_task = {}
for anime, title in zip(anilist_results, titles):
for anime, title in zip(anilist_results, titles, strict=False):
# actual image url
image_url = ""
if os.environ.get("FASTANIME_IMAGE_PREVIEWS", "true").lower() == "true":
@@ -212,7 +215,7 @@ def get_rofi_icons(
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
# load the jobs
future_to_url = {}
for anime, title in zip(anilist_results, titles):
for anime, title in zip(anilist_results, titles, strict=False):
# actual link to download image from
image_url = anime["coverImage"]["large"]

182
fastanime/cli/options.py Normal file
View File

@@ -0,0 +1,182 @@
from collections.abc import Callable
from pathlib import Path
from typing import Any, Literal, get_origin, get_args
import click
from pydantic import BaseModel
from pydantic.fields import FieldInfo
from pydantic_core import PydanticUndefined
from .config.model import External
# Mapping from Python/Pydantic types to Click types
TYPE_MAP = {
str: click.STRING,
int: click.INT,
bool: click.BOOL,
float: click.FLOAT,
Path: click.Path(),
}
class ConfigOption(click.Option):
"""
Custom click option that allows for more flexible handling of Pydantic models.
This is used to ensure that options can be generated dynamically from Pydantic models.
"""
model_name: str | None
field_name: str | None
def __init__(self, *args, **kwargs):
self.model_name = kwargs.pop("model_name", None)
self.field_name = kwargs.pop("field_name", None)
super().__init__(*args, **kwargs)
def options_from_model(model: type[BaseModel], parent_name: str = "") -> Callable:
"""
A decorator factory that generates click.option decorators from a Pydantic model.
This function introspects a Pydantic model and creates a stack of decorators
that can be applied to a click command function, ensuring the CLI options
always match the configuration model.
Args:
model: The Pydantic BaseModel class to generate options from.
Returns:
A decorator that applies the generated options to a function.
"""
decorators = []
# Check if this model inherits from ExternalTool
is_external_tool = issubclass(model, External)
model_name = model.__name__.lower().replace("config", "")
# Introspect the model's fields
for field_name, field_info in model.model_fields.items():
# Handle nested models by calling this function recursively
if isinstance(field_info.annotation, type) and issubclass(
field_info.annotation, BaseModel
):
# Apply decorators from the nested model with current model as parent
nested_decorators = options_from_model(field_info.annotation, field_name)
nested_decorator_list = getattr(nested_decorators, "decorators", [])
decorators.extend(nested_decorator_list)
continue
# Determine the option name for the CLI
if is_external_tool:
# For ExternalTool subclasses, use --model_name-field_name format
cli_name = f"--{model_name}-{field_name.replace('_', '-')}"
else:
cli_name = f"--{field_name.replace('_', '-')}"
# Build the arguments for the click.option decorator
kwargs = {
"type": _get_click_type(field_info),
"help": field_info.description or "",
}
# Handle boolean flags (e.g., --foo/--no-foo)
if field_info.annotation is bool:
# Set default value for boolean flags
if field_info.default is not PydanticUndefined:
kwargs["default"] = field_info.default
kwargs["show_default"] = True
if is_external_tool:
cli_name = (
f"{cli_name}/--no-{model_name}-{field_name.replace('_', '-')}"
)
else:
# For non-external tools, we use the --no- prefix directly
cli_name = f"{cli_name}/--no-{field_name.replace('_', '-')}"
# For other types, set default if one is provided in the model
elif field_info.default is not PydanticUndefined:
kwargs["default"] = field_info.default
kwargs["show_default"] = True
decorators.append(
click.option(
cli_name,
cls=ConfigOption,
model_name=model_name,
field_name=field_name,
**kwargs,
)
)
def decorator(f: Callable) -> Callable:
# Apply the decorators in reverse order to the function
for deco in reversed(decorators):
f = deco(f)
return f
# Store the list of decorators as an attribute for nested calls
setattr(decorator, "decorators", decorators)
return decorator
def _get_click_type(field_info: FieldInfo) -> Any:
"""Maps a Pydantic field's type to a corresponding click type."""
field_type = field_info.annotation
# Check if the type is a Literal
if (
field_type is not None
and hasattr(field_type, "__origin__")
and get_origin(field_type) is Literal
):
args = get_args(field_type)
if args:
return click.Choice(args)
# Check for examples in field_info - use as choices
if hasattr(field_info, "examples") and field_info.examples:
return click.Choice(field_info.examples)
# Check for numeric constraints and create click.Range
if field_type in (int, float):
constraints = {}
# Extract constraints from field_info.metadata
if hasattr(field_info, "metadata") and field_info.metadata:
for constraint in field_info.metadata:
constraint_type = type(constraint).__name__
if constraint_type == "Ge" and hasattr(constraint, "ge"):
constraints["min"] = constraint.ge
elif constraint_type == "Le" and hasattr(constraint, "le"):
constraints["max"] = constraint.le
elif constraint_type == "Gt" and hasattr(constraint, "gt"):
# gt means strictly greater than, so min should be gt + 1 for int
if field_type is int:
constraints["min"] = constraint.gt + 1
else:
# For float, we can't easily handle strict inequality in click.Range
constraints["min"] = constraint.gt
elif constraint_type == "Lt" and hasattr(constraint, "lt"):
# lt means strictly less than, so max should be lt - 1 for int
if field_type is int:
constraints["max"] = constraint.lt - 1
else:
# For float, we can't easily handle strict inequality in click.Range
constraints["max"] = constraint.lt
# Create click.Range if we have constraints
if constraints:
range_kwargs = {}
if "min" in constraints:
range_kwargs["min"] = constraints["min"]
if "max" in constraints:
range_kwargs["max"] = constraints["max"]
if range_kwargs:
if field_type is int:
return click.IntRange(**range_kwargs)
else:
return click.FloatRange(**range_kwargs)
return TYPE_MAP.get(
field_type, click.STRING
) # Default to STRING if type is not found

View File

@@ -9,4 +9,4 @@ def feh_manga_viewer(image_links: list[str], window_title: str):
print("feh not found")
exit(1)
commands = [FEH_EXECUTABLE, *image_links, "--title", window_title]
subprocess.run(commands)
subprocess.run(commands, check=False)

View File

@@ -5,9 +5,9 @@ import termios
import tty
from sys import exit
from rich.align import Align
from rich.console import Console
from rich.panel import Panel
from rich.align import Align
from rich.text import Text
console = Console()
@@ -72,7 +72,8 @@ def icat_manga_viewer(image_links: list[str], window_title: str):
"--z-index",
"-1",
image_links[idx],
]
],
check=False,
)
if show_banner:

View File

@@ -0,0 +1,40 @@
import importlib
import click
class LazyGroup(click.Group):
def __init__(self, root:str, *args, lazy_subcommands=None, **kwargs):
super().__init__(*args, **kwargs)
# lazy_subcommands is a map of the form:
#
# {command-name} -> {module-name}.{command-object-name}
#
self.root = root
self.lazy_subcommands = lazy_subcommands or {}
def list_commands(self, ctx):
base = super().list_commands(ctx)
lazy = sorted(self.lazy_subcommands.keys())
return base + lazy
def get_command(self, ctx, cmd_name): # pyright:ignore
if cmd_name in self.lazy_subcommands:
return self._lazy_load(cmd_name)
return super().get_command(ctx, cmd_name)
def _lazy_load(self, cmd_name: str):
# lazily loading a command, first get the module name and attribute name
import_path: str = self.lazy_subcommands[cmd_name]
modname, cmd_object_name = import_path.rsplit(".", 1)
# do the import
mod = importlib.import_module(f".{modname}", package=self.root)
# get the Command object from that module
cmd_object = getattr(mod, cmd_object_name)
# check the result to make debugging easier
if not isinstance(cmd_object, click.Command):
raise ValueError(
f"Lazy loading of {import_path} failed by returning "
"a non-command object"
)
return cmd_object

View File

@@ -0,0 +1,30 @@
import logging
from rich.traceback import install as rich_install
from ..constants import LOG_FILE_PATH
def setup_logging(log: bool, log_file: bool, rich_traceback: bool) -> None:
"""Configures the application's logging based on CLI flags."""
if rich_traceback:
rich_install(show_locals=True)
if log:
from rich.logging import RichHandler
logging.basicConfig(
level="DEBUG",
format="%(message)s",
datefmt="[%X]",
handlers=[RichHandler()],
)
logging.getLogger(__name__).info("Rich logging initialized.")
elif log_file:
logging.basicConfig(
level="DEBUG",
filename=LOG_FILE_PATH,
format="%(asctime)s %(levelname)s: %(message)s",
datefmt="[%d/%m/%Y@%H:%M:%S]",
filemode="w",
)
else:
logging.basicConfig(level="CRITICAL")

View File

@@ -64,6 +64,7 @@ def stream_video(MPV, url, mpv_args, custom_args, pre_args=[]):
capture_output=True,
text=True,
encoding="utf-8",
check=False,
)
if proc.stdout:
for line in reversed(proc.stdout.split("\n")):
@@ -101,7 +102,7 @@ def run_mpv(
time.sleep(120)
return "0", "0"
cmd = [WEBTORRENT_CLI, link, f"--{player}"]
subprocess.run(cmd, encoding="utf-8")
subprocess.run(cmd, encoding="utf-8", check=False)
return "0", "0"
if player == "vlc":
VLC = shutil.which("vlc")
@@ -141,7 +142,7 @@ def run_mpv(
title,
]
subprocess.run(args)
subprocess.run(args, check=False)
return "0", "0"
else:
args = ["vlc", link]
@@ -152,7 +153,7 @@ def run_mpv(
if title:
args.append("--video-title")
args.append(title)
subprocess.run(args, encoding="utf-8")
subprocess.run(args, encoding="utf-8", check=False)
return "0", "0"
else:
# Determine if mpv is available
@@ -191,7 +192,7 @@ def run_mpv(
"is.xyz.mpv/.MPVActivity",
]
subprocess.run(args)
subprocess.run(args, check=False)
return "0", "0"
else:
# General mpv command with custom arguments

View File

@@ -20,7 +20,7 @@ def format_time(duration_in_secs: float):
return f"{int(h):2d}:{int(m):2d}:{int(s):2d}".replace(" ", "0")
class MpvPlayer(object):
class MpvPlayer:
anime_provider: "AnimeProvider"
config: "Config"
subs = []
@@ -97,8 +97,7 @@ class MpvPlayer(object):
else:
self.mpv_player.show_text("Fetching previous episode...")
prev_episode = total_episodes.index(current_episode_number) - 1
if prev_episode <= 0:
prev_episode = 0
prev_episode = max(0, prev_episode)
fastanime_runtime_state.provider_current_episode_number = total_episodes[
prev_episode
]

View File

@@ -11,7 +11,7 @@ def print_img(url: str):
url: [TODO:description]
"""
if EXECUTABLE := shutil.which("icat"):
subprocess.run([EXECUTABLE, url])
subprocess.run([EXECUTABLE, url], check=False)
else:
EXECUTABLE = shutil.which("chafa")
@@ -30,4 +30,4 @@ def print_img(url: str):
subprocess.run([EXECUTABLE, url, "--size=15x15"], input=img_bytes)
"""
subprocess.run([EXECUTABLE, "--size=15x15"], input=img_bytes)
subprocess.run([EXECUTABLE, "--size=15x15"], input=img_bytes, check=False)

View File

@@ -27,7 +27,8 @@ def SyncPlayer(url: str, anime_title=None, headers={}, subtitles=[], *args):
[
SYNCPLAY_EXECUTABLE,
url,
]
],
check=False,
)
else:
subprocess.run(
@@ -37,7 +38,8 @@ def SyncPlayer(url: str, anime_title=None, headers={}, subtitles=[], *args):
"--",
f"--force-media-title={anime_title}",
*mpv_args,
]
],
check=False,
)
# for compatability

View File

@@ -1,13 +1,14 @@
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from typing import Any, Callable
from collections.abc import Callable
from typing import Any
from ...libs.anilist.types import AnilistBaseMediaDataSchema
from ...libs.anime_provider.types import Anime, EpisodeStream, SearchResult, Server
class FastAnimeRuntimeState(object):
class FastAnimeRuntimeState:
"""A class that manages fastanime runtime during anilist command runtime"""
provider_current_episode_stream_link: str

View File

@@ -9,7 +9,7 @@ import sys
import requests
from rich import print
from .. import APP_NAME, AUTHOR, GIT_REPO, __version__
from ... import APP_NAME, AUTHOR, GIT_REPO, __version__
API_URL = f"https://api.{GIT_REPO}/repos/{AUTHOR}/{APP_NAME}/releases/latest"
@@ -95,7 +95,9 @@ def update_app(force=False):
print("[red]Cannot find nix, it looks like your system is broken.[/]")
return False, release_json
process = subprocess.run([NIX, "profile", "upgrade", APP_NAME.lower()])
process = subprocess.run(
[NIX, "profile", "upgrade", APP_NAME.lower()], check=False
)
elif is_git_repo(AUTHOR, APP_NAME):
GIT_EXECUTABLE = shutil.which("git")
args = [
@@ -111,31 +113,31 @@ def update_app(force=False):
process = subprocess.run(
args,
check=False,
)
elif UV := shutil.which("uv"):
process = subprocess.run([UV, "tool", "upgrade", APP_NAME], check=False)
elif PIPX := shutil.which("pipx"):
process = subprocess.run([PIPX, "upgrade", APP_NAME], check=False)
else:
if UV := shutil.which("uv"):
process = subprocess.run([UV, "tool", "upgrade", APP_NAME])
elif PIPX := shutil.which("pipx"):
process = subprocess.run([PIPX, "upgrade", APP_NAME])
else:
PYTHON_EXECUTABLE = sys.executable
PYTHON_EXECUTABLE = sys.executable
args = [
PYTHON_EXECUTABLE,
"-m",
"pip",
"install",
APP_NAME,
"-U",
"--no-warn-script-location",
]
if sys.prefix == sys.base_prefix:
# ensure NOT in a venv, where --user flag can cause an error.
# TODO: Get value of 'include-system-site-packages' in pyenv.cfg.
args.append("--user")
args = [
PYTHON_EXECUTABLE,
"-m",
"pip",
"install",
APP_NAME,
"-U",
"--no-warn-script-location",
]
if sys.prefix == sys.base_prefix:
# ensure NOT in a venv, where --user flag can cause an error.
# TODO: Get value of 'include-system-site-packages' in pyenv.cfg.
args.append("--user")
process = subprocess.run(args)
process = subprocess.run(args, check=False)
if process.returncode == 0:
print(
"[green]Its recommended to run the following after updating:\n\tfastanime config --update (to get the latest config docs)\n\tfastanime cache --clean (to get rid of any potential issues)[/]",

View File

@@ -39,8 +39,7 @@ def get_requested_quality_or_default_to_first(url, quality):
m3u8_format["height"] < quality_u and m3u8_format["height"] > quality_l
):
return m3u8_format["url"]
else:
return m3u8_formats[0]["url"]
return m3u8_formats[0]["url"]
def move_preferred_subtitle_lang_to_top(sub_list, lang_str):
@@ -78,20 +77,19 @@ def filter_by_quality(quality: str, stream_links: "list[EpisodeStream]", default
# some providers have inaccurate/weird/non-standard eg qualities 718 instead of 720
if Q <= q + 80 and Q >= q - 80:
return stream_link
else:
if stream_links and default:
from rich import print
if stream_links and default:
from rich import print
try:
print("[yellow bold]WARNING Qualities were:[/] ", stream_links)
print(
"[cyan bold]Using default of quality:[/] ",
stream_links[0]["quality"],
)
return stream_links[0]
except Exception as e:
print(e)
return
try:
print("[yellow bold]WARNING Qualities were:[/] ", stream_links)
print(
"[cyan bold]Using default of quality:[/] ",
stream_links[0]["quality"],
)
return stream_links[0]
except Exception as e:
print(e)
return
def format_bytes_to_human(num_of_bytes: float, suffix: str = "B"):
@@ -195,4 +193,8 @@ def which_bashlike():
Returns:
the path to the bash executable or None if not found
"""
return (shutil.which("bash") or "bash") if S_PLATFORM != "win32" else which_win32_gitbash()
return (
(shutil.which("bash") or "bash")
if S_PLATFORM != "win32"
else which_win32_gitbash()
)

View File

@@ -1,5 +1,5 @@
from importlib import resources
import os
from importlib import resources
APP_NAME = os.environ.get("FASTANIME_APPNAME", "fastanime")

View File

@@ -1,6 +1,3 @@
from typing import Optional
class FastAnimeError(Exception):
"""
Base exception for all custom errors raised by the FastAnime library and application.
@@ -34,7 +31,7 @@ class DependencyNotFoundError(FastAnimeError):
This indicates a problem with the user's environment setup.
"""
def __init__(self, dependency_name: str, hint: Optional[str] = None):
def __init__(self, dependency_name: str, hint: str | None = None):
self.dependency_name = dependency_name
message = (
f"Required dependency '{dependency_name}' not found in your system's PATH."
@@ -71,7 +68,7 @@ class ProviderAPIError(ProviderError):
"""
def __init__(
self, provider_name: str, http_status: Optional[int] = None, details: str = ""
self, provider_name: str, http_status: int | None = None, details: str = ""
):
self.http_status = http_status
message = "An API communication error occurred."

View File

@@ -80,7 +80,7 @@ class AniListApi:
return
if not success or not user:
return
user_info: "AnilistUser_" = user["data"]["Viewer"]
user_info: AnilistUser_ = user["data"]["Viewer"]
self.user_id = user_info["id"]
return user_info

View File

@@ -2,7 +2,7 @@ from .allanime.constants import SERVERS_AVAILABLE as ALLANIME_SERVERS
from .animepahe.constants import SERVERS_AVAILABLE as ANIMEPAHE_SERVERS
from .hianime.constants import SERVERS_AVAILABLE as HIANIME_SERVERS
anime_sources = {
PROVIDERS_AVAILABLE = {
"allanime": "api.AllAnime",
"animepahe": "api.AnimePahe",
"hianime": "api.HiAnime",

View File

@@ -200,7 +200,6 @@ class AllAnime(AnimeProvider):
"""
url = embed.get("sourceUrl")
#
if not url:
return
if url.startswith("--"):
@@ -498,4 +497,4 @@ if __name__ == "__main__":
for header_name, header_value in headers.items():
mpv_headers += f"{header_name}:{header_value},"
mpv_args.append(mpv_headers)
subprocess.run(mpv_args)
subprocess.run(mpv_args, check=False)

View File

@@ -37,7 +37,7 @@ class AnimePahe(AnimeProvider):
ANIMEPAHE_ENDPOINT, params={"m": "search", "q": search_keywords}
)
response.raise_for_status()
data: "AnimePaheSearchPage" = response.json()
data: AnimePaheSearchPage = response.json()
results = []
for result in data["data"]:
results.append(
@@ -81,9 +81,8 @@ class AnimePahe(AnimeProvider):
response.raise_for_status()
if not data:
data.update(response.json())
else:
if ep_data := response.json().get("data"):
data["data"].extend(ep_data)
elif ep_data := response.json().get("data"):
data["data"].extend(ep_data)
if response.json()["next_page_url"]:
# TODO: Refine this
time.sleep(
@@ -110,15 +109,15 @@ class AnimePahe(AnimeProvider):
page=page,
standardized_episode_number=standardized_episode_number,
)
else:
for episode in data.get("data", []):
if episode["episode"] % 1 == 0:
standardized_episode_number += 1
episode.update({"episode": standardized_episode_number})
else:
standardized_episode_number += episode["episode"] % 1
episode.update({"episode": standardized_episode_number})
standardized_episode_number = int(standardized_episode_number)
else:
for episode in data.get("data", []):
if episode["episode"] % 1 == 0:
standardized_episode_number += 1
episode.update({"episode": standardized_episode_number})
else:
standardized_episode_number += episode["episode"] % 1
episode.update({"episode": standardized_episode_number})
standardized_episode_number = int(standardized_episode_number)
return data
@debug_provider
@@ -126,8 +125,8 @@ class AnimePahe(AnimeProvider):
page = 1
standardized_episode_number = 0
if d := self.store.get(str(session_id), "search_result"):
anime_result: "AnimePaheSearchResult" = d
data: "AnimePaheAnimePage" = {} # pyright:ignore
anime_result: AnimePaheSearchResult = d
data: AnimePaheAnimePage = {} # pyright:ignore
data = self._pages_loader(
data,
@@ -335,4 +334,4 @@ if __name__ == "__main__":
for header_name, header_value in headers.items():
mpv_headers += f"{header_name}:{header_value},"
mpv_args.append(mpv_headers)
subprocess.run(mpv_args)
subprocess.run(mpv_args, check=False)

View File

@@ -67,7 +67,7 @@ if __name__ == "__main__":
# Testing time
filepath = input("Enter file name: ")
if filepath:
with open(filepath, "r") as file:
with open(filepath) as file:
data = file.read()
else:
data = """<script>eval(function(p,a,c,k,e,d){e=function(c){return(c<a?'':e(parseInt(c/a)))+((c=c%a)>35?String.fromCharCode(c+29):c.toString(36))};if(!''.replace(/^/,String)){while(c--){d[e(c)]=k[c]||e(c)}k=[function(e){return d[e]}];e=function(){return'\\w+'};c=1};while(c--){if(k[c]){p=p.replace(new RegExp('\\b'+e(c)+'\\b','g'),k[c])}}return p}('f $7={H:a(2){4 B(9.7.h(y z("(?:(?:^|.*;)\\\\s*"+d(2).h(/[\\-\\.\\+\\*]/g,"\\\\$&")+"\\\\s*\\\\=\\\\s*([^;]*).*$)|^.*$"),"$1"))||G},E:a(2,q,3,6,5,t){k(!2||/^(?:8|r\\-v|o|m|p)$/i.D(2)){4 w}f b="";k(3){F(3.J){j K:b=3===P?"; 8=O, I N Q M:u:u A":"; r-v="+3;n;j L:b="; 8="+3;n;j S:b="; 8="+3.Z();n}}9.7=d(2)+"="+d(q)+b+(5?"; m="+5:"")+(6?"; o="+6:"")+(t?"; p":"");4 x},Y:a(2,6,5){k(!2||!11.C(2)){4 w}9.7=d(2)+"=; 8=12, R 10 W l:l:l A"+(5?"; m="+5:"")+(6?"; o="+6:"");4 x},C:a(2){4(y z("(?:^|;\\\\s*)"+d(2).h(/[\\-\\.\\+\\*]/g,"\\\\$&")+"\\\\s*\\\\=")).D(9.7)},X:a(){f c=9.7.h(/((?:^|\\s*;)[^\\=]+)(?=;|$)|^\\s*|\\s*(?:\\=[^;]*)?(?:\\1|$)/g,"").T(/\\s*(?:\\=[^;]*)?;\\s*/);U(f e=0;e<c.V;e++){c[e]=B(c[e])}4 c}};',62,65,'||sKey|vEnd|return|sDomain|sPath|cookie|expires|document|function|sExpires|aKeys|encodeURIComponent|nIdx|var||replace||case|if|00|domain|break|path|secure|sValue|max||bSecure|59|age|false|true|new|RegExp|GMT|decodeURIComponent|hasItem|test|setItem|switch|null|getItem|31|constructor|Number|String|23|Dec|Fri|Infinity|9999|01|Date|split|for|length|1970|keys|removeItem|toUTCString|Jan|this|Thu'.split('|'),0,{}));eval(function(p,a,c,k,e,d){e=function(c){return(c<a?'':e(parseInt(c/a)))+((c=c%a)>35?String.fromCharCode(c+29):c.toString(36))};if(!''.replace(/^/,String)){while(c--){d[e(c)]=k[c]||e(c)}k=[function(e){return d[e]}];e=function(){return'\\w+'};c=1};while(c--){if(k[c]){p=p.replace(new RegExp('\\b'+e(c)+'\\b','g'),k[c])}}return p}('h o=\'1D://1C-E.1B.1A.1z/1y/E/1x/1w/1v.1u\';h d=s.r(\'d\');h 0=B 1t(d,{\'1s\':{\'1r\':i},\'1q\':\'16:9\',\'D\':1,\'1p\':5,\'1o\':{\'1n\':\'1m\'},1l:[\'7-1k\',\'7\',\'1j\',\'1i-1h\',\'1g\',\'1f-1e\',\'1d\',\'D\',\'1c\',\'1b\',\'1a\',\'19\',\'C\',\'18\'],\'C\':{\'17\':i}});8(!A.15()){d.14=o}x{j z={13:12,11:10,Z:Y,X:i,W:i};h c=B A(z);c.V(o);c.U(d);g.c=c}0.3("T",6=>{g.S.R.Q("P")});0.O=1;k v(b,n,m){8(b.y){b.y(n,m,N)}x 8(b.w){b.w(\'3\'+n,m)}}j 4=k(l){g.M.L(l,\'*\')};v(g,\'l\',k(e){j a=e.a;8(a===\'7\')0.7();8(a===\'f\')0.f();8(a===\'u\')0.u()});0.3(\'t\',6=>{4(\'t\')});0.3(\'7\',6=>{4(\'7\')});0.3(\'f\',6=>{4(\'f\')});0.3(\'K\',6=>{4(0.q);s.r(\'.J-I\').H=G(0.q.F(2))});0.3(\'p\',6=>{4(\'p\')});',62,102,'player|||on|sendMessage||event|play|if||data|element|hls|video||pause|window|const|true|var|function|message|eventHandler|eventName|source|ended|currentTime|querySelector|document|ready|stop|bindEvent|attachEvent|else|addEventListener|config|Hls|new|fullscreen|volume|01|toFixed|String|innerHTML|timestamp|ss|timeupdate|postMessage|parent|false|speed|landscape|lock|orientation|screen|enterfullscreen|attachMedia|loadSource|lowLatencyMode|enableWorker|Infinity|backBufferLength|600|maxMaxBufferLength|180|maxBufferLength|src|isSupported||iosNative|capture|airplay|pip|settings|captions|mute|time|current|progress|forward|fast|rewind|large|controls|kwik|key|storage|seekTime|ratio|global|keyboard|Plyr|m3u8|uwu|b92a392054c041a3f9c6eecabeb0e127183f44e547828447b10bca8d77523e6f|03|stream|org|nextcdn|files|eu|https'.split('|'),0,{}))</script>"""

View File

@@ -242,7 +242,7 @@ class HiAnime(AnimeProvider):
link_to_streams
)
if link_to_streams_response.ok:
juicy_streams_json: "HiAnimeStream" = (
juicy_streams_json: HiAnimeStream = (
link_to_streams_response.json()
)

View File

@@ -3,7 +3,7 @@ import json
import re
import time
from base64 import b64decode
from typing import TYPE_CHECKING, Dict, List
from typing import TYPE_CHECKING
from Crypto.Cipher import AES
@@ -28,9 +28,9 @@ class HiAnimeError(Exception):
# Adapted from https://github.com/ghoshRitesh12/aniwatch
class MegaCloud:
def __init__(self, session):
self.session: "CachedRequestsSession" = session
self.session: CachedRequestsSession = session
def extract(self, video_url: str) -> Dict:
def extract(self, video_url: str) -> dict:
try:
extracted_data = {
"tracks": [],
@@ -113,7 +113,7 @@ class MegaCloud:
except Exception as err:
raise err
def extract_variables(self, text: str) -> List[List[int]]:
def extract_variables(self, text: str) -> list[list[int]]:
regex = r"case\s*0x[0-9a-f]+:(?![^;]*=partKey)\s*\w+\s*=\s*(\w+)\s*,\s*\w+\s*=\s*(\w+);"
matches = re.finditer(regex, text)
vars_ = []
@@ -127,7 +127,7 @@ class MegaCloud:
return vars_
def get_secret(
self, encrypted_string: str, values: List[List[int]]
self, encrypted_string: str, values: list[list[int]]
) -> tuple[str, str]:
secret = []
encrypted_source_array = list(encrypted_string)

View File

@@ -32,9 +32,7 @@ class Nyaa(AnimeProvider):
@debug_provider
def search_for_anime(self, user_query: str, *args, **_):
self.search_results = search_for_anime_with_anilist(
user_query, True
) # pyright: ignore
self.search_results = search_for_anime_with_anilist(user_query, True) # pyright: ignore
self.user_query = user_query
return self.search_results
@@ -74,7 +72,7 @@ class Nyaa(AnimeProvider):
try:
url_arguments: dict[str, str] = {
"c": "1_2", # Language (English)
"q": f"{title} {'0' if len(episode_number)==1 else ''}{episode_number}", # Search Query
"q": f"{title} {'0' if len(episode_number) == 1 else ''}{episode_number}", # Search Query
}
# url_arguments["q"] = anime_title
@@ -160,7 +158,7 @@ class Nyaa(AnimeProvider):
if not torrent_anchor_tag_atrrs:
continue
torrent_file_url = (
f'{NYAA_ENDPOINT}{torrent_anchor_tag_atrrs["href"]}'
f"{NYAA_ENDPOINT}{torrent_anchor_tag_atrrs['href']}"
)
if server in servers:
link = {
@@ -235,7 +233,7 @@ class Nyaa(AnimeProvider):
if not torrent_anchor_tag_atrrs:
continue
torrent_file_url = (
f'{NYAA_ENDPOINT}{torrent_anchor_tag_atrrs["href"]}'
f"{NYAA_ENDPOINT}{torrent_anchor_tag_atrrs['href']}"
)
if server in servers:
link = {
@@ -312,7 +310,7 @@ class Nyaa(AnimeProvider):
if not torrent_anchor_tag_atrrs:
continue
torrent_file_url = (
f'{NYAA_ENDPOINT}{torrent_anchor_tag_atrrs["href"]}'
f"{NYAA_ENDPOINT}{torrent_anchor_tag_atrrs['href']}"
)
if server in servers:
link = {

View File

@@ -40,7 +40,7 @@ def give_random_quality(links):
return [
{**episode_stream, "quality": quality}
for episode_stream, quality in zip(links, qualities)
for episode_stream, quality in zip(links, qualities, strict=False)
]

View File

@@ -78,7 +78,6 @@ class Yugen(AnimeProvider):
if excl is not None:
if "dub" in excl.lower():
languages["dub"] = 1
#
results.append(
{
"id": identifier,
@@ -200,7 +199,6 @@ class Yugen(AnimeProvider):
video_query = f"{id_num}|{episode_number}|dub"
else:
video_query = f"{id_num}|{episode_number}"
#
res = self.session.post(
f"{YUGEN_ENDPOINT}/api/embed/",

View File

@@ -67,7 +67,7 @@ def search_for_manga_with_anilist(manga_title: str):
timeout=10,
)
if response.status_code == 200:
anilist_data: "AnilistDataSchema" = response.json()
anilist_data: AnilistDataSchema = response.json()
return {
"pageInfo": anilist_data["data"]["Page"]["pageInfo"],
"results": [
@@ -133,7 +133,7 @@ query ($query: String) {
timeout=10,
)
if response.status_code == 200:
anilist_data: "AnilistDataSchema" = response.json()
anilist_data: AnilistDataSchema = response.json()
return {
"pageInfo": anilist_data["data"]["Page"]["pageInfo"],
"results": [
@@ -233,7 +233,7 @@ def get_mal_id_and_anilist_id(anime_title: str) -> "dict[str,int] | None":
json={"query": query, "variables": variables},
timeout=10,
)
anilist_data: "AnilistDataSchema" = response.json()
anilist_data: AnilistDataSchema = response.json()
if response.status_code == 200:
anime = max(
anilist_data["data"]["Page"]["media"],
@@ -291,7 +291,7 @@ def get_basic_anime_info_by_title(anime_title: str):
json={"query": query, "variables": variables},
timeout=10,
)
anilist_data: "AnilistDataSchema" = response.json()
anilist_data: AnilistDataSchema = response.json()
if response.status_code == 200:
anime = max(
anilist_data["data"]["Page"]["media"],

View File

@@ -157,10 +157,12 @@ class CachedRequestsSession(requests.Session):
response = super().request(method, url, *args, **kwargs)
if response.ok and (
force_caching
or self.is_content_type_cachable(
response.headers.get("content-type"), caching_mimetypes
or (
self.is_content_type_cachable(
response.headers.get("content-type"), caching_mimetypes
)
and len(response.content) < self.max_size
)
and len(response.content) < self.max_size
):
logger.debug("Caching the current request")
cursor.execute(

View File

@@ -19,9 +19,7 @@ class SqliteDB:
start_time = time.time()
self.connection = sqlite3.connect(self.db_path)
logger.debug(
"Successfully got a new connection in {} seconds".format(
time.time() - start_time
)
f"Successfully got a new connection in {time.time() - start_time} seconds"
)
return self.connection

View File

@@ -1,11 +1,13 @@
from pypresence import Presence
import time
from pypresence import Presence
def discord_connect(show, episode, switch):
presence = Presence(client_id = '1292070065583165512')
presence = Presence(client_id="1292070065583165512")
presence.connect()
if not switch.is_set():
presence.update(details = show, state = "Watching episode "+episode)
presence.update(details=show, state="Watching episode " + episode)
time.sleep(10)
else:
presence.close()
presence.close()

View File

@@ -3,7 +3,8 @@ import os
import shutil
import subprocess
import sys
from typing import Callable, List
from collections.abc import Callable
from typing import List
from click import clear
from rich import print
@@ -62,7 +63,7 @@ class FZF:
"--wrap",
]
def _with_filter(self, command: str, work: Callable) -> List[str]:
def _with_filter(self, command: str, work: Callable) -> list[str]:
"""ported from the fzf docs demo
Args:
@@ -125,9 +126,9 @@ class FZF:
[self.FZF_EXECUTABLE, *commands],
input=fzf_input,
stdout=subprocess.PIPE,
universal_newlines=True,
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
@@ -200,7 +201,6 @@ class FZF:
return result
def search_for_anime(self):
commands = [
"--preview",
f"{FETCH_ANIME_SCRIPT}fetch_anime_details {{}}",
@@ -225,9 +225,9 @@ class FZF:
[self.FZF_EXECUTABLE, *commands],
input="",
stdout=subprocess.PIPE,
universal_newlines=True,
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

View File

@@ -30,6 +30,7 @@ class RofiApi:
input=rofi_input,
stdout=subprocess.PIPE,
text=True,
check=False,
)
choice = result.stdout.strip()
@@ -66,6 +67,7 @@ class RofiApi:
input=rofi_input,
stdout=subprocess.PIPE,
text=True,
check=False,
)
choice = result.stdout.strip()
@@ -100,6 +102,7 @@ class RofiApi:
input=rofi_choices,
stdout=subprocess.PIPE,
text=True,
check=False,
)
choice = result.stdout.strip()
@@ -136,6 +139,7 @@ class RofiApi:
args,
stdout=subprocess.PIPE,
text=True,
check=False,
)
user_input = result.stdout.strip()

View File

@@ -1,7 +1,8 @@
from pathlib import Path
import pytest
from unittest.mock import patch
import pytest
from fastanime.cli.config.loader import ConfigLoader
from fastanime.cli.config.model import AppConfig, GeneralConfig
from fastanime.core.exceptions import ConfigError