285 lines
10 KiB
Python
285 lines
10 KiB
Python
"""
|
|
OneTagger CLI integration helpers.
|
|
|
|
This module wraps the `onetagger-cli autotagger` command so the rest of the
|
|
application can tag a single audio file in isolation. Key expectations:
|
|
- The user has installed onetagger-cli (>= 1.7.0) and created an Auto Tag
|
|
profile via the OneTagger GUI, exporting its JSON config. Authentication
|
|
(Discogs token, Spotify client ID/secret, etc.) is managed in OneTagger.
|
|
- The profile should enable platforms Beatport/Spotify/Discogs/iTunes/Bandcamp,
|
|
enable tag overwrites and album art (1000px for Beatport/iTunes), include
|
|
Artist/Title/Album/Label/Genre/Style/Release/Publish dates, enable Shazam
|
|
matching and filename parsing, ID3v2.4 + short title, and save album art.
|
|
- OneTagger writes run reports as success-*.m3u and failed-*.m3u under its
|
|
data_dir/runs; this module inspects those to decide whether tagging succeeded.
|
|
|
|
Usage:
|
|
from onetagger_integration import auto_tag_postprocess, OneTaggerResult
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
class OneTaggerError(Exception):
|
|
"""Base exception for OneTagger integration errors."""
|
|
|
|
|
|
class OneTaggerCliNotFoundError(OneTaggerError):
|
|
"""Raised when onetagger-cli cannot be located or executed."""
|
|
|
|
|
|
@dataclass
|
|
class OneTaggerResult:
|
|
success: bool
|
|
tagged_file: Path
|
|
log_file: Optional[Path]
|
|
success_playlist: Optional[Path]
|
|
failed_playlist: Optional[Path]
|
|
stdout: str
|
|
stderr: str
|
|
|
|
|
|
def get_default_onetagger_data_dir() -> Path:
|
|
"""Return the default OneTagger data directory based on platform."""
|
|
if sys.platform.startswith("linux"):
|
|
return Path.home() / ".config" / "onetagger"
|
|
if sys.platform == "darwin":
|
|
return Path.home() / "Library" / "Preferences" / "com.OneTagger.OneTagger"
|
|
if sys.platform.startswith("win"):
|
|
appdata = os.environ.get("APPDATA") or str(Path.home() / "AppData" / "Roaming")
|
|
return Path(appdata) / "OneTagger" / "OneTagger"
|
|
return Path.home() / ".onetagger"
|
|
|
|
|
|
def find_latest_run_playlists(data_dir: Path) -> tuple[Optional[Path], Optional[Path]]:
|
|
"""Return the most recent (by mtime, then name) success/failed playlists."""
|
|
runs_dir = data_dir / "runs"
|
|
if not runs_dir.exists():
|
|
return None, None
|
|
|
|
def _latest(prefix: str) -> Optional[Path]:
|
|
candidates = list(runs_dir.glob(f"{prefix}-*.m3u"))
|
|
if not candidates:
|
|
return None
|
|
candidates.sort(key=lambda p: (p.stat().st_mtime, p.name))
|
|
return candidates[-1]
|
|
|
|
return _latest("success"), _latest("failed")
|
|
|
|
|
|
def playlist_contains_path(playlist: Path, target: Path) -> bool:
|
|
"""Check if target path appears in the .m3u playlist."""
|
|
if not playlist or not playlist.exists():
|
|
return False
|
|
try:
|
|
target_norm = str(target.resolve())
|
|
except Exception:
|
|
target_norm = str(target)
|
|
target_norm = os.path.normcase(os.path.normpath(target_norm))
|
|
try:
|
|
with playlist.open("r", encoding="utf-8", errors="ignore") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
try:
|
|
cand = os.path.normcase(os.path.normpath(str(Path(line).expanduser().resolve())))
|
|
except Exception:
|
|
cand = os.path.normcase(os.path.normpath(line))
|
|
if cand == target_norm:
|
|
return True
|
|
except FileNotFoundError:
|
|
return False
|
|
return False
|
|
|
|
|
|
def _resolve_cli_path(onetagger_cli_path: Path | str) -> Path:
|
|
"""Resolve CLI path or raise OneTaggerCliNotFoundError.
|
|
|
|
Tries (in order):
|
|
- Exact path if provided (absolute or relative to cwd)
|
|
- Binary alongside this module (useful when bundled in the project root)
|
|
- cwd / <name> when only a bare name was provided
|
|
- PATH lookup
|
|
"""
|
|
cli_path = Path(onetagger_cli_path) if not isinstance(onetagger_cli_path, Path) else onetagger_cli_path
|
|
candidates = []
|
|
candidates.append(cli_path)
|
|
if not cli_path.is_absolute():
|
|
module_dir = Path(__file__).resolve().parent
|
|
candidates.append(module_dir / cli_path.name)
|
|
candidates.append(Path.cwd() / cli_path)
|
|
for cand in candidates:
|
|
if cand.exists():
|
|
if os.access(cand, os.X_OK):
|
|
return cand.resolve()
|
|
raise OneTaggerCliNotFoundError(f"onetagger-cli not executable: {cand}")
|
|
found = shutil.which(str(onetagger_cli_path))
|
|
if found:
|
|
return Path(found).resolve()
|
|
raise OneTaggerCliNotFoundError(f"onetagger-cli not found at {onetagger_cli_path}")
|
|
|
|
|
|
def _list_run_playlists(data_dir: Path) -> set[Path]:
|
|
runs_dir = data_dir / "runs"
|
|
if not runs_dir.exists():
|
|
return set()
|
|
return set(runs_dir.glob("*.m3u"))
|
|
|
|
|
|
def _sanitize_config(config_path: Path, work_dir: Path) -> Path:
|
|
"""Create a sanitized copy of the config that drops platforms lacking creds."""
|
|
try:
|
|
cfg = json.loads(config_path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return config_path
|
|
|
|
platforms = cfg.get("platforms")
|
|
if not isinstance(platforms, list):
|
|
return config_path
|
|
|
|
modified = False
|
|
spotify_cfg = cfg.get("spotify") or {}
|
|
discogs_cfg = ((cfg.get("custom") or {}).get("discogs")) if isinstance(cfg.get("custom"), dict) else {}
|
|
|
|
def _missing(val):
|
|
return val is None or (isinstance(val, str) and not val.strip())
|
|
|
|
if "spotify" in platforms:
|
|
if _missing(spotify_cfg.get("clientId")) or _missing(spotify_cfg.get("clientSecret")):
|
|
platforms = [p for p in platforms if p != "spotify"]
|
|
cfg["platforms"] = platforms
|
|
cfg["spotify"] = None
|
|
modified = True
|
|
|
|
if "discogs" in platforms:
|
|
token = discogs_cfg.get("token") if isinstance(discogs_cfg, dict) else None
|
|
if _missing(token):
|
|
platforms = [p for p in platforms if p != "discogs"]
|
|
cfg["platforms"] = platforms
|
|
if isinstance(cfg.get("custom"), dict) and "discogs" in cfg["custom"]:
|
|
cfg["custom"]["discogs"]["token"] = ""
|
|
modified = True
|
|
|
|
if not modified:
|
|
return config_path
|
|
|
|
sanitized_path = work_dir / "sanitized-config.json"
|
|
sanitized_path.write_text(json.dumps(cfg, indent=2), encoding="utf-8")
|
|
return sanitized_path
|
|
|
|
|
|
def tag_audio_with_onetagger(
|
|
audio_file: Path,
|
|
onetagger_cli_path: Path | str,
|
|
config_path: Path,
|
|
data_dir: Optional[Path] = None,
|
|
timeout: int = 1800,
|
|
) -> OneTaggerResult:
|
|
"""Tag a single audio file using OneTagger CLI Auto Tag.
|
|
|
|
The file is copied into a temporary directory and only that copy is tagged.
|
|
If tagging succeeds (per success-*.m3u), the original file is atomically
|
|
replaced by the tagged copy.
|
|
"""
|
|
src = Path(audio_file)
|
|
if not src.exists() or not src.is_file():
|
|
raise FileNotFoundError(f"Audio file not found: {src}")
|
|
cfg = Path(config_path)
|
|
if not cfg.exists() or not cfg.is_file():
|
|
raise FileNotFoundError(
|
|
f"OneTagger config not found: {cfg}. Export a valid Auto Tag profile from OneTagger GUI."
|
|
)
|
|
cli = _resolve_cli_path(onetagger_cli_path)
|
|
data_dir = Path(data_dir) if data_dir else get_default_onetagger_data_dir()
|
|
runs_before = _list_run_playlists(data_dir)
|
|
|
|
log_file = data_dir / "onetagger.log"
|
|
log_file = log_file if log_file.exists() else None
|
|
|
|
with tempfile.TemporaryDirectory(prefix="onetagger_work_") as tmpdir:
|
|
tmpdir_path = Path(tmpdir)
|
|
tmp_file = tmpdir_path / src.name
|
|
shutil.copy2(src, tmp_file)
|
|
|
|
config_to_use = _sanitize_config(cfg, tmpdir_path)
|
|
|
|
cmd = [str(cli), "autotagger", "--config", str(config_to_use), "--path", str(tmpdir_path)]
|
|
LOG.debug("Running OneTagger: %s", " ".join(cmd))
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
timeout=timeout,
|
|
)
|
|
LOG.debug("OneTagger exited %s", result.returncode)
|
|
|
|
runs_after = _list_run_playlists(data_dir)
|
|
new_runs = runs_after - runs_before
|
|
|
|
success_playlist, failed_playlist = find_latest_run_playlists(data_dir)
|
|
# Prefer freshly created playlists if present
|
|
def _select(prefix: str, fallback: Optional[Path]) -> Optional[Path]:
|
|
candidates = [p for p in new_runs if p.name.startswith(f"{prefix}-")]
|
|
if candidates:
|
|
candidates.sort(key=lambda p: (p.stat().st_mtime, p.name))
|
|
return candidates[-1]
|
|
return fallback
|
|
|
|
success_playlist = _select("success", success_playlist)
|
|
failed_playlist = _select("failed", failed_playlist)
|
|
|
|
success_in_playlist = playlist_contains_path(success_playlist, tmp_file) if success_playlist else False
|
|
failed_in_playlist = playlist_contains_path(failed_playlist, tmp_file) if failed_playlist else False
|
|
|
|
cli_ok = result.returncode == 0
|
|
tagging_success = cli_ok and success_in_playlist and not failed_in_playlist
|
|
|
|
if tagging_success:
|
|
os.replace(tmp_file, src)
|
|
else:
|
|
if not cli_ok and not result.stderr:
|
|
result.stderr = f"onetagger-cli exited with code {result.returncode}"
|
|
if cli_ok and not (success_in_playlist or failed_in_playlist):
|
|
result.stderr = (
|
|
result.stderr
|
|
+ "\n"
|
|
if result.stderr
|
|
else ""
|
|
) + "File not found in success or failed playlists; tagging uncertain."
|
|
|
|
return OneTaggerResult(
|
|
success=tagging_success,
|
|
tagged_file=src,
|
|
log_file=log_file,
|
|
success_playlist=success_playlist,
|
|
failed_playlist=failed_playlist,
|
|
stdout=result.stdout,
|
|
stderr=result.stderr,
|
|
)
|
|
|
|
|
|
def auto_tag_postprocess(audio_file: Path, settings) -> OneTaggerResult:
|
|
"""Convenience wrapper that reads OneTagger settings from an app settings object."""
|
|
return tag_audio_with_onetagger(
|
|
audio_file=audio_file,
|
|
onetagger_cli_path=settings.onetagger_cli_path,
|
|
config_path=settings.onetagger_config_path,
|
|
data_dir=getattr(settings, "onetagger_data_dir", None),
|
|
timeout=getattr(settings, "onetagger_timeout", 1800),
|
|
)
|