import os import requests import xml.etree.ElementTree as ET from datetime import datetime from typing import List, Dict, Optional from dotenv import load_dotenv load_dotenv() class PublicDataAPIClient: """국토교통부 공공데이터 API 클라이언트""" def __init__(self): """API 키 및 기본 설정 초기화""" self.base_url = "http://apis.data.go.kr/1613000" # API 엔드포인트 매핑 self.endpoints = { # 아파트 ("아파트", "매매"): { "url": f"{self.base_url}/RTMSDataSvcAptTrade/getRTMSDataSvcAptTrade", "key": os.getenv("APT_TRADE_API_KEY") }, ("아파트", "전세"): { "url": f"{self.base_url}/RTMSDataSvcAptRent/getRTMSDataSvcAptRent", "key": os.getenv("APT_RENT_API_KEY") }, ("아파트", "월세"): { "url": f"{self.base_url}/RTMSDataSvcAptRent/getRTMSDataSvcAptRent", "key": os.getenv("APT_RENT_API_KEY") }, # 오피스텔 ("오피스텔", "매매"): { "url": f"{self.base_url}/RTMSDataSvcOffiTrade/getRTMSDataSvcOffiTrade", "key": os.getenv("OFFI_TRADE_API_KEY") }, ("오피스텔", "전세"): { "url": f"{self.base_url}/RTMSDataSvcOffiRent/getRTMSDataSvcOffiRent", "key": os.getenv("OFFI_RENT_API_KEY") }, ("오피스텔", "월세"): { "url": f"{self.base_url}/RTMSDataSvcOffiRent/getRTMSDataSvcOffiRent", "key": os.getenv("OFFI_RENT_API_KEY") }, # 빌라/연립 ("빌라", "매매"): { "url": f"{self.base_url}/RTMSDataSvcSHTrade/getRTMSDataSvcSHTrade", "key": os.getenv("SH_TRADE_API_KEY") }, ("빌라", "전세"): { "url": f"{self.base_url}/RTMSDataSvcSHRent/getRTMSDataSvcSHRent", "key": os.getenv("SH_RENT_API_KEY") }, ("빌라", "월세"): { "url": f"{self.base_url}/RTMSDataSvcSHRent/getRTMSDataSvcSHRent", "key": os.getenv("SH_RENT_API_KEY") }, ("연립주택", "매매"): { "url": f"{self.base_url}/RTMSDataSvcSHTrade/getRTMSDataSvcSHTrade", "key": os.getenv("SH_TRADE_API_KEY") }, ("연립주택", "전세"): { "url": f"{self.base_url}/RTMSDataSvcSHRent/getRTMSDataSvcSHRent", "key": os.getenv("SH_RENT_API_KEY") }, # 주택 ("주택", "매매"): { "url": f"{self.base_url}/RTMSDataSvcSHHouseTrade/getRTMSDataSvcSHHouseTrade", "key": os.getenv("HOUSE_TRADE_API_KEY") }, ("주택", "전세"): { "url": f"{self.base_url}/RTMSDataSvcSHHouseRent/getRTMSDataSvcSHHouseRent", "key": os.getenv("HOUSE_RENT_API_KEY") }, ("단독주택", "매매"): { "url": f"{self.base_url}/RTMSDataSvcSHHouseTrade/getRTMSDataSvcSHHouseTrade", "key": os.getenv("HOUSE_TRADE_API_KEY") }, ("다가구주택", "매매"): { "url": f"{self.base_url}/RTMSDataSvcSHHouseTrade/getRTMSDataSvcSHHouseTrade", "key": os.getenv("HOUSE_TRADE_API_KEY") } } def get_real_estate_data( self, property_type: str, transaction_type: str, region_code: str, deal_ymd: Optional[str] = None ) -> List[Dict]: """ 부동산 실거래가 데이터 조회 Args: property_type: 매물 형태 (아파트, 오피스텔, 빌라 등) transaction_type: 거래 유형 (매매, 전세, 월세) region_code: 지역 코드 (5자리) deal_ymd: 계약년월 (YYYYMM 형식, 기본값: 현재 월) Returns: 실거래가 데이터 리스트 """ # 기본값 설정 if not deal_ymd: deal_ymd = datetime.now().strftime("%Y%m") # 매물 형태 정규화 property_type = self._normalize_property_type(property_type) transaction_type = self._normalize_transaction_type(transaction_type) print(property_type) print(transaction_type) # 엔드포인트 찾기 endpoint_info = self.endpoints.get((property_type, transaction_type)) if not endpoint_info: return [] if not endpoint_info["key"] or endpoint_info["key"] == "your-api-key-here": print(f"Warning: API key not configured for {property_type} {transaction_type}") return [] # API 호출 params = { "serviceKey": endpoint_info["key"], "LAWD_CD": region_code, "DEAL_YMD": deal_ymd, "pageNo": 1, "numOfRows": 100 } try: response = requests.get(endpoint_info["url"], params=params, timeout=10) response.raise_for_status() # XML 파싱 return self._parse_xml_response(response.text, property_type, transaction_type) except Exception as e: print(f"Error fetching data: {e}") return [] def _normalize_property_type(self, property_type: str) -> str: """매물 형태 정규화""" if not property_type: return "아파트" property_type = property_type.strip().lower() # 매핑 테이블 mappings = { "아파트": ["아파트", "apt", "apartment"], "오피스텔": ["오피스텔", "officetel", "오피스"], "빌라": ["빌라", "villa", "빌딩"], "연립주택": ["연립", "연립주택"], "주택": ["주택", "house"], "단독주택": ["단독", "단독주택"], "다가구주택": ["다가구", "다가구주택"] } for normalized, variants in mappings.items(): for variant in variants: if variant in property_type: return normalized return "아파트" # 기본값 def _normalize_transaction_type(self, transaction_type: str) -> str: """거래 유형 정규화""" if not transaction_type: return "매매" transaction_type = transaction_type.strip().lower() if "매매" in transaction_type or "매도" in transaction_type: return "매매" elif "전세" in transaction_type: return "전세" elif "월세" in transaction_type or "렌트" in transaction_type: return "월세" return "매매" # 기본값 def _parse_xml_response(self, xml_text: str, property_type: str, transaction_type: str) -> List[Dict]: """XML 응답 파싱""" try: root = ET.fromstring(xml_text) # 응답 코드 확인 result_code = root.find(".//resultCode") items = root.findall(".//item") results = [] for item in items: data = {} # 공통 필드 data["거래유형"] = transaction_type data["매물형태"] = property_type data["년"] = self._get_xml_value(item, "dealYear") data["월"] = self._get_xml_value(item, "dealMonth") data["일"] = self._get_xml_value(item, "dealDay") data["법정동"] = self._get_xml_value(item, "umdNm") data["지번"] = self._get_xml_value(item, "jibun") data["전용면적"] = self._get_xml_value(item, "excluUseAr") data["층"] = self._get_xml_value(item, "floor") # 아파트/오피스텔 특화 필드 if property_type in ["아파트", "오피스텔"]: data["아파트"] = self._get_xml_value(item, "aptNm") data["건축년도"] = self._get_xml_value(item, "buildYear") # 전월세 특화 필드 if transaction_type in ["전세", "월세"]: data["보증금액"] = self._get_xml_value(item, "deposit") data["월세금액"] = self._get_xml_value(item, "monthlyRent") if transaction_type in ["매매"]: data["거래금액"] = self._get_xml_value(item, "dealAmount") results.append(data) return results except Exception as e: print(f"Error parsing XML: {e}") return [] def _get_xml_value(self, element, tag: str) -> Optional[str]: """XML 요소에서 값 추출""" found = element.find(tag) if found is not None and found.text: return found.text.strip() return None