Update app.py

app.py

@@ -4,36 +4,48 @@ from pydub import AudioSegment
 import shazamio
 import os
 import time
+import tempfile
+from pathlib import Path
+import sys
+import asyncio
 
-# Environment variables
-token = os.environ.get('TOKEN')
+# 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 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'
+        }
+        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)}"
 
-# Function to download audio
 def download_audio(streaming_url, output_path, headers):
     try:
         response = requests.get(streaming_url, headers=headers, stream=True)
-        response.raise_for_status()
         with open(output_path, 'wb') as f:
             for chunk in response.iter_content(chunk_size=8192):
                 f.write(chunk)
-        return "Audio downloaded successfully."
     except Exception as e:
-        return f"Error downloading audio: {e}"
+        raise Exception(f"Error downloading audio: {str(e)}")
 
-# Function to identify track using Shazam
-async def identify_track(shazam, audio_chunk):
+async def identify_track(shazam, audio_chunk, temp_dir):
     try:
-        temp_file = 'temp_chunk.wav'
+        temp_file = os.path.join(temp_dir, 'temp_chunk.wav')
         audio_chunk.export(temp_file, format='wav')
-        
-        # Ensure temp_file exists and is not a directory
-        if not os.path.isfile(temp_file):
-            raise FileNotFoundError(f"Temporary file '{temp_file}' does not exist or is not a valid file.")
-        
         result = await shazam.recognize_file(temp_file)
-        os.remove(temp_file)  # Clean up temporary file
         if result and 'track' in result:
             track_data = result['track']
             return {
@@ -43,70 +55,92 @@ async def identify_track(shazam, audio_chunk):
         else:
             return None
     except Exception as e:
-        return f"Error identifying track: {e}"
+        return f"Error identifying track: {str(e)}"
 
-# Main processing function
 async def process_dj_set(track_url, progress=gr.Progress()):
+    temp_dir = "/tmp"
+    output_path = os.path.join(temp_dir, 'track.wav')
+    tracklist_path = os.path.join(temp_dir, 'tracklist.txt')
+    status_messages = []
+
+    # Check environment variables
+    if not client_id or not client_secret or not token:
+        return "", "", "Environment variables not set."
+
+    # 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': f'Bearer {access_token}'
+    }
+
+    # Resolve track URL
     try:
-        status = "Resolving SoundCloud URL..."
-        headers = {'Authorization': f'Bearer {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.", ""
-        
+            return "", "", f"Failed to resolve track URL: {resolve_response.text}"
         track_data = resolve_response.json()
-        streaming_url = track_data.get('stream_url')
-        
-        if not streaming_url:
-            return "Streaming URL not found.", ""
-        
-        status = "Downloading audio..."
-        download_status = download_audio(streaming_url, 'track.wav', headers)
-        
-        if "Error" in download_status:
-            return download_status, ""
-        
-        status = "Processing audio..."
-        
-        # Ensure 'track.wav' exists and is not a directory
-        if not os.path.isfile('track.wav'):
-            raise FileNotFoundError("Downloaded audio file 'track.wav' does not exist or is not valid.")
-        
-        audio = AudioSegment.from_wav('track.wav')
-        
-        chunk_duration = 30000  # 30 seconds
-        overlap = 10000  # 10 seconds
-        
-        chunks = []
-        start = 0
-        
-        while start + chunk_duration < len(audio):
-            end = start + chunk_duration
-            chunk = audio[start:end]
-            chunks.append((start, chunk))
-            start += chunk_duration - overlap
-        
-        shazam = shazamio.Shazam()
-        tracklist = []
-        
-        for start_time, chunk in chunks:
-            progress(0.1)
-            track_info = await identify_track(shazam, chunk)
-            
-            if isinstance(track_info, str) and "Error" in track_info:  # Check for errors in identification
-                return track_info, ""
-            
-            if track_info:
-                timestamp = time.strftime("%M:%S", time.gmtime(start_time / 1000))
-                tracklist.append(f"{timestamp} - {track_info['title']} by {track_info['subtitle']}")
-        
-        status = "Generating output..."
-        tracklist_output = "\n".join(tracklist)
-        
-        return tracklist_output, tracklist_output
+        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"An error occurred during processing: {e}", ""
+        return "", "", f"Error loading audio: {str(e)}"
+
+    # Split audio into chunks
+    chunk_duration = 30000  # 30 seconds
+    overlap = 10000  # 10 seconds
+    chunks = []
+    start = 0
+    while start + chunk_duration < len(audio):
+        end = start + chunk_duration
+        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, 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)
+    with open(tracklist_path, 'w') as f:
+        f.write(tracklist_output)
+
+    # Clean up temporary directory
+    import shutil
+    shutil.rmtree(temp_dir)
+
+    # 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 {
@@ -118,26 +152,22 @@ 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_input = gr.Text(
+            track_url = gr.Text(
                 label="SoundCloud DJ Set URL",
                 show_label=False,
                 max_lines=1,
                 placeholder="Enter SoundCloud DJ set URL",
                 container=False,
             )
-
             run_button = gr.Button("Process", scale=0, variant="primary")
-
-        result_output = gr.Textbox(label="Tracklist", show_label=False)
-
+        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])
 
-        gr.Examples(examples=["https://soundcloud.com/your-track-url"], inputs=[track_url_input])
-
-    run_button.click(process_dj_set, inputs=track_url_input, outputs=[result_output, download_button])
+    run_button.click(process_dj_set, inputs=track_url, outputs=[result, download_button, status_text])
 
 if __name__ == "__main__":
-    demo.launch(share=False)
+    demo.launch()