feat: use device to get m3u8

src/adb.py

@@ -42,6 +42,7 @@ class Device:
     suMethod: str
     decryptLock: asyncio.Lock
     hyperDecryptDevices: list[HyperDecryptDevice] = []
+    m3u8Script: frida.core.Script
 
     def __init__(self, host="127.0.0.1", port=5037, su_method: str = "su -c"):
         self.client = AdbClient(host, port)
@@ -49,6 +50,18 @@ class Device:
         self.host = host
         self.decryptLock = asyncio.Lock()
 
+    async def get_m3u8(self, adam_id: str):
+        try:
+            result: str = await self.m3u8Script.exports_async.getm3u8(adam_id)
+        except frida.core.RPCException:
+            # The script takes 8 seconds to start.
+            # If the script does not start when the function is called, wait 8 seconds and call again.
+            await asyncio.sleep(8)
+            result: str = await self.m3u8Script.exports_async.getm3u8(adam_id)
+        if result.isdigit():
+            return None
+        return result
+
     def connect(self, host: str, port: int):
         try:
             status = self.client.remote_connect(host, port)
@@ -96,6 +109,10 @@ class Device:
             self.fridaDevice = frida.get_device_manager().get_device(self.device.serial)
         self.pid = self.fridaDevice.spawn("com.apple.android.music")
         self.fridaSession = self.fridaDevice.attach(self.pid)
+        with open("m3u8.js", "r") as f:
+            m3u8_script = f.read()
+        self.m3u8Script = self.fridaSession.create_script(m3u8_script)
+        self.m3u8Script.load()
         script: frida.core.Script = self.fridaSession.create_script(agent)
         script.load()
         self.fridaDevice.resume(self.pid)
@@ -172,6 +189,10 @@ class Device:
             self.fridaDevice = frida.get_device_manager().get_device(self.device.serial)
         self.pid = self.fridaDevice.spawn("com.apple.android.music")
         self.fridaSession = self.fridaDevice.attach(self.pid)
+        with open("m3u8.js", "r") as f:
+            m3u8_script = f.read()
+        self.m3u8Script = self.fridaSession.create_script(m3u8_script)
+        self.m3u8Script.load()
         for port in ports:
             self._start_forward(port, port)
             with open("agent.js", "r") as f:

src/rip.py

@@ -69,6 +69,11 @@ async def rip_song(song: Song, auth_params: GlobalAuthParams, codec: str, config
         if not specified_m3u8 and not song_data.attributes.extendedAssetUrls.enhancedHls:
             logger.error(f"Failed to download song: {song_metadata.artist} - {song_metadata.title}. Lossless audio does not exist")
             return
+        if not specified_m3u8:
+            device_m3u8 = await device.get_m3u8(song.id)
+            if device_m3u8:
+                specified_m3u8 = device_m3u8
+                logger.info(f"Use m3u8 from device for song: {song_metadata.artist} - {song_metadata.title}")
         if specified_m3u8:
             song_uri, keys, codec_id = await extract_media(specified_m3u8, codec, song_metadata,
                                                            config.download.codecPriority,