import json import logging from typing import Dict, Any, List from .db import _conn, _profile_row_to_dict logger = logging.getLogger("realestate-lab") TIER_WEIGHTS = {"S": 1.00, "A": 0.80, "B": 0.60, "C": 0.40, "D": 0.20} def _region_score(profile: Dict[str, Any], ann: Dict[str, Any]) -> tuple[int, list[str]]: """지역 점수 계산. 광역 10점 + 자치구 5티어 가중치 0~25점. 자치구 기준 미설정 시 광역 매칭만으로 35점 풀 점수(기존 호환). """ region_name = ann.get("region_name") or "" district = ann.get("district") or "" preferred_regions = profile.get("preferred_regions") or [] preferred_districts = profile.get("preferred_districts") or {} region_match = bool(region_name and any(r in region_name for r in preferred_regions)) if not region_match: return 0, [] has_districts = any(preferred_districts.get(t) for t in TIER_WEIGHTS) if not has_districts: return 35, [f"선호 지역 일치: {region_name}"] score = 10 reasons = [f"광역 일치: {region_name}"] for tier, weight in TIER_WEIGHTS.items(): if district and district in (preferred_districts.get(tier) or []): tier_score = round(25 * weight) score += tier_score reasons.append(f"자치구 {tier}티어: {district} (+{tier_score})") break return score, reasons def _eligibility_score(eligible_types: List[str]) -> int: """자격 점수 0~25. 첫 자격 15점 + 추가 자격당 5점, 최대 +10.""" if not eligible_types: return 0 return 15 + min((len(eligible_types) - 1) * 5, 10) # 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순위") # 특별공급 — 신혼부부 # NOTE: 소득기준 검증은 향후 구현 예정 (income_level 필드 활용) 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)와 사유를 계산한다. 배분: 지역 35 / 유형 10 / 면적 15 / 가격 15 / 자격 25. """ score = 0 reasons: List[str] = [] # 1. 지역 (35점) — 광역 + 자치구 5티어 region_score, region_reasons = _region_score(profile, ann) score += region_score reasons.extend(region_reasons) # 2. 주택유형 (10점) — binary 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점) — binary, 범위 안 모델 1개라도 있으면 통과 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점) — binary, 예산 이하 모델 1개라도 있으면 통과 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. 자격 (25점) — 첫 자격 15 + 추가당 5 eligible_types = _check_eligible_types(profile, ann) elig_score = _eligibility_score(eligible_types) if elig_score > 0: score += elig_score reasons.append(f"자격 유형 {len(eligible_types)}개: {', '.join(eligible_types)}") return { "match_score": score, "match_reasons": reasons, "eligible_types": eligible_types, } def run_matching(): """프로필 기반 매칭을 실행하여 결과를 저장한다. 단일 connection으로 프로필 조회 + 매칭 + 저장을 처리하여 DB lock 방지. """ with _conn() as conn: profile_row = conn.execute("SELECT * FROM user_profile WHERE id = 1").fetchone() if not profile_row: logger.info("프로필 미설정 — 매칭 건너뜀") return profile = _profile_row_to_dict(profile_row) anns = conn.execute( "SELECT * FROM announcements WHERE status IN ('청약예정', '청약중')" ).fetchall() for ann_row in anns: ann = {c: ann_row[c] for c in ann_row.keys()} models = conn.execute( "SELECT * FROM announcement_models WHERE house_manage_no = ? AND pblanc_no = ?", (ann["house_manage_no"], ann["pblanc_no"]), ).fetchall() model_list = [dict(m) for m in models] result = _compute_score(profile, ann, model_list) if result["match_score"] > 0: conn.execute(""" INSERT INTO match_results (announcement_id, model_id, match_score, match_reasons, eligible_types, is_new) VALUES (?, ?, ?, ?, ?, 1) ON CONFLICT(announcement_id, model_id) DO UPDATE SET match_score=excluded.match_score, match_reasons=excluded.match_reasons, eligible_types=excluded.eligible_types """, ( ann["id"], None, result["match_score"], json.dumps(result["match_reasons"]), json.dumps(result["eligible_types"]), )) # Clean up stale match results for completed announcements conn.execute( "DELETE FROM match_results WHERE announcement_id NOT IN " "(SELECT id FROM announcements WHERE status IN ('청약예정', '청약중'))" ) logger.info("매칭 완료")