diff --git a/realestate-lab/app/db.py b/realestate-lab/app/db.py index d795b48..7601821 100644 --- a/realestate-lab/app/db.py +++ b/realestate-lab/app/db.py @@ -177,6 +177,72 @@ def init_db(): ); """) + # ── listings (매물 파이프라인) ────────────────────────────────────── + conn.execute(""" + CREATE TABLE IF NOT EXISTS listings ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + article_no TEXT UNIQUE, source TEXT NOT NULL, deal_type TEXT NOT NULL, + deposit INTEGER, monthly_rent INTEGER, sale_price INTEGER, + area_exclusive REAL, floor TEXT, complex_name TEXT, address TEXT, + dong TEXT, dong_code TEXT, url TEXT, posted_at TEXT, raw_json TEXT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), + notified_at TEXT + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS market_deals ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + house_type TEXT, dong_code TEXT, dong TEXT, complex_name TEXT, area REAL, + deal_type TEXT, deposit INTEGER, monthly_rent INTEGER, sale_amount INTEGER, + deal_ym TEXT, floor TEXT, source TEXT DEFAULT 'molit', + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) + ) + """) + # 플랜 원안은 table-level UNIQUE(...)였으나 SQLite는 UNIQUE 제약에서 NULL을 서로 다른 값으로 + # 취급해 deposit/sale_amount(둘 중 하나는 항상 NULL)가 있는 행에서 dedup이 깨진다. + # (표현식은 table-level UNIQUE에 사용 불가 → CREATE UNIQUE INDEX + COALESCE로 대체. + # 실제 컬럼값은 그대로 NULL 유지 — downstream get_market_deals_for/listing_matcher 영향 없음.) + conn.execute(""" + CREATE UNIQUE INDEX IF NOT EXISTS ux_market_deals_dedup ON market_deals( + dong_code, COALESCE(complex_name,''), area, deal_ym, deal_type, + COALESCE(deposit,-1), COALESCE(sale_amount,-1), COALESCE(floor,'') + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS listing_matches ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + listing_id INTEGER UNIQUE REFERENCES listings(id) ON DELETE CASCADE, + category TEXT, passed INTEGER, match_score INTEGER, + jeonse_ratio REAL, safety_tier TEXT, price_ratio REAL, valuation_tier TEXT, + market_median INTEGER, sample_size INTEGER, budget_ok INTEGER, + regulation_flags TEXT, reasons TEXT, is_new INTEGER, + notified_at TEXT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS listing_criteria ( + id INTEGER PRIMARY KEY DEFAULT 1, + dongs TEXT NOT NULL DEFAULT '[]', deal_types TEXT NOT NULL DEFAULT '[]', + max_deposit INTEGER, max_sale_price INTEGER, min_area REAL, + house_types TEXT NOT NULL DEFAULT '[]', min_safety_tier TEXT, + notify_enabled INTEGER NOT NULL DEFAULT 1, + equity INTEGER, annual_income INTEGER, + is_homeless INTEGER NOT NULL DEFAULT 1, is_householder INTEGER NOT NULL DEFAULT 0, + is_first_home INTEGER NOT NULL DEFAULT 0, + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS listing_collect_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + new_count INTEGER, total_count INTEGER, naver_ok INTEGER, + error TEXT, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) + ) + """) + conn.execute("CREATE INDEX IF NOT EXISTS idx_md_lookup ON market_deals(dong_code, deal_type, area, deal_ym);") + conn.execute("CREATE INDEX IF NOT EXISTS idx_listing_dong ON listings(dong, deal_type);") + # ── 상태 자동 계산 ─────────────────────────────────────────────────────────── @@ -841,3 +907,187 @@ def get_dashboard() -> Dict[str, Any]: "upcoming_schedules": schedules, "bookmarked": bookmarked_items, } + + +# ── 매물 파이프라인 (listings/market_deals/listing_matches/listing_criteria) ── + +_LISTING_CRITERIA_SEED = { + "dongs": ["신대방동", "대방동", "상도동", "봉천동", "대림동", "신길동"], + "deal_types": ["전세", "반전세", "매매"], + "max_deposit": 32000, "max_sale_price": None, "min_area": 40.0, + "house_types": ["아파트", "오피스텔", "빌라"], "min_safety_tier": None, + "notify_enabled": 1, "equity": None, "annual_income": None, + "is_homeless": 1, "is_householder": 0, "is_first_home": 0, +} +_LC_JSON_COLS = ("dongs", "deal_types", "house_types") + + +def get_listing_criteria() -> Dict[str, Any]: + with _conn() as conn: + row = conn.execute("SELECT * FROM listing_criteria WHERE id=1").fetchone() + if row is None: + conn.execute( + """INSERT INTO listing_criteria + (id,dongs,deal_types,max_deposit,max_sale_price,min_area,house_types, + min_safety_tier,notify_enabled,equity,annual_income,is_homeless,is_householder,is_first_home) + VALUES (1,:dongs,:deal_types,:max_deposit,:max_sale_price,:min_area,:house_types, + :min_safety_tier,:notify_enabled,:equity,:annual_income,:is_homeless,:is_householder,:is_first_home)""", + {**_LISTING_CRITERIA_SEED, + "dongs": json.dumps(_LISTING_CRITERIA_SEED["dongs"], ensure_ascii=False), + "deal_types": json.dumps(_LISTING_CRITERIA_SEED["deal_types"], ensure_ascii=False), + "house_types": json.dumps(_LISTING_CRITERIA_SEED["house_types"], ensure_ascii=False)}, + ) + row = conn.execute("SELECT * FROM listing_criteria WHERE id=1").fetchone() + d = dict(row) + for c in _LC_JSON_COLS: + d[c] = json.loads(d[c] or "[]") + return d + + +def update_listing_criteria(fields: Dict[str, Any]) -> None: + if not fields: + return + sets, vals = [], [] + for k, v in fields.items(): + if v is None and k not in ("max_deposit", "max_sale_price", "equity", "annual_income", "min_safety_tier"): + continue + sets.append(f"{k}=?") + vals.append(json.dumps(v, ensure_ascii=False) if k in _LC_JSON_COLS else v) + if not sets: + return + sets.append("updated_at=strftime('%Y-%m-%dT%H:%M:%fZ','now')") + with _conn() as conn: + conn.execute("INSERT OR IGNORE INTO listing_criteria(id) VALUES(1)") + conn.execute(f"UPDATE listing_criteria SET {','.join(sets)} WHERE id=1", vals) + + +def upsert_listing(data: Dict[str, Any]) -> tuple: + cols = ("article_no", "source", "deal_type", "deposit", "monthly_rent", "sale_price", + "area_exclusive", "floor", "complex_name", "address", "dong", "dong_code", + "url", "posted_at", "raw_json") + d = {c: data.get(c) for c in cols} + with _conn() as conn: + exists = conn.execute("SELECT id FROM listings WHERE article_no=?", (d["article_no"],)).fetchone() + is_new = exists is None + conn.execute(f""" + INSERT INTO listings ({','.join(cols)}) VALUES ({','.join(':'+c for c in cols)}) + ON CONFLICT(article_no) DO UPDATE SET + deposit=excluded.deposit, monthly_rent=excluded.monthly_rent, sale_price=excluded.sale_price, + floor=excluded.floor, complex_name=excluded.complex_name, address=excluded.address, + url=excluded.url, posted_at=excluded.posted_at, raw_json=excluded.raw_json + """, d) + row = conn.execute("SELECT * FROM listings WHERE article_no=?", (d["article_no"],)).fetchone() + return dict(row), is_new + + +def upsert_market_deal(data: Dict[str, Any]) -> None: + 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} + d.setdefault("source", "molit") + with _conn() as conn: + conn.execute(f"""INSERT OR IGNORE INTO market_deals ({','.join(cols)}) + VALUES ({','.join(':'+c for c in cols)})""", d) + + +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 upsert_listing_match(data: Dict[str, Any]) -> None: + 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") + d = {c: data.get(c) for c 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) + with _conn() as conn: + conn.execute(f""" + INSERT INTO listing_matches ({','.join(cols)}) VALUES ({','.join(':'+c for c 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) + + +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") + p = [] + if dong: sql += " AND l.dong=?"; p.append(dong) + if deal_type: sql += " AND l.deal_type=?"; p.append(deal_type) + if tier: sql += " AND (m.safety_tier=? OR m.valuation_tier=?)"; p += [tier, tier] + if matched_only: sql += " AND m.passed=1" + sql += " ORDER BY l.created_at DESC LIMIT ? OFFSET ?"; p += [limit, offset] + with _conn() as conn: + return [dict(r) for r in conn.execute(sql, p).fetchall()] + + +def get_listing_matches() -> List[Dict[str, Any]]: + with _conn() as conn: + rows = conn.execute( + "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] + + +_TIER_ORDER = {"위험": 0, "고가": 0, "주의": 1, "시세": 1, "안전": 2, "저평가": 2, "보류": -1} + + +def get_unnotified_listing_matches(min_tier=None) -> List[Dict[str, Any]]: + with _conn() as conn: + rows = conn.execute( + "SELECT m.*, l.complex_name, l.dong, l.deal_type, l.deposit, l.monthly_rent, " + "l.sale_price, l.area_exclusive, l.floor, l.url " + "FROM listing_matches m JOIN listings l ON l.id=m.listing_id " + "WHERE m.passed=1 AND m.notified_at IS NULL" + ).fetchall() + 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) + if min_tier: + floor = _TIER_ORDER.get(min_tier, -1) + out = [d for d in out if _TIER_ORDER.get(d.get("safety_tier") or d.get("valuation_tier"), -1) >= floor] + return out + + +def mark_listings_notified(match_ids: List[int]) -> None: + if not match_ids: + return + with _conn() as conn: + conn.executemany( + "UPDATE listing_matches SET notified_at=strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id=?", + [(i,) for i in match_ids], + ) + + +def save_listing_collect_log(new_count, total_count, naver_ok, error=None) -> None: + with _conn() as conn: + conn.execute( + "INSERT INTO listing_collect_log (new_count,total_count,naver_ok,error) VALUES (?,?,?,?)", + (new_count, total_count, 1 if naver_ok else 0, error), + ) + + +def get_last_listing_collect_log() -> Optional[Dict[str, Any]]: + with _conn() as conn: + row = conn.execute("SELECT * FROM listing_collect_log ORDER BY id DESC LIMIT 1").fetchone() + return dict(row) if row else None diff --git a/realestate-lab/tests/conftest.py b/realestate-lab/tests/conftest.py index 553b4fe..e0e360c 100644 --- a/realestate-lab/tests/conftest.py +++ b/realestate-lab/tests/conftest.py @@ -16,6 +16,11 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # 테이블 목록 — init_db가 생성하는 모든 테이블 _USER_TABLES = ( + "listing_matches", # FK CASCADE 대비 자식 테이블 먼저 + "listings", + "market_deals", + "listing_criteria", + "listing_collect_log", "match_results", # FK CASCADE 대비 자식 테이블 먼저 "announcement_models", "announcements", diff --git a/realestate-lab/tests/test_listing_db.py b/realestate-lab/tests/test_listing_db.py new file mode 100644 index 0000000..8c65ef1 --- /dev/null +++ b/realestate-lab/tests/test_listing_db.py @@ -0,0 +1,40 @@ +def test_listing_criteria_seed_and_update(): + from app import db + c = db.get_listing_criteria() + assert c["id"] == 1 and "신대방동" in c["dongs"] and c["max_deposit"] == 32000 + db.update_listing_criteria({"max_deposit": 30000, "equity": 15000}) + assert db.get_listing_criteria()["max_deposit"] == 30000 + assert db.get_listing_criteria()["equity"] == 15000 + +def test_upsert_listing_idempotent_and_is_new(): + from app import db + row = {"article_no": "A1", "source": "naver", "deal_type": "전세", "deposit": 29000, + "area_exclusive": 42.0, "complex_name": "OO", "dong": "신대방동", "dong_code": "11590"} + _, is_new = db.upsert_listing(row) + assert is_new is True + _, is_new2 = db.upsert_listing({**row, "deposit": 28000}) + assert is_new2 is False + ls = db.get_listings() + assert len(ls) == 1 and ls[0]["deposit"] == 28000 + +def test_market_deals_dedup_and_query(): + from app import db + base = {"house_type": "아파트", "dong_code": "11590", "dong": "신대방동", + "complex_name": "OO", "area": 42.0, "deal_type": "전세", + "deposit": 43000, "sale_amount": None, "deal_ym": "202605", "floor": "5", "source": "molit"} + db.upsert_market_deal(base) + db.upsert_market_deal(base) # 동일 → dedup + deals = db.get_market_deals_for("11590", "OO", 42.0, "전세", months=120) + assert len(deals) == 1 + +def test_unnotified_listing_match_and_mark(): + from app import db + lid, _ = db.upsert_listing({"article_no": "A2", "source": "naver", "deal_type": "전세", + "deposit": 29000, "area_exclusive": 42.0, "dong": "신대방동"}) + db.upsert_listing_match({"listing_id": lid[0] if isinstance(lid, tuple) else lid["id"], + "category": "임차", "passed": 1, "match_score": 90, + "safety_tier": "안전", "sample_size": 7, "reasons": ["ok"], "is_new": 1}) + un = db.get_unnotified_listing_matches() + assert len(un) == 1 + db.mark_listings_notified([un[0]["id"]]) + assert db.get_unnotified_listing_matches() == []