490 lines
19 KiB
Python
490 lines
19 KiB
Python
"""
|
|
Cross-Tabulated CSV 내보내기 모듈 (수정 버전)
|
|
JSON 형태의 분석 결과를 key-value 형태의 cross-tabulated CSV로 저장하는 기능을 제공합니다.
|
|
|
|
Author: Claude Assistant
|
|
Created: 2025-07-15
|
|
Updated: 2025-07-16 (디버깅 개선 버전)
|
|
Version: 1.1.0
|
|
"""
|
|
|
|
import pandas as pd
|
|
import json
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import List, Dict, Any, Optional, Union
|
|
import os
|
|
import re
|
|
|
|
# 로깅 설정
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CrossTabulatedCSVExporter:
|
|
"""Cross-Tabulated CSV 내보내기 클래스 (수정 버전)"""
|
|
|
|
def __init__(self):
|
|
"""Cross-Tabulated CSV 내보내기 초기화"""
|
|
self.coordinate_pattern = re.compile(r'\b(\d+)\s*,\s*(\d+)\b') # x,y 좌표 패턴
|
|
self.debug_mode = True # 디버깅 모드 활성화
|
|
|
|
def export_cross_tabulated_csv(
|
|
self,
|
|
processing_results: List[Any],
|
|
output_path: str,
|
|
include_coordinates: bool = True,
|
|
coordinate_source: str = "auto" # "auto", "text_blocks", "analysis_result", "none"
|
|
) -> bool:
|
|
"""
|
|
처리 결과를 cross-tabulated CSV 형태로 저장
|
|
|
|
Args:
|
|
processing_results: 다중 파일 처리 결과 리스트
|
|
output_path: 출력 CSV 파일 경로
|
|
include_coordinates: 좌표 정보 포함 여부
|
|
coordinate_source: 좌표 정보 출처 ("auto", "text_blocks", "analysis_result", "none")
|
|
|
|
Returns:
|
|
저장 성공 여부
|
|
"""
|
|
try:
|
|
if self.debug_mode:
|
|
logger.info(f"=== Cross-tabulated CSV 저장 시작 ===")
|
|
logger.info(f"입력된 결과 수: {len(processing_results)}")
|
|
logger.info(f"출력 경로: {output_path}")
|
|
logger.info(f"좌표 포함: {include_coordinates}, 좌표 출처: {coordinate_source}")
|
|
|
|
# 입력 데이터 검증
|
|
if not processing_results:
|
|
logger.warning("입력된 처리 결과가 비어있습니다.")
|
|
return False
|
|
|
|
# 각 결과 객체의 구조 분석
|
|
for i, result in enumerate(processing_results):
|
|
if self.debug_mode:
|
|
logger.info(f"결과 {i+1}: {self._analyze_result_structure(result)}")
|
|
|
|
# 모든 파일의 key-value 쌍을 수집
|
|
all_data_rows = []
|
|
|
|
for i, result in enumerate(processing_results):
|
|
try:
|
|
if not hasattr(result, 'success'):
|
|
logger.warning(f"결과 {i+1}: 'success' 속성이 없습니다. 스킵합니다.")
|
|
continue
|
|
|
|
if not result.success:
|
|
if self.debug_mode:
|
|
logger.info(f"결과 {i+1}: 실패한 파일, 스킵합니다 ({getattr(result, 'error_message', 'Unknown error')})")
|
|
continue # 실패한 파일은 제외
|
|
|
|
file_data = self._extract_key_value_pairs(result, include_coordinates, coordinate_source)
|
|
if file_data:
|
|
all_data_rows.extend(file_data)
|
|
if self.debug_mode:
|
|
logger.info(f"결과 {i+1}: {len(file_data)}개 key-value 쌍 추출")
|
|
else:
|
|
if self.debug_mode:
|
|
logger.warning(f"결과 {i+1}: key-value 쌍을 추출할 수 없습니다")
|
|
|
|
except Exception as e:
|
|
logger.error(f"결과 {i+1} 처리 중 오류: {str(e)}")
|
|
continue
|
|
|
|
if not all_data_rows:
|
|
logger.warning("저장할 데이터가 없습니다. 모든 파일에서 유효한 key-value 쌍을 추출할 수 없었습니다.")
|
|
if self.debug_mode:
|
|
self._print_debug_summary(processing_results)
|
|
return False
|
|
|
|
# DataFrame 생성
|
|
df = pd.DataFrame(all_data_rows)
|
|
|
|
# 컬럼 순서 정렬
|
|
column_order = ['file_name', 'file_type', 'key', 'value']
|
|
if include_coordinates and coordinate_source != "none":
|
|
column_order.extend(['x', 'y'])
|
|
|
|
# 추가 컬럼들을 뒤에 배치
|
|
existing_columns = [col for col in column_order if col in df.columns]
|
|
additional_columns = [col for col in df.columns if col not in existing_columns]
|
|
df = df[existing_columns + additional_columns]
|
|
|
|
# 출력 디렉토리 생성
|
|
os.makedirs(os.path.dirname(output_path), exist_ok=True)
|
|
|
|
# UTF-8 BOM으로 저장 (한글 호환성)
|
|
df.to_csv(output_path, index=False, encoding='utf-8-sig')
|
|
|
|
logger.info(f"Cross-tabulated CSV 저장 완료: {output_path}")
|
|
logger.info(f"총 {len(all_data_rows)}개 key-value 쌍 저장")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Cross-tabulated CSV 저장 오류: {str(e)}")
|
|
return False
|
|
|
|
def _analyze_result_structure(self, result: Any) -> str:
|
|
"""결과 객체의 구조를 분석하여 문자열로 반환"""
|
|
try:
|
|
info = []
|
|
|
|
# 기본 속성들 확인
|
|
if hasattr(result, 'file_name'):
|
|
info.append(f"file_name='{result.file_name}'")
|
|
if hasattr(result, 'file_type'):
|
|
info.append(f"file_type='{result.file_type}'")
|
|
if hasattr(result, 'success'):
|
|
info.append(f"success={result.success}")
|
|
|
|
# PDF 관련 속성
|
|
if hasattr(result, 'pdf_analysis_result'):
|
|
pdf_result = result.pdf_analysis_result
|
|
if pdf_result:
|
|
if isinstance(pdf_result, str):
|
|
info.append(f"pdf_analysis_result=str({len(pdf_result)} chars)")
|
|
else:
|
|
info.append(f"pdf_analysis_result={type(pdf_result).__name__}")
|
|
else:
|
|
info.append("pdf_analysis_result=None")
|
|
|
|
# DXF 관련 속성
|
|
if hasattr(result, 'dxf_title_blocks'):
|
|
dxf_blocks = result.dxf_title_blocks
|
|
if dxf_blocks:
|
|
info.append(f"dxf_title_blocks=list({len(dxf_blocks)} blocks)")
|
|
else:
|
|
info.append("dxf_title_blocks=None")
|
|
|
|
return " | ".join(info) if info else "구조 분석 실패"
|
|
|
|
except Exception as e:
|
|
return f"분석 오류: {str(e)}"
|
|
|
|
def _print_debug_summary(self, processing_results: List[Any]):
|
|
"""디버깅을 위한 요약 정보 출력"""
|
|
logger.info("=== 디버깅 요약 ===")
|
|
|
|
success_count = 0
|
|
pdf_count = 0
|
|
dxf_count = 0
|
|
has_pdf_data = 0
|
|
has_dxf_data = 0
|
|
|
|
for i, result in enumerate(processing_results):
|
|
try:
|
|
if hasattr(result, 'success') and result.success:
|
|
success_count += 1
|
|
|
|
file_type = getattr(result, 'file_type', 'unknown').lower()
|
|
if file_type == 'pdf':
|
|
pdf_count += 1
|
|
if getattr(result, 'pdf_analysis_result', None):
|
|
has_pdf_data += 1
|
|
elif file_type == 'dxf':
|
|
dxf_count += 1
|
|
if getattr(result, 'dxf_title_blocks', None):
|
|
has_dxf_data += 1
|
|
|
|
except Exception as e:
|
|
logger.error(f"결과 {i+1} 분석 중 오류: {str(e)}")
|
|
|
|
logger.info(f"총 결과: {len(processing_results)}개")
|
|
logger.info(f"성공한 결과: {success_count}개")
|
|
logger.info(f"PDF 파일: {pdf_count}개 (분석 데이터 있음: {has_pdf_data}개)")
|
|
logger.info(f"DXF 파일: {dxf_count}개 (타이틀블록 데이터 있음: {has_dxf_data}개)")
|
|
|
|
def _extract_key_value_pairs(
|
|
self,
|
|
result: Any,
|
|
include_coordinates: bool,
|
|
coordinate_source: str
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
단일 파일 결과에서 key-value 쌍 추출
|
|
|
|
Args:
|
|
result: 파일 처리 결과
|
|
include_coordinates: 좌표 정보 포함 여부
|
|
coordinate_source: 좌표 정보 출처
|
|
|
|
Returns:
|
|
key-value 쌍 리스트
|
|
"""
|
|
data_rows = []
|
|
|
|
try:
|
|
# 기본 정보 확인
|
|
file_name = getattr(result, 'file_name', 'Unknown')
|
|
file_type = getattr(result, 'file_type', 'Unknown')
|
|
|
|
base_info = {
|
|
'file_name': file_name,
|
|
'file_type': file_type,
|
|
}
|
|
|
|
if self.debug_mode:
|
|
logger.info(f"처리 중: {file_name} ({file_type})")
|
|
|
|
# PDF 분석 결과 처리
|
|
if file_type.lower() == 'pdf':
|
|
pdf_result = getattr(result, 'pdf_analysis_result', None)
|
|
if pdf_result:
|
|
pdf_rows = self._extract_pdf_key_values(result, base_info, include_coordinates, coordinate_source)
|
|
data_rows.extend(pdf_rows)
|
|
if self.debug_mode:
|
|
logger.info(f"PDF에서 {len(pdf_rows)}개 key-value 쌍 추출")
|
|
else:
|
|
if self.debug_mode:
|
|
logger.warning(f"PDF 분석 결과가 없습니다: {file_name}")
|
|
|
|
# DXF 분석 결과 처리
|
|
elif file_type.lower() == 'dxf':
|
|
dxf_blocks = getattr(result, 'dxf_title_blocks', None)
|
|
if dxf_blocks:
|
|
dxf_rows = self._extract_dxf_key_values(result, base_info, include_coordinates, coordinate_source)
|
|
data_rows.extend(dxf_rows)
|
|
if self.debug_mode:
|
|
logger.info(f"DXF에서 {len(dxf_rows)}개 key-value 쌍 추출")
|
|
else:
|
|
if self.debug_mode:
|
|
logger.warning(f"DXF 타이틀블록 데이터가 없습니다: {file_name}")
|
|
|
|
else:
|
|
if self.debug_mode:
|
|
logger.warning(f"지원하지 않는 파일 형식: {file_type}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Key-value 추출 오류 ({getattr(result, 'file_name', 'Unknown')}): {str(e)}")
|
|
|
|
return data_rows
|
|
|
|
def _extract_pdf_key_values(
|
|
self,
|
|
result: Any,
|
|
base_info: Dict[str, str],
|
|
include_coordinates: bool,
|
|
coordinate_source: str
|
|
) -> List[Dict[str, Any]]:
|
|
"""PDF 분석 결과에서 key-value 쌍 추출"""
|
|
data_rows = []
|
|
|
|
try:
|
|
# PDF 분석 결과를 JSON으로 파싱
|
|
analysis_result = getattr(result, 'pdf_analysis_result', None)
|
|
|
|
if not analysis_result:
|
|
return data_rows
|
|
|
|
if isinstance(analysis_result, str):
|
|
try:
|
|
analysis_data = json.loads(analysis_result)
|
|
except json.JSONDecodeError:
|
|
# JSON이 아닌 경우 텍스트로 처리
|
|
analysis_data = {"분석결과": analysis_result}
|
|
else:
|
|
analysis_data = analysis_result
|
|
|
|
if self.debug_mode:
|
|
logger.info(f"PDF 분석 데이터 구조: {type(analysis_data).__name__}")
|
|
if isinstance(analysis_data, dict):
|
|
logger.info(f"PDF 분석 데이터 키: {list(analysis_data.keys())}")
|
|
|
|
# 중첩된 구조를 평탄화하여 key-value 쌍 생성
|
|
flattened_data = self._flatten_dict(analysis_data)
|
|
|
|
for key, value in flattened_data.items():
|
|
if value is None or str(value).strip() == "":
|
|
continue # 빈 값 제외
|
|
|
|
row_data = base_info.copy()
|
|
row_data.update({
|
|
'key': key,
|
|
'value': str(value),
|
|
})
|
|
|
|
# 좌표 정보 추가
|
|
if include_coordinates and coordinate_source != "none":
|
|
coordinates = self._extract_coordinates(key, value, coordinate_source)
|
|
row_data.update(coordinates)
|
|
|
|
data_rows.append(row_data)
|
|
|
|
except Exception as e:
|
|
logger.error(f"PDF key-value 추출 오류: {str(e)}")
|
|
|
|
return data_rows
|
|
|
|
def _extract_dxf_key_values(
|
|
self,
|
|
result: Any,
|
|
base_info: Dict[str, str],
|
|
include_coordinates: bool,
|
|
coordinate_source: str
|
|
) -> List[Dict[str, Any]]:
|
|
"""DXF 분석 결과에서 key-value 쌍 추출"""
|
|
data_rows = []
|
|
|
|
try:
|
|
title_blocks = getattr(result, 'dxf_title_blocks', None)
|
|
|
|
if not title_blocks:
|
|
return data_rows
|
|
|
|
if self.debug_mode:
|
|
logger.info(f"DXF 타이틀블록 수: {len(title_blocks)}")
|
|
|
|
for block_idx, title_block in enumerate(title_blocks):
|
|
if not isinstance(title_block, dict):
|
|
continue
|
|
|
|
block_name = title_block.get('block_name', 'Unknown')
|
|
|
|
# 블록 정보
|
|
row_data = base_info.copy()
|
|
row_data.update({
|
|
'key': f"{block_name}_블록명",
|
|
'value': block_name,
|
|
})
|
|
|
|
if include_coordinates and coordinate_source != "none":
|
|
coordinates = self._extract_coordinates('블록명', block_name, coordinate_source)
|
|
row_data.update(coordinates)
|
|
|
|
data_rows.append(row_data)
|
|
|
|
# 속성 정보
|
|
attributes = title_block.get('attributes', [])
|
|
if self.debug_mode:
|
|
logger.info(f"블록 {block_idx+1} ({block_name}): {len(attributes)}개 속성")
|
|
|
|
for attr_idx, attr in enumerate(attributes):
|
|
if not isinstance(attr, dict):
|
|
continue
|
|
|
|
attr_text = attr.get('text', '')
|
|
if not attr_text or str(attr_text).strip() == "":
|
|
continue # 빈 속성 제외
|
|
|
|
# 속성별 key-value 쌍 생성
|
|
attr_key = attr.get('tag', attr.get('prompt', f'Unknown_Attr_{attr_idx}'))
|
|
attr_value = str(attr_text)
|
|
|
|
row_data = base_info.copy()
|
|
row_data.update({
|
|
'key': attr_key,
|
|
'value': attr_value,
|
|
})
|
|
|
|
# DXF 속성의 경우 insert 좌표 사용
|
|
if include_coordinates and coordinate_source != "none":
|
|
x_coord = attr.get('insert_x', '')
|
|
y_coord = attr.get('insert_y', '')
|
|
|
|
if x_coord and y_coord:
|
|
row_data.update({
|
|
'x': round(float(x_coord), 2) if isinstance(x_coord, (int, float)) else x_coord,
|
|
'y': round(float(y_coord), 2) if isinstance(y_coord, (int, float)) else y_coord,
|
|
})
|
|
else:
|
|
row_data.update({'x': '', 'y': ''})
|
|
|
|
data_rows.append(row_data)
|
|
|
|
except Exception as e:
|
|
logger.error(f"DXF key-value 추출 오류: {str(e)}")
|
|
|
|
return data_rows
|
|
|
|
def _flatten_dict(self, data: Dict[str, Any], parent_key: str = '', sep: str = '_') -> Dict[str, Any]:
|
|
"""
|
|
중첩된 딕셔너리를 평탄화
|
|
|
|
Args:
|
|
data: 평탄화할 딕셔너리
|
|
parent_key: 부모 키
|
|
sep: 구분자
|
|
|
|
Returns:
|
|
평탄화된 딕셔너리
|
|
"""
|
|
items = []
|
|
|
|
for k, v in data.items():
|
|
new_key = f"{parent_key}{sep}{k}" if parent_key else k
|
|
|
|
if isinstance(v, dict):
|
|
# 중첩된 딕셔너리인 경우 재귀 호출
|
|
items.extend(self._flatten_dict(v, new_key, sep=sep).items())
|
|
elif isinstance(v, list):
|
|
# 리스트인 경우 인덱스와 함께 처리
|
|
for i, item in enumerate(v):
|
|
if isinstance(item, dict):
|
|
items.extend(self._flatten_dict(item, f"{new_key}_{i}", sep=sep).items())
|
|
else:
|
|
items.append((f"{new_key}_{i}", item))
|
|
else:
|
|
items.append((new_key, v))
|
|
|
|
return dict(items)
|
|
|
|
def _extract_coordinates(self, key: str, value: str, coordinate_source: str) -> Dict[str, str]:
|
|
"""
|
|
텍스트에서 좌표 정보 추출
|
|
|
|
Args:
|
|
key: 키
|
|
value: 값
|
|
coordinate_source: 좌표 정보 출처
|
|
|
|
Returns:
|
|
좌표 딕셔너리
|
|
"""
|
|
coordinates = {'x': '', 'y': ''}
|
|
|
|
try:
|
|
# 값에서 좌표 패턴 찾기
|
|
matches = self.coordinate_pattern.findall(str(value))
|
|
|
|
if matches:
|
|
# 첫 번째 매치 사용
|
|
x, y = matches[0]
|
|
coordinates = {'x': x, 'y': y}
|
|
else:
|
|
# 키에서 좌표 정보 찾기
|
|
key_matches = self.coordinate_pattern.findall(str(key))
|
|
if key_matches:
|
|
x, y = key_matches[0]
|
|
coordinates = {'x': x, 'y': y}
|
|
|
|
except Exception as e:
|
|
logger.warning(f"좌표 추출 오류: {str(e)}")
|
|
|
|
return coordinates
|
|
|
|
|
|
def generate_cross_tabulated_csv_filename(base_name: str = "cross_tabulated_analysis") -> str:
|
|
"""기본 Cross-tabulated CSV 파일명 생성"""
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
return f"{base_name}_results_{timestamp}.csv"
|
|
|
|
|
|
# 사용 예시
|
|
if __name__ == "__main__":
|
|
# 테스트용 예시
|
|
exporter = CrossTabulatedCSVExporter()
|
|
|
|
# 샘플 처리 결과 (실제 데이터 구조에 맞게 수정)
|
|
sample_results = []
|
|
|
|
# 실제 사용 시에는 processing_results를 전달
|
|
# success = exporter.export_cross_tabulated_csv(
|
|
# sample_results,
|
|
# "test_cross_tabulated.csv",
|
|
# include_coordinates=True
|
|
# )
|
|
|
|
print("Cross-tabulated CSV 내보내기 모듈 (수정 버전) 테스트 완료")
|