feat: better file validation

Author: slhck

src/ffmpeg_normalize/__main__.py

@@ -723,6 +723,18 @@ def _split_options(opts: str) -> list[str]:
     if not input_files:
         error("No input files specified. Use positional arguments or --input-list.")
 
+    # Validate all input files upfront before processing
+    _logger.debug("Validating all input files before processing...")
+    validation_errors = FFmpegNormalize.validate_input_files(input_files)
+    if validation_errors:
+        _logger.error("Validation failed for the following files:")
+        for err in validation_errors:
+            _logger.error(f"  - {err}")
+        error(
+            f"Validation failed for {len(validation_errors)} file(s). "
+            "Please fix the issues above and try again."
+        )
+
     for index, input_file in enumerate(input_files):
         if cli_args.output is not None and index < len(cli_args.output):
             if cli_args.output_folder and cli_args.output_folder != "normalized":

src/ffmpeg_normalize/_cmd_utils.py

@@ -190,3 +190,76 @@ def ffmpeg_has_loudnorm() -> bool:
             "Please make sure you are running ffmpeg v4.2 or above."
         )
     return supports_loudnorm
+
+
+def validate_input_file(input_file: str) -> tuple[bool, str | None]:
+    """
+    Validate that an input file exists, is readable, and contains audio streams.
+
+    This function performs a lightweight probe of the file using ffmpeg to check
+    if it can be read and contains at least one audio stream.
+
+    Args:
+        input_file: Path to the input file to validate
+
+    Returns:
+        tuple: (is_valid, error_message)
+            - is_valid: True if the file is valid, False otherwise
+            - error_message: None if valid, otherwise a descriptive error message
+    """
+    # Check if file exists
+    if not os.path.exists(input_file):
+        return False, f"File does not exist: {input_file}"
+
+    # Check if it's actually a file (not a directory)
+    if not os.path.isfile(input_file):
+        return False, f"Path is not a file: {input_file}"
+
+    # Check if file is readable
+    if not os.access(input_file, os.R_OK):
+        return False, f"File is not readable (permission denied): {input_file}"
+
+    # Check if file has audio streams using ffmpeg probe
+    ffmpeg_exe = get_ffmpeg_exe()
+    cmd = [
+        ffmpeg_exe,
+        "-i",
+        input_file,
+        "-c",
+        "copy",
+        "-t",
+        "0",
+        "-map",
+        "0",
+        "-f",
+        "null",
+        os.devnull,
+    ]
+
+    try:
+        output = CommandRunner().run_command(cmd).get_output()
+    except RuntimeError as e:
+        error_str = str(e)
+        # Extract a cleaner error message from ffmpeg output
+        if "Invalid data found" in error_str:
+            return False, f"Invalid or corrupted media file: {input_file}"
+        if "No such file or directory" in error_str:
+            return False, f"File not found or cannot be opened: {input_file}"
+        if "Permission denied" in error_str:
+            return False, f"Permission denied when reading file: {input_file}"
+        if "does not contain any stream" in error_str:
+            return False, f"File contains no media streams: {input_file}"
+        # Generic error for other ffmpeg failures
+        return False, f"Cannot read media file: {input_file}"
+
+    # Check for audio streams in the output
+    has_audio = False
+    for line in output.split("\n"):
+        if line.strip().startswith("Stream") and "Audio" in line:
+            has_audio = True
+            break
+
+    if not has_audio:
+        return False, f"File does not contain any audio streams: {input_file}"
+
+    return True, None

src/ffmpeg_normalize/_ffmpeg_normalize.py

@@ -9,7 +9,7 @@
 
 from tqdm import tqdm
 
-from ._cmd_utils import ffmpeg_has_loudnorm, get_ffmpeg_exe
+from ._cmd_utils import ffmpeg_has_loudnorm, get_ffmpeg_exe, validate_input_file
 from ._errors import FFmpegNormalizeError
 from ._media_file import MediaFile
 
@@ -317,6 +317,28 @@ def add_media_file(self, input_file: str, output_file: str) -> None:
         self.media_files.append(MediaFile(self, input_file, output_file))
         self.file_count += 1
 
+    @staticmethod
+    def validate_input_files(input_files: list[str]) -> list[str]:
+        """
+        Validate all input files before processing.
+
+        This method checks that each input file exists, is readable, and contains
+        at least one audio stream. All files are validated upfront so that users
+        can fix all issues before rerunning the batch.
+
+        Args:
+            input_files: List of input file paths to validate
+
+        Returns:
+            list: List of error messages for invalid files. Empty if all files are valid.
+        """
+        errors = []
+        for input_file in input_files:
+            is_valid, error_msg = validate_input_file(input_file)
+            if not is_valid and error_msg:
+                errors.append(error_msg)
+        return errors
+
     def _calculate_batch_reference(self) -> float | None:
         """
         Calculate the batch reference loudness by averaging measurements across all files.

tests/test_all.py

@@ -599,3 +599,62 @@ def test_keep_other_and_keep_original_conflict(self):
             ]
         )
         assert "Cannot use both" in stderr
+
+
+class TestFileValidation:
+    """Tests for pre-batch file validation."""
+
+    def test_nonexistent_file(self):
+        """Test that validation fails for non-existent files."""
+        _, stderr = ffmpeg_normalize_call(["nonexistent_file.mp4"])
+        assert "Validation failed" in stderr
+        assert "does not exist" in stderr
+
+    def test_multiple_invalid_files(self):
+        """Test that validation reports all invalid files at once."""
+        _, stderr = ffmpeg_normalize_call(
+            ["nonexistent1.mp4", "nonexistent2.mp4", "nonexistent3.mp4"]
+        )
+        assert "Validation failed for 3 file(s)" in stderr
+        assert "nonexistent1.mp4" in stderr
+        assert "nonexistent2.mp4" in stderr
+        assert "nonexistent3.mp4" in stderr
+
+    def test_file_without_audio(self, tmp_path):
+        """Test that validation fails for files without audio streams."""
+        # Create a video-only file using ffmpeg
+        video_only = tmp_path / "video_only.mp4"
+        cmd = [
+            "ffmpeg",
+            "-y",
+            "-f",
+            "lavfi",
+            "-i",
+            "color=c=blue:s=320x240:d=1",
+            "-an",  # no audio
+            "-c:v",
+            "libx264",
+            str(video_only),
+        ]
+        subprocess.run(cmd, check=True, capture_output=True)
+
+        _, stderr = ffmpeg_normalize_call([str(video_only)])
+        assert "Validation failed" in stderr
+        assert "does not contain any audio streams" in stderr
+
+    def test_mixed_valid_invalid_files(self, tmp_path):
+        """Test that validation fails when mix of valid and invalid files."""
+        _, stderr = ffmpeg_normalize_call(
+            ["tests/test.mp4", "nonexistent.mp4", "tests/test.m4a"]
+        )
+        # Should report the invalid file but not the valid ones
+        assert "Validation failed for 1 file(s)" in stderr
+        assert "nonexistent.mp4" in stderr
+        # Valid files should not appear in error messages
+        assert "test.mp4" not in stderr or "does not exist" not in stderr
+
+    def test_valid_file_passes_validation(self):
+        """Test that valid files pass validation."""
+        # Should succeed (file exists and has audio)
+        ffmpeg_normalize_call(["tests/test.mp4", "-n"])  # dry run
+        # No assertion needed - if validation fails, the command will error

tests/test_api_ground_truth.py

@@ -11,6 +11,7 @@
 Run with: pytest -m "slow" to run only the slow integration tests
 """
 
+import subprocess
 import tempfile
 import zipfile
 from pathlib import Path
@@ -19,6 +20,7 @@
 import pytest
 
 from ffmpeg_normalize import FFmpegNormalize
+from ffmpeg_normalize._cmd_utils import validate_input_file
 from ffmpeg_normalize._streams import LoudnessStatisticsWithMetadata
 
 
@@ -640,3 +642,88 @@ def test_batch_mode_preserves_relative_loudness(self, temp_output_dir):
                 f"Normal file should remain louder than quiet file in batch mode. "
                 f"Got normal={loudness_values[1]:.2f}, quiet={loudness_values[2]:.2f}"
             )
+
+
+class TestFileValidationAPI:
+    """API-level tests for input file validation."""
+
+    def test_validate_input_file_exists(self):
+        """Test validation of existing audio file."""
+        test_file = Path(__file__).parent / "test.mp4"
+        is_valid, error_msg = validate_input_file(str(test_file))
+        assert is_valid is True
+        assert error_msg is None
+
+    def test_validate_input_file_not_exists(self):
+        """Test validation fails for non-existent file."""
+        is_valid, error_msg = validate_input_file("nonexistent_file.mp4")
+        assert is_valid is False
+        assert error_msg is not None
+        assert "does not exist" in error_msg
+
+    def test_validate_input_file_directory(self, tmp_path):
+        """Test validation fails for directories."""
+        is_valid, error_msg = validate_input_file(str(tmp_path))
+        assert is_valid is False
+        assert error_msg is not None
+        assert "not a file" in error_msg
+
+    def test_validate_input_file_no_audio(self, tmp_path):
+        """Test validation fails for files without audio streams."""
+        # Create a video-only file using ffmpeg
+        video_only = tmp_path / "video_only.mp4"
+        cmd = [
+            "ffmpeg",
+            "-y",
+            "-f",
+            "lavfi",
+            "-i",
+            "color=c=blue:s=320x240:d=1",
+            "-an",  # no audio
+            "-c:v",
+            "libx264",
+            str(video_only),
+        ]
+        subprocess.run(cmd, check=True, capture_output=True)
+
+        is_valid, error_msg = validate_input_file(str(video_only))
+        assert is_valid is False
+        assert error_msg is not None
+        assert "audio" in error_msg.lower()
+
+    def test_validate_input_files_batch(self):
+        """Test batch validation with multiple files."""
+        test_dir = Path(__file__).parent
+        valid_files = [
+            str(test_dir / "test.mp4"),
+            str(test_dir / "test.m4a"),
+        ]
+
+        errors = FFmpegNormalize.validate_input_files(valid_files)
+        assert len(errors) == 0
+
+    def test_validate_input_files_batch_mixed(self):
+        """Test batch validation with mix of valid and invalid files."""
+        test_dir = Path(__file__).parent
+        files = [
+            str(test_dir / "test.mp4"),
+            "nonexistent1.mp4",
+            str(test_dir / "test.m4a"),
+            "nonexistent2.mp4",
+        ]
+
+        errors = FFmpegNormalize.validate_input_files(files)
+        assert len(errors) == 2
+        assert any("nonexistent1.mp4" in e for e in errors)
+        assert any("nonexistent2.mp4" in e for e in errors)
+
+    def test_validate_input_files_all_invalid(self):
+        """Test batch validation with all invalid files."""
+        files = [
+            "nonexistent1.mp4",
+            "nonexistent2.mp4",
+            "nonexistent3.mp4",
+        ]
+
+        errors = FFmpegNormalize.validate_input_files(files)
+        assert len(errors) == 3