""" 여러 모듈에서 공통으로 사용하는 유틸리티 함수 """ import os import json import uuid import logging from datetime import datetime from typing import Dict, List, Any, Optional, Union from src import config logger = config.setup_logger("common") def generate_unique_id(prefix: str = "") -> str: """ 고유 ID 생성 Args: prefix: ID 접두사 Returns: 고유 ID 문자열 """ unique_id = str(uuid.uuid4()) if prefix: return f"{prefix}-{unique_id}" return unique_id def save_json_data(data: Any, filename: str, directory: str = None) -> str: """ 데이터를 JSON 파일로 저장 Args: data: 저장할 데이터 (JSON 직렬화 가능) filename: 파일명 directory: 저장 디렉토리 (기본값: config.DATA_DIR) Returns: 저장된 파일의 전체 경로 """ directory = directory or config.DATA_DIR os.makedirs(directory, exist_ok=True) # 타임스탬프 추가 (파일명에 이미 있으면 추가하지 않음) if not any(part.isdigit() and len(part) == 8 for part in filename.split('_')): timestamp = datetime.now().strftime("%Y%m%d") base_name, ext = os.path.splitext(filename) filename = f"{base_name}_{timestamp}{ext}" filepath = os.path.join(directory, filename) with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) logger.info(f"데이터 저장 완료: {filepath}") return filepath def format_error_response(error: Exception) -> Dict[str, Any]: """ 예외를 API 응답 형식으로 변환 Args: error: 예외 객체 Returns: 오류 정보를 포함한 딕셔너리 """ error_code = getattr(error, 'error_code', 'UNKNOWN') error_detail = { "status": "error", "error": { "type": error.__class__.__name__, "code": error_code, "message": str(error), "timestamp": datetime.now().isoformat() } } # 추가 상세 정보가 있으면 포함 if hasattr(error, 'status_code'): error_detail["error"]["status_code"] = error.status_code if hasattr(error, 'response') and error.response: error_detail["error"]["response"] = error.response return error_detail def sanitize_string(text: str) -> str: """ 문자열에서 위험한 문자 또는 코드 삽입 시도 제거 Args: text: 정제할 문자열 Returns: 정제된 문자열 """ if not text: return "" # 기본적인 정제 (줄바꿈, 스크립트 태그 등 처리) sanitized = text.replace('', '</script>') # 필요한 경우 추가 정제 로직 구현 return sanitized def merge_metadata(base: Dict[str, Any], updates: Dict[str, Any]) -> Dict[str, Any]: """ 메타데이터 병합 Args: base: 기본 메타데이터 updates: 업데이트할 메타데이터 Returns: 병합된 메타데이터 """ result = base.copy() if updates: for key, value in updates.items(): if key in result and isinstance(result[key], dict) and isinstance(value, dict): # 중첩된 딕셔너리는 재귀적으로 병합 result[key] = merge_metadata(result[key], value) else: # 그 외에는 덮어쓰기 result[key] = value return result def normalize_doc_type(doc_type: str) -> str: """ 문서 유형 정규화 Args: doc_type: 정규화할 문서 유형 Returns: 정규화된 문서 유형 """ if not doc_type: return "" # 대문자로 변환하고 공백 제거 normalized = doc_type.upper().strip() # 알려진 문서 유형 매핑 mapping = { "KDS": "KDS", "KOREAN DESIGN STANDARD": "KDS", "KCS": "KCS", "KOREAN CONSTRUCTION SPECIFICATION": "KCS" } return mapping.get(normalized, normalized)