转自微信公众号 侯plus
Qobuz音乐平台介绍
Qobuz是一个受欢迎的音乐流媒体平台,提供超过1亿首歌曲的访问。 所有歌曲均以高保真的FLAC(无损音频编解码器)格式提供。 凭借其令人印象深刻的音频质量和广泛的目录,Qobuz已成为音响发烧友和音乐爱好者的首选平台。 Qobuz于2007年在法国创立。 Qubuz在无损音乐方面最高能提供24bit/192kHz规格的高解析度无损音乐。 但与TIDAL不同的是,TIDAL在高解析度音乐是采用MQA编码后进行传输的,同样规格下的传输码流要比Qobuz要小,而Qobuz不采用MQA编码(MQA为有损编码),传输码流相对更大。 月付费是13刀一个月,我们要搬运Qobuz的音乐需要至少一个收费账户。
搬运机器人框架添加proxy模块github上有qobuz的下载机器人框架,我选择的是Sorrow446写的Qo-DL下载机器人框架。 不过这个代码年久失修,已经有5年没有更新,和最新的网页已经不能适配,需要自己修补和改造。 具体网页解析功能修补的过程不再赘述,每次网页改版均需要升级代码做适配。主要记录下主程序改造的过程。 首先qobuz在国内是无法登录的,需要飞机场。 机器人框架不支持飞机场需要做第一个改造,添加proxy模块。 在qopy.py中添加proxy,代码参考如下: - class Client:
- def __init__(self, email, <font color="#0086b3">pwd</font>, app_id, secrets):
- logger.info(f<font color="#d14">"{YELLOW}Logging...,app_id是{app_id}"</font>)
- self.secrets = secrets
- self.id = str(app_id)
- self.session = requests.Session()
- self.session.proxies = {<font color="#d14">'https://127.0.0.1'</font>: <font color="#d14">'foo.bar:1081'</font>}
- self.session.headers.update(
- {
- <font color="#d14">"User-Agent"</font>: <font color="#d14">"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:83.0) Gecko/20100101 Firefox/83.0"</font>,
- <font color="#d14">"X-App-Id"</font>: self.id,
- <font color="#d14">"Content-Type"</font>: <font color="#d14">"application/json;charset=UTF-8"</font>,
- <font color="#d14">'Connection'</font>: <font color="#d14">'close'</font>
- }
- )
- self.base = <font color="#d14">"https://www.qobuz.com/api.json/0.2/"</font>
- self.sec = None
- self.auth(email, <font color="#0086b3">pwd</font>)
- self.cfg_setup()
复制代码
搬运机器人框架改造专辑下载模块在对qobuz平台的地址解析改造完成后,需要对专辑下载模块进行改造。 专辑下载模块存在的主要问题是limit固定是500,也就是只能一次最多下载500首歌。 通过改造可以绕过limit限制,进行无限制的下载。 机器人框架download模块的改造 download模块改造函数def download_release(self),改造前代码: - def download_release(self):
- count = 0
- meta = self.client.get_album_meta(self.item_id)
- <font color="#333"><b>if</b></font> not meta.get(<font color="#d14">"streamable"</font>):
- raise NonStreamable(<font color="#d14">"This release is not streamable"</font>)
- <font color="#333"><b>if</b></font> self.albums_only and (
- meta.get(<font color="#d14">"release_type"</font>) != <font color="#d14">"album"</font>
- or meta.get(<font color="#d14">"artist"</font>).get(<font color="#d14">"name"</font>) == <font color="#d14">"Various Artists"</font>
- ):
- logger.info(f<font color="#d14">'{OFF}Ignoring Single/EP/VA: {meta.get("title", "n/a")}'</font>)
- <font color="#0086b3">return</font>
- album_title = _get_title(meta)
- format_info = self._get_format(meta)
- file_format, quality_met, bit_depth, sampling_rate = format_info
- <font color="#333"><b>if</b></font> not self.downgrade_quality and not quality_met:
- logger.info(
- f<font color="#d14">"{OFF}Skipping {album_title} as it doesn't meet quality requirement"</font>
- )
- <font color="#0086b3">return</font>
- logger.info(
- f<font color="#d14">"\n{YELLOW}Downloading: {album_title}\nQuality: {file_format}"</font>
- f<font color="#d14">" ({bit_depth}/{sampling_rate})\n"</font>
- )
- album_attr = self._get_album_attr(
- meta, album_title, file_format, bit_depth, sampling_rate
- )
- folder_format, track_format = _clean_format_str(
- self.folder_format, self.track_format, file_format
- )
- sanitized_title = sanitize_filepath(folder_format.format(**album_attr))
- dirn = os.path.join(self.path, sanitized_title)
- os.makedirs(dirn, exist_ok=True)
- <font color="#333"><b>if</b></font> self.no_cover:
- logger.info(f<font color="#d14">"{OFF}Skipping cover"</font>)
- <font color="#333"><b>else</b></font>:
- _get_extra(meta[<font color="#d14">"image"</font>][<font color="#d14">"large"</font>], dirn, og_quality=self.cover_og_quality)
- <font color="#333"><b>if</b></font> <font color="#d14">"goodies"</font> <font color="#333"><b>in</b></font> meta:
- try:
- _get_extra(meta[<font color="#d14">"goodies"</font>][0][<font color="#d14">"url"</font>], dirn, <font color="#d14">"booklet.pdf"</font>)
- except: <font color="#998"><i># noqa</i></font>
- pass
- media_numbers = [track[<font color="#d14">"media_number"</font>] <font color="#333"><b>for</b></font> track <font color="#333"><b>in</b></font> meta[<font color="#d14">"tracks"</font>][<font color="#d14">"items"</font>]]
- is_multiple = True <font color="#333"><b>if</b></font> len([*{*media_numbers}]) > 1 <font color="#333"><b>else</b></font> False
- <font color="#333"><b>for</b></font> i <font color="#333"><b>in</b></font> meta[<font color="#d14">"tracks"</font>][<font color="#d14">"items"</font>]:
- parse = self.client.get_track_url(i[<font color="#d14">"id"</font>], fmt_id=self.quality)
- <font color="#333"><b>if</b></font> <font color="#d14">"sample"</font> not <font color="#333"><b>in</b></font> parse and parse[<font color="#d14">"sampling_rate"</font>]:
- is_mp3 = True <font color="#333"><b>if</b></font> int(self.quality) == 5 <font color="#333"><b>else</b></font> False
- self._download_and_tag(
- dirn,
- count,
- parse,
- i,
- meta,
- False,
- is_mp3,
- i[<font color="#d14">"media_number"</font>] <font color="#333"><b>if</b></font> is_multiple <font color="#333"><b>else</b></font> None,
- )
- <font color="#333"><b>else</b></font>:
- logger.info(f<font color="#d14">"{OFF}Demo. Skipping"</font>)
- count = count + 1
- logger.info(f<font color="#d14">"{GREEN}Completed"</font>)
复制代码
改造后代码: - def download_release(self):
- count = 0
- meta_lists = self.client.get_album_meta(self.item_id)
- <font color="#333"><b>for</b></font> meta <font color="#333"><b>in</b></font> meta_lists:
- <font color="#333"><b>if</b></font> not meta.get(<font color="#d14">"streamable"</font>):
- raise NonStreamable(<font color="#d14">"This release is not streamable"</font>)
- <font color="#333"><b>if</b></font> self.albums_only and (
- meta.get(<font color="#d14">"release_type"</font>) != <font color="#d14">"album"</font>
- or meta.get(<font color="#d14">"artist"</font>).get(<font color="#d14">"name"</font>) == <font color="#d14">"Various Artists"</font>
- ):
- logger.info(f<font color="#d14">'{OFF}Ignoring Single/EP/VA: {meta.get("title", "n/a")}'</font>)
- <font color="#0086b3">return</font>
- album_title = _get_title(meta)
- format_info = self._get_format(meta)
- file_format, quality_met, bit_depth, sampling_rate = format_info
- <font color="#333"><b>if</b></font> not self.downgrade_quality and not quality_met:
- logger.info(
- f<font color="#d14">"{OFF}Skipping {album_title} as it doesn't meet quality requirement"</font>
- )
- <font color="#0086b3">return</font>
- logger.info(
- f<font color="#d14">"\n{YELLOW}Downloading: {album_title}\nQuality: {file_format}"</font>
- f<font color="#d14">" ({bit_depth}/{sampling_rate})\n"</font>
- )
- album_attr = self._get_album_attr(
- meta, album_title, file_format, bit_depth, sampling_rate
- )
- folder_format, track_format = _clean_format_str(
- self.folder_format, self.track_format, file_format
- )
- sanitized_title = sanitize_filepath(folder_format.format(**album_attr))
- dirn = os.path.join(self.path, sanitized_title)
- os.makedirs(dirn, exist_ok=True)
- <font color="#333"><b>if</b></font> self.no_cover:
- logger.info(f<font color="#d14">"{OFF}Skipping cover"</font>)
- <font color="#333"><b>else</b></font>:
- _get_extra(meta[<font color="#d14">"image"</font>][<font color="#d14">"large"</font>], dirn, og_quality=self.cover_og_quality)
- <font color="#333"><b>if</b></font> <font color="#d14">"goodies"</font> <font color="#333"><b>in</b></font> meta:
- try:
- _get_extra(meta[<font color="#d14">"goodies"</font>][0][<font color="#d14">"url"</font>], dirn, <font color="#d14">"booklet.pdf"</font>)
- except: <font color="#998"><i># noqa</i></font>
- pass
- media_numbers = [track[<font color="#d14">"media_number"</font>] <font color="#333"><b>for</b></font> track <font color="#333"><b>in</b></font> meta[<font color="#d14">"tracks"</font>][<font color="#d14">"items"</font>]]
- is_multiple = True <font color="#333"><b>if</b></font> len([*{*media_numbers}]) > 1 <font color="#333"><b>else</b></font> False
- <font color="#333"><b>for</b></font> i <font color="#333"><b>in</b></font> meta[<font color="#d14">"tracks"</font>][<font color="#d14">"items"</font>]:
- parse = self.client.get_track_url(i[<font color="#d14">"id"</font>], fmt_id=self.quality)
- <font color="#333"><b>if</b></font> <font color="#d14">"sample"</font> not <font color="#333"><b>in</b></font> parse and parse[<font color="#d14">"sampling_rate"</font>]:
- is_mp3 = True <font color="#333"><b>if</b></font> int(self.quality) == 5 <font color="#333"><b>else</b></font> False
- self._download_and_tag(
- dirn,
- count,
- parse,
- i,
- meta,
- False,
- is_mp3,
- i[<font color="#d14">"media_number"</font>] <font color="#333"><b>if</b></font> is_multiple <font color="#333"><b>else</b></font> None,
- )
- <font color="#333"><b>else</b></font>:
- logger.info(f<font color="#d14">"{OFF}Demo. Skipping"</font>)
- count = count + 1
- logger.info(f<font color="#d14">"{GREEN}Completed"</font>)
复制代码
机器人框架qopy模块的改造 qopy模块主要改造def get_album_meta(self, id)函数,改造前代码: - def get_album_meta(self, id):
- <font color="#0086b3">return</font> self.api_call(<font color="#d14">"album/get"</font>, id=id)
复制代码
改造后代码: - def get_album_meta(self, id):
- <font color="#0086b3">limit</font> = 500
- offset = 0
- meta_lists = []
- meta_url = self.api_call(<font color="#d14">"album/get"</font>, id=id, offset=offset)
- total = meta_url[<font color="#d14">"tracks"</font>][<font color="#d14">"total"</font>]
- logger.info(f<font color="#d14">"{YELLOW}{total} albums in the albums !"</font>)
- <font color="#333"><b>while</b></font> total > 0:
- meta = self.api_call(
- <font color="#d14">"album/get"</font>, id=id, offset=offset)
- meta_lists.append(meta)
- total -= <font color="#0086b3">limit</font>
- offset += 1
- <font color="#0086b3">return</font> meta_lists
复制代码
主要作用是添加了offset参数给网址做解析,每次解析是500条,超过500条offset参数加一,再解析第501到第1000条。 qopy模块改造def api_call(self, epoint, **kwargs)函数,改造前代码: - def api_call(self, epoint, **kwargs):
- <font color="#333"><b>if</b></font> epoint == <font color="#d14">"user/login"</font>:
- params = {
- <font color="#d14">"email"</font>: kwargs[<font color="#d14">"email"</font>],
- <font color="#d14">"password"</font>: kwargs[<font color="#d14">"pwd"</font>],
- <font color="#d14">"app_id"</font>: self.id,
- }
- <font color="#333"><b>elif</b></font> epoint == <font color="#d14">"track/get"</font>:
- params = {<font color="#d14">"track_id"</font>: kwargs[<font color="#d14">"id"</font>]}
- <font color="#333"><b>elif</b></font> epoint == <font color="#d14">"album/get"</font>:
- params = {<font color="#d14">"album_id"</font>: kwargs[<font color="#d14">"id"</font>]}
复制代码
改造后代码: - def api_call(self, epoint, **kwargs):
- <font color="#333"><b>if</b></font> epoint == <font color="#d14">"user/login"</font>:
- params = {
- <font color="#d14">"email"</font>: kwargs[<font color="#d14">"email"</font>],
- <font color="#d14">"password"</font>: kwargs[<font color="#d14">"pwd"</font>],
- <font color="#d14">"app_id"</font>: self.id,
- }
- <font color="#333"><b>elif</b></font> epoint == <font color="#d14">"track/get"</font>:
- params = {<font color="#d14">"track_id"</font>: kwargs[<font color="#d14">"id"</font>]}
- <font color="#333"><b>elif</b></font> epoint == <font color="#d14">"album/get"</font>:
- params = {<font color="#d14">"album_id"</font>: kwargs[<font color="#d14">"id"</font>],
- <font color="#d14">"offset"</font>: kwargs[<font color="#d14">"offset"</font>],
- }
复制代码
qobuz机器人下载进度条改造 进度条由download模块的def tqdm_download(url, fname, desc, max_retries=10)函数进行控制,改造前代码: - def tqdm_download(url, fname, desc):
- r = requests.get(url, allow_redirects=True, stream=True)
- total = int(r.headers.get(<font color="#d14">"content-length"</font>, 0))
- download_size = 0
- with open(fname, <font color="#d14">"wb"</font>) as file, tqdm(
- total=total,
- unit=<font color="#d14">"iB"</font>,
- unit_scale=True,
- unit_divisor=1024,
- desc=desc,
- bar_format=CYAN + <font color="#d14">"{n_fmt}/{total_fmt} /// {desc}"</font>,
- ) as bar:
- <font color="#333"><b>for</b></font> data <font color="#333"><b>in</b></font> r.iter_content(chunk_size=1024):
- size = file.write(data)
- bar.update(size)
- download_size += size
- <font color="#333"><b>if</b></font> total != download_size:
- <font color="#998"><i># <a href="https://stackoverflow.com/questions/69919912/requests-iter-content-thinks-file-is-complete-but-its-not" target="_blank">https://stackoverflow.com/questi ... omplete-but-its-not</a></i></font>
- raise ConnectionError(<font color="#d14">"File download was interrupted for "</font> + fname)
复制代码
改造后代码: - def tqdm_download(url, fname, desc, max_retries=10):
- pbar = None
- miniters = 10 <font color="#998"><i># 设置 miniters 参数,以达到动态进度条效果</i></font>
- <font color="#333"><b>for</b></font> retry_count <font color="#333"><b>in</b></font> range(max_retries):
- try:
- r = requests.get(url,
- allow_redirects=True,
- stream=False,
- timeout=30,
- headers={<font color="#d14">'Connection'</font>: <font color="#d14">'close'</font>}, ) <font color="#998"><i># 增加超时时间</i></font>
- r.raise_for_status()
- total = int(r.headers.get(<font color="#d14">"content-length"</font>, 0))
- download_size = 0
- with open(fname, <font color="#d14">"wb"</font>) as file:
- pbar = tqdm(
- total=total,
- unit=<font color="#d14">"iB"</font>,
- unit_scale=True,
- unit_divisor=1024,
- desc=desc,
- dynamic_ncols=True, <font color="#998"><i># 使用 dynamic_ncols 参数显示实时速度</i></font>
- miniters=miniters,
- bar_format=CYAN + <font color="#d14">"{n_fmt}/{total_fmt} /// {desc} /// {rate_fmt}"</font>,
- )
- start_time = time.time()
- <font color="#333"><b>for</b></font> data <font color="#333"><b>in</b></font> r.iter_content(chunk_size=1024):
- size = file.write(data)
- pbar.update(size)
- download_size += size
- elapsed_time = time.time() - start_time
- current_speed = size / elapsed_time <font color="#333"><b>if</b></font> elapsed_time > 0 <font color="#333"><b>else</b></font> 0
- pbar.set_postfix(speed=f<font color="#d14">"{current_speed / 1024:.2f} KB/s"</font>)
- <font color="#333"><b>if</b></font> total == download_size:
- <font color="#0086b3">break</font> <font color="#998"><i># 下载成功,退出循环</i></font>
- <font color="#333"><b>else</b></font>:
- <font color="#998"><i># <a href="https://stackoverflow.com/questions/69919912/requests-iter-content-thinks-file-is-complete-but-its-not" target="_blank">https://stackoverflow.com/questi ... omplete-but-its-not</a></i></font>
- raise ConnectionError(<font color="#d14">"File download was interrupted for "</font> + fname)
- except (requests.exceptions.RequestException, ConnectionError, urllib3.exceptions.ProtocolError) as e:
- logger.error(f<font color="#d14">"{RED}Error during download: {e}"</font>, exc_info=True)
- <font color="#333"><b>if</b></font> retry_count < max_retries - 1:
- <font color="#998"><i># 间隔一段时间后进行下一次重试</i></font>
- time.sleep(5)
- <font color="#0086b3">continue</font>
- <font color="#333"><b>else</b></font>:
- logger.error(f<font color="#d14">"{RED}Max retries reached. Unable to download: {fname}"</font>)
- <font color="#0086b3">break</font> <font color="#998"><i># 如果达到最大重试次数仍然失败,退出循环</i></font>
- finally:
- <font color="#333"><b>if</b></font> pbar is not None:
- pbar.close()
复制代码
经过改造后就能正常搬运qobuz音乐平台几乎所有的音乐到本地存储了。
|