diff options
| -rw-r--r-- | README.md | 28 | ||||
| -rw-r--r-- | gui.py | 29 | ||||
| -rw-r--r-- | optical.py | 290 |
3 files changed, 311 insertions, 36 deletions
@@ -37,6 +37,23 @@ All other dependencies can be installed via `uv` uv sync ``` +# CD Ripping +A bonus useful feature for ripping CDs right into expergo. Only supported on Windows and Linux. + +**Linux:** Requires [abcde](https://github.com/poddmo/abcde) to be installed. Please follow distro specific instructions + +**Windows:** All of the optical drive related functionalities are relegated to [CueRipper](https://github.com/gchudov/cuetools.net), expergo can automatically install this for you + +# GUI +```bash +uv run expergo.py gui +``` +<img width="1913" height="1038" alt="image" src="https://github.com/user-attachments/assets/192fabd5-a3bb-48f7-9b92-f90e01987352" /> +<img width="1000" height="600" alt="image" src="https://github.com/user-attachments/assets/967e3a67-46a2-4abd-8c07-df56d6cea9ca" /> + +> [!TIP] +> If you're using a Snowsky device, you can "emulate" playlist functionality by grouping songs together via the "Genre" tag + # CLI **Apply Echo Rules:** @@ -57,16 +74,9 @@ What each command does: - `auto-lrc`: only fetch LRC files for FLACs (skips existing by default, `--force` overwrites) - `auto-cover`: only fetch album art using MusicBrainz + Cover Art Archive, grouped per album (uses album artist when present) - `reorg`: reorganize library into `Artist/Album/` folders, moving matching `.lrc` sidecars too -- -# GUI -```bash -uv run expergo.py gui -``` -<img width="1913" height="1038" alt="image" src="https://github.com/user-attachments/assets/192fabd5-a3bb-48f7-9b92-f90e01987352" /> -<img width="1000" height="600" alt="image" src="https://github.com/user-attachments/assets/967e3a67-46a2-4abd-8c07-df56d6cea9ca" /> -> [!TIP] -> If you're using a Snowsky device, you can "emulate" playlist functionality by grouping songs together via the "Genre" tag +**CD Ripping is not supported from the CLI at this time** + # Other Tools `apply_album_art.sh` - Recursively traverse all folders, automatically apply album art and resize to all FLACs that have a JPG/PNG in the same directory @@ -1241,7 +1241,7 @@ class MainWindow(QMainWindow): reorg_act = action("Reorganize…", self.reorganize, ("folder-new", SP.SP_FileDialogDetailedView), "Ctrl+Shift+O", "Move every track in the library into Artist / Album folders (with preview)") rip_cd_act = action("Rip CD", self.rip_cd, ("media-optical", SP.SP_DirOpenIcon), - tip="Rip tracks from an audio CD (stub)") + tip="Rip tracks from an audio CD") self.preview_action = action("Preview", self.preview_track, ("media-playback-start", SP.SP_MediaPlay), tip="Open the selected track in the system default player") rules_act = action("Apply Echo Rules", self.process, ("system-run", SP.SP_MediaPlay), "Ctrl+R", @@ -1619,8 +1619,29 @@ class MainWindow(QMainWindow): ok, msg = optical.check_dependencies() if not ok: - QMessageBox.critical(self, "Rip CD", msg) - return + if optical.needs_cuetools_install(): + answer = QMessageBox.question( + self, + "Rip CD", + ( + "CUERipper was not found.\n\n" + "Download and install CUETools now?\n\n" + f"Source: {optical.CUETOOLS_DOWNLOAD_URL}" + ), + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, + QMessageBox.StandardButton.Yes, + ) + if answer == QMessageBox.StandardButton.Yes: + try: + optical.ensure_windows_cuetools(on_output=self.log) + ok, msg = optical.check_dependencies() + except Exception as err: + QMessageBox.critical(self, "Rip CD", f"Failed to install CUETools.\n\n{err}") + return + + if not ok: + QMessageBox.critical(self, "Rip CD", msg) + return drives = optical.get_optical_drives() if not drives: @@ -1637,7 +1658,7 @@ class MainWindow(QMainWindow): self.rip_drive = selected_drive self.log(f"Rip CD selected drive: {self.rip_drive}") - self.log("Starting abcde rip (default config)…") + self.log("Starting CD ripping process…") self._set_busy(True) self.stop_action.setEnabled(False) @@ -1,14 +1,178 @@ +import json import os import platform import shutil import subprocess +import tempfile +import time +import urllib.request +import zipfile from collections.abc import Callable from pathlib import Path +CUETOOLS_VERSION = "2.2.6" +CUETOOLS_DOWNLOAD_URL = ( + "https://github.com/gchudov/cuetools.net/releases/download/" + "v2.2.6/CUETools_2.2.6.zip" +) + + +def _get_powershell_executable() -> str | None: + return shutil.which("powershell") or shutil.which("pwsh") + + +def _get_windows_tools_root() -> Path: + local_app_data = os.environ.get("LOCALAPPDATA") + if local_app_data: + return Path(local_app_data) / "fiio-snowsky-echo-flac-tool" / "tools" + return Path.home() / ".fiio-snowsky-echo-flac-tool" / "tools" + + +def _get_windows_cuetools_dir() -> Path: + return _get_windows_tools_root() / f"cuetools-{CUETOOLS_VERSION}" + + +def _find_cueripper_in_dir(base_dir: Path) -> str | None: + if not base_dir.exists(): + return None + + for name in ("CUERipper.Console.exe", "CUERipper.exe"): + matches = sorted(base_dir.rglob(name)) + if matches: + return str(matches[0]) + + return None + + +def _get_cueripper_executable() -> str | None: + explicit = os.environ.get("CUERIPPER_EXE") + if explicit: + return explicit + + for name in ["CUERipper.Console.exe", "CUERipper.exe"]: + found = shutil.which(name) + if found: + return found + + return _find_cueripper_in_dir(_get_windows_cuetools_dir()) + + +def needs_cuetools_install() -> bool: + return platform.system() == "Windows" and _get_cueripper_executable() is None + + +def ensure_windows_cuetools( + on_output: Callable[[str], None] | None = None, +) -> str: + if platform.system() != "Windows": + raise RuntimeError("CUETools bootstrap is only supported on Windows") + + existing = _get_cueripper_executable() + if existing: + return existing + + install_dir = _get_windows_cuetools_dir() + install_dir.mkdir(parents=True, exist_ok=True) + + zip_path = install_dir / "CUETools_2.2.6.zip" + + if on_output: + on_output(f"Downloading CUETools from {CUETOOLS_DOWNLOAD_URL}") + + with urllib.request.urlopen(CUETOOLS_DOWNLOAD_URL) as resp, zip_path.open("wb") as dst: + shutil.copyfileobj(resp, dst) + + if on_output: + on_output(f"Extracting {zip_path} to {install_dir}") + + with zipfile.ZipFile(zip_path, "r") as zf: + zf.extractall(install_dir) + + cueripper = _find_cueripper_in_dir(install_dir) + if not cueripper: + raise RuntimeError( + "Downloaded CUETools archive but could not find CUERipper executable" + ) + + if on_output: + on_output(f"CUERipper installed: {cueripper}") + + return cueripper + + +def _iter_existing_dirs(paths: list[Path]) -> list[Path]: + seen: set[str] = set() + out: list[Path] = [] + for p in paths: + key = str(p).lower() + if key in seen: + continue + seen.add(key) + if p.exists() and p.is_dir(): + out.append(p) + return out + + +def _find_recent_flacs(search_roots: list[Path], since_epoch: float) -> list[str]: + found: list[str] = [] + for root in _iter_existing_dirs(search_roots): + for path in root.rglob("*.flac"): + try: + if path.stat().st_mtime >= since_epoch: + found.append(str(path.resolve())) + except OSError: + continue + return sorted(set(found)) + + +def _run_streaming_command( + cmd: list[str], + cwd: Path, + on_output: Callable[[str], None] | None = None, +) -> list[str]: + proc = subprocess.Popen( + cmd, + cwd=cwd, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + env=os.environ.copy(), + ) + + lines: list[str] = [] + if proc.stdout: + for raw in proc.stdout: + line = raw.rstrip("\n") + lines.append(line) + if on_output and line: + on_output(line) + + rc = proc.wait() + if rc != 0: + raise subprocess.CalledProcessError(rc, cmd, output="\n".join(lines)) + + return lines + + def check_dependencies() -> tuple[bool, str]: if platform.system() == "Windows": - return False, "Windows is currently not supported" # stub for now + missing: list[str] = [] + if _get_powershell_executable() is None: + missing.append("powershell") + if _get_cueripper_executable() is None: + missing.append("CUERipper.Console.exe") + + if missing: + return ( + False, + "Ripping CDs on Windows requires: " + + ", ".join(missing) + + ". CUERipper can be auto-downloaded.", + ) + return True, "OK" + if platform.system() == "Linux": try: subprocess.run(["abcde", "-h"], check=True, capture_output=True) @@ -18,6 +182,7 @@ def check_dependencies() -> tuple[bool, str]: False, "Ripping CDs requires the 'abcde' package to be installed, please refer to the README for more information", ) + return True, "You are on an unsupported platform, CD features may not be fully functional" @@ -25,7 +190,47 @@ def get_optical_drives() -> list[tuple[str, str]]: import re if platform.system() == "Windows": - return [] # stub for now + powershell = _get_powershell_executable() + if not powershell: + return [] + + try: + res = subprocess.run( + [ + powershell, + "-NoProfile", + "-Command", + "Get-CimInstance Win32_CDROMDrive | " + "Select-Object Drive, Caption, Name, MediaLoaded | " + "ConvertTo-Json -Compress", + ], + capture_output=True, + text=True, + check=True, + ) + + payload = res.stdout.strip() + if not payload: + return [] + + parsed = json.loads(payload) + entries = parsed if isinstance(parsed, list) else [parsed] + + drives: list[tuple[str, str]] = [] + for entry in entries: + drive = entry.get("Drive") + if not drive: + continue + caption = entry.get("Caption") or entry.get("Name") or drive + media_loaded = entry.get("MediaLoaded") + if media_loaded is False: + caption = f"{caption} (no media)" + drives.append((drive, caption)) + + return drives + except (subprocess.CalledProcessError, json.JSONDecodeError): + return [] + if platform.system() == "Linux": unix_path_regex = re.compile(r"(/dev/[^\s]+)") try: @@ -43,6 +248,7 @@ def get_optical_drives() -> list[tuple[str, str]]: return drives except subprocess.CalledProcessError: return [] + return [] @@ -50,32 +256,70 @@ def run_abcde_rip( drive: str, on_output: Callable[[str], None] | None = None, ) -> list[str]: - workspace = Path("/tmp/expergo-rip-workspace") + workspace = Path(tempfile.gettempdir()) / "expergo-rip-workspace" if workspace.exists(): shutil.rmtree(workspace) workspace.mkdir(parents=True, exist_ok=True) - cmd = ["abcde", "-d", drive, "-L", "-n", "-N", "-o", "flac"] - proc = subprocess.Popen( - cmd, - cwd=workspace, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - bufsize=1, - env=os.environ.copy(), - ) + if platform.system() == "Windows": + cueripper = _get_cueripper_executable() + if not cueripper: + raise RuntimeError( + "CUERipper executable not found. Install CUETools and ensure " + "CUERipper.Console.exe is in PATH, set CUERIPPER_EXE, " + "or use auto-download from the Rip CD prompt." + ) - lines: list[str] = [] - if proc.stdout: - for raw in proc.stdout: - line = raw.rstrip("\n") - lines.append(line) - if on_output and line: - on_output(line) + is_console = Path(cueripper).name.lower() == "cueripper.console.exe" + rip_start = time.time() - rc = proc.wait() - if rc != 0: - raise subprocess.CalledProcessError(rc, cmd, output="\n".join(lines)) + if is_console: + cmd = [ + cueripper, + "--drive", + drive, + "--output", + str(workspace), + "--format", + "flac", + ] + else: + if on_output: + on_output("CUERipper GUI opened. Start rip there and close it when done.") + cmd = [cueripper] + + _run_streaming_command(cmd, workspace, on_output) + + flacs_in_workspace = sorted(str(p.resolve()) for p in workspace.rglob("*.flac")) + if flacs_in_workspace: + return flacs_in_workspace + + configured_output = os.environ.get("CUERIPPER_OUTPUT_DIR") + search_roots = [ + workspace, + Path(cueripper).parent, + Path.home() / "Music", + Path.home() / "Documents", + Path.home() / "Desktop", + Path.home() / "Downloads", + Path(tempfile.gettempdir()), + ] + if configured_output: + search_roots.insert(0, Path(configured_output)) + + recent_flacs = _find_recent_flacs(search_roots, rip_start - 2) + if recent_flacs: + return recent_flacs + + searched = "\n".join(f"- {p}" for p in _iter_existing_dirs(search_roots)) + raise RuntimeError( + "CUERipper completed but no FLAC files were detected.\n\n" + "Searched these locations for newly created .flac files:\n" + f"{searched}\n\n" + "If CUERipper saves elsewhere, set CUERIPPER_OUTPUT_DIR to that folder." + ) + + cmd = ["abcde", "-d", drive, "-L", "-n", "-N", "-o", "flac"] + _run_streaming_command(cmd, workspace, on_output) return sorted(str(p.resolve()) for p in workspace.rglob("*.flac")) |