king/dbx-main
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
85c7333e72c55c3cbae289b361298e4c4b6a0677
dbx-main/app/modules/cafe24/routes_product_info.py
T
king 85c7333e72 feat(cafe24): 품목 추가·삭제 + 옵션 재생성 
- 추가(「+ 행 추가」/불러오기의 새 이름): 저장 시 옵션값을 끝에 덧붙여
  PUT options(allow_append — 앞 위치는 그대로라 기존 이름이 밀리지 않음).
  카페24가 품목을 자동 생성하면 wait_for_variants 로 새 코드를 받아
  자체코드·추가금액·진열/판매를 이어서 PUT. 거부 시 오류 그대로 표시.
- 삭제(행 ✕): DELETE /variants/{code}. 지연 중 GET 에 남는 삭제 품목은
  스냅샷 {"_deleted": true} 로 걸러냄. 품목 없는 옵션값은 하단 안내로 분리.
- 옵션값 줄이기(개수 감소)는 서버가 거부 — 위치 짝맞춤으로 이름이 밀림.
- 옵션 재생성: 전체 삭제 → 현재 행으로 생성 → 코드/금액/썸네일 재반영
  (품목코드 새로 부여, 강한 확인창).
- 유닛 91 통과, 통합 하네스(덧붙이기·삭제·지연 필터) 통과, 헤드리스 렌더 확인.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-18 23:47:42 +09:00

541 lines
27 KiB
Python
Raw Blame History

"""카페24 상품 정보 패널 — 오른쪽 칸 JSON API.
화면(products.html 의 `_side.html` 조각)이 fetch 로 부른다. 전부 `def`(동기) 핸들러.
POST /products/{no}/basic 상품명 · 판매가 · 공급가 · 소비자가
POST /products/{no}/image 대표 이미지 교체 (multipart 파일)
GET /products/{no}/options 옵션 + 품목 조회 (읽기 지연 보정 포함)
POST /products/{no}/options 옵션 생성 (옵션 없는 상품 — 품목 자동 생성)
PUT /products/{no}/options 옵션명 · 옵션값 이름/이미지/표시방식 수정
DELETE /products/{no}/options 옵션 삭제 (품목도 함께 삭제 — 확인 후)
POST /products/{no}/options/image 옵션값 썸네일 업로드만 (경로 반환)
PUT /products/{no}/variants 품목 자체코드 · 추가금액 · 진열 · 판매
공통 규칙
- 카페24 호출은 `app.integrations.cafe24` 만 통한다(httpx 직접 호출 금지).
- 쓰기 전에 현재값을 읽되, **읽기 지연을 보정한 유효 현재값**과 비교해 바뀐
것만 보낸다(routes_products.load_product). 이걸 안 하면 "A→B 로 바꾼 직후
다시 B→A" 가 카페24의 예전 값(A)과 같다고 판단돼 무시된다.
- PUT 응답이 곧 현재값이다. 응답을 스냅샷(`save_write_snapshot`)에 남기고 화면도
응답으로 그린다 — 다시 GET 하지 않는다(GET 은 한동안 예전 값을 돌려준다).
- 모든 쓰기는 감사로그(`cafe24_audit_logs`)에 남긴다.
"""
from __future__ import annotations
import logging
from typing import Any
from fastapi import APIRouter, Body, File, HTTPException, Request, UploadFile
from app.integrations.cafe24 import Cafe24Error, build_cafe24_api, products
from . import store
from .common import read_lag_grace_minutes, require_store
from .routes_products import load_product, remember_write
logger = logging.getLogger("cafe24.product_info")
product_info_router = APIRouter()
# 업로드 허용 이미지 형식 (카페24 상품 이미지 규격)
_IMAGE_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}
def _http_from_cafe24(exc: Cafe24Error) -> HTTPException:
return HTTPException(status_code=502, detail=str(exc))
def _read_image(upload: UploadFile) -> bytes:
"""업로드 파일 검증 + 바이트. 형식/크기 오류는 400."""
content_type = (upload.content_type or "").lower()
if content_type not in _IMAGE_TYPES:
raise HTTPException(status_code=400, detail="JPG·PNG·GIF·WEBP 이미지만 올릴 수 있습니다.")
data = upload.file.read(products.IMAGE_MAX_BYTES + 1)
if not data:
raise HTTPException(status_code=400, detail="빈 파일입니다.")
if len(data) > products.IMAGE_MAX_BYTES:
raise HTTPException(status_code=400, detail="이미지는 10MB 를 넘을 수 없습니다.")
return data
def _scalar_view(product: dict[str, Any]) -> dict[str, Any]:
"""화면이 바로 쓰는 스칼라 값들(PUT 응답 또는 보정된 GET 에서)."""
snap = store.product_snapshot(product)
return {
"product_name": snap.get("product_name", ""),
"price": snap.get("price", ""),
"supply_price": snap.get("supply_price", ""),
"retail_price": snap.get("retail_price", ""),
"display": products._flag(snap.get("display")), # noqa: SLF001 — 같은 규칙 재사용
"selling": products._flag(snap.get("selling")), # noqa: SLF001
"detail_image": snap.get("detail_image", ""),
"list_image": snap.get("list_image", ""),
"tiny_image": snap.get("tiny_image", ""),
"small_image": snap.get("small_image", ""),
"updated_date": snap.get("updated_date", ""),
}
# ════════════════════════════════════════════════════════════
# 기본 정보 — 상품명 · 가격
# ════════════════════════════════════════════════════════════
@product_info_router.post("/products/{product_no}/basic")
def product_basic(
request: Request,
product_no: int,
payload: dict[str, Any] = Body(default_factory=dict),
) -> dict[str, Any]:
st, user = require_store(request)
actor = str(user.get("email") or "")
want: dict[str, str] = {}
if "product_name" in payload:
name = str(payload.get("product_name") or "").strip()
if not name:
raise HTTPException(status_code=400, detail="상품명을 입력하세요.")
if len(name) > store.NAME_MAX:
raise HTTPException(status_code=400, detail=f"상품명은 {store.NAME_MAX}자를 넘을 수 없습니다.")
want["product_name"] = name
try:
for key, label in (("price", "판매가"), ("supply_price", "공급가"), ("retail_price", "소비자가")):
if key in payload:
parsed = store.parse_price(payload.get(key), field=label)
if parsed is not None:
want[key] = parsed
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
if not want:
raise HTTPException(status_code=400, detail="바꿀 항목이 없습니다.")
api = build_cafe24_api(st)
try:
current, _ = load_product(st, api, product_no)
except Cafe24Error as exc:
st.log_audit(actor=actor, action="update_basic", product_no=product_no,
result="FAIL", detail=f"현재값 조회 실패: {exc}")
raise _http_from_cafe24(exc) from exc
if not current:
raise HTTPException(status_code=404, detail="카페24에서 상품을 찾지 못했습니다.")
changes: dict[str, str] = {}
before: dict[str, str] = {}
for key, value in want.items():
old = str(current.get(key) or "")
same = old == value if key == "product_name" else store.price_equal(old, value)
if not same:
changes[key] = value
before[key] = old
if not changes:
return {"ok": True, "changed": [], "product": _scalar_view(current)}
try:
updated = products.update_product(api.client, product_no, **changes)
except Cafe24Error as exc:
st.log_audit(actor=actor, action="update_basic", product_no=product_no,
result="FAIL", detail=f"{changes} 실패: {exc}")
logger.warning("카페24 상품 %s 기본정보 변경 실패: %s", product_no, exc)
raise _http_from_cafe24(exc) from exc
remember_write(st, product_no, updated if isinstance(updated, dict) else {})
merged = {**current, **(updated if isinstance(updated, dict) else {}), **changes}
detail = ", ".join(f"{k}: '{before[k]}' → '{v}'" for k, v in changes.items())
st.log_audit(actor=actor, action="update_basic", product_no=product_no,
result="SUCCESS", detail=detail)
logger.info("카페24 상품 %s 기본정보 변경 (%s): %s", product_no, actor, detail)
return {"ok": True, "changed": list(changes.keys()), "product": _scalar_view(merged)}
# ════════════════════════════════════════════════════════════
# 대표 이미지
# ════════════════════════════════════════════════════════════
@product_info_router.post("/products/{product_no}/image")
def product_image(
request: Request,
product_no: int,
file: UploadFile = File(...),
) -> dict[str, Any]:
"""대표 이미지 교체. 업로드(/products/images) → PUT detail_image(A 타입).
A(대표이미지등록) 타입이면 목록/작은목록/축소 이미지는 카페24가 리사이징한다.
이전 이미지 경로는 감사로그에 남긴다(되돌리려면 그 경로를 다시 넣는다).
"""
st, user = require_store(request)
actor = str(user.get("email") or "")
data = _read_image(file)
api = build_cafe24_api(st)
try:
current, _ = load_product(st, api, product_no)
except Cafe24Error as exc:
raise _http_from_cafe24(exc) from exc
before = str(current.get("detail_image") or "")
try:
path = products.upload_image_bytes(api.client, data)
if not path:
raise Cafe24Error("카페24가 업로드 경로를 돌려주지 않았습니다.")
updated = products.set_main_image(api.client, product_no, path)
except Cafe24Error as exc:
st.log_audit(actor=actor, action="set_main_image", product_no=product_no,
result="FAIL", detail=f"업로드/적용 실패: {exc}")
logger.warning("카페24 상품 %s 대표이미지 변경 실패: %s", product_no, exc)
raise _http_from_cafe24(exc) from exc
remember_write(st, product_no, updated if isinstance(updated, dict) else {})
merged = {**current, **(updated if isinstance(updated, dict) else {})}
if not merged.get("detail_image"):
merged["detail_image"] = path
st.log_audit(actor=actor, action="set_main_image", product_no=product_no,
result="SUCCESS", detail=f"'{before}' → '{merged.get('detail_image')}' (업로드 {path})")
logger.info("카페24 상품 %s 대표이미지 변경 (%s)", product_no, actor)
return {"ok": True, "product": _scalar_view(merged)}
# ════════════════════════════════════════════════════════════
# 옵션 · 품목
# ════════════════════════════════════════════════════════════
def _variant_view(v: dict[str, Any]) -> dict[str, Any]:
opts = v.get("options") if isinstance(v.get("options"), list) else []
return {
"variant_code": str(v.get("variant_code") or ""),
"options": [
{"name": str(o.get("name") or ""), "value": str(o.get("value") or "")}
for o in opts if isinstance(o, dict)
],
"custom_variant_code": str(v.get("custom_variant_code") or ""),
"additional_amount": str(v.get("additional_amount") or "0.00"),
"display": products._flag(v.get("display")), # noqa: SLF001
"selling": products._flag(v.get("selling")), # noqa: SLF001
"quantity": v.get("quantity"),
"use_inventory": products._flag(v.get("use_inventory"), default=False), # noqa: SLF001
"image": str(v.get("image") or ""),
}
def _options_view(option: dict[str, Any]) -> dict[str, Any]:
raw_options = option.get("options") if isinstance(option.get("options"), list) else []
out_options = []
for o in raw_options:
if not isinstance(o, dict):
continue
values = o.get("option_value") if isinstance(o.get("option_value"), list) else []
out_options.append(
{
"option_code": str(o.get("option_code") or ""),
"option_name": str(o.get("option_name") or ""),
"option_display_type": str(o.get("option_display_type") or "S"),
"required_option": str(o.get("required_option") or "T"),
"option_value": [
{
"option_text": str(v.get("option_text") or ""),
"option_image_file": str(v.get("option_image_file") or ""),
"option_link_image": str(v.get("option_link_image") or ""),
"option_color": str(v.get("option_color") or ""),
"value_no": v.get("value_no"),
}
for v in values if isinstance(v, dict)
],
}
)
return {
"has_option": products._flag(option.get("has_option"), default=False), # noqa: SLF001
"option_type": str(option.get("option_type") or ""),
"option_list_type": str(option.get("option_list_type") or ""),
"option_preset_code": str(option.get("option_preset_code") or ""),
"options": out_options,
}
def _apply_variant_snapshot(st: Any, product_no: int, variants: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""GET 품목 목록에 유예시간 안의 우리 쓰기(스냅샷)를 덮어씌운다.
삭제한 품목(`{"_deleted": true}`)은 카페24 GET 이 한동안 계속 돌려주므로 걸러낸다.
"""
section = (st.get_write_snapshot(product_no) or {}).get("variants") or {}
data = section.get("data") or {}
if not data or not store.within_grace(section.get("written_at"), grace_minutes=read_lag_grace_minutes()):
return variants
out = []
for v in variants:
code = str(v.get("variant_code") or "")
patch = data.get(code)
if isinstance(patch, dict) and patch.get("_deleted"):
continue
out.append({**v, **patch} if isinstance(patch, dict) else v)
return out
def _apply_options_snapshot(st: Any, product_no: int, option: dict[str, Any]) -> dict[str, Any]:
"""GET 옵션이 우리 마지막 쓰기와 다르고 유예시간 안이면 스냅샷(우리 쓰기)을 쓴다."""
section = (st.get_write_snapshot(product_no) or {}).get("options") or {}
data = section.get("data")
if not isinstance(data, dict) or not store.within_grace(
section.get("written_at"), grace_minutes=read_lag_grace_minutes()
):
return option
return data
def _load_options_and_variants(st: Any, api: Any, product_no: int) -> dict[str, Any]:
option = _apply_options_snapshot(st, product_no, products.get_options(api.client, product_no))
view = _options_view(option)
variants: list[dict[str, Any]] = []
if view["has_option"]:
variants = _apply_variant_snapshot(st, product_no, products.list_variants(api.client, product_no))
return {"option": view, "variants": [_variant_view(v) for v in variants], "raw_options": option}
@product_info_router.get("/products/{product_no}/options")
def options_get(request: Request, product_no: int) -> dict[str, Any]:
st, _user = require_store(request)
api = build_cafe24_api(st)
try:
loaded = _load_options_and_variants(st, api, product_no)
except Cafe24Error as exc:
raise _http_from_cafe24(exc) from exc
return {"ok": True, "option": loaded["option"], "variants": loaded["variants"],
"display_types": store.OPTION_DISPLAY_LABELS}
@product_info_router.post("/products/{product_no}/options")
def options_create(
request: Request,
product_no: int,
payload: dict[str, Any] = Body(default_factory=dict),
) -> dict[str, Any]:
"""옵션 없는 상품에 조합형 옵션 1개(옵션명 + 옵션값들)를 만든다. 품목은 자동 생성."""
st, user = require_store(request)
actor = str(user.get("email") or "")
try:
body = store.build_create_options_request(
str(payload.get("option_name") or ""),
store.parse_option_values(payload.get("values")),
display_type=str(payload.get("display_type") or "S"),
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
api = build_cafe24_api(st)
try:
existing = products.get_options(api.client, product_no)
if products._flag(existing.get("has_option"), default=False): # noqa: SLF001
raise HTTPException(status_code=409, detail="이미 옵션이 있는 상품입니다. 수정하거나 삭제한 뒤 다시 만드세요.")
created = products.create_options(api.client, product_no, body)
except Cafe24Error as exc:
st.log_audit(actor=actor, action="create_options", product_no=product_no,
result="FAIL", detail=f"{body['options'][0]['option_name']} 실패: {exc}")
raise _http_from_cafe24(exc) from exc
st.save_write_snapshot(product_no, "options", created)
values = [v["option_text"] for v in body["options"][0]["option_value"]]
st.log_audit(actor=actor, action="create_options", product_no=product_no, result="SUCCESS",
detail=f"옵션 '{body['options'][0]['option_name']}' 생성: {', '.join(values)}")
logger.info("카페24 상품 %s 옵션 생성 (%s)", product_no, actor)
# 카페24가 자동 생성한 품목(코드 부여)을 바로 돌려준다 — 읽기 지연이 있어 짧게 재시도.
# 화면은 이 코드로 자체코드·추가금액을 이어서 PUT 한다.
variants = products.wait_for_variants(api.client, product_no, len(values))
if len(variants) < len(values):
logger.warning("카페24 상품 %s 옵션 생성 후 품목 %s/%s 건만 조회됨", product_no, len(variants), len(values))
return {"ok": True, "option": _options_view(created), "variants": [_variant_view(v) for v in variants]}
@product_info_router.put("/products/{product_no}/options")
def options_update(
request: Request,
product_no: int,
payload: dict[str, Any] = Body(default_factory=dict),
) -> dict[str, Any]:
"""옵션명 · 옵션값 이름/썸네일/연결이미지/색상 · 표시방식 수정.
카페24 PUT 은 `original_options`(수정 전)와 `options`(수정 후)를 짝지어 받는다.
수정 전 값은 화면이 들고 있던 것이 아니라 **지금 카페24에서 다시 읽은 값**(읽기
지연 보정 후)을 쓴다 — 그래야 다른 곳에서 바뀐 이름과 어긋나지 않는다.
"""
st, user = require_store(request)
actor = str(user.get("email") or "")
edited = payload.get("options")
if not isinstance(edited, list) or not edited:
raise HTTPException(status_code=400, detail="옵션 목록이 비어 있습니다.")
api = build_cafe24_api(st)
try:
loaded = _load_options_and_variants(st, api, product_no)
except Cafe24Error as exc:
raise _http_from_cafe24(exc) from exc
if not loaded["option"]["has_option"]:
raise HTTPException(status_code=409, detail="옵션이 없는 상품입니다. 먼저 옵션을 만드세요.")
original = loaded["option"]["options"]
try:
body = store.build_update_options_request(
original, edited, option_list_type=str(payload.get("option_list_type") or ""),
allow_append=True,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
if loaded["option"]["option_preset_code"]:
body["option_preset_code"] = loaded["option"]["option_preset_code"]
# 뒤에 덧붙인 옵션값 수 — 카페24가 받아들이면 품목을 자동 생성한다(재시도 조회).
added = sum(
max(0, len(n["option_value"]) - len(o["option_value"]))
for o, n in zip(body["original_options"], body["options"])
)
try:
updated = products.update_options(api.client, product_no, body)
except Cafe24Error as exc:
st.log_audit(actor=actor, action="update_options", product_no=product_no,
result="FAIL", detail=str(exc))
logger.warning("카페24 상품 %s 옵션 수정 실패: %s", product_no, exc)
raise _http_from_cafe24(exc) from exc
st.save_write_snapshot(product_no, "options", updated)
summary = "; ".join(
f"{o['option_name']}: " + ", ".join(v["option_text"] for v in o["option_value"])
for o in body["options"]
) + (f" (+{added} 추가)" if added else "")
st.log_audit(actor=actor, action="update_options", product_no=product_no,
result="SUCCESS", detail=summary[:900])
logger.info("카페24 상품 %s 옵션 수정 (%s)", product_no, actor)
if added:
# 새 옵션값의 품목코드를 받아 돌려준다 — 화면이 자체코드/추가금액을 이어서 PUT 한다.
fresh = products.wait_for_variants(api.client, product_no, len(loaded["variants"]) + added)
fresh = _apply_variant_snapshot(st, product_no, fresh)
return {"ok": True, "option": _options_view(updated), "variants": [_variant_view(v) for v in fresh],
"added": added}
# 품목의 옵션값 표기는 이름을 바꾼 만큼 달라진다 — 응답값으로 다시 맞춘다.
variants = loaded["variants"]
rename: dict[tuple[str, str], tuple[str, str]] = {}
for o, n in zip(original, body["options"]):
for ov, nv in zip(o["option_value"], n["option_value"]):
rename[(o["option_name"], ov["option_text"])] = (n["option_name"], nv["option_text"])
for v in variants:
v["options"] = [
dict(zip(("name", "value"), rename.get((opt["name"], opt["value"]), (opt["name"], opt["value"]))))
for opt in v["options"]
]
return {"ok": True, "option": _options_view(updated), "variants": variants}
@product_info_router.delete("/products/{product_no}/options")
def options_delete(request: Request, product_no: int) -> dict[str, Any]:
"""옵션 삭제 — 카페24가 품목도 함께 지운다. 화면에서 확인창을 거친다."""
st, user = require_store(request)
actor = str(user.get("email") or "")
api = build_cafe24_api(st)
try:
before = products.get_options(api.client, product_no)
products.delete_options(api.client, product_no)
except Cafe24Error as exc:
st.log_audit(actor=actor, action="delete_options", product_no=product_no,
result="FAIL", detail=str(exc))
raise _http_from_cafe24(exc) from exc
names = ", ".join(
str(o.get("option_name") or "") for o in (before.get("options") or []) if isinstance(o, dict)
)
st.save_write_snapshot(product_no, "options", {"has_option": "F", "options": []})
st.save_write_snapshot(product_no, "variants", {})
st.log_audit(actor=actor, action="delete_options", product_no=product_no,
result="SUCCESS", detail=f"삭제된 옵션: {names or '(없음)'}")
logger.info("카페24 상품 %s 옵션 삭제 (%s)", product_no, actor)
return {"ok": True, "option": _options_view({"has_option": "F"}), "variants": []}
@product_info_router.post("/products/{product_no}/options/image")
def option_image_upload(
request: Request,
product_no: int,
file: UploadFile = File(...),
) -> dict[str, Any]:
"""옵션값 썸네일 업로드만 한다. 경로를 돌려주면 화면이 옵션 저장(PUT)에 담는다."""
st, user = require_store(request)
actor = str(user.get("email") or "")
data = _read_image(file)
api = build_cafe24_api(st)
try:
path = products.upload_image_bytes(api.client, data)
except Cafe24Error as exc:
st.log_audit(actor=actor, action="upload_option_image", product_no=product_no,
result="FAIL", detail=str(exc))
raise _http_from_cafe24(exc) from exc
if not path:
raise HTTPException(status_code=502, detail="카페24가 업로드 경로를 돌려주지 않았습니다.")
st.log_audit(actor=actor, action="upload_option_image", product_no=product_no,
result="SUCCESS", detail=path)
return {"ok": True, "path": path}
@product_info_router.delete("/products/{product_no}/variants/{variant_code}")
def variant_delete(request: Request, product_no: int, variant_code: str) -> dict[str, Any]:
"""품목 1건 삭제. 옵션값은 남을 수 있다(카페24 모델) — 정리는 옵션 재생성으로."""
st, user = require_store(request)
actor = str(user.get("email") or "")
code = str(variant_code or "").strip().upper()
if not store.VARIANT_CODE_RE.match(code):
raise HTTPException(status_code=400, detail=f"품목코드 형식이 올바르지 않습니다: {code}")
api = build_cafe24_api(st)
try:
products.delete_variant(api.client, product_no, code)
except Cafe24Error as exc:
st.log_audit(actor=actor, action="delete_variant", product_no=product_no,
result="FAIL", detail=f"{code}: {exc}")
raise _http_from_cafe24(exc) from exc
# 카페24 GET 은 한동안 삭제된 품목을 계속 돌려준다 — 스냅샷에 삭제 표식을 남겨 걸러낸다.
st.save_write_snapshot(product_no, "variants", {code: {"_deleted": True}})
st.log_audit(actor=actor, action="delete_variant", product_no=product_no,
result="SUCCESS", detail=code)
logger.info("카페24 상품 %s 품목 %s 삭제 (%s)", product_no, code, actor)
return {"ok": True, "variant_code": code}
@product_info_router.put("/products/{product_no}/variants")
def variants_update(
request: Request,
product_no: int,
payload: dict[str, Any] = Body(default_factory=dict),
) -> dict[str, Any]:
"""품목 여러 건의 자체코드 · 추가금액 · 진열 · 판매를 한 번에 바꾼다."""
st, user = require_store(request)
actor = str(user.get("email") or "")
rows = payload.get("rows")
if not isinstance(rows, list) or not rows:
raise HTTPException(status_code=400, detail="바꿀 품목이 없습니다.")
try:
requests = store.build_variant_updates(rows)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
if not requests:
raise HTTPException(status_code=400, detail="바뀐 값이 없습니다.")
api = build_cafe24_api(st)
try:
results = products.update_variants(api.client, product_no, requests)
except Cafe24Error as exc:
st.log_audit(actor=actor, action="update_variants", product_no=product_no,
result="FAIL", detail=f"{len(requests)}건 실패: {exc}")
logger.warning("카페24 상품 %s 품목 수정 실패: %s", product_no, exc)
raise _http_from_cafe24(exc) from exc
# 우리가 보낸 값이 곧 현재값이다(카페24가 받아들였으므로). 응답에 담긴 값이 있으면
# 그것을 우선한다.
by_code: dict[str, dict[str, Any]] = {}
for r in requests:
by_code[r["variant_code"]] = {k: v for k, v in r.items() if k != "variant_code"}
for r in results:
code = str(r.get("variant_code") or "")
if code in by_code:
for key in ("custom_variant_code", "additional_amount", "display", "selling", "display_order"):
if key in r and r[key] is not None:
by_code[code][key] = r[key]
st.save_write_snapshot(product_no, "variants", by_code)
detail = "; ".join(
f"{code}: " + ", ".join(f"{k}={v}" for k, v in patch.items()) for code, patch in by_code.items()
)
st.log_audit(actor=actor, action="update_variants", product_no=product_no,
result="SUCCESS", detail=detail[:900])
logger.info("카페24 상품 %s 품목 %s건 수정 (%s)", product_no, len(by_code), actor)
return {"ok": True, "updated": {code: _variant_view({"variant_code": code, **patch})
for code, patch in by_code.items()}}
Reference in New Issue View Git Blame Copy Permalink