src/utils/common.py

"""
여러 모듈에서 공통으로 사용하는 유틸리티 함수
"""

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>', '&lt;script&gt;') \
                    .replace('</script>', '&lt;/script&gt;')
    
    # 필요한 경우 추가 정제 로직 구현
    
    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)