""" 한국건설기준센터(KCSC) API 클라이언트 모듈 """ import time import json import logging from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Dict, List, Any, Optional, Union, Tuple from src import config from src.utils import ( HTTPClient, APIError, ConfigError, DataProcessingError, save_json_data, normalize_doc_type ) logger = config.setup_logger("kcsc_api") class KCSCApiClient: """ 한국건설기준센터(KCSC) API와 통신하는 클라이언트 """ def __init__(self, api_key: str = None, base_url: str = None): """ KCSC API 클라이언트 초기화 Args: api_key: KCSC API 키 base_url: KCSC API 기본 URL """ self.api_key = api_key or config.KCSC_API_KEY if not self.api_key: raise ConfigError("KCSC API 키가 설정되지 않았습니다. .env 파일을 확인하세요.") self.base_url = base_url or "https://www.kcsc.re.kr/OpenApi" self.data_dir = config.DATA_DIR # HTTP 클라이언트 초기화 self.http_client = HTTPClient( base_url=self.base_url, default_headers={ "Accept": "application/json", "Content-Type": "application/json" } ) logger.info(f"KCSC API 클라이언트 초기화 완료 (기본 URL: {self.base_url})") def get_code_list(self, doc_type: str = "KDS") -> List[Dict[str, Any]]: """ 코드 목록 가져오기 Args: doc_type: 문서 유형 (예: "KDS", "KCS") Returns: 코드 목록 Raises: APIError: API 요청 실패 시 """ doc_type = normalize_doc_type(doc_type) params = { "Type": doc_type, "key": self.api_key } try: logger.info(f"{doc_type} 코드 목록 요청 중...") response = self.http_client.get("CodeList", params=params) data = response.json() if "List" not in data: raise APIError(f"{doc_type} 코드 목록에 'List' 필드가 없습니다.", response=data) code_list = data["List"] logger.info(f"{doc_type} 코드 목록 요청 성공 (총 {len(code_list)}개 항목)") return code_list except APIError: raise except Exception as e: raise APIError(f"{doc_type} 코드 목록 요청 실패: {str(e)}") def get_code_details(self, doc_type: str, code: str) -> Dict[str, Any]: """ 코드 상세 정보 가져오기 Args: doc_type: 문서 유형 (예: "KDS", "KCS") code: 코드 번호 Returns: 코드 상세 정보 Raises: APIError: API 요청 실패 시 """ doc_type = normalize_doc_type(doc_type) params = { "key": self.api_key } try: logger.info(f"{doc_type}/{code} 상세 정보 요청 중...") endpoint = f"CodeViewer/{doc_type}/{code}" response = self.http_client.get(endpoint, params=params) data = response.json() logger.info(f"{doc_type}/{code} 상세 정보 요청 성공") # 응답에 문서 유형 정보 추가 if data and isinstance(data, dict): data["DocType"] = doc_type return data except APIError: raise except Exception as e: raise APIError(f"{doc_type}/{code} 상세 정보 요청 실패: {str(e)}") def collect_codes(self, doc_types: List[str] = None, save: bool = True) -> Dict[str, List[Dict[str, Any]]]: """ 문서 유형별 코드 수집 Args: doc_types: 수집할 문서 유형 목록 save: 수집한 데이터를 파일로 저장할지 여부 Returns: 문서 유형별 코드 목록 딕셔너리 Raises: APIError: API 요청 실패 시 """ if doc_types is None: doc_types = ["KDS", "KCS"] all_codes = {} for doc_type in doc_types: normalized_type = normalize_doc_type(doc_type) logger.info(f"{normalized_type} 코드 수집 시작") try: code_list = self.get_code_list(normalized_type) if not code_list: logger.warning(f"{normalized_type} 코드 목록이 비어 있습니다.") continue all_codes[normalized_type] = code_list if save: filename = f"{normalized_type}_codes.json" save_json_data(code_list, filename) except Exception as e: logger.error(f"{normalized_type} 코드 수집 중 오류 발생: {str(e)}") return all_codes def fetch_code_details( self, doc_type: str, codes: List[Dict[str, Any]], save: bool = True, max_workers: int = 5 ) -> List[Dict[str, Any]]: """ 특정 문서 유형의 여러 코드에 대한 상세 정보 가져오기 Args: doc_type: 문서 유형 codes: 코드 목록 save: 수집한 데이터를 파일로 저장할지 여부 max_workers: 병렬 처리할 최대 작업자 수 Returns: 코드 상세 정보 목록 Raises: APIError: API 요청 실패 시 """ doc_type = normalize_doc_type(doc_type) details = [] failed_codes = [] logger.info(f"{doc_type} 상세 정보 수집 시작 (총 {len(codes)}개 코드)") # 병렬 처리로 여러 코드 정보 수집 with ThreadPoolExecutor(max_workers=max_workers) as executor: # 각 코드에 대한 작업 생성 future_to_code = { executor.submit(self._fetch_single_code_detail, doc_type, code.get("Code")): code for code in codes if code.get("Code") } # 작업 결과 처리 for i, future in enumerate(as_completed(future_to_code)): code = future_to_code[future] try: detail = future.result() if detail: details.append(detail) # 진행 상황 로깅 (10% 단위) completion = (i + 1) / len(codes) * 100 if completion % 10 < 100 / len(codes): logger.info(f"{doc_type} 상세 정보 수집 진행 중: {i+1}/{len(codes)} ({completion:.1f}%)") except Exception as e: logger.error(f"{doc_type}/{code.get('Code')} 상세 정보 수집 실패: {str(e)}") failed_codes.append(code.get("Code")) logger.info(f"{doc_type} 상세 정보 수집 완료: {len(details)}개 성공, {len(failed_codes)}개 실패") if failed_codes: logger.warning(f"실패한 코드: {', '.join(failed_codes[:10])}{' 외 {}개'.format(len(failed_codes)-10) if len(failed_codes) > 10 else ''}") if save and details: filename = f"{doc_type}_details.json" save_json_data(details, filename) return details def _fetch_single_code_detail(self, doc_type: str, code: str) -> Optional[Dict[str, Any]]: """ 단일 코드 상세 정보 가져오기 (내부 메서드) Args: doc_type: 문서 유형 code: 코드 번호 Returns: 코드 상세 정보 또는 None (오류 시) """ try: # API 요청 detail = self.get_code_details(doc_type, code) # API 서버 부하 방지를 위한 지연 time.sleep(config.API_RETRY_DELAY) return detail except Exception as e: logger.error(f"{doc_type}/{code} 상세 정보 가져오기 실패: {str(e)}") return None # 실행 예시 if __name__ == "__main__": # 설정 초기화 config.init() # API 클라이언트 초기화 kcsc = KCSCApiClient() # 코드 목록 수집 all_codes = kcsc.collect_codes() # 코드별 상세 정보 수집 for doc_type, codes in all_codes.items(): logger.info(f"{doc_type} 상세 정보 수집 시작 (총 {len(codes)}개 코드)") kcsc.fetch_code_details(doc_type, codes)