Как реализована выгрузка музыки из ВК?

[lgg]

Если я правильно понял, то текущее репо поддерживает выгрузку музыки? А можно этот компонент отдельно использовать вытянуть? @Bizordec, можешь, пожалуйста, подсказать - куда смотреть/копать в коде?

[Bizordec]

Выгрузка музыки реализована с помощью библиотеки vkaudiotoken-python.
Запрос в API с токеном Kate Mobile возвращает ссылку на MP3, а с токеном официального VK клиента - на M3U8 (для скачивания требуется FFmpeg).

Конкретно в данном репозитории такая реализация:

1. Сначала берутся токены:
   

   vk-to-tgm/app/config.py

   Lines 221 to 231 in 61f3175

   	values["KATE_TOKEN"] = await get_vk_token(
   	VK_LOGIN,
   	VK_PASSWORD,
   	KATE_TOKEN,
   	)
   	values["VK_OFFICIAL_TOKEN"] = await get_vk_token(
   	VK_LOGIN,
   	VK_PASSWORD,
   	VK_OFFICIAL_TOKEN,
   	is_kate=False,
   	)

   
   

   vk-to-tgm/app/config.py

   Lines 37 to 122 in 61f3175

   	async def get_vk_token(
   	login: str,
   	password: str,
   	current_token: str = "",
   	exit_on_error: bool = False,
   	is_kate: bool = True,
   	):
   	new_token: str = current_token
   	if is_kate:
   	user_agent = supported_clients.KATE.user_agent
   	get_token = get_kate_token
   	t_name = "KATE_TOKEN"
   	else:
   	user_agent = supported_clients.VK_OFFICIAL.user_agent
   	get_token = get_vk_official_token
   	t_name = "VK_OFFICIAL_TOKEN"
   	if not current_token:
   	if not login and not password:
   	logger.critical("Environment variables VK_LOGIN and VK_PASSWORD are required.")
   	sys.exit(1)
   	logger.info(f"[{t_name}] Getting new token...")
   	auth_code = "GET_CODE"
   	captcha_sid = None
   	captcha_key = None
   	while True:
   	try:
   	new_token = get_token(
   	login,
   	password,
   	auth_code=auth_code,
   	captcha_sid=captcha_sid,
   	captcha_key=captcha_key,
   	)["token"]
   	if new_token:
   	set_key(".env", t_name, new_token)
   	break
   	except TokenException as err:
   	err_code: int = err.code
   	err_extra: dict = err.extra or {}
   	if err_code == TokenException.TOKEN_NOT_RECEIVED:
   	auth_code = "GET_CODE"
   	captcha_sid = None
   	captcha_key = None
   	continue
   	if err_code == TokenException.CAPTCHA_REQ and "captcha_sid" in err_extra:
   	captcha_sid = err_extra["captcha_sid"]
   	captcha_key = input("Enter captcha key from image (" + err_extra["captcha_img"] + "): ")
   	continue
   	if err_code == TokenException.TWOFA_REQ and "validation_sid" in err_extra:
   	params = CommonParams(user_agent)
   	try:
   	TwoFAHelper(params).validate_phone(err_extra["validation_sid"])
   	logger.warning("SMS should be sent")
   	except TokenException as error:
   	error_extra: dict = error.extra or {}
   	error_obj: dict = error_extra.get("error", {})
   	error_code = error_obj.get("error_code") if error_obj else None
   	if error_code is None or error_code != 103:
   	raise
   	while True:
   	try:
   	auth_code = str(int(input(f"[{t_name}] Enter auth code: ")))
   	break
   	except ValueError:
   	logger.warning("Please enter integer code.")
   	else:
   	raise
   	vk_user = API(
   	token=new_token,
   	http_client=AiohttpClient(session=ClientSession(headers={"User-agent": user_agent})),
   	)
   	try:
   	logger.info(f"[{t_name}] Checking if token is valid...")
   	await vk_user.users.get()
   	logger.info(f"[{t_name}] Token is valid.")
   	except VKAPIError[5] as error:
   	err_msg = error.description
   	logger.warning(f"[{t_name}] Token is not valid: {err_msg}")
   	if exit_on_error:
   	sys.exit(err_msg)
   	else:
   	new_token = await get_vk_token("", login, password, exit_on_error=True, is_kate=is_kate)
   	finally:
   	await vk_user.http_client.close()
   	return new_token

2. Затем создаются клиенты:
   

   vk-to-tgm/app/handlers/wall.py

   Lines 53 to 70 in 61f3175

   	self.kate_user = API(
   	token=settings.KATE_TOKEN,
   	http_client=AiohttpClient(
   	session=ClientSession(
   	headers={"User-agent": supported_clients.KATE.user_agent},
   	),
   	),
   	)
   	self.kate_user.request_validators.append(VkLangRequestValidator())
   	self.vk_user = API(
   	token=settings.VK_OFFICIAL_TOKEN,
   	http_client=AiohttpClient(
   	session=ClientSession(
   	headers={"User-agent": supported_clients.VK_OFFICIAL.user_agent},
   	),
   	),
   	)

3. Затем из API запрашиваются MP3 ссылки на аудио:
   

   vk-to-tgm/app/handlers/wall.py

   Lines 194 to 202 in 61f3175

   	audios: List[dict] = (
   	await self.kate_user.request(
   	"audio.getById",
   	{
   	"audios": ",".join(audio_ids),
   	},
   	)
   	)["response"]
   	attachments.audios = [AudioAudio(**audio) for audio in audios if audio.get("url")]

4. И здесь уже идет скачивание аудио по 10 штук (максимум в Telegram на одно сообщение). В случае битой ссылки будет попытка скачивания по второму токену:
   

   vk-to-tgm/app/handlers/wall.py

   Lines 334 to 344 in 61f3175

   	async with Downloader() as downloader:
   	downloaded_audios = await downloader.download_audios(attachments.audios[:10], self.vk_user)
   	main_msg = (
   	await self.tgm_bot.send_file(
   	self.channel_id,
   	caption=[""] * (len(downloaded_audios) - 1) + [caption_first_html],
   	file=downloaded_audios,
   	voice_note=True,
   	reply_to=reply_to_message_id,
   	)
   	)[0]

   
   

   vk-to-tgm/app/utils/downloader.py

   Lines 74 to 122 in 61f3175

   	async def download_audio(
   	self,
   	audio: AudioAudio,
   	fallback_vk_api: Optional[API] = None,
   	try_fallback: bool = True,
   	) -> str:
   	url = audio.url
   	audio_full_id = f"{audio.owner_id}_{audio.id}_{audio.access_key}"
   	audio_full_title = f"{audio.artist} - {audio.title}"
   	logger.info(f"Downloading audio [{audio_full_id}] from URL: {url}")
   	filepath = ""
   	await asyncio.sleep(randbelow(3))
   	async with NamedTemporaryFile(suffix=".mp3", delete=False) as temp:
   	if not urlparse(url).path.endswith("m3u8"):
   	logger.info("Downloading audio by http...")
   	async with self.session.get(url, timeout=3600) as response:
   	async for chunk, _ in response.content.iter_chunks():
   	await temp.write(chunk)
   	else:
   	logger.info("Downloading audio by ffmpeg...")
   	stream = ffmpeg.input(url, http_persistent=False)
   	stream = ffmpeg.output(stream, "-", c="copy", f="mp3")
   	out, _ = ffmpeg.run(stream, capture_stdout=True, quiet=True)
   	await temp.write(out)
   	filepath = temp.name
   	# Setting audio metadata
   	audiofile = eyed3.load(filepath)
   	if audiofile:
   	audiofile.tag.artist = audio.artist
   	audiofile.tag.title = audio.title
   	audiofile.tag.save()
   	elif fallback_vk_api and try_fallback:
   	logger.warning(f"Broken audio [{audio_full_id}] [{audio_full_title}], trying with another token...")
   	audios: List[dict] = (
   	await fallback_vk_api.request(
   	"audio.getById",
   	{
   	"audios": audio_full_id,
   	},
   	)
   	)["response"]
   	fb_audio = next(iter(audios), None)
   	if fb_audio and fb_audio.get("url"):
   	audio = AudioAudio(**fb_audio)
   	filepath = await self.download_audio(audio, fallback_vk_api, try_fallback=False)
   	logger.info(f"Audio succesfully downloaded: [{audio_full_id}] [{audio_full_title}]")
   	return filepath

[lgg]

Спасибо большое за такой подробный ответ!