zvukdown.py

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")))