Merge pull request #16 from MK2112/11-batch-pdf-splitting-merging

Integrate PDF splitting and merging

Author: MK2112

README.md

@@ -59,6 +59,7 @@ You can structure a command in three fundamental ways:
 | `-q` or </br>`--quality`     | Set output file quality, either `low`, `medium`, or `high`; default is same as input. |
 | `-m` or </br>`--merge`       | Merge movie file with equally named audio file to become its audio track. |
 | `-c` or </br>`--concat`      | Concatenate input files of the same type (images, audio, video) into one output file (e.g. `concatenated_video.mp4` for movie files, `concatenated_audio.mp3` for audio files). |
+| `-s` or </br>`--split`       | Split a PDF into multiple files, either by page count or page ranges, e.g. `1-2,3-5` or `10` or `1-3,2-6,8-end` or `1-5,rest`. |
 | `-a` or </br>`--across`      | Merge/Concatenate across directories when multiple directories are provided. |
 | `-w` or </br>`--web`         | Ignores all other arguments, starts browser + a web server at `http://localhost:5000`. |
 | `-d` or </br>`--delete`      | Delete input files after conversion. |
@@ -103,6 +104,11 @@ You may also convert the images contained in a DOCX file to an MP4 video:
 python any_to_any.py -i /path/to/file.docx -f mp4
 ```
 
+Split a PDF into multiple files, either by page count or page ranges, e.g. `1-2,3-5` or `10` or `1-3,2-6,8-end` or `1-5,rest`:
+```python
+python any_to_any.py -i /path/to/file.pdf -s 1-3,2-6,8-end
+```
+
 ### Directory Processing
 Directory Processing is useful when you want to work with multiple files in a directory
 

any_to_any.py

@@ -122,9 +122,26 @@
         type=str,
         required=False,
     )
+    parser.add_argument(
+        "-s",
+        "--split",
+        help=f"{lang.get_translation('split_help', controller.locale)}",
+        type=str,
+        required=False,
+    )
 
     args = vars(parser.parse_args())
 
+    if args["split"] and (args["merge"] or args["concat"]):
+        parser.error(
+            f"{lang.get_translation('split_merge_error', controller.locale)}"
+        )
+
+    if args["merge"] and args["concat"]:
+        parser.error(
+            f"{lang.get_translation('merge_concat_error', controller.locale)}"
+        )
+
     if args["web"]:
         # Check for web frontend request
         if os.name in ["nt"]:
@@ -139,6 +156,7 @@
             output=args["output"],
             framerate=args["framerate"],
             quality=args["quality"],
+            split=args["split"],
             merge=args["merge"],
             concat=args["concat"],
             delete=args["delete"],

core/controller.py

@@ -23,8 +23,6 @@
     clips_array,
 )
 
-# TODO: Add converter-wise tests
-
 class Controller:
     """
     Taking an input directory of files, convert them to a multitude of formats.
@@ -227,6 +225,7 @@ def run(
         output: str,
         framerate: int,
         quality: str,
+        split: str,
         merge: bool,
         concat: bool,
         delete: bool,
@@ -300,6 +299,8 @@ def run(
         self.merging = merge
         # Concatenate files of same type (img/movie/audio) back to back
         self.concatenating = concat
+        # Split files into smaller parts
+        self.page_ranges = split
 
         file_paths = {}
         was_none, found_files = False, False
@@ -347,7 +348,7 @@ def run(
                         f"[!] {lang.get_translation('error', self.locale)}: {lang.get_translation('dropzone_diff', self.locale)}",
                     )
                 self.event_logger.info(
-                    f"[+] {lang.get_translation('dropzone_active', self.locale)} {self.input}"
+                    f"[>] {lang.get_translation('dropzone_active', self.locale)} {self.input}"
                 )
                 self.watchdropzone(self.input)
                 return
@@ -475,6 +476,8 @@ def process_file_paths(self, file_paths: dict) -> None:
             self.merge(file_paths, getattr(self, "across", False))
         elif self.concatenating:
             self.concat(file_paths, self.target_format)
+        elif self.page_ranges is not None:
+            self.split(file_paths, self.page_ranges)
         else:
             # Handle unsupported formats here
             end_with_msg(
@@ -559,6 +562,25 @@ def handle_file_event(event_type: str, file_path: str) -> None:
             )
             raise
 
+    def split(self, file_paths: dict, page_ranges) -> None:
+        for doc_path_set in file_paths[Category.DOCUMENT]:
+            if hasattr(self.prog_logger, 'shared_progress_dict') and self.prog_logger.shared_progress_dict:
+                import threading
+                with threading.Lock():
+                    if self.prog_logger.job_id in self.prog_logger.shared_progress_dict:
+                        self.prog_logger.shared_progress_dict[self.prog_logger.job_id].update({
+                            'status': f'splitting {doc_path_set[1]}',
+                            'last_updated': time.time()
+                        })
+            if doc_path_set[2] == "pdf":
+                self.doc_converter.split_pdf(
+                    output=self.output,
+                    doc_path_set=doc_path_set,
+                    format='pdf',
+                    delete=self.delete,
+                    page_ranges=page_ranges
+                )
+
     def concat(self, file_paths: dict, format: str) -> None:
         # Concatenate files of same type (img/movie/audio) back to back
         # Concatenate audio files
@@ -634,7 +656,7 @@ def concat(self, file_paths: dict, format: str) -> None:
         if file_paths[Category.DOCUMENT] and (
             format is None or format in self._supported_formats[Category.DOCUMENT]
         ):
-            pdf_out_path = os.path.join(self.output, "concatenated_pdfs.pdf")
+            pdf_out_path = os.path.join(self.output, "concatenated.pdf")
             pdfs = sorted(
                 [
                     doc_path_set if doc_path_set[2] == "pdf" else None
@@ -737,7 +759,6 @@ def concat(self, file_paths: dict, format: str) -> None:
             f"[+] {lang.get_translation('concat_success', self.locale)}"
         )
 
-
     def merge(self, file_paths: dict, across: bool = False) -> None:
         # For movie files and equally named audio file, merge them together under same name
         # (movie with audio with '_merged' addition to name)

core/doc_converter.py

@@ -179,7 +179,7 @@ def to_subtitles(
             input_path = self.file_handler.join_back(movie_path_set)
             out_path = os.path.abspath(os.path.join(output, f"{movie_path_set[1]}.srt"))
             self.event_logger.info(
-                f"[+] {lang.get_translation('extract_subtitles', self.locale)} '{input_path}'"
+                f"[>] {lang.get_translation('extract_subtitles', self.locale)} '{input_path}'"
             )
             try:
                 # Use FFmpeg to extract subtitles
@@ -200,7 +200,7 @@ def to_subtitles(
 
                 if os.path.exists(out_path) and os.path.getsize(out_path) > 0:
                     self.event_logger.info(
-                        f"[+] {lang.get_translation('subtitles_success', self.locale)} '{out_path}'"
+                        f"[>] {lang.get_translation('subtitles_success', self.locale)} '{out_path}'"
                     )
                     self.file_handler.post_process(
                         movie_path_set, out_path, delete, show_status=False
@@ -217,7 +217,7 @@ def to_subtitles(
                     )
                     if os.path.exists(out_path) and os.path.getsize(out_path) > 0:
                         self.event_logger.info(
-                            f"[+] {lang.get_translation('embed_subtitles_success')} '{out_path}'"
+                            f"[>] {lang.get_translation('embed_subtitles_success')} '{out_path}'"
                         )
                         self.file_handler.post_process(
                             movie_path_set, out_path, delete, show_status=False
@@ -400,3 +400,132 @@ def _place_img(page, img_path, full_page=False):
                         doc.add_page_break()
                 doc.save(out_path)
                 self.file_handler.post_process(document_path_set, out_path, delete)
+
+    def split_pdf(self, output: str, doc_path_set: tuple, page_ranges: str, delete: bool, format: str = "pdf"):
+        # output: str - output directory
+        # doc_path_set: tuple - (path, name, format)
+        # page_ranges: str - page ranges to split the document into
+        # delete: bool - delete the original document after completion
+        # format: str - format of the document
+        doc_path = self.file_handler.join_back(doc_path_set)
+        if format == "pdf":
+            # Open the PDF document
+            pdf = fitz.open(doc_path)
+            total_pages = len(pdf)
+            if total_pages == 0:
+                pdf.close()
+                return
+            # Parse page_ranges into a list of tuples
+            parsed_ranges = self._parse_page_ranges(page_ranges, total_pages)
+
+            # None means any splitting efforts would be futile, skip
+            if parsed_ranges is None:
+                pdf.close()
+                return
+
+            # Create output directory if it doesn't exist
+            os.makedirs(output, exist_ok=True)
+            for i, (start, end) in enumerate(parsed_ranges):
+                new_pdf = fitz.open()
+                new_pdf.insert_pdf(pdf, from_page=start-1, to_page=end-1)
+                # Generate output filename
+                if len(parsed_ranges) == 1:
+                    out_filename = f"{doc_path_set[1]}_{start}-{end}.{format}"
+                else:
+                    out_filename = f"{doc_path_set[1]}_split_{i+1}_{start}-{end}.{format}"
+                out_path = os.path.abspath(os.path.join(output, out_filename))
+                # Save new PDF
+                new_pdf.save(out_path)
+                new_pdf.close()
+                self.event_logger.info(f"[>] {lang.get_translation('split_produced', self.locale)}: {out_path}")
+            # Close source PDF
+            pdf.close()
+            self.file_handler.post_process(doc_path_set, out_path, delete)
+
+    def _parse_page_ranges(self, page_ranges: str, total_pages: int) -> list[tuple[int, int]] | None:
+        """
+        Parse page_ranges string into list of (start, end) tuples.
+        Handles: None, '1-5', '2-5,3-4', '3-6, 8-20, 23-45, rest', 'all', 'rest', 
+        '3-6;15-22', '1-7;8-15;20-22;rest', '3', '35', '12-end', '2-end',
+        '2-5, 3-6, 12-end', '2-end, 3-6, 12-end', '2-end, 3-6'
+        """
+        if not page_ranges or page_ranges.strip() == "":
+            # Default to all pages
+            return None #[(1, total_pages)], we don't need that, skip that
+        page_ranges = page_ranges.strip()
+        if page_ranges.lower() in ["all", "rest"]:
+            return None #[(1, total_pages)], we don't need that, skip that
+
+        ranges = []
+        for delim in [',', ';']:
+            # Split by comma or semicolon
+            if delim in page_ranges:
+                ranges = [r.strip() for r in page_ranges.split(delim)]
+                break
+        if not ranges:
+            # No delimiters found
+            ranges = [page_ranges.strip()]
+        
+        parsed_ranges = []
+        # Track where 'rest' should start
+        rest_start = 1
+        
+        for range_str in ranges:
+            range_str = range_str.strip()
+            if range_str.lower() == "rest":
+                # 'rest': from highest processed page + 1 to end
+                if parsed_ranges:
+                    # Find highest end page processed so far
+                    max_end = max(end for _, end in parsed_ranges)
+                    rest_start = max_end + 1
+                parsed_ranges.append((rest_start, total_pages))
+                continue
+            
+            # Handle single page numbers
+            if range_str.isdigit():
+                page_num = int(range_str)
+                if 1 <= page_num <= total_pages:
+                    parsed_ranges.append((page_num, page_num))
+                    rest_start = max(rest_start, page_num + 1)
+                continue
+            
+            # Handle ranges like "1-5", "12-end", "2-end"
+            if '-' in range_str:
+                parts = range_str.split('-', 1)
+                if len(parts) == 2:
+                    start_str, end_str = parts[0].strip(), parts[1].strip()
+                    
+                    # Parse start
+                    try:
+                        start = int(start_str)
+                    except ValueError:
+                        continue # Some invalid range
+                    
+                    # Parse end
+                    if end_str.lower() == "end":
+                        end = total_pages
+                    else:
+                        try:
+                            end = int(end_str)
+                        except ValueError:
+                            continue # Some invalid range
+                    
+                    # Validate, add range
+                    if 1 <= start <= total_pages and 1 <= end <= total_pages and start <= end:
+                        parsed_ranges.append((start, end))
+                        rest_start = max(rest_start, end + 1)
+        
+        # Remove duplicate range tuples, preserve order
+        # Ranges still are allowed to overlap, must each be unique though
+        seen = set()
+        unique_ranges = []
+        for range_tuple in parsed_ranges:
+            if range_tuple not in seen:
+                seen.add(range_tuple)
+                unique_ranges.append(range_tuple)
+        
+        if not unique_ranges:
+            # If no valid ranges parsed, default to all pages, which is None
+            unique_ranges = None
+        
+        return unique_ranges

core/movie_converter.py

@@ -349,7 +349,7 @@ def to_protocol(
                             exist_ok=True,
                         )
                     self.event_logger.info(
-                        f"[+] {lang.get_translation('get_hls', self.locale)} {self.file_handler.join_back(movie_path_set)}: {resolution} at {v_bitrate} video, {a_bitrate} audio"
+                        f"[>] {lang.get_translation('get_hls', self.locale)} {self.file_handler.join_back(movie_path_set)}: {resolution} at {v_bitrate} video, {a_bitrate} audio"
                     )
                     stream = [
                         "-map",
@@ -383,7 +383,7 @@ def to_protocol(
                     cmd += stream
                     variant_playlist += f"#EXT-X-STREAM-INF:BANDWIDTH={int(v_bitrate[:-1]) * 1000},RESOLUTION={resolution}\n{i}.m3u8\n"
                 self.event_logger.info(
-                    f"[+] {lang.get_translation('get_hls_master', self.locale)} {self.file_handler.join_back(movie_path_set)}"
+                    f"[>] {lang.get_translation('get_hls_master', self.locale)} {self.file_handler.join_back(movie_path_set)}"
                 )
                 master_playlist_path = os.path.join(current_out_dir, "master.m3u8")
 
@@ -402,7 +402,7 @@ def to_protocol(
                     )
             elif protocol[0] == "dash":
                 self.event_logger.info(
-                    f"[+] {lang.get_translation('create_dash', self.locale)} {self.file_handler.join_back(movie_path_set)}"
+                    f"[>] {lang.get_translation('create_dash', self.locale)} {self.file_handler.join_back(movie_path_set)}"
                 )
                 out_path = os.path.join(current_out_dir, "manifest.mpd")
                 cmd = [

core/utils/file_handler.py

@@ -28,7 +28,7 @@ def post_process(
             # Only log if the conversion was successful and output exists
             if show_status and os.path.exists(out_path):
                 self.event_logger.info(
-                    f"[+] {lang.get_translation('converted', self.locale)} "
+                    f"[>] {lang.get_translation('converted', self.locale)} "
                     f'"{source_path}" 🡢 "{out_path}"'
                 )
 

tests/test_cli.py

@@ -38,6 +38,7 @@ def test_blank_start_no_files_in_cli_output(controller_instance, caplog):
             None,
             None,
             None,
+            None,
             False,
             False,
             False,