157 lines
5.3 KiB
Python
157 lines
5.3 KiB
Python
import json
|
|
import logging
|
|
from typing import Dict, Any, List
|
|
|
|
from .db import get_profile, save_match_result, _conn
|
|
|
|
logger = logging.getLogger("realestate-lab")
|
|
|
|
# house_secd → 주택유형 이름 매핑
|
|
_HOUSE_TYPE_MAP = {
|
|
"01": "APT",
|
|
"02": "오피스텔",
|
|
"04": "무순위",
|
|
"09": "민간사전청약",
|
|
"10": "신혼희망타운",
|
|
}
|
|
|
|
|
|
def _check_eligible_types(profile: Dict[str, Any], ann: Dict[str, Any]) -> List[str]:
|
|
"""프로필 기반으로 신청 가능한 공급유형 목록을 반환한다."""
|
|
eligible: List[str] = []
|
|
is_homeless = profile.get("is_homeless", False)
|
|
is_speculative = ann.get("is_speculative_area") == "Y"
|
|
required_months = 24 if is_speculative else 12
|
|
subscription_months = profile.get("subscription_months") or 0
|
|
|
|
# 일반공급
|
|
if is_homeless and profile.get("is_householder") and subscription_months >= required_months:
|
|
eligible.append("일반1순위")
|
|
elif is_homeless:
|
|
eligible.append("일반2순위")
|
|
|
|
# 특별공급
|
|
if profile.get("is_newlywed") and is_homeless:
|
|
eligible.append("특별-신혼부부")
|
|
|
|
if profile.get("is_first_home") and is_homeless:
|
|
eligible.append("특별-생애최초")
|
|
|
|
children_count = profile.get("children_count") or 0
|
|
if children_count >= 2 and is_homeless:
|
|
eligible.append("특별-다자녀")
|
|
|
|
if profile.get("has_dependents") and is_homeless:
|
|
eligible.append("특별-노부모부양")
|
|
|
|
age = profile.get("age") or 0
|
|
if 19 <= age <= 39 and is_homeless:
|
|
eligible.append("특별-청년")
|
|
|
|
if profile.get("has_newborn") and is_homeless:
|
|
eligible.append("특별-신생아")
|
|
|
|
return eligible
|
|
|
|
|
|
def _compute_score(
|
|
profile: Dict[str, Any],
|
|
ann: Dict[str, Any],
|
|
models: List[Dict[str, Any]],
|
|
) -> Dict[str, Any]:
|
|
"""매칭 점수(0-100)와 사유를 계산한다."""
|
|
score = 0
|
|
reasons: List[str] = []
|
|
|
|
# 1. 지역 (30점)
|
|
preferred_regions = profile.get("preferred_regions") or []
|
|
region_name = ann.get("region_name") or ""
|
|
if region_name and any(r in region_name for r in preferred_regions):
|
|
score += 30
|
|
reasons.append(f"선호 지역 일치: {region_name}")
|
|
|
|
# 2. 주택유형 (10점)
|
|
preferred_types = profile.get("preferred_types") or []
|
|
house_secd = ann.get("house_secd") or ""
|
|
type_name = _HOUSE_TYPE_MAP.get(house_secd, house_secd)
|
|
if type_name and type_name in preferred_types:
|
|
score += 10
|
|
reasons.append(f"선호 유형 일치: {type_name}")
|
|
|
|
# 3. 면적 (15점)
|
|
min_area = profile.get("min_area")
|
|
max_area = profile.get("max_area")
|
|
if min_area is not None and max_area is not None and models:
|
|
for m in models:
|
|
supply_area = m.get("supply_area")
|
|
if supply_area is not None and min_area <= supply_area <= max_area:
|
|
score += 15
|
|
reasons.append(f"희망 면적 범위 내 모델 존재 ({supply_area}㎡)")
|
|
break
|
|
|
|
# 4. 가격 (15점)
|
|
max_price = profile.get("max_price")
|
|
if max_price is not None and models:
|
|
for m in models:
|
|
top_amount = m.get("top_amount")
|
|
if top_amount is not None and top_amount <= max_price:
|
|
score += 15
|
|
reasons.append(f"예산 범위 내 모델 존재 (최고가 {top_amount:,}만원)")
|
|
break
|
|
|
|
# 5. 자격 (30점)
|
|
eligible_types = _check_eligible_types(profile, ann)
|
|
eligibility_score = min(len(eligible_types) * 10, 30)
|
|
if eligibility_score > 0:
|
|
score += eligibility_score
|
|
reasons.append(f"자격 유형 {len(eligible_types)}개: {', '.join(eligible_types)}")
|
|
|
|
return {
|
|
"match_score": score,
|
|
"match_reasons": reasons,
|
|
"eligible_types": eligible_types,
|
|
}
|
|
|
|
|
|
def run_matching():
|
|
"""프로필 기반 매칭을 실행하여 결과를 저장한다."""
|
|
profile = get_profile()
|
|
if not profile:
|
|
logger.info("매칭 스킵: 프로필이 설정되지 않음")
|
|
return
|
|
|
|
with _conn() as conn:
|
|
anns = conn.execute(
|
|
"SELECT * FROM announcements WHERE status IN ('청약예정', '청약중')"
|
|
).fetchall()
|
|
|
|
saved = 0
|
|
for row in anns:
|
|
ann = {c: row[c] for c in row.keys()}
|
|
|
|
models_rows = conn.execute(
|
|
"SELECT * FROM announcement_models WHERE house_manage_no = ? AND pblanc_no = ?",
|
|
(ann["house_manage_no"], ann["pblanc_no"]),
|
|
).fetchall()
|
|
models = [{c: m[c] for c in m.keys()} for m in models_rows]
|
|
|
|
result = _compute_score(profile, ann, models)
|
|
|
|
if result["match_score"] > 0:
|
|
save_match_result({
|
|
"announcement_id": ann["id"],
|
|
"model_id": None,
|
|
"match_score": result["match_score"],
|
|
"match_reasons": result["match_reasons"],
|
|
"eligible_types": result["eligible_types"],
|
|
})
|
|
saved += 1
|
|
|
|
# 완료/결과발표 공고의 매칭 결과 정리
|
|
conn.execute(
|
|
"DELETE FROM match_results WHERE announcement_id NOT IN "
|
|
"(SELECT id FROM announcements WHERE status IN ('청약예정', '청약중'))"
|
|
)
|
|
|
|
logger.info("매칭 완료: %d건 공고 중 %d건 매칭됨", len(anns), saved)
|