[Из песочницы] Реализация offline режима для Yandex.Music

Сегодня мы будем рассматривать такой достаточно известный музыкальный сервис, как Yandex.Music. Хороший в целом сервис, но с существенным недостатком — невозможностью работы оффлайн. Мы попробуем исправить это досадное недоразумение, используя подручные инструменты.

Итак, нам понадобится:

• Относительно свежий python: 3.7 и выше
• Всякая асинхронщина: aiohttp и aiofile
• Классический инструмент для работы с html-API: BeautifulSoup
• Для развлечения пользователя во время процесса: tqdm
• Для заполнения тэгов: mutagen

Неавторизованным пользователям сервиса доступны только отрезки песен длинной до 30 секунд. Этого явно недостаточно для качественного прослушивания. Авторизоваться мы будем самым что ни на есть естественным способом, через веб-форму с получением печенюшек. В этом нам поможет opener для выполнения запросов и HTMLParser для парсинга форм.

def resolve_cookie(login: str, password: str) -> str:
    cookies = CookieJar()
    opener = urllib.request.build_opener(
        urllib.request.HTTPCookieProcessor(cookies),
        urllib.request.HTTPRedirectHandler())
    response = opener.open("https://passport.yandex.ru")
    doc = response.read()
    parser = FormParser()
    parser.feed(doc.decode("utf-8"))
    parser.close()
    parser.params["login"] = login
    response = opener.open(parser.url or response.url, urllib.parse.urlencode(parser.params).encode("utf-8"))
    doc = response.read()
    parser = FormParser()
    parser.feed(doc.decode("utf-8"))
    parser.close()
    parser.params["login"] = login
    parser.params["passwd"] = password
    response = opener.open(parser.url or response.url, urllib.parse.urlencode(parser.params).encode("utf-8"))
    cookie_data = {}
    for item in cookies:
        if item.domain == ".yandex.ru":
            cookie_data[item.name] = item.value
    if "yandex_login" not in cookie_data:
        keys = ", ".join(cookie_data.keys())
        raise Exception(f"Invalid cookie_data {keys}")
    return "; ".join(map(lambda v: f"{v[0]}={v[1]}", cookie_data.items()))

По адресу https://passport.yandex.ru нас ожидает первая форма с полем login. Если будет указан актуальный логин, то следующая форма предложит ввести пароль, что мы незамедлительно сделаем. Результат отправки формы нас особо не интересует. Наша цель — кукисы. Если среди полученых кук окажется yandex_login, то авторизация прошла успешно. Вариант с двухфакторной аутентификацией мы пока не рассматриваем.

Следующим нашим шагом будет получение списка избранных пользователем исполнителей. Чтобы показать наш стиль и молодёжность, для запросов применим aiohttp. Полученный html распарсим с помощью BeautifulSoup. Это классический инструмент, который используют ребята, заинтересованные в получении контента с чужих веб-страниц.

class YandexMusicApi:
    host = "music.yandex.ru"
    base_url = f"https://{host}"

    def __init__(self, cookie: str):
        self.headers = Headers(self.host, cookie)

    async def _request(self, end_point: str):
        async with aiohttp.ClientSession() as session:
            url = f"{self.base_url}/{end_point}"
            async with session.request(method="GET", url=url) as response:
                return await response.read()

    async def get_favorite_artists(self, login: str) -> List[Artist]:
        body = await self._request(f"users/{login}/artists")
        soup = BeautifulSoup(body, "lxml")
        artists_soup = soup.find("div", class_="page-users__artists")
        if artists_soup is None:
            caption = soup.find("div", class_="page-users__caption")
            if caption:
                raise Exception(caption.contents[0])
        result = []
        for artist_soup in artists_soup.find_all("div", class_="artist"):
            title_soup = artist_soup.find("div", class_="artist__name")
            title = title_soup.attrs["title"]
            title_href_soup = title_soup.find("a")
            id_ = int(title_href_soup.attrs["href"].split("/")[-1])
            result.append(Artist(id_, title))
        return result

Здесь всё довольно просто, на странице https://music.yandex.ru/users/<login>/artists в блоке page-users__artists отображается требуемый нам список. Имя исполнителя можно обнаружить в аттрибуте title блока artist__name. Id исполнителя лёгким движением split извлекается из адреса ссылки.
Аналогичные действия позволят нам получить список альбомов исполнителя, а также список треков в альбоме.

В этом месте мы уже потираем наши шаловливые ручки в ожидании прослушивания своей музыкальной коллекции. Но если бы всё было так просто, то я бы не писал эту статью. Нас ожидает новый противник — yandex-хранилище. Для проникновения в хранилище потребуется воспроизвести жонглирование ссылками, которое происходит на страницах музыкального ресурса. При внимательном изучении вкладки Network браузерного инструментария выяснилось, что все элементы запроса легко выводятся за исключением одного участка в адресе https://{host}/get-mp3/{sign}/{ts}/{path}, который после непродолжительного гугления получил имя sign. Соль для сигнатуры (XGRlBW9FXlekgbPrRHuSiA) тоже была обнаружена на просторах интернета. Скорее всего не составит большого труда извлечь это значение из исходников страницы, но это уже не требуется.

    async def get_track_url(self, album_id: int, track_id: int) -> str:
        async with aiohttp.ClientSession() as session:
            url = f"{self.base_url}/api/v2.1/handlers/track/{track_id}:{album_id}/" \
                  f"web-album-track-track-main/download/m?hq=0&external-domain={self.host}&overembed=no&__t={timestamp()}"
            page = f"album/{album_id}"
            headers = self.headers.build(page)
            async with session.request(method="GET", url=url, headers=headers) as response:
                body = await response.json()
                src = body["src"]
                src += f"&format=json&external-domain={self.host}&overembed=no&__t={timestamp()}"
                result = parse.urlparse(src)
                headers = self.headers.build(page, {
                    ":authority": "storage.mds.yandex.net",
                    ":method": "GET",
                    ":path": f"{result.path}/{result.query}",
                    ":scheme": "https",
                }, True)
                async with session.request(method="GET", url=src, headers=headers) as response:
                    body = await response.json()
                    host = body["host"]
                    path = body["path"]
                    s = body["s"]
                    ts = body["ts"]
                    sign = md5(f"XGRlBW9FXlekgbPrRHuSiA{path[1::]}{s}".encode("utf-8")).hexdigest()
                    url = f"https://{host}/get-mp3/{sign}/{ts}/{path}"
                    return url

Время жизни ссылки ограничено, поэтому нужно незамедлительно перейти к скачиванию.

На этом этапе уже немного надоедает всё это дело, поэтому берём в одну руку ранее используемое нами aiohttp, в другую — aiofile и реализуем скачивание в лоб.

    async def download_file(cls, url: str, filename: str):
        async with aiohttp.ClientSession() as session:
            async with session.request(method="GET", url=url) as response:
                data = await response.read()
                async with AIOFile(filename, "wb") as afp:
                    await afp.write(data)
                    await afp.fsync()

Опытные программисты могут предложить реализовать скачивание с использованием буфера и записью во временный файл. Но в данном случае, учитывая, что максимальный размер mp3, как правило, укладывается в оперативную память, то при достаточно быстром интернете можно этим не заморачиваться. Ладно, не надо кидать в меня подручными материалами, я всё понял. На самом деле это одно из ключевых мест нашего проекта, которое в следующей итерации требуется привести к адекватному виду. В идеале ещё стоит добавить повторные запросы при возникновении сетевых ошибок.

Далее, нам нужно поставить закачку на поток. Методично перебираем исполнителей, альбомы, треки. Создаём директории, скачиваем альбомный арт, скачиваем сами треки. Прописываем тэги (почему-то у полученных mp3 они отсутствовали).

    async def download_artist(self, artist: Artist, depth: Depth = Depth.NORMAL):
        artist_progress = tqdm(total=0, desc=artist.title, position=1, ascii=True)
        albums = await self.api.get_artist_albums(artist.id)
        artist_progress.total = len(albums)
        artist_progress.refresh()
        for album in albums:
            album_dir = os.path.join(self.target_dir, normalize(artist.title), f"{album.year} - {normalize(album.title)}")
            if depth < Depth.ALBUMS and os.path.exists(album_dir):
                artist_progress.update()
                continue
            album_progress = tqdm(total=0, desc=f"> {album.title}", position=0, ascii=True)
            tracks = await self.api.get_album_tracks(album.id)
            album_progress.total = len(tracks)
            album_progress.refresh()
            os.makedirs(album_dir, exist_ok=True)
            if album.cover:
                album_progress.total += 1
                cover_filename = os.path.join(album_dir, "cover.jpg")
                if not os.path.exists(cover_filename):
                    await self.download_file(album.cover, cover_filename)
                album_progress.update()
            for track in tracks:
                target_filename = os.path.join(album_dir, f"{track.num:02d}. {normalize(track.title)}.mp3")
                if depth >= Depth.TRACKS or not os.path.exists(target_filename):
                    url = await self.api.get_track_url(track.album_id, track.id)
                    await self.download_file(url, target_filename)
                    self.write_tags(target_filename, {
                        "title": track.title,
                        "tracknumber": str(track.num),
                        "artist": artist.title,
                        "album": album.title,
                        "date": str(album.year),
                    })
                album_progress.update()
            album_progress.close()
            artist_progress.update()
        artist_progress.close()

После первой загрузки, я с удивлением обнаружил, что любимая мною группа AC/DC оказалась разделена на две директории. Это послужило поводом к реализации метода normalize:

def normalize(name: str) -> str:
    return name.replace("/", "-")

Как несложно заметить, процесс скачивания можно (точнее нужно) распараллелить. Почва для этого подготовлена благодаря использованию асинхронных библиотек для сетевых запросов и записи в файлы. Для этого можно использовать asyncio.Semaphore в комбинации с asyncio.gather.
Но реализация асинхронной очереди не является предметом данной статьи, поэтому пока и так сойдёт.

Теперь, когда у нас всё уже практически готово, нужно подумать о том, как это запускать. Явки и пароли мы будем вычитывать из файла .credentials, ненавязчиво бросая эксепшен в случае его отсутствия. Полученную куку бережно сохраним в файл .cookie для дальнейшего использования.

def resolve_cookie() -> str:
    base_dir = os.path.normpath(os.path.join(os.path.dirname(__file__), os.path.pardir))
    cookie_file = os.path.join(base_dir, ".cookie")
    if os.path.exists(cookie_file):
        with open(cookie_file, "rt") as file:
            return file.read()
    credentials_file = os.path.join(base_dir, ".credentials")
    if os.path.exists(credentials_file):
        config = configparser.ConfigParser()
        config.read(credentials_file)
        login = config["yandex"]["login"]
        password = config["yandex"]["password"]
    else:
        raise Exception(f"""Create \"{credentials_file}\" with content

[yandex]
login=<user_login>
password=<user_password>
""")
    cookie = auth.resolve_cookie(login, password)
    with open(cookie_file, "wt") as file:
        file.write(cookie)
    return cookie

В этом месте стоит немного украсить жизнь пользователя, позволив ему выбрать директорию для хранения музыкальной коллекции, а так же возможность скачивания треков отдельного исполнителя. Для разбора аргументов командной строки используем стандартный инструмент argparse, который отличается простотой конфигурации.
Опишем немного опциональных параметров:

• -a (--artist), Id исполнителя, если указан, то скачиваются только треки этого исполнителя
• -o (--output), директория для хранения музыкальной коллекции, по умолчанию — Music в домашней директории пользователя.
• -d (--depth), параметр родился как костыль, вызванный возможным прерыванием процесса
  
  • При значении по-умолчанию 0 (NORMAL) проверяется наличие директории с альбомом, и, если она существует, то загрузка альбома пропускается
  • Значение 1 (ALBUMS) перебирает все треки в альбоме и скачивает недостающие
  • Значение 2 (TRACKS) скачивает и перезаписывает треки, даже если они уже присутствуют в файловой системе

async def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("-a", "--artist", help="Artist ID")
    parser.add_argument("-o", "--output", default=f"{Path.home()}/Music",
                        help=f"Output directory, default {Path.home()}/Music")
    parser.add_argument("-d", "--depth", default=0, type=int,
                        help=f"Exists files check depth, {enum_print(Depth)}")
    args = parser.parse_args()
    cookie = resolve_cookie()
    api = YandexMusicApi(cookie)
    agent = YandexMusicAgent(api, args.output)
    if args.artist:
        artist = await api.get_artist(args.artist)
        await agent.download_artist(artist, args.depth)
    else:
        email = re.compile(".*?yandex_login=(.*?);.*?", re.M).match(cookie).group(1)
        await agent.download_favorites(email, args.depth)

И вот, наконец-то, мы можем всё это запустить:

if __name__ == "__main__":
    asyncio.run(main())

Спасибо за внимание. Теперь вы знаете, как реализовать API которого нет, скачать нескачиваемое и стать счастливым обладателем собственной музыкальной коллекции.

Результат можно увидеть в репозитории yandex.music.agent

Let's block ads! (Why?)