From e3c7f2479440c6047fbc24cd7556b4047ed6d6d4 Mon Sep 17 00:00:00 2001 From: gahusb Date: Fri, 10 Jul 2026 10:21:03 +0900 Subject: [PATCH] feat(naver-fetch): fetch_worker (run_cycle/loop/heartbeat) --- services/naver-fetch/fetch_worker.py | 130 ++++++++++++++++++ .../naver-fetch/tests/test_fetch_worker.py | 106 ++++++++++++++ 2 files changed, 236 insertions(+) create mode 100644 services/naver-fetch/fetch_worker.py create mode 100644 services/naver-fetch/tests/test_fetch_worker.py diff --git a/services/naver-fetch/fetch_worker.py b/services/naver-fetch/fetch_worker.py new file mode 100644 index 0000000..664c791 --- /dev/null +++ b/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) diff --git a/services/naver-fetch/tests/test_fetch_worker.py b/services/naver-fetch/tests/test_fetch_worker.py new file mode 100644 index 0000000..5e4d2ec --- /dev/null +++ b/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