implemented the automated fetching of the api key

This commit is contained in:
Hellow
2023-07-19 21:40:23 +02:00
parent d10b4fc1a1
commit 31bce2fcb1
3 changed files with 94 additions and 18 deletions

View File

@@ -1,10 +1,14 @@
from typing import Dict, List, Optional, Set, Type
from urllib.parse import urlparse
import logging
import random
import json
from dataclasses import dataclass
import re
from music_kraken.utils.shared import PROXIES_LIST, YOUTUBE_MUSIC_LOGGER
from ..utils.exception.config import SettingValueError
from ..utils.shared import PROXIES_LIST, YOUTUBE_MUSIC_LOGGER
from ..utils.config import CONNECTION_SECTION, write_config
from ..objects import Source, DatabaseObject
from .abstract import Page
@@ -30,21 +34,22 @@ class YoutubeMusicConnection(Connection):
103.82
--> average delay in between: 1.8875 min
-->
===API=KEY===
AIzaSyC9XL3ZjWddXya6X74dJoCTL-WEYFDNX30
can be found at `view-source:https://music.youtube.com/`
search for: "innertubeApiKey"
"""
def __init__(self, logger: logging.Logger):
super().__init__(
host="https://music.youtube.com/",
logger=logger,
hearthbeat=True,
hearthbeat_interval=113.25
hearthbeat_interval=113.25,
)
# cookie consent for youtube
# https://stackoverflow.com/a/66940841/16804841
self.session.cookies.set(
name='CONSENT', value='YES+cb.20210328-17-p0.en-GB+FX+{}'.format(random.randint(100, 999)),
path='/', domain='.youtube.com'
)
self.start_hearthbeat()
def hearthbeat(self):
r = self.get("https://music.youtube.com/verify_session", is_hearthbeat=True)
if r is None:
@@ -59,6 +64,15 @@ class YoutubeMusicConnection(Connection):
self.hearthbeat_failed()
@dataclass
class YouTubeMusicCredentials:
api_key: str
# ctoken is probably short for continue-token
# It is probably not strictly necessary, but hey :))
ctoken: str
class YoutubeMusic(Page):
# CHANGE
SOURCE_TYPE = SourcePages.PRESET
@@ -66,9 +80,59 @@ class YoutubeMusic(Page):
def __init__(self, *args, **kwargs):
self.connection: YoutubeMusicConnection = YoutubeMusicConnection(logger=self.LOGGER)
self.credentials: YouTubeMusicCredentials = YouTubeMusicCredentials(
api_key=CONNECTION_SECTION.YOUTUBE_MUSIC_API_KEY.object_from_value,
ctoken=""
)
if self.credentials.api_key == "":
self._fetch_from_main_page()
super().__init__(*args, **kwargs)
def _fetch_from_main_page(self):
"""
===API=KEY===
AIzaSyC9XL3ZjWddXya6X74dJoCTL-WEYFDNX30
can be found at `view-source:https://music.youtube.com/`
search for: "innertubeApiKey"
"""
r = self.connection.get("https://music.youtube.com/")
if r is None:
return
content = r.text
# api key
api_key_pattern = (
r"(?<=\"innertubeApiKey\":\")(.*?)(?=\")",
r"(?<=\"INNERTUBE_API_KEY\":\")(.*?)(?=\")",
)
api_keys = []
for api_key_patter in api_key_pattern:
api_keys.extend(re.findall(api_key_patter, content))
found_a_good_api_key = False
for api_key in api_keys:
# save the first api key
api_key = api_keys[0]
try:
CONNECTION_SECTION.YOUTUBE_MUSIC_API_KEY.set_value(api_key)
except SettingValueError:
continue
found_a_good_api_key = True
break
if found_a_good_api_key:
write_config()
self.LOGGER.info(f"Found a valid API-KEY for {type(self).__name__}: \"{api_key}\"")
else:
self.LOGGER.error(f"Couldn't find an API-KEY for {type(self).__name__}. :((")
def get_source_type(self, source: Source) -> Optional[Type[DatabaseObject]]:
return super().get_source_type(source)