o2o-negosium-original/lps/services/naver_hub/shopping_insight.py

"""쇼핑 인사이트 — 분야별 트렌드 조회 (POST /shopping/v1/categories).

네이버 데이터랩 쇼핑인사이트의 **분야별 검색 클릭 추이**를 조회한다.
반환값 ratio 는 절대 클릭수가 아니라 **구간 내 최대값을 100 으로 둔 상대 지표**다
(같은 응답 안에서만 비교 가능 — 다른 조회 결과와 절대 비교하면 안 된다).
⇒ 상품명·가격·판매처가 없으므로 최저가 검색 소스로는 쓸 수 없다. 수요 추이 분석용.

요청 제약은 전부 호출 전에 검증한다 — 게이트웨이 왕복 없이 바로 틀린 곳을 알려주는 게 낫다.
"""

from datetime import date
from typing import Optional, Sequence

from pydantic import BaseModel, Field

from common.logger import LOG
from services.naver_hub.client import NaverApiHubClient

_PATH = "/shopping/v1/categories"
_MAX_CATEGORIES = 3                 # 문서: 최대 3개 쌍
_MIN_START = date(2017, 8, 1)       # 문서: 2017년 8월 1일부터 조회 가능
_TIME_UNITS = ("date", "week", "month")
_DEVICES = ("pc", "mo")
_GENDERS = ("m", "f")
_AGES = ("10", "20", "30", "40", "50", "60")


class ShoppingCategory(BaseModel):
    """조회할 쇼핑 분야. param 은 네이버쇼핑 카테고리 URL 의 cat_id 값."""

    name: str = Field(description="쇼핑 분야 이름(응답 title 로 되돌아옴)")
    param: list[str] = Field(description="쇼핑 분야 코드 목록(cat_id)")


class InsightPoint(BaseModel):
    period: str = Field(description="구간 시작 날짜(yyyy-mm-dd)")
    ratio: float = Field(description="구간별 클릭량의 상대 비율 — 결과 내 최대값이 100")


class InsightSeries(BaseModel):
    title: str = Field(description="쇼핑 분야 이름")
    category: list[str] = Field(default_factory=list, description="쇼핑 분야 코드")
    data: list[InsightPoint] = Field(default_factory=list, description="구간별 추이")


class ShoppingInsightResult(BaseModel):
    start_date: str
    end_date: str
    time_unit: str
    results: list[InsightSeries] = Field(default_factory=list)


class ShoppingInsightClient:
    """쇼핑 인사이트 조회기. 허브 공통 클라이언트를 감싼다."""

    def __init__(self, client: Optional[NaverApiHubClient] = None):
        self._client = client or NaverApiHubClient()

    @property
    def enabled(self) -> bool:
        return self._client.enabled

    async def categories(
        self,
        *,
        start_date: str,
        end_date: str,
        categories: Sequence[ShoppingCategory],
        time_unit: str = "date",
        device: Optional[str] = None,
        gender: Optional[str] = None,
        ages: Optional[Sequence[str]] = None,
    ) -> ShoppingInsightResult:
        """분야별 트렌드 조회. 검증 실패는 ValueError, 호출 실패는 NaverApiHubError."""
        body = build_categories_body(
            start_date=start_date, end_date=end_date, categories=categories,
            time_unit=time_unit, device=device, gender=gender, ages=ages,
        )
        data = await self._client.post(_PATH, body)
        result = parse_categories_response(data)
        LOG.d(f"[naver-hub] 쇼핑인사이트 {start_date}~{end_date} {time_unit} → {len(result.results)}개 분야")
        return result


def build_categories_body(
    *,
    start_date: str,
    end_date: str,
    categories: Sequence[ShoppingCategory],
    time_unit: str = "date",
    device: Optional[str] = None,
    gender: Optional[str] = None,
    ages: Optional[Sequence[str]] = None,
) -> dict:
    """요청 바디 생성 + 문서상의 제약 검증(순수 함수 — 네트워크 없이 테스트 가능)."""
    start = _parse_date(start_date, "startDate")
    end = _parse_date(end_date, "endDate")
    if start < _MIN_START:
        raise ValueError(f"startDate 는 {_MIN_START.isoformat()} 이후여야 합니다(문서 제약): {start_date}")
    if start > end:
        raise ValueError(f"startDate 가 endDate 보다 늦습니다: {start_date} > {end_date}")

    if time_unit not in _TIME_UNITS:
        raise ValueError(f"timeUnit 은 {_TIME_UNITS} 중 하나여야 합니다: {time_unit!r}")

    if not categories:
        raise ValueError("category 는 최소 1개가 필요합니다")
    if len(categories) > _MAX_CATEGORIES:
        raise ValueError(f"category 는 최대 {_MAX_CATEGORIES}개입니다(요청 {len(categories)}개)")
    for c in categories:
        if not c.name or not c.param:
            raise ValueError(f"category 의 name/param 이 비었습니다: {c!r}")

    body: dict = {
        "startDate": start_date,
        "endDate": end_date,
        "timeUnit": time_unit,
        "category": [{"name": c.name, "param": list(c.param)} for c in categories],
    }

    # 선택 파라미터는 값이 있을 때만 싣는다 — 빈 값을 보내면 게이트웨이가 검증 오류로 되돌린다.
    if device is not None:
        if device not in _DEVICES:
            raise ValueError(f"device 는 {_DEVICES} 중 하나여야 합니다: {device!r}")
        body["device"] = device
    if gender is not None:
        if gender not in _GENDERS:
            raise ValueError(f"gender 는 {_GENDERS} 중 하나여야 합니다: {gender!r}")
        body["gender"] = gender
    if ages:
        bad = [a for a in ages if a not in _AGES]
        if bad:
            raise ValueError(f"ages 는 {_AGES} 중에서 골라야 합니다: {bad}")
        body["ages"] = list(ages)
    return body


def parse_categories_response(data: dict) -> ShoppingInsightResult:
    """응답 → 타입 있는 결과. 응답 키가 카멜케이스라 여기서 한 번만 변환한다."""
    return ShoppingInsightResult(
        start_date=data.get("startDate", ""),
        end_date=data.get("endDate", ""),
        time_unit=data.get("timeUnit", ""),
        results=[
            InsightSeries(
                title=r.get("title", ""),
                category=list(r.get("category") or []),
                data=[InsightPoint(period=p.get("period", ""), ratio=float(p.get("ratio", 0)))
                      for p in (r.get("data") or [])],
            )
            for r in (data.get("results") or [])
        ],
    )


def _parse_date(value: str, field: str) -> date:
    try:
        return date.fromisoformat(value)
    except (TypeError, ValueError):
        raise ValueError(f"{field} 형식은 yyyy-mm-dd 여야 합니다: {value!r}") from None