Convert the downloaded video to M4A and save to disk

For the first time, the downloaded video is converted (again) into M4A in order to be able to normalize the audio and adapt the downloaded file for removing the extra streams not needed (video)

YouTubeMDBot/__init__.py

@@ -27,6 +27,7 @@
 from .audio import FFmpegMP3
 from .audio import FFmpegOGG
 from .audio import FFmpegOpener
+from .audio import FFmpegM4A
 from .audio import FPCalc
 from .audio import ffmpeg_available
 
@@ -41,6 +42,7 @@
 
 from .downloader import YouTubeDownloader
 from .downloader import MultipleYouTubeDownloader
+from .downloader import M4AYouTubeDownloader
 
 from .metadata import AudioMetadata
 from .metadata import MetadataIdentifier

YouTubeMDBot/audio/__init__.py

@@ -16,5 +16,6 @@
 from ..audio.ffmpeg import FFmpegOpener
 from ..audio.ffmpeg import FFmpegOGG
 from ..audio.ffmpeg import FFmpegMP3
+from ..audio.ffmpeg import FFmpegM4A
 from ..audio.ffmpeg import ffmpeg_available
 from ..audio.fpcalc import FPCalc

YouTubeMDBot/audio/ffmpeg.py

@@ -21,6 +21,7 @@
 
 from ..constants import FFMPEG_OPENER
 from ..constants import FFMPEG_CONVERTER
+from ..constants import FFMPEG_VOLUME
 
 
 def ffmpeg_available() -> bool:
@@ -99,6 +100,16 @@ def get_extra(self) -> bytes:
         """
         return self.__err
 
+    def get_volume(self) -> float:
+        """
+        Gets the maximum volume of the data input.
+        :return: the volume.
+        """
+        command = FFMPEG_VOLUME
+        proc = Popen(command, stdout=PIPE, stderr=PIPE, stdin=PIPE, shell=True)
+        out, err = proc.communicate(self._data)
+        return float(out.decode("utf-8")) if out.decode("utf-8") != '' else 0.0
+
 
 class FFmpegOpener(FFmpeg):
     """
@@ -169,3 +180,26 @@ def convert(self) -> int:
         command.append("ogg")
         command.append("-")
         return self.process()
+
+
+class FFmpegM4A(FFmpegExporter):
+    def __init__(self, data: bytes, filename: str, bitrate: str = None):
+        super().__init__(data, bitrate)
+        self.filename = filename
+
+    def convert(self) -> int:
+        command = super().get_command()
+        vol = self.get_volume() * -1
+        command.append("-af")
+        command.append(f"volume={vol}dB")
+        if self._bitrate:
+            command.append("-b:a")
+            command.append(self._bitrate)
+        command.append("-c:a")
+        command.append("aac")
+        command.append("-movflags")
+        command.append("faststart")
+        command.append("-f")
+        command.append("ipod")
+        command.append(self.filename)
+        return self.process()

YouTubeMDBot/constants/__init__.py

@@ -20,6 +20,7 @@
 from ..constants.app_constants import PROGRAM_ARGS
 from ..constants.app_constants import FFMPEG_OPENER
 from ..constants.app_constants import FFMPEG_CONVERTER
+from ..constants.app_constants import FFMPEG_VOLUME
 from ..constants.app_constants import DB_HOST
 from ..constants.app_constants import DB_NAME
 from ..constants.app_constants import DB_PASSWORD

YouTubeMDBot/constants/app_constants.py

@@ -20,7 +20,8 @@
 
 PROGRAM_ARGS = sys.argv
 # YouTube DL options
-YDL_CLI_OPTIONS = ["youtube-dl", "--format", "bestaudio[ext=m4a]", "--quiet", "--output",
+YDL_CLI_OPTIONS = ["youtube-dl", "--format", "bestaudio[ext=m4a]", "--quiet",
+                   "--output",
                    "-"]
 
 # FPCalc command
@@ -41,8 +42,10 @@
 
 # FFmpeg commands
 FFMPEG_OPENER = "ffmpeg -i - -f s16le -".split(" ")
-FFMPEG_CONVERTER = ["ffmpeg", "-i", "-", "-vn", "-map_metadata", "0",
-                    "-movflags", "use_metadata_tags"]
+FFMPEG_CONVERTER = ["ffmpeg", "-y", "-i", "-", "-vn", "-map_metadata", "0",
+                    "-map", "a:a", "-movflags", "use_metadata_tags"]
+FFMPEG_VOLUME = "ffmpeg -i - -af \"volumedetect\" -f null /dev/null 2>&1 | " \
+                "grep \"max_volume\" | awk -F': ' '{print $2}' | cut -d ' ' -f1"
 
 MAX_PROCESS = cpu_count()
 

YouTubeMDBot/downloader/__init__.py

@@ -15,3 +15,4 @@
 #    along with this program. If not, see <http://www.gnu.org/licenses/>.
 from ..downloader.youtube_downloader import MultipleYouTubeDownloader
 from ..downloader.youtube_downloader import YouTubeDownloader
+from ..downloader.youtube_downloader import M4AYouTubeDownloader

YouTubeMDBot/downloader/youtube_downloader.py

@@ -17,8 +17,10 @@
 from typing import Any
 from typing import Callable
 from typing import Tuple
+from tempfile import NamedTemporaryFile
 
 from .. import ThreadPoolBase
+from .. import FFmpegM4A
 from ..constants.app_constants import YDL_CLI_OPTIONS
 
 
@@ -63,6 +65,27 @@ def get_url(self) -> str:
         return self.__url
 
 
+class M4AYouTubeDownloader(YouTubeDownloader):
+    def __init__(self, url: str, bitrate: str = None):
+        super().__init__(url)
+        self.user_bitrate = bitrate
+
+    def download(self) -> Tuple[BytesIO, bytes]:
+        io, data = super().download()
+        m4a_file = NamedTemporaryFile(suffix=".m4a")
+        m4a_converter = FFmpegM4A(data=data,
+                                  filename=m4a_file.name,
+                                  bitrate=self.user_bitrate)
+        ret = m4a_converter.convert()
+        if ret != 0:
+            raise RuntimeError("ffmpeg is unable to convert file - output: "
+                               + m4a_converter.get_extra().decode("utf-8"))
+        with open(m4a_file.name, "rb") as out_m4a:
+            m4a_data = out_m4a.read()
+        m4a_file.close()
+        return BytesIO(m4a_data), m4a_data
+
+
 class MultipleYouTubeDownloader(ThreadPoolBase):
     def __new__(cls,
                 max_processes: int = 4,

YouTubeMDBot/tests/m4adownloader.py

@@ -0,0 +1,27 @@
+import unittest
+import mutagen
+
+from YouTubeMDBot.downloader import M4AYouTubeDownloader
+from YouTubeMDBot.audio import FFmpegM4A
+
+
+class MyTestCase(unittest.TestCase):
+    def test_download(self):
+        dl = M4AYouTubeDownloader(
+            url="https://www.youtube.com/watch?v=s6VaeFCxta8",
+            bitrate="128k")
+        io, data = dl.download()
+        with open("outex.m4a", "wb") as of:
+            of.write(data)
+        print(mutagen.File(io).pprint())
+        io.seek(0)
+        return io, data
+
+    def test_normalization(self):
+        io, data = self.test_download()
+        ctr = FFmpegM4A(data, "filename")
+        assert ctr.get_volume() == 0.0
+
+
+if __name__ == '__main__':
+    unittest.main()