diff --git a/src/music_kraken/connection/connection.py b/src/music_kraken/connection/connection.py index e1166c2..d1e89c5 100644 --- a/src/music_kraken/connection/connection.py +++ b/src/music_kraken/connection/connection.py @@ -3,6 +3,7 @@ from typing import List, Dict, Callable, Optional, Set from urllib.parse import urlparse, urlunsplit, ParseResult import logging +import threading import requests from tqdm import tqdm @@ -23,7 +24,9 @@ class Connection: header_values: Dict[str, str] = None, accepted_response_codes: Set[int] = None, semantic_not_found: bool = True, - sleep_after_404: float = 0.0 + sleep_after_404: float = 0.0, + hearthbeat: bool = False, + hearthbeat_interval = 0 ): if proxies is None: proxies = PROXIES_LIST @@ -46,6 +49,36 @@ class Connection: self.session.headers = self.get_header(**self.HEADER_VALUES) self.session.proxies = self.rotating_proxy.current_proxy + self.session_is_occupied: bool = False + + self.hearthbeat_thread = None + if hearthbeat: + self.hearthbeat_thread = threading.Thread(target=self._hearthbeat_loop, args=(hearthbeat_interval, ), daemon=True) + self.hearthbeat_thread.start() + + + def hearthbeat_failed(self): + self.LOGGER.warning(f"I just died... (The hearthbeat failed)") + + + def hearthbeat(self): + # Your code to send heartbeat requests goes here + print("the hearth is beating, but it needs to be implemented ;-;\nFuck youuuu for setting hearthbeat in the constructor to true, but not implementing the method Connection.hearbeat()") + + def _hearthbeat_loop(self, interval: float): + def hearthbeat_wrapper(): + self.session_is_occupied = True + self.LOGGER.info(f"I am living. (sending a hearthbeat)") + self.hearthbeat() + self.LOGGER.debug(f"finished the hearthbeat") + self.session_is_occupied = False + + while True: + hearthbeat_wrapper() + time.sleep(interval) + + + def base_url(self, url: ParseResult = None): if url is None: url = self.HOST @@ -89,6 +122,7 @@ class Connection: refer_from_origin: bool = True, raw_url: bool = False, sleep_after_404: float = None, + is_hearthbeat: bool = False, **kwargs ) -> Optional[requests.Response]: if sleep_after_404 is None: @@ -111,6 +145,10 @@ class Connection: connection_failed = False try: + while self.session_is_occupied and not is_hearthbeat: + if self.session_is_occupied and not is_hearthbeat: + self.LOGGER.info(f"Waiting for the hearthbeat to finish.") + r: requests.Response = request(request_url, timeout=timeout, headers=headers, **kwargs) if r.status_code in accepted_response_codes: @@ -145,6 +183,7 @@ class Connection: timeout=timeout, headers=headers, sleep_after_404=sleep_after_404, + is_hearthbeat=is_hearthbeat, **kwargs ) diff --git a/src/music_kraken/download/page_attributes.py b/src/music_kraken/download/page_attributes.py index 9b58087..60e85f7 100644 --- a/src/music_kraken/download/page_attributes.py +++ b/src/music_kraken/download/page_attributes.py @@ -5,23 +5,31 @@ from ..objects import DatabaseObject, Source from ..utils.enums.source import SourcePages from ..utils.support_classes import Query, DownloadResult from ..utils.exception.download import UrlNotFoundException -from ..pages import Page, EncyclopaediaMetallum, Musify, YouTube, INDEPENDENT_DB_OBJECTS +from ..pages import Page, EncyclopaediaMetallum, Musify, YouTube, YoutubeMusic, INDEPENDENT_DB_OBJECTS + +DEBUGGING_PAGE = YoutubeMusic ALL_PAGES: Set[Type[Page]] = { EncyclopaediaMetallum, Musify, YouTube, + YoutubeMusic } AUDIO_PAGES: Set[Type[Page]] = { Musify, YouTube, + YoutubeMusic } SHADY_PAGES: Set[Type[Page]] = { Musify, } +if DEBUGGING_PAGE is not None: + print(f"The DEBUGGING_PAGE is not None, but {DEBUGGING_PAGE}. Only using this page") + ALL_PAGES = {DEBUGGING_PAGE} + AUDIO_PAGES = ALL_PAGES.union(AUDIO_PAGES) class Pages: diff --git a/src/music_kraken/pages/__init__.py b/src/music_kraken/pages/__init__.py index b562686..a317147 100644 --- a/src/music_kraken/pages/__init__.py +++ b/src/music_kraken/pages/__init__.py @@ -1,5 +1,6 @@ from .encyclopaedia_metallum import EncyclopaediaMetallum from .musify import Musify from .youtube import YouTube +from .youtube_music import YoutubeMusic from .abstract import Page, INDEPENDENT_DB_OBJECTS diff --git a/src/music_kraken/pages/youtube_music.py b/src/music_kraken/pages/youtube_music.py new file mode 100644 index 0000000..3458efc --- /dev/null +++ b/src/music_kraken/pages/youtube_music.py @@ -0,0 +1,98 @@ +from typing import Dict, List, Optional, Set, Type +from urllib.parse import urlparse +import logging +import json + +from music_kraken.utils.shared import PROXIES_LIST, YOUTUBE_LOGGER + + +from ..objects import Source, DatabaseObject +from .abstract import Page +from ..objects import ( + Artist, + Source, + SourcePages, + Song, + Album, + Label, + Target +) +from ..connection import Connection +from ..utils.support_classes import DownloadResult + +class YoutubeMusicConnection(Connection): + """ + ===Hearthbeat=timings=for=YOUTUBEMUSIC=== + 96.27 + 98.16 + 100.04 + 101.93 + 103.82 + + --> average delay in between: 1.8875 min + --> + """ + def __init__(self, logger: logging.Logger): + super().__init__( + host="https://music.youtube.com/", + logger=logger, + hearthbeat=True, + hearthbeat_interval=113.25 + ) + + def hearthbeat(self): + r = self.get("https://music.youtube.com/verify_session", is_hearthbeat=True) + if r is None: + self.hearthbeat_failed() + + string = r.content.decode("utf-8") + + data = json.loads(string[string.index("{"):]) + success: bool = data["success"] + + if not success: + self.hearthbeat_failed() + + +class YoutubeMusic(Page): + # CHANGE + SOURCE_TYPE = SourcePages.PRESET + LOGGER = YOUTUBE_LOGGER + + def __init__(self, *args, **kwargs): + self.connection: YoutubeMusicConnection = YoutubeMusicConnection(logger=self.LOGGER) + + super().__init__(*args, **kwargs) + + def get_source_type(self, source: Source) -> Optional[Type[DatabaseObject]]: + return super().get_source_type(source) + + def general_search(self, search_query: str) -> List[DatabaseObject]: + return [] + + def label_search(self, label: Label) -> List[Label]: + return [] + + def artist_search(self, artist: Artist) -> List[Artist]: + return [] + + def album_search(self, album: Album) -> List[Album]: + return [] + + def song_search(self, song: Song) -> List[Song]: + return [] + + def fetch_song(self, source: Source, stop_at_level: int = 1) -> Song: + return Song() + + def fetch_album(self, source: Source, stop_at_level: int = 1) -> Album: + return Album() + + def fetch_artist(self, source: Source, stop_at_level: int = 1) -> Artist: + return Artist() + + def fetch_label(self, source: Source, stop_at_level: int = 1) -> Label: + return Label() + + def download_song_to_target(self, source: Source, target: Target, desc: str = None) -> DownloadResult: + return DownloadResult()