feat(music_new): new interfaces for music, platform and API

app/music_new/music.py

@@ -0,0 +1,154 @@
+import asyncio
+import math
+import time
+from abc import abstractmethod as abstract
+from asyncio.locks import Lock
+from typing import Optional, Type
+
+
+class MusicPiece:
+    """The abstract class for music in a single platform.
+    If one day we need music playable from multiple platforms, then build a CompositeMusic class :)
+
+    Most properties are essentially getters thus we can perform lazy loading.
+    Online resources that can expire (e.g. playable media links) should be re-obtained when necessary.
+
+    Fetching multiple properties may lead to race condition. Avoid them while coding with a lock.
+    """
+
+    def __init__(self, platform: Type['Platform'], requestors: list['PropertyRequestor']):
+        self.platform = platform
+        self.requestors: dict[str, 'PropertyRequestor'] = {}
+        self.endtime_ms: Optional[int] = None
+
+        for r in requestors:
+            r.bind(self)
+            self.requestors[r.name] = r
+
+    @property
+    @abstract
+    async def name(self) -> str:
+        pass
+
+    @property
+    @abstract
+    async def artists(self) -> list[str]:
+        pass
+
+    @property
+    @abstract
+    async def playable(self) -> bool:
+        pass
+
+    @property
+    @abstract
+    async def media_url(self) -> str:
+        pass
+
+    @property
+    @abstract
+    async def duration_ms(self) -> int:
+        pass
+
+    @property
+    @abstract
+    async def cover_url(self) -> str:
+        pass
+
+    def __repr__(self):
+        return f'<{self.__class__.__name__} with attributes {self.__dict__}>'
+
+    class PropertyRequestor:
+        def __init__(self, name: str = None, expiration_time_sec: float = math.inf):
+            self.name = name if name is not None else self.__class__.__name__
+            self.expiration_time_sec = expiration_time_sec
+            self.owner: Optional[MusicPiece] = None
+            self.lock = Lock()
+            self.cache = None
+            self.__counter = 0  # for profiling
+            self.lastrun_sec = -math.inf
+
+        def bind(self, music: 'MusicPiece'):
+            self.owner = music
+
+        @abstract
+        async def __invoke(self) -> any:
+            pass
+
+        async def __call__(self) -> any:
+            async with self.lock:
+                if time.time() - self.lastrun_sec > self.expiration_time_sec or self.__counter == 0:
+                    self.__counter += 1
+                    self.lastrun_sec = time.time()
+                    self.result = await self.invoke()
+            return self.result
+
+        def __repr__(self):
+            return f'<{self.__class__.__name__} with attributes {self.__dict__}>'
+
+
+PropertyRequestor = MusicPiece.PropertyRequestor
+
+
+class Platform:
+    """The abstract class for a music platform.
+
+    Some platforms may have additional feature (e.g. radio of netease).
+    What we should do is create another platform with different implementation.
+    Methods not implemented should be discovered by checking the @abstract annotation.
+    """
+
+    def __init__(self, name: str, *alias: str):
+        self.name = name
+        self.alias = [*alias]
+
+    def names(self):
+        return {self.name, *self.alias}
+
+    # ============================================= Music =============================================
+
+    @abstract
+    async def search_music(self, keywords: str, limit: int) -> list[MusicPiece]:
+        """Search for at most `limit` musics, but do not get all information at once.
+        It can return an empty list."""
+        pass
+
+    @staticmethod
+    @abstract
+    def is_music_url(text: str):
+        pass
+
+    @staticmethod
+    @abstract
+    def is_music_id(text: str):
+        pass
+
+    async def play_by_keywords(self, keywords: str, limit: int) -> Optional[MusicPiece]:
+        result = await self.search_music(keywords, limit)
+        return result[0] if result else None
+
+    @abstract
+    async def play_by_url(self, url: str) -> Optional[MusicPiece]:
+        """A music can have multiple urls, which potentially come from phone app, PC client and webpage."""
+        pass
+
+    @abstract
+    async def play_by_id(self, music_id: int):
+        pass
+
+    # ============================================= Album =============================================
+
+    @abstract
+    async def import_album_by_url(self, url: str, range_from: int, range_to_inclusive: int) -> Optional[MusicPiece]:
+        """An album can have multiple urls, which potentially come from phone app, PC client and webpage."""
+        pass
+
+    # ============================================= Playlist =============================================
+
+    @abstract
+    async def import_playlist_by_url(self, url: str) -> Optional[MusicPiece]:
+        """A playlist can have multiple urls, which potentially come from phone app, PC client and webpage."""
+        pass
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}({repr(self.name)}, {', '.join(map(repr, self.alias))})"

app/music_new/netease/netease_music.py

@@ -0,0 +1,87 @@
+import json
+from typing import Optional
+
+import aiohttp
+
+from app.music_new.music import MusicPiece, Platform, PropertyRequestor
+
+
+class NeteaseMusicPlatform(Platform):
+    def __init__(self):
+        super(NeteaseMusicPlatform, self).__init__('netease', 'netease-music', '网易', '网易云', '网易云音乐')
+
+
+class SourceRequestor(PropertyRequestor):
+    def __init__(self, song_id: int):
+        super(SourceRequestor, self).__init__(expiration_time_sec=5 * 60)
+        self.song_id = song_id
+
+    async def invoke(self, *args, **kwargs):
+        async with aiohttp.ClientSession() as sess:
+            async with sess.post('https://music.163.com/api/song/enhance/player/url/', params={
+                'br': '3200000',
+                'Aids': f'[{self.song_id}]'
+            }) as resp:
+                text = await resp.text()
+                data = json.loads(text)
+                return data
+
+
+class DetailRequestor(PropertyRequestor):
+    def __init__(self, song_id: int):
+        super(DetailRequestor, self).__init__()
+        self.song_id = song_id
+
+    async def __invoke(self):
+        async with aiohttp.ClientSession() as sess:
+            async with sess.post('https://music.163.com/api/song/detail/', params={
+                'ids': f'[{self.song_id}]'
+            }) as resp:
+                text = await resp.text()
+                data = json.loads(text)
+                return data
+
+
+class NeteaseMusic(MusicPiece):
+    def __init__(self, song_id):
+        super().__init__(NeteaseMusicPlatform, [
+            SourceRequestor(song_id),
+            DetailRequestor(song_id)
+        ])
+        self.song_id = song_id
+        self.__name: Optional[str] = None
+        self.__artists: Optional[list[str]] = None
+        self.__playable: Optional[bool] = None
+        self.__duration_ms: Optional[int] = None
+        self.__cover_url: Optional[str] = None
+
+    @property
+    async def name(self) -> str:
+        if self.__name is None:
+            data = await self.requestors['DetailRequestor']()
+            self.__name = data.get('songs', [{}])[0].get('name', 'N/A')
+        return self.__name
+
+    @property
+    async def artists(self) -> list[str]:
+        if self.artists is None:
+            data = await self.requestors['DetailRequestor']()
+            artists = data.get('songs', [{}])[0].get('artists', [])
+            self.__artists = [artists.get('name', 'Unknown') for artist in artists]
+        return self.__artists
+
+    @property
+    async def media_url(self) -> str:
+        return await self.requestors['SourceRequestor']()
+
+
+async def main():
+    music = NeteaseMusic(36226134)
+    print(repr(music))
+    print(await music.media_url)
+    print(await asyncio.gather(music.name, music.artists))
+
+
+if __name__ == '__main__':
+    import asyncio
+    asyncio.run(main())