Merge branch 'worktree-agent-a36803ff'

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)