File conversion: Make sample rate optional

maestro_worker_python/convert_files.py

@@ -22,9 +22,9 @@ def __init__(self, message):
 class FileToConvert:
     input_file_path: str
     file_format: str
-    output_file_path: str = None
+    output_file_path: str | None = None
     max_duration: int = 1200
-    sample_rate: int = 44100
+    sample_rate: int | None = 44100
 
 
 def convert_files(convert_files: List[FileToConvert]):
@@ -92,48 +92,52 @@ def convert_files_manager(*convert_files: FileToConvert) -> None | str | list[st
             obj.close()
 
 
-def _convert_to_wav(input_file_path, output_file_path, max_duration, sample_rate=44100):
-    _run_subprocess(
-        [
-            "ffmpeg",
-            "-y",
-            "-hide_banner",
-            "-loglevel",
-            "error",
-            "-t",
-            str(max_duration),
-            "-i",
-            str(input_file_path),
-            "-ar",
-            str(sample_rate),
-            str(output_file_path),
-        ]
-    )
-
-
-def _convert_to_m4a(input_file_path, output_file_path, max_duration, sample_rate=44100):
-    _run_subprocess(
-        [
-            "ffmpeg",
-            "-y",
-            "-hide_banner",
-            "-loglevel",
-            "error",
-            "-t",
-            str(max_duration),
-            "-i",
-            str(input_file_path),
-            "-c:a",
-            "aac",
-            "-b:a",
-            "192k",
-            "-ar",
-            str(sample_rate),
-            "-movflags",
-            "+faststart",
-            str(output_file_path),
-        ]
-    )
+def _convert_to_wav(input_file_path: str, output_file_path: str, max_duration: int, sample_rate: int | None = 44100):
+    command_list = [
+        "ffmpeg",
+        "-y",
+        "-hide_banner",
+        "-loglevel",
+        "error",
+        "-t",
+        str(max_duration),
+        "-i",
+        str(input_file_path),
+    ]
+
+    if sample_rate is not None:
+        command_list.extend(["-ar", str(sample_rate)])
+
+    command_list.append(str(output_file_path))
+
+    _run_subprocess(command_list)
+
+
+def _convert_to_m4a(input_file_path: str, output_file_path: str, max_duration: int, sample_rate: int | None = 44100):
+    command_list = [
+        "ffmpeg",
+        "-y",
+        "-hide_banner",
+        "-loglevel",
+        "error",
+        "-t",
+        str(max_duration),
+        "-i",
+        str(input_file_path),
+        "-c:a",
+        "aac",
+        "-b:a",
+        "192k",
+        "-movflags",
+        "+faststart",
+    ]
+
+    if sample_rate is not None:
+        command_list.extend(["-ar", str(sample_rate)])
+
+    command_list.append(str(output_file_path))
+
+    _run_subprocess(command_list)
 
 
 def _run_subprocess(command):

pyproject.toml

@@ -9,14 +9,15 @@ packages = [{include = "maestro_worker_python"}]
 exclude = ["tests/**/*"]
 
 [tool.poetry.dependencies]
-python = "^3.9"
+python = ">3.9,<3.14"
 json-logging = ">=1.3,<2"
 uvicorn = ">=0.34.0,<0.34.3"
 sentry-sdk = {version = ">=1.16,<3", extras = ["fastapi"]}
 requests = ">=2.31,<3"
 psutil = ">=6.0.0,<7.1"
 fastapi = ">=0.100.0,<0.115.13"
 pydantic-settings = ">=2.9,<3"
+pydantic_core = ">=2.9,<3"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.2.0"

tests/test_convert_files.py

@@ -58,7 +58,7 @@ def test_should_raise_validation_error_if_audio_file_is_invalid(
             ]
         )
 
-    assert "Invalid data" in str(exc.value)
+    assert "invalid" in str(exc.value).lower()
 
 
 @pytest.mark.parametrize("file_format", ["m4a", "wav"])
@@ -76,7 +76,7 @@ def test_should_raise_validation_error_if_audio_file_is_corrupt(
             ]
         )
 
-    assert "Invalid argument" in str(exc.value)
+    assert "invalid" in str(exc.value).lower()
 
 
 @pytest.mark.parametrize("file_format", ["m4a", "wav"])
@@ -102,6 +102,7 @@ def test_should_raise_validation_error_if_source_has_no_audio(file_format, caplo
 @pytest.mark.parametrize(
     "input_name, output_name, format, sample_rate",
     [
+        ("silent.ogg", "converted_44100.wav", "wav", None),
         ("silent.ogg", "converted_44100.wav", "wav", 44100),
         ("silent with space.ogg", "converted_44100.wav", "wav", 44100),
         ("silent.ogg", "converted_48000.wav", "wav", 48000),
@@ -130,6 +131,7 @@ def test_should_convert_valid_wav_audio_file(input_name, output_name, format, sa
 @pytest.mark.parametrize(
     "input_name, output_name, format, sample_rate",
     [
+        ("silent.ogg", "converted_44100.m4a", "m4a", None),
         ("silent.ogg", "converted_44100.m4a", "m4a", 44100),
         ("silent with space.wav", "converted_44100.m4a", "m4a", 44100),
         ("silent.ogg", "converted_48000.m4a", "m4a", 48000),
@@ -184,14 +186,19 @@ def _get_hash(file_name, sample_rate):
         "error",
         "-i",
         str(file_name),
-        "-ar",
-        str(sample_rate),
+    ]
+    if sample_rate:
+        command.extend([
+            "-ar",
+            str(sample_rate),
+        ])
+    command.extend([
         "-map",
         "0",
         "-f",
         "hash",
-        "-"
-    ]
+        "-",
+    ])
 
     process = subprocess.run(command, shell=False, capture_output=True, check=True)
     return process.stdout.split(b"=")[1].strip()