Update app.py

app.py

@@ -4,41 +4,99 @@ from pydub import AudioSegment
 import shazamio
 import os
 import time
-token = os.environ.get('TOKEN')
+import tempfile
+from pathlib import Path
+import sys
+
 # Replace with your actual client ID and client secret
 client_id = os.environ.get('SOUNDCLOUD_CLIENT_ID')
 client_secret = os.environ.get('SOUNDCLOUD_CLIENT_SECRET')
+token = os.environ.get('TOKEN')
 
-def download_audio(streaming_url, output_path, headers):
-    response = requests.get(streaming_url, headers=headers, stream=True)
-    with open(output_path, 'wb') as f:
-        for chunk in response.iter_content(chunk_size=8192):
-            f.write(chunk)
-
-async def identify_track(shazam, audio_chunk):
-    temp_file = 'temp_chunk.wav'
-    audio_chunk.export(temp_file, format='wav')
-    result = await shazam.recognize_file(temp_file)
-    if result and 'track' in result:
-        track_data = result['track']
-        return {
-            'title': track_data['title'],
-            'subtitle': track_data['subtitle']
+def get_soundcloud_access_token(client_id, client_secret):
+    try:
+        auth_string = f'{client_id}:{client_secret}'
+        auth_headers = {
+            'Authorization': 'Basic ' + base64.b64encode(auth_string.encode()).decode()
+        }
+        data = {
+            'grant_type': 'client_credentials'
         }
-    else:
-        return None
+        response = requests.post('https://api.soundcloud.com/oauth2/token', headers=auth_headers, data=data)
+        if response.status_code == 200:
+            token_data = response.json()
+            return token_data['access_token']
+        else:
+            raise Exception(f"Failed to obtain access token: {response.text}")
+    except Exception as e:
+        return f"Error obtaining access token: {str(e)}"
+
+def download_audio(streaming_url, output_path, headers):
+    try:
+        response = requests.get(streaming_url, headers=headers, stream=True)
+        with open(output_path, 'wb') as f:
+            for chunk in response.iter_content(chunk_size=8192):
+                f.write(chunk)
+    except Exception as e:
+        raise Exception(f"Error downloading audio: {str(e)}")
+
+async def identify_track(shazam, audio_chunk, temp_dir):
+    try:
+        temp_file = os.path.join(temp_dir, 'temp_chunk.wav')
+        audio_chunk.export(temp_file, format='wav')
+        result = await shazam.recognize_file(temp_file)
+        if result and 'track' in result:
+            track_data = result['track']
+            return {
+                'title': track_data['title'],
+                'subtitle': track_data['subtitle']
+            }
+        else:
+            return None
+    except Exception as e:
+        return f"Error identifying track: {str(e)}"
 
 async def process_dj_set(track_url, progress=gr.Progress()):
+    temp_dir = tempfile.mkdtemp()
+    output_path = os.path.join(temp_dir, 'track.wav')
+    tracklist_path = os.path.join(temp_dir, 'tracklist.txt')
+    status_messages = []
+
+    # Get access token
+    access_token = get_soundcloud_access_token(client_id, client_secret)
+    if isinstance(access_token, str) and access_token.startswith("Error"):
+        return "", "", access_token
+
     headers = {
-        'Authorization': 'Bearer ' + f'{token}'
+        'Authorization': f'Bearer {access_token}'
     }
-    resolve_response = requests.get(f'https://api.soundcloud.com/resolve.json?url={track_url}', headers=headers)
-    if resolve_response.status_code != 200:
-        return "Failed to resolve track URL.", ""
-    track_data = resolve_response.json()
-    streaming_url = track_data['stream_url']
-    download_audio(streaming_url, 'track.wav', headers)
-    audio = AudioSegment.from_wav('track.wav')
+
+    # Resolve track URL
+    try:
+        resolve_response = requests.get(f'https://api.soundcloud.com/resolve.json?url={track_url}', headers=headers)
+        if resolve_response.status_code != 200:
+            return "", "", f"Failed to resolve track URL: {resolve_response.text}"
+        track_data = resolve_response.json()
+        streaming_url = track_data['stream_url']
+        status_messages.append("Track URL resolved successfully.")
+    except Exception as e:
+        return "", "", f"Error resolving track URL: {str(e)}"
+
+    # Download audio
+    try:
+        download_audio(streaming_url, output_path, headers)
+        status_messages.append("Audio downloaded successfully.")
+    except Exception as e:
+        return "", "", f"Error downloading audio: {str(e)}"
+
+    # Load audio
+    try:
+        audio = AudioSegment.from_wav(output_path)
+        status_messages.append("Audio loaded successfully.")
+    except Exception as e:
+        return "", "", f"Error loading audio: {str(e)}"
+
+    # Split audio into chunks
     chunk_duration = 30000  # 30 seconds
     overlap = 10000  # 10 seconds
     chunks = []
@@ -48,17 +106,32 @@ async def process_dj_set(track_url, progress=gr.Progress()):
         chunk = audio[start:end]
         chunks.append((start, chunk))
         start += chunk_duration - overlap
+
+    # Initialize Shazam
     shazam = shazamio.Shazam()
+
+    # Identify tracks
     tracklist = []
     for start_time, chunk in chunks:
         progress(0.1)
-        track_info = await identify_track(shazam, chunk)
-        if track_info:
+        track_info = await identify_track(shazam, chunk, temp_dir)
+        if isinstance(track_info, dict):
             timestamp = time.strftime("%M:%S", time.gmtime(start_time / 1000))
             tracklist.append(f"{timestamp} - {track_info['title']} by {track_info['subtitle']}")
+        elif isinstance(track_info, str):
+            status_messages.append(track_info)
+
+    # Prepare tracklist output
     tracklist_output = "\n".join(tracklist)
-    download_content = tracklist_output
-    return tracklist_output, download_content
+    with open(tracklist_path, 'w') as f:
+        f.write(tracklist_output)
+
+    # Prepare status message
+    status_message = "\n".join(status_messages)
+    if not tracklist:
+        status_message += "\nNo tracks identified."
+
+    return tracklist_output, tracklist_path, status_message
 
 css = """
 #col-container {
@@ -70,7 +143,7 @@ css = """
 with gr.Blocks(css=css) as demo:
     with gr.Column(elem_id="col-container"):
         gr.Markdown("# SoundCloud DJ Set Track Identifier")
-
+        status_text = gr.Markdown(elem_id="status_text")
         with gr.Row():
             track_url = gr.Text(
                 label="SoundCloud DJ Set URL",
@@ -79,17 +152,13 @@ with gr.Blocks(css=css) as demo:
                 placeholder="Enter SoundCloud DJ set URL",
                 container=False,
             )
-
             run_button = gr.Button("Process", scale=0, variant="primary")
-
         result = gr.Textbox(label="Tracklist", show_label=False)
-
         with gr.Accordion("Download Tracklist", open=False):
             download_button = gr.File(label="Download")
-
         gr.Examples(examples=["https://soundcloud.com/your-track-url"], inputs=[track_url])
 
-    run_button.click(process_dj_set, inputs=track_url, outputs=[result, download_button])
+    run_button.click(process_dj_set, inputs=track_url, outputs=[result, download_button, status_text])
 
 if __name__ == "__main__":
     demo.launch()