diff --git a/realestate-lab/app/internal_router.py b/realestate-lab/app/internal_router.py
index bb363db..95c68e6 100644
--- a/realestate-lab/app/internal_router.py
+++ b/realestate-lab/app/internal_router.py
@@ -1,11 +1,16 @@
 """naver-fetch 워커 내부 계약: targets 조회 + listings ingest."""
 import os
 import logging
+from typing import List, Dict, Any
 from fastapi import APIRouter, Depends
+from pydantic import BaseModel
 
 from .auth import verify_internal_key
-from .db import get_listing_criteria
+from .db import get_listing_criteria, upsert_listing
 from .lawd_codes import naver_cortar
+from .listing_collector import _parse_naver_article
+from .listing_matcher import run_listing_matching
+from .notifier import notify_new_listings
 
 logger = logging.getLogger("realestate-lab")
 router = APIRouter()
@@ -25,3 +30,38 @@ def listing_targets():
         dongs.append({"dong": d, "cortar_no": code})
     return {"dongs": dongs, "deal_types": crit.get("deal_types", []),
             "page_limit": NAVER_PAGE_LIMIT}
+
+
+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}
diff --git a/realestate-lab/tests/test_internal_ingest.py b/realestate-lab/tests/test_internal_ingest.py
new file mode 100644
index 0000000..31aa9ea
--- /dev/null
+++ b/realestate-lab/tests/test_internal_ingest.py
@@ -0,0 +1,46 @@
+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)
+    # 헤더 없음 — FastAPI Header(...)는 dependency 실행 전 자체 검증하므로 422.
+    # test_internal_targets.py의 동일 패턴과 일치(같은 서비스 선례).
+    assert c.post("/api/internal/realestate/listings-ingest", json={"batches": []}).status_code in (401, 422)
+    assert c.post("/api/internal/realestate/listings-ingest",
+                  headers={"X-Internal-Key": "WRONG"}, 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