""" 배치 처리 명령줄 인터페이스 WPF 애플리케이션에서 호출 가능한 간단한 배치 처리 도구 Usage: python batch_cli.py --files "file1.pdf,file2.dxf" --schema "한국도로공사" --concurrent 3 --batch-mode true --save-intermediate false --include-errors true --output "results.csv" """ import argparse import asyncio import logging import os import sys import time from datetime import datetime from typing import List # 프로젝트 모듈 임포트 from config import Config from multi_file_processor import MultiFileProcessor, BatchProcessingConfig, generate_default_csv_filename # 로깅 설정 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class BatchCLI: """배치 처리 명령줄 인터페이스 클래스""" def __init__(self): self.processor = None self.start_time = None def setup_processor(self) -> bool: """다중 파일 처리기 설정""" try: # 설정 검증 config_errors = Config.validate_config() if config_errors: for error in config_errors: print(f"ERROR: {error}") return False # Gemini API 키 확인 gemini_api_key = Config.get_gemini_api_key() if not gemini_api_key: print("ERROR: Gemini API 키가 설정되지 않았습니다") return False # 처리기 초기화 self.processor = MultiFileProcessor(gemini_api_key) print("START: 배치 처리기 초기화 완료") return True except Exception as e: print(f"ERROR: 처리기 초기화 실패: {e}") return False def parse_file_paths(self, files_arg: str) -> List[str]: """파일 경로 문자열을 리스트로 파싱""" if not files_arg: return [] # 쉼표로 구분된 파일 경로들을 분리 file_paths = [path.strip().strip('"\'') for path in files_arg.split(',')] # 파일 존재 여부 확인 valid_paths = [] for path in file_paths: if os.path.exists(path): valid_paths.append(path) print(f"START: 파일 확인: {os.path.basename(path)}") else: print(f"ERROR: 파일을 찾을 수 없습니다: {path}") return valid_paths def parse_file_list_from_file(self, file_list_path: str) -> List[str]: """파일 리스트 파일에서 파일 경로들을 읽어옴""" if not file_list_path or not os.path.exists(file_list_path): return [] valid_paths = [] try: with open(file_list_path, 'r', encoding='utf-8') as f: for line in f: path = line.strip().strip('"\'') if path and os.path.exists(path): valid_paths.append(path) print(f"START: 파일 확인: {os.path.basename(path)}") elif path: print(f"ERROR: 파일을 찾을 수 없음: {path}") print(f"START: 총 {len(valid_paths)}개 파일 로드됨") return valid_paths except Exception as e: print(f"ERROR: 파일 리스트 읽기 실패: {e}") return [] def create_batch_config(self, args) -> BatchProcessingConfig: """명령줄 인수에서 배치 설정 생성""" config = BatchProcessingConfig( organization_type=args.schema, enable_gemini_batch_mode=args.batch_mode, max_concurrent_files=args.concurrent, save_intermediate_results=args.save_intermediate, output_csv_path=args.output, include_error_files=args.include_errors ) return config def progress_callback(self, current: int, total: int, status: str): """진행률 콜백 함수 - WPF가 기대하는 형식으로 출력""" # WPF가 파싱할 수 있는 간단한 형식으로 출력 print(f"PROGRESS: {current}/{total}") print(f"COMPLETED: {status}") async def run_batch_processing(self, file_paths: List[str], config: BatchProcessingConfig) -> bool: """배치 처리 실행""" try: self.start_time = time.time() total_files = len(file_paths) print(f"START: 배치 처리 시작: {total_files}개 파일") # 처리 실행 results = await self.processor.process_multiple_files( file_paths, config, self.progress_callback ) # 처리 완료 end_time = time.time() total_time = end_time - self.start_time # 요약 정보 summary = self.processor.get_processing_summary() print(f"COMPLETED: 배치 처리 완료!") print(f"COMPLETED: 총 처리 시간: {total_time:.1f}초") print(f"COMPLETED: 성공: {summary['success_files']}개, 실패: {summary['failed_files']}개") print(f"COMPLETED: CSV 결과 저장: {config.output_csv_path}") print(f"COMPLETED: JSON 결과 저장: {config.output_csv_path.replace('.csv', '.json')}") return True except Exception as e: print(f"ERROR: 배치 처리 중 오류: {e}") return False def str_to_bool(value: str) -> bool: """문자열을 boolean으로 변환""" return value.lower() in ["true", "1", "yes", "on"] async def main(): """메인 함수""" parser = argparse.ArgumentParser(description="PDF/DXF 파일 배치 처리 도구") # 파일 입력 방식 (둘 중 하나 필수) input_group = parser.add_mutually_exclusive_group(required=True) input_group.add_argument("--files", "-f", help="처리할 파일 경로들 (쉼표로 구분)") input_group.add_argument("--file-list", "-fl", help="처리할 파일 경로가 담긴 텍스트 파일") # 선택적 인수들 parser.add_argument("--schema", "-s", default="한국도로공사", help="분석 스키마") parser.add_argument("--concurrent", "-c", type=int, default=3, help="동시 처리할 파일 수") parser.add_argument("--batch-mode", "-b", default="false", help="배치 모드 사용 여부") parser.add_argument("--save-intermediate", "-i", default="true", help="중간 결과 저장 여부") parser.add_argument("--include-errors", "-e", default="true", help="오류 파일 포함 여부") parser.add_argument("--output", "-o", help="출력 CSV 파일 경로 (JSON 파일도 함께 생성됨)") args = parser.parse_args() # CLI 인스턴스 생성 cli = BatchCLI() # 처리기 설정 if not cli.setup_processor(): sys.exit(1) # 파일 경로 파싱 if args.files: input_files = cli.parse_file_paths(args.files) else: input_files = cli.parse_file_list_from_file(args.file_list) if not input_files: print("ERROR: 처리할 파일이 없습니다.") sys.exit(1) # boolean 변환 args.batch_mode = str_to_bool(args.batch_mode) args.save_intermediate = str_to_bool(args.save_intermediate) args.include_errors = str_to_bool(args.include_errors) # 배치 설정 생성 config = cli.create_batch_config(args) # 배치 처리 실행 success = await cli.run_batch_processing(input_files, config) if success: print("SUCCESS: 배치 처리가 성공적으로 완료되었습니다.") sys.exit(0) else: print("ERROR: 배치 처리 중 오류가 발생했습니다.") sys.exit(1) if __name__ == "__main__": asyncio.run(main())