109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
import argparse
|
|
import os
|
|
from fledermaus_formatter.logger import logger
|
|
|
|
from fledermaus_formatter.table_handler.serializer import TableSerializer
|
|
import fledermaus_formatter.instructions.countings_formater as countings_formater
|
|
import fledermaus_formatter.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
|
|
|
|
logger.info(f"Processing: {input_file}")
|
|
|
|
serializer = TableSerializer(
|
|
input_file,
|
|
formatter.EXPECTED_INPUT_TABLE_SCHEMA,
|
|
formatter.EXPECTED_OUTPUT_TABLE_SCHEMA
|
|
)
|
|
|
|
serializer.run(
|
|
formatter.COL_INSTRUCTIONS,
|
|
row_instructions
|
|
)
|
|
|
|
serializer.write_output(output_file)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Serialize Excel tables.")
|
|
|
|
# Single file mode (optional fallback)
|
|
parser.add_argument("-i", "--input", help="Input Excel file")
|
|
parser.add_argument("-o", "--output", help="Output Excel file")
|
|
parser.add_argument("-v", help="Version", action="store_true")
|
|
|
|
# Directory mode
|
|
parser.add_argument("--input-dir", help="Input directory")
|
|
parser.add_argument("--output-dir", help="Output directory")
|
|
parser.add_argument("-s", help="Serial mode", action="store_true")
|
|
|
|
parser.add_argument(
|
|
"-t", "--type",
|
|
required=False,
|
|
choices=["ring", "counting"],
|
|
help="Type of table to process"
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.v:
|
|
print("1.4")
|
|
return
|
|
if not args.type:
|
|
parser.error("The --type argument is required.")
|
|
retur
|
|
|
|
# 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")):
|
|
continue
|
|
|
|
input_path = os.path.join(root, file)
|
|
|
|
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)
|
|
|
|
tasks.append((input_path, output_path, args.type))
|
|
|
|
max_workers = os.cpu_count() or 1 if not args.s else 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, args.type)
|
|
|
|
else:
|
|
parser.error("You must provide either input/output files or input-dir/output-dir")
|
|
|
|
logger.info("Done")
|