From 96f8f793408bd3305cb8ce7f122bd781bcb28de3 Mon Sep 17 00:00:00 2001 From: lis1991 <127238619+lis1991@users.noreply.github.com> Date: Sun, 19 Mar 2023 17:40:41 +0300 Subject: [PATCH] Create zvukdown_fixed.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Изменения относительно родного скрипта: Исправил таймер ожидания ссылки на трек. Убрал изменение размера обложки, считаю это излишним усложнением. Каждый может сам под себя размер поменять. Добавил вывод дополнительной информации при скачивании. В имени папки поменял местати год и название альбома. --- zvukdown_fixed.py | 333 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 333 insertions(+) create mode 100644 zvukdown_fixed.py diff --git a/zvukdown_fixed.py b/zvukdown_fixed.py new file mode 100644 index 0000000..63a40d4 --- /dev/null +++ b/zvukdown_fixed.py @@ -0,0 +1,333 @@ +import glob +import os +import subprocess +import sys +import time +from pathlib import Path +from shutil import copyfile +from typing import Any + +import requests +from mutagen.flac import FLAC, Picture + +class zvukdown_: + def __init__(self): + self.verify = True + self.headers = [] + # self.token = self.read_token() + pass + + def read_token(self): + import os.path + if os.path.exists('token.txt'): + with open("token.txt", "r", encoding="utf8") as f: + token = f.read() + if len(token) != 32: + raise Exception("Wrong token length") + self.headers = {"x-auth-token": token} + else: + print('Нет файла token.txt') + + def save_token(self, login, password): + url = "https://zvuk.com/api/tiny/login/email" + params = { + "register": "true" + } + data = { + "email": login, + "password": password, + } + r = requests.post(url, params=params, data=data, verify=self.verify) + r.raise_for_status() + resp = r.json(strict=False) + if "result" in resp: + if "token" in resp["result"]: + with open("token.txt", "w", encoding="utf8") as f: + token = resp["result"]["token"] + f.write(token) + if len(token) != 32: + raise Exception("Wrong token length") + self.headers = {"x-auth-token": token} + + @staticmethod + def __ntfs(filename): + for ch in ['<', '>', '@', '%', '!', '+', ':', '"', '/', '\\', '|', '?', '*']: + if ch in filename: + filename = filename.replace(ch, "_") + filename = " ".join(filename.split()) + filename = filename.replace(" .flac", ".flac") + return filename + + @staticmethod + def __launch(args): + try: + pipe = subprocess.Popen(args, creationflags=0x08000000, stdin=subprocess.PIPE, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True) + output, err = pipe.communicate() + pipe.wait() + + if pipe.returncode != 0: + print(args + '\n') + print(output + '\n') + print(err) + raise Exception("Unable to launch") + return output + except FileNotFoundError: + return "Install pingo and imagemagick!" + + @staticmethod + def __to_str(l): + global int + if isinstance(l, int): + return [l] + elif not isinstance(l, str): + l = [str(int) for int in l] + l = ",".join(l) + l = str(l.strip('[]')) + return l + + def __get_copyright(self, label_ids): + label_ids = self.__to_str(label_ids) + url = f"https://zvuk.com/api/tiny/labels" + params = { + "ids": label_ids + } + r = requests.get(url, params=params, verify=self.verify) + r.raise_for_status() + resp = r.json(strict=False) + + info = {} + for i in resp['result']['labels'].values(): + info[i["id"]] = i['title'] + return (info) + + def __get_tracks_metadata(self, track_ids): + track_ids = self.__to_str(track_ids) + params = { + "ids": track_ids + } + url = "https://zvuk.com/api/tiny/tracks" + r = requests.get(url, params=params, + headers=self.headers, verify=self.verify) + r.raise_for_status() + resp = r.json(strict=False) + info = {} + for s in resp['result']['tracks'].values(): + if s['has_flac']: + author = s['credits'] + name = s['title'] + album = s['release_title'] + release_id = s['release_id'] + track_id = s['id'] + if s['genres']: + # genre = s['genres'][0] + genre = ", ".join(s['genres']) + else: + genre = "" + + number = s["position"] + image = s['image']['src'].replace(r"&size={size}&ext=jpg", "") + + info[track_id] = {"author": author, "name": name, "album": album, "release_id": release_id, + "track_id": track_id, "genre": genre, "number": number, "image": image} + else: + if s['highest_quality'] != "flac": + raise Exception( + "has_flac, but highest_quality is not flac, token is invalid") + raise Exception(f"Skipping track {s['title']}, no flac") + return info + + def __get_tracks_link(self, track_ids): + links = {} + print(" ") + print("Поиск треков:") + print(" ") + index = 0 + for i in track_ids: + url = "https://zvuk.com/api/tiny/track/stream" + params = { + "id": i, + "quality": "flac" + } + r = requests.get(url, params=params, + headers=self.headers, verify=self.verify) + # r.raise_for_status() + # print(r.raise_for_status()) + resp = r.json(strict=False) + links[i] = resp['result']['stream'] + if links[i] != 0: + index += 1 + print(index,": ",i, "- ", resp['result']['stream']) + time.sleep(3) + return links + + def __get_releases_info(self, release_ids: object) -> object: + release_ids = self.__to_str(release_ids) + + info = {} + url = "https://zvuk.com/api/tiny/releases" + params = { + "ids": release_ids + } + r = requests.get(url, params=params, + headers=self.headers, verify=self.verify) + r.raise_for_status() + resp = r.json(strict=False) + + labels = set() + for i in resp['result']["releases"].values(): + labels.add(i["label_id"]) + labels_info = self.__get_copyright(labels) + + # print(resp) + for a in resp['result']["releases"].values(): + info[a["id"]] = {"track_ids": a["track_ids"], "tracktotal": len(a["track_ids"]), + "copyright": labels_info[a['label_id']], "date": a["date"], "album": a["title"], + "author": a["credits"]} + # print(info) + return info + + def __download_image(self, release_id, image_link): + global err + pic: Path = Path(f"temp_{release_id}.jpg") + comp_pic = Path(f"temp_{release_id}_comp.jpg") + if not pic.is_file(): + # os.system("pause") + r = requests.get(image_link, allow_redirects=True, + verify=self.verify) + open(pic, 'wb').write(r.content) + # print(self.__launch(f'pingo -sa -notime -strip {pic}')) + # if os.path.getsize(pic) > 2 * 1000 * 1000: + # print(self.__launch(f"magick convert {pic} -define jpeg:extent=1MB {comp_pic}")) + # print(self.__launch(f'pingo -sa -notime -strip {comp_pic}')) + # else: + # copyfile(pic, comp_pic) + copyfile(pic, comp_pic) + # pingo optimize, compress + return {"original": pic, "compressed": comp_pic} + # return {"original": pic} + + def __save_track(self, url, metadata, releases, single): + pic = self.__download_image(metadata["release_id"], metadata["image"]) + if not single and releases["tracktotal"] != 1: + folder = f'{releases["author"]} - {str(releases["date"])[0:4]} {releases["album"]}' + folder = self.__ntfs(folder) + if not os.path.exists(folder): + os.makedirs(folder) + copyfile(pic["original"], os.path.join(folder, "cover.jpg")) + # else: + # print("Folder already exist, continue?") + # a = input() + # if not a: + # os._exit() + # os.chdir(folder) + pic = pic["compressed"] + filename = f'{metadata["number"]:02d} - {metadata["name"]}.flac' + else: + pic = pic["original"] + folder = "" + filename = f'{metadata["author"]} - {metadata["name"]}.flac' + + filename = self.__ntfs(filename) + filename = os.path.join(folder, filename) + + r = requests.get(url, allow_redirects=True, verify=self.verify) + open(filename, 'wb').write(r.content) + + audio = FLAC(filename) + audio["ARTIST"] = metadata["author"] + audio["TITLE"] = metadata["name"] + audio["ALBUM"] = metadata["album"] + audio["TRACKNUMBER"] = str(metadata["number"]) + audio["TRACKTOTAL"] = str(releases["tracktotal"]) + + audio["GENRE"] = metadata["genre"] + audio["COPYRIGHT"] = releases["copyright"] + audio["DATE"] = str(releases["date"]) + audio["YEAR"] = str(releases["date"])[0:4] + + audio["RELEASE_ID"] = str(metadata["release_id"]) + audio["TRACK_ID"] = str(metadata["track_id"]) + + covart = Picture() + covart.data = open(pic, 'rb').read() + covart.type = 3 # as the front cover + covart.mime = "image/jpeg" + audio.add_picture(covart) + + # Printing the metadata + print(audio.pprint() + '\n') + + # Saving the changes + audio.save() + time.sleep(1) + + def download_tracks(self, track_ids, single=False, releases=""): + metadata = self.__get_tracks_metadata(track_ids) + link = self.__get_tracks_link(track_ids) + + if len(metadata) != len(link): + raise Exception("metadata != link") + + if not releases: + release_ids = set() + for i in metadata.values(): + release_ids.add(i["release_id"]) + releases = self.__get_releases_info(release_ids) + + print(" ") + print("Скачивание треков") + print(" ") + index = 0 + for i in metadata.keys(): + print(" ") + index += 1 + print("Скачивание трека № ", index, ":") + self.__save_track(link[i], metadata[i], + releases[metadata[i]["release_id"]], single) + + def download_albums(self, release_ids): + track_ids = [] + releases: dict[Any, dict[str, int | Any]] = self.__get_releases_info(release_ids) + #print(releases) + print(" ") + print("Информация о релизе: \n") + from pprint import pprint + pprint(releases) + #for key, value in releases.items(): + # if key := track_ids: + # print(" ") + # else: + # print("{0}: {1}".format(key, value) + "\n") + # os.system("pause") + for i in releases.values(): + track_ids += i["track_ids"] + self.download_tracks(track_ids, releases=releases) + + +if __name__ == '__main__': + + release_ids = [] + track_ids = [] + z = zvukdown_() + + if "login" in sys.argv: + # print(sys.argv[2], sys.argv[3]) + z.save_token(sys.argv[2], sys.argv[3]) + print("Token saved!") + else: + if "debug" in sys.argv: + z.verify = False + for i in sys.argv: + # print(i) + if "release" in i: + release_ids.append(int(i.strip("https://sber-zvuk.com/release/"))) + elif "track" in i: + track_ids.append(int(i.strip("https://sber-zvuk.com/track/"))) + + z.read_token() + if release_ids: + z.download_albums(release_ids) + if track_ids: + z.download_tracks(track_ids, True) + list(map(os.remove, glob.glob("temp*.jpg")))