(aid, output_path="output")
| 58 | |
| 59 | |
| 60 | def save_song(aid, output_path="output"): |
| 61 | start_time = time.time() |
| 62 | while True: |
| 63 | audio_url, metadata = get_info(aid) |
| 64 | if audio_url: |
| 65 | break |
| 66 | elif time.time() - start_time > 90: |
| 67 | raise TimeoutError("Failed to get audio_url within 90 seconds") |
| 68 | time.sleep(30) |
| 69 | response = rget(audio_url, allow_redirects=False, stream=True) |
| 70 | if response.status_code != 200: |
| 71 | raise Exception("Could not download song") |
| 72 | index = 0 |
| 73 | while os.path.exists(os.path.join(output_path, f"suno_{index}.mp3")): |
| 74 | index += 1 |
| 75 | path = os.path.join(output_path, f"suno_{index}.mp3") |
| 76 | with open(path, "wb") as output_file: |
| 77 | for chunk in response.iter_content(chunk_size=1024): |
| 78 | # If the chunk is not empty, write it to the file. |
| 79 | if chunk: |
| 80 | output_file.write(chunk) |
nothing calls this directly
no test coverage detected