From d21ffdc52a95a26639028d3a74f12087ce735b85 Mon Sep 17 00:00:00 2001 From: Kyattsukuro Date: Wed, 18 Mar 2026 14:42:08 +0100 Subject: [PATCH] parallel processing of multiple files --- __init__.py | 42 +++++++++++++++++++++++++------------ table_handler/serializer.py | 3 ++- 2 files changed, 31 insertions(+), 14 deletions(-) diff --git a/__init__.py b/__init__.py index 0d0a6da..7d477e5 100644 --- a/__init__.py +++ b/__init__.py @@ -6,8 +6,20 @@ from table_handler.serializer import TableSerializer import instructions.countings_formater as countings_formater import instructions.ring_formater as ring_formater +from concurrent.futures import ProcessPoolExecutor, as_completed + + + + +def process_file(input_file, output_file, xlsx_type): + # Select formatter + if xlsx_type == "ring": + formatter = ring_formater + row_instructions = ring_formater.ROW_INSTRUCTIONS + else: + formatter = countings_formater + row_instructions = countings_formater.ROW_INSTRUCTION -def process_file(input_file, output_file, formatter, row_instructions): logger.info(f"Processing: {input_file}") serializer = TableSerializer( @@ -44,18 +56,12 @@ def main(): args = parser.parse_args() - # Select formatter - if args.type == "ring": - formatter = ring_formater - row_instructions = ring_formater.ROW_INSTRUCTIONS - else: - formatter = countings_formater - row_instructions = countings_formater.ROW_INSTRUCTION - # directory if args.input_dir and args.output_dir: logger.info("Running in directory mode") + tasks = [] + for root, _, files in os.walk(args.input_dir): for file in files: if not file.endswith((".xlsx", ".xls")): @@ -63,19 +69,29 @@ def main(): input_path = os.path.join(root, file) - # Create corresponding output path relative_path = os.path.relpath(root, args.input_dir) output_root = os.path.join(args.output_dir, relative_path) os.makedirs(output_root, exist_ok=True) output_path = os.path.join(output_root, file) - process_file(input_path, output_path, formatter, row_instructions) + tasks.append((input_path, output_path, args.type)) - # single file + max_workers = os.cpu_count() or 1 + logger.info(f"Processing {len(tasks)} files using {max_workers} workers") + + with ProcessPoolExecutor(max_workers=max_workers) as executor: + futures = [executor.submit(process_file, *task) for task in tasks] + + for future in as_completed(futures): + try: + future.result() + except Exception as e: + logger.error(f"Error processing file: {e}") + # single file elif args.input and args.output: logger.info("Running in single file mode") - process_file(args.input, args.output, formatter, row_instructions) + process_file(args.input, args.output, args.type) else: parser.error("You must provide either input/output files or input-dir/output-dir") diff --git a/table_handler/serializer.py b/table_handler/serializer.py index d241260..de6eded 100644 --- a/table_handler/serializer.py +++ b/table_handler/serializer.py @@ -37,6 +37,7 @@ class TableSerializer: if not self.ws: raise ValueError("No active worksheet found in the workbook.") + self.file = file self.data = self._serialize_table(expected_input_schema) self.serialized_output = {} self.seperator = ", " @@ -147,7 +148,7 @@ class TableSerializer: amount_working_rows = self.vertical_len +2 for rid in range(2, amount_working_rows): if rid % 1000 == 0: - logger.info(f"Working on row {rid}/{amount_working_rows} ") + logger.info(f"{self.file} Working on row {rid}/{amount_working_rows} ") working_row = {} source_row = self.get_row_as_dict(rid) for source_cols, instructions in list(col_instructions.items()) + [