gahusb/ai-trade
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(naver-fetch): realestate base :18800 + 동별 ingest + 180s no-retry

Browse Source
This commit is contained in:
gahusb
2026-07-10 10:55:51 +09:00
parent 0b4e4c2e0e
commit 8e64c4feb8
8 changed files with 56 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
services/docker-compose.yml
View File

@@ -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}

3
services/naver-fetch/.env.example
View File

@@ -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=

14
services/naver-fetch/DESIGN.md
View File

@@ -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 주기 |

2
services/naver-fetch/config.py
View File

@@ -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")),

22
services/naver-fetch/fetch_worker.py
View File

@@ -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})
fetched_at = _iso_utc_now()
continue
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"))
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 실패")
logger.exception("listings-ingest 실패 dong=%s", dong)
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"

27
services/naver-fetch/nas_client.py
View File

@@ -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

2
services/naver-fetch/tests/test_config.py
View File

@@ -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

25
services/naver-fetch/tests/test_fetch_worker.py
View File

@@ -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"