# 네이버 매물 fetch 워커 — BE 구현 Plan

> **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:** naver-fetch 워커가 네이버 호가 매물을 push할 수 있도록 realestate-lab에 내부 계약(targets 조회 + listings ingest), 관측 등재, cortarNo 매핑을 구축한다. 워커(web-ai)·FE(/infra)는 별도 세션.

**Architecture:** 워커는 `GET /api/internal/realestate/targets`로 대상 동+10자리 cortarNo를 받고, 네이버 raw 매물을 `POST /api/internal/realestate/listings-ingest`로 push. NAS는 기존 `_parse_naver_article`로 파싱→`upsert_listing`→`run_listing_matching`→`notify_new_listings`. 인증은 render 워커와 동형(X-Internal-Key + nginx IP화이트리스트). 관측은 heartbeat + node_monitor `fetcher` kind.

**Tech Stack:** Python 3.12, FastAPI(APIRouter+Depends), SQLite, pytest. 외부 호출 테스트는 전부 mock.

**스펙:** `realestate-lab/docs/superpowers/specs/2026-07-09-naver-listing-worker-design.md`

## Global Constraints

- **기존 매물/청약 파이프라인 불변** — 신규 내부 라우터·매핑·등재만 추가. 안전마진 판정 로직 불변.
- **인증 패턴 준수**: `verify_internal_key(x_internal_key: str = Header(...))`, env **`INTERNAL_API_KEY`**, 실패 시 `HTTPException(401, "Invalid X-Internal-Key")`. (스펙의 `REALESTATE_INTERNAL_KEY` 대신 기존 서비스 관례 `INTERNAL_API_KEY` 사용 — image/video/insta-lab 일관성.)
- **파싱은 한 곳**: 네이버 article 파싱은 기존 `listing_collector._parse_naver_article` 재사용(중복 구현 금지).
- **멱등**: `listings.article_no` UNIQUE upsert + `listing_matches.notified_at` → 재push해도 재알림 없음.
- **테스트**: `cd realestate-lab && PYTHONPATH="C:/Users/jaeoh/Desktop/workspace/web-backend" python -m pytest tests/<file> -q` (main import은 `_shared` 때문에 PYTHONPATH에 web-backend 루트 필요). agent-office 테스트는 `cd agent-office && python -m pytest`.
- **cortarNo는 10자리 법정동코드** (기존 `LAWD_CODES` 5자리 시군구는 MOLIT 전용 유지).
- **nginx/deploy는 locked 리소스**: nginx 변경·push는 `nginx-conf`·`nas-deploy` 락 획득 후(Task 6).

## File Structure

| 파일 | 신규/수정 | 책임 |
|---|---|---|
| `realestate-lab/app/lawd_codes.py` | Modify | `NAVER_CORTAR`(동→10자리) + `naver_cortar()` 추가 |
| `realestate-lab/app/auth.py` | Create | `verify_internal_key` (X-Internal-Key) |
| `realestate-lab/app/internal_router.py` | Create | `GET targets` + `POST listings-ingest` |
| `realestate-lab/app/main.py` | Modify | `include_router(internal_router)` |
| `nginx/default.conf` | Modify | `/api/internal/realestate/` 블록 |
| `realestate-lab/app/listing_collector.py` | Modify | `collect_listings` MOLIT 전용화(naver deprecate) |
| `agent-office/app/node_monitor.py` | Modify | `WORKER_REGISTRY` fetcher 등재 + link 분기 |

---

### Task 1: NAVER_CORTAR 매핑 (10자리 법정동코드)

**Files:**
- Modify: `realestate-lab/app/lawd_codes.py`
- Test: `realestate-lab/tests/test_lawd_codes.py` (기존 파일에 append)

**Interfaces:**
- Produces: `NAVER_CORTAR: dict[str,str]`, `naver_cortar(dong: str) -> str | None`

- [ ] **Step 1: 실패 테스트 append** — `tests/test_lawd_codes.py`

```python
def test_naver_cortar_10digit():
    from app.lawd_codes import naver_cortar, NAVER_CORTAR
    assert naver_cortar("신대방동") == "1159010100"
    assert naver_cortar("봉천동") == "1162010100"
    assert naver_cortar("신길동") == "1156013200"
    assert naver_cortar("없는동") is None
    # 6개 초기 대상 동 전부 10자리
    for d in ["신대방동","대방동","상도동","봉천동","대림동","신길동"]:
        assert len(NAVER_CORTAR[d]) == 10
```

- [ ] **Step 2: 실패 확인** — `cd realestate-lab && PYTHONPATH="C:/Users/jaeoh/Desktop/workspace/web-backend" python -m pytest tests/test_lawd_codes.py -q` → FAIL(ImportError naver_cortar)

- [ ] **Step 3: 구현** — `app/lawd_codes.py` 하단에 추가

```python
# 네이버부동산 cortarNo = 10자리 법정동코드 (LAWD_CODES 5자리 시군구와 별개).
# 출처: 행정표준코드관리시스템 법정동코드. 대상 동 추가 시 여기에.
NAVER_CORTAR = {
    "신대방동": "1159010100",
    "대방동":   "1159010200",
    "상도동":   "1159010600",
    "봉천동":   "1162010100",
    "대림동":   "1156013600",
    "신길동":   "1156013200",
}


def naver_cortar(dong: str) -> str | None:
    return NAVER_CORTAR.get((dong or "").strip())
```

- [ ] **Step 4: 통과 확인** — 같은 pytest → PASS
- [ ] **Step 5: 커밋**

```bash
git add realestate-lab/app/lawd_codes.py realestate-lab/tests/test_lawd_codes.py
git commit -m "feat(realestate): 네이버 cortarNo 10자리 법정동 매핑(NAVER_CORTAR)"
```

---

### Task 2: 내부 인증 + targets 엔드포인트 + 라우터 배선

**Files:**
- Create: `realestate-lab/app/auth.py`, `realestate-lab/app/internal_router.py`
- Modify: `realestate-lab/app/main.py`
- Test: `realestate-lab/tests/test_internal_targets.py`

**Interfaces:**
- Consumes: Task1 `naver_cortar`; 기존 `db.get_listing_criteria`.
- Produces: `verify_internal_key` dependency; `GET /api/internal/realestate/targets` → `{"dongs":[{"dong","cortar_no"}],"deal_types":[...],"page_limit":int}`; `router` (APIRouter, main에 include).

- [ ] **Step 1: 실패 테스트** — `tests/test_internal_targets.py`

```python
import os
from fastapi.testclient import TestClient

def _client(monkeypatch):
    monkeypatch.setenv("INTERNAL_API_KEY", "SECRET")
    import importlib
    from app import auth as _a
    importlib.reload(_a)  # env 반영
    from app.main import app
    return TestClient(app)

def test_targets_requires_key(monkeypatch):
    c = _client(monkeypatch)
    assert c.get("/api/internal/realestate/targets").status_code == 401           # 헤더 없음
    assert c.get("/api/internal/realestate/targets",
                 headers={"X-Internal-Key": "WRONG"}).status_code == 401

def test_targets_returns_dongs_with_cortar(monkeypatch):
    c = _client(monkeypatch)
    r = c.get("/api/internal/realestate/targets", headers={"X-Internal-Key": "SECRET"})
    assert r.status_code == 200
    body = r.json()
    dongs = {d["dong"]: d["cortar_no"] for d in body["dongs"]}
    assert dongs.get("신대방동") == "1159010100"     # criteria seed 6동 + cortar join
    assert "전세" in body["deal_types"]
    assert isinstance(body["page_limit"], int)
```

- [ ] **Step 2: 실패 확인** — `PYTHONPATH=... python -m pytest tests/test_internal_targets.py -q` → FAIL(404/ImportError)

- [ ] **Step 3: 구현**

`app/auth.py`:
```python
"""Windows naver-fetch 워커 → NAS realestate-lab 내부 엔드포인트 인증."""
import os
from fastapi import Header, HTTPException


def verify_internal_key(x_internal_key: str = Header(...)):
    expected = os.getenv("INTERNAL_API_KEY")
    if not expected:
        raise HTTPException(401, "INTERNAL_API_KEY not configured on server")
    if x_internal_key != expected:
        raise HTTPException(401, "Invalid X-Internal-Key")
```

`app/internal_router.py`:
```python
"""naver-fetch 워커 내부 계약: targets 조회 + listings ingest."""
import os
import logging
from fastapi import APIRouter, Depends

from .auth import verify_internal_key
from .db import get_listing_criteria
from .lawd_codes import naver_cortar

logger = logging.getLogger("realestate-lab")
router = APIRouter()

NAVER_PAGE_LIMIT = int(os.getenv("NAVER_PAGE_LIMIT", "2"))


@router.get("/api/internal/realestate/targets", dependencies=[Depends(verify_internal_key)])
def listing_targets():
    crit = get_listing_criteria()
    dongs = []
    for d in crit.get("dongs", []):
        code = naver_cortar(d)
        if not code:
            logger.warning("naver cortarNo 매핑 없음 — 대상 제외: %s", d)
            continue
        dongs.append({"dong": d, "cortar_no": code})
    return {"dongs": dongs, "deal_types": crit.get("deal_types", []),
            "page_limit": NAVER_PAGE_LIMIT}
```

`app/main.py`: import 블록에 추가(기존 `from .models import (...)` 아래)
```python
from .internal_router import router as internal_router
```
그리고 `app = FastAPI(lifespan=lifespan)` + `install_access_log(app)` **직후**에:
```python
app.include_router(internal_router)
```

`nginx/default.conf`: `/api/realestate/` location 블록 **앞**에 추가(더 구체적 prefix 우선):
```nginx
location /api/internal/realestate/ {
    allow 192.168.45.0/24;
    allow 100.64.0.0/10;
    allow 127.0.0.1;
    deny all;

    proxy_http_version 1.1;
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Internal-Key $http_x_internal_key;
    proxy_pass http://realestate-lab:8000/api/internal/realestate/;
}
```

- [ ] **Step 4: 통과 확인** — 같은 pytest → PASS (nginx는 로컬 테스트 무관, 배포 시 검증)
- [ ] **Step 5: 커밋**

```bash
git add realestate-lab/app/auth.py realestate-lab/app/internal_router.py realestate-lab/app/main.py nginx/default.conf realestate-lab/tests/test_internal_targets.py
git commit -m "feat(realestate): 내부 인증 + naver 워커 targets 엔드포인트 + nginx internal 블록"
```

---

### Task 3: listings-ingest 엔드포인트

**Files:**
- Modify: `realestate-lab/app/internal_router.py`
- Test: `realestate-lab/tests/test_internal_ingest.py`

**Interfaces:**
- Consumes: 기존 `db.upsert_listing`, `listing_collector._parse_naver_article`, `listing_matcher.run_listing_matching`, `notifier.notify_new_listings`.
- Produces: `POST /api/internal/realestate/listings-ingest` body `{fetched_at:str, batches:[{dong:str, articles:[dict]}]}` → `{"received":int,"new":int,"matched":int}`.

- [ ] **Step 1: 실패 테스트** — `tests/test_internal_ingest.py`

```python
import os, importlib
from unittest.mock import patch
from fastapi.testclient import TestClient

def _client(monkeypatch):
    monkeypatch.setenv("INTERNAL_API_KEY", "SECRET")
    from app import auth as _a
    importlib.reload(_a)
    from app.main import app
    return TestClient(app)

_ARTICLE = {"articleNo": "N100", "tradeTypeName": "전세", "dealOrWarrantPrc": "2억 9,000",
            "area2": "42.5", "floorInfo": "5/15", "articleName": "OO아파트",
            "buildingName": "OO", "cortarNo": "1159010100"}

def test_ingest_auth(monkeypatch):
    c = _client(monkeypatch)
    assert c.post("/api/internal/realestate/listings-ingest", json={"batches": []}).status_code == 401

def test_ingest_upserts_and_matches(monkeypatch):
    c = _client(monkeypatch)
    with patch("app.internal_router.notify_new_listings", return_value={"sent": 1, "sent_ids": [1]}) as noti:
        r = c.post("/api/internal/realestate/listings-ingest",
                   headers={"X-Internal-Key": "SECRET"},
                   json={"fetched_at": "2026-07-09T00:00:00Z",
                         "batches": [{"dong": "신대방동", "articles": [_ARTICLE]}]})
    assert r.status_code == 200
    body = r.json()
    assert body["received"] == 1 and body["new"] == 1 and body["matched"] == 1
    assert noti.call_count == 1
    # 저장 확인
    from app import db
    ls = db.get_listings()
    assert any(l["article_no"] == "N100" and l["dong"] == "신대방동" for l in ls)

def test_ingest_idempotent(monkeypatch):
    c = _client(monkeypatch)
    payload = {"batches": [{"dong": "신대방동", "articles": [_ARTICLE]}]}
    with patch("app.internal_router.notify_new_listings", return_value={"sent": 0}):
        c.post("/api/internal/realestate/listings-ingest", headers={"X-Internal-Key": "SECRET"}, json=payload)
        r2 = c.post("/api/internal/realestate/listings-ingest", headers={"X-Internal-Key": "SECRET"}, json=payload)
    assert r2.json()["new"] == 0   # 재push → 신규 0
```

- [ ] **Step 2: 실패 확인** — `PYTHONPATH=... python -m pytest tests/test_internal_ingest.py -q` → FAIL(404)

- [ ] **Step 3: 구현** — `app/internal_router.py`에 추가

상단 import 확장:
```python
from pydantic import BaseModel
from typing import List, Dict, Any
from .db import get_listing_criteria, upsert_listing
from .listing_collector import _parse_naver_article
from .listing_matcher import run_listing_matching
from .notifier import notify_new_listings
```
(기존 `from .db import get_listing_criteria`는 위 확장 라인으로 합치거나 중복 없이 정리)

모델 + 핸들러:
```python
class _Batch(BaseModel):
    dong: str
    articles: List[Dict[str, Any]] = []


class ListingsIngest(BaseModel):
    fetched_at: str | None = None
    batches: List[_Batch] = []


@router.post("/api/internal/realestate/listings-ingest",
             dependencies=[Depends(verify_internal_key)])
def listings_ingest(body: ListingsIngest):
    received = new = 0
    for batch in body.batches:
        for raw in batch.articles:
            try:
                d = _parse_naver_article(raw)
                if not d.get("article_no"):
                    continue
                d["dong"] = batch.dong
                _, is_new = upsert_listing(d)
                received += 1
                if is_new:
                    new += 1
            except Exception as e:
                logger.warning("네이버 ingest 파싱 실패: %s", e)
    matched = 0
    if received:
        run_listing_matching()
        noti = notify_new_listings()
        matched = int(noti.get("sent", 0)) if isinstance(noti, dict) else 0
    return {"received": received, "new": new, "matched": matched}
```

- [ ] **Step 4: 통과 확인** — 같은 pytest → PASS
- [ ] **Step 5: 커밋**

```bash
git add realestate-lab/app/internal_router.py realestate-lab/tests/test_internal_ingest.py
git commit -m "feat(realestate): naver 워커 listings-ingest(파싱→upsert→매칭→알림)"
```

---

### Task 4: node_monitor fetcher 등재

**Files:**
- Modify: `agent-office/app/node_monitor.py`
- Test: `agent-office/tests/test_node_monitor.py` (기존 파일에 append; 없으면 `test_node_monitor_fetcher.py` 생성)

**Interfaces:**
- Produces: `WORKER_REGISTRY`에 `{"name":"naver-fetch","kind":"fetcher","queue":None}`; `collect_status()`가 fetcher에 대해 `{from:"naver-fetch", to:"nas-realestate", type:"http-pull", status:...}` link 생성.

- [ ] **Step 1: 실패 테스트** — 등재 여부(트리비얼) + 링크 분기

```python
def test_naver_fetch_registered():
    from app.node_monitor import WORKER_REGISTRY
    entry = next((w for w in WORKER_REGISTRY if w["name"] == "naver-fetch"), None)
    assert entry is not None
    assert entry["kind"] == "fetcher" and entry["queue"] is None
```
> collect_status의 fetcher-link는 기존 `test_node_monitor.py`의 redis-mock 픽스처(trader/render 링크 검증부)를 그대로 따라, "naver-fetch heartbeat 존재 시 links에 from=naver-fetch·type=http-pull이 있다"를 추가 검증하라. 픽스처가 없으면 이 트리비얼 등재 테스트만으로 충분.

- [ ] **Step 2: 실패 확인** — `cd agent-office && python -m pytest tests/test_node_monitor.py -q` (또는 신규 파일) → FAIL

- [ ] **Step 3: 구현** — `agent-office/app/node_monitor.py`

`WORKER_REGISTRY` 리스트에 추가:
```python
    {"name": "naver-fetch",  "kind": "fetcher", "queue": None},
```

`collect_status()`의 kind 분기(`if w["kind"] == "trader": ... elif w["kind"] == "render": ...`)에 추가:
```python
    elif w["kind"] == "fetcher":
        out["links"].append({"from": w["name"], "to": "nas-realestate", "type": "http-pull",
                             "status": "healthy" if w["alive"] else "down"})
```

- [ ] **Step 4: 통과 확인** — 같은 pytest → PASS + 회귀 `cd agent-office && python -m pytest -q`
- [ ] **Step 5: 커밋**

```bash
git add agent-office/app/node_monitor.py agent-office/tests/test_node_monitor.py
git commit -m "feat(agent-office): node_monitor에 naver-fetch(fetcher) 등재 + http-pull 링크"
```

---

### Task 5: collect_listings MOLIT 전용화 (naver deprecate)

**Files:**
- Modify: `realestate-lab/app/listing_collector.py`
- Test: `realestate-lab/tests/test_listing_collector.py` (기존 수정)

**Interfaces:**
- `collect_listings()` → `{"new_count":0,"total_count":0,"naver_ok":None,"diag":str}` (naver는 워커가 담당하므로 NAS는 MOLIT만). `_fetch_naver_all`/`_parse_naver_article`/`_won_to_manwon`는 **존치**(ingest가 `_parse_naver_article` 사용).

- [ ] **Step 1: 실패 테스트** — 기존 `test_collect_naver_block_sets_naver_ok_false`를 대체

```python
def test_collect_is_molit_only(monkeypatch):
    """collect_listings는 MOLIT만 수행(naver fetch 미호출). naver는 워커 담당."""
    from app import listing_collector as lc
    monkeypatch.setattr(lc, "_fetch_molit_all", lambda: (5, "calls=1 saved=5"))
    called = {"naver": False}
    def naver_spy():
        called["naver"] = True
        return (0, 0)
    monkeypatch.setattr(lc, "_fetch_naver_all", naver_spy)
    r = lc.collect_listings()
    assert called["naver"] is False          # naver 미호출
    assert "molit" in r["diag"]
```

- [ ] **Step 2: 실패 확인** — `PYTHONPATH=... python -m pytest tests/test_listing_collector.py -q` → FAIL(naver 호출됨)

- [ ] **Step 3: 구현** — `collect_listings` 교체

```python
def collect_listings() -> dict:
    """MOLIT 실거래(합법 baseline)만 수집. 네이버 호가는 naver-fetch 워커가
    /api/internal/realestate/listings-ingest로 push (spec 2026-07-09-naver-listing-worker)."""
    molit_diag = None
    try:
        _, molit_diag = _fetch_molit_all()
    except Exception as e:
        molit_diag = f"exc={type(e).__name__}:{str(e)[:80]}"
        logger.error("MOLIT 수집 오류(안전마진 baseline): %s", e)
    diag = f"molit[{molit_diag}] naver[worker]"
    save_listing_collect_log(0, 0, None, error=diag)
    return {"new_count": 0, "total_count": 0, "naver_ok": None, "diag": diag}
```
> `save_listing_collect_log`의 naver_ok 인자는 `1 if naver_ok else 0` 처리이므로 None→0 저장(무해). `_fetch_naver_all`/`_parse_naver_article`는 삭제하지 말 것(ingest 재사용).
> 기존 `test_collect_writes_molit_diag_to_log`(계측 테스트)가 `_fetch_naver_all`을 `lambda: (0,0)`로 mock하는데 이제 미호출이므로 여전히 통과(무해). `test_molit_call_returns_diag_on_http_error`·`_parse_*`도 무관.

- [ ] **Step 4: 통과 확인** — `PYTHONPATH=... python -m pytest tests/test_listing_collector.py -q` → PASS (신규 + 기존 계측/파서 테스트)
- [ ] **Step 5: 커밋**

```bash
git add realestate-lab/app/listing_collector.py realestate-lab/tests/test_listing_collector.py
git commit -m "refactor(realestate): collect_listings MOLIT 전용화(네이버는 워커 ingest로 이전)"
```

---

### Task 6: 회귀 + 카탈로그/메모리 + 배포 + 계약 핸드오프

**Files:** `CLAUDE.md`, 메모리 `service_realestate.md`/`infra_distributed_workers.md`

- [ ] **Step 1: 전체 회귀** — `cd realestate-lab && PYTHONPATH="C:/Users/jaeoh/Desktop/workspace/web-backend" python -m pytest -q` (0 fail) + `cd agent-office && python -m pytest -q` (0 fail)
- [ ] **Step 2: CLAUDE.md** — §5 nginx 표에 `/api/internal/realestate/`(IP화이트리스트) 추가, §9 realestate 표에 `GET /api/internal/realestate/targets`·`POST /listings-ingest` 추가
- [ ] **Step 3: 메모리** — `service_realestate.md`(내부 계약·NAVER_CORTAR·collect MOLIT전용화), `infra_distributed_workers.md`(naver-fetch fetcher kind 등재, /api/internal/realestate/ 계약)
- [ ] **Step 4: 계약 잠금·핸드오프** — co-gahusb `acquire_lock("nginx-conf","BE")`+`acquire_lock("nas-deploy","BE")` → 커밋 push(자동배포). AI(web-ai)에 §4 계약(targets/ingest 스키마·heartbeat kind=fetcher·INTERNAL_API_KEY)·병렬 spike 요청, FE(web-ui)에 /infra fetcher 노드 요청 `post_message`. 배포 후 `GET /api/internal/realestate/targets`(X-Internal-Key) 200 검증 → 락 해제.

```bash
git add CLAUDE.md
git commit -m "docs(CLAUDE.md): naver 워커 internal 엔드포인트 + nginx 등재"
git push origin main
```

---

## Self-Review

**Spec coverage:** §4.1 targets→T2, §4.2 ingest→T3, §4.3 cortarNo→T1, §4.4 heartbeat(계약, 워커측)→T4 등재로 관측, §5 관측→T4, §6 인증(nginx+X-Internal-Key)→T2, §7 멱등/에러→T3(try/except·article_no UNIQUE), §8 테스트→각 Task, §3 _fetch_naver_all deprecate→T5, §10 핸드오프→T6. ✓
- 스펙 대비 조정: env명 `INTERNAL_API_KEY`(기존 관례, 스펙의 REALESTATE_INTERNAL_KEY 대체) — Global Constraints에 명시. matched=notify sent 수로 구체화(스펙 {received,new,matched} 유지).

**Placeholder scan:** 각 Step 실제 코드/명령. T4의 collect_status fetcher-link 상세 테스트만 "기존 test_node_monitor 픽스처 따라"로 위임(기존 파일 참조, 트리비얼 등재 테스트는 완전). ✓

**Type consistency:** `naver_cortar`(T1)→internal_router(T2) 사용 일치. `upsert_listing`→`(dict,is_new)`(기존)·`_parse_naver_article`(기존 반환 dict)·`notify_new_listings`→`{sent,...}`(기존)→ingest 소비 일치. targets 반환 `{dongs:[{dong,cortar_no}],deal_types,page_limit}`=스펙 §4.1. ingest 반환 `{received,new,matched}`=스펙 §4.2. WORKER_REGISTRY 항목 형식(T4)=기존 구조 일치. ✓

**범위:** BE 단일 세션(realestate T1-3,5 + agent-office T4 + 통합 T6). 워커(web-ai)·FE(web-ui)는 별도. 단일 실행 계획 적정(6 tasks).