perf(realestate): run_listing_matching 단일 conn+배치 + POST /listings/rematch(MOLIT없이 즉시 재판정)

- get_market_deals_for에 optional conn 파라미터 추가(재사용 시 open/close 안 함, 기존 단건 호출부는 불변)
- bulk_upsert_listing_matches 추가(listing_matches 다건 upsert, 단일 connection)
- run_listing_matching이 listings 조회+매물별 market_deals 조회+최종 upsert를 단일 _conn()으로 처리
  (기존: 269건 매물마다 개별 _conn() 3회 → 병목)
- POST /api/realestate/listings/rematch: MOLIT 재수집 없이 기존 listings+market_deals로 즉시 재판정
  (알림 미발송, 즉시 피드백용)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EqCYBhvTcdeCTUDX3RhWx9

realestate-lab/app/db.py

@@ -1019,10 +1019,14 @@ def bulk_upsert_market_deals(rows: List[Dict[str, Any]]) -> int:
     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 get_market_deals_for(dong_code, complex_name, area, deal_type, months=6, conn=None) -> List[Dict[str, Any]]:
+    """conn을 넘기면 재사용(open/close 안 함, 호출부가 소유) — run_listing_matching 단일 conn 배치용.
+    conn 없으면 기존대로 자체 _conn() 열고 닫음(safety-check 등 단건 호출 경로 불변)."""
+    own = conn is None
+    c = conn or _conn()
+    try:
         def _query(cn):
-            rows = conn.execute(
+            rows = c.execute(
                 """SELECT * FROM market_deals
                    WHERE dong_code=? AND deal_type=?
                      AND (complex_name=? OR ? IS NULL OR complex_name IS NULL)
@@ -1038,7 +1042,10 @@ def get_market_deals_for(dong_code, complex_name, area, deal_type, months=6) ->
             # 단지 표본이 부족하면 complex 제약을 풀고 같은 동/거래유형/면적±5/최근성으로
             # 광역 재조회한다. 표본을 늘리기만 하므로 안전(잘못된 판정 유발 X).
             deals = _query(None)
-    return deals
+        return deals
+    finally:
+        if own:
+            c.close()
 
 
 def upsert_listing_match(data: Dict[str, Any]) -> None:
@@ -1061,6 +1068,39 @@ def upsert_listing_match(data: Dict[str, Any]) -> None:
         """, d)
 
 
+def bulk_upsert_listing_matches(recs: List[Dict[str, Any]], conn=None) -> None:
+    """listing_matches 다건 upsert(단일 connection). conn 주면 재사용(호출부 with가 commit) —
+    run_listing_matching 269건×3conn 병목 개선(Task 2). ON CONFLICT 절은 upsert_listing_match와 동일
+    (notified_at 미포함=보존, 재알림 방지)."""
+    if not recs:
+        return
+    cols = ("listing_id", "category", "passed", "match_score", "jeonse_ratio", "safety_tier",
+            "price_ratio", "valuation_tier", "market_median", "sample_size", "budget_ok",
+            "regulation_flags", "reasons", "is_new")
+    own = conn is None
+    c = conn or _conn()
+    try:
+        for data in recs:
+            d = {col: data.get(col) for col in cols}
+            d["regulation_flags"] = json.dumps(data.get("regulation_flags") or [], ensure_ascii=False)
+            d["reasons"] = json.dumps(data.get("reasons") or [], ensure_ascii=False)
+            c.execute(f"""
+                INSERT INTO listing_matches ({','.join(cols)}) VALUES ({','.join(':'+col for col in cols)})
+                ON CONFLICT(listing_id) DO UPDATE SET
+                  passed=excluded.passed, match_score=excluded.match_score,
+                  jeonse_ratio=excluded.jeonse_ratio, safety_tier=excluded.safety_tier,
+                  price_ratio=excluded.price_ratio, valuation_tier=excluded.valuation_tier,
+                  market_median=excluded.market_median, sample_size=excluded.sample_size,
+                  budget_ok=excluded.budget_ok, regulation_flags=excluded.regulation_flags,
+                  reasons=excluded.reasons, category=excluded.category
+            """, d)
+        if own:
+            c.commit()
+    finally:
+        if own:
+            c.close()
+
+
 def get_listings(dong=None, deal_type=None, tier=None, matched_only=False, limit=50, offset=0) -> List[Dict[str, Any]]:
     sql = ("SELECT l.*, m.safety_tier, m.valuation_tier, m.category, m.match_score, m.passed "
            "FROM listings l LEFT JOIN listing_matches m ON m.listing_id=l.id WHERE 1=1")