web-page-backend/realestate-lab/app/matcher.py

import json
import logging
from typing import Dict, Any, List

from .db import _conn, _profile_row_to_dict

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순위")

    # 특별공급 — 신혼부부
    # 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)와 사유를 계산한다."""
    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():
    """프로필 기반 매칭을 실행하여 결과를 저장한다.
    단일 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("매칭 완료")