implemented the hearthbeat of youtube music

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
         )
 

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:

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

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