Мои лайки
Отдельный список лайкнутых треков и артистов из этих лайков.
Лайки еще не загружены.
Активные задачи
Смотрим какой трек качается и прогресс.
"""
class DownloadRequest(BaseModel):
url: Optional[str] = None
artist_id: Optional[str] = None
album_id: Optional[str] = None
track_id: Optional[str] = None
playlist_id: Optional[str] = None
dir: str = Field(default=DEFAULT_DIR, description="Папка для загрузки музыки")
path_pattern: str = Field(
default=DEFAULT_PATH_PATTERN,
description="Паттерн сохранения файла",
)
quality: int = Field(default=DEFAULT_QUALITY, ge=0, le=2)
skip_existing: bool = DEFAULT_SKIP_EXISTING
lyrics_format: Literal["none", "text", "lrc"] = DEFAULT_LYRICS_FORMAT
embed_cover: bool = DEFAULT_EMBED_COVER
cover_resolution: int = Field(default=DEFAULT_COVER_RESOLUTION)
delay: int = Field(default=DEFAULT_DELAY, ge=0)
stick_to_artist: bool = DEFAULT_STICK_TO_ARTIST
only_music: bool = DEFAULT_ONLY_MUSIC
compatibility_level: int = Field(
default=DEFAULT_COMPATIBILITY_LEVEL,
ge=core.MIN_COMPATIBILITY_LEVEL,
le=core.MAX_COMPATIBILITY_LEVEL,
)
timeout: int = Field(default=DEFAULT_TIMEOUT, ge=1)
tries: int = Field(default=DEFAULT_TRIES, ge=0)
retry_delay: int = Field(default=DEFAULT_RETRY_DELAY, ge=0)
unsafe_path: bool = DEFAULT_UNSAFE_PATH
class DiscographyRequest(BaseModel):
url: Optional[str] = None
artist_id: Optional[str] = None
stick_to_artist: bool = DEFAULT_STICK_TO_ARTIST
only_music: bool = DEFAULT_ONLY_MUSIC
timeout: int = Field(default=DEFAULT_TIMEOUT, ge=1)
tries: int = Field(default=DEFAULT_DISCO_TRIES, ge=0)
retry_delay: int = Field(default=DEFAULT_DISCO_RETRY_DELAY, ge=0)
JobStatus = Literal["pending", "running", "done", "failed"]
JobType = Literal[
"download",
"discography",
"likes_tracks",
"likes_artists_albums",
]
@dataclass
class Job:
id: str
type: JobType
status: JobStatus = "pending"
message: str = ""
total: Optional[int] = None
progress: int = 0
created_at: float = field(default_factory=time.time)
finished_at: Optional[float] = None
logs: list[str] = field(default_factory=list)
current_track: Optional[str] = None
def log(self, text: str) -> None:
timestamp = time.strftime("%H:%M:%S", time.localtime())
line = f"[{timestamp}] {text}"
logger.debug(line)
self.logs.append(line)
job_lock = threading.Lock()
jobs: dict[str, Job] = {}
def register_job(job: Job) -> Job:
with job_lock:
jobs[job.id] = job
return job
def snapshot_job(job_id: str) -> Job:
with job_lock:
job = jobs.get(job_id)
if job is None:
raise KeyError(job_id)
return job
def parse_ids_from_url(url: str) -> dict[str, str]:
parsed = {}
if match := ARTIST_RE.search(url):
parsed["artist_id"] = match.group(1)
if match := ALBUM_RE.search(url):
parsed["album_id"] = match.group(1)
if match := TRACK_RE.search(url):
parsed["track_id"] = match.group(1)
if match := PLAYLIST_RE.search(url):
parsed["playlist_id"] = match.group(1) + "/" + match.group(2)
return parsed
def validate_and_fill_request(req: DownloadRequest) -> DownloadRequest:
require_default_token()
if req.url:
parsed = parse_ids_from_url(req.url)
for key, value in parsed.items():
if getattr(req, key) is None:
setattr(req, key, value)
return req
def album_matches_request(
album: Album, req: DownloadRequest, artist_id: Optional[str] = None
) -> bool:
if album.id is None or not album.available:
return False
if req.only_music and album.meta_type != "music":
return False
if (
artist_id is not None
and req.stick_to_artist
and album.artists
and album.artists[0].id != int(artist_id)
):
return False
return True
def build_track_iter(
client, req: DownloadRequest
) -> Tuple[Iterable[Track], Optional[int]]:
def album_tracks_gen(album_ids: Iterable[str]) -> Iterable[Track]:
for album_id in album_ids:
if full_album := client.albums_with_tracks(album_id):
if volumes := full_album.volumes:
for volume in volumes:
for track in volume:
yield track
total_track_count = None
if req.artist_id is not None:
def filter_album(album: Album) -> bool:
return album_matches_request(album, req, req.artist_id)
def albums_id_gen() -> Iterable[str]:
has_next = True
page = 0
while has_next:
albums_info = client.artists_direct_albums(req.artist_id, page)
if not albums_info:
break
for album in albums_info.albums:
if filter_album(album) and album.id:
yield str(album.id)
else:
nonlocal total_track_count
if album.track_count and total_track_count is not None:
total_track_count -= album.track_count
if pager := albums_info.pager:
page = pager.page + 1
has_next = pager.per_page * page < pager.total
else:
break
result_tracks = album_tracks_gen(albums_id_gen())
artist = client.artists(req.artist_id)[0]
if counts := artist.counts:
total_track_count = counts.tracks
elif req.album_id is not None:
result_tracks = album_tracks_gen((req.album_id,))
if album := client.albums_with_tracks(req.album_id):
total_track_count = album.track_count
elif req.track_id is not None:
result_tracks = client.tracks(req.track_id)
total_track_count = 1
elif req.playlist_id is not None:
user, kind = req.playlist_id.split("/")
playlist = client.users_playlists(kind, user)
playlist = playlist if isinstance(playlist, Playlist) else None
if playlist is None:
raise HTTPException(404, detail="Плейлист не найден")
total_track_count = playlist.track_count
def playlist_tracks_gen() -> Iterable[Track]:
tracks = playlist.fetch_tracks()
for i in range(0, len(tracks), FETCH_PAGE_SIZE):
chunk = tracks[i : i + FETCH_PAGE_SIZE]
yield from client.tracks([track.id for track in chunk])
result_tracks = playlist_tracks_gen()
else:
raise HTTPException(400, detail="Нужно указать URL или ID")
return result_tracks, total_track_count
def to_core_quality(quality: int) -> core.CoreTrackQuality:
return core.CoreTrackQuality(quality)
def build_track_iter_from_ids(client, track_ids: list[str]) -> Iterable[Track]:
def generator() -> Iterable[Track]:
for i in range(0, len(track_ids), FETCH_PAGE_SIZE):
chunk = track_ids[i : i + FETCH_PAGE_SIZE]
for track in client.tracks(chunk):
if track is not None:
yield track
return generator()
def collect_artist_album_ids(
client, artist_ids: Iterable[str], req: DownloadRequest
) -> list[str]:
result: list[str] = []
seen: set[str] = set()
for artist_id in artist_ids:
has_next = True
page = 0
while has_next:
albums_info = client.artists_direct_albums(artist_id, page)
if not albums_info:
break
for album in albums_info.albums:
if not album_matches_request(album, req, artist_id):
continue
album_id = str(album.id)
if album_id in seen:
continue
seen.add(album_id)
result.append(album_id)
if pager := albums_info.pager:
page = pager.page + 1
has_next = pager.per_page * page < pager.total
else:
break
return result
def collect_album_track_ids(client, album_ids: Iterable[str]) -> list[str]:
result: list[str] = []
seen: set[str] = set()
for album_id in album_ids:
if full_album := client.albums_with_tracks(album_id):
if volumes := full_album.volumes:
for volume in volumes:
for track in volume:
if track.id is None:
continue
track_id = str(track.id)
if track_id in seen:
continue
seen.add(track_id)
result.append(track_id)
return result
def collect_liked_track_ids(client) -> list[str]:
likes = client.users_likes_tracks()
if not likes:
return []
return unique_ids(likes.tracks_ids)
def collect_liked_artist_ids(client) -> list[str]:
artist_ids: list[str] = []
seen: set[str] = set()
for track in fetch_liked_tracks(client):
for artist in track.artists or []:
if artist.id is None:
continue
artist_id = str(artist.id)
if artist_id in seen:
continue
seen.add(artist_id)
artist_ids.append(artist_id)
return artist_ids
def download_tracks_for_job(
job: Job,
req: DownloadRequest,
tracks: Iterable[Track],
total: Optional[int],
) -> None:
job.total = total
covers_cache: dict[int, core.AlbumCover] = {}
base_dir = Path(req.dir).expanduser()
base_dir.mkdir(parents=True, exist_ok=True)
path_pattern = Path(req.path_pattern)
lyrics_format = core.LyricsFormat(req.lyrics_format)
quality = to_core_quality(req.quality)
downloaded = 0
seq = 0
for track in tracks:
seq += 1
if job.status == "failed":
break
if job.total:
job.progress += 1
if not track.available:
job.current_track = core.full_title(track)
job.message = f"Пропущено: {job.current_track}"
job.log(f"[track] {job.message}")
continue
save_path = base_dir / core.prepare_base_path(
path_pattern,
track,
req.unsafe_path,
fallback_number=seq,
fallback_total=job.total,
)
if req.skip_existing:
if any(Path(str(save_path) + s).is_file() for s in core.AUDIO_FILE_SUFFIXES):
job.current_track = core.full_title(track)
job.message = f"Пропуск: уже есть {job.current_track}"
job.log(f"[track] {job.message}")
continue
save_path.parent.mkdir(parents=True, exist_ok=True)
downloadable = core.to_downloadable_track(track, quality, save_path)
bitrate = downloadable.download_info.bitrate
format_info = "[" + downloadable.download_info.file_format.codec.name
if bitrate > 0:
format_info += f" {bitrate}kbps"
format_info += "]"
display_title = core.full_title(track)
job.current_track = display_title
job.message = f"Скачивается: {display_title}"
job.log(f"[track] {job.message}")
job.log(f"{format_info} {downloadable.path}")
core.download_track(
track_info=downloadable,
lyrics_format=lyrics_format,
embed_cover=req.embed_cover,
cover_resolution=req.cover_resolution,
covers_cache=covers_cache if req.embed_cover else None,
compatibility_level=req.compatibility_level,
path_pattern=path_pattern,
unsafe_path=req.unsafe_path,
fallback_number=seq,
fallback_total=job.total,
rename_using_tags=True,
)
downloaded += 1
job.current_track = None
if req.delay:
time.sleep(req.delay)
job.status = "done"
job.message = f"Готово. Скачано {downloaded} трек(ов)"
job.finished_at = time.time()
job.log(job.message)
def start_download_job(req: DownloadRequest) -> Job:
req = validate_and_fill_request(req)
job = register_job(Job(id=str(uuid.uuid4()), type="download", status="running"))
def worker():
try:
job.log("Инициализация клиента")
client = init_web_client(req.timeout, req.tries, req.retry_delay)
tracks, total = build_track_iter(client, req)
download_tracks_for_job(job, req, tracks, total)
except Exception as exc: # noqa: BLE001
job.status = "failed"
job.message = str(exc)
job.finished_at = time.time()
job.log(f"Ошибка: {exc}")
thread = threading.Thread(target=worker, name=f"download-{job.id}", daemon=True)
thread.start()
return job
def start_likes_download_job(
req: DownloadRequest, job_type: Literal["likes_tracks", "likes_artists_albums"]
) -> Job:
require_default_token()
job = register_job(Job(id=str(uuid.uuid4()), type=job_type, status="running"))
def worker():
try:
job.log("Инициализация клиента")
client = init_web_client(req.timeout, req.tries, req.retry_delay)
if job_type == "likes_tracks":
track_ids = collect_liked_track_ids(client)
if not track_ids:
raise ValueError("Лайкнутые треки не найдены")
job.log(f"Лайкнутых треков: {len(track_ids)}")
else:
job.log("Собираем артистов из лайкнутых треков")
artist_ids = collect_liked_artist_ids(client)
if not artist_ids:
raise ValueError("Артисты в лайках не найдены")
job.log(f"Артистов из лайков: {len(artist_ids)}")
album_ids = collect_artist_album_ids(client, artist_ids, req)
if not album_ids:
raise ValueError("У артистов из лайков не найдено подходящих альбомов")
job.log(f"Подходящих альбомов: {len(album_ids)}")
track_ids = collect_album_track_ids(client, album_ids)
if not track_ids:
raise ValueError("Не удалось собрать треки из альбомов артистов")
job.log(f"Собрано треков: {len(track_ids)}")
tracks = build_track_iter_from_ids(client, track_ids)
download_tracks_for_job(job, req, tracks, len(track_ids))
except Exception as exc: # noqa: BLE001
job.status = "failed"
job.message = str(exc)
job.finished_at = time.time()
job.log(f"Ошибка: {exc}")
thread = threading.Thread(
target=worker, name=f"{job_type}-{job.id}", daemon=True
)
thread.start()
return job
def export_discography(req: DiscographyRequest):
require_default_token()
if req.url and not req.artist_id:
parsed = parse_ids_from_url(req.url)
req.artist_id = parsed.get("artist_id")
if not req.artist_id:
raise HTTPException(400, detail="Нужен artist_id или ссылка на артиста")
client = init_web_client(req.timeout, req.tries, req.retry_delay)
artist = client.artists(req.artist_id)[0]
result = {
"artist": {
"id": artist.id,
"name": artist.name,
"counts": artist.counts.dict() if artist.counts else None,
},
"albums": [],
}
page = 0
while True:
albums_info = client.artists_direct_albums(req.artist_id, page)
if not albums_info:
break
for album in albums_info.albums:
if not album.id or not album.available:
continue
if req.only_music and album.meta_type != "music":
continue
if req.stick_to_artist and album.artists and album.artists[0].id != int(
req.artist_id
):
continue
album_full = client.albums_with_tracks(album.id)
tracks_data = []
if album_full and album_full.volumes:
for volume in album_full.volumes:
for track in volume:
tracks_data.append(
{
"id": track.id,
"title": core.full_title(track),
"available": track.available,
"duration_ms": track.duration_ms,
"version": track.version,
"has_lyrics": bool(track.lyrics_info),
}
)
result["albums"].append(
{
"id": album.id,
"title": core.full_title(album),
"year": album.year,
"genre": album.genre,
"meta_type": album.meta_type,
"track_count": album.track_count,
"tracks": tracks_data,
}
)
if pager := albums_info.pager:
page = pager.page + 1
if pager.per_page * page >= pager.total:
break
else:
break
return result
app = FastAPI(title="Yandex Music Downloader Web", version="1.0.0")
@app.get("/", response_class=HTMLResponse)
def index():
config_json = json.dumps(frontend_config(), ensure_ascii=False)
return HTMLResponse(INDEX_HTML.replace("__CONFIG_JSON__", config_json))
@app.get("/health")
def health():
return {"status": "ok"}
@app.get("/api/likes")
def api_likes():
return JSONResponse(export_likes())
@app.post("/api/download")
def api_download(req: DownloadRequest):
job = start_download_job(req)
return serialize_job(job)
@app.post("/api/download/likes/all")
def api_download_likes_all(req: DownloadRequest):
job = start_likes_download_job(req, "likes_tracks")
return serialize_job(job)
@app.post("/api/download/likes/artists-albums")
def api_download_likes_artists(req: DownloadRequest):
job = start_likes_download_job(req, "likes_artists_albums")
return serialize_job(job)
@app.get("/api/jobs")
def api_jobs():
with job_lock:
return [serialize_job(job) for job in jobs.values()]
@app.get("/api/jobs/{job_id}")
def api_job(job_id: str):
try:
job = snapshot_job(job_id)
except KeyError:
raise HTTPException(404, detail="Задача не найдена")
return serialize_job(job)
@app.post("/api/discography")
def api_discography(req: DiscographyRequest):
data = export_discography(req)
return JSONResponse(data)
def serialize_job(job: Job) -> dict:
percent = 0
if job.total:
percent = int(min(100, round(job.progress / job.total * 100)))
return {
"id": job.id,
"type": job.type,
"status": job.status,
"message": job.message,
"progress": job.progress,
"total": job.total,
"percent": percent,
"created_at": job.created_at,
"finished_at": job.finished_at,
"current_track": job.current_track,
"logs": job.logs,
"recent_logs": job.logs[-8:],
}
def main():
import argparse
import uvicorn
parser = argparse.ArgumentParser(description="Yandex Music Downloader Web UI")
parser.add_argument("--host", default="127.0.0.1", help="Хост для прослушивания")
parser.add_argument("--port", default=8000, type=int, help="Порт")
parser.add_argument(
"--reload", action="store_true", help="Перезагрузка при изменении файлов"
)
args = parser.parse_args()
uvicorn.run(
"ymd.webui:app", host=args.host, port=args.port, reload=args.reload, factory=False
)
if __name__ == "__main__":
main()