dbx-main/app/integrations/cafe24/products.py

"""카페24 상품 엔드포인트 래퍼.

전송/재시도/인증은 Cafe24Client 가 담당하고, 여기서는 경로와 payload 모양만
안다. 향후 주문관리는 같은 클라이언트로 `orders.py` 를 추가하면 된다.

상세설명은 **별도 리소스가 아니다.** 실제 쇼핑몰(miraskitchen)에 확인한 결과
`/admin/products/{no}/description` 은 존재하지 않는다(`No API found.`).
상세설명은 상품 리소스의 필드로 읽고 쓴다.

    GET  /admin/products/{no}   → description · mobile_description ·
                                  separated_mobile_description
    PUT  /admin/products/{no}   → {"request": {"description": ...}}

목록 API(`/admin/products`) 응답에는 description 이 **없다**. 그래서 상세설명은
상품 1건씩 조회해야 한다(목록 화면에서 미리보기를 뿌리지 않는 이유).
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

from .client import Cafe24Client

# 카페24 상품 목록 API 의 1회 최대 조회 수
PAGE_LIMIT = 100


def _flag(value: Any, *, default: bool = True) -> bool:
    """카페24는 boolean 을 'T'/'F' 문자열로 준다."""
    if isinstance(value, bool):
        return value
    text = str(value or "").strip().upper()
    if text in ("T", "TRUE", "Y", "1"):
        return True
    if text in ("F", "FALSE", "N", "0"):
        return False
    return default


def count_products(client: Cafe24Client, *, product_name: str = "") -> int:
    params: dict[str, Any] = {}
    if product_name:
        params["product_name"] = product_name
    payload = client.get("/admin/products/count", params=params)
    try:
        return int(payload.get("count") or 0)
    except (TypeError, ValueError):
        return 0


def list_products(
    client: Cafe24Client,
    *,
    limit: int = PAGE_LIMIT,
    offset: int = 0,
    product_name: str = "",
    product_no: int | None = None,
) -> list[dict[str, Any]]:
    """상품 목록 1페이지. 검색어가 있으면 상품명 부분일치로 조회한다."""
    params: dict[str, Any] = {
        "limit": max(1, min(int(limit), PAGE_LIMIT)),
        "offset": max(0, int(offset)),
    }
    if product_name:
        params["product_name"] = product_name
    if product_no:
        params["product_no"] = int(product_no)
    payload = client.get("/admin/products", params=params)
    products = payload.get("products")
    return products if isinstance(products, list) else []


def list_all_products(
    client: Cafe24Client,
    *,
    product_name: str = "",
    max_items: int = 1000,
) -> tuple[list[dict[str, Any]], bool]:
    """전체 상품을 페이지를 넘겨가며 모두 가져온다.

    2분할 화면의 왼쪽 목록은 페이지 없이 한 번에 보여주고 필터·정렬을 브라우저에서
    처리한다. 그래야 "진열중만" 같은 필터가 전체 기준으로 정확해진다
    (한 페이지만 받아 걸러내면 다음 페이지의 해당 상품이 빠진다).

    반환: (상품 목록, 상한에 걸려 잘렸는지)
    상품이 max_items 를 넘으면 거기서 멈춘다 — 무한 호출로 API 제한에 걸리는
    것을 막기 위한 안전장치다(현재 쇼핑몰 87개, 1회 100개 조회).
    """
    collected: list[dict[str, Any]] = []
    while len(collected) < max_items:
        want = min(PAGE_LIMIT, max_items - len(collected))
        batch = list_products(
            client,
            limit=want,
            offset=len(collected),
            product_name=product_name,
        )
        collected.extend(batch)
        if len(batch) < want:
            return collected, False  # 요청한 만큼 못 받았다 = 마지막 페이지
        if len(collected) >= max_items:
            return collected, True  # 상한에서 멈췄다 — 뒤에 더 있을 수 있다
    return collected, False


def get_product(client: Cafe24Client, product_no: int) -> dict[str, Any]:
    """상품 1건 상세. 이 응답에 상세설명 필드까지 들어 있다."""
    no = int(product_no)
    payload = client.get(f"/admin/products/{no}", product_no=no)
    product = payload.get("product")
    return product if isinstance(product, dict) else {}


@dataclass(frozen=True)
class Descriptions:
    """상품 1건의 상세설명 묶음. 카페24가 언제나 source of truth 다."""

    product_no: int
    product_name: str
    description: str
    mobile_description: str
    # separated_mobile_description = 'T' 면 PC/모바일 상세설명을 따로 쓴다.
    # 'F' 면 모바일도 PC 값을 쓰므로 수정 시 두 필드를 함께 맞춰야 한다.
    separated_mobile: bool

    @property
    def mobile_differs(self) -> bool:
        return self.mobile_description != self.description


def descriptions_from_product(raw: dict[str, Any]) -> Descriptions:
    """`get_product` 응답 dict → Descriptions."""
    try:
        product_no = int(raw.get("product_no") or 0)
    except (TypeError, ValueError):
        product_no = 0
    return Descriptions(
        product_no=product_no,
        product_name=str(raw.get("product_name") or ""),
        description=str(raw.get("description") or ""),
        mobile_description=str(raw.get("mobile_description") or ""),
        separated_mobile=_flag(raw.get("separated_mobile_description"), default=False),
    )


def fetch_descriptions(client: Cafe24Client, product_no: int) -> Descriptions:
    """상품의 현재 상세설명. 로컬 DB 의 마지막 버전을 현재값으로 가정하지 않는다."""
    return descriptions_from_product(get_product(client, product_no))


def _flag_value(flag: bool) -> str:
    """카페24는 boolean 을 'T'/'F' 문자열로 받는다."""
    return "T" if flag else "F"


def build_update_payload(
    *,
    description: str | None = None,
    mobile_description: str | None = None,
    product_name: str | None = None,
    display: bool | None = None,
    selling: bool | None = None,
    shop_no: int | None = None,
) -> dict[str, Any]:
    """상품 수정 PUT body. 준 필드만 바뀌고 나머지는 유지된다(부분 수정).

    `None` 인 항목은 payload 에 넣지 않는다 = 그 필드를 건드리지 않는다.
    예약에서 "진열만 켜기"처럼 상세설명 없이 상태만 바꾸는 경우가 있으므로
    description 도 생략할 수 있다.
    """
    request: dict[str, Any] = {}
    if description is not None:
        request["description"] = description
    if product_name is not None:
        request["product_name"] = product_name
    if mobile_description is not None:
        request["mobile_description"] = mobile_description
    if display is not None:
        request["display"] = _flag_value(display)
    if selling is not None:
        request["selling"] = _flag_value(selling)
    payload: dict[str, Any] = {"request": request}
    if shop_no:
        payload["shop_no"] = int(shop_no)
    return payload


def update_product(
    client: Cafe24Client,
    product_no: int,
    *,
    description: str | None = None,
    mobile_description: str | None = None,
    product_name: str | None = None,
    display: bool | None = None,
    selling: bool | None = None,
    shop_no: int | None = None,
) -> dict[str, Any]:
    """상품 부분 수정. 상세설명·상품명·진열·판매를 한 번의 호출로 바꿀 수 있다.

    바꿀 것이 하나도 없으면 호출하지 않고 빈 dict 를 돌려준다.

    ⚠️ 상세설명을 바꿀 때는 쓰기 직전 카페24 현재 HTML 을 다시 읽어 BACKUP
       revision 을 남길 것(`docs/CAFE24_MODULE.md` 규칙). 이 함수는 백업하지 않는다.
    """
    payload = build_update_payload(
        description=description,
        mobile_description=mobile_description,
        product_name=product_name,
        display=display,
        selling=selling,
        shop_no=shop_no,
    )
    if not payload["request"]:
        return {}
    no = int(product_no)
    response = client.put(f"/admin/products/{no}", json=payload, product_no=no)
    product = response.get("product")
    return product if isinstance(product, dict) else response


def update_descriptions(
    client: Cafe24Client,
    product_no: int,
    *,
    description: str,
    mobile_description: str | None = None,
    shop_no: int | None = None,
) -> dict[str, Any]:
    """상세설명만 교체하는 지름길. 실제 전송은 `update_product` 가 한다."""
    return update_product(
        client,
        product_no,
        description=description,
        mobile_description=mobile_description,
        shop_no=shop_no,
    )


def normalize_product(raw: dict[str, Any]) -> dict[str, Any]:
    """카페24 상품 dict → 캐시 테이블(cafe24_products) 컬럼 모양으로 정규화."""
    try:
        product_no = int(raw.get("product_no") or 0)
    except (TypeError, ValueError):
        product_no = 0

    return {
        "product_no": product_no,
        "product_code": str(raw.get("product_code") or ""),
        "product_name": str(raw.get("product_name") or ""),
        "display": _flag(raw.get("display")),
        "selling": _flag(raw.get("selling")),
    }