diff --git a/realestate-lab/app/internal_router.py b/realestate-lab/app/internal_router.py index 8d9dd12..88123de 100644 --- a/realestate-lab/app/internal_router.py +++ b/realestate-lab/app/internal_router.py @@ -1,8 +1,9 @@ """naver-fetch 워커 내부 계약: targets 조회 + listings ingest.""" import os +import time import logging from typing import List, Dict, Any -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, BackgroundTasks from pydantic import BaseModel from .auth import verify_internal_key @@ -45,7 +46,7 @@ class ListingsIngest(BaseModel): @router.post("/api/internal/realestate/listings-ingest", dependencies=[Depends(verify_internal_key)]) -def listings_ingest(body: ListingsIngest): +def listings_ingest(body: ListingsIngest, background_tasks: BackgroundTasks): received = new = 0 for batch in body.batches: for raw in batch.articles: @@ -60,10 +61,18 @@ def listings_ingest(body: ListingsIngest): new += 1 except Exception as e: logger.warning("네이버 ingest 파싱 실패: %s", e) - matched = 0 if received: - with match_notify_lock: - run_listing_matching() - noti = notify_new_listings() - matched = int(noti.get("sent", 0)) if isinstance(noti, dict) else 0 - return {"received": received, "new": new, "matched": matched} + background_tasks.add_task(_match_and_notify) + return {"received": received, "new": new, "queued": bool(received)} + + +def _match_and_notify(): + """ingest 후처리(백그라운드): 매칭+알림을 공유 락으로 직렬화. 타이밍 로그로 병목 관측.""" + with match_notify_lock: + t0 = time.time() + run_listing_matching() + t1 = time.time() + noti = notify_new_listings() + t2 = time.time() + sent = (noti or {}).get("sent") if isinstance(noti, dict) else noti + logger.info("ingest 후처리: match=%.1fs notify=%.1fs sent=%s", t1 - t0, t2 - t1, sent) diff --git a/realestate-lab/app/notifier.py b/realestate-lab/app/notifier.py index 40f6976..30ce5c1 100644 --- a/realestate-lab/app/notifier.py +++ b/realestate-lab/app/notifier.py @@ -10,6 +10,8 @@ logger = logging.getLogger("realestate-lab") AGENT_OFFICE_URL = os.getenv("AGENT_OFFICE_URL", "http://agent-office:8000") NOTIFY_TIMEOUT_SECONDS = int(os.getenv("REALESTATE_NOTIFY_TIMEOUT", "15")) +LISTING_NOTIFY_MAX = int(os.getenv("LISTING_NOTIFY_MAX", "8")) +LISTING_NOTIFY_TIMEOUT = int(os.getenv("LISTING_NOTIFY_TIMEOUT", "60")) def notify_new_matches() -> dict: @@ -49,21 +51,28 @@ def notify_new_matches() -> dict: def notify_new_listings() -> dict: """임계 통과 미알림 매물을 agent-office로 push (notify_new_matches 대칭). - passed 매물은 전 안전등급 알림(위험도 회피 목적 유용). criteria.min_safety_tier 설정 시 그 이상만.""" + passed 매물은 전 안전등급 알림(위험도 회피 목적 유용). criteria.min_safety_tier 설정 시 그 이상만. + 콜드스타트로 backlog가 쌓이면 한 사이클 최대 LISTING_NOTIFY_MAX건만 알림 발송하고 + 나머지는 조용히 notified_at 마킹(baseline)해 스팸·재전송 루프를 막는다.""" crit = get_listing_criteria() if not crit.get("notify_enabled"): return {"sent": 0, "skipped": "notify_disabled"} listings = get_unnotified_listing_matches(min_tier=crit.get("min_safety_tier")) if not listings: return {"sent": 0} + to_send = listings[:LISTING_NOTIFY_MAX] + to_baseline = listings[LISTING_NOTIFY_MAX:] + if to_baseline: # 콜드스타트 backlog: 조용히 baseline 마킹(알림 X) + mark_listings_notified([m["id"] for m in to_baseline]) + logger.info("매물 알림 baseline(무알림) 마킹: %d건", len(to_baseline)) url = f"{AGENT_OFFICE_URL}/api/agent-office/realestate/notify-listing" try: - resp = requests.post(url, json={"listings": listings}, timeout=NOTIFY_TIMEOUT_SECONDS) + resp = requests.post(url, json={"listings": to_send}, timeout=LISTING_NOTIFY_TIMEOUT) resp.raise_for_status() body = resp.json() except requests.RequestException as e: logger.error("agent-office 매물 push 실패: %s", e) - return {"sent": 0, "error": str(e)} + return {"sent": 0, "error": str(e), "baselined": len(to_baseline)} sent_ids = body.get("sent_ids") or [] if sent_ids: mark_listings_notified(sent_ids) diff --git a/realestate-lab/docs/superpowers/specs/2026-07-09-naver-listing-worker-design.md b/realestate-lab/docs/superpowers/specs/2026-07-09-naver-listing-worker-design.md index 2b43edc..e46e94f 100644 --- a/realestate-lab/docs/superpowers/specs/2026-07-09-naver-listing-worker-design.md +++ b/realestate-lab/docs/superpowers/specs/2026-07-09-naver-listing-worker-design.md @@ -110,7 +110,7 @@ ## 6. 인증 / 보안 - **nginx**: 신규 location `/api/internal/realestate/` → `realestate-lab:8000`, **IP 화이트리스트**(기존 render 워커 화이트리스트에 쓰는 Windows 워커 IP 재사용). (nginx-conf 변경 → 배포) -- **앱 레벨**: realestate-lab이 `X-Internal-Key` 검증(env `REALESTATE_INTERNAL_KEY`). 다른 서비스 `internal_router` 패턴 준용. GET targets·POST ingest 둘 다 검증. +- **앱 레벨**: realestate-lab이 `X-Internal-Key` 검증(env `INTERNAL_API_KEY`). 다른 서비스 `internal_router` 패턴 준용. GET targets·POST ingest 둘 다 검증. - 워커는 `.env`로 키 주입(커밋 금지). ## 7. 에러처리 / 멱등성 diff --git a/realestate-lab/tests/test_internal_ingest.py b/realestate-lab/tests/test_internal_ingest.py index 4dfb16e..37b01cf 100644 --- a/realestate-lab/tests/test_internal_ingest.py +++ b/realestate-lab/tests/test_internal_ingest.py @@ -23,6 +23,8 @@ def test_ingest_auth(monkeypatch): def test_ingest_upserts_and_matches(monkeypatch): c = _client(monkeypatch) + # TestClient는 BackgroundTask를 응답 반환 전 동기 실행하므로 patch된 notify는 + # c.post()가 리턴하는 시점에 이미 호출되어 있다. with patch("app.internal_router.notify_new_listings", return_value={"sent": 1, "sent_ids": [1]}) as noti: r = c.post("/api/internal/realestate/listings-ingest", headers={"X-Internal-Key": "SECRET"}, @@ -30,7 +32,7 @@ def test_ingest_upserts_and_matches(monkeypatch): "batches": [{"dong": "신대방동", "articles": [_ARTICLE]}]}) assert r.status_code == 200 body = r.json() - assert body["received"] == 1 and body["new"] == 1 and body["matched"] == 1 + assert body["received"] == 1 and body["new"] == 1 and body["queued"] is True assert noti.call_count == 1 # 저장 확인 from app import db diff --git a/realestate-lab/tests/test_listing_notifier.py b/realestate-lab/tests/test_listing_notifier.py index 906362f..d1bce99 100644 --- a/realestate-lab/tests/test_listing_notifier.py +++ b/realestate-lab/tests/test_listing_notifier.py @@ -26,3 +26,26 @@ def test_notify_listings_no_mark_on_failure(): with patch.object(notifier.requests, "post", side_effect=rq.RequestException("down")): notifier.notify_new_listings() assert len(db.get_unnotified_listing_matches()) == 1 # 미마킹→재시도 + + +def test_notify_caps_and_baselines(monkeypatch): + from app import notifier, db + from unittest.mock import MagicMock + for i in range(10): + row, _ = db.upsert_listing({"article_no": f"C{i}", "source": "naver", "deal_type": "전세", + "deposit": 29000, "area_exclusive": 42.0, "dong": "신대방동"}) + db.upsert_listing_match({"listing_id": row["id"], "category": "임차", "passed": 1, + "match_score": 100, "safety_tier": "안전", "sample_size": 7, + "reasons": ["ok"], "is_new": 1}) + monkeypatch.setattr(notifier, "LISTING_NOTIFY_MAX", 8) + captured = {} + fake = MagicMock(); fake.raise_for_status.return_value = None + def fake_post(url, json=None, timeout=None): + captured["n"] = len(json["listings"]) + fake.json.return_value = {"sent": len(json["listings"]), + "sent_ids": [m["id"] for m in json["listings"]]} + return fake + monkeypatch.setattr(notifier.requests, "post", fake_post) + notifier.notify_new_listings() + assert captured["n"] == 8 # 8건만 텔레그램 전송 + assert len(db.get_unnotified_listing_matches()) == 0 # 나머지 2건 baseline 마킹