Update api_v2.py

api_v2.py

@@ -192,6 +192,56 @@ def process_audio_path(audio_path) -> Tuple[str, bool]:
     # If not a URL or download failed, return the original path
     return audio_path, False  # Not a temporary file
 
+# Function to process weight files (similar to process_audio_path)
+def process_weights_path(weights_path) -> Tuple[str, bool]:
+    """
+    Process a weights path, downloading it if it's a URL.
+    
+    Args:
+        weights_path (str): Path or URL to weights file
+        
+    Returns:
+        Tuple[str, bool]: (local_path, is_temporary)
+    """
+    if weights_path and (weights_path.startswith('http://') or weights_path.startswith('https://') or 
+                        weights_path.startswith('s3://')):
+        try:
+            # Create temp directory if it doesn't exist
+            temp_dir = os.path.join(now_dir, "temp_weights")
+            os.makedirs(temp_dir, exist_ok=True)
+            
+            # Generate a filename from the URL
+            parsed_url = urllib.parse.urlparse(weights_path)
+            filename = os.path.basename(parsed_url.path)
+            if not filename:
+                filename = f"temp_weights_{hash(weights_path)}.pth"
+            
+            # Full path for downloaded file
+            local_path = os.path.join(temp_dir, filename)
+            
+            # Download file
+            if weights_path.startswith('s3://'):
+                # S3 implementation placeholder
+                print(f"Downloading from S3: {weights_path}")
+                raise NotImplementedError("S3 download not implemented. Add boto3 library and implementation.")
+            else:
+                # HTTP/HTTPS download
+                print(f"Downloading weights from URL: {weights_path}")
+                response = requests.get(weights_path, stream=True)
+                response.raise_for_status()
+                with open(local_path, 'wb') as f:
+                    for chunk in response.iter_content(chunk_size=8192):
+                        f.write(chunk)
+            
+            print(f"Downloaded weights to: {local_path}")
+            return local_path, True  # Return path and flag indicating it's temporary
+        except Exception as e:
+            print(f"Error downloading weights file: {e}")
+            raise Exception(f"Failed to download weights from URL: {e}")
+    
+    # If not a URL or download failed, return the original path
+    return weights_path, False  # Not a temporary file
+
 parser = argparse.ArgumentParser(description="GPT-SoVITS api")
 parser.add_argument("-c", "--tts_config", type=str, default="GPT_SoVITS/configs/tts_infer.yaml", help="tts_infer路径")
 parser.add_argument("-a", "--bind_addr", type=str, default="127.0.0.1", help="default: 127.0.0.1")
@@ -543,25 +593,71 @@ async def set_refer_aduio(refer_audio_path: str = None):
 
 @APP.get("/set_gpt_weights")
 async def set_gpt_weights(weights_path: str = None):
+    temp_file = None
     try:
         if weights_path in ["", None]:
             return JSONResponse(status_code=400, content={"message": "gpt weight path is required"})
-        tts_pipeline.init_t2s_weights(weights_path)
+        
+        # Process the path (download if it's a URL)
+        local_path, is_temp = process_weights_path(weights_path)
+        if is_temp:
+            temp_file = local_path
+            
+        # Load the weights
+        tts_pipeline.init_t2s_weights(local_path)
+        
+        # Clean up if it was a temporary file
+        # Note: Depending on how init_t2s_weights works, you might need to keep the file
+        # If the function loads the file into memory, you can delete it right away
+        if temp_file and os.path.exists(temp_file):
+            os.remove(temp_file)
+            print(f"Removed temporary weights file: {temp_file}")
+            
+        return JSONResponse(status_code=200, content={"message": "success"})
     except Exception as e:
+        # Clean up temp file in case of error
+        if temp_file and os.path.exists(temp_file):
+            try:
+                os.remove(temp_file)
+                print(f"Removed temporary weights file: {temp_file}")
+            except Exception as cleanup_error:
+                print(f"Error removing temporary file {temp_file}: {cleanup_error}")
+                
         return JSONResponse(status_code=400, content={"message": f"change gpt weight failed", "Exception": str(e)})
 
-    return JSONResponse(status_code=200, content={"message": "success"})
-
-
 @APP.get("/set_sovits_weights")
 async def set_sovits_weights(weights_path: str = None):
+    temp_file = None
     try:
         if weights_path in ["", None]:
             return JSONResponse(status_code=400, content={"message": "sovits weight path is required"})
-        tts_pipeline.init_vits_weights(weights_path)
+        
+        # Process the path (download if it's a URL)
+        local_path, is_temp = process_weights_path(weights_path)
+        if is_temp:
+            temp_file = local_path
+            
+        # Load the weights
+        tts_pipeline.init_vits_weights(local_path)
+        
+        # Clean up if it was a temporary file
+        # Note: Depending on how init_vits_weights works, you might need to keep the file
+        # If the function loads the file into memory, you can delete it right away
+        if temp_file and os.path.exists(temp_file):
+            os.remove(temp_file)
+            print(f"Removed temporary weights file: {temp_file}")
+            
+        return JSONResponse(status_code=200, content={"message": "success"})
     except Exception as e:
+        # Clean up temp file in case of error
+        if temp_file and os.path.exists(temp_file):
+            try:
+                os.remove(temp_file)
+                print(f"Removed temporary weights file: {temp_file}")
+            except Exception as cleanup_error:
+                print(f"Error removing temporary file {temp_file}: {cleanup_error}")
+                
         return JSONResponse(status_code=400, content={"message": f"change sovits weight failed", "Exception": str(e)})
-    return JSONResponse(status_code=200, content={"message": "success"})