Update utils.py

utils.py

@@ -1,140 +1,44 @@
-import os, zipfile, shutil, subprocess, shlex, sys # noqa
-from urllib.parse import urlparse
-import re
-import logging
-
-
-def load_file_from_url(
-    url: str,
-    model_dir: str,
-    file_name: str | None = None,
-    overwrite: bool = False,
-    progress: bool = True,
-) -> str:
-    """Download a file from `url` into `model_dir`,
-    using the file present if possible.
-    Returns the path to the downloaded file.
-    """
-    os.makedirs(model_dir, exist_ok=True)
-    if not file_name:
-        parts = urlparse(url)
-        file_name = os.path.basename(parts.path)
-    cached_file = os.path.abspath(os.path.join(model_dir, file_name))
-
-    # Overwrite
-    if os.path.exists(cached_file):
-        if overwrite or os.path.getsize(cached_file) == 0:
-            remove_files(cached_file)
-
-    # Download
-    if not os.path.exists(cached_file):
-        logger.info(f'Downloading: "{url}" to {cached_file}\n')
-        from torch.hub import download_url_to_file
-
-        download_url_to_file(url, cached_file, progress=progress)
-    else:
-        logger.debug(cached_file)
-
-    return cached_file
-
-
-def friendly_name(file: str):
-    if file.startswith("http"):
-        file = urlparse(file).path
-
-    file = os.path.basename(file)
-    model_name, extension = os.path.splitext(file)
-    return model_name, extension
-
-
-def download_manager(
-    url: str,
-    path: str,
-    extension: str = "",
-    overwrite: bool = False,
-    progress: bool = True,
-):
-    url = url.strip()
-
-    name, ext = friendly_name(url)
-    name += ext if not extension else f".{extension}"
-
-    if url.startswith("http"):
-        filename = load_file_from_url(
-            url=url,
-            model_dir=path,
-            file_name=name,
-            overwrite=overwrite,
-            progress=progress,
-        )
-    else:
-        filename = path
-
-    return filename
-
+# utils.py
 
-def remove_files(file_list):
-    if isinstance(file_list, str):
-        file_list = [file_list]
-
-    for file in file_list:
-        if os.path.exists(file):
-            os.remove(file)
-
-
-def remove_directory_contents(directory_path):
-    """
-    Removes all files and subdirectories within a directory.
-    Parameters:
-    directory_path (str): Path to the directory whose
-    contents need to be removed.
-    """
-    if os.path.exists(directory_path):
-        for filename in os.listdir(directory_path):
-            file_path = os.path.join(directory_path, filename)
-            try:
-                if os.path.isfile(file_path):
-                    os.remove(file_path)
-                elif os.path.isdir(file_path):
-                    shutil.rmtree(file_path)
-            except Exception as e:
-                logger.error(f"Failed to delete {file_path}. Reason: {e}")
-        logger.info(f"Content in '{directory_path}' removed.")
-    else:
-        logger.error(f"Directory '{directory_path}' does not exist.")
-
-
-# Create directory if not exists
-def create_directories(directory_path):
-    if isinstance(directory_path, str):
-        directory_path = [directory_path]
-    for one_dir_path in directory_path:
-        if not os.path.exists(one_dir_path):
-            os.makedirs(one_dir_path)
-            logger.debug(f"Directory '{one_dir_path}' created.")
-
-
-def setup_logger(name_log):
-    logger = logging.getLogger(name_log)
-    logger.setLevel(logging.INFO)
-
-    _default_handler = logging.StreamHandler()  # Set sys.stderr as stream.
-    _default_handler.flush = sys.stderr.flush
-    logger.addHandler(_default_handler)
-
-    logger.propagate = False
-
-    handlers = logger.handlers
-
-    for handler in handlers:
-        formatter = logging.Formatter("[%(levelname)s] >> %(message)s")
-        handler.setFormatter(formatter)
-
-    # logger.handlers
-
-    return logger
+import os
+import logging
+import requests
 
 
-logger = setup_logger("ss")
+# Basic logger setup
+logger = logging.getLogger("audio_splitter")
 logger.setLevel(logging.INFO)
-
+ch = logging.StreamHandler()
+ch.setFormatter(logging.Formatter("[%(levelname)s] %(message)s"))
+logger.addHandler(ch)
+
+
+def create_directories(path):
+    os.makedirs(path, exist_ok=True)
+
+
+def remove_directory_contents(path):
+    if not os.path.exists(path):
+        return
+    for filename in os.listdir(path):
+        file_path = os.path.join(path, filename)
+        if os.path.isfile(file_path):
+            os.remove(file_path)
+        elif os.path.isdir(file_path):
+            import shutil
+            shutil.rmtree(file_path)
+
+
+def download_manager(url, output_dir):
+    filename = os.path.basename(url)
+    output_path = os.path.join(output_dir, filename)
+    if os.path.exists(output_path):
+        return output_path
+    logger.info(f"Downloading: {filename}")
+    with requests.get(url, stream=True) as r:
+        r.raise_for_status()
+        with open(output_path, 'wb') as f:
+            for chunk in r.iter_content(chunk_size=8192):
+                f.write(chunk)
+    logger.info(f"Downloaded: {filename}")
+    return output_path