parallel processing of multiple files

This commit is contained in:
Kyattsukuro 2026-03-18 14:42:08 +01:00
parent 95e95f407f
commit d21ffdc52a
2 changed files with 31 additions and 14 deletions

View File

@ -6,8 +6,20 @@ from table_handler.serializer import TableSerializer
import instructions.countings_formater as countings_formater import instructions.countings_formater as countings_formater
import instructions.ring_formater as ring_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}") logger.info(f"Processing: {input_file}")
serializer = TableSerializer( serializer = TableSerializer(
@ -44,18 +56,12 @@ def main():
args = parser.parse_args() 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 # directory
if args.input_dir and args.output_dir: if args.input_dir and args.output_dir:
logger.info("Running in directory mode") logger.info("Running in directory mode")
tasks = []
for root, _, files in os.walk(args.input_dir): for root, _, files in os.walk(args.input_dir):
for file in files: for file in files:
if not file.endswith((".xlsx", ".xls")): if not file.endswith((".xlsx", ".xls")):
@ -63,19 +69,29 @@ def main():
input_path = os.path.join(root, file) input_path = os.path.join(root, file)
# Create corresponding output path
relative_path = os.path.relpath(root, args.input_dir) relative_path = os.path.relpath(root, args.input_dir)
output_root = os.path.join(args.output_dir, relative_path) output_root = os.path.join(args.output_dir, relative_path)
os.makedirs(output_root, exist_ok=True) os.makedirs(output_root, exist_ok=True)
output_path = os.path.join(output_root, file) 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: elif args.input and args.output:
logger.info("Running in single file mode") 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: else:
parser.error("You must provide either input/output files or input-dir/output-dir") parser.error("You must provide either input/output files or input-dir/output-dir")

View File

@ -37,6 +37,7 @@ class TableSerializer:
if not self.ws: if not self.ws:
raise ValueError("No active worksheet found in the workbook.") raise ValueError("No active worksheet found in the workbook.")
self.file = file
self.data = self._serialize_table(expected_input_schema) self.data = self._serialize_table(expected_input_schema)
self.serialized_output = {} self.serialized_output = {}
self.seperator = ", " self.seperator = ", "
@ -147,7 +148,7 @@ class TableSerializer:
amount_working_rows = self.vertical_len +2 amount_working_rows = self.vertical_len +2
for rid in range(2, amount_working_rows): for rid in range(2, amount_working_rows):
if rid % 1000 == 0: 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 = {} working_row = {}
source_row = self.get_row_as_dict(rid) source_row = self.get_row_as_dict(rid)
for source_cols, instructions in list(col_instructions.items()) + [ for source_cols, instructions in list(col_instructions.items()) + [