27e2c17606
- 다운로드 파일은 재고가 아니라 사방넷에 올려 출고로 잡는 파일(업로드 시 재고 차감). 달력 카드 제목/안내/버튼 문구를 출고로 정정, 파일명 _sabangnet_outbound.xlsx - 컬럼/집계 로직은 동일 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
254 lines
8.9 KiB
Python
254 lines
8.9 KiB
Python
"""출고 배치 → 취합 엑셀(openpyxl) 생성.
|
|
|
|
다운로드 zip 에 업로드 원본과 함께 넣는 '출고 엑셀'을 만든다. 1박스 안에 여러
|
|
SKU 가 있으면 SKU 당 1행으로 펼치고 박스 단위(받는 사람/주문/택배) 정보는 반복한다.
|
|
|
|
파일명 형식: 2026.06.22(Mon)_tictoc.xlsx (Shopee 는 끝이 _shopee).
|
|
- TikTok 은 운영 요청 철자에 맞춰 'tictoc' 을 쓴다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import re
|
|
from datetime import date, datetime
|
|
from typing import Any
|
|
|
|
# 엑셀 컬럼(요청 순서). (헤더, 박스/배치/상품 키)
|
|
_COLUMNS: tuple[tuple[str, str], ...] = (
|
|
("발송 날짜", "dispatch_date"),
|
|
("고객 이름", "recipient_name"),
|
|
("전화번호", "recipient_phone"),
|
|
("주소", "recipient_address"),
|
|
("상품이름", "product_name"),
|
|
("아이템 코드", "seller_sku"),
|
|
("주문 수량", "quantity"),
|
|
("오더번호", "order_id"),
|
|
("택배사", "shipping_provider"),
|
|
("택배송장번호", "tracking_id"),
|
|
("주문처", "platform"),
|
|
)
|
|
|
|
# 플랫폼 → 파일명 접미사.
|
|
_PLATFORM_SUFFIX: dict[str, str] = {
|
|
"TikTok": "tictoc",
|
|
"Shopee": "shopee",
|
|
}
|
|
|
|
|
|
def export_filename(*, dispatch_date: str, platform: str) -> str:
|
|
"""2026.06.22(Mon)_tictoc.xlsx 형식 파일명. 날짜 파싱 실패 시 원문 사용."""
|
|
d = _parse_date(dispatch_date)
|
|
if d is not None:
|
|
stamp = d.strftime("%Y.%m.%d(%a)")
|
|
else:
|
|
stamp = (dispatch_date or "").strip() or "dispatch"
|
|
suffix = _PLATFORM_SUFFIX.get((platform or "").strip(), (platform or "dispatch").strip().lower())
|
|
return f"{stamp}_{suffix}.xlsx"
|
|
|
|
|
|
def split_sku(seller_sku: str) -> tuple[str, int]:
|
|
"""seller_sku → (아이템코드, 배수).
|
|
|
|
끝의 `_<숫자>` 는 '같은 상품 N개'를 뜻한다(예: MT-0320_3 → 코드 MT-0320, 3개).
|
|
접미사가 없으면 배수 1.
|
|
"""
|
|
s = (seller_sku or "").strip()
|
|
m = re.search(r"_(\d+)$", s)
|
|
if m:
|
|
return s[: m.start()], int(m.group(1))
|
|
return s, 1
|
|
|
|
|
|
def resolve_product_name(seller_sku: str, name_map: dict[str, str], fallback: str = "") -> str:
|
|
"""아이템코드(seller_sku)로 itemcode_db 상품명을 찾는다.
|
|
|
|
seller_sku 는 변형 접미사가 붙을 수 있다(예: MT-0320_3 → DB 는 MT-0320).
|
|
정확 일치 → 접미사(_숫자) 제거 후 일치 → 없으면 fallback(파싱된 상품명).
|
|
"""
|
|
key = (seller_sku or "").strip().upper()
|
|
if not key:
|
|
return fallback
|
|
if key in name_map:
|
|
return name_map[key]
|
|
base = re.sub(r"_\d+$", "", key)
|
|
if base != key and base in name_map:
|
|
return name_map[base]
|
|
return fallback
|
|
|
|
|
|
def build_workbook_bytes(
|
|
*,
|
|
batch: dict[str, Any],
|
|
parcels: list[dict[str, Any]],
|
|
name_map: dict[str, str] | None = None,
|
|
) -> bytes:
|
|
"""배치 + 박스(상품 포함) → xlsx 바이트.
|
|
|
|
parcels 각 항목은 db.list_parcels 결과(items 포함, recipient_* 포함).
|
|
name_map: itemcode_db 의 {코드(대문자): 상품명}. 있으면 아이템코드로 상품명을
|
|
조회해 채운다(없으면 파싱된 상품명 사용).
|
|
"""
|
|
name_map = name_map or {}
|
|
from openpyxl import Workbook # noqa: WPS433 — 지연 import
|
|
from openpyxl.styles import Font
|
|
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "출고"
|
|
|
|
headers = [h for h, _ in _COLUMNS]
|
|
ws.append(headers)
|
|
for cell in ws[1]:
|
|
cell.font = Font(bold=True)
|
|
|
|
dispatch_date = batch.get("dispatch_date", "") or ""
|
|
platform = batch.get("platform", "") or ""
|
|
|
|
for p in parcels:
|
|
box = {
|
|
"dispatch_date": dispatch_date,
|
|
"platform": platform,
|
|
"recipient_name": p.get("recipient_name", "") or "",
|
|
"recipient_phone": p.get("recipient_phone", "") or "",
|
|
"recipient_address": p.get("recipient_address", "") or "",
|
|
"order_id": p.get("order_id", "") or "",
|
|
"shipping_provider": p.get("shipping_provider", "") or "",
|
|
"tracking_id": p.get("tracking_id", "") or "",
|
|
}
|
|
items = p.get("items") or [{}]
|
|
for it in items:
|
|
row_data = dict(box)
|
|
sku = it.get("seller_sku", "") or ""
|
|
base_code, mult = split_sku(sku)
|
|
row_data["product_name"] = resolve_product_name(
|
|
base_code, name_map, fallback=it.get("product_name", "") or ""
|
|
)
|
|
row_data["seller_sku"] = base_code # 접미사(_N) 제거한 실제 아이템코드
|
|
qty = it.get("quantity")
|
|
row_data["quantity"] = (int(qty) * mult) if qty is not None else ""
|
|
ws.append([_cell(row_data.get(key, "")) for _, key in _COLUMNS])
|
|
|
|
_autosize(ws, headers)
|
|
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
# ── 사방넷 재고 업로드 엑셀(낱개 분해) ──
|
|
_STOCK_HEADERS: tuple[str, ...] = ("상품코드[필수]", "가용수량", "불용수량", "바코드")
|
|
_LID_PREFIX = "MD-" # 뚜껑 — 재고 집계 제외
|
|
|
|
|
|
def explode_to_singles(
|
|
rows: list[dict[str, Any]],
|
|
set_bom: dict[str, list[dict[str, Any]]],
|
|
) -> dict[str, int]:
|
|
"""SKU별 합산 행 → 낱개 코드별 수량.
|
|
|
|
- seller_sku 의 _N 접미사는 배수(같은 상품 N개)로 곱한다.
|
|
- 콤보(세트, set_bom 에 존재)는 구성 낱개로 분해해 (배수 × 구성수량) 더한다.
|
|
- MD-(뚜껑)은 재고 집계에서 제외한다.
|
|
반환: {낱개코드(대문자): 수량}.
|
|
"""
|
|
singles: dict[str, int] = {}
|
|
for r in rows:
|
|
base, mult = split_sku(r.get("seller_sku", ""))
|
|
base = base.upper()
|
|
if not base:
|
|
continue
|
|
pieces = int(r.get("qty", 0)) * mult
|
|
if pieces <= 0 or base.startswith(_LID_PREFIX):
|
|
continue
|
|
comps = set_bom.get(base)
|
|
if comps: # 콤보 → 낱개 분해
|
|
for c in comps:
|
|
cc = str(c.get("component_code", "")).strip().upper()
|
|
if not cc or cc.startswith(_LID_PREFIX):
|
|
continue
|
|
singles[cc] = singles.get(cc, 0) + pieces * int(c.get("component_qty", 0) or 0)
|
|
else:
|
|
singles[base] = singles.get(base, 0) + pieces
|
|
return {k: v for k, v in singles.items() if v > 0}
|
|
|
|
|
|
def build_stock_xlsx(
|
|
*,
|
|
rows: list[dict[str, Any]],
|
|
set_bom: dict[str, list[dict[str, Any]]],
|
|
sabangnet_map: dict[str, str],
|
|
) -> bytes:
|
|
"""사방넷 재고 업로드용 xlsx(낱개 분해). 4열: 상품코드[필수]/가용수량/불용수량/바코드.
|
|
|
|
A=사방넷코드, B=수량, C=0(불용), D=아이템코드(바코드). 코드 오름차순.
|
|
"""
|
|
from openpyxl import Workbook # noqa: WPS433
|
|
from openpyxl.styles import Alignment, Font, PatternFill
|
|
|
|
singles = explode_to_singles(rows, set_bom)
|
|
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "재고"
|
|
ws.append(list(_STOCK_HEADERS))
|
|
head_fill = PatternFill("solid", fgColor="C55A11") # 사방넷 템플릿 주황
|
|
head_font = Font(bold=True, color="FFFFFF")
|
|
for cell in ws[1]:
|
|
cell.fill = head_fill
|
|
cell.font = head_font
|
|
cell.alignment = Alignment(horizontal="center", vertical="center")
|
|
|
|
for code in sorted(singles):
|
|
sabangnet = sabangnet_map.get(code, "")
|
|
ws.append([sabangnet, singles[code], 0, code])
|
|
|
|
widths = [22, 12, 12, 18]
|
|
from openpyxl.utils import get_column_letter
|
|
for i, w in enumerate(widths, start=1):
|
|
ws.column_dimensions[get_column_letter(i)].width = w
|
|
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
def stock_filename(dispatch_date: str) -> str:
|
|
"""사방넷 출고 엑셀 파일명: 2026.06.22(Mon)_sabangnet_outbound.xlsx."""
|
|
d = _parse_date(dispatch_date)
|
|
stamp = d.strftime("%Y.%m.%d(%a)") if d else (dispatch_date or "outbound").strip()
|
|
return f"{stamp}_sabangnet_outbound.xlsx"
|
|
|
|
|
|
def _cell(value: Any) -> Any:
|
|
"""송장/전화번호 등이 숫자로 보이면 엑셀이 과학표기/반올림 하므로 문자열 보존."""
|
|
if isinstance(value, int):
|
|
return value
|
|
return "" if value is None else str(value)
|
|
|
|
|
|
def _autosize(ws: Any, headers: list[str]) -> None:
|
|
"""대략적인 컬럼 폭. 주소는 길어 상한을 둔다."""
|
|
from openpyxl.utils import get_column_letter
|
|
|
|
widths: list[int] = [len(h) for h in headers]
|
|
for row in ws.iter_rows(min_row=2, values_only=True):
|
|
for i, val in enumerate(row):
|
|
ln = len(str(val)) if val is not None else 0
|
|
if ln > widths[i]:
|
|
widths[i] = ln
|
|
for i, w in enumerate(widths, start=1):
|
|
ws.column_dimensions[get_column_letter(i)].width = min(max(w + 2, 8), 50)
|
|
|
|
|
|
def _parse_date(value: str) -> date | None:
|
|
s = (value or "").strip()
|
|
if not s:
|
|
return None
|
|
for fmt in ("%Y-%m-%d", "%Y.%m.%d", "%Y/%m/%d"):
|
|
try:
|
|
return datetime.strptime(s, fmt).date()
|
|
except ValueError:
|
|
continue
|
|
return None
|