ETC
Parse Json
Eunbig
2025. 5. 23. 14:34
Project Discovery의 httpx json 데이터 파싱용 코드
#!/usr/bin/env python3
"""
Enhanced JSON to Excel/CSV Converter
Supports multiple JSON formats with high performance and flexibility
"""
import json
import pandas as pd
import sys
import os
import argparse
import time
from pathlib import Path
from typing import Dict, List, Any, Union
import logging
# 로깅 설정
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class JSONConverter:
def __init__(self, max_depth: int = 10, batch_size: int = 1000):
"""
JSON 변환기 초기화
Args:
max_depth: JSON 플래튼 최대 깊이
batch_size: 배치 처리 크기 (메모리 효율성)
"""
self.max_depth = max_depth
self.batch_size = batch_size
self.stats = {
'total_records': 0,
'success_records': 0,
'error_records': 0,
'processing_time': 0
}
def flatten_json(self, data: Any, max_depth: int = None, parent_key: str = '', sep: str = '_') -> Dict[str, Any]:
"""
고급 JSON 플래튼 함수 - 깊이 제한 및 다양한 데이터 타입 지원
Args:
data: 플래튼할 JSON 데이터
max_depth: 최대 깊이 (None이면 self.max_depth 사용)
parent_key: 부모 키 이름
sep: 키 구분자
Returns:
Dict: 플래튼된 딕셔너리
"""
if max_depth is None:
max_depth = self.max_depth
items = []
if isinstance(data, dict) and max_depth > 0:
for key, value in data.items():
new_key = f"{parent_key}{sep}{key}" if parent_key else key
items.extend(self.flatten_json(value, max_depth - 1, new_key, sep).items())
elif isinstance(data, list):
if len(data) == 0:
items.append((parent_key, None))
elif all(isinstance(item, (str, int, float, bool, type(None))) for item in data):
# 원시 타입의 리스트는 쉼표로 구분된 문자열로 변환
items.append((parent_key, ', '.join(map(str, filter(lambda x: x is not None, data)))))
elif len(data) <= 10: # 작은 리스트는 인덱스별로 전개
for i, item in enumerate(data):
new_key = f"{parent_key}_{i}" if parent_key else str(i)
items.extend(self.flatten_json(item, max_depth - 1, new_key, sep).items())
else:
# 큰 리스트는 요약 정보만 저장
items.append((f"{parent_key}_length", len(data)))
items.append((f"{parent_key}_sample", str(data[0]) if data else None))
else:
# 원시 타입 또는 최대 깊이 도달
if isinstance(data, (dict, list)) and max_depth <= 0:
items.append((parent_key, f"[Complex object - max depth reached]"))
else:
items.append((parent_key, data))
return dict(items)
def detect_json_format(self, file_path: str) -> str:
"""
JSON 파일 형식 자동 감지
Returns:
str: 'jsonl', 'json_array', 'single_json'
"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
first_line = f.readline().strip()
if not first_line:
return 'empty'
# 첫 줄이 JSON 객체로 파싱되는지 확인
try:
json.loads(first_line)
# 두 번째 줄도 확인
second_line = f.readline().strip()
if second_line:
try:
json.loads(second_line)
return 'jsonl' # JSON Lines 형식
except:
pass
return 'jsonl' # 단일 라인일 수도 있음
except:
pass
# 전체 파일을 JSON으로 파싱 시도
f.seek(0)
content = f.read()
data = json.loads(content)
if isinstance(data, list):
return 'json_array'
else:
return 'single_json'
except Exception as e:
logger.warning(f"JSON 형식 감지 실패: {e}")
return 'unknown'
def parse_jsonl(self, file_path: str) -> List[Dict[str, Any]]:
"""JSON Lines 형식 파싱 (배치 처리)"""
data = []
with open(file_path, 'r', encoding='utf-8') as f:
batch = []
for line_num, line in enumerate(f, 1):
line = line.strip()
if not line:
continue
try:
json_obj = json.loads(line)
flat_json = self.flatten_json(json_obj)
batch.append(flat_json)
self.stats['success_records'] += 1
# 배치 처리
if len(batch) >= self.batch_size:
data.extend(batch)
batch = []
logger.info(f"배치 처리 완료: {len(data)} 레코드")
except json.JSONDecodeError as e:
logger.warning(f"라인 {line_num} JSON 파싱 오류: {e}")
self.stats['error_records'] += 1
except Exception as e:
logger.error(f"라인 {line_num} 처리 오류: {e}")
self.stats['error_records'] += 1
# 마지막 배치 처리
if batch:
data.extend(batch)
return data
def parse_json_array(self, file_path: str) -> List[Dict[str, Any]]:
"""JSON 배열 형식 파싱"""
with open(file_path, 'r', encoding='utf-8') as f:
json_array = json.load(f)
if not isinstance(json_array, list):
raise ValueError("JSON 파일이 배열 형식이 아닙니다.")
data = []
for i, item in enumerate(json_array):
try:
flat_json = self.flatten_json(item)
data.append(flat_json)
self.stats['success_records'] += 1
except Exception as e:
logger.warning(f"배열 인덱스 {i} 처리 오류: {e}")
self.stats['error_records'] += 1
return data
def parse_single_json(self, file_path: str) -> List[Dict[str, Any]]:
"""단일 JSON 객체 파싱"""
with open(file_path, 'r', encoding='utf-8') as f:
json_obj = json.load(f)
flat_json = self.flatten_json(json_obj)
self.stats['success_records'] += 1
return [flat_json]
def convert_to_dataframe(self, data: List[Dict[str, Any]]) -> pd.DataFrame:
"""데이터를 DataFrame으로 변환"""
if not data:
return pd.DataFrame()
# 메모리 효율적인 DataFrame 생성
df = pd.DataFrame(data)
# 데이터 타입 최적화
for col in df.columns:
if df[col].dtype == 'object':
# 문자열 길이가 짧으면 category로 변환 (메모리 절약)
if df[col].astype(str).str.len().max() < 50:
unique_ratio = df[col].nunique() / len(df)
if unique_ratio < 0.5: # 중복이 많으면 category로 변환
df[col] = df[col].astype('category')
return df
def save_output(self, df: pd.DataFrame, output_path: str, format_type: str = 'excel'):
"""결과 저장"""
if format_type.lower() == 'csv':
df.to_csv(output_path, index=False, encoding='utf-8-sig')
else:
# Excel 저장 시 시트 크기 제한 고려
if len(df) > 1048576: # Excel 최대 행 수
logger.warning("데이터가 Excel 한계를 초과합니다. 여러 시트로 분할합니다.")
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
chunk_size = 1000000
for i in range(0, len(df), chunk_size):
chunk = df.iloc[i:i+chunk_size]
sheet_name = f'Sheet_{i//chunk_size + 1}'
chunk.to_excel(writer, sheet_name=sheet_name, index=False)
else:
df.to_excel(output_path, index=False, engine='openpyxl')
def convert_file(self, input_path: str, output_path: str = None,
output_format: str = 'excel', force_format: str = None) -> bool:
"""
메인 변환 함수
Args:
input_path: 입력 JSON 파일 경로
output_path: 출력 파일 경로 (None이면 자동 생성)
output_format: 출력 형식 ('excel' 또는 'csv')
force_format: 강제로 지정할 JSON 형식
Returns:
bool: 변환 성공 여부
"""
start_time = time.time()
try:
# 입력 파일 검증
if not os.path.isfile(input_path):
logger.error(f"파일이 존재하지 않습니다: {input_path}")
return False
# JSON 형식 감지
if force_format:
json_format = force_format
logger.info(f"강제 지정된 JSON 형식: {json_format}")
else:
json_format = self.detect_json_format(input_path)
logger.info(f"감지된 JSON 형식: {json_format}")
# 출력 경로 생성
if not output_path:
input_file = Path(input_path)
suffix = '.xlsx' if output_format.lower() == 'excel' else '.csv'
output_path = input_file.with_suffix(suffix)
# JSON 파싱
logger.info(f"JSON 파일 파싱 시작: {input_path}")
if json_format == 'jsonl':
data = self.parse_jsonl(input_path)
elif json_format == 'json_array':
data = self.parse_json_array(input_path)
elif json_format == 'single_json':
data = self.parse_single_json(input_path)
else:
# 알 수 없는 형식은 모든 방법 시도
logger.info("알 수 없는 형식입니다. 모든 파싱 방법을 시도합니다.")
try:
data = self.parse_jsonl(input_path)
except:
try:
data = self.parse_json_array(input_path)
except:
data = self.parse_single_json(input_path)
self.stats['total_records'] = self.stats['success_records'] + self.stats['error_records']
if not data:
logger.warning("파싱된 데이터가 없습니다.")
return False
# DataFrame 변환
logger.info("DataFrame 변환 중...")
df = self.convert_to_dataframe(data)
# 결과 저장
logger.info(f"결과 저장 중: {output_path}")
self.save_output(df, str(output_path), output_format)
# 통계 출력
self.stats['processing_time'] = time.time() - start_time
self.print_stats(str(output_path), df.shape)
return True
except Exception as e:
logger.error(f"변환 중 오류 발생: {e}")
return False
def print_stats(self, output_path: str, df_shape: tuple):
"""변환 통계 출력"""
print("\n" + "="*60)
print("🎉 변환 완료!")
print("="*60)
print(f"📁 출력 파일: {output_path}")
print(f"📊 데이터 크기: {df_shape[0]:,} 행 × {df_shape[1]:,} 열")
print(f"✅ 성공 레코드: {self.stats['success_records']:,}")
print(f"❌ 오류 레코드: {self.stats['error_records']:,}")
print(f"⏱️ 처리 시간: {self.stats['processing_time']:.2f}초")
print(f"⚡ 처리 속도: {self.stats['success_records']/self.stats['processing_time']:.0f} 레코드/초")
print("="*60)
def main():
parser = argparse.ArgumentParser(
description='고성능 JSON to Excel/CSV 변환기',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
사용 예시:
python json_converter.py data.json # Excel로 변환
python json_converter.py data.jsonl -f csv # CSV로 변환
python json_converter.py data.json -o result.xlsx # 출력 파일명 지정
python json_converter.py data.json --format jsonl # JSON 형식 강제 지정
python json_converter.py data.json --depth 5 # 플래튼 깊이 제한
python json_converter.py large.json --batch 5000 # 배치 크기 조정
지원하는 JSON 형식:
- JSON Lines (.jsonl): 라인별 JSON 객체
- JSON Array: JSON 객체 배열
- Single JSON: 단일 JSON 객체
"""
)
parser.add_argument('input_file', help='입력 JSON 파일 경로')
parser.add_argument('-o', '--output', help='출력 파일 경로 (자동 생성됨)')
parser.add_argument('-f', '--format', choices=['excel', 'csv'],
default='excel', help='출력 형식 (기본값: excel)')
parser.add_argument('--json-format', choices=['jsonl', 'json_array', 'single_json'],
help='JSON 형식 강제 지정 (자동 감지됨)')
parser.add_argument('--depth', type=int, default=10,
help='JSON 플래튼 최대 깊이 (기본값: 10)')
parser.add_argument('--batch', type=int, default=1000,
help='배치 처리 크기 (기본값: 1000)')
parser.add_argument('--verbose', action='store_true',
help='상세 로그 출력')
args = parser.parse_args()
# 로깅 레벨 설정
if args.verbose:
logging.getLogger().setLevel(logging.DEBUG)
# 변환기 생성 및 실행
converter = JSONConverter(max_depth=args.depth, batch_size=args.batch)
success = converter.convert_file(
input_path=args.input_file,
output_path=args.output,
output_format=args.format,
force_format=args.json_format
)
sys.exit(0 if success else 1)
if __name__ == "__main__":
main()