perf(realestate): market_deals 배치 upsert(단일 conn)로 collect 속도 개선

realestate-lab/app/db.py

@@ -997,6 +997,28 @@ def upsert_market_deal(data: Dict[str, Any]) -> None:
                          VALUES ({','.join(':'+c for c in cols)})""", d)
 
 
+def bulk_upsert_market_deals(rows: List[Dict[str, Any]]) -> int:
+    """market_deals 다건 INSERT OR IGNORE (단일 connection). area=None은 skip(무용/bloat 방지).
+    저장시도 건수 반환. 기존 upsert_market_deal과 동일 dedup(ux_market_deals_dedup 인덱스)."""
+    cols = ("house_type", "dong_code", "dong", "complex_name", "area", "deal_type",
+            "deposit", "monthly_rent", "sale_amount", "deal_ym", "floor", "source")
+    payload = []
+    for data in rows:
+        if data.get("area") is None:
+            continue
+        d = {c: data.get(c) for c in cols}
+        d["source"] = data.get("source") or "molit"
+        payload.append(tuple(d[c] for c in cols))
+    if not payload:
+        return 0
+    with _conn() as conn:
+        conn.executemany(
+            f"INSERT OR IGNORE INTO market_deals ({','.join(cols)}) VALUES ({','.join(['?'] * len(cols))})",
+            payload,
+        )
+    return len(payload)
+
+
 def get_market_deals_for(dong_code, complex_name, area, deal_type, months=6) -> List[Dict[str, Any]]:
     with _conn() as conn:
         def _query(cn):

realestate-lab/app/listing_collector.py

@@ -7,8 +7,8 @@ import requests
 from datetime import date
 
 from .lawd_codes import LAWD_CODES
-from .db import (get_listing_criteria, upsert_listing, upsert_market_deal,
-                 save_listing_collect_log)
+from .db import (get_listing_criteria, upsert_listing,
+                 bulk_upsert_market_deals, save_listing_collect_log)
 
 logger = logging.getLogger("realestate-lab")
 
@@ -108,7 +108,7 @@ def _molit_call(op: str, lawd: str, ym: str):
 def _fetch_molit_all():
     """대상 동 시군구코드 × 유형 × 최근 2개월 전월세·매매 실거래 수집→market_deals.
     Returns (saved_count, diag). diag = 'calls=N saved=M [err=<첫에러>]' 관측용."""
-    saved = 0
+    all_deals = []
     calls = 0
     first_err = None
     lawds = set(LAWD_CODES.values())
@@ -128,11 +128,11 @@ def _fetch_molit_all():
                             d = parser(raw, htype)
                             d["dong_code"] = lawd
                             d["dong"] = (raw.get("umdNm") or "").strip() or None
-                            upsert_market_deal(d)
-                            saved += 1
+                            all_deals.append(d)
                         except Exception as e:
                             logger.warning("MOLIT 파싱 실패: %s", e)
                     time.sleep(0.3)
+    saved = bulk_upsert_market_deals(all_deals)
     diag = f"calls={calls} saved={saved}" + (f" err={first_err}" if first_err else "")
     return saved, diag