From 95fadaa8efb1ffe20001c2fc3b7c2a4c98da0185 Mon Sep 17 00:00:00 2001 From: gahusb Date: Thu, 9 Jul 2026 14:03:54 +0900 Subject: [PATCH] =?UTF-8?q?fix(realestate):=20=EC=8B=9C=EC=84=B8=20?= =?UTF-8?q?=ED=91=9C=EB=B3=B8=20=EA=B4=91=EC=97=AD=ED=8F=B4=EB=B0=B1(M1)+a?= =?UTF-8?q?rea-NULL=20upsert=20skip(M3)+matches=20JSON=ED=8C=8C=EC=8B=B1(M?= =?UTF-8?q?4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01EqCYBhvTcdeCTUDX3RhWx9 --- realestate-lab/app/db.py | 43 +++++++++++++++------ realestate-lab/tests/test_listing_db.py | 50 +++++++++++++++++++++++++ 2 files changed, 82 insertions(+), 11 deletions(-) diff --git a/realestate-lab/app/db.py b/realestate-lab/app/db.py index 7601821..0629210 100644 --- a/realestate-lab/app/db.py +++ b/realestate-lab/app/db.py @@ -10,6 +10,9 @@ logger = logging.getLogger("realestate-lab") DB_PATH = os.getenv("REALESTATE_DB_PATH", "/app/data/realestate.db") +# listing_matcher.MIN_SAMPLE와 동일 값 유지 — 시세 표본이 이 미만이면 광역 폴백. +_MARKET_SAMPLE_MIN = 3 + def _conn(): c = sqlite3.connect(DB_PATH, timeout=120.0) @@ -981,6 +984,10 @@ def upsert_listing(data: Dict[str, Any]) -> tuple: def upsert_market_deal(data: Dict[str, Any]) -> None: + if data.get("area") is None: + # area 없는 실거래는 area BETWEEN 조회에서 절대 매칭 안 돼 median 계산에 무용하고, + # dedup 인덱스도 못 걸려(area가 NOT NULL 컬럼이 아니라 조건 매칭 불가) 무한 중복을 유발한다. + return cols = ("house_type", "dong_code", "dong", "complex_name", "area", "deal_type", "deposit", "monthly_rent", "sale_amount", "deal_ym", "floor", "source") d = {c: data.get(c) for c in cols} @@ -992,16 +999,24 @@ def upsert_market_deal(data: Dict[str, Any]) -> None: def get_market_deals_for(dong_code, complex_name, area, deal_type, months=6) -> List[Dict[str, Any]]: with _conn() as conn: - rows = conn.execute( - """SELECT * FROM market_deals - WHERE dong_code=? AND deal_type=? - AND (complex_name=? OR ? IS NULL OR complex_name IS NULL) - AND area BETWEEN ? AND ? - AND deal_ym >= strftime('%Y%m', 'now', ?)""", - (dong_code, deal_type, complex_name, complex_name, - (area or 0) - 5, (area or 0) + 5, f"-{int(months)} months"), - ).fetchall() - return [dict(r) for r in rows] + def _query(cn): + rows = conn.execute( + """SELECT * FROM market_deals + WHERE dong_code=? AND deal_type=? + AND (complex_name=? OR ? IS NULL OR complex_name IS NULL) + AND area BETWEEN ? AND ? + AND deal_ym >= strftime('%Y%m', 'now', ?)""", + (dong_code, deal_type, cn, cn, + (area or 0) - 5, (area or 0) + 5, f"-{int(months)} months"), + ).fetchall() + return [dict(r) for r in rows] + + deals = _query(complex_name) + if len(deals) < _MARKET_SAMPLE_MIN and complex_name: + # 단지 표본이 부족하면 complex 제약을 풀고 같은 동/거래유형/면적±5/최근성으로 + # 광역 재조회한다. 표본을 늘리기만 하므로 안전(잘못된 판정 유발 X). + deals = _query(None) + return deals def upsert_listing_match(data: Dict[str, Any]) -> None: @@ -1043,7 +1058,13 @@ def get_listing_matches() -> List[Dict[str, Any]]: "SELECT m.*, l.complex_name, l.dong, l.deal_type, l.deposit, l.sale_price, l.area_exclusive, l.url " "FROM listing_matches m JOIN listings l ON l.id=m.listing_id ORDER BY m.created_at DESC" ).fetchall() - return [dict(r) for r in rows] + out = [] + for r in rows: + d = dict(r) + d["regulation_flags"] = json.loads(d.get("regulation_flags") or "[]") + d["reasons"] = json.loads(d.get("reasons") or "[]") + out.append(d) + return out _TIER_ORDER = {"위험": 0, "고가": 0, "주의": 1, "시세": 1, "안전": 2, "저평가": 2, "보류": -1} diff --git a/realestate-lab/tests/test_listing_db.py b/realestate-lab/tests/test_listing_db.py index 8c65ef1..ff39f88 100644 --- a/realestate-lab/tests/test_listing_db.py +++ b/realestate-lab/tests/test_listing_db.py @@ -38,3 +38,53 @@ def test_unnotified_listing_match_and_mark(): assert len(un) == 1 db.mark_listings_notified([un[0]["id"]]) assert db.get_unnotified_listing_matches() == [] + + +# ── 최종 리뷰 fix (M1/M3/M4) ───────────────────────────────────────────────── + +def test_market_deals_broad_fallback_when_complex_sample_thin(): + """complex_name 지정 시 표본이 MIN_SAMPLE(3) 미만이면 complex 제약을 풀고 + 같은 dong_code/deal_type/area±5/recency 조건으로 광역 재조회해야 한다.""" + from app import db + base = {"house_type": "아파트", "dong_code": "11590", "dong": "신대방동", + "area": 42.0, "deal_type": "전세", "sale_amount": None, "floor": "5", "source": "molit"} + # complex "A" 단독 1건 (< MIN_SAMPLE=3) + db.upsert_market_deal({**base, "complex_name": "A", "deposit": 30000, "deal_ym": "202601"}) + # complex "B" 3건 (dedup 인덱스 회피 위해 deal_ym 다르게) + db.upsert_market_deal({**base, "complex_name": "B", "deposit": 31000, "deal_ym": "202602"}) + db.upsert_market_deal({**base, "complex_name": "B", "deposit": 32000, "deal_ym": "202603"}) + db.upsert_market_deal({**base, "complex_name": "B", "deposit": 33000, "deal_ym": "202604"}) + + deals = db.get_market_deals_for("11590", "A", 42.0, "전세", months=120) + assert len(deals) == 4 # A 1건 + B 3건 광역 폴백 + + +def test_upsert_market_deal_area_null_skipped(): + """area=None인 실거래는 median 계산에 무용 + dedup 인덱스도 못 걸려 무한중복 + 유발하므로 저장 자체를 skip해야 한다.""" + from app import db + d = {"house_type": "아파트", "dong_code": "11590", "dong": "신대방동", + "complex_name": "OO", "area": None, "deal_type": "전세", + "deposit": 43000, "sale_amount": None, "deal_ym": "202605", "floor": "5", "source": "molit"} + db.upsert_market_deal(d) + db.upsert_market_deal(d) # 재호출해도 여전히 저장 안 됨(무한중복 방지 확인) + with db._conn() as conn: + cnt = conn.execute("SELECT COUNT(*) FROM market_deals").fetchone()[0] + assert cnt == 0 + + +def test_get_listing_matches_json_parsed(): + """get_listing_matches()도 get_unnotified_listing_matches()와 동일하게 + regulation_flags/reasons를 list로 파싱해 반환해야 한다.""" + from app import db + lid, _ = db.upsert_listing({"article_no": "A3", "source": "naver", "deal_type": "전세", + "deposit": 29000, "area_exclusive": 42.0, "dong": "신대방동"}) + db.upsert_listing_match({"listing_id": lid["id"], "category": "임차", "passed": 1, + "match_score": 90, "safety_tier": "안전", "sample_size": 7, + "regulation_flags": ["토지거래허가구역"], "reasons": ["ok"], "is_new": 1}) + matches = db.get_listing_matches() + assert len(matches) == 1 + assert isinstance(matches[0]["regulation_flags"], list) + assert matches[0]["regulation_flags"] == ["토지거래허가구역"] + assert isinstance(matches[0]["reasons"], list) + assert matches[0]["reasons"] == ["ok"]