feat: add ability to redownload directly from YM broken tracks

cli_core.py

@@ -6,7 +6,7 @@
 
 from dotenv import load_dotenv
 
-from core import soundcloud_client
+from core import navidrome_client, soundcloud_client
 from core.database import MigrationDB
 from core.lyrics import generate_lrc_for_track
 from core.models.appconfig import AppConfig
@@ -128,6 +128,92 @@ def process_single_track(
     db.mark_success(track.track_id, str(final_audio_dest))
 
 
+def process_single_track_replace(
+    track: TrackMetadata,
+    cfg: AppConfig,
+    db: MigrationDB,
+    existing_path: Path,
+) -> None:
+    """Download track from Yandex Music and replace the existing file at existing_path.
+    Skips SoundCloud tracks (track_id starting with 'sc_'). Deletes old file before writing new one.
+    """
+    if track.track_id.startswith("sc_"):
+        _logger.info(f"skip_soundcloud_track for {track.title} (not from Yandex)")
+        return
+    album_dir = existing_path.parent
+    ensure_directory(album_dir)
+    try:
+        try:
+            download_path, actual_extension = download_track_ytdlp(
+                track=track,
+                timeout_seconds=cfg.download_timeout_seconds,
+            )
+        except DownloadError as e:
+            _logger.warning("download_error from yt-dlp: " + str(e))
+            if "The current session has been rate-limited by YouTube" in str(e):
+                raise Exception(
+                    "The current session has been rate-limited by YouTube. Retry after an hour"
+                )
+            download_path, actual_extension = download_track_yandex(
+                track=track,
+                max_retries=cfg.max_download_retries,
+            )
+    except (DownloadError, RuntimeError) as exc:
+        _logger.error(f"redownload_failed for {track.title}: {exc}")
+        db.mark_failed(track.track_id, str(exc))
+        return
+    _logger.info(f"redownload_successful for {track.title}")
+    final_audio_dest = album_dir / build_track_filename(track, extension=actual_extension)
+    if existing_path.exists():
+        existing_path.unlink()
+        _logger.info(f"removed_old_file {existing_path}")
+    download_path.replace(final_audio_dest)
+    cover_bytes = download_cover_image(track)
+    cover_path = album_dir / "album-cover.jpg"
+    if cover_bytes and not cover_path.exists():
+        cover_path.write_bytes(cover_bytes)
+    try:
+        embed_tags(final_audio_dest, track, cover_bytes)
+    except Exception:
+        _logger.warning(f"tagging_failed for {track.title}")
+    try:
+        generate_lrc_for_track(final_audio_dest, track)
+    except Exception:
+        _logger.warning(f"lyrics_failed for {track.title}")
+    db.mark_success(track.track_id, str(final_audio_dest))
+
+
+def run_redownload_playlist(playlist_name: str, cfg: AppConfig, limit: int | None = None) -> None:
+    """Fetch tracks from Navidrome playlist (e.g. REDOWNLOAD), resolve to Yandex track_ids,
+    redownload from Yandex Music, and replace existing files.
+    Skips tracks not in migration DB or from SoundCloud.
+    """
+    data_dir = _get_data_dir()
+    music_root = cfg.music_root
+    pl = navidrome_client.get_playlist_by_name(playlist_name)
+    if pl is None:
+        _logger.error(f"playlist_not_found: {playlist_name!r}")
+        raise RuntimeError(f"Playlist {playlist_name!r} not found in Navidrome")
+    _logger.info(f"fetched_playlist {playlist_name!r}: {len(pl.entries)} tracks")
+    with MigrationDB(data_dir / "migration.db") as db:
+        processed = 0
+        for entry in pl.entries:
+            if limit is not None and processed >= limit:
+                break
+            full_path = (music_root / entry.path).resolve()
+            track_id = db.get_track_id_by_dest_path(full_path)
+            if track_id is None:
+                _logger.warning(f"track_not_in_migration_db: {full_path}")
+                continue
+            if track_id.startswith("sc_"):
+                _logger.info(f"skip_soundcloud {entry.title}")
+                continue
+            track = fetch_failed_track_metadata(track_id)
+            process_single_track_replace(track, cfg, db, full_path)
+            processed += 1
+    _logger.info(f"redownloaded {processed} tracks from {playlist_name!r}")
+
+
 def run_sync_like_tracks(cfg: AppConfig, limit: int | None = None) -> None:
     """Sync liked tracks from Yandex Music. If limit is set, process only that many tracks (for testing)."""
     data_dir = _get_data_dir()

core/database.py

@@ -3,7 +3,7 @@
 import datetime as _dt
 import sqlite3
 from pathlib import Path
-from typing import List, Tuple
+from typing import List, Optional, Tuple
 
 FailedTrack = Tuple[str, str]  # (track_id, error)
 
@@ -114,3 +114,26 @@ def get_total_count(self) -> int:
         cursor = self._connection.execute("SELECT COUNT(*) FROM migrations")
         row = cursor.fetchone()
         return row[0] if row else 0
+
+    def get_track_id_by_dest_path(self, path: Path) -> Optional[str]:
+        """Return track_id for a successful migration whose dest_path matches the given path.
+        path is the full filesystem path. Only considers successful migrations. Returns None if not found.
+        """
+        resolved = path.resolve()
+        cursor = self._connection.execute(
+            "SELECT track_id, dest_path FROM migrations WHERE status = 'success' AND dest_path IS NOT NULL"
+        )
+        for row in cursor.fetchall():
+            tid, dest = row[0], row[1]
+            if not dest:
+                continue
+            try:
+                dest_resolved = Path(dest).resolve()
+            except Exception:
+                continue
+            if dest_resolved == resolved:
+                return tid
+            # Normalize separators for cross-platform match
+            if str(dest_resolved).replace("\\", "/") == str(resolved).replace("\\", "/"):
+                return tid
+        return None

core/navidrome_client.py

@@ -0,0 +1,155 @@
+"""Navidrome/Subsonic API client for fetching playlists."""
+
+from __future__ import annotations
+
+import logging
+import os
+from dataclasses import dataclass
+from typing import List, Optional
+
+import requests
+
+_logger = logging.getLogger("navidrome_client")
+
+_NAVIDROME_URL = "NAVIDROME_URL"
+_NAVIDROME_USER = "NAVIDROME_USER"
+_NAVIDROME_PASSWORD = "NAVIDROME_PASSWORD"
+
+@dataclass
+class PlaylistTrack:
+    """A track entry from a Navidrome playlist (Subsonic getPlaylist)."""
+
+    id: str
+    title: str
+    artist: str
+    album: str
+    path: str  # Relative to music folder
+    duration: Optional[int] = None
+
+
+@dataclass
+class Playlist:
+    """A Navidrome playlist with its tracks."""
+
+    id: str
+    name: str
+    owner: str
+    song_count: int
+    duration: int
+    entries: List[PlaylistTrack]
+
+
+def _get_base_url() -> str:
+    url = os.getenv(_NAVIDROME_URL)
+    if not url:
+        raise RuntimeError(
+            f"Environment variable {_NAVIDROME_URL!r} is not set. "
+            "Set it to your Navidrome server URL (e.g. https://navidrome.example.com)."
+        )
+    return url.rstrip("/")
+
+
+def _get_auth_params() -> dict:
+    user = os.getenv(_NAVIDROME_USER)
+    password = os.getenv(_NAVIDROME_PASSWORD)
+    if not user or not password:
+        raise RuntimeError(
+            f"Environment variables {_NAVIDROME_USER!r} and {_NAVIDROME_PASSWORD!r}"
+            "are required for Navidrome API authentication."
+        )
+    return {"u": user, "p": password}
+
+
+def _parse_playlist_entry(entry: dict) -> Optional[PlaylistTrack]:
+    """Parse a single entry from getPlaylist response."""
+    path = entry.get("path")
+    if not path:
+        return None
+    return PlaylistTrack(
+        id=str(entry.get("id", "")),
+        title=entry.get("title") or "Unknown",
+        artist=entry.get("artist") or "Unknown Artist",
+        album=entry.get("album") or "",
+        path=path,
+        duration=entry.get("duration"),
+    )
+
+
+def fetch_playlists() -> List[dict]:
+    """Fetch all playlists. Returns raw playlist objects with id, name, etc."""
+    base_url = _get_base_url()
+    params = {"v": "1.16.1", "c": "navidrome_rw", "f": "json", **_get_auth_params()}
+    resp = requests.get(
+        f"{base_url}/rest/getPlaylists.view",
+        params=params,
+        timeout=30,
+    )
+    resp.raise_for_status()
+    data = resp.json()
+    sub = data.get("subsonic-response", {})
+    if sub.get("status") != "ok":
+        err = sub.get("error", {})
+        raise RuntimeError(
+            f"Navidrome API error {err.get('code', 'unknown')}: {err.get('message', 'Unknown error')}"
+        )
+    playlists = sub.get("playlists", {})
+    items = playlists.get("playlist")
+    if items is None:
+        items = playlists if isinstance(playlists, list) else []
+    if not isinstance(items, list):
+        items = [items] if items else []
+    return items
+
+
+def fetch_playlist(playlist_id: str) -> Playlist:
+    """Fetch a single playlist with all track entries."""
+    base_url = _get_base_url()
+    params = {
+        "v": "1.16.1",
+        "c": "navidrome_rw",
+        "f": "json",
+        "id": playlist_id,
+        **_get_auth_params(),
+    }
+    resp = requests.get(
+        f"{base_url}/rest/getPlaylist.view",
+        params=params,
+        timeout=60,
+    )
+    resp.raise_for_status()
+    data = resp.json()
+    sub = data.get("subsonic-response", {})
+    if sub.get("status") != "ok":
+        err = sub.get("error", {})
+        raise RuntimeError(
+            f"Navidrome API error {err.get('code', 'unknown')}: {err.get('message', 'Unknown error')}"
+        )
+    pl = sub.get("playlist", {})
+    entries_raw = pl.get("entry")
+    if entries_raw is None:
+        entries_raw = []
+    if not isinstance(entries_raw, list):
+        entries_raw = [entries_raw] if entries_raw else []
+    entries: List[PlaylistTrack] = []
+    for e in entries_raw:
+        if isinstance(e, dict):
+            t = _parse_playlist_entry(e)
+            if t:
+                entries.append(t)
+    return Playlist(
+        id=str(pl.get("id", "")),
+        name=pl.get("name", ""),
+        owner=pl.get("owner", ""),
+        song_count=int(pl.get("songCount", 0)),
+        duration=int(pl.get("duration", 0)),
+        entries=entries,
+    )
+
+
+def get_playlist_by_name(name: str) -> Optional[Playlist]:
+    """Find a playlist by exact name and return it with tracks. Returns None if not found."""
+    playlists = fetch_playlists()
+    for pl in playlists:
+        if isinstance(pl, dict) and pl.get("name") == name:
+            return fetch_playlist(str(pl.get("id", "")))
+    return None

docs/README.md

@@ -63,6 +63,11 @@ pip install -r requirements.txt
 - `NAVIDROME_FOLDER` – Целевая директория, откуда navidrome читает музыку/альбомы/артистов
 - `YANDEX_MUSIC_PERIOD_BETWEEN_REQUESTS` - Период между запросами в ЯМ, дабы избежать блокировки rate limit
 
+Для команды `redownload-playlist` (перекачка треков из плейлиста Navidrome):
+- `NAVIDROME_URL` – URL сервера Navidrome (напр. https://navidrome.example.com)
+- `NAVIDROME_USER` – Логин пользователя Navidrome
+- `NAVIDROME_PASSWORD` – Пароль пользователя Navidrome. Если пароль содержит `#`, `&`, запятые и др. спецсимволы — обязательно возьмите значение в кавычки: `NAVIDROME_PASSWORD="2uY#hG,#^Tw&;_4"`
+
 ## Использование
 
 1. Запустить выгрузку с ЯМ в navidrome
@@ -95,7 +100,17 @@ python -m main count-successful
 python -m main soundcloud-import <your_nickname>
 ```
 
-6. Запуск webui:
+6. Перекачать треки из плейлиста Navidrome (напр. `_REDOWNLOAD`) заново с ЯМ:
+
+```bash
+python -m main redownload-playlist
+# или указать другой плейлист:
+python -m main redownload-playlist --playlist _REDOWNLOAD
+# тест на одном треке:
+python -m main redownload-playlist --limit 1
+```
+
+7. Запуск webui:
 
 ```bash
 python -m main web

main.py

@@ -14,6 +14,7 @@
     run_count_successful,
     run_import_soundcloud_likes,
     run_list_failed,
+    run_redownload_playlist,
     run_retry_failed,
     run_sync_like_tracks,
 )
@@ -72,6 +73,31 @@ def soundcloud_import_test_command(
     _logger.info("finished soundcloud-import-test")
 
 
+@app.command("redownload-playlist")
+def redownload_playlist_command(
+    playlist_name: str = typer.Option(
+        "_REDOWNLOAD",
+        "--playlist",
+        "-p",
+        help="Name of the Navidrome playlist to redownload (default: _REDOWNLOAD).",
+    ),
+    limit: int | None = typer.Option(
+        None,
+        "--limit",
+        "-n",
+        help="Process at most N tracks (for testing).",
+    ),
+) -> None:
+    """Fetch tracks from a Navidrome playlist, redownload from Yandex Music, and replace existing files.
+    Requires NAVIDROME_URL, NAVIDROME_USER, NAVIDROME_PASSWORD in .env.
+    Skips tracks not in migration DB or from SoundCloud."""
+    data_dir = _get_data_dir()
+    configure_logging(data_dir / "migration.log")
+    cfg = _build_config()
+    run_redownload_playlist(playlist_name, cfg, limit=limit)
+    _logger.info("finished redownload-playlist")
+
+
 @app.command("retry-failed")
 def retry_failed_command() -> None:
     """Retry previously failed downloads recorded in migration.db."""