mirror of
https://github.com/n0stal6ic/unshackle-services.git
synced 2026-07-15 17:40:02 +02:00
Add files via upload
This commit is contained in:
n0stal6ic
@@ -0,0 +1,501 @@
|
||||
from __future__ import annotations
|
||||
import base64
|
||||
import hashlib
|
||||
import re
|
||||
import time
|
||||
from http.cookiejar import CookieJar
|
||||
from typing import Any, Optional
|
||||
import click
|
||||
import requests
|
||||
from unshackle.core.config import config
|
||||
from unshackle.core.credential import Credential
|
||||
from unshackle.core.music import MusicTrackOption
|
||||
from unshackle.core.service import Service
|
||||
from unshackle.core.titles import Music, Song, Titles_T
|
||||
from unshackle.core.tracks import Audio, Chapters, Tracks
|
||||
from unshackle.core.tracks.track import Track
|
||||
|
||||
|
||||
class QOBZ(Service):
|
||||
"""
|
||||
Service code for Qobuz (https://qobuz.com)
|
||||
|
||||
Author: n0stal6ic
|
||||
Authorization: Credentials, Tokens.
|
||||
Security: None
|
||||
"""
|
||||
|
||||
ALIASES = ("QOBZ", "qobuz")
|
||||
GROUP_AUDIO_DOWNLOADS = True
|
||||
|
||||
TITLE_RE = r"^(?:https?://(?:www\.|open\.|play\.)?qobuz\.com/(?:[a-z]{2}-[a-z]{2}/)?(?P<type>album|track|playlist|interpreter|artist|label)/(?:[^/]+/)*)?(?P<id>[A-Za-z0-9]+)"
|
||||
|
||||
FORMATS = {
|
||||
5: ("MP3", "MP3 320 kb/s"),
|
||||
6: ("FLAC", "FLAC 16-bit/44.1kHz"),
|
||||
7: ("FLAC", "FLAC 24-bit ≤96kHz"),
|
||||
27: ("FLAC", "FLAC 24-bit ≤192kHz"),
|
||||
}
|
||||
|
||||
QUALITY_MAP = {
|
||||
"MP3": 5, "CD": 6, "HIFI": 7, "HIRES": 27,
|
||||
"5": 5, "6": 6, "7": 7, "27": 27,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
@click.command(name="QOBZ", short_help="https://qobuz.com", help=__doc__)
|
||||
@click.argument("title", type=str)
|
||||
@click.option("-q", "--quality", "quality",
|
||||
type=click.Choice(["MP3", "CD", "HIFI", "HIRES", "5", "6", "7", "27"], case_sensitive=False),
|
||||
default=None,
|
||||
help="Quality: MP3/5=MP3 320, CD/6=FLAC 16/44.1, HIFI/7=FLAC 24/96, "
|
||||
"HIRES/27=FLAC 24/192 (default).")
|
||||
@click.pass_context
|
||||
def cli(ctx, **kwargs):
|
||||
return QOBZ(ctx, **kwargs)
|
||||
|
||||
def __init__(self, ctx, title: str, quality: Optional[str]):
|
||||
super().__init__(ctx)
|
||||
self.title = title
|
||||
if quality:
|
||||
self.quality = self.QUALITY_MAP[quality.upper()]
|
||||
else:
|
||||
self.quality = int(self.config.get("default_format_id", 27))
|
||||
|
||||
self.app_id: Optional[str] = None
|
||||
self.secrets: list[str] = []
|
||||
self.valid_secret: Optional[str] = None
|
||||
self.user_auth_token: Optional[str] = None
|
||||
|
||||
m = re.search(self.TITLE_RE, self.title)
|
||||
self.item_type = (m.group("type") if m and m.group("type") else "album")
|
||||
self.item_id = m.group("id") if m else self.title
|
||||
|
||||
def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None:
|
||||
super().authenticate(cookies, credential)
|
||||
self.session.headers.update({"User-Agent": self.config["user_agent"]})
|
||||
|
||||
self.app_id = str(self.config.get("app_id") or "").strip() or None
|
||||
self.secrets = [s for s in (self.config.get("secrets") or []) if s]
|
||||
if not self.app_id or not self.secrets:
|
||||
spoofed_id, spoofed_secrets = self._spoof_app_credentials()
|
||||
self.app_id = self.app_id or spoofed_id
|
||||
if not self.secrets:
|
||||
self.secrets = spoofed_secrets
|
||||
if not self.app_id:
|
||||
self.log.error(" - Could not find a Qobuz app_id."); raise SystemExit(1)
|
||||
self.session.headers["X-App-Id"] = self.app_id
|
||||
|
||||
token = None
|
||||
if credential:
|
||||
user = (credential.username or "").strip()
|
||||
pw = (credential.password or "").strip()
|
||||
if user.lower() in ("token", "auth", "authtoken", "user_auth_token"):
|
||||
token = pw
|
||||
elif user and pw:
|
||||
token = self._login(user, pw)
|
||||
elif pw and not user:
|
||||
token = pw
|
||||
elif user and not pw:
|
||||
token = None if "@" in user else user
|
||||
if not token:
|
||||
token = str(self.config.get("user_auth_token") or "").strip() or None
|
||||
if not token and cookies:
|
||||
for cookie in cookies:
|
||||
if cookie.name in ("X-User-Auth-Token", "user_auth_token", "qobuz_token"):
|
||||
token = cookie.value
|
||||
break
|
||||
|
||||
if not token:
|
||||
self.log.error(
|
||||
" - No Qobuz auth. In unshackle.yaml credentials, use either "
|
||||
"'token:YOUR_AUTH_TOKEN' or 'email:password'."
|
||||
)
|
||||
raise SystemExit(1)
|
||||
|
||||
self.user_auth_token = token
|
||||
self.session.headers["X-User-Auth-Token"] = token
|
||||
self.log.info(" + Authenticated with Qobuz")
|
||||
|
||||
def _login(self, email: str, password: str) -> str:
|
||||
login_app_id, _ = self._spoof_app_credentials()
|
||||
login_app_id = login_app_id or self.app_id
|
||||
self.log.info(f" + Logging in with app_id={login_app_id}")
|
||||
attempts = [("plain", password), ("md5", hashlib.md5(password.encode()).hexdigest())]
|
||||
for label, pwd in attempts:
|
||||
resp = self.session.get(
|
||||
self.config["base_url"] + "user/login",
|
||||
params={"email": email, "password": pwd, "app_id": login_app_id},
|
||||
headers={"X-App-Id": login_app_id},
|
||||
)
|
||||
self.log.info(f" login[{label}] HTTP {resp.status_code}: {resp.text[:160]}")
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
token = data.get("user_auth_token")
|
||||
if token:
|
||||
self.log.info(f" + Logged in as {data.get('user', {}).get('display_name', email)}")
|
||||
return token
|
||||
self.log.error(
|
||||
" - Qobuz password login failed. If it's a 401 'User authentication is required', the "
|
||||
"login app_id couldn't be extracted from the web bundle. Use a bare user_auth_token in "
|
||||
"unshackle.yaml credentials instead of email:password."
|
||||
)
|
||||
raise SystemExit(1)
|
||||
|
||||
def _spoof_app_credentials(self) -> tuple[Optional[str], list[str]]:
|
||||
try:
|
||||
login_page = self.session.get(self.config["web_url"] + "/login").text
|
||||
bundle_match = re.search(
|
||||
r'<script src="(/resources/\d+\.\d+\.\d+-[a-z]\d{3}/bundle\.js)"></script>', login_page
|
||||
)
|
||||
if not bundle_match:
|
||||
self.log.warning(" - Could not locate Qobuz bundle.js for credential extraction.")
|
||||
return None, []
|
||||
bundle = self.session.get(self.config["web_url"] + bundle_match.group(1)).text
|
||||
|
||||
app_id_match = re.search(r'production:\{api:\{appId:"(\d+)",appSecret:', bundle)
|
||||
app_id = app_id_match.group(1) if app_id_match else None
|
||||
|
||||
seeds: dict[str, str] = {}
|
||||
for m in re.finditer(
|
||||
r'[a-z]\.initialSeed\("([\w=]+)",window\.utimezone\.(?P<tz>[a-z]+)\)', bundle
|
||||
):
|
||||
seeds[m.group("tz").capitalize()] = m.group(1)
|
||||
|
||||
secrets: list[str] = []
|
||||
for m in re.finditer(
|
||||
r'name:"\w+/(?P<tz>[A-Z][a-z]+)",info:"(?P<info>[\w=]+)",extras:"(?P<extras>[\w=]+)"', bundle
|
||||
):
|
||||
tz = m.group("tz")
|
||||
if tz not in seeds:
|
||||
continue
|
||||
combined = seeds[tz] + m.group("info") + m.group("extras")
|
||||
try:
|
||||
secret = base64.standard_b64decode(combined[:-44]).decode("utf-8")
|
||||
if secret:
|
||||
secrets.append(secret)
|
||||
except Exception:
|
||||
continue
|
||||
self.log.debug(f" + Extracted app_id={app_id}, {len(secrets)} secret(s)")
|
||||
return app_id, secrets
|
||||
except Exception as e:
|
||||
self.log.warning(f" - Qobuz credential extraction failed: {e}")
|
||||
return None, []
|
||||
|
||||
def _api(self, endpoint: str, params: dict, allow_error: bool = False) -> Optional[dict]:
|
||||
resp = self.session.get(self.config["base_url"] + endpoint, params=params)
|
||||
if resp.status_code != 200:
|
||||
if allow_error:
|
||||
return None
|
||||
self.log.error(f" - Qobuz API error on {endpoint}: {resp.status_code} {resp.text[:200]}")
|
||||
raise SystemExit(1)
|
||||
return resp.json()
|
||||
|
||||
def get_titles(self) -> Titles_T:
|
||||
if self.item_type == "track":
|
||||
track = self._api("track/get", params={"track_id": self.item_id})
|
||||
album = self._api("album/get", params={"album_id": track["album"]["id"]})
|
||||
songs = [self._build_song(track, album)]
|
||||
return self._build_music(album, songs, kind="single")
|
||||
|
||||
if self.item_type == "playlist":
|
||||
return self._get_playlist()
|
||||
|
||||
album = self._get_album_full(self.item_id)
|
||||
tracks = album.get("tracks", {}).get("items", [])
|
||||
songs = [self._build_song(t, album) for t in tracks]
|
||||
return self._build_music(album, songs, kind=self._release_kind(album))
|
||||
|
||||
def _get_album_full(self, album_id: str) -> dict:
|
||||
album = self._api("album/get", params={"album_id": album_id, "limit": 500, "offset": 0})
|
||||
items = album.get("tracks", {}).get("items", [])
|
||||
total = album.get("tracks", {}).get("total", len(items))
|
||||
offset = 500
|
||||
while len(items) < total:
|
||||
page = self._api("album/get", params={"album_id": album_id, "limit": 500, "offset": offset})
|
||||
page_items = page.get("tracks", {}).get("items", [])
|
||||
if not page_items:
|
||||
break
|
||||
items.extend(page_items)
|
||||
offset += 500
|
||||
album.setdefault("tracks", {})["items"] = items
|
||||
return album
|
||||
|
||||
def _get_playlist(self) -> Music:
|
||||
playlist = self._api("playlist/get", params={
|
||||
"playlist_id": self.item_id, "extra": "tracks", "limit": 500, "offset": 0,
|
||||
})
|
||||
items = playlist.get("tracks", {}).get("items", [])
|
||||
total = playlist.get("tracks", {}).get("total", len(items))
|
||||
offset = 500
|
||||
while len(items) < total:
|
||||
page = self._api("playlist/get", params={
|
||||
"playlist_id": self.item_id, "extra": "tracks", "limit": 500, "offset": offset,
|
||||
})
|
||||
page_items = page.get("tracks", {}).get("items", [])
|
||||
if not page_items:
|
||||
break
|
||||
items.extend(page_items)
|
||||
offset += 500
|
||||
|
||||
songs = []
|
||||
for position, track in enumerate(items, start=1):
|
||||
album = track.get("album") or {}
|
||||
song = self._build_song(track, album, playlist_position=position)
|
||||
songs.append(song)
|
||||
|
||||
music = Music(
|
||||
songs,
|
||||
kind="playlist",
|
||||
title=playlist.get("name"),
|
||||
artist=(playlist.get("owner") or {}).get("name"),
|
||||
total_tracks=len(songs) or None,
|
||||
owner=(playlist.get("owner") or {}).get("name"),
|
||||
description=(playlist.get("description") or None),
|
||||
)
|
||||
return music
|
||||
|
||||
def _build_music(self, album: dict, songs: list[Song], kind: str) -> Music:
|
||||
artwork = self._cover_url(album)
|
||||
year = self._year(album)
|
||||
return Music(
|
||||
songs,
|
||||
kind=kind,
|
||||
title=self._album_title(album),
|
||||
artist=(album.get("artist") or {}).get("name"),
|
||||
year=year,
|
||||
total_tracks=album.get("tracks_count") or (len(songs) or None),
|
||||
total_discs=album.get("media_count") or None,
|
||||
artwork_url=artwork,
|
||||
total_duration=int(album.get("duration") or 0) or None,
|
||||
)
|
||||
|
||||
def _build_song(self, track: dict, album: dict, playlist_position: Optional[int] = None) -> Song:
|
||||
album = album or track.get("album") or {}
|
||||
album_artist = (album.get("artist") or {}).get("name") or "Various Artists"
|
||||
performer = (track.get("performer") or {}).get("name") or album_artist
|
||||
year = self._year(album) or self._year(track) or 1
|
||||
artwork = self._cover_url(album)
|
||||
release_date = self._release_date(album) or self._release_date(track)
|
||||
|
||||
title = track.get("title") or ""
|
||||
if track.get("version"):
|
||||
title = f"{title.strip()} ({track['version']})"
|
||||
|
||||
genre = (album.get("genre") or {}).get("name")
|
||||
label = (album.get("label") or {}).get("name")
|
||||
composer = (track.get("composer") or {}).get("name")
|
||||
|
||||
data = {
|
||||
"service": self.ALIASES[0],
|
||||
"source": self.ALIASES[0],
|
||||
"track_id": str(track.get("id")),
|
||||
"album_id": str(album.get("id")) if album.get("id") else None,
|
||||
"track_url": track.get("url"),
|
||||
"album_url": album.get("url"),
|
||||
"title": title,
|
||||
"artist": performer,
|
||||
"album": self._album_title(album),
|
||||
"album_artist": album_artist,
|
||||
"performer": performer,
|
||||
"composer": composer,
|
||||
"track_number": track.get("track_number"),
|
||||
"total_tracks": album.get("tracks_count"),
|
||||
"disc_number": track.get("media_number") or 1,
|
||||
"total_discs": album.get("media_count") or 1,
|
||||
"release_date": release_date,
|
||||
"year": year,
|
||||
"genre": genre,
|
||||
"label": label,
|
||||
"isrc": track.get("isrc"),
|
||||
"upc": album.get("upc"),
|
||||
"copyright": album.get("copyright"),
|
||||
"explicit": bool(track.get("parental_warning")),
|
||||
"duration": track.get("duration"),
|
||||
"channels": 2,
|
||||
"artwork_url": artwork,
|
||||
"quality": self.FORMATS.get(self.quality, ("", ""))[1],
|
||||
"maximum_bit_depth": track.get("maximum_bit_depth") or album.get("maximum_bit_depth"),
|
||||
"maximum_sampling_rate": track.get("maximum_sampling_rate") or album.get("maximum_sampling_rate"),
|
||||
}
|
||||
if config.tag:
|
||||
data["comment"] = config.tag
|
||||
|
||||
return Song(
|
||||
id_=str(track.get("id")),
|
||||
service=self.__class__,
|
||||
name=title or "Unknown",
|
||||
artist=performer,
|
||||
album=self._album_title(album) or "Unknown Album",
|
||||
track=int(playlist_position or track.get("track_number") or 1),
|
||||
disc=int(track.get("media_number") or 1),
|
||||
year=int(year),
|
||||
album_artist=album_artist,
|
||||
release_type=self._release_kind(album),
|
||||
total_tracks=int(album["tracks_count"]) if album.get("tracks_count") else None,
|
||||
total_discs=int(album["media_count"]) if album.get("media_count") else None,
|
||||
genre=genre,
|
||||
explicit=bool(track.get("parental_warning")),
|
||||
isrc=track.get("isrc") or None,
|
||||
upc=str(album.get("upc")) if album.get("upc") else None,
|
||||
copyright=album.get("copyright") or None,
|
||||
label=label,
|
||||
artwork_url=artwork,
|
||||
data=data,
|
||||
)
|
||||
|
||||
def get_music_track_options(self, song: Song) -> list[MusicTrackOption]:
|
||||
data = song.data if isinstance(song.data, dict) else {}
|
||||
codec = self.FORMATS.get(self.quality, ("FLAC", ""))[0]
|
||||
bit_depth = None
|
||||
sample_rate = None
|
||||
if self.quality != 5:
|
||||
max_bd = data.get("maximum_bit_depth")
|
||||
max_sr = data.get("maximum_sampling_rate")
|
||||
bit_depth = min(int(max_bd or 24), 16 if self.quality == 6 else 24)
|
||||
sr_cap = {6: 44.1, 7: 96.0, 27: 192.0}.get(self.quality, 192.0)
|
||||
sample_rate = int(min(float(max_sr or sr_cap), sr_cap) * 1000)
|
||||
hires = bool(bit_depth and bit_depth > 16) or bool(sample_rate and sample_rate > 48000)
|
||||
return [MusicTrackOption(
|
||||
codec=codec,
|
||||
bit_depth=bit_depth,
|
||||
sample_rate=sample_rate,
|
||||
bitrate=320000 if self.quality == 5 else None,
|
||||
channels=2.0,
|
||||
lossless=self.quality != 5,
|
||||
hires=hires,
|
||||
explicit=bool(data.get("explicit")),
|
||||
duration=int(data["duration"]) if data.get("duration") else None,
|
||||
quality_label=self.FORMATS.get(self.quality, ("", ""))[1],
|
||||
)]
|
||||
|
||||
def get_tracks(self, title: Song) -> Tracks:
|
||||
track_id = str(title.id)
|
||||
file_info = self._get_file_url(track_id, self.quality)
|
||||
url = file_info.get("url")
|
||||
if not url:
|
||||
self.log.error(f" - No file URL for track {track_id}."); raise SystemExit(1)
|
||||
|
||||
actual_format = int(file_info.get("format_id") or self.quality)
|
||||
is_mp3 = actual_format == 5 or (file_info.get("mime_type") or "").endswith("mpeg")
|
||||
codec = Audio.Codec.FLAC if not is_mp3 else None
|
||||
extension = "mp3" if is_mp3 else "flac"
|
||||
|
||||
bit_depth = file_info.get("bit_depth")
|
||||
sampling_rate = file_info.get("sampling_rate")
|
||||
bitrate = 320000 if is_mp3 else None
|
||||
|
||||
audio = Audio(
|
||||
url,
|
||||
language=title.language or "en",
|
||||
codec=codec,
|
||||
bitrate=bitrate,
|
||||
channels=2,
|
||||
descriptor=Track.Descriptor.URL,
|
||||
id_=track_id,
|
||||
data={
|
||||
"qobuz_ext": extension,
|
||||
"bit_depth": bit_depth,
|
||||
"sampling_rate": sampling_rate,
|
||||
},
|
||||
)
|
||||
return Tracks([audio])
|
||||
|
||||
def get_chapters(self, title: Song) -> Chapters:
|
||||
return Chapters()
|
||||
|
||||
def _get_file_url(self, track_id: str, format_id: int) -> dict:
|
||||
secrets_to_try = ([self.valid_secret] if self.valid_secret else []) + [
|
||||
s for s in self.secrets if s != self.valid_secret
|
||||
]
|
||||
last_error = ""
|
||||
for secret in secrets_to_try:
|
||||
ts = int(time.time())
|
||||
sig_str = f"trackgetFileUrlformat_id{format_id}intentstreamtrack_id{track_id}{ts}{secret}"
|
||||
request_sig = hashlib.md5(sig_str.encode()).hexdigest()
|
||||
resp = self.session.get(
|
||||
self.config["base_url"] + "track/getFileUrl",
|
||||
params={
|
||||
"request_ts": ts,
|
||||
"request_sig": request_sig,
|
||||
"track_id": track_id,
|
||||
"format_id": format_id,
|
||||
"intent": "stream",
|
||||
},
|
||||
)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
if data.get("url"):
|
||||
self.valid_secret = secret
|
||||
return data
|
||||
last_error = data.get("message", "no url in response")
|
||||
else:
|
||||
last_error = f"HTTP {resp.status_code}: {resp.text[:120]}"
|
||||
self.log.error(f" - Failed to get file URL for track {track_id}: {last_error}")
|
||||
raise SystemExit(1)
|
||||
|
||||
def on_track_downloaded(self, track: Any) -> None:
|
||||
try:
|
||||
path = getattr(track, "path", None)
|
||||
data = getattr(track, "data", None)
|
||||
if not path or not path.exists() or not isinstance(data, dict):
|
||||
return
|
||||
extension = data.get("qobuz_ext")
|
||||
if not extension or path.suffix.lower() == f".{extension}":
|
||||
return
|
||||
new_path = path.with_suffix(f".{extension}")
|
||||
if new_path.exists():
|
||||
new_path.unlink()
|
||||
path.rename(new_path)
|
||||
track.path = new_path
|
||||
except Exception as e:
|
||||
self.log.debug(f"Extension rename skipped: {e}")
|
||||
|
||||
@staticmethod
|
||||
def _album_title(album: dict) -> str:
|
||||
title = album.get("title") or ""
|
||||
if album.get("version"):
|
||||
title = f"{title.strip()} ({album['version']})"
|
||||
return title.strip()
|
||||
|
||||
@staticmethod
|
||||
def _year(obj: dict) -> int:
|
||||
for key in ("release_date_original", "release_date_stream", "released_at", "release_date"):
|
||||
value = obj.get(key)
|
||||
if not value:
|
||||
continue
|
||||
if isinstance(value, (int, float)):
|
||||
try:
|
||||
return int(time.gmtime(int(value)).tm_year)
|
||||
except Exception:
|
||||
continue
|
||||
match = re.match(r"(\d{4})", str(value))
|
||||
if match:
|
||||
return int(match.group(1))
|
||||
return 0
|
||||
|
||||
@staticmethod
|
||||
def _release_date(obj: dict) -> Optional[str]:
|
||||
for key in ("release_date_original", "release_date_stream", "release_date"):
|
||||
value = obj.get(key)
|
||||
if value and re.match(r"\d{4}-\d{2}-\d{2}", str(value)):
|
||||
return str(value)
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _cover_url(album: dict) -> Optional[str]:
|
||||
image = album.get("image") or {}
|
||||
url = image.get("large") or image.get("small") or image.get("thumbnail")
|
||||
if not url:
|
||||
return None
|
||||
return re.sub(r"_(\d+|max|org)\.jpg", "_org.jpg", url)
|
||||
|
||||
@staticmethod
|
||||
def _release_kind(album: dict) -> str:
|
||||
release_type = (album.get("release_type") or album.get("product_type") or "").lower()
|
||||
if release_type in ("single", "ep", "compilation", "album"):
|
||||
return release_type
|
||||
tracks_count = album.get("tracks_count") or 0
|
||||
if tracks_count and tracks_count <= 3:
|
||||
return "single"
|
||||
return "album"
|
||||
Reference in New Issue
Block a user