ai-trade/services/naver-fetch/PLAN.md

naver-fetch 워커 구현 계획

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 네이버 매물을 가정 IP에서 fetch해 NAS로 전달하는 naver-fetch 워커를 구현한다 — Playwright로 Bearer 토큰 획득 → httpx로 동별 article fetch → NAS listings-ingest로 raw 전달 + heartbeat.

Architecture: FastAPI + asyncio 워커(trade-monitor·insta-render 관례). naver_client가 Playwright(토큰)+httpx(fetch)를 캡슐화, fetch_worker가 2~3h 주간 스케줄로 오케스트레이션. 얇은 프록시(파싱은 NAS). heartbeat는 fetcher 전용 스키마(epoch ts).

Tech Stack: Python 3.12, FastAPI, playwright==1.48.0(async, chromium), httpx(async), redis.asyncio, pytest + pytest-asyncio + respx. WSL2 Docker.

설계 원본: services/naver-fetch/DESIGN.md · 권위 계약: web-backend realestate-lab/docs/superpowers/specs/2026-07-09-naver-listing-worker-design.md.

Global Constraints

• Python 3.12, playwright==1.48.0 (async API, chromium).
• 얇은 프록시 — raw article 그대로 전달, 파싱/매칭/알림/dedup은 NAS.
• 인증: .env INTERNAL_API_KEY → X-Internal-Key 헤더(render 워커 공유키). (스펙 §config의 REALESTATE_INTERNAL_KEY와 상충 — BE 메시지 우선, 배포 전 확인.)
• heartbeat 키 worker:naver-fetch:heartbeat, TTL EX45, kind fetcher, epoch ts: {name,kind,state:idle|fetching,ts,last_fetch_at,listings_pushed,errors}.
• ingest fetched_at은 ISO UTC 문자열(YYYY-MM-DDTHH:MM:SSZ).
• 스케줄: 주간창(기본 08:00~20:00 KST) 안에서 FETCH_INTERVAL_HOURS(기본 2.5) 간격.
• 테스트 실행: C:\Users\jaeoh\AppData\Local\Programs\Python\Python312\python.exe -m pytest services/naver-fetch/tests -q (web-ai 루트). 이하 python = 이 경로.
• 커밋: web-ai repo에서만. Docker 포트 18716.
• Spike 검증됨(2026-07-10): Playwright 토큰 인터셉트 + httpx /api/articles 200 + articleList.

---

File Structure

파일	책임
services/naver-fetch/config.py	Settings + load_settings()
services/naver-fetch/nas_client.py	NASClient: get_targets / post_ingest (X-Internal-Key)
services/naver-fetch/naver_client.py	NaverClient: acquire_token(Playwright) + fetch_articles(httpx)
services/naver-fetch/fetch_worker.py	FetchState, in_daytime, run_cycle, fetch_loop, build_heartbeat, heartbeat_loop
services/naver-fetch/main.py	FastAPI lifespan + /health
services/naver-fetch/conftest.py	services/ 루트 sys.path
services/naver-fetch/tests/*	단위 테스트
services/naver-fetch/Dockerfile	Playwright 베이스(insta-render 관례) + _shared
services/naver-fetch/requirements.txt, .env.example, pytest.ini
services/docker-compose.yml (수정)	naver-fetch 서비스(18716)

---

Task 1: 스캐폴딩 + config

Files:

• Create: services/naver-fetch/requirements.txt, conftest.py, pytest.ini, tests/__init__.py, config.py
• Test: services/naver-fetch/tests/test_config.py

Interfaces:

• Produces: Settings dataclass fields nas_base_url:str, internal_api_key:str, redis_url:str, fetch_interval_hours:float, daytime_start:str, daytime_end:str, naver_real_estate_type:str, naver_token_url:str, naver_headless:bool. load_settings() -> Settings.

• Step 1: 스캐폴딩

services/naver-fetch/requirements.txt:

fastapi==0.115.6
uvicorn[standard]==0.34.0
playwright==1.48.0
redis>=5.0
httpx>=0.27
pytest>=8.0
pytest-asyncio>=0.24
respx>=0.21

services/naver-fetch/conftest.py:

"""services/ 루트를 sys.path에 추가 — from _shared... import 가능하게."""
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

services/naver-fetch/pytest.ini:

[pytest]
asyncio_mode = auto

services/naver-fetch/tests/__init__.py: (빈 파일)

• Step 2: 실패 테스트 — services/naver-fetch/tests/test_config.py

"""Settings env 로드."""
from config import load_settings


def test_defaults(monkeypatch):
    for k in ("NAS_BASE_URL", "INTERNAL_API_KEY", "REDIS_URL", "FETCH_INTERVAL_HOURS",
              "DAYTIME_START", "DAYTIME_END", "NAVER_REAL_ESTATE_TYPE",
              "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.fetch_interval_hours == 2.5
    assert s.daytime_start == "08:00" and s.daytime_end == "20:00"
    assert s.naver_headless is True
    assert "new.land.naver.com" in s.naver_token_url


def test_override(monkeypatch):
    monkeypatch.setenv("INTERNAL_API_KEY", "sekret")
    monkeypatch.setenv("FETCH_INTERVAL_HOURS", "3")
    monkeypatch.setenv("NAVER_HEADLESS", "0")
    s = load_settings()
    assert s.internal_api_key == "sekret"
    assert s.fetch_interval_hours == 3.0
    assert s.naver_headless is False

• Step 3: 실패 확인

Run: python -m pytest services/naver-fetch/tests/test_config.py -q Expected: FAIL — ModuleNotFoundError: No module named 'config'

• Step 4: config.py 구현

"""Settings — 환경변수 로드."""
from __future__ import annotations

import os
from dataclasses import dataclass

_DEFAULT_TOKEN_URL = (
    "https://new.land.naver.com/houses?ms=37.489,126.921,16&a=APT:ABYG:JGC:PRE&e=RETAIL"
)


@dataclass
class Settings:
    nas_base_url: str
    internal_api_key: str
    redis_url: str
    fetch_interval_hours: float
    daytime_start: str
    daytime_end: str
    naver_real_estate_type: str
    naver_token_url: str
    naver_headless: bool


def load_settings() -> Settings:
    return Settings(
        nas_base_url=os.getenv("NAS_BASE_URL", "http://192.168.45.54:18500"),
        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")),
        daytime_start=os.getenv("DAYTIME_START", "08:00"),
        daytime_end=os.getenv("DAYTIME_END", "20:00"),
        naver_real_estate_type=os.getenv("NAVER_REAL_ESTATE_TYPE", "APT:ABYG:JGC:PRE"),
        naver_token_url=os.getenv("NAVER_TOKEN_URL", _DEFAULT_TOKEN_URL),
        naver_headless=os.getenv("NAVER_HEADLESS", "1") == "1",
    )

• Step 5: 통과 확인

Run: python -m pytest services/naver-fetch/tests/test_config.py -q Expected: PASS (2 passed)

• Step 6: 커밋

git add services/naver-fetch/requirements.txt services/naver-fetch/conftest.py services/naver-fetch/pytest.ini services/naver-fetch/tests/__init__.py services/naver-fetch/config.py services/naver-fetch/tests/test_config.py services/naver-fetch/DESIGN.md services/naver-fetch/PLAN.md
git commit -m "feat(naver-fetch): 스캐폴딩 + config"

---

Task 2: nas_client

Files:

• Create: services/naver-fetch/nas_client.py
• Test: services/naver-fetch/tests/test_nas_client.py

Interfaces:

• Produces NASClient:

  • __init__(base_url, api_key, timeout=15.0)
  • async get_targets() -> dict (GET /api/internal/realestate/targets)
  • async post_ingest(fetched_at: str, batches: list[dict]) -> dict (POST /api/internal/realestate/listings-ingest)
  • async close()

• Step 1: 실패 테스트 — services/naver-fetch/tests/test_nas_client.py

"""NASClient — targets/ingest + X-Internal-Key (respx)."""
import json as _json

import httpx
import respx

from nas_client import NASClient

BASE = "http://nas.test"


@respx.mock
async def test_get_targets_sends_key():
    route = respx.get(f"{BASE}/api/internal/realestate/targets").mock(
        return_value=httpx.Response(200, json={
            "dongs": [{"dong": "신대방동", "cortar_no": "1159010100"}],
            "deal_types": ["전세", "매매"], "page_limit": 2}))
    c = NASClient(BASE, "KEY")
    tg = await c.get_targets()
    assert tg["page_limit"] == 2
    assert tg["dongs"][0]["cortar_no"] == "1159010100"
    assert route.calls.last.request.headers["X-Internal-Key"] == "KEY"
    await c.close()


@respx.mock
async def test_post_ingest_payload():
    captured = {}

    def _resp(request):
        captured.update(_json.loads(request.content))
        return httpx.Response(200, json={"received": 3, "new": 2, "matched": 1})

    respx.post(f"{BASE}/api/internal/realestate/listings-ingest").mock(side_effect=_resp)
    c = NASClient(BASE, "KEY")
    batches = [{"dong": "신대방동", "articles": [{"articleNo": "1"}, {"articleNo": "2"}]}]
    out = await c.post_ingest("2026-07-10T05:00:00Z", batches)
    assert out == {"received": 3, "new": 2, "matched": 1}
    assert captured["fetched_at"] == "2026-07-10T05:00:00Z"
    assert captured["batches"] == batches
    await c.close()

• Step 2: 실패 확인

Run: python -m pytest services/naver-fetch/tests/test_nas_client.py -q Expected: FAIL — ModuleNotFoundError: No module named 'nas_client'

• Step 3: nas_client.py 구현

"""NAS realestate 내부 계약 — X-Internal-Key + retry."""
from __future__ import annotations

import asyncio
import logging

import httpx

logger = logging.getLogger(__name__)

_MAX_ATTEMPTS = 3
_RETRY_STATUSES = {429, 500, 502, 503, 504}


class NASClient:
    def __init__(self, base_url: str, api_key: str, timeout: float = 15.0):
        self._base_url = base_url.rstrip("/")
        self._api_key = api_key
        self._client = httpx.AsyncClient(timeout=timeout)

    async def close(self) -> None:
        await self._client.aclose()

    async def get_targets(self) -> dict:
        return await self._request("GET", "/api/internal/realestate/targets")

    async def post_ingest(self, fetched_at: str, batches: list[dict]) -> dict:
        return await self._request(
            "POST", "/api/internal/realestate/listings-ingest",
            json={"fetched_at": fetched_at, "batches": batches})

    async def _request(self, method: str, path: str, **kwargs) -> dict:
        url = f"{self._base_url}{path}"
        headers = {"X-Internal-Key": self._api_key}
        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:
                    await asyncio.sleep(2 ** attempt)
                    continue
                resp.raise_for_status()
                return resp.json()
            except httpx.TimeoutException:
                if attempt < _MAX_ATTEMPTS - 1:
                    await asyncio.sleep(2 ** attempt)
                    continue
                raise
        raise RuntimeError("retry exhausted")

• Step 4: 통과 확인

Run: python -m pytest services/naver-fetch/tests/test_nas_client.py -q Expected: PASS (2 passed)

• Step 5: 커밋

git add services/naver-fetch/nas_client.py services/naver-fetch/tests/test_nas_client.py
git commit -m "feat(naver-fetch): NAS realestate 클라이언트 (targets/ingest)"

---

Task 3: naver_client (Playwright 토큰 + httpx fetch)

Files:

• Create: services/naver-fetch/naver_client.py
• Test: services/naver-fetch/tests/test_naver_client.py

Interfaces:

• Produces NaverClient:

  • __init__(real_estate_type: str, token_url: str, headless: bool = True, timeout: float = 20.0)
  • async acquire_token() -> None — Playwright로 self._token(Bearer str), self._cookies(str) 설정. 실패 시 RuntimeError. (실브라우저 — 단위테스트 제외, spike/첫운영 검증)
  • async fetch_articles(cortar_no: str, page: int = 1) -> dict — httpx GET, raw JSON dict. 토큰 없으면 RuntimeError.
  • async close()

• Step 1: 실패 테스트 — services/naver-fetch/tests/test_naver_client.py

"""NaverClient.fetch_articles — httpx 경로 (respx). acquire_token은 실브라우저라 제외."""
import httpx
import pytest
import respx

from naver_client import NaverClient

API = "https://new.land.naver.com/api/articles"


def _client():
    return NaverClient(real_estate_type="APT:ABYG", token_url="https://x", headless=True)


async def test_fetch_requires_token():
    c = _client()
    with pytest.raises(RuntimeError):
        await c.fetch_articles("1159010100", 1)
    await c.close()


@respx.mock
async def test_fetch_articles_request_and_parse():
    captured = {}

    def _resp(request):
        captured["auth"] = request.headers.get("authorization")
        captured["cookie"] = request.headers.get("cookie")
        captured["url"] = str(request.url)
        return httpx.Response(200, json={
            "isMoreData": True,
            "articleList": [{"articleNo": "111", "articleName": "테스트아파트"}]})

    respx.get(API).mock(side_effect=_resp)
    c = _client()
    c._token = "Bearer TESTJWT"          # acquire_token 우회(단위테스트)
    c._cookies = "NNB=abc; page_uid=xyz"
    out = await c.fetch_articles("1159010100", page=2)
    assert out["articleList"][0]["articleNo"] == "111"
    assert captured["auth"] == "Bearer TESTJWT"
    assert captured["cookie"] == "NNB=abc; page_uid=xyz"
    assert "cortarNo=1159010100" in captured["url"]
    assert "page=2" in captured["url"]
    assert "realEstateType=APT" in captured["url"]
    await c.close()


@respx.mock
async def test_fetch_non_200_raises():
    respx.get(API).mock(return_value=httpx.Response(429, json={"code": "TOO_MANY_REQUESTS"}))
    c = _client()
    c._token = "Bearer X"
    with pytest.raises(httpx.HTTPStatusError):
        await c.fetch_articles("1159010100", 1)
    await c.close()

• Step 2: 실패 확인

Run: python -m pytest services/naver-fetch/tests/test_naver_client.py -q Expected: FAIL — ModuleNotFoundError: No module named 'naver_client'

• Step 3: naver_client.py 구현

"""Naver land client — Playwright 토큰 획득 + httpx article fetch."""
from __future__ import annotations

import logging

import httpx

logger = logging.getLogger(__name__)

_UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
       "(KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36")
_API = "https://new.land.naver.com/api/articles"


class NaverClient:
    def __init__(self, real_estate_type: str, token_url: str,
                 headless: bool = True, timeout: float = 20.0):
        self._ret = real_estate_type
        self._token_url = token_url
        self._headless = headless
        self._client = httpx.AsyncClient(timeout=timeout)
        self._token: str | None = None
        self._cookies: str = ""

    async def close(self) -> None:
        await self._client.aclose()

    async def acquire_token(self) -> None:
        """new.land.naver.com 로드 → XHR Authorization 헤더 인터셉트."""
        from playwright.async_api import async_playwright

        captured: dict[str, str] = {}
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=self._headless)
            try:
                ctx = await browser.new_context(user_agent=_UA, locale="ko-KR")
                page = await ctx.new_page()

                def on_request(req):
                    auth = req.headers.get("authorization")
                    if auth and "bearer" in auth.lower() and "auth" not in captured:
                        captured["auth"] = auth

                page.on("request", on_request)
                try:
                    await page.goto(self._token_url, wait_until="networkidle", timeout=45000)
                except Exception:
                    logger.warning("naver token goto 지연/부분 실패 — 캡처 계속")
                await page.wait_for_timeout(4000)
                cookies = await ctx.cookies()
                self._cookies = "; ".join(f"{c['name']}={c['value']}" for c in cookies)
            finally:
                await browser.close()

        token = captured.get("auth")
        if not token:
            raise RuntimeError("naver Bearer token 획득 실패")
        self._token = token
        logger.info("naver token 획득 (len=%d)", len(token))

    async def fetch_articles(self, cortar_no: str, page: int = 1) -> dict:
        if not self._token:
            raise RuntimeError("token 없음 — acquire_token 먼저 호출")
        params = {
            "cortarNo": cortar_no, "order": "rank", "realEstateType": self._ret,
            "tradeType": "", "rentPriceMin": 0, "rentPriceMax": 900000000,
            "priceMin": 0, "priceMax": 900000000, "areaMin": 0, "areaMax": 900000000,
            "showArticle": "false", "sameAddressGroup": "false",
            "priceType": "RETAIL", "page": page,
        }
        headers = {
            "authorization": self._token, "user-agent": _UA,
            "referer": "https://new.land.naver.com/",
            "accept": "application/json, text/plain, */*",
            "accept-language": "ko-KR,ko;q=0.9", "cookie": self._cookies,
        }
        resp = await self._client.get(_API, params=params, headers=headers)
        resp.raise_for_status()
        return resp.json()

• Step 4: 통과 확인

Run: python -m pytest services/naver-fetch/tests/test_naver_client.py -q Expected: PASS (3 passed)

• Step 5: 커밋

git add services/naver-fetch/naver_client.py services/naver-fetch/tests/test_naver_client.py
git commit -m "feat(naver-fetch): NaverClient (Playwright 토큰 + httpx fetch)"

---

Task 4: fetch_worker (오케스트레이션 + heartbeat)

Files:

• Create: services/naver-fetch/fetch_worker.py
• Test: services/naver-fetch/tests/test_fetch_worker.py

Interfaces:

• Consumes: NASClient, NaverClient, config.Settings.

• Produces:

  • class FetchState — .state="idle", .last_fetch_at=None, .listings_pushed=0, .errors=0
  • in_daytime(now: datetime, start: str, end: str) -> bool
  • async run_cycle(nas, naver, state, settings) -> None
  • async fetch_loop(nas, naver, state, settings) -> None
  • build_heartbeat(state) -> str (JSON)
  • async heartbeat_loop(redis, state, interval=15, ttl=45) -> None

• Step 1: 실패 테스트 — services/naver-fetch/tests/test_fetch_worker.py

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

• Step 2: 실패 확인

Run: python -m pytest services/naver-fetch/tests/test_fetch_worker.py -q Expected: FAIL — ModuleNotFoundError: No module named 'fetch_worker'

• Step 3: fetch_worker.py 구현

"""오케스트레이션 — 주간 스케줄 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)

• Step 4: 통과 확인

Run: python -m pytest services/naver-fetch/tests/test_fetch_worker.py -q Expected: PASS (5 passed)

• Step 5: 커밋

git add services/naver-fetch/fetch_worker.py services/naver-fetch/tests/test_fetch_worker.py
git commit -m "feat(naver-fetch): fetch_worker (run_cycle/loop/heartbeat)"

---

Task 5: main.py (FastAPI lifespan + /health)

Files:

• Create: services/naver-fetch/main.py
• Test: services/naver-fetch/tests/test_health.py

Interfaces:

• Consumes: config.load_settings, NASClient, NaverClient, fetch_worker.*, redis.asyncio.

• Produces: FastAPI app + health() → {"ok": True, "service": "naver-fetch"}.

• Step 1: 실패 테스트 — services/naver-fetch/tests/test_health.py

"""/health — 핸들러 직접 검증."""
from main import health


def test_health():
    assert health() == {"ok": True, "service": "naver-fetch"}

• Step 2: 실패 확인

Run: python -m pytest services/naver-fetch/tests/test_health.py -q Expected: FAIL — ModuleNotFoundError: No module named 'main'

• Step 3: main.py 구현

"""naver-fetch FastAPI entry — lifespan(fetch_loop + heartbeat_loop) + /health."""
from __future__ import annotations

import asyncio
import logging
from contextlib import asynccontextmanager

import redis.asyncio as aioredis
from fastapi import FastAPI

import fetch_worker
from config import load_settings
from nas_client import NASClient
from naver_client import NaverClient

logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s %(name)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI):
    settings = load_settings()
    nas = NASClient(settings.nas_base_url, settings.internal_api_key)
    naver = NaverClient(settings.naver_real_estate_type, settings.naver_token_url,
                        settings.naver_headless)
    state = fetch_worker.FetchState()
    redis = aioredis.from_url(settings.redis_url, decode_responses=False)

    ft = asyncio.create_task(fetch_worker.fetch_loop(nas, naver, state, settings))
    hb = asyncio.create_task(fetch_worker.heartbeat_loop(redis, state))
    logger.info("naver-fetch lifespan 시작")
    try:
        yield
    finally:
        for t in (ft, hb):
            t.cancel()
            try:
                await t
            except asyncio.CancelledError:
                pass
        await naver.close()
        await nas.close()
        await redis.aclose()
        logger.info("naver-fetch lifespan 종료")


app = FastAPI(lifespan=lifespan)


@app.get("/health")
def health():
    return {"ok": True, "service": "naver-fetch"}

• Step 4: 통과 확인 + 전체 스위트

Run: python -m pytest services/naver-fetch/tests -q Expected: PASS (config 2 + nas 2 + naver 3 + fetch_worker 5 + health 1 = 13)

• Step 5: 커밋

git add services/naver-fetch/main.py services/naver-fetch/tests/test_health.py
git commit -m "feat(naver-fetch): FastAPI lifespan + heartbeat 배선 + /health"

---

Task 6: 배포 (Dockerfile + compose + .env.example)

Files:

• Create: services/naver-fetch/Dockerfile, services/naver-fetch/.env.example
• Modify: services/docker-compose.yml

Interfaces: 없음(배포). 검증 docker compose config.

• Step 1: Dockerfile — services/naver-fetch/Dockerfile (insta-render Chromium 관례 복제)

FROM python:3.12-slim-bookworm
ENV PYTHONUNBUFFERED=1

WORKDIR /app

# Chromium 런타임 deps (Debian 12 / bookworm) — insta-render 관례
RUN apt-get update && apt-get install -y --no-install-recommends \
    ca-certificates tzdata fonts-noto-cjk \
    libnss3 libnspr4 libdbus-1-3 libatk1.0-0 libatk-bridge2.0-0 \
    libcups2 libdrm2 libxkbcommon0 libxcomposite1 libxdamage1 \
    libxfixes3 libxrandr2 libgbm1 libxshmfence1 libpango-1.0-0 \
    libcairo2 libasound2 libatspi2.0-0 \
 && rm -rf /var/lib/apt/lists/*

COPY naver-fetch/requirements.txt /app/
RUN pip install --no-cache-dir --timeout 600 --retries 5 -r requirements.txt
RUN playwright install chromium

# 공통 heartbeat 등 모듈 (services/_shared) — 빌드 컨텍스트 . 필요
COPY _shared /app/_shared
COPY naver-fetch/. /app/
ENV PYTHONPATH=/app

EXPOSE 8000
CMD ["python", "-m", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]

• Step 2: .env.example — services/naver-fetch/.env.example

# naver-fetch — realestate 매물 fetch 워커

REDIS_URL=redis://192.168.45.54:6379
NAS_BASE_URL=http://192.168.45.54:18500
# render 워커 공유키 (X-Internal-Key)
INTERNAL_API_KEY=

# 스케줄 (KST)
FETCH_INTERVAL_HOURS=2.5
DAYTIME_START=08:00
DAYTIME_END=20:00

# 네이버
NAVER_REAL_ESTATE_TYPE=APT:ABYG:JGC:PRE
NAVER_HEADLESS=1
# NAVER_TOKEN_URL=  # 기본값(new.land.naver.com houses) 사용

• Step 3: docker-compose.yml에 서비스 추가

services/docker-compose.yml의 trade-monitor 블록 뒤(파일 끝, 들여쓰기 2칸)에 추가:

  naver-fetch:
    build:
      context: .
      dockerfile: naver-fetch/Dockerfile
    container_name: naver-fetch
    restart: unless-stopped
    ports:
      - "18716:8000"
    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}
      - INTERNAL_API_KEY=${INTERNAL_API_KEY:-}
      - FETCH_INTERVAL_HOURS=${FETCH_INTERVAL_HOURS:-2.5}
      - DAYTIME_START=${DAYTIME_START:-08:00}
      - DAYTIME_END=${DAYTIME_END:-20:00}
      - NAVER_REAL_ESTATE_TYPE=${NAVER_REAL_ESTATE_TYPE:-APT:ABYG:JGC:PRE}
      - NAVER_HEADLESS=${NAVER_HEADLESS:-1}
    healthcheck:
      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"]
      interval: 60s
      timeout: 5s
      retries: 3

• Step 4: compose 검증

Run: cd services && docker compose config (실패 시 YAML 들여쓰기 육안 확인) Expected: naver-fetch 서비스 파싱, 포트 18716.

• Step 5: 커밋

git add services/naver-fetch/Dockerfile services/naver-fetch/.env.example services/docker-compose.yml
git commit -m "feat(naver-fetch): Dockerfile(Playwright) + compose 서비스(18716) + .env.example"

---

Self-Review

1. Spec coverage:

• §4 GET /targets → Task 2 get_targets + Task 4 사용. ✅
• §4 POST /listings-ingest(fetched_at ISO, batches) → Task 2 post_ingest + Task 4 조립. ✅
• §4.4 heartbeat(EX45, kind=fetcher, epoch ts, 필드) → Task 4 build_heartbeat/heartbeat_loop. ✅
• 워커 루프(targets→네이버 fetch→ingest) → Task 4 run_cycle. ✅
• 네이버 Bearer 토큰(Playwright) + fetch → Task 3. ✅
• 스케줄(주간 2~3h) → Task 4 in_daytime + fetch_loop. ✅
• 얇은 프록시(raw 전달) → Task 4(파싱 없음, articleList 그대로). ✅
• 배포(_shared COPY, Playwright, 18716) → Task 6. ✅

2. Placeholder scan: 실제 코드/명령/기대출력 기재. TODO 없음. ✅

3. Type consistency: FetchState(state/last_fetch_at/listings_pushed/errors)가 Task 4 생성 ↔ build_heartbeat 소비 일치. NaverClient(acquire_token/fetch_articles) Task 3 정의 ↔ Task 4 run_cycle 호출 일치. NASClient(get_targets/post_ingest) Task 2 ↔ Task 4 일치. fetched_at ISO 문자열/heartbeat ts epoch 구분 유지. ✅

미해결 플래그(DESIGN §11): INTERNAL_API_KEY vs REALESTATE_INTERNAL_KEY(배포 전 BE 확인), 헤드리스 탐지, deal_types 미사용(NAS 필터), 토큰 만료 재획득.