From 437838c28bd85e7a3a4184f2e3daeb9ec24a5d50 Mon Sep 17 00:00:00 2001 From: gahusb Date: Mon, 6 Apr 2026 08:29:36 +0900 Subject: [PATCH] =?UTF-8?q?feat(realestate-lab):=20=ED=94=84=EB=A1=9C?= =?UTF-8?q?=ED=95=84=20=EA=B8=B0=EB=B0=98=20=EB=A7=A4=EC=B9=AD=20=EC=97=94?= =?UTF-8?q?=EC=A7=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 --- realestate-lab/app/matcher.py | 152 ++++++++++++++++++++++++++++++++++ 1 file changed, 152 insertions(+) create mode 100644 realestate-lab/app/matcher.py diff --git a/realestate-lab/app/matcher.py b/realestate-lab/app/matcher.py new file mode 100644 index 0000000..297e101 --- /dev/null +++ b/realestate-lab/app/matcher.py @@ -0,0 +1,152 @@ +import json +import logging +from typing import Dict, Any, List + +from .db import get_profile, save_match_result, clear_match_results, _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 + + clear_match_results() + + 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 + + logger.info("매칭 완료: %d건 공고 중 %d건 매칭됨", len(anns), saved)