005ebc3d76
네이버가 2026-07-31 검색 오픈API 중 쇼핑·책·전문자료를 종료(유예·대체 없음)해 shop.json 이 404 SE05 를 반환한다. 후속 플랫폼인 NCP NAVER API HUB 를 붙인다. - NaverApiHubConfig: 게이트웨이 base_url + NCP Client ID/Secret(둘 다 차야 enabled) - services/naver_hub/client.py: X-NCP-APIGW-API-KEY-ID/KEY 인증, 오류 바디 3형식(게이트웨이/Search/인사이트)을 NaverApiHubError 로 정규화(auth_failed·retryable) - services/naver_hub/shopping_insight.py: POST /shopping/v1/categories. 문서 제약(기간 2017-08-01~, 분야 최대 3개, timeUnit·device·gender·ages)을 호출 전에 검증하고 카멜케이스 응답을 타입으로 변환 - tests: MockTransport 로 경로·헤더·오류형식 계약 검증 17건 + LPS_LIVE 스모크 주의: 허브에도 쇼핑 '검색'(상품명·가격·판매처)은 없다. 인사이트의 ratio 는 구간 내 최대값 100 기준 상대지표라 최저가 파이프라인 소스로는 쓸 수 없다. 기존 services/search/naver 어댑터는 손대지 않았다(사문화 상태 유지).
112 lines
4.6 KiB
Python
112 lines
4.6 KiB
Python
"""NAVER API HUB 공통 호출 계층 — 인증 헤더·요청·오류 해석을 한곳에 모은다.
|
|
|
|
옛 오픈API(openapi.naver.com) 와 달라진 점:
|
|
엔드포인트 https://naverapihub.apigw.ntruss.com
|
|
인증 헤더 X-NCP-APIGW-API-KEY-ID / X-NCP-APIGW-API-KEY (NCP 콘솔 발급 Client ID/Secret)
|
|
경로 규칙 /search/v1/{type} · /shopping/v1/{...} (옛 /v1/search/{type}.json 과 순서가 반대)
|
|
|
|
오류 바디가 계층마다 다르다(공식 문서 'NAVER API HUB 개요' 기준). 호출부가 세 형식을 다 알
|
|
필요는 없으므로 _parse_error 가 하나의 NaverApiHubError 로 정규화한다:
|
|
1) 게이트웨이(인증 실패·라우팅 실패) {"error": {"errorCode", "message", "details"}}
|
|
2) Search API 파라미터 검증 {"errorCode", "errorMessage"}
|
|
3) 검색어 트렌드·쇼핑 인사이트 검증 {"errMsg", "errId"}
|
|
"""
|
|
|
|
import json
|
|
from typing import Any, Optional
|
|
|
|
import httpx
|
|
|
|
from common.logger import LOG
|
|
from config.config_models import NaverApiHubConfig
|
|
from config.server_configs import naver_api_hub_config
|
|
|
|
|
|
class NaverApiHubError(Exception):
|
|
"""허브 호출 실패. status/code 로 재시도 가치를 구분한다.
|
|
|
|
auth_failed(401·403) 와 not_found(404) 는 재시도해도 그대로다 — 키 설정이나 경로가 틀린 것.
|
|
쿼터 소진(429)·5xx 만 재시도 가치가 있다(retryable).
|
|
"""
|
|
|
|
def __init__(self, message: str, *, status: int, code: Optional[str] = None, path: str = ""):
|
|
super().__init__(message)
|
|
self.status = status
|
|
self.code = code
|
|
self.path = path
|
|
|
|
@property
|
|
def auth_failed(self) -> bool:
|
|
return self.status in (401, 403)
|
|
|
|
@property
|
|
def not_found(self) -> bool:
|
|
return self.status == 404
|
|
|
|
@property
|
|
def retryable(self) -> bool:
|
|
return self.status == 429 or self.status >= 500
|
|
|
|
|
|
class NaverApiHubClient:
|
|
"""허브 공통 클라이언트. 개별 API(쇼핑 인사이트 등)는 이 위에 얹는다.
|
|
|
|
transport 는 테스트에서 httpx.MockTransport 를 꽂기 위한 주입점이다(네트워크 없이 계약 검증).
|
|
"""
|
|
|
|
def __init__(self, cfg: Optional[NaverApiHubConfig] = None, *, transport: Any = None):
|
|
self._cfg = cfg if cfg is not None else naver_api_hub_config
|
|
self._transport = transport
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return self._cfg.enabled
|
|
|
|
def _headers(self) -> dict:
|
|
return {
|
|
"X-NCP-APIGW-API-KEY-ID": self._cfg.client_id,
|
|
"X-NCP-APIGW-API-KEY": self._cfg.client_secret,
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
async def post(self, path: str, body: dict) -> dict:
|
|
"""POST 요청 1건. 성공이면 파싱된 JSON, 실패면 NaverApiHubError."""
|
|
if not self.enabled:
|
|
raise NaverApiHubError(
|
|
"NAVER API HUB 미설정 — config.local.toml [NaverApiHubConfig] 의 client_id/client_secret 을 채우세요",
|
|
status=0, path=path,
|
|
)
|
|
|
|
url = f"{self._cfg.base_url.rstrip('/')}{path}"
|
|
async with httpx.AsyncClient(timeout=self._cfg.timeout_sec, transport=self._transport) as client:
|
|
r = await client.post(url, headers=self._headers(), json=body)
|
|
|
|
if r.status_code != 200:
|
|
raise self._parse_error(r, path)
|
|
return r.json()
|
|
|
|
@staticmethod
|
|
def _parse_error(r: httpx.Response, path: str) -> NaverApiHubError:
|
|
"""세 가지 오류 바디 형식을 하나로 정규화. JSON 이 아니면 본문 앞부분을 그대로 싣는다."""
|
|
code = None
|
|
message = r.text[:200]
|
|
try:
|
|
body = r.json()
|
|
except (json.JSONDecodeError, ValueError):
|
|
body = None
|
|
|
|
if isinstance(body, dict):
|
|
if isinstance(body.get("error"), dict): # 1) 게이트웨이
|
|
err = body["error"]
|
|
code = str(err.get("errorCode") or "") or None
|
|
message = " ".join(str(v) for v in (err.get("message"), err.get("details")) if v)
|
|
elif "errorMessage" in body or "errorCode" in body: # 2) Search API
|
|
code = str(body.get("errorCode") or "") or None
|
|
message = str(body.get("errorMessage") or message)
|
|
elif "errMsg" in body or "errId" in body: # 3) 트렌드·인사이트
|
|
code = str(body.get("errId") or "") or None
|
|
message = str(body.get("errMsg") or message)
|
|
|
|
LOG.w(f"[naver-hub] {r.status_code} {path} code={code} {message}")
|
|
return NaverApiHubError(message, status=r.status_code, code=code, path=path)
|