feat: generate m4b audiobook from files

Dockerfile

@@ -4,7 +4,7 @@ WORKDIR /opt/kobodl/src
 ENV PATH="/opt/kobodl/local/venv/bin:$PATH"
 ENV VIRTUAL_ENV="/opt/kobodl/local/venv"
 
-RUN apk add --no-cache gcc libc-dev libffi-dev
+RUN apk add --no-cache gcc libc-dev libffi-dev ffmpeg
 ADD https://install.python-poetry.org /install-poetry.py
 RUN POETRY_VERSION=1.1.7 POETRY_HOME=/opt/kobodl/local python /install-poetry.py
 

kobodl/actions.py

@@ -1,6 +1,7 @@
 import json
 import os
 import platform
+import subprocess
 from typing import List, TextIO, Tuple, Union
 
 import click
@@ -51,12 +52,16 @@ def __MakeFileNameForBook(bookMetadata: dict, formatStr: str) -> str:
     fileName = ''
     author = __SanitizeString(__GetBookAuthor(bookMetadata))
     title = __SanitizeString(bookMetadata['Title'])
+    bookSeries = __SanitizeString(bookMetadata.get('Series', {}).get('Name', ''))
+    bookSeriesNumber = __SanitizeString(bookMetadata.get('Series', {}).get('Number', ''))
 
     return formatStr.format_map(
         {
             **bookMetadata,
             'Author': author,
             'Title': title,
+            'Series': bookSeries,
+            'SeriesNumber': bookSeriesNumber,
             # Append a portion of revisionId to prevent name collisions.
             'ShortRevisionId': bookMetadata['RevisionId'][:8],
         }
@@ -144,6 +149,90 @@ def __GetBookList(kobo: Kobo, listAll: bool, exportFile: Union[TextIO, None]) ->
     return rows
 
 
+def __CreateM4BFile(outputPath: str, filename: str, bookMetadata: dict) -> None:
+    # Check if ffmpeg is installed
+    try:
+        subprocess.run(["ffmpeg", "-version"], check=True)
+    except subprocess.CalledProcessError:
+        click.echo("ffmpeg is not installed. Please install ffmpeg and try again.")
+        return
+
+    # Concatenate all mp3 files into one file
+    subprocess.run(
+        [
+            "ffmpeg",
+            "-f",
+            "concat",
+            "-safe",
+            "0",
+            "-i",
+            "files.txt",
+            "-c",
+            "copy",
+            "-y",
+            "build_01_concat.mp3",
+        ],
+        check=True,
+        cwd=outputPath,
+    )
+
+    # Add cover
+    subprocess.run(
+        [
+            "ffmpeg",
+            "-i",
+            "build_01_concat.mp3",
+            "-i",
+            "cover.jpg",
+            "-c",
+            "copy",
+            "-map",
+            "0",
+            "-map",
+            "1",
+            "-y",
+            "build_02_cover.mp3",
+        ],
+        check=True,
+        cwd=outputPath,
+    )
+
+    # Convert mp3 to m4a
+    subprocess.run(
+        ["ffmpeg", "-y", "-i", "build_02_cover.mp3", "-c:v", "copy", "build_03_m4a.m4a"],
+        check=True,
+        cwd=outputPath,
+    )
+
+    # Add metadata to the m4a file and convert to m4b
+    subprocess.run(
+        [
+            "ffmpeg",
+            "-i",
+            "build_03_m4a.m4a",
+            "-i",
+            "metadata.txt",
+            "-map",
+            "0",
+            "-map_metadata",
+            "1",
+            "-c",
+            "copy",
+            "-y",
+            f'{filename}.m4b',
+        ],
+        check=True,
+        cwd=outputPath,
+    )
+
+    # Remove all build_* files using python
+    for f in os.listdir(outputPath):
+        if f.startswith("build_"):
+            os.remove(os.path.join(outputPath, f))
+
+    return os.path.join(outputPath, f'{bookMetadata["Title"]}.m4b')
+
+
 def ListBooks(users: List[User], listAll: bool, exportFile: Union[TextIO, None]) -> List[Book]:
     '''list all books currently in the account'''
     for user in users:
@@ -174,6 +263,7 @@ def GetBookOrBooks(
     outputPath: str,
     formatStr: str = r'{Author} - {Title} {ShortRevisionId}',
     productId: str = '',
+    generateAudiobook: bool = False,
 ) -> Union[None, str]:
     """
     download 1 or all books to file
@@ -203,14 +293,26 @@ def GetBookOrBooks(
             click.echo('Skipping subscribtion entity')
             continue
 
+        # Save metadata to JSON file
+        with open(os.path.join(outputPath, f'{bookMetadata["Title"]}.json'), 'w') as f:
+            f.write(json.dumps(bookMetadata, indent=2))
+
         fileName = __MakeFileNameForBook(bookMetadata, formatStr)
         if book_type == BookType.EBOOK:
             # Audiobooks go in sub-directories
             # but epub files go directly in outputPath
             fileName += '.epub'
         outputFilePath = os.path.join(outputPath, fileName)
 
-        if not productId and os.path.exists(outputFilePath):
+        if (
+            not productId
+            and os.path.exists(outputFilePath)
+            and (
+                generateAudiobook
+                and book_type == BookType.AUDIOBOOK
+                and os.path.exists(os.path.join(outputFilePath, f'{fileName}.m4b'))
+            )
+        ):
             # when downloading ALL books, skip books we've downloaded before
             click.echo(f'Skipping already downloaded book {outputFilePath}')
             continue
@@ -242,6 +344,13 @@ def GetBookOrBooks(
                     err=True,
                 )
 
+        # Create final audiobook file
+        if book_type == BookType.AUDIOBOOK and generateAudiobook:
+            try:
+                __CreateM4BFile(outputFilePath, fileName, bookMetadata)
+            except Exception as e:
+                click.echo(f'Failed to create audiobook file: {str(e)}', err=True)
+
         if productId:
             # TODO: support audiobook downloads from web
             return outputFilePath

kobodl/commands/book.py

@@ -45,9 +45,18 @@ def book():
     default=r'{Author} - {Title} {ShortRevisionId}',
     help=r"default: '{Author} - {Title} {ShortRevisionId}'",
 )
+@click.option('--generate-audiobook', is_flag=True, help='generate m4b audiobook bundle file')
 @click.argument('product-id', nargs=-1, type=click.STRING)
 @click.pass_obj
-def get(ctx, user, output_dir: Path, get_all: bool, format_str: str, product_id: List[str]):
+def get(
+    ctx,
+    user,
+    output_dir: Path,
+    get_all: bool,
+    format_str: str,
+    generate_audiobook: bool,
+    product_id: List[str],
+):
     if len(Globals.Settings.UserList.users) == 0:
         click.echo('error: no users found.  Did you `kobodl user add`?', err=True)
         exit(1)
@@ -77,10 +86,18 @@ def get(ctx, user, output_dir: Path, get_all: bool, format_str: str, product_id:
 
     os.makedirs(output_dir, exist_ok=True)
     if get_all:
-        actions.GetBookOrBooks(usercls, output_dir, formatStr=format_str)
+        actions.GetBookOrBooks(
+            usercls, output_dir, formatStr=format_str, generateAudiobook=generate_audiobook
+        )
     else:
         for pid in product_id:
-            actions.GetBookOrBooks(usercls, output_dir, formatStr=format_str, productId=pid)
+            actions.GetBookOrBooks(
+                usercls,
+                output_dir,
+                formatStr=format_str,
+                productId=pid,
+                generateAudiobook=generate_audiobook,
+            )
 
 
 @book.command(name='list', help='list books')