202 lines
7.5 KiB
Python
202 lines
7.5 KiB
Python
import json
|
|
import logging
|
|
from typing import Dict, Any, List
|
|
|
|
from .db import _conn, _profile_row_to_dict
|
|
|
|
logger = logging.getLogger("realestate-lab")
|
|
|
|
TIER_POINTS = {"S": 25, "A": 20, "B": 15, "C": 10, "D": 5}
|
|
|
|
|
|
def _region_score(profile: Dict[str, Any], ann: Dict[str, Any]) -> tuple[int, list[str]]:
|
|
"""지역 점수 계산. 광역 10점 + 자치구 5티어 점수 0~25점.
|
|
preferred_districts에 자치구가 하나라도 등록되면 티어 가중 모드로 동작.
|
|
자치구가 하나도 등록되지 않으면(빈 dict 또는 모든 티어가 빈 리스트) 광역 매칭만으로 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.values())
|
|
if not has_districts:
|
|
return 35, [f"선호 지역 일치: {region_name}"]
|
|
|
|
score = 10
|
|
reasons = [f"광역 일치: {region_name}"]
|
|
for tier, tier_score in TIER_POINTS.items():
|
|
if district and district in (preferred_districts.get(tier) or []):
|
|
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("매칭 완료")
|