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

feat(naver-fetch): 스캐폴딩 + config

Browse Source
This commit is contained in:
gahusb
2026-07-10 10:21:03 +09:00
parent 9abca3eeab
commit 6b0e91342c
8 changed files with 1205 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

150
services/naver-fetch/DESIGN.md Normal file
View File

@@ -0,0 +1,150 @@
# naver-fetch 워커 — 구현 설계
> **2026-07-10 · web-ai 소유.** realestate 매물 알림 파이프라인의 가정-IP fetch 워커.
> **권위 계약(원본 스펙)**: web-backend repo `realestate-lab/docs/superpowers/specs/2026-07-09-naver-listing-worker-design.md` §4.
> 이 문서는 그 계약을 Windows Docker 워커로 구현하기 위한 **구현 설계**.
---
## 1. 역할 · 경계
네이버가 NAS(데이터센터) IP를 429/ReadTimeout으로 막기 때문에, **가정 IP(LAN 192.168.45.x)에서 네이버 매물을 fetch해 NAS로 전달**하는 얇은 프록시 워커. trade-monitor·render 워커와 동일한 FastAPI + asyncio 관례. WSL2 Docker.
**책임**
1. 2~3h 주간 주기로 NAS `targets` 조회(감시 동 목록).
2. Playwright로 네이버 Bearer 토큰 획득 → httpx로 동별 매물 fetch(페이지네이션).
3. raw article을 그대로 NAS `listings-ingest`로 전달(**파싱·매칭·알림은 NAS**).
4. Redis heartbeat(`worker:naver-fetch:heartbeat` EX45, kind=fetcher).
5. 토큰/네트워크 오류를 사이클·동 단위로 격리.
**경계 밖(안 함)**: article 파싱, 매물 매칭, 텔레그램 발송(전부 NAS), dedup.
### Spike 검증 완료 (2026-07-10)
- Playwright 헤드리스 → `new.land.naver.com` 로드 → `/api/articles` XHR의 **Bearer JWT 토큰 인터셉트 성공**.
- 인터셉트 토큰+쿠키를 httpx로 재사용 → `GET /api/articles?cortarNo=1159010100&...` **HTTP 200, articleList 20건**(isMoreData 페이지네이션). 토큰 있으면 이 LAN에서 429 회피됨.
- raw article 키: `articleNo, articleName, tradeTypeName, dealOrWarrantPrc, areaName, buildingName, floorInfo` 등 → NAS로 그대로 전달.
---
## 2. 모듈 분해
| 파일 | 책임 | 인터페이스 |
|------|------|-----------|
| `main.py` | FastAPI + lifespan. `fetch_loop` + `heartbeat_loop` 스폰, `/health` | 부작용 |
| `fetch_worker.py` | `FetchState`, `run_cycle`, `fetch_loop`, `heartbeat_loop`(전용) | 부작용 |
| `nas_client.py` | `get_targets()` / `post_ingest(fetched_at, batches)` (X-Internal-Key + retry) | 부작용(HTTP) |
| `naver_client.py` | `NaverClient`: `acquire_token()`(Playwright) + `fetch_articles(cortar_no, page)`(httpx) | 부작용(브라우저/HTTP) |
| `config.py` | `Settings` — env 로드 | 순수 |
순수 로직이 적은 워커(얇은 프록시)라 파싱은 없음. 테스트는 HTTP mock(respx) + fetch_worker 조립 중심.
---
## 3. 데이터 흐름 (run_cycle)
```
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 격리
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)
state.last_fetch_at = epoch_now
state.state = "idle"
```
> ⚠️ **시간 포맷 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 페이로드 발신.
---
## 4. naver_client (Playwright 토큰 + httpx fetch)
- `async acquire_token()`: `async_playwright`로 chromium 헤드리스 기동 → `new.land.naver.com/houses?ms=...`(설정 `NAVER_TOKEN_URL`) 로드 → `page.on("request")`로 첫 `Authorization: Bearer ...` 헤더 캡처 → `self._token`, `self._cookies`(context.cookies) 저장 → 브라우저 종료. 실패 시 예외(사이클 skip).
- **토큰 수명 정책**: 사이클마다 fresh 획득(2~3h당 1회). JWT 만료 무관, 캐시/갱신 로직 없음(단순).
- `async fetch_articles(cortar_no, page) -> dict`: httpx GET `https://new.land.naver.com/api/articles?cortarNo=<cortar_no>&order=rank&realEstateType=<NAVER_REAL_ESTATE_TYPE>&tradeType=&...&page=<page>` with headers `{authorization: self._token, user-agent, referer: https://new.land.naver.com/, accept, cookie: self._cookies}`. 200 아니면 예외. raw JSON dict 반환.
- `async close()`: httpx client 정리.
**얇은 프록시**: tradeType 비움(전체 유형) + realEstateType 설정값(기본 `APT:ABYG:JGC:PRE`). deal_types 필터는 NAS 담당(targets의 deal_types는 워커가 그대로 두고 NAS가 매칭 시 사용). 추후 네이버 쿼리 최적화 여지.
---
## 5. heartbeat (§4.4 — 전용 스키마)
`worker:naver-fetch:heartbeat` EX45. **trade-monitor의 `_shared.build_payload`(ISO ts + jobs_done/jobs_failed)와 스키마가 달라 전용 emit 작성**:
```json
{"name":"naver-fetch","kind":"fetcher","state":"idle|fetching",
"ts":<epoch int>,"last_fetch_at":<epoch int|null>,"listings_pushed":<int>,"errors":<int>}
```
- 15초 독립 태스크(주기 fetch가 2~3h라 TTL45 유지 위해 필수). `FetchState`를 읽어 발신.
- BE가 node_monitor.WORKER_REGISTRY에 `naver-fetch`(kind=fetcher)로 등재 완료 → 발신 시 `/nodes`·`/infra`에서 alive.
---
## 6. 스케줄 (fetch_loop)
- env `FETCH_INTERVAL_HOURS`(기본 2.5), `DAYTIME_START`/`DAYTIME_END`(기본 08:00/20:00 KST).
- 루프: ~5분마다 판정 — **주간창 안 && (now - last_run) ≥ interval** 이면 `run_cycle` 실행 후 `last_run=now`. 창 밖이면 idle(state=idle, fetch 안 함).
- 최상위 try/except로 루프 불사(어떤 예외도 죽이지 않음).
---
## 7. 설정 (env)
| env | 기본값 | 용도 |
|-----|--------|------|
| `NAS_BASE_URL` | `http://192.168.45.54:18500` | targets/ingest |
| `INTERNAL_API_KEY` | (필수) | `X-Internal-Key` (render 워커 공유키) |
| `REDIS_URL` | `redis://192.168.45.54:6379` | heartbeat |
| `FETCH_INTERVAL_HOURS` | `2.5` | fetch 주기 |
| `DAYTIME_START` / `DAYTIME_END` | `08:00` / `20:00` | 주간창(KST) |
| `NAVER_REAL_ESTATE_TYPE` | `APT:ABYG:JGC:PRE` | 네이버 realEstateType |
| `NAVER_TOKEN_URL` | (기본 상수) | Playwright 토큰 획득용 네이버 URL |
| `NAVER_HEADLESS` | `1` | Playwright headless |
---
## 8. 에러 처리
- **토큰 획득 실패**: 사이클 skip, `errors++`, state=idle, 다음 주기 재시도.
- **targets 실패**: 사이클 skip, `errors++`.
- **동별 fetch 실패**: 해당 동 skip(warning), 나머지 동 계속.
- **ingest 실패**: 로그 error, `errors++`(다음 주기 재fetch).
- 루프 최상위 try/except.
---
## 9. 테스트 (pytest, 시스템 Python)
| 파일 | 검증 |
|------|------|
| `test_config.py` | env 기본/override |
| `test_nas_client.py` | respx — targets 파싱, ingest 페이로드/응답, X-Internal-Key, retry |
| `test_naver_client.py` | respx — fetch_articles URL·헤더(authorization/cookie)·articleList 파싱·비200 예외. (Playwright acquire_token은 실브라우저 통합이라 단위 mock: 토큰 주입 후 fetch 경로만) |
| `test_fetch_worker.py` | mock: 주간창 게이트(창 밖 skip), run_cycle 동별 조립·batches, 동 fetch 실패 격리, listings_pushed/last_fetch_at 갱신, heartbeat 필드(epoch ts/kind=fetcher) |
---
## 10. 배포
- `Dockerfile`: Playwright 베이스(`mcr.microsoft.com/playwright/python` 또는 insta-render 관례) + chromium, `COPY _shared /app/_shared`(빌드 컨텍스트 `.`), `COPY naver-fetch/. /app/`, `PYTHONPATH=/app`, uvicorn `:8000`.
- `services/docker-compose.yml`: `naver-fetch` 서비스, 포트 **18716**, `TZ=Asia/Seoul`, INTERNAL_API_KEY/REDIS/NAS env, healthcheck `/health`.
- 엔드포인트는 nginx `/api/internal/realestate/`(IP 화이트리스트, render 워커 IP 재사용).
---
## 11. 미해결 플래그 / 후속
1. **`INTERNAL_API_KEY` vs `REALESTATE_INTERNAL_KEY`** — 스펙 §config는 후자, BE 메시지는 전자(render 공유키). **BE 메시지 우선(INTERNAL_API_KEY)**, 배포 전 co-gahusb로 1줄 확인.
2. **네이버 헤드리스 탐지** — 현재 spike는 통과. 차단 시 stealth 헤더/headed 필요할 수 있음(첫 운영 모니터링).
3. **`isMoreData`/페이지 상한** — page_limit로 제한, 네이버 rate limit 재유입 시 동 간 지연 고려(첫 운영).
4. **deal_types 미사용** — 워커는 전체 유형 fetch, NAS가 필터. 필요 시 tradeType 매핑으로 최적화.
5. **토큰 재사용 실패(만료) 대비** — 사이클 중 fetch가 401 반환 시 acquire_token 1회 재시도(구현 시 반영 검토).

979
services/naver-fetch/PLAN.md Normal file
View File

@@ -0,0 +1,979 @@
# 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`:
```python
"""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`
```python
"""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 구현**
```python
"""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: 커밋**
```bash
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`
```python
"""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 구현**
```python
"""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: 커밋**
```bash
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`
```python
"""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 구현**
```python
"""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: 커밋**
```bash
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`
```python
"""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 구현**
```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))
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: 커밋**
```bash
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`
```python
"""/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 구현**
```python
"""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: 커밋**
```bash
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 관례 복제)
```dockerfile
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칸)에 추가:
```yaml
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: 커밋**
```bash
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 필터), 토큰 만료 재획득.

36
services/naver-fetch/config.py Normal file
View File

@@ -0,0 +1,36 @@
"""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",
)

5
services/naver-fetch/conftest.py Normal file
View File

@@ -0,0 +1,5 @@
"""services/ 루트를 sys.path에 추가 — from _shared... import 가능하게."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

2
services/naver-fetch/pytest.ini Normal file
View File

@@ -0,0 +1,2 @@
[pytest]
asyncio_mode = auto

8
services/naver-fetch/requirements.txt Normal file
View File

@@ -0,0 +1,8 @@
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

0
services/naver-fetch/tests/__init__.py Normal file
View File

25
services/naver-fetch/tests/test_config.py Normal file
View File

@@ -0,0 +1,25 @@
"""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