Merge pull request #18 from MK2112/12-multi-threaded-multi-file-conversion

multi threaded multi file conversion

Author: MK2112

README.md

@@ -17,6 +17,7 @@
 - Offering a flexible **Command Line Interface**, a **Web Interface**, and a **Graphical User Interface (GUI)**
 - Control output quality and video framerate for the conversion
 - Automatically monitor a **"dropzone" directory** for new files and process them as they are dropped
+- Fast batch conversion with configurable **parallel processing** for audio and video conversions
 
 ## Setup
 1. **Clone/Download**:
@@ -96,6 +97,7 @@ This is the most detailed way to use `any_to_any.py`. You can structure a comman
 | `-r` or </br>`--recursive`   | Recursively process all input files in subdirectories from the input directory. Outputs by default will be placed in their respective subdirectory, unless different output path provided. |
 | `-z` or </br>`--dropzone`    | While running, a specified directory will be monitored for new files. When a file is added, it will be converted to the specified format, saved in the output directory and deleted from the input directory. |
 | `-fps` or</br>`--framerate`  | Set the framerate (fps) when converting to a movie format or codec; default maintains input fps. |
+| `--workers`                   | Set the maximum number of parallel worker threads for per-file conversions (`1` to `cpu_count - 1` are supported). Defaults to `1`. |
 
 ### Single File Processing
 Convert a WEBP file to PNG:
@@ -202,6 +204,23 @@ python any_to_any.py -i -1 /path/to/folder1 -2 /path/to/folder2 -o /path/to/outp
 ```
 Omitting the `-a`/`--across` parameter will execute merges or concatenations seperately, per each input directory.
 
+#### Parallel Processing
+- Per-file conversions (e.g., audio-to-audio, movie-to-movie, gif-to-video) are processed in parallel.
+- Override default (`1` worker) by setting `--workers N` where `N` can be any integer from `1` to `cpu_count - 1`.
+
+Directory with many audio files:
+```bash
+python any_to_any.py -i /path/to/folder -f mp3 --workers 4
+```
+Explicit multiple files:
+```bash
+python any_to_any.py -i /path/to/a.wav /path/to/b.flac /path/to/c.ogg -f mp3 --workers 4 -o /path/to/output_dir
+```
+Recursive scan (include subdirectories):
+```bash
+python any_to_any.py -i /path/to/input_dir -f mp3 --workers 4 --recursive
+```
+
 ## Supported Formats
 **Audio:** MP2, MP3, FLAC, AAC, AC3, DTS, OGG, OGA, WMA, WAV, M4A, AIFF, WEBA, MKA, WV, CAF, TTA, M4B, EAC3, SPX, AU, OPUS, M3U8, W64, MLP, ADTS, SBC, THD<br><br>
 **Image:** JPEG, PNG, GIF, BMP, WEBP, TIFF, TGA, EPS, PS, ICO, EPS, JPEG2000, IM, PCX, PPM<br><br>

any_to_any.py

@@ -129,6 +129,13 @@
         type=str,
         required=False,
     )
+    parser.add_argument(
+        "--workers",
+        help="Maximum worker threads to use for per-file conversions (default: 1)",
+        type=int,
+        default=1,
+        required=False,
+    )
 
     args = vars(parser.parse_args())
 
@@ -168,4 +175,5 @@
             recursive=args["recursive"],
             dropzone=args["dropzone"],
             language=args["language"],
+            workers=args["workers"],
         )

core/audio_converter.py

@@ -2,6 +2,7 @@
 import utils.language_support as lang
 from utils.category import Category
 from moviepy import AudioFileClip, VideoFileClip
+from concurrent.futures import ThreadPoolExecutor, as_completed
 
 
 class AudioConverter:
@@ -26,87 +27,141 @@ def to_audio(
         output: str,
         delete: str,
     ) -> None:
-        # Audio to audio conversion
-        for audio_path_set in file_paths[Category.AUDIO]:
+        try:
+            # Decide worker count with env A2A_MAX_WORKERS if set
+            env_workers = int(os.environ.get("A2A_MAX_WORKERS", "1"))
+            env_workers = 1 if env_workers < 1 else env_workers
+            env_workers = os.cpu_count() - 1 if env_workers >= os.cpu_count() else env_workers
+        except ValueError:
+            # If this variable doesn't exist, flag wasn't invoked: Default to 1
+            env_workers = 1
+
+        # Helper to convert a single audio file
+        def _convert_audio_file(audio_path_set: tuple):
             if audio_path_set[2] == format:
-                continue
-            audio = AudioFileClip(self.file_handler.join_back(audio_path_set))
-            # If recursive, create file outright where its source was found
-            if not recursive or input != output:
-                out_path = os.path.abspath(
-                    os.path.join(output, f"{audio_path_set[1]}.{format}")
-                )
-            else:
-                out_path = os.path.abspath(
-                    os.path.join(audio_path_set[0], f"{audio_path_set[1]}.{format}")
-                )
-            # Write audio to file
+                return None
+            audio = None
+            out_path = None
             try:
-                audio.write_audiofile(
-                    out_path,
-                    codec=codec,
-                    bitrate=bitrate,
-                    fps=audio.fps,
-                    logger=self.prog_logger,
-                )
-            except Exception as _:
-                self.event_logger.info(
-                    f"\n\n[!] {lang.get_translation('error', self.locale)}: {lang.get_translation('source_rate_incompatible', self.locale).replace('[format]', f'{format}')}\n"
-                )
-                audio.write_audiofile(
-                    out_path,
-                    codec=codec,
-                    bitrate=bitrate,
-                    fps=48000,
-                    logger=self.prog_logger,
-                )
-            audio.close()
-            self.file_handler.post_process(audio_path_set, out_path, delete)
-
-        # Movie to audio conversion
-        for movie_path_set in file_paths[Category.MOVIE]:
-            out_path = os.path.abspath(
-                os.path.join(output, f"{movie_path_set[1]}.{format}")
-            )
-
-            if self.file_handler.has_visuals(movie_path_set):
-                video = VideoFileClip(
-                    self.file_handler.join_back(movie_path_set),
-                    audio=True,
-                    fps_source="tbr",
-                )
-                audio = video.audio
-                # Check if audio was found
-                if audio is None:
-                    self.event_logger.info(
-                        f"[!] {lang.get_translation('no_audio', self.locale).replace('[path]', f'"{self.file_handler.join_back(movie_path_set)}"')} - {lang.get_translation('skipping', self.locale)}\n"
+                audio = AudioFileClip(self.file_handler.join_back(audio_path_set))
+                # If recursive, create file outright where its source was found
+                if not recursive or input != output:
+                    out_path = os.path.abspath(
+                        os.path.join(output, f"{audio_path_set[1]}.{format}")
+                    )
+                else:
+                    out_path = os.path.abspath(
+                        os.path.join(audio_path_set[0], f"{audio_path_set[1]}.{format}")
                     )
-                    video.close()
-                    continue
-
-                audio.write_audiofile(
-                    out_path,
-                    codec=codec,
-                    bitrate=bitrate,
-                    logger=self.prog_logger,
-                )
-
-                audio.close()
-                video.close()
-            else:
                 try:
-                    # AudioFileClip works for audio-only video files
-                    audio = AudioFileClip(self.file_handler.join_back(movie_path_set))
                     audio.write_audiofile(
                         out_path,
                         codec=codec,
                         bitrate=bitrate,
+                        fps=audio.fps,
                         logger=self.prog_logger,
                     )
-                    audio.close()
                 except Exception as _:
                     self.event_logger.info(
-                        f"[!] {lang.get_translation('audio_extract_fail', self.locale).replace('[path]', f'"{self.file_handler.join_back(movie_path_set)}"')} - {lang.get_translation('skipping', self.locale)}\n"
+                        f"\n\n[!] {lang.get_translation('error', self.locale)}: {lang.get_translation('source_rate_incompatible', self.locale).replace('[format]', f'{format}')}\n"
+                    )
+                    audio.write_audiofile(
+                        out_path,
+                        codec=codec,
+                        bitrate=bitrate,
+                        fps=48000,
+                        logger=self.prog_logger,
                     )
-                    continue
-            self.file_handler.post_process(movie_path_set, out_path, delete)
+                return (audio_path_set, out_path)
+            finally:
+                if audio is not None:
+                    audio.close()
+
+        audio_items = list(file_paths[Category.AUDIO])
+        if len(audio_items) <= 1:
+            # Fallback to sequential for 0/1 items
+            result = _convert_audio_file(audio_items[0]) if audio_items else None
+            if result is not None:
+                src, out_path = result
+                self.file_handler.post_process(src, out_path, delete)
+        else:
+            max_workers = env_workers if env_workers > 1 else 1
+            with ThreadPoolExecutor(max_workers=max_workers) as ex:
+                futures = [ex.submit(_convert_audio_file, a) for a in audio_items]
+                for fut in as_completed(futures):
+                    res = fut.result()
+                    if res is None:
+                        continue
+                    src, out_path = res
+                    self.file_handler.post_process(src, out_path, delete)
+
+        # Movie to audio conversion
+        def _extract_from_movie(movie_path_set: tuple):
+            out_path_local = os.path.abspath(
+                os.path.join(output, f"{movie_path_set[1]}.{format}")
+            )
+            video = None
+            audio = None
+            try:
+                if self.file_handler.has_visuals(movie_path_set):
+                    video = VideoFileClip(
+                        self.file_handler.join_back(movie_path_set),
+                        audio=True,
+                        fps_source="tbr",
+                    )
+                    audio = video.audio
+                    # Check if audio was found
+                    if audio is None:
+                        self.event_logger.info(
+                            f"[!] {lang.get_translation('no_audio', self.locale).replace('[path]', f'\"{self.file_handler.join_back(movie_path_set)}\"')} - {lang.get_translation('skipping', self.locale)}\n"
+                        )
+                        return None
+                    audio.write_audiofile(
+                        out_path_local,
+                        codec=codec,
+                        bitrate=bitrate,
+                        logger=self.prog_logger,
+                    )
+                else:
+                    try:
+                        # AudioFileClip works for audio-only video files
+                        audio = AudioFileClip(self.file_handler.join_back(movie_path_set))
+                        audio.write_audiofile(
+                            out_path_local,
+                            codec=codec,
+                            bitrate=bitrate,
+                            logger=self.prog_logger,
+                        )
+                    except Exception as _:
+                        self.event_logger.info(
+                            f"[!] {lang.get_translation('audio_extract_fail', self.locale).replace('[path]', f'\"{self.file_handler.join_back(movie_path_set)}\"')} - {lang.get_translation('skipping', self.locale)}\n"
+                        )
+                        return None
+                return (movie_path_set, out_path_local)
+            finally:
+                try:
+                    if audio is not None:
+                        audio.close()
+                except Exception:
+                    pass
+                try:
+                    if video is not None:
+                        video.close()
+                except Exception:
+                    pass
+
+        movie_items = list(file_paths[Category.MOVIE])
+        if len(movie_items) <= 1:
+            res = _extract_from_movie(movie_items[0]) if movie_items else None
+            if res is not None:
+                src, out_path = res
+                self.file_handler.post_process(src, out_path, delete)
+        else:
+            with ThreadPoolExecutor(max_workers=env_workers) as ex:
+                futures = [ex.submit(_extract_from_movie, m) for m in movie_items]
+                for fut in as_completed(futures):
+                    res = fut.result()
+                    if res is None:
+                        continue
+                    src, out_path = res
+                    self.file_handler.post_process(src, out_path, delete)

core/controller.py

@@ -238,10 +238,21 @@ def run(
         recursive: bool,
         dropzone: bool,
         language: str,
+        workers: int,
     ) -> None:
         # Main function, convert media files to defined formats
         # or merge or concatenate, according to the arguments
         input_paths = []
+        # Apply worker override via environment for converters
+        try:
+            if workers is not None:
+                # Workers flag was set, do some validation, hand it to env
+                os.environ["A2A_MAX_WORKERS"] = str(max(1, int(workers)))
+        except Exception:
+            # Ignore invalid values; converters will fall back to CPU-based default
+            pass
+
+        # Apply input paths
         input_path_args = (
             input_path_args
             if input_path_args is not None