From 8e64c4feb801a3f4ac09eac8b58d0cc532f27293 Mon Sep 17 00:00:00 2001 From: gahusb Date: Fri, 10 Jul 2026 10:55:51 +0900 Subject: [PATCH] =?UTF-8?q?fix(naver-fetch):=20realestate=20base=20:18800?= =?UTF-8?q?=20+=20=EB=8F=99=EB=B3=84=20ingest=20+=20180s=20no-retry?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- services/docker-compose.yml | 2 +- services/naver-fetch/.env.example | 3 +- services/naver-fetch/DESIGN.md | 14 ++++++---- services/naver-fetch/config.py | 2 +- services/naver-fetch/fetch_worker.py | 28 +++++++++---------- services/naver-fetch/nas_client.py | 27 +++++++++++------- services/naver-fetch/tests/test_config.py | 2 +- .../naver-fetch/tests/test_fetch_worker.py | 25 +++++++++-------- 8 files changed, 56 insertions(+), 47 deletions(-) diff --git a/services/docker-compose.yml b/services/docker-compose.yml index 77e4d32..16596da 100644 --- a/services/docker-compose.yml +++ b/services/docker-compose.yml @@ -164,7 +164,7 @@ services: environment: - TZ=Asia/Seoul - REDIS_URL=${REDIS_URL:-redis://192.168.45.54:6379} - - NAS_BASE_URL=${NAS_BASE_URL:-http://192.168.45.54:18500} + - NAS_BASE_URL=${NAVER_NAS_BASE_URL:-http://192.168.45.54:18800} - INTERNAL_API_KEY=${INTERNAL_API_KEY:-} - FETCH_INTERVAL_HOURS=${FETCH_INTERVAL_HOURS:-2.5} - DAYTIME_START=${DAYTIME_START:-08:00} diff --git a/services/naver-fetch/.env.example b/services/naver-fetch/.env.example index b09640d..c8e2ea4 100644 --- a/services/naver-fetch/.env.example +++ b/services/naver-fetch/.env.example @@ -1,7 +1,8 @@ # naver-fetch — realestate 매물 fetch 워커 REDIS_URL=redis://192.168.45.54:6379 -NAS_BASE_URL=http://192.168.45.54:18500 +# realestate-lab 직결 (targets/listings-ingest). compose가 NAVER_NAS_BASE_URL로 주입 +NAVER_NAS_BASE_URL=http://192.168.45.54:18800 # render 워커 공유키 (X-Internal-Key) INTERNAL_API_KEY= diff --git a/services/naver-fetch/DESIGN.md b/services/naver-fetch/DESIGN.md index 75c0aba..2a18b33 100644 --- a/services/naver-fetch/DESIGN.md +++ b/services/naver-fetch/DESIGN.md @@ -46,20 +46,22 @@ state.state = "fetching" await naver.acquire_token() # Playwright 1회 (사이클당) tg = await nas.get_targets() # {dongs:[{dong,cortar_no}], deal_types, page_limit} -batches = [] -for d in tg.dongs: # 동 단위 try/except 격리 +for d in tg.dongs: # 동 단위 try/except 격리 (fetch/ingest 각각) articles = [] for page in 1..tg.page_limit: resp = await naver.fetch_articles(d.cortar_no, page) # httpx GET articles += resp.get("articleList", []) if not resp.get("isMoreData"): break - batches.append({"dong": d.dong, "articles": articles}) -r = await nas.post_ingest(fetched_at=iso_utc_now, batches=batches) # {received,new,matched} -state.listings_pushed += sum(len(b["articles"]) for b in batches) + # ⚠️ 동별로 즉시 ingest — 전체 batch를 한 POST로 보내면 NAS 동기 처리(매물 수 비례)가 + # >180s 타임아웃. 동 단위(각 수십건)로 쪼개 각 POST가 timeout 내 완료 + 부분 실패 격리. + r = await nas.post_ingest(fetched_at=iso_utc_now, batches=[{"dong": d.dong, "articles": articles}]) + state.listings_pushed += len(articles) # {received,new,matched} state.last_fetch_at = epoch_now state.state = "idle" ``` +> ⚠️ **ingest 분할**: 스펙은 batches(복수)를 한 POST로 예시하지만, 첫 실행 대량(6동×~40건)이 NAS에서 >180s 걸려 타임아웃 → **동별 1 POST**로 분할(batches 리스트 계약 호환, 각 POST는 batches 1개). nas_client의 ingest는 timeout 180s·재시도 없음(재전송 시 NAS 중복 알림 방지). BE에 분할 통지 필요. + > ⚠️ **시간 포맷 2종**: ingest `fetched_at`은 **ISO UTC 문자열**(스펙 §4 ingest 스키마, 예 `2026-07-09T05:44:23Z`), heartbeat `ts`/`last_fetch_at`은 **epoch 정수**(스펙 §4.4). 혼동 주의. `heartbeat_loop`(별도 15초)이 `FetchState`를 읽어 §4.4 페이로드 발신. @@ -101,7 +103,7 @@ state.state = "idle" | env | 기본값 | 용도 | |-----|--------|------| -| `NAS_BASE_URL` | `http://192.168.45.54:18500` | targets/ingest | +| `NAS_BASE_URL` (compose: `NAVER_NAS_BASE_URL`) | `http://192.168.45.54:18800` | **realestate-lab 직결** targets/ingest (stock :18500 아님) | | `INTERNAL_API_KEY` | (필수) | `X-Internal-Key` (render 워커 공유키) | | `REDIS_URL` | `redis://192.168.45.54:6379` | heartbeat | | `FETCH_INTERVAL_HOURS` | `2.5` | fetch 주기 | diff --git a/services/naver-fetch/config.py b/services/naver-fetch/config.py index fdd78f6..ee2ca6a 100644 --- a/services/naver-fetch/config.py +++ b/services/naver-fetch/config.py @@ -24,7 +24,7 @@ class Settings: def load_settings() -> Settings: return Settings( - nas_base_url=os.getenv("NAS_BASE_URL", "http://192.168.45.54:18500"), + nas_base_url=os.getenv("NAS_BASE_URL", "http://192.168.45.54:18800"), internal_api_key=os.getenv("INTERNAL_API_KEY", ""), redis_url=os.getenv("REDIS_URL", "redis://192.168.45.54:6379"), fetch_interval_hours=float(os.getenv("FETCH_INTERVAL_HOURS", "2.5")), diff --git a/services/naver-fetch/fetch_worker.py b/services/naver-fetch/fetch_worker.py index 664c791..0041562 100644 --- a/services/naver-fetch/fetch_worker.py +++ b/services/naver-fetch/fetch_worker.py @@ -58,12 +58,14 @@ async def run_cycle(nas, naver, state, settings) -> None: dongs = tg.get("dongs", []) page_limit = int(tg.get("page_limit", 1)) - batches: list[dict] = [] + fetched_at = _iso_utc_now() + # 동별로 fetch → 즉시 ingest. 전체 batch를 한 POST로 보내면 NAS 동기 처리(매물 수 비례)가 + # >180s로 타임아웃하므로 동 단위로 쪼갬(각 ~수십건 → 빠름, 부분 실패 격리). batches 리스트 계약 호환. for d in dongs: cortar = d.get("cortar_no") dong = d.get("dong") - articles: list[dict] = [] try: + articles: list[dict] = [] for page in range(1, page_limit + 1): resp = await naver.fetch_articles(cortar, page) articles += resp.get("articleList", []) @@ -72,20 +74,16 @@ async def run_cycle(nas, naver, state, settings) -> None: except Exception: logger.exception("naver fetch 실패 dong=%s", dong) state.errors += 1 - batches.append({"dong": dong, "articles": articles}) + continue + try: + r = await nas.post_ingest(fetched_at, [{"dong": dong, "articles": articles}]) + logger.info("ingest dong=%s received=%s new=%s matched=%s", + dong, r.get("received"), r.get("new"), r.get("matched")) + state.listings_pushed += len(articles) + except Exception: + logger.exception("listings-ingest 실패 dong=%s", dong) + state.errors += 1 - fetched_at = _iso_utc_now() - try: - r = await nas.post_ingest(fetched_at, batches) - logger.info("ingest received=%s new=%s matched=%s", - r.get("received"), r.get("new"), r.get("matched")) - except Exception: - logger.exception("listings-ingest 실패") - state.errors += 1 - state.state = "idle" - return - - state.listings_pushed += sum(len(b["articles"]) for b in batches) state.last_fetch_at = _epoch_now() state.state = "idle" diff --git a/services/naver-fetch/nas_client.py b/services/naver-fetch/nas_client.py index deb535f..149a18c 100644 --- a/services/naver-fetch/nas_client.py +++ b/services/naver-fetch/nas_client.py @@ -8,40 +8,47 @@ import httpx logger = logging.getLogger(__name__) -_MAX_ATTEMPTS = 3 _RETRY_STATUSES = {429, 500, 502, 503, 504} +_TARGETS_TIMEOUT = 15.0 +_INGEST_TIMEOUT = 180.0 # 첫 실행 bulk upsert/매칭/알림이 느림 (초 단위 수십) class NASClient: - def __init__(self, base_url: str, api_key: str, timeout: float = 15.0): + def __init__(self, base_url: str, api_key: str): self._base_url = base_url.rstrip("/") self._api_key = api_key - self._client = httpx.AsyncClient(timeout=timeout) + self._client = httpx.AsyncClient() async def close(self) -> None: await self._client.aclose() async def get_targets(self) -> dict: - return await self._request("GET", "/api/internal/realestate/targets") + return await self._request("GET", "/api/internal/realestate/targets", + timeout=_TARGETS_TIMEOUT, max_attempts=3) async def post_ingest(self, fetched_at: str, batches: list[dict]) -> dict: + # ingest는 재시도 안 함(max_attempts=1) — 타임아웃 후 재전송 시 + # NAS가 같은 batch를 재처리/재알림할 위험. 실패 시 다음 사이클에 재시도. return await self._request( "POST", "/api/internal/realestate/listings-ingest", - json={"fetched_at": fetched_at, "batches": batches}) + json={"fetched_at": fetched_at, "batches": batches}, + timeout=_INGEST_TIMEOUT, max_attempts=1) - async def _request(self, method: str, path: str, **kwargs) -> dict: + async def _request(self, method: str, path: str, *, timeout: float, + max_attempts: int, **kwargs) -> dict: url = f"{self._base_url}{path}" headers = {"X-Internal-Key": self._api_key} - for attempt in range(_MAX_ATTEMPTS): + for attempt in range(max_attempts): try: - resp = await self._client.request(method, url, headers=headers, **kwargs) - if resp.status_code in _RETRY_STATUSES and attempt < _MAX_ATTEMPTS - 1: + resp = await self._client.request(method, url, headers=headers, + timeout=timeout, **kwargs) + if resp.status_code in _RETRY_STATUSES and attempt < max_attempts - 1: await asyncio.sleep(2 ** attempt) continue resp.raise_for_status() return resp.json() except httpx.TimeoutException: - if attempt < _MAX_ATTEMPTS - 1: + if attempt < max_attempts - 1: await asyncio.sleep(2 ** attempt) continue raise diff --git a/services/naver-fetch/tests/test_config.py b/services/naver-fetch/tests/test_config.py index 335d02d..26ce8f1 100644 --- a/services/naver-fetch/tests/test_config.py +++ b/services/naver-fetch/tests/test_config.py @@ -8,7 +8,7 @@ def test_defaults(monkeypatch): "NAVER_TOKEN_URL", "NAVER_HEADLESS"): monkeypatch.delenv(k, raising=False) s = load_settings() - assert s.nas_base_url == "http://192.168.45.54:18500" + assert s.nas_base_url == "http://192.168.45.54:18800" assert s.fetch_interval_hours == 2.5 assert s.daytime_start == "08:00" and s.daytime_end == "20:00" assert s.naver_headless is True diff --git a/services/naver-fetch/tests/test_fetch_worker.py b/services/naver-fetch/tests/test_fetch_worker.py index 5e4d2ec..e4a9b50 100644 --- a/services/naver-fetch/tests/test_fetch_worker.py +++ b/services/naver-fetch/tests/test_fetch_worker.py @@ -23,14 +23,15 @@ def test_in_daytime_boundaries(): class _FakeNAS: def __init__(self, targets): self._targets = targets - self.ingested = None + self.ingests = [] # 동별 post_ingest 호출 기록 async def get_targets(self): return self._targets async def post_ingest(self, fetched_at, batches): - self.ingested = {"fetched_at": fetched_at, "batches": batches} - return {"received": sum(len(b["articles"]) for b in batches), "new": 1, "matched": 1} + self.ingests.append({"fetched_at": fetched_at, "batches": batches}) + n = sum(len(b["articles"]) for b in batches) + return {"received": n, "new": n, "matched": 0} class _FakeNaver: @@ -48,7 +49,7 @@ class _FakeNaver: "articleList": [{"articleNo": f"{cortar_no}-{page}"}]} -async def test_run_cycle_assembles_batches_and_ingests(): +async def test_run_cycle_ingests_per_dong(): nas = _FakeNAS({"dongs": [{"dong": "신대방동", "cortar_no": "1159010100"}, {"dong": "봉천동", "cortar_no": "1162010100"}], "deal_types": ["매매"], "page_limit": 1}) @@ -56,13 +57,14 @@ async def test_run_cycle_assembles_batches_and_ingests(): state = FetchState() await run_cycle(nas, naver, state, load_settings()) assert naver.token_acquired is True - assert nas.ingested is not None - dongs = [b["dong"] for b in nas.ingested["batches"]] + assert len(nas.ingests) == 2 # 동별 1 POST씩 + dongs = [ing["batches"][0]["dong"] for ing in nas.ingests] assert dongs == ["신대방동", "봉천동"] + assert all(len(ing["batches"]) == 1 for ing in nas.ingests) assert state.listings_pushed == 2 assert state.last_fetch_at is not None assert state.state == "idle" - assert nas.ingested["fetched_at"].endswith("Z") + assert nas.ingests[0]["fetched_at"].endswith("Z") async def test_run_cycle_dong_failure_isolated(): @@ -72,11 +74,10 @@ async def test_run_cycle_dong_failure_isolated(): naver = _FakeNaver(fail_dongs={"1159010100"}) state = FetchState() await run_cycle(nas, naver, state, load_settings()) - assert nas.ingested is not None # 실패해도 ingest 진행 + # A는 fetch 실패 → ingest skip, B만 ingest (부분 실패 격리) + assert len(nas.ingests) == 1 + assert nas.ingests[0]["batches"][0]["dong"] == "B" assert state.errors >= 1 - batches = {b["dong"]: b["articles"] for b in nas.ingested["batches"]} - assert batches["A"] == [] # 실패 동 빈 배열 - assert len(batches["B"]) == 1 # 성공 동 async def test_run_cycle_token_failure_skips_ingest(): @@ -87,7 +88,7 @@ async def test_run_cycle_token_failure_skips_ingest(): nas = _FakeNAS({"dongs": [], "page_limit": 1}) state = FetchState() await run_cycle(nas, _BadNaver(), state, load_settings()) - assert nas.ingested is None + assert nas.ingests == [] assert state.errors == 1 assert state.state == "idle"