216 lines
7.9 KiB
Python
216 lines
7.9 KiB
Python
"""
|
|
배치 처리 명령줄 인터페이스
|
|
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()) |