web-page-backend/realestate-lab/docs/superpowers/plans/2026-07-09-naver-listing-worker-be.md

네이버 매물 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

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 하단에 추가

# 네이버부동산 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: 커밋

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

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:

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

"""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 (...) 아래)

from .internal_router import router as internal_router

그리고 app = FastAPI(lifespan=lifespan) + install_access_log(app) 직후에:

app.include_router(internal_router)

nginx/default.conf: /api/realestate/ location 블록 앞에 추가(더 구체적 prefix 우선):

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: 커밋

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

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 확장:

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는 위 확장 라인으로 합치거나 중복 없이 정리)

모델 + 핸들러:

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: 커밋

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: 실패 테스트 — 등재 여부(트리비얼) + 링크 분기

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 리스트에 추가:

    {"name": "naver-fetch",  "kind": "fetcher", "queue": None},

collect_status()의 kind 분기(if w["kind"] == "trader": ... elif w["kind"] == "render": ...)에 추가:

    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: 커밋

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를 대체

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 교체

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: 커밋

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 검증 → 락 해제.

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).