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

feat(feedback-service): make it configurable

Browse Source
This commit is contained in:
Benexl
2025-07-29 14:00:44 +03:00
parent 87372e41be
commit 7f52d8cb39
17 changed files with 292 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
fastanime/cli/commands/anilist/commands/auth.py
View File

@@ -16,7 +16,7 @@ def auth(config: AppConfig, status: bool, logout: bool):
from ....service.feedback import FeedbackService
auth_service = AuthService("anilist")
feedback = FeedbackService(config.general.icons)
feedback = FeedbackService(config)
selector = create_selector(config)
feedback.clear_console()

2
fastanime/cli/commands/anilist/commands/download.py
View File

@@ -123,7 +123,7 @@ def download(config: AppConfig, **options: "Unpack[DownloadOptions]"):
from fastanime.libs.selectors import create_selector
from rich.progress import Progress
feedback = FeedbackService(config.general.icons)
feedback = FeedbackService(config)
selector = create_selector(config)
media_api = create_api_client(config.general.media_api, config)
provider = create_provider(config.general.provider)

2
fastanime/cli/commands/anilist/commands/notifications.py
View File

@@ -16,7 +16,7 @@ def notifications(config: AppConfig):
from ....service.auth import AuthService
feedback = FeedbackService(config.general.icons)
feedback = FeedbackService(config)
console = Console()
auth = AuthService(config.general.media_api)
api_client = create_api_client(config.general.media_api, config)

3
fastanime/cli/commands/anilist/commands/search.py
View File

@@ -19,6 +19,7 @@ from .. import examples
if TYPE_CHECKING:
from typing import TypedDict
from typing_extensions import Unpack
class SearchOptions(TypedDict, total=False):
@@ -196,7 +197,7 @@ def search(config: AppConfig, **options: "Unpack[SearchOptions]"):
from .....libs.media_api.params import MediaSearchParams
from ....service.feedback import FeedbackService
feedback = FeedbackService(config.general.icons)
feedback = FeedbackService(config)
try:
# Create API client

2
fastanime/cli/commands/anilist/commands/stats.py
View File

@@ -23,7 +23,7 @@ def stats(config: "AppConfig"):
console = Console()
feedback = FeedbackService(config.general.icons)
feedback = FeedbackService(config)
auth = AuthService(config.general.media_api)
registry_service = MediaRegistryService(
config.general.media_api, config.media_registry

61
fastanime/cli/commands/download.py
View File

@@ -11,6 +11,7 @@ if TYPE_CHECKING:
from pathlib import Path
from typing import TypedDict
from fastanime.cli.service.feedback.service import FeedbackService
from typing_extensions import Unpack
from ...libs.provider.anime.base import BaseAnimeProvider
@@ -102,8 +103,7 @@ if TYPE_CHECKING:
)
@click.pass_obj
def download(config: AppConfig, **options: "Unpack[Options]"):
from rich import print
from rich.progress import Progress
from fastanime.cli.service.feedback.service import FeedbackService
from ...core.exceptions import FastAnimeError
from ...libs.provider.anime.params import (
@@ -113,16 +113,16 @@ def download(config: AppConfig, **options: "Unpack[Options]"):
from ...libs.provider.anime.provider import create_provider
from ...libs.selectors.selector import create_selector
feedback = FeedbackService(config)
provider = create_provider(config.general.provider)
selector = create_selector(config)
anime_titles = options["anime_title"]
print(f"[green bold]Streaming:[/] {anime_titles}")
feedback.info(f"[green bold]Streaming:[/] {anime_titles}")
for anime_title in anime_titles:
# ---- search for anime ----
print(f"[green bold]Searching for:[/] {anime_title}")
with Progress() as progress:
progress.add_task("Fetching Search Results...", total=None)
feedback.info(f"[green bold]Searching for:[/] {anime_title}")
with feedback.progress(f"Fetching anime search results for {anime_title}"):
search_results = provider.search(
SearchParams(
query=anime_title, translation_type=config.stream.translation_type
@@ -144,8 +144,7 @@ def download(config: AppConfig, **options: "Unpack[Options]"):
anime_result = _search_results[selected_anime_title]
# ---- fetch selected anime ----
with Progress() as progress:
progress.add_task("Fetching Anime...", total=None)
with feedback.progress(f"Fetching {anime_result.title}"):
anime = provider.get(AnimeParams(id=anime_result.id, query=anime_title))
if not anime:
@@ -165,7 +164,14 @@ def download(config: AppConfig, **options: "Unpack[Options]"):
for episode in episodes_range:
download_anime(
config, options, provider, selector, anime, anime_title, episode
config,
options,
provider,
selector,
feedback,
anime,
anime_title,
episode,
)
except (ValueError, IndexError) as e:
raise FastAnimeError(f"Invalid episode range: {e}") from e
@@ -177,7 +183,14 @@ def download(config: AppConfig, **options: "Unpack[Options]"):
if not episode:
raise FastAnimeError("No episode selected")
download_anime(
config, options, provider, selector, anime, anime_title, episode
config,
options,
provider,
selector,
feedback,
anime,
anime_title,
episode,
)
@@ -186,35 +199,19 @@ def download_anime(
download_options: "Options",
provider: "BaseAnimeProvider",
selector: "BaseSelector",
feedback: "FeedbackService",
anime: "Anime",
anime_title: str,
episode: str,
):
from rich import print
from rich.progress import Progress
from ...core.downloader import DownloadParams, create_downloader
from ...libs.provider.anime.params import EpisodeStreamsParams
downloader = create_downloader(config.downloads)
with Progress() as progress:
progress.add_task("Fetching Episode Streams...", total=None)
streams = provider.episode_streams(
EpisodeStreamsParams(
anime_id=anime.id,
episode=episode,
query=anime_title,
translation_type=config.stream.translation_type,
)
)
if not streams:
raise FastAnimeError(
f"Failed to get streams for anime: {anime.title}, episode: {episode}"
)
with Progress() as progress:
progress.add_task("Fetching Episode Streams...", total=None)
with feedback.progress(f"Fetching episode streams"):
streams = provider.episode_streams(
EpisodeStreamsParams(
anime_id=anime.id,
@@ -229,16 +226,14 @@ def download_anime(
)
if config.stream.server.value == "TOP":
with Progress() as progress:
progress.add_task("Fetching top server...", total=None)
with feedback.progress(f"Fetching top server"):
server = next(streams, None)
if not server:
raise FastAnimeError(
f"Failed to get server for anime: {anime.title}, episode: {episode}"
)
else:
with Progress() as progress:
progress.add_task("Fetching servers", total=None)
with feedback.progress(f"Fetching servers"):
servers = {server.name: server for server in streams}
servers_names = list(servers.keys())
if config.stream.server in servers_names:
@@ -253,7 +248,7 @@ def download_anime(
raise FastAnimeError(
f"Failed to get stream link for anime: {anime.title}, episode: {episode}"
)
print(f"[green bold]Now Downloading:[/] {anime.title} Episode: {episode}")
feedback.info(f"[green bold]Now Downloading:[/] {anime.title} Episode: {episode}")
downloader.download(
DownloadParams(
url=stream_link,

29
fastanime/cli/commands/queue.py
View File

@@ -1,11 +1,16 @@
import click
from fastanime.core.config import AppConfig
from fastanime.libs.media_api.params import MediaSearchParams
from fastanime.core.exceptions import FastAnimeError
from fastanime.libs.media_api.params import MediaSearchParams
@click.command(help="Queue episodes for the background worker to download.")
@click.option("--title", "-t", required=True, multiple=True, help="Anime title to queue.")
@click.option("--episode-range", "-r", required=True, help="Range of episodes (e.g., '1-10').")
@click.option(
"--title", "-t", required=True, multiple=True, help="Anime title to queue."
)
@click.option(
"--episode-range", "-r", required=True, help="Range of episodes (e.g., '1-10')."
)
@click.pass_obj
def queue(config: AppConfig, title: tuple, episode_range: str):
"""
@@ -14,12 +19,12 @@ def queue(config: AppConfig, title: tuple, episode_range: str):
"""
from fastanime.cli.service.download.service import DownloadService
from fastanime.cli.service.feedback import FeedbackService
from fastanime.cli.service.registry import MediaRegistryService
from fastanime.cli.utils.parser import parse_episode_range
from fastanime.libs.media_api.api import create_api_client
from fastanime.libs.provider.anime.provider import create_provider
from fastanime.cli.service.registry import MediaRegistryService
feedback = FeedbackService(config.general.icons)
feedback = FeedbackService(config)
media_api = create_api_client(config.general.media_api, config)
provider = create_provider(config.general.provider)
registry = MediaRegistryService(config.general.media_api, config.media_registry)
@@ -28,7 +33,9 @@ def queue(config: AppConfig, title: tuple, episode_range: str):
for anime_title in title:
try:
feedback.info(f"Searching for '{anime_title}'...")
search_result = media_api.search_media(MediaSearchParams(query=anime_title, per_page=1))
search_result = media_api.search_media(
MediaSearchParams(query=anime_title, per_page=1)
)
if not search_result or not search_result.media:
feedback.warning(f"Could not find '{anime_title}' on AniList.")
@@ -36,14 +43,18 @@ def queue(config: AppConfig, title: tuple, episode_range: str):
media_item = search_result.media[0]
available_episodes = [str(i + 1) for i in range(media_item.episodes or 0)]
episodes_to_queue = list(parse_episode_range(episode_range, available_episodes))
episodes_to_queue = list(
parse_episode_range(episode_range, available_episodes)
)
queued_count = 0
for ep in episodes_to_queue:
if download_service.add_to_queue(media_item, ep):
queued_count += 1
feedback.success(f"Successfully queued {queued_count} episodes for '{media_item.title.english}'.")
feedback.success(
f"Successfully queued {queued_count} episodes for '{media_item.title.english}'."
)
except FastAnimeError as e:
feedback.error(f"Failed to queue '{anime_title}'", str(e))

2
fastanime/cli/commands/registry/cmd.py
View File

@@ -42,7 +42,7 @@ def registry(ctx: click.Context, api: str):
from ...service.registry import MediaRegistryService
config: AppConfig = ctx.obj
feedback = FeedbackService()
feedback = FeedbackService(config)
if ctx.invoked_subcommand is None:
# Show registry overview and statistics

233
fastanime/cli/commands/registry/commands/sync.py
View File

@@ -3,6 +3,7 @@ Registry sync command - synchronize local registry with remote media API
"""
import click
from fastanime.cli.service.feedback.service import FeedbackService
from fastanime.cli.service.registry.service import MediaRegistryService
from rich.progress import Progress
@@ -60,7 +61,7 @@ def sync(
from ....service.feedback import FeedbackService
from ....service.registry import MediaRegistryService
feedback = FeedbackService(config.general.icons)
feedback = FeedbackService(config)
auth = AuthService(config.general.media_api)
registry_service = MediaRegistryService(api, config.media_registry)
@@ -100,93 +101,91 @@ def sync(
statuses_to_sync = [status_map[s] for s in status_list]
with Progress() as progress:
if download:
_sync_download(
media_api_client,
registry_service,
statuses_to_sync,
feedback,
progress,
dry_run,
force,
)
if download:
_sync_download(
media_api_client,
registry_service,
statuses_to_sync,
feedback,
dry_run,
force,
)
if upload:
_sync_upload(
media_api_client,
registry_service,
statuses_to_sync,
feedback,
progress,
dry_run,
force,
)
if upload:
_sync_upload(
media_api_client,
registry_service,
statuses_to_sync,
feedback,
dry_run,
force,
)
feedback.success("Sync Complete", "Registry synchronization finished successfully")
def _sync_download(
api_client, registry_service, statuses, feedback, progress, dry_run, force
api_client, registry_service, statuses, feedback: "FeedbackService", dry_run, force
):
"""Download remote media list to local registry."""
from .....libs.media_api.params import UserMediaListSearchParams
feedback.info("Starting Download", "Fetching remote media lists...")
download_task = progress.add_task("Downloading media lists...", total=len(statuses))
total_downloaded = 0
total_updated = 0
with feedback.progress("Downloading media lists...", total=len(statuses)) as (
task_id,
progress,
):
for status in statuses:
try:
# Fetch all pages for this status
page = 1
while True:
params = UserMediaListSearchParams(
status=status, page=page, per_page=50
)
for status in statuses:
try:
# Fetch all pages for this status
page = 1
while True:
params = UserMediaListSearchParams(
status=status, page=page, per_page=50
)
result = api_client.search_media_list(params)
if not result or not result.media:
break
result = api_client.search_media_list(params)
if not result or not result.media:
break
for media_item in result.media:
if dry_run:
feedback.info(
"Would download",
f"{media_item.title.english or media_item.title.romaji} ({status.value})",
)
else:
# Get or create record and update with user status
record = registry_service.get_or_create_record(media_item)
# Update index entry with latest status
if media_item.user_status:
registry_service.update_media_index_entry(
media_item.id,
media_item=media_item,
status=media_item.user_status.status,
progress=str(media_item.user_status.progress or 0),
score=media_item.user_status.score,
repeat=media_item.user_status.repeat,
notes=media_item.user_status.notes,
for media_item in result.media:
if dry_run:
feedback.info(
"Would download",
f"{media_item.title.english or media_item.title.romaji} ({status.value})",
)
total_updated += 1
else:
# Get or create record and update with user status
record = registry_service.get_or_create_record(media_item)
registry_service.save_media_record(record)
total_downloaded += 1
# Update index entry with latest status
if media_item.user_status:
registry_service.update_media_index_entry(
media_item.id,
media_item=media_item,
status=media_item.user_status.status,
progress=str(media_item.user_status.progress or 0),
score=media_item.user_status.score,
repeat=media_item.user_status.repeat,
notes=media_item.user_status.notes,
)
total_updated += 1
if not result.page_info.has_next_page:
break
page += 1
registry_service.save_media_record(record)
total_downloaded += 1
except Exception as e:
feedback.error(f"Download Error ({status.value})", str(e))
continue
if not result.page_info.has_next_page:
break
page += 1
progress.advance(download_task)
except Exception as e:
feedback.error(f"Download Error ({status.value})", str(e))
continue
progress.advance(task_id) # type:ignore
if not dry_run:
feedback.success(
@@ -200,76 +199,72 @@ def _sync_upload(
registry_service: MediaRegistryService,
statuses,
feedback,
progress,
dry_run,
force,
):
"""Upload local registry changes to remote API."""
feedback.info("Starting Upload", "Syncing local changes to remote...")
upload_task = progress.add_task("Uploading changes...", total=None)
total_uploaded = 0
total_errors = 0
try:
# Get all media records from registry
all_records = registry_service.get_all_media_records()
with feedback.progress("Uploading changes..."):
try:
# Get all media records from registry
all_records = registry_service.get_all_media_records()
for record in all_records:
try:
# Get the index entry for this media
index_entry = registry_service.get_media_index_entry(
record.media_item.id
)
if not index_entry or not index_entry.status:
continue
# Only sync if status is in our target list
if index_entry.status.value not in statuses:
continue
if dry_run:
feedback.info(
"Would upload",
f"{record.media_item.title.english or record.media_item.title.romaji} "
f"({index_entry.status.value}, progress: {index_entry.progress or 0})",
)
else:
# Update remote list entry
from .....libs.media_api.params import (
UpdateUserMediaListEntryParams,
for record in all_records:
try:
# Get the index entry for this media
index_entry = registry_service.get_media_index_entry(
record.media_item.id
)
if not index_entry or not index_entry.status:
continue
update_params = UpdateUserMediaListEntryParams(
media_id=record.media_item.id,
status=index_entry.status,
progress=index_entry.progress,
score=index_entry.score,
)
# Only sync if status is in our target list
if index_entry.status.value not in statuses:
continue
if api_client.update_list_entry(update_params):
total_uploaded += 1
if dry_run:
feedback.info(
"Would upload",
f"{record.media_item.title.english or record.media_item.title.romaji} "
f"({index_entry.status.value}, progress: {index_entry.progress or 0})",
)
else:
total_errors += 1
feedback.warning(
"Upload Failed",
f"Failed to upload {record.media_item.title.english or record.media_item.title.romaji}",
# Update remote list entry
from .....libs.media_api.params import (
UpdateUserMediaListEntryParams,
)
except Exception as e:
total_errors += 1
feedback.error(
"Upload Error",
f"Failed to upload media {record.media_item.id}: {e}",
)
continue
update_params = UpdateUserMediaListEntryParams(
media_id=record.media_item.id,
status=index_entry.status,
progress=index_entry.progress,
score=index_entry.score,
)
except Exception as e:
feedback.error("Upload Error", f"Failed to get local records: {e}")
return
if api_client.update_list_entry(update_params):
total_uploaded += 1
else:
total_errors += 1
feedback.warning(
"Upload Failed",
f"Failed to upload {record.media_item.title.english or record.media_item.title.romaji}",
)
progress.remove_task(upload_task)
except Exception as e:
total_errors += 1
feedback.error(
"Upload Error",
f"Failed to upload media {record.media_item.id}: {e}",
)
continue
except Exception as e:
feedback.error("Upload Error", f"Failed to get local records: {e}")
return
if not dry_run:
feedback.success(

43
fastanime/cli/commands/search.py
View File

@@ -1,7 +1,6 @@
from typing import TYPE_CHECKING
import click
from fastanime.cli.service.player.service import PlayerService
from ...core.config import AppConfig
from ...core.exceptions import FastAnimeError
@@ -11,6 +10,7 @@ from . import examples
if TYPE_CHECKING:
from typing import TypedDict
from fastanime.cli.service.feedback.service import FeedbackService
from typing_extensions import Unpack
from ...libs.provider.anime.base import BaseAnimeProvider
@@ -42,8 +42,7 @@ if TYPE_CHECKING:
)
@click.pass_obj
def search(config: AppConfig, **options: "Unpack[Options]"):
from rich import print
from rich.progress import Progress
from fastanime.cli.service.feedback.service import FeedbackService
from ...core.exceptions import FastAnimeError
from ...libs.provider.anime.params import (
@@ -53,16 +52,16 @@ def search(config: AppConfig, **options: "Unpack[Options]"):
from ...libs.provider.anime.provider import create_provider
from ...libs.selectors.selector import create_selector
feedback = FeedbackService(config)
provider = create_provider(config.general.provider)
selector = create_selector(config)
anime_titles = options["anime_title"]
print(f"[green bold]Streaming:[/] {anime_titles}")
feedback.info(f"[green bold]Streaming:[/] {anime_titles}")
for anime_title in anime_titles:
# ---- search for anime ----
print(f"[green bold]Searching for:[/] {anime_title}")
with Progress() as progress:
progress.add_task("Fetching Search Results...", total=None)
feedback.info(f"[green bold]Searching for:[/] {anime_title}")
with feedback.progress(f"Fetching anime search results for {anime_title}"):
search_results = provider.search(
SearchParams(
query=anime_title, translation_type=config.stream.translation_type
@@ -84,8 +83,7 @@ def search(config: AppConfig, **options: "Unpack[Options]"):
anime_result = _search_results[selected_anime_title]
# ---- fetch selected anime ----
with Progress() as progress:
progress.add_task("Fetching Anime...", total=None)
with feedback.progress(f"Fetching {anime_result.title}"):
anime = provider.get(AnimeParams(id=anime_result.id, query=anime_title))
if not anime:
@@ -105,7 +103,13 @@ def search(config: AppConfig, **options: "Unpack[Options]"):
for episode in episodes_range:
stream_anime(
config, provider, selector, anime, episode, anime_title
config,
provider,
selector,
feedback,
anime,
episode,
anime_title,
)
except (ValueError, IndexError) as e:
raise FastAnimeError(f"Invalid episode range: {e}") from e
@@ -116,27 +120,28 @@ def search(config: AppConfig, **options: "Unpack[Options]"):
)
if not episode:
raise FastAnimeError("No episode selected")
stream_anime(config, provider, selector, anime, episode, anime_title)
stream_anime(
config, provider, selector, feedback, anime, episode, anime_title
)
def stream_anime(
config: AppConfig,
provider: "BaseAnimeProvider",
selector: "BaseSelector",
feedback: "FeedbackService",
anime: "Anime",
episode: str,
anime_title: str,
):
from rich import print
from rich.progress import Progress
from fastanime.cli.service.player.service import PlayerService
from ...libs.player.params import PlayerParams
from ...libs.provider.anime.params import EpisodeStreamsParams
player_service = PlayerService(config, provider)
with Progress() as progress:
progress.add_task("Fetching Episode Streams...", total=None)
with feedback.progress(f"Fetching episode streams"):
streams = provider.episode_streams(
EpisodeStreamsParams(
anime_id=anime.id,
@@ -151,16 +156,14 @@ def stream_anime(
)
if config.stream.server.value == "TOP":
with Progress() as progress:
progress.add_task("Fetching top server...", total=None)
with feedback.progress(f"Fetching top server"):
server = next(streams, None)
if not server:
raise FastAnimeError(
f"Failed to get server for anime: {anime.title}, episode: {episode}"
)
else:
with Progress() as progress:
progress.add_task("Fetching servers", total=None)
with feedback.progress(f"Fetching servers"):
servers = {server.name: server for server in streams}
servers_names = list(servers.keys())
if config.stream.server.value in servers_names:
@@ -175,7 +178,7 @@ def stream_anime(
raise FastAnimeError(
f"Failed to get stream link for anime: {anime.title}, episode: {episode}"
)
print(f"[green bold]Now Streaming:[/] {anime.title} Episode: {episode}")
feedback.info(f"[green bold]Now Streaming:[/] {anime.title} Episode: {episode}")
player_service.play(
PlayerParams(

2
fastanime/cli/commands/worker.py
View File

@@ -19,7 +19,7 @@ def worker(config: AppConfig):
from fastanime.libs.media_api.api import create_api_client
from fastanime.libs.provider.anime.provider import create_provider
feedback = FeedbackService(config.general.icons)
feedback = FeedbackService(config)
if not config.worker.enabled:
feedback.warning("Worker is disabled in the configuration. Exiting.")
return

3
fastanime/cli/interactive/menu/media/media_actions.py
View File

@@ -292,7 +292,8 @@ def _manage_user_media_list_in_bulk(ctx: Context, state: State) -> MenuAction:
)
):
print(f"Failed to update {media_item.title.english}")
progress.update(task_id, advance=1)
progress.update(task_id, advance=1) # type: ignore
return InternalDirective.RELOAD
return action

2
fastanime/cli/interactive/session.py
View File

@@ -143,7 +143,7 @@ class Context:
if not self._feedback:
from ..service.feedback.service import FeedbackService
self._feedback = FeedbackService()
self._feedback = FeedbackService(self.config)
return self._feedback
@property

26
fastanime/cli/service/feedback/service.py
View File

@@ -11,18 +11,20 @@ from rich.progress import (
TextColumn,
)
from ....core.config import AppConfig
console = Console()
class FeedbackService:
"""Centralized manager for user feedback in interactive menus."""
def __init__(self, icons_enabled: bool = True):
self.icons_enabled = icons_enabled
def __init__(self, config: AppConfig):
self.config = config
def success(self, message: str, details: Optional[str] = None) -> None:
"""Show a success message with optional details."""
icon = "✅ " if self.icons_enabled else ""
icon = "✅ " if self.config.general.icons else ""
main_msg = f"[bold green]{icon}{message}[/bold green]"
if details:
@@ -32,7 +34,7 @@ class FeedbackService:
def error(self, message: str, details: Optional[str] = None) -> None:
"""Show an error message with optional details."""
icon = "❌ " if self.icons_enabled else ""
icon = "❌ " if self.config.general.icons else ""
main_msg = f"[bold red]{icon}Error: {message}[/bold red]"
if details:
@@ -43,7 +45,7 @@ class FeedbackService:
def warning(self, message: str, details: Optional[str] = None) -> None:
"""Show a warning message with optional details."""
icon = "⚠️ " if self.icons_enabled else ""
icon = "⚠️ " if self.config.general.icons else ""
main_msg = f"[bold yellow]{icon}Warning: {message}[/bold yellow]"
if details:
@@ -53,7 +55,7 @@ class FeedbackService:
def info(self, message: str, details: Optional[str] = None) -> None:
"""Show an informational message with optional details."""
icon = "ℹ️ " if self.icons_enabled else ""
icon = "" if self.config.general.icons else ""
main_msg = f"[bold blue]{icon}{message}[/bold blue]"
if details:
@@ -67,20 +69,24 @@ class FeedbackService:
self,
message: str,
total: Optional[float] = None,
transient: bool = True,
transient: bool = False,
auto_add_task: bool = True,
success_msg: Optional[str] = None,
error_msg: Optional[str] = None,
):
"""Context manager for operations with loading indicator and result feedback."""
with Progress(
SpinnerColumn(),
SpinnerColumn(self.config.general.preferred_spinner),
TextColumn(f"[cyan]{message}..."),
BarColumn(),
TaskProgressColumn(),
transient=transient,
console=console,
) as progress:
task_id = progress.add_task("", total=total)
task_id = None
if auto_add_task:
# FIXME: for some reason task id is still none
task_id = progress.add_task("", total=total)
try:
yield task_id, progress
if success_msg:
@@ -93,7 +99,7 @@ class FeedbackService:
def pause_for_user(self, message: str = "Press Enter to continue") -> None:
"""Pause execution and wait for user input."""
icon = "⏸️ " if self.icons_enabled else ""
icon = "⏸️ " if self.config.general.icons else ""
click.pause(f"{icon}{message}...")
def clear_console(self):

1
fastanime/core/config/defaults.py
View File

@@ -3,6 +3,7 @@ from ..utils import detect
# GeneralConfig
GENERAL_PYGMENT_STYLE = "github-dark"
GENERAL_PREFERRED_SPINNER = "smiley"
GENERAL_API_CLIENT = "anilist"
GENERAL_PREFERRED_TRACKER = "local"
GENERAL_PROVIDER = "allanime"

1
fastanime/core/config/descriptions.py
View File

@@ -2,6 +2,7 @@
from .defaults import SESSIONS_DIR
GENERAL_PYGMENT_STYLE = "The pygment style to use"
GENERAL_PREFERRED_SPINNER = "The spinner to use"
GENERAL_API_CLIENT = "The media database API to use (e.g., 'anilist', 'jikan')."
GENERAL_PREFERRED_TRACKER = (
"The preferred watch history tracker (local,remote) in cases of conflicts"

78
fastanime/core/config/model.py
View File

@@ -20,6 +20,84 @@ class GeneralConfig(BaseModel):
pygment_style: str = Field(
default=defaults.GENERAL_PYGMENT_STYLE, description=desc.GENERAL_PYGMENT_STYLE
)
preferred_spinner: Literal[
"dots",
"dots2",
"dots3",
"dots4",
"dots5",
"dots6",
"dots7",
"dots8",
"dots9",
"dots10",
"dots11",
"dots12",
"dots8Bit",
"line",
"line2",
"pipe",
"simpleDots",
"simpleDotsScrolling",
"star",
"star2",
"flip",
"hamburger",
"growVertical",
"growHorizontal",
"balloon",
"balloon2",
"noise",
"bounce",
"boxBounce",
"boxBounce2",
"triangle",
"arc",
"circle",
"squareCorners",
"circleQuarters",
"circleHalves",
"squish",
"toggle",
"toggle2",
"toggle3",
"toggle4",
"toggle5",
"toggle6",
"toggle7",
"toggle8",
"toggle9",
"toggle10",
"toggle11",
"toggle12",
"toggle13",
"arrow",
"arrow2",
"arrow3",
"bouncingBar",
"bouncingBall",
"smiley",
"monkey",
"hearts",
"clock",
"earth",
"material",
"moon",
"runner",
"pong",
"shark",
"dqpb",
"weather",
"christmas",
"grenade",
"point",
"layer",
"betaWave",
"aesthetic",
] = Field(
default=defaults.GENERAL_PREFERRED_SPINNER,
description=desc.GENERAL_PREFERRED_SPINNER,
)
media_api: Literal["anilist", "jikan"] = Field(
default=defaults.GENERAL_API_CLIENT,
description=desc.GENERAL_API_CLIENT,