236 lines
9.1 KiB
Python
236 lines
9.1 KiB
Python
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 |