feat: replace typer with cycplots & improve cli
This commit is contained in:
+175
-134
@@ -4,67 +4,86 @@ Date: 2026-03-26 02:04:39
|
||||
Description: CLI interface
|
||||
"""
|
||||
|
||||
import typer
|
||||
import sys
|
||||
import time
|
||||
from typing import Optional
|
||||
from loguru import logger
|
||||
import os
|
||||
from typing import Annotated
|
||||
import cyclopts
|
||||
from loguru import logger
|
||||
|
||||
from .config import enable_debug
|
||||
from .models import TrackMeta, CacheStatus
|
||||
from .mpris import get_current_track
|
||||
from .core import LrcManager
|
||||
from .core import LrcManager, FetcherMethodType
|
||||
|
||||
app = typer.Typer(
|
||||
|
||||
app = cyclopts.App(
|
||||
help="LRCFetch — Fetch line-synced lyrics for your music player.",
|
||||
add_completion=True,
|
||||
)
|
||||
app.register_install_completion_command()
|
||||
|
||||
cache_app = cyclopts.App(name="cache", help="Manage the local SQLite cache.")
|
||||
app.command(cache_app)
|
||||
|
||||
manager = LrcManager()
|
||||
|
||||
# Global state set by the app callback
|
||||
_player: Optional[str] = None
|
||||
# Global state set by the meta launcher
|
||||
_player: str | None = None
|
||||
|
||||
|
||||
@app.callback()
|
||||
def main(
|
||||
debug: bool = typer.Option(False, "--debug", "-d", help="Enable debug logging."),
|
||||
player: Optional[str] = typer.Option(
|
||||
None,
|
||||
"--player",
|
||||
"-p",
|
||||
help="Target a specific MPRIS player using its DBus name or a portion thereof.",
|
||||
),
|
||||
@app.meta.default
|
||||
def launcher(
|
||||
*tokens: Annotated[str, cyclopts.Parameter(show=False, allow_leading_hyphen=True)],
|
||||
debug: Annotated[
|
||||
bool,
|
||||
cyclopts.Parameter(
|
||||
name=["--debug", "-d"], negative="", help="Enable debug logging."
|
||||
),
|
||||
] = False,
|
||||
player: Annotated[
|
||||
str | None,
|
||||
cyclopts.Parameter(
|
||||
name=["--player", "-p"],
|
||||
help="Target a specific MPRIS player using its DBus name or a portion thereof.",
|
||||
),
|
||||
] = None,
|
||||
):
|
||||
global _player
|
||||
if debug:
|
||||
enable_debug()
|
||||
_player = player
|
||||
app(tokens)
|
||||
|
||||
|
||||
# fetch
|
||||
|
||||
|
||||
@app.command()
|
||||
@app.command
|
||||
def fetch(
|
||||
method: Optional[str] = typer.Option(
|
||||
None,
|
||||
"--method",
|
||||
help="Force a specific source (local, spotify, lrclib, lrclib-search, netease).",
|
||||
),
|
||||
no_cache: bool = typer.Option(
|
||||
False, "--no-cache", help="Bypass the cache for this request."
|
||||
),
|
||||
only_synced: bool = typer.Option(
|
||||
False, "--only-synced", help="Only accept synced (timed) lyrics."
|
||||
),
|
||||
*,
|
||||
method: Annotated[
|
||||
FetcherMethodType | None,
|
||||
cyclopts.Parameter(help="Force a specific source."),
|
||||
] = None,
|
||||
no_cache: Annotated[
|
||||
bool,
|
||||
cyclopts.Parameter(
|
||||
name="--no-cache", negative="", help="Bypass the cache for this request."
|
||||
),
|
||||
] = False,
|
||||
only_synced: Annotated[
|
||||
bool,
|
||||
cyclopts.Parameter(
|
||||
name="--only-synced", negative="", help="Only accept synced (timed) lyrics."
|
||||
),
|
||||
] = False,
|
||||
):
|
||||
"""Fetch and print lyrics for the currently playing track."""
|
||||
track = get_current_track(_player)
|
||||
|
||||
if not track:
|
||||
logger.error("No active playing track found.")
|
||||
raise typer.Exit(1)
|
||||
sys.exit(1)
|
||||
|
||||
logger.info(f"Track: {track.display_name()}")
|
||||
|
||||
@@ -72,11 +91,11 @@ def fetch(
|
||||
|
||||
if not result or not result.lyrics:
|
||||
logger.error("No lyrics found.")
|
||||
raise typer.Exit(1)
|
||||
sys.exit(1)
|
||||
|
||||
if only_synced and result.status != CacheStatus.SUCCESS_SYNCED:
|
||||
logger.error("Only unsynced lyrics available (--only-synced requested).")
|
||||
raise typer.Exit(1)
|
||||
sys.exit(1)
|
||||
|
||||
print(result.lyrics)
|
||||
|
||||
@@ -84,27 +103,41 @@ def fetch(
|
||||
# search
|
||||
|
||||
|
||||
@app.command()
|
||||
@app.command
|
||||
def search(
|
||||
title: str = typer.Option(..., "--title", "-t", help="Track title."),
|
||||
artist: Optional[str] = typer.Option(None, "--artist", "-a", help="Artist name."),
|
||||
album: Optional[str] = typer.Option(None, "--album", help="Album name."),
|
||||
trackid: Optional[str] = typer.Option(None, "--trackid", help="Spotify track ID."),
|
||||
length: Optional[int] = typer.Option(
|
||||
None, "--length", "-l", help="Track duration in milliseconds."
|
||||
),
|
||||
url: Optional[str] = typer.Option(
|
||||
None, "--url", help="Local file URL (file:///...)."
|
||||
),
|
||||
method: Optional[str] = typer.Option(
|
||||
None, "--method", help="Force a specific source."
|
||||
),
|
||||
no_cache: bool = typer.Option(
|
||||
False, "--no-cache", help="Bypass the cache for this request."
|
||||
),
|
||||
only_synced: bool = typer.Option(
|
||||
False, "--only-synced", help="Only accept synced (timed) lyrics."
|
||||
),
|
||||
*,
|
||||
title: Annotated[
|
||||
str, cyclopts.Parameter(name=["--title", "-t"], help="Track title.")
|
||||
],
|
||||
artist: Annotated[
|
||||
str | None, cyclopts.Parameter(name=["--artist", "-a"], help="Artist name.")
|
||||
] = None,
|
||||
album: Annotated[str | None, cyclopts.Parameter(help="Album name.")] = None,
|
||||
trackid: Annotated[str | None, cyclopts.Parameter(help="Spotify track ID.")] = None,
|
||||
length: Annotated[
|
||||
int | None,
|
||||
cyclopts.Parameter(
|
||||
name=["--length", "-l"], help="Track duration in milliseconds."
|
||||
),
|
||||
] = None,
|
||||
url: Annotated[
|
||||
str | None, cyclopts.Parameter(help="Local file URL (file:///...).")
|
||||
] = None,
|
||||
method: Annotated[
|
||||
FetcherMethodType | None, cyclopts.Parameter(help="Force a specific source.")
|
||||
] = None,
|
||||
no_cache: Annotated[
|
||||
bool,
|
||||
cyclopts.Parameter(
|
||||
name="--no-cache", negative="", help="Bypass the cache for this request."
|
||||
),
|
||||
] = False,
|
||||
only_synced: Annotated[
|
||||
bool,
|
||||
cyclopts.Parameter(
|
||||
name="--only-synced", negative="", help="Only accept synced (timed) lyrics."
|
||||
),
|
||||
] = False,
|
||||
):
|
||||
"""Search for lyrics by metadata (bypasses MPRIS)."""
|
||||
track = TrackMeta(
|
||||
@@ -122,11 +155,11 @@ def search(
|
||||
|
||||
if not result or not result.lyrics:
|
||||
logger.error("No lyrics found.")
|
||||
raise typer.Exit(1)
|
||||
sys.exit(1)
|
||||
|
||||
if only_synced and result.status != CacheStatus.SUCCESS_SYNCED:
|
||||
logger.error("Only unsynced lyrics available (--only-synced requested).")
|
||||
raise typer.Exit(1)
|
||||
sys.exit(1)
|
||||
|
||||
print(result.lyrics)
|
||||
|
||||
@@ -134,32 +167,39 @@ def search(
|
||||
# export
|
||||
|
||||
|
||||
@app.command()
|
||||
@app.command
|
||||
def export(
|
||||
output: Optional[str] = typer.Option(
|
||||
None,
|
||||
"--output",
|
||||
"-o",
|
||||
help="Output file path (default: <Artist> - <Title>.lrc).",
|
||||
),
|
||||
method: Optional[str] = typer.Option(
|
||||
None, "--method", help="Force a specific source."
|
||||
),
|
||||
no_cache: bool = typer.Option(False, "--no-cache", help="Bypass cache."),
|
||||
overwrite: bool = typer.Option(
|
||||
False, "--overwrite", "-f", help="Overwrite existing file."
|
||||
),
|
||||
*,
|
||||
output: Annotated[
|
||||
str | None,
|
||||
cyclopts.Parameter(
|
||||
name=["--output", "-o"],
|
||||
help="Output file path (default: <Artist> - <Title>.lrc).",
|
||||
),
|
||||
] = None,
|
||||
method: Annotated[
|
||||
FetcherMethodType | None, cyclopts.Parameter(help="Force a specific source.")
|
||||
] = None,
|
||||
no_cache: Annotated[
|
||||
bool, cyclopts.Parameter(name="--no-cache", negative="", help="Bypass cache.")
|
||||
] = False,
|
||||
overwrite: Annotated[
|
||||
bool,
|
||||
cyclopts.Parameter(
|
||||
name=["--overwrite", "-f"], negative="", help="Overwrite existing file."
|
||||
),
|
||||
] = False,
|
||||
):
|
||||
"""Export lyrics of the current track to a .lrc file."""
|
||||
track = get_current_track(_player)
|
||||
if not track:
|
||||
logger.error("No active playing track found.")
|
||||
raise typer.Exit(1)
|
||||
sys.exit(1)
|
||||
|
||||
result = manager.fetch_for_track(track, force_method=method, bypass_cache=no_cache)
|
||||
if not result or not result.lyrics:
|
||||
logger.error("No lyrics available to export.")
|
||||
raise typer.Exit(1)
|
||||
sys.exit(1)
|
||||
|
||||
# Build default output path
|
||||
if not output:
|
||||
@@ -176,7 +216,7 @@ def export(
|
||||
|
||||
if os.path.exists(output) and not overwrite:
|
||||
logger.error(f"File exists: {output} (use -f to overwrite)")
|
||||
raise typer.Exit(1)
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
with open(output, "w", encoding="utf-8") as f:
|
||||
@@ -184,69 +224,22 @@ def export(
|
||||
logger.info(f"Exported lyrics to {output}")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to write file: {e}")
|
||||
raise typer.Exit(1)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
# cache
|
||||
# cache subcommands
|
||||
|
||||
|
||||
@app.command()
|
||||
def cache(
|
||||
clear: bool = typer.Option(False, "--clear", help="Clear the entire cache."),
|
||||
clear_current: bool = typer.Option(
|
||||
False, "--clear-current", help="Clear cache for the current track."
|
||||
),
|
||||
prune: bool = typer.Option(False, "--prune", help="Remove expired entries."),
|
||||
stats: bool = typer.Option(False, "--stats", help="Show cache statistics."),
|
||||
query: bool = typer.Option(
|
||||
False, "--query", "-q", help="Show detailed cache info for the current track."
|
||||
),
|
||||
query_all: bool = typer.Option(
|
||||
False, "--query-all", help="Dump all cache entries."
|
||||
),
|
||||
@cache_app.command
|
||||
def query(
|
||||
*,
|
||||
all: Annotated[
|
||||
bool,
|
||||
cyclopts.Parameter(name="--all", negative="", help="Dump all cache entries."),
|
||||
] = False,
|
||||
):
|
||||
"""Manage the local SQLite cache."""
|
||||
if clear:
|
||||
manager.cache.clear_all()
|
||||
return
|
||||
|
||||
if clear_current:
|
||||
track = get_current_track(_player)
|
||||
if not track:
|
||||
logger.error("No active playing track found.")
|
||||
raise typer.Exit(1)
|
||||
manager.cache.clear_track(track)
|
||||
return
|
||||
|
||||
if prune:
|
||||
manager.cache.prune()
|
||||
return
|
||||
|
||||
if stats:
|
||||
s = manager.cache.stats()
|
||||
print("=== Cache Statistics ===")
|
||||
print(f"Total entries : {s['total']}")
|
||||
print(f"Active : {s['active']}")
|
||||
print(f"Expired : {s['expired']}")
|
||||
if s["by_status"]:
|
||||
print("\nBy status:")
|
||||
for status, count in s["by_status"].items():
|
||||
print(f" {status}: {count}")
|
||||
if s["by_source"]:
|
||||
print("\nBy source:")
|
||||
for source, count in s["by_source"].items():
|
||||
print(f" {source}: {count}")
|
||||
return
|
||||
|
||||
if query:
|
||||
track = get_current_track(_player)
|
||||
if not track:
|
||||
logger.error("No active playing track found.")
|
||||
raise typer.Exit(1)
|
||||
_print_track_cache(track)
|
||||
return
|
||||
|
||||
if query_all:
|
||||
"""Show cached entries for the current track."""
|
||||
if all:
|
||||
rows = manager.cache.query_all()
|
||||
if not rows:
|
||||
print("Cache is empty.")
|
||||
@@ -256,10 +249,58 @@ def cache(
|
||||
print()
|
||||
return
|
||||
|
||||
logger.info(
|
||||
"No action specified. Try --stats, --query, --query-all, "
|
||||
"--prune, --clear, or --clear-current."
|
||||
)
|
||||
track = get_current_track(_player)
|
||||
if not track:
|
||||
logger.error("No active playing track found.")
|
||||
sys.exit(1)
|
||||
_print_track_cache(track)
|
||||
|
||||
|
||||
@cache_app.command
|
||||
def clear(
|
||||
*,
|
||||
all: Annotated[
|
||||
bool,
|
||||
cyclopts.Parameter(name="--all", negative="", help="Clear the entire cache."),
|
||||
] = False,
|
||||
):
|
||||
"""Clear cached entries for the current track."""
|
||||
if all:
|
||||
manager.cache.clear_all()
|
||||
return
|
||||
|
||||
track = get_current_track(_player)
|
||||
if not track:
|
||||
logger.error("No active playing track found.")
|
||||
sys.exit(1)
|
||||
manager.cache.clear_track(track)
|
||||
|
||||
|
||||
@cache_app.command
|
||||
def prune():
|
||||
"""Remove expired cache entries."""
|
||||
manager.cache.prune()
|
||||
|
||||
|
||||
@cache_app.command
|
||||
def stats():
|
||||
"""Show cache statistics."""
|
||||
s = manager.cache.stats()
|
||||
print("=== Cache Statistics ===")
|
||||
print(f"Total entries : {s['total']}")
|
||||
print(f"Active : {s['active']}")
|
||||
print(f"Expired : {s['expired']}")
|
||||
if s["by_status"]:
|
||||
print("\nBy status:")
|
||||
for status, count in s["by_status"].items():
|
||||
print(f" {status}: {count}")
|
||||
if s["by_source"]:
|
||||
print("\nBy source:")
|
||||
for source, count in s["by_source"].items():
|
||||
print(f" {source}: {count}")
|
||||
|
||||
|
||||
# helpers
|
||||
|
||||
|
||||
def _print_track_cache(track: TrackMeta) -> None:
|
||||
@@ -317,7 +358,7 @@ def _print_cache_row(row: dict, indent: str = "") -> None:
|
||||
|
||||
|
||||
def run():
|
||||
app()
|
||||
app.meta()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
+12
-3
@@ -14,6 +14,7 @@ Fetch pipeline:
|
||||
|
||||
from typing import Optional
|
||||
from loguru import logger
|
||||
from typing import Literal
|
||||
|
||||
from .fetchers.netease import NeteaseFetcher
|
||||
from .fetchers.lrclib_search import LrclibSearchFetcher
|
||||
@@ -27,6 +28,11 @@ from .lrc import LRC_LINE_RE, normalize_tags
|
||||
from .config import TTL_SYNCED, TTL_UNSYNCED, TTL_NOT_FOUND, TTL_NETWORK_ERROR
|
||||
from .models import TrackMeta, LyricResult, CacheStatus
|
||||
|
||||
METHODS = ("local", "cache-search", "spotify", "lrclib", "lrclib-search", "netease")
|
||||
FetcherMethodType = Literal[
|
||||
"local", "cache-search", "spotify", "lrclib", "lrclib-search", "netease"
|
||||
]
|
||||
|
||||
|
||||
def _normalize_unsynced(lyrics: str) -> str:
|
||||
"""Normalize unsynced lyrics so every line has a [00:00.00] tag.
|
||||
@@ -65,7 +71,7 @@ class LrcManager:
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.cache = CacheEngine()
|
||||
self.fetchers: dict[str, BaseFetcher] = {
|
||||
self.fetchers: dict[FetcherMethodType, BaseFetcher] = {
|
||||
"local": LocalFetcher(),
|
||||
"cache-search": CacheSearchFetcher(self.cache),
|
||||
"spotify": SpotifyFetcher(),
|
||||
@@ -73,9 +79,12 @@ class LrcManager:
|
||||
"lrclib-search": LrclibSearchFetcher(),
|
||||
"netease": NeteaseFetcher(),
|
||||
}
|
||||
assert set(self.fetchers) == set(METHODS), (
|
||||
f"METHODS and fetchers out of sync: {set(METHODS) ^ set(self.fetchers)}"
|
||||
)
|
||||
|
||||
def _build_sequence(
|
||||
self, track: TrackMeta, force_method: Optional[str] = None
|
||||
self, track: TrackMeta, force_method: Optional[FetcherMethodType] = None
|
||||
) -> list[BaseFetcher]:
|
||||
"""Determine the ordered list of fetchers to try."""
|
||||
if force_method:
|
||||
@@ -103,7 +112,7 @@ class LrcManager:
|
||||
def fetch_for_track(
|
||||
self,
|
||||
track: TrackMeta,
|
||||
force_method: Optional[str] = None,
|
||||
force_method: Optional[FetcherMethodType] = None,
|
||||
bypass_cache: bool = False,
|
||||
) -> Optional[LyricResult]:
|
||||
"""Fetch lyrics for *track* using the fallback pipeline.
|
||||
|
||||
Reference in New Issue
Block a user