feat(naver-fetch): fetch_worker (run_cycle/loop/heartbeat)

services/naver-fetch/fetch_worker.py

@@ -0,0 +1,130 @@
+"""오케스트레이션 — 주간 스케줄 fetch + fetcher heartbeat."""
+from __future__ import annotations
+
+import asyncio
+import datetime as dt
+import json
+import logging
+from zoneinfo import ZoneInfo
+
+logger = logging.getLogger(__name__)
+KST = ZoneInfo("Asia/Seoul")
+HEARTBEAT_KEY = "worker:naver-fetch:heartbeat"
+
+
+class FetchState:
+    """fetch_loop가 갱신, heartbeat_loop가 읽는 공유 상태."""
+    def __init__(self):
+        self.state = "idle"                 # idle | fetching
+        self.last_fetch_at: int | None = None
+        self.listings_pushed = 0
+        self.errors = 0
+
+
+def _parse_hhmm(s: str) -> dt.time:
+    hh, mm = s.split(":")
+    return dt.time(int(hh), int(mm))
+
+
+def in_daytime(now: dt.datetime, start: str, end: str) -> bool:
+    t = now.timetz().replace(tzinfo=None)
+    return _parse_hhmm(start) <= t < _parse_hhmm(end)
+
+
+def _epoch_now() -> int:
+    return int(dt.datetime.now(dt.timezone.utc).timestamp())
+
+
+def _iso_utc_now() -> str:
+    return dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+async def run_cycle(nas, naver, state, settings) -> None:
+    state.state = "fetching"
+    try:
+        await naver.acquire_token()
+    except Exception:
+        logger.exception("naver 토큰 획득 실패")
+        state.errors += 1
+        state.state = "idle"
+        return
+    try:
+        tg = await nas.get_targets()
+    except Exception:
+        logger.exception("targets 조회 실패")
+        state.errors += 1
+        state.state = "idle"
+        return
+
+    dongs = tg.get("dongs", [])
+    page_limit = int(tg.get("page_limit", 1))
+    batches: list[dict] = []
+    for d in dongs:
+        cortar = d.get("cortar_no")
+        dong = d.get("dong")
+        articles: list[dict] = []
+        try:
+            for page in range(1, page_limit + 1):
+                resp = await naver.fetch_articles(cortar, page)
+                articles += resp.get("articleList", [])
+                if not resp.get("isMoreData"):
+                    break
+        except Exception:
+            logger.exception("naver fetch 실패 dong=%s", dong)
+            state.errors += 1
+        batches.append({"dong": dong, "articles": articles})
+
+    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"
+
+
+async def fetch_loop(nas, naver, state, settings) -> None:
+    interval = dt.timedelta(hours=settings.fetch_interval_hours)
+    last_run: dt.datetime | None = None
+    logger.info("naver-fetch loop 시작 interval=%.1fh 주간창 %s-%s",
+                settings.fetch_interval_hours, settings.daytime_start, settings.daytime_end)
+    while True:
+        try:
+            now = dt.datetime.now(KST)
+            if in_daytime(now, settings.daytime_start, settings.daytime_end) and (
+                    last_run is None or (now - last_run) >= interval):
+                await run_cycle(nas, naver, state, settings)
+                last_run = now
+        except asyncio.CancelledError:
+            logger.info("fetch_loop cancelled")
+            raise
+        except Exception:
+            logger.exception("fetch_loop iteration 실패")
+        await asyncio.sleep(300)
+
+
+def build_heartbeat(state) -> str:
+    return json.dumps({
+        "name": "naver-fetch", "kind": "fetcher", "state": state.state,
+        "ts": _epoch_now(), "last_fetch_at": state.last_fetch_at,
+        "listings_pushed": state.listings_pushed, "errors": state.errors,
+    })
+
+
+async def heartbeat_loop(redis, state, interval: int = 15, ttl: int = 45) -> None:
+    logger.info("naver-fetch heartbeat 시작 ttl=%ds", ttl)
+    while True:
+        try:
+            await redis.set(HEARTBEAT_KEY, build_heartbeat(state), ex=ttl)
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            logger.exception("heartbeat 발신 실패")
+        await asyncio.sleep(interval)

services/naver-fetch/tests/test_fetch_worker.py

@@ -0,0 +1,106 @@
+"""fetch_worker — 주간창/조립/격리/heartbeat."""
+import datetime as dt
+import json
+from zoneinfo import ZoneInfo
+
+from fetch_worker import FetchState, in_daytime, run_cycle, build_heartbeat
+from config import load_settings
+
+KST = ZoneInfo("Asia/Seoul")
+
+
+def _kst(hh, mm):
+    return dt.datetime(2026, 7, 10, hh, mm, tzinfo=KST)
+
+
+def test_in_daytime_boundaries():
+    assert in_daytime(_kst(8, 0), "08:00", "20:00") is True
+    assert in_daytime(_kst(19, 59), "08:00", "20:00") is True
+    assert in_daytime(_kst(20, 0), "08:00", "20:00") is False
+    assert in_daytime(_kst(7, 59), "08:00", "20:00") is False
+
+
+class _FakeNAS:
+    def __init__(self, targets):
+        self._targets = targets
+        self.ingested = None
+
+    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}
+
+
+class _FakeNaver:
+    def __init__(self, fail_dongs=None):
+        self._fail = fail_dongs or set()
+        self.token_acquired = False
+
+    async def acquire_token(self):
+        self.token_acquired = True
+
+    async def fetch_articles(self, cortar_no, page=1):
+        if cortar_no in self._fail:
+            raise RuntimeError("naver 429")
+        return {"isMoreData": False,
+                "articleList": [{"articleNo": f"{cortar_no}-{page}"}]}
+
+
+async def test_run_cycle_assembles_batches_and_ingests():
+    nas = _FakeNAS({"dongs": [{"dong": "신대방동", "cortar_no": "1159010100"},
+                              {"dong": "봉천동", "cortar_no": "1162010100"}],
+                    "deal_types": ["매매"], "page_limit": 1})
+    naver = _FakeNaver()
+    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 dongs == ["신대방동", "봉천동"]
+    assert state.listings_pushed == 2
+    assert state.last_fetch_at is not None
+    assert state.state == "idle"
+    assert nas.ingested["fetched_at"].endswith("Z")
+
+
+async def test_run_cycle_dong_failure_isolated():
+    nas = _FakeNAS({"dongs": [{"dong": "A", "cortar_no": "1159010100"},
+                              {"dong": "B", "cortar_no": "1162010100"}],
+                    "page_limit": 1})
+    naver = _FakeNaver(fail_dongs={"1159010100"})
+    state = FetchState()
+    await run_cycle(nas, naver, state, load_settings())
+    assert nas.ingested is not None            # 실패해도 ingest 진행
+    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():
+    class _BadNaver(_FakeNaver):
+        async def acquire_token(self):
+            raise RuntimeError("token 실패")
+
+    nas = _FakeNAS({"dongs": [], "page_limit": 1})
+    state = FetchState()
+    await run_cycle(nas, _BadNaver(), state, load_settings())
+    assert nas.ingested is None
+    assert state.errors == 1
+    assert state.state == "idle"
+
+
+def test_build_heartbeat_schema():
+    state = FetchState()
+    state.state = "fetching"
+    state.listings_pushed = 7
+    d = json.loads(build_heartbeat(state))
+    assert d["name"] == "naver-fetch"
+    assert d["kind"] == "fetcher"
+    assert d["state"] == "fetching"
+    assert isinstance(d["ts"], int)
+    assert d["last_fetch_at"] is None
+    assert d["listings_pushed"] == 7
+    assert d["errors"] == 0