232 lines
9.6 KiB
Python
232 lines
9.6 KiB
Python
import os
|
|
import json
|
|
from openai import OpenAI
|
|
from dotenv import load_dotenv
|
|
from models import ParsedRealEstate
|
|
from typing import Optional, Tuple
|
|
|
|
load_dotenv()
|
|
|
|
def get_region_code(location: str) -> Tuple[Optional[str], Optional[str]]:
|
|
"""위치 정보를 시군구 코드로 변환"""
|
|
|
|
# 지역 코드 데이터 로드
|
|
base_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
with open(os.path.join(base_path, 'data', 'region_codes.json'), 'r', encoding='utf-8') as f:
|
|
region_codes = json.load(f)
|
|
|
|
if not location:
|
|
return None, None
|
|
|
|
location = location.strip()
|
|
|
|
# 1. 정확한 매칭 시도 (전체 이름)
|
|
for full_name, code in region_codes.items():
|
|
if location in full_name or full_name in location:
|
|
return code, full_name
|
|
|
|
# 2. 부분 매칭 시도 (구, 시, 군 단위)
|
|
location_parts = location.split()
|
|
for part in location_parts:
|
|
if part.endswith('구') or part.endswith('시') or part.endswith('군'):
|
|
for full_name, code in region_codes.items():
|
|
if part in full_name:
|
|
return code, full_name
|
|
|
|
# 3. 동 이름으로 구 추정
|
|
dong_to_gu = {
|
|
# 서울 강남구
|
|
'역삼': '강남구', '삼성': '강남구', '청담': '강남구', '논현': '강남구', '대치': '강남구',
|
|
'도곡': '강남구', '개포': '강남구', '일원': '강남구', '수서': '강남구', '세곡': '강남구',
|
|
# 서울 서초구
|
|
'반포': '서초구', '서초': '서초구', '방배': '서초구', '양재': '서초구', '잠원': '서초구',
|
|
# 서울 송파구
|
|
'잠실': '송파구', '신천': '송파구', '석촌': '송파구', '송파': '송파구', '가락': '송파구',
|
|
'문정': '송파구', '장지': '송파구', '방이': '송파구', '오금': '송파구',
|
|
# 서울 강동구
|
|
'천호': '강동구', '성내': '강동구', '길동': '강동구', '둔촌': '강동구', '암사': '강동구',
|
|
# 서울 마포구
|
|
'홍대': '마포구', '합정': '마포구', '상수': '마포구', '망원': '마포구', '연남': '마포구',
|
|
# 서울 성동구
|
|
'성수': '성동구', '왕십리': '성동구', '행당': '성동구', '금호': '성동구', '옥수': '성동구',
|
|
# 경기도 성남시 분당구
|
|
'판교': '분당구', '정자': '분당구', '서현': '분당구', '수내': '분당구', '야탑': '분당구',
|
|
'분당': '분당구', '이매': '분당구',
|
|
# 경기도 용인시
|
|
'동백': '기흥구', '보정': '기흥구', '죽전': '수지구', '수지': '수지구',
|
|
# 경기도 고양시
|
|
'일산': '일산동구', '백석': '일산동구', '마두': '일산서구', '주엽': '일산서구'
|
|
}
|
|
|
|
for dong, gu in dong_to_gu.items():
|
|
if dong in location:
|
|
for full_name, code in region_codes.items():
|
|
if gu in full_name:
|
|
return code, full_name
|
|
|
|
return None, None
|
|
|
|
class OpenAIParser:
|
|
"""OpenAI를 이용한 부동산 정보 파싱"""
|
|
|
|
def __init__(self):
|
|
api_key = os.getenv("OPENAI_API_KEY")
|
|
if not api_key or api_key == "your-api-key-here":
|
|
raise ValueError("OpenAI API key not configured in .env file")
|
|
self.client = OpenAI(api_key=api_key)
|
|
|
|
async def parse_real_estate_query(self, text: str) -> ParsedRealEstate:
|
|
"""자연어 텍스트에서 부동산 정보 추출"""
|
|
|
|
system_prompt = """
|
|
당신은 부동산 정보를 추출하는 전문가입니다.
|
|
사용자의 자연어 입력에서 다음 정보를 추출하세요:
|
|
1. price: 가격 (전세금, 월세, 매매가 등)
|
|
2. location: 위치 (지역명, 동, 구 등)
|
|
3. area: 면적 (평수, 제곱미터 등)
|
|
4. rooms: 방 개수 (숫자만)
|
|
5. transaction_type: 거래 유형 (전세, 월세, 매매)
|
|
6. property_type: 매물 형태 (아파트, 오피스텔, 주택, 빌라, 원룸, 투룸, 쓰리룸, 단독주택, 다가구주택, 연립주택, 상가주택 등)
|
|
|
|
JSON 형식으로만 응답하세요.
|
|
정보가 없는 항목은 null로 표시하세요.
|
|
rooms는 숫자(integer)로만 표현하세요.
|
|
"""
|
|
|
|
try:
|
|
response = self.client.chat.completions.create(
|
|
model="gpt-3.5-turbo",
|
|
messages=[
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": text}
|
|
],
|
|
temperature=0.1,
|
|
response_format={"type": "json_object"}
|
|
)
|
|
|
|
result = json.loads(response.choices[0].message.content)
|
|
|
|
# 위치 정보를 시군구 코드로 변환
|
|
region_code = None
|
|
region_name = None
|
|
if result.get("location"):
|
|
region_code, region_name = get_region_code(result.get("location"))
|
|
|
|
return ParsedRealEstate(
|
|
price=result.get("price"),
|
|
location=result.get("location"),
|
|
area=result.get("area"),
|
|
rooms=result.get("rooms"),
|
|
transaction_type=result.get("transaction_type"),
|
|
property_type=result.get("property_type"),
|
|
region_code=region_code,
|
|
region_name=region_name,
|
|
raw_text=text
|
|
)
|
|
|
|
except Exception as e:
|
|
# 에러 발생 시 기본값 반환
|
|
return ParsedRealEstate(
|
|
raw_text=text,
|
|
price=None,
|
|
location=None,
|
|
area=None,
|
|
rooms=None,
|
|
transaction_type=None,
|
|
property_type=None,
|
|
region_code=None,
|
|
region_name=None
|
|
)
|
|
|
|
async def filter_listings(self, user_query: str, listings: list, top_k: int = 10) -> list:
|
|
"""
|
|
OpenAI를 사용하여 실거래가 목록에서 사용자 요구사항과 가장 일치하는 매물 선별
|
|
|
|
Args:
|
|
user_query: 사용자의 원본 검색 쿼리
|
|
listings: 공공데이터 API에서 받은 실거래가 목록
|
|
top_k: 선별할 매물 개수 (기본값: 10)
|
|
|
|
Returns:
|
|
필터링된 매물 목록
|
|
"""
|
|
|
|
if not listings:
|
|
return []
|
|
|
|
# 매물 정보를 간략하게 정리
|
|
simplified_listings = []
|
|
for idx, item in enumerate(listings[:50]): # 최대 50개만 처리 (토큰 제한)
|
|
simplified = {
|
|
"index": idx,
|
|
"거래유형": item.get("거래유형", ""),
|
|
"매물형태": item.get("매물형태", ""),
|
|
"거래금액": item.get("거래금액", ""),
|
|
"보증금액": item.get("보증금액", ""),
|
|
"월세금액": item.get("월세금액", ""),
|
|
"전용면적": item.get("전용면적", ""),
|
|
"층": item.get("층", ""),
|
|
"건축년도": item.get("건축년도", ""),
|
|
"법정동": item.get("법정동", ""),
|
|
"거래일": f"{item.get('년', '')}.{item.get('월', '')}.{item.get('일', '')}"
|
|
}
|
|
# 빈 값 제거
|
|
simplified = {k: v for k, v in simplified.items() if v}
|
|
simplified_listings.append(simplified)
|
|
|
|
system_prompt = """
|
|
당신은 부동산 전문가입니다.
|
|
사용자의 요구사항과 실거래가 데이터를 비교하여 가장 적합한 매물을 선별해주세요.
|
|
|
|
평가 기준:
|
|
1. 가격 조건 일치도 (사용자가 언급한 가격 범위)
|
|
2. 면적 조건 일치도 (평수, 방 개수 등)
|
|
3. 층수 선호도
|
|
4. 건축년도 (신축/구축 선호도)
|
|
5. 위치 적합성
|
|
|
|
JSON 형식으로 응답하세요:
|
|
{
|
|
"selected_indices": [가장 적합한 매물의 index 10개],
|
|
"reason": "선별 이유 간단 설명"
|
|
}
|
|
"""
|
|
|
|
user_prompt = f"""
|
|
사용자 요구사항: {user_query}
|
|
|
|
실거래가 데이터:
|
|
{json.dumps(simplified_listings, ensure_ascii=False, indent=2)}
|
|
|
|
위 데이터에서 사용자 요구사항과 가장 잘 맞는 매물 {top_k}개를 선별해주세요.
|
|
"""
|
|
|
|
try:
|
|
response = self.client.chat.completions.create(
|
|
model="gpt-3.5-turbo",
|
|
messages=[
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": user_prompt}
|
|
],
|
|
temperature=0.1,
|
|
response_format={"type": "json_object"}
|
|
)
|
|
|
|
result = json.loads(response.choices[0].message.content)
|
|
selected_indices = result.get("selected_indices", [])
|
|
|
|
# 선별된 매물 반환
|
|
filtered_listings = []
|
|
for idx in selected_indices:
|
|
if 0 <= idx < len(listings):
|
|
listing = listings[idx].copy()
|
|
listing["match_reason"] = result.get("reason", "")
|
|
listing["rank"] = len(filtered_listings) + 1
|
|
filtered_listings.append(listing)
|
|
|
|
return filtered_listings[:top_k]
|
|
|
|
except Exception as e:
|
|
print(f"Error filtering listings: {e}")
|
|
# 에러 시 원본 데이터의 상위 10개 반환
|
|
return listings[:top_k] |