Repository: Zulko/moviepy
Pull request #2470 (2025-06-27 14:00:52)

FFmpeg Parser refactor
<!--
Please tick when you have done these. They don't need to all be completed before the PR is submitted.
Delete them if they are not appropriate for this pull request.

- [ ] I have provided code that clearly demonstrates the bug and that only works correctly when applying this fix
- [ ] I have added suitable tests demonstrating a fixed bug or new/changed feature to the test suite in `tests/`
- [ ] I have properly documented new or changed features in the documentation or in the docstrings
- [ ] I have properly explained unusual or unexpected code in the comments around it
-->
Work in progress, rewriting FFmpeg parser for something more readable and robust



Patch:
diff --git a/docs/conf.py b/docs/conf.py
index 4989020..2dc3577 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -121,8 +121,8 @@ html_theme_options = {
             "type": "fontawesome",
         }
     ],
-    "announcement": f"""<p>MoviePy v2.0 have introduced breaking changes, 
-    see <a href="{v2_page}">Updating from v1.X to v2.X</a> for more info.</p>""",
+    "announcement": f"""<p>MoviePy v2.0 have introduced breaking changes, see
+<a href="{v2_page}">Updating from v1.X to v2.X</a> for more info.</p>""",
 }
 
 html_context = {
diff --git a/moviepy/audio/io/readers.py b/moviepy/audio/io/readers.py
index 2f11286..607456f 100644
--- a/moviepy/audio/io/readers.py
+++ b/moviepy/audio/io/readers.py
@@ -61,7 +61,9 @@ class FFMPEG_AudioReader:
         self.format = "s%dle" % (8 * nbytes)
         self.codec = "pcm_s%dle" % (8 * nbytes)
         self.nchannels = nchannels
-        infos = ffmpeg_parse_infos(filename, decode_file=decode_file)
+        infos = ffmpeg_parse_infos(
+            filename, decode_file=decode_file, print_infos=print_infos
+        )
         self.duration = infos["duration"]
         self.bitrate = infos["audio_bitrate"]
         self.infos = infos
diff --git a/moviepy/video/io/ffmpeg_reader.py b/moviepy/video/io/ffmpeg_reader.py
index f8dcb74..24c6a0b 100644
--- a/moviepy/video/io/ffmpeg_reader.py
+++ b/moviepy/video/io/ffmpeg_reader.py
@@ -4,6 +4,7 @@ import os
 import re
 import subprocess as sp
 import warnings
+from typing import List
 
 import numpy as np
 
@@ -327,10 +328,9 @@ def ffmpeg_read_image(filename, with_mask=True, pixel_format=None):
 
 
 class FFmpegInfosParser:
-    """Finite state ffmpeg `-i` command option file information parser.
-    Is designed to parse the output fast, in one loop. Iterates line by
-    line of the `ffmpeg -i <filename> [-f null -]` command output changing
-    the internal state of the parser.
+    """An (hopefully) robuste ffmpeg `-i` command option file information parser.
+    Is designed to parse the output by extracting the different blocks of informations,
+    based on the indentation, in order to create an easy to handle block tree
 
     Parameters
     ----------
@@ -353,6 +353,35 @@ class FFmpegInfosParser:
       will differ depending on this argument.
     """
 
+    class ParseDimensionError(VideoCorruptedError):
+        """Error raised when we cannot find dimensions in a video stream"""
+
+        pass
+
+    class ParseDurationError(VideoCorruptedError):
+        """Error raised when we cannot find duration in a video stream"""
+
+        pass
+
+    class InfoBlock:
+        """Represents a block of output from ffmpeg, which can be an input file,
+        stream, chapter or metadata.
+        """
+
+        def __init__(self, block_line, indent_level=0):
+            self.type = "unknown"
+            self.childs: List[FFmpegInfosParser.InfoBlock] = []
+            self.parent = None
+            self.indent_level = indent_level
+            self.head_line = block_line
+            self.raw_data = []
+            self.data = {}
+
+        def add_child(self, child):
+            """Adds a child to the current block."""
+            child.parent = self
+            self.childs.append(child)
+
     def __init__(
         self,
         infos,
@@ -366,367 +395,205 @@ class FFmpegInfosParser:
         self.check_duration = check_duration
         self.fps_source = fps_source
         self.duration_tag_separator = "time=" if decode_file else "Duration: "
+        self.blocks = None
+        self.video_stream = None
+        self.audio_stream = None
+        self.data_stream = None
 
-        self._reset_state()
-
-    def _reset_state(self):
-        """Reinitializes the state of the parser. Used internally at
-        initialization and at the end of the parsing process.
-        """
-        # could be 2 possible types of metadata:
-        #   - file_metadata: Metadata of the container. Here are the tags set
-        #     by the user using `-metadata` ffmpeg option
-        #   - stream_metadata: Metadata for each stream of the container.
-        self._inside_file_metadata = False
-
-        # this state is needed if `duration_tag_separator == "time="` because
-        # execution of ffmpeg decoding the whole file using `-f null -` appends
-        # to the output the blocks "Stream mapping:" and "Output:", which
-        # should be ignored
-        self._inside_output = False
-
-        # map from stream type to default stream
-        # if a default stream is not indicated, pick the first one available
-        self._default_streams = {}
-
-        # current input file, stream and chapter, which will be built at runtime
-        self._current_input_file = {"streams": []}
-        self._current_stream = None
-        self._current_chapter = None
-
-        # if stream type is video then has side data
-        self._inside_stream_sidedata = False
-
-        # resulting data of the parsing process
         self.result = {
             "video_found": False,
             "audio_found": False,
             "metadata": {},
-            "inputs": [],
+            "blocks": None,
+            "inputs": {},
         }
 
-        # keep the value of latest metadata value parsed so we can build
-        # at next lines a multiline metadata value
-        self._last_metadata_field_added = None
+    def _extract_block(self, index, start_indent, block: InfoBlock = None):
+        lines = self.infos.splitlines()
+        block.content = []
+        multiline = None
+        is_last_line = False
+        while index < len(lines) - 1:
+            index += 1
+            is_last_line = index == (len(lines) - 1)
+            line = lines[index]
+            indent_level = (len(line) - len(line.lstrip())) / 2
+            line = line.strip()
+
+            if not is_last_line:
+                next_line = lines[index + 1].strip()
+            else:
+                next_line = False
 
-    def parse(self):
-        """Parses the information returned by FFmpeg in stderr executing their binary
-        for a file with ``-i`` option and returns a dictionary with all data needed
-        by MoviePy.
-        """
-        # chapters by input file
-        input_chapters = []
-
-        for line in self.infos.splitlines()[1:]:
-            if (
-                self.duration_tag_separator == "time="
-                and self.check_duration
-                and "time=" in line
-            ):
-                # parse duration using file decodification
-                self.result["duration"] = self.parse_duration(line)
-            elif self._inside_output or line[0] != " ":
-                if self.duration_tag_separator == "time=" and not self._inside_output:
-                    self._inside_output = True
-                # skip lines like "At least one output file must be specified"
-            elif not self._inside_file_metadata and line.startswith("  Metadata:"):
-                # enter "  Metadata:" group
-                self._inside_file_metadata = True
-            elif line.startswith("  Duration:"):
-                # exit "  Metadata:" group
-                self._inside_file_metadata = False
-                if self.check_duration and self.duration_tag_separator == "Duration: ":
-                    self.result["duration"] = self.parse_duration(line)
-
-                # parse global bitrate (in kb/s)
-                bitrate_match = re.search(r"bitrate: (\d+) k(i?)b/s", line)
-                self.result["bitrate"] = (
-                    int(bitrate_match.group(1)) if bitrate_match else None
-                )
+            # End of block
+            if indent_level <= start_indent:
+                index -= 1
+                break
 
-                # parse start time (in seconds)
-                start_match = re.search(r"start: (\d+\.?\d+)", line)
-                self.result["start"] = (
-                    float(start_match.group(1)) if start_match else None
-                )
-            elif self._inside_file_metadata:
-                # file metadata line
-                field, value = self.parse_metadata_field_value(line)
-
-                # multiline metadata value parsing
-                if field == "":
-                    field = self._last_metadata_field_added
-                    value = self.result["metadata"][field] + "\n" + value
-                else:
-                    self._last_metadata_field_added = field
-
-                self.result["metadata"][field] = value
-            elif line.lstrip().startswith("Stream "):
-                # exit stream "    Metadata:"
-                self._inside_stream_sidedata = False
-
-                if self._current_stream:
-                    self._current_input_file["streams"].append(self._current_stream)
-
-                # get input number, stream number, language and type
-                main_info_match = re.search(
-                    r"^Stream\s#(\d+):(\d+)(?:\[\w+\])?\(?(\w+)?\)?:\s(\w+):",
-                    line.lstrip(),
+            # New block
+            if line.lstrip().startswith(
+                ("Metadata", "Stream", "Side data", "Chapter", "Chapters")
+            ):
+                (child_block, index) = self._extract_block(
+                    index, indent_level, self.InfoBlock(line.lstrip(), indent_level)
                 )
-                (
-                    input_number,
-                    stream_number,
-                    language,
-                    stream_type,
-                ) = main_info_match.groups()
-                input_number = int(input_number)
-                stream_number = int(stream_number)
-                stream_type_lower = stream_type.lower()
-
-                if language == "und":
-                    language = None
-
-                # start builiding the current stream
-                self._current_stream = {
-                    "input_number": input_number,
-                    "stream_number": stream_number,
-                    "stream_type": stream_type_lower,
-                    "language": language,
-                    "default": (stream_type_lower not in self._default_streams)
-                    and line.endswith("(default)"),
-                }
-
-                # for default streams, set their numbers globally, so it's
-                # easy to get without iterating all
-                if self._current_stream["default"]:
-                    self._default_streams[stream_type_lower] = self._current_stream
-
-                # exit chapter
-                if self._current_chapter:
-                    input_chapters[input_number].append(self._current_chapter)
-                    self._current_chapter = None
-
-                if "input_number" not in self._current_input_file:
-                    # first input file
-                    self._current_input_file["input_number"] = input_number
-                elif self._current_input_file["input_number"] != input_number:
-                    # new input file
-
-                    # include their chapters if there are for this input file
-                    if len(input_chapters) >= input_number + 1:
-                        self._current_input_file["chapters"] = input_chapters[
-                            input_number
-                        ]
-
-                    # add new input file to result
-                    self.result["inputs"].append(self._current_input_file)
-                    self._current_input_file = {"input_number": input_number}
-
-                # parse relevant data by stream type
-                try:
-                    stream_data = self.parse_data_by_stream_type(stream_type, line)
-                except NotImplementedError as exc:
-                    warnings.warn(
-                        f"{str(exc)}\nffmpeg output:\n\n{self.infos}", UserWarning
-                    )
-                else:
-                    self._current_stream.update(stream_data)
-            elif line.startswith('      Metadata:'):
-                # enter group "    Metadata:"
+                self._parse_headline_data(child_block)
+                block.add_child(child_block)
                 continue
-            elif self._current_stream:
-                # stream metadata line
-                if "metadata" not in self._current_stream:
-                    self._current_stream["metadata"] = {}
-
-                if line.strip().startswith("Side data:"):
-                    self._inside_stream_sidedata = True
-                    if "side_data" not in self._current_stream:
-                        self._current_stream["sidedata"] = {}
-                    continue
 
-                field, value = self.parse_metadata_field_value(line)
-
-                if self._current_stream["stream_type"] == "video":
-                    field, value = self.video_metadata_type_casting(field, value)
-                    # ffmpeg 7 now use displaymatrix instead of rotate
-                    if field == "rotate":
-                        self.result["video_rotation"] = value
-                    elif field == "displaymatrix":
-                        self.result["video_rotation"] = value
-
-                # multiline metadata value parsing
-                if field == "" and self._last_metadata_field_added not in ['rotate', 'displaymatrix']:
-                    field = self._last_metadata_field_added
-                    value = self._current_stream["metadata"][field] + "\n" + value
-                else:
-                    self._last_metadata_field_added = field
-
-                if not self._inside_stream_sidedata:
-                    self._current_stream["metadata"][field] = value
-                else:
-                    self._current_stream["sidedata"][field] = value
-                
-            elif line.startswith("    Chapter"):
-                # Chapter data line
-                if self._current_chapter:
-                    # there is a previews chapter?
-                    if len(input_chapters) < self._current_chapter["input_number"] + 1:
-                        input_chapters.append([])
-                    # include in the chapters by input matrix
-                    input_chapters[self._current_chapter["input_number"]].append(
-                        self._current_chapter
-                    )
-
-                # extract chapter data
-                chapter_data_match = re.search(
-                    r"^    Chapter #(\d+):(\d+): start (\d+\.?\d+?), end (\d+\.?\d+?)",
-                    line,
-                )
-                input_number, chapter_number, start, end = chapter_data_match.groups()
-
-                # start building the chapter
-                self._current_chapter = {
-                    "input_number": int(input_number),
-                    "chapter_number": int(chapter_number),
-                    "start": float(start),
-                    "end": float(end),
-                }
-            elif self._current_chapter:
-                # inside chapter metadata
-                if "metadata" not in self._current_chapter:
-                    self._current_chapter["metadata"] = {}
-                field, value = self.parse_metadata_field_value(line)
-
-                # multiline metadata value parsing
-                if field == "":
-                    field = self._last_metadata_field_added
-                    value = self._current_chapter["metadata"][field] + "\n" + value
-                else:
-                    self._last_metadata_field_added = field
-                self._current_chapter["metadata"][field] = value
-
-        # last input file, must be included in the result
-        if self._current_input_file:
-            self._current_input_file["streams"].append(self._current_stream)
-            # include their chapters, if there are any
-            if (
-                "input_number" in self._current_input_file
-                and len(input_chapters) == self._current_input_file["input_number"] + 1
-            ):
-                self._current_input_file["chapters"] = input_chapters[
-                    self._current_input_file["input_number"]
-                ]
-            self.result["inputs"].append(self._current_input_file)
-
-        # set any missing default automatically
-        for stream in self._current_input_file["streams"]:
-            if stream["stream_type"] not in self._default_streams:
-                self._default_streams[stream["stream_type"]] = stream
-                stream["default"] = True
-
-        # set some global info based on the defaults
-        for stream_type_lower, stream_data in self._default_streams.items():
-            self.result[f"default_{stream_type_lower}_input_number"] = stream_data[
-                "input_number"
-            ]
-            self.result[f"default_{stream_type_lower}_stream_number"] = stream_data[
-                "stream_number"
-            ]
+            # Support for multiline entries
+            if line.startswith(":") or (next_line and next_line.startswith(":")):
+                if not multiline:
+                    multiline = line
+                    continue
+                elif next_line.startswith(":"):
+                    multiline += "\n" + line[1:].strip()
+                    continue
 
-            if stream_type_lower == "audio":
-                self.result["audio_found"] = True
-                self.result["audio_fps"] = stream_data["fps"]
-                self.result["audio_bitrate"] = stream_data["bitrate"]
-            elif stream_type_lower == "video":
-                self.result["video_found"] = True
-                self.result["video_size"] = stream_data.get("size", None)
-                self.result["video_bitrate"] = stream_data.get("bitrate", None)
-                self.result["video_fps"] = stream_data["fps"]
-                self.result["video_codec_name"] = stream_data.get("codec_name", None)
-                self.result["video_profile"] = stream_data.get("profile", None)
-
-        # some video duration utilities
-        if self.result["video_found"] and self.check_duration:
-            self.result["video_duration"] = self.result["duration"]
-            self.result["video_n_frames"] = int(
-                self.result["duration"] * self.result.get("video_fps", 0)
-            )
+            if multiline:
+                line = multiline + "\n" + line[1:].strip()
+                multiline = None
+
+            # Standard line, add to block raw data and parsed data
+            block.raw_data.append(line)
+            field, value = self._parse_line(line)
+            block.data[field] = value
+
+        return (block, index)
+
+    def _parse_headline_data(self, block: InfoBlock):
+        line = block.head_line.lstrip()
+        if line.startswith("Input "):
+            block.type = "input"
+        elif line.startswith("Metadata:"):
+            block.type = "metadata"
+        elif line.startswith("Stream "):
+            block.type = "stream"
+            self._parse_stream(block)
+        elif line.startswith("Side data:"):
+            block.type = "side_data"
+        elif line.startswith("Chapters"):
+            block.type = "chapters"
+        elif line.startswith("Chapter"):
+            block.type = "chapter"
+            self._parse_chapter(block)
+
+    def _parse_line(self, line):
+        """Parse a standard line to return (field, value) with typecasting
+        when needed (rotate, displaymatrix)
+        """
+        specials = (
+            "Ambient Viewing Environment,",
+            "Content Light Level Metadata,",
+            "Mastering Display Metadata,",
+        )
+        line = line.strip()
+        if line.startswith(specials):
+            infos = line.split(",", 1)
         else:
-            self.result["video_n_frames"] = 0
-            self.result["video_duration"] = 0.0
-        # We could have also recomputed duration from the number of frames, as follows:
-        # >>> result['video_duration'] = result['video_n_frames'] / result['video_fps']
-
-        # not default audio found, assume first audio stream is the default
-        if self.result["audio_found"] and not self.result.get("audio_bitrate"):
-            self.result["audio_bitrate"] = None
-            for streams_input in self.result["inputs"]:
-                for stream in streams_input["streams"]:
-                    if stream["stream_type"] == "audio" and stream.get("bitrate"):
-                        self.result["audio_bitrate"] = stream["bitrate"]
-                        break
-
-                if self.result["audio_bitrate"] is not None:
-                    break
+            infos = line.split(":", 1)
 
-        result = self.result
+        field = infos[0].strip()
+        value = infos[1].strip()
 
-        # reset state of the parser
-        self._reset_state()
+        if field == "rotate":
+            value = float(value)
 
-        return result
+        elif field == "displaymatrix":
+            match = re.search(r"[-+]?\d+(\.\d+)?", value)
+            if match:
+                # We must multiply by -1 because displaymatrix return info
+                # about how to rotate to show video, not about video rotation
+                value = float(match.group()) * -1
 
-    def parse_data_by_stream_type(self, stream_type, line):
-        """Parses data from "Stream ... {stream_type}" line."""
+        return (field, value)
+
+    def _parse_duration(self, line):
+        """Parse the duration from the block data."""
         try:
-            return {
-                "Audio": self.parse_audio_stream_data,
-                "Video": self.parse_video_stream_data,
-                "Data": lambda _line: {},
-            }[stream_type](line)
-        except KeyError:
-            raise NotImplementedError(
-                f"{stream_type} stream parsing is not supported by moviepy and"
-                " will be ignored"
+            match_duration = re.search(
+                r"([0-9][0-9]:[0-9][0-9]:[0-9][0-9].[0-9][0-9])",
+                line,
+            )
+            if match_duration is None:
+                raise VideoCorruptedError(f"Could not parse duration from {line!r}")
+            return convert_to_seconds(match_duration.group(1))
+        except VideoCorruptedError:
+            raise
+        except Exception:
+            raise IOError(
+                (
+                    "MoviePy error: failed to read the duration of file '%s'.\n"
+                    "Here are the file infos returned by ffmpeg:\n\n%s"
+                )
+                % (self.filename, self.infos)
             )
 
-    def parse_audio_stream_data(self, line):
+    def _parse_stream(self, block: InfoBlock):
+        # get input number, stream number, language and type
+        main_info_match = re.search(
+            r"^Stream\s#(\d+):(\d+)(?:\[\w+\])?\(?(\w+)?\)?:\s(\w+):",
+            block.head_line.lstrip(),
+        )
+        (
+            input_number,
+            stream_number,
+            language,
+            stream_type,
+        ) = main_info_match.groups()
+        block.data["input_number"] = int(input_number)
+        block.data["stream_number"] = int(stream_number)
+        block.data["stream_type_lower"] = stream_type.lower()
+
+        if language == "und":
+            language = None
+
+        block.data["language"] = language
+        block.data["default"] = block.head_line.rstrip().endswith("(default)")
+
+        if block.data["stream_type_lower"] == "audio":
+            self._parse_stream_audio(block)
+        elif block.data["stream_type_lower"] == "video":
+            self._parse_stream_video(block)
+        elif block.data["stream_type_lower"] == "data":
+            self._parse_stream_data(block)
+
+    def _parse_stream_audio(self, block: InfoBlock):
         """Parses data from "Stream ... Audio" line."""
-        stream_data = {}
         try:
-            stream_data["fps"] = int(re.search(r" (\d+) Hz", line).group(1))
+            block.data["fps"] = int(re.search(r" (\d+) Hz", block.head_line).group(1))
         except (AttributeError, ValueError):
             # AttributeError: 'NoneType' object has no attribute 'group'
             # ValueError: invalid literal for int() with base 10: '<string>'
-            stream_data["fps"] = "unknown"
-        match_audio_bitrate = re.search(r"(\d+) k(i?)b/s", line)
-        stream_data["bitrate"] = (
+            block.data["fps"] = "unknown"
+
+        match_audio_bitrate = re.search(r"(\d+) k(i?)b/s", block.head_line)
+        block.data["bitrate"] = (
             int(match_audio_bitrate.group(1)) if match_audio_bitrate else None
         )
-        return stream_data
 
-    def parse_video_stream_data(self, line):
-        """Parses data from "Stream ... Video" line."""
-        stream_data = {}
+        # Store default stream, or first stream if we dont find any default
+        if block.data["default"] or not self.audio_stream:
+            self.audio_stream = block
+
+    def _parse_stream_data(self, block: InfoBlock):
+        """Parses data from "Stream ... Data" line."""
+        # Store default stream, or first stream if we dont find any default
+        if block.data["default"] or not self.data_stream:
+            self.data_stream = block
 
+    def _parse_stream_video(self, block: InfoBlock):
+        """Parses data from "Stream ... Video" line."""
         try:
-            match_video_size = re.search(r" (\d+)x(\d+)[,\s]", line)
+            match_video_size = re.search(r" (\d+)x(\d+)[,\s]", block.head_line)
             if match_video_size:
                 # size, of the form 460x320 (w x h)
-                stream_data["size"] = [int(num) for num in match_video_size.groups()]
+                block.data["size"] = [int(num) for num in match_video_size.groups()]
         except Exception:
-            raise IOError(
-                (
-                    "MoviePy error: failed to read video dimensions in"
-                    " file '%s'.\nHere are the file infos returned by"
-                    "ffmpeg:\n\n%s"
-                )
-                % (self.filename, self.infos)
-            )
+            raise FFmpegInfosParser.ParseDimensionError()
 
-        match_bitrate = re.search(r"(\d+) k(i?)b/s", line)
-        stream_data["bitrate"] = int(match_bitrate.group(1)) if match_bitrate else None
+        match_bitrate = re.search(r"(\d+) k(i?)b/s", block.head_line)
+        block.data["bitrate"] = int(match_bitrate.group(1)) if match_bitrate else None
 
         # Get the frame rate. Sometimes it's 'tbr', sometimes 'fps', sometimes
         # tbc, and sometimes tbc/2...
@@ -738,14 +605,14 @@ class FFmpegInfosParser:
 
         if self.fps_source == "fps":
             try:
-                fps = self.parse_fps(line)
+                fps = self._parse_fps(block.head_line)
             except (AttributeError, ValueError):
-                fps = self.parse_tbr(line)
+                fps = self._parse_tbr(block.head_line)
         elif self.fps_source == "tbr":
             try:
-                fps = self.parse_tbr(line)
+                fps = self._parse_tbr(block.head_line)
             except (AttributeError, ValueError):
-                fps = self.parse_fps(line)
+                fps = self._parse_fps(block.head_line)
         else:
             raise ValueError(
                 ("fps source '%s' not supported parsing the video '%s'")
@@ -758,25 +625,28 @@ class FFmpegInfosParser:
         for x in [23, 24, 25, 30, 50]:
             if (fps != x) and abs(fps - x * coef) < 0.01:
                 fps = x * coef
-        stream_data["fps"] = fps
+        block.data["fps"] = fps
 
         # Try to extract video codec and profile
         main_info_match = re.search(
             r"Video:\s(\w+)?\s?(\([^)]+\))?",
-            line.lstrip(),
+            block.head_line.lstrip(),
         )
+
         if main_info_match is not None:
             (codec_name, profile) = main_info_match.groups()
-            stream_data["codec_name"] = codec_name
-            stream_data["profile"] = profile
+            block.data["codec_name"] = codec_name
+            block.data["profile"] = profile
 
-        return stream_data
+        # Store default stream, or first stream if we dont find any default
+        if block.data["default"] or not self.video_stream:
+            self.video_stream = block
 
-    def parse_fps(self, line):
+    def _parse_fps(self, line):
         """Parses number of FPS from a line of the ``ffmpeg -i`` command output."""
         return float(re.search(r" (\d+.?\d*) fps", line).group(1))
 
-    def parse_tbr(self, line):
+    def _parse_tbr(self, line):
         """Parses number of TBS from a line of the ``ffmpeg -i`` command output."""
         s_tbr = re.search(r" (\d+.?\d*k?) tbr", line).group(1)
 
@@ -787,66 +657,180 @@ class FFmpegInfosParser:
             tbr = float(s_tbr)
         return tbr
 
-    def parse_duration(self, line):
-        """Parse the duration from the line that outputs the duration of
-        the container.
-        """
-        try:
-            time_raw_string = line.split(self.duration_tag_separator)[-1]
-            match_duration = re.search(
-                r"([0-9][0-9]:[0-9][0-9]:[0-9][0-9].[0-9][0-9])",
-                time_raw_string,
-            )
-            if match_duration is None:
-                raise VideoCorruptedError(
-                    f"Could not parse duration from {time_raw_string!r}"
+    def _parse_chapter(self, block: InfoBlock):
+        # extract chapter data
+        chapter_data_match = re.search(
+            r"^Chapter #(\d+):(\d+): start (\d+\.?\d+?), end (\d+\.?\d+?)",
+            block.head_line.strip(),
+        )
+        input_number, chapter_number, start, end = chapter_data_match.groups()
+
+        # start building the chapter
+        block.data = {
+            "input_number": int(input_number),
+            "chapter_number": int(chapter_number),
+            "start": float(start),
+            "end": float(end),
+        }
+
+    def _parse_blocks(self, root: InfoBlock):
+        for key, data in root.data.items():
+            if key == "Duration":
+                self.result["duration"] = self._parse_duration(data)
+
+                bitrate_match = re.search(r"bitrate: (\d+) k(i?)b/s", data)
+                self.result["bitrate"] = (
+                    int(bitrate_match.group(1)) if bitrate_match else None
                 )
-            return convert_to_seconds(match_duration.group(1))
-        except VideoCorruptedError:
-            raise
-        except Exception:
-            raise IOError(
-                (
-                    "MoviePy error: failed to read the duration of file '%s'.\n"
-                    "Here are the file infos returned by ffmpeg:\n\n%s"
+
+                start_match = re.search(r"start: (\d+\.?\d+)", data)
+                self.result["start"] = (
+                    float(start_match.group(1)) if start_match else None
                 )
-                % (self.filename, self.infos)
+            else:
+                if "metadata" not in self.result:
+                    self.result["metadata"] = {}
+
+                self.result["metadata"][key] = data
+
+        # For input direct metadata blocks, add meta to results
+        for child in root.childs:
+            if child.type in ("metadata", "side_data"):
+                for key, data in child.data.items():
+                    if "metadata" not in self.result:
+                        self.result["metadata"] = {}
+
+                    self.result["metadata"][key] = data
+
+        if self.video_stream:
+            self.result["video_found"] = True
+            self.result["video_size"] = self.video_stream.data.get("size", None)
+            self.result["video_bitrate"] = self.video_stream.data.get("bitrate", None)
+            self.result["video_fps"] = self.video_stream.data["fps"]
+            self.result["video_codec_name"] = self.video_stream.data.get(
+                "codec_name", None
             )
+            self.result["video_profile"] = self.video_stream.data.get("profile", None)
+            for child in self.video_stream.childs:
+                if child.type in ("metadata", "side_data"):
+                    for key, data in child.data.items():
+                        if key in ("rotate", "displaymatrix"):
+                            self.result["video_rotation"] = data
+
+        if self.audio_stream:
+            self.result["audio_found"] = True
+            self.result["audio_fps"] = self.audio_stream.data["fps"]
+            self.result["audio_bitrate"] = self.audio_stream.data["bitrate"]
 
-    def parse_metadata_field_value(
-        self,
-        line,
-    ):
-        """Returns a tuple with a metadata field-value pair given a ffmpeg `-i`
-        command output line.
-        """
-        if line.startswith("Ambient Viewing Environment, ") or line.startswith(
-            "Content Light Level Metadata, "
-        ) or line.startswith("Mastering Display Metadata, "):
-            field = line.split(",")[0]
-            value = line.split(",")[1:]
-            return (field.strip(" "), ",".join(value).strip(" "))
-
-        info = line.split(":", 1)
-        if len(info) == 2:
-            raw_field, raw_value = info
-            return (raw_field.strip(" "), raw_value.strip(" "))
+        if self.result["video_found"] and self.check_duration:
+            if "duration" not in self.result:
+                raise self.ParseDurationError()
+
+            self.result["video_duration"] = self.result["duration"]
+            self.result["video_n_frames"] = int(
+                self.result["duration"] * self.result.get("video_fps", 0)
+            )
         else:
-            return ("", "")
+            self.result["video_n_frames"] = 0
+            self.result["video_duration"] = 0.0
 
-    def video_metadata_type_casting(self, field, value):
-        """Cast needed video metadata fields to other types than the default str."""
-        if field == "rotate":
-            return (field, float(value))
+        self._populate_inputs(root=root)
 
-        elif field == "displaymatrix":
-            match = re.search(r"[-+]?\d+(\.\d+)?", value)
-            if match:
-                # We must multiply by -1 because displaymatrix return info
-                # about how to rotate to show video, not about video rotation
-                return (field, float(match.group()) * -1)
+    def _populate_inputs(self, root: InfoBlock):
+        """Forge inputs for compatibility with old versions, not used anywhere though"""
+        for child in root.childs:
+            if child.type == "stream":
+                if "streams" not in self.result["inputs"]:
+                    self.result["inputs"]["streams"] = []
 
-        return (field, value)
+                stream = child.data
+
+                for stream_child in child.childs:
+                    if stream_child.type == "metadata":
+                        stream["metadata"] = stream_child.data
+                    elif stream_child.type == "side_data":
+                        stream["side_data"] = stream_child.data
+
+                self.result["inputs"]["streams"].append(stream)
+
+            elif child.type == "chapters":
+                for chapter in child.childs:
+                    if "chapters" not in self.result["inputs"]:
+                        self.result["inputs"]["chapters"] = []
+
+                    chap = chapter.data
+
+                    for chapter_child in chapter.childs:
+                        if chapter_child.type == "metadata":
+                            chap["metadata"] = chapter_child.data
+                        elif chapter_child.type == "side_data":
+                            chap["side_data"] = chapter_child.data
+
+                    self.result["inputs"]["chapters"].append(chap)
+
+            elif child.type == "metadata":
+                self.result["metadata"] = child.data
+
+        if self.audio_stream:
+            self.result["default_audio_input_number"] = self.audio_stream.data[
+                "input_number"
+            ]
+            self.result["default_audio_stream_number"] = self.audio_stream.data[
+                "stream_number"
+            ]
+
+        if self.video_stream:
+            self.result["default_video_input_number"] = self.video_stream.data[
+                "input_number"
+            ]
+            self.result["default_video_stream_number"] = self.video_stream.data[
+                "stream_number"
+            ]
+
+        if self.data_stream:
+            self.result["default_data_input_number"] = self.data_stream.data[
+                "input_number"
+            ]
+            self.result["default_data_stream_number"] = self.data_stream.data[
+                "stream_number"
+            ]
+
+    def parse(self):
+        """Parses the information returned by FFmpeg in stderr executing their binary
+        for a file with ``-i`` option and returns a dictionary with all data needed
+        by MoviePy.
+        """
+        try:
+            first_input = 0
+            for line in self.infos.splitlines():
+                if line.startswith("Input"):
+                    break
+                first_input += 1
+
+            root_block = self.InfoBlock(self.infos.splitlines()[first_input], 0)
+            self._extract_block(first_input, 0, root_block)
+            self._parse_blocks(root_block)
+            self.blocks = root_block
+            self.result["blocks"] = root_block
+            return self.result
+        except self.ParseDimensionError:
+            raise IOError(
+                (
+                    "MoviePy error: failed to read video dimensions in"
+                    " file '%s'.\nHere are the file infos returned by"
+                    "ffmpeg:\n\n%s"
+                )
+                % (self.filename, self.infos)
+            )
+        except self.ParseDurationError:
+            raise IOError(
+                (
+                    "MoviePy error: failed to read video duration in"
+                    " file '%s'.\nHere are the file infos returned by"
+                    "ffmpeg:\n\n%s"
+                )
+                % (self.filename, self.infos)
+            )
 
 
 def ffmpeg_parse_infos(
@@ -860,21 +844,23 @@ def ffmpeg_parse_infos(
 
     Returns a dictionary with next fields:
 
+    - ``"audio_bitrate"``
+    - ``"audio_found"``
+    - ``"audio_fps"``
+    - ``"bitrate"``
     - ``"duration"``
-    - ``"metadata"``
     - ``"inputs"``
-    - ``"video_found"``
-    - ``"video_fps"``
-    - ``"video_n_frames"``
-    - ``"video_duration"``
+    - ``"metadata"``
+    - ``"start"``
     - ``"video_bitrate"``
-    - ``"video_metadata"``
-    - ``"audio_found"``
-    - ``"audio_fps"``
-    - ``"audio_bitrate"``
-    - ``"audio_metadata"``
     - ``"video_codec_name"``
+    - ``"video_duration"``
+    - ``"video_fps"``
+    - ``"video_found"``
+    - ``"video_n_frames"``
     - ``"video_profile"``
+    - ``"video_rotation"``
+    - ``"video_size"``
 
     Note that "video_duration" is slightly smaller than "duration" to avoid
     fetching the incomplete frames at the end, which raises an error.
@@ -940,4 +926,4 @@ def ffmpeg_parse_infos(
             raise IsADirectoryError(f"'{filename}' is a directory")
         elif not os.path.exists(filename):
             raise FileNotFoundError(f"'{filename}' not found")
-        raise IOError(f"Error passing `ffmpeg -i` command output:\n\n{infos}") from exc
+        raise IOError(f"Error passing `ffmpeg -i` command output: \n\n{infos}") from exc
diff --git a/moviepy/video/io/ffmpeg_tools.py b/moviepy/video/io/ffmpeg_tools.py
index 3d27147..b9ea29c 100644
--- a/moviepy/video/io/ffmpeg_tools.py
+++ b/moviepy/video/io/ffmpeg_tools.py
@@ -4,7 +4,6 @@ import os
 import re
 import subprocess
 from pathlib import Path
-from typing import Union
 
 from moviepy.config import FFMPEG_BINARY, FFPLAY_BINARY
 from moviepy.decorators import convert_parameter_to_seconds, convert_path_to_string
@@ -213,7 +212,8 @@ def ffmpeg_stabilize_video(
     subprocess_call(cmd, logger=logger)
 
 
-def ffmpeg_copy(input_file: Union[str, Path], output_file: Union[str, Path]):
+@convert_path_to_string(("input_file", "output_file"))
+def ffmpeg_copy(input_file, output_file):
     """
     Re-mix a video file using ffmpeg.
     This may fix issues with corrupted video file.
@@ -235,13 +235,11 @@ def ffmpeg_copy(input_file: Union[str, Path], output_file: Union[str, Path]):
     if not input_path.exists():
         raise FileNotFoundError(f"Input file '{input_file}' not found.")
 
-    if output_path.exists():
-        raise FileExistsError(f"Output file '{output_file}' already exists.")
-
     try:
         # Construct the ffmpeg command
         command = [
             FFMPEG_BINARY,
+            "-y",
             "-i",
             str(input_path),
             "-c",


Test patch:
diff --git a/tests/test_ffmpeg_reader.py b/tests/test_ffmpeg_reader.py
index df7468d..e3a1956 100644
--- a/tests/test_ffmpeg_reader.py
+++ b/tests/test_ffmpeg_reader.py
@@ -117,16 +117,16 @@ def test_ffmpeg_parse_infos_decode_file(decode_file, expected_duration):
     assert len(d["inputs"]) == 1
 
     # check streams
-    streams = d["inputs"][0]["streams"]
+    streams = d["inputs"]["streams"]
     assert len(streams) == 2
-    assert streams[0]["stream_type"] == "video"
+    assert streams[0]["stream_type_lower"] == "video"
     assert streams[0]["stream_number"] == 0
     assert streams[0]["fps"] == 24
     assert streams[0]["size"] == [1280, 720]
     assert streams[0]["default"] is True
     assert streams[0]["language"] is None
 
-    assert streams[1]["stream_type"] == "audio"
+    assert streams[1]["stream_type_lower"] == "audio"
     assert streams[1]["stream_number"] == 1
     assert streams[1]["fps"] == 44100
     assert streams[1]["default"] is True
@@ -173,10 +173,10 @@ def test_ffmpeg_parse_infos_multiple_audio_streams(util, mono_wave):
 
     # number of inputs and streams
     assert len(d["inputs"]) == 1
-    assert len(d["inputs"][0]["streams"]) == 2
+    assert len(d["inputs"]["streams"]) == 2
 
-    default_stream = d["inputs"][0]["streams"][0]
-    ignored_stream = d["inputs"][0]["streams"][1]
+    default_stream = d["inputs"]["streams"][0]
+    ignored_stream = d["inputs"]["streams"][1]
 
     # default, only the first
     assert default_stream["default"]
@@ -189,8 +189,8 @@ def test_ffmpeg_parse_infos_multiple_audio_streams(util, mono_wave):
     assert ignored_stream["input_number"] == 0
 
     # stream type
-    assert default_stream["stream_type"] == "audio"
-    assert ignored_stream["stream_type"] == "audio"
+    assert default_stream["stream_type_lower"] == "audio"
+    assert ignored_stream["stream_type_lower"] == "audio"
 
     # cleanup
     for filepath in [clip_440_filepath, clip_880_filepath, multiple_streams_filepath]:
@@ -269,8 +269,8 @@ def test_ffmpeg_parse_infos_metadata(util, mono_wave):
 
     # assert streams metadata
     streams = {"audio": None, "video": None}
-    for stream in d["inputs"][0]["streams"]:
-        streams[stream["stream_type"]] = stream
+    for stream in d["inputs"]["streams"]:
+        streams[stream["stream_type_lower"]] = stream
 
     for stream_type, stream in streams.items():
         for field, value in metadata[stream_type].items():
@@ -289,7 +289,7 @@ def test_ffmpeg_parse_infos_chapters():
     """Check that `ffmpeg_parse_infos` can parse chapters with their metadata."""
     d = ffmpeg_parse_infos("media/sintel_with_14_chapters.mp4")
 
-    chapters = d["inputs"][0]["chapters"]
+    chapters = d["inputs"]["chapters"]
 
     num_chapters_expected = 14
 
@@ -315,10 +315,10 @@ def test_ffmpeg_parse_infos_metadata_with_attached_pic():
     assert d["audio_fps"] == 44100
 
     assert len(d["inputs"]) == 1
-    streams = d["inputs"][0]["streams"]
+    streams = d["inputs"]["streams"]
     assert len(streams) == 2
-    assert streams[0]["stream_type"] == "audio"
-    assert streams[1]["stream_type"] == "video"
+    assert streams[0]["stream_type_lower"] == "audio"
+    assert streams[1]["stream_type_lower"] == "video"
 
     assert len(d["metadata"].keys()) == 7
 
@@ -386,20 +386,20 @@ def test_ffmpeg_parse_infos_multiline_metadata():
                     :
                     : <?xpacket end="w"?>
   Duration: 00:02:10.67, start: 0.000000, bitrate: 26287 kb/s
-    Stream #0:0(eng): Video: mjpeg 768x576 26213 kb/s, 24 fps, 24 tbr (default)
+  Stream #0:0(eng): Video: mjpeg 768x576 26213 kb/s, 24 fps, 24 tbr (default)
     Metadata:
       creation_time   : 2015-09-14 14:57:32
       handler_name    : Foo
                       : Bar
       encoder         : Photo - JPEG
       timecode        : 00:00:00:00
-    Stream #0:1(eng): Audio: aac (mp4a / 0x6), 44100 Hz, mono, fltp, 64 kb/s (default)
+  Stream #0:1(eng): Audio: aac (mp4a / 0x6), 44100 Hz, mono, fltp, 64 kb/s (default)
     Metadata:
       creation_time   : 2015-09-14 14:57:33
       handler_name    : Bar
                       : Foo
       timecode        : 00:00:00:00
-    Stream #0:2(eng): Data: none (tmcd / 0x64636D74) (default)
+  Stream #0:2(eng): Data: none (tmcd / 0x64636D74) (default)
     Metadata:
       creation_time   : 2015-09-14 14:58:24
       handler_name    : Baz
@@ -472,7 +472,7 @@ stDim:unit="pixel"/>
     # streams
     assert len(d["inputs"]) == 1
 
-    streams = d["inputs"][0]["streams"]
+    streams = d["inputs"]["streams"]
     assert len(streams) == 3
 
     # video stream
@@ -481,7 +481,7 @@ stDim:unit="pixel"/>
     assert streams[0]["input_number"] == 0
     assert streams[0]["language"] == "eng"
     assert streams[0]["stream_number"] == 0
-    assert streams[0]["stream_type"] == "video"
+    assert streams[0]["stream_type_lower"] == "video"
     assert streams[0]["size"] == [768, 576]
 
     assert streams[0]["metadata"]["creation_time"] == "2015-09-14 14:57:32"
@@ -495,7 +495,7 @@ stDim:unit="pixel"/>
     assert streams[1]["input_number"] == 0
     assert streams[1]["language"] == "eng"
     assert streams[1]["stream_number"] == 1
-    assert streams[1]["stream_type"] == "audio"
+    assert streams[1]["stream_type_lower"] == "audio"
 
     assert streams[1]["metadata"]["creation_time"] == "2015-09-14 14:57:33"
     assert streams[1]["metadata"]["timecode"] == "00:00:00:00"
@@ -506,7 +506,7 @@ stDim:unit="pixel"/>
     assert streams[2]["input_number"] == 0
     assert streams[2]["language"] == "eng"
     assert streams[2]["stream_number"] == 2
-    assert streams[2]["stream_type"] == "data"
+    assert streams[2]["stream_type_lower"] == "data"
 
     assert streams[2]["metadata"]["creation_time"] == "2015-09-14 14:58:24"
     assert streams[2]["metadata"]["timecode"] == "00:00:00:00"
@@ -547,7 +547,7 @@ At least one output file must be specified"""
     d = FFmpegInfosParser(infos, "clip.mp4").parse()
 
     assert d
-    assert len(d["inputs"][0]["streams"]) == 1
+    assert len(d["inputs"]["streams"]) == 1
 
 
 def test_stream_square_brackets():
@@ -561,9 +561,9 @@ At least one output file must be specified"""
     d = FFmpegInfosParser(infos, "clip.mp4").parse()
 
     assert d
-    assert len(d["inputs"][0]["streams"]) == 2
-    assert d["inputs"][0]["streams"][0]["language"] is None
-    assert d["inputs"][0]["streams"][1]["language"] is None
+    assert len(d["inputs"]["streams"]) == 2
+    assert d["inputs"]["streams"][0]["language"] is None
+    assert d["inputs"]["streams"][1]["language"] is None
 
 
 def test_stream_square_brackets_and_language():
@@ -577,9 +577,9 @@ At least one output file must be specified"""
     d = FFmpegInfosParser(infos, "clip.mp4").parse()
 
     assert d
-    assert len(d["inputs"][0]["streams"]) == 2
-    assert d["inputs"][0]["streams"][0]["language"] == "eng"
-    assert d["inputs"][0]["streams"][1]["language"] is None
+    assert len(d["inputs"]["streams"]) == 2
+    assert d["inputs"]["streams"][0]["language"] == "eng"
+    assert d["inputs"]["streams"][1]["language"] is None
 
 
 def test_stream_missing_audio_bitrate():
@@ -593,7 +593,7 @@ At least one output file must be specified"""
     d = FFmpegInfosParser(infos, "clip.mp4").parse()
 
     assert d
-    assert len(d["inputs"][0]["streams"]) == 2
+    assert len(d["inputs"]["streams"]) == 2
     assert d["audio_found"]
     assert d["audio_bitrate"] is None