130 lines
4.5 KiB
Python
130 lines
4.5 KiB
Python
"""오케스트레이션 — 주간 스케줄 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))
|
|
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")
|
|
try:
|
|
articles: list[dict] = []
|
|
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
|
|
continue
|
|
try:
|
|
r = await nas.post_ingest(fetched_at, [{"dong": dong, "articles": articles}])
|
|
# BE ingest async화(main 8e28ce9): 응답 {received,new,queued} (matched는 백그라운드라 제거됨)
|
|
logger.info("ingest dong=%s received=%s new=%s queued=%s",
|
|
dong, r.get("received"), r.get("new"), r.get("queued"))
|
|
state.listings_pushed += len(articles)
|
|
except Exception:
|
|
logger.exception("listings-ingest 실패 dong=%s", dong)
|
|
state.errors += 1
|
|
|
|
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)
|