feat(dispatch): 말레이시아 TikTok 출고관리 모듈 추가
03_TikTok_Order_Export.xlsx 업로드 → 1박스=1카드 출고 작업 리스트, SKU 피킹 요약, Kagayaku 전달표(A4 인쇄)를 자동 생성. - 1박스 묶음 기준: Package ID > Tracking ID > Order ID, 같은 박스 같은 SKU 합산 - 작업 상태 5단계 토글(AJAX 즉시 저장) + dispatch_logs 기록 - 고객 이름/주소/전화 미저장(파싱 시 폐기, 스키마에 컬럼 없음) - 엑셀은 openpyxl 파싱(pandas 미사용), DB는 dispatch_db 전용 - 인증은 기존 Google OAuth + 신규 권한키 dispatch 재사용 - scripts/sql/dispatch_db_init.sql, docs/DISPATCH_MODULE.md 동봉 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,223 @@
|
||||
"""말레이시아 TikTok 출고관리 모듈 — 상수 및 순수 계산/검증 헬퍼.
|
||||
|
||||
- 데이터 저장은 dispatch_db(PostgreSQL) 전용이다(`db.py`).
|
||||
- 엑셀 읽기는 `parser.py`(openpyxl)가 담당하고, 여기에는 DB/파일 의존이 없는
|
||||
순수 함수(컬럼 정규화, 박스 묶기, SKU 합산)만 둔다(테스트 용이).
|
||||
|
||||
핵심 업무 규칙:
|
||||
- 03_TikTok_Order_Export.xlsx 가 자동 출고 리스트의 기준 데이터다.
|
||||
- 엑셀 1줄 ≠ 1박스. 1박스 묶음 기준은 아래 우선순위로 정한다:
|
||||
1순위 package_id → 2순위 tracking_id → 3순위 order_id
|
||||
- 같은 박스 안 같은 SKU 는 수량을 합산한다.
|
||||
- 고객 이름/주소/전화 등 개인정보는 절대 보관하지 않는다(허용 컬럼만 사용).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Iterable
|
||||
|
||||
# ── 작업 상태 필드(토글 대상) ──
|
||||
# 순서 = 작업 진행 순서. label_ko/label_en 은 버튼 표기.
|
||||
STATUS_FIELDS: tuple[str, ...] = (
|
||||
"product_ready",
|
||||
"packed",
|
||||
"label_attached",
|
||||
"handed_to_kagayaku",
|
||||
"courier_scanned",
|
||||
)
|
||||
|
||||
STATUS_LABELS: dict[str, dict[str, str]] = {
|
||||
"product_ready": {"ko": "상품준비 완료", "en": "Product Ready"},
|
||||
"packed": {"ko": "포장완료", "en": "Packed"},
|
||||
"label_attached": {"ko": "라벨부착 완료", "en": "Label Attached"},
|
||||
"handed_to_kagayaku": {"ko": "Kagayaku 전달 완료", "en": "Handed to Kagayaku"},
|
||||
"courier_scanned": {"ko": "택배스캔 확인", "en": "Courier Scanned"},
|
||||
}
|
||||
|
||||
# ── 엑셀 컬럼명(앞뒤 공백 제거 후, 소문자 비교). 표준키 → 가능한 헤더들 ──
|
||||
# 개인정보 컬럼은 의도적으로 매핑하지 않는다(이름/주소/전화는 버린다).
|
||||
COLUMN_ALIASES: dict[str, tuple[str, ...]] = {
|
||||
"order_id": ("order id", "order_id", "주문번호"),
|
||||
"package_id": ("package id", "package_id", "패키지번호"),
|
||||
"tracking_id": ("tracking id", "tracking_id", "tracking number", "송장번호"),
|
||||
"shipping_provider": ("shipping provider name", "shipping provider", "shipping_provider_name", "배송사", "택배사"),
|
||||
"seller_sku": ("seller sku", "seller_sku", "sku", "판매자 sku"),
|
||||
"product_name": ("product name", "product_name", "상품명"),
|
||||
"quantity": ("quantity", "qty", "수량"),
|
||||
}
|
||||
|
||||
# 사용자에게 보여줄 컬럼 이름(누락 안내 메시지용). 비교키는 소문자라 별도 표기.
|
||||
COLUMN_DISPLAY: dict[str, str] = {
|
||||
"order_id": "Order ID",
|
||||
"package_id": "Package ID",
|
||||
"tracking_id": "Tracking ID",
|
||||
"shipping_provider": "Shipping Provider Name",
|
||||
"seller_sku": "Seller SKU",
|
||||
"product_name": "Product Name",
|
||||
"quantity": "Quantity",
|
||||
}
|
||||
|
||||
# 박스 묶음 기준 우선순위. 앞에서부터 비어있지 않은 첫 값으로 묶는다.
|
||||
BOX_KEY_PRIORITY: tuple[str, ...] = ("package_id", "tracking_id", "order_id")
|
||||
|
||||
# 파싱에 최소한 필요한 표준키. (개인정보는 불필요)
|
||||
# - SKU/수량이 없으면 포장할 상품을 못 만든다 → seller_sku 필수.
|
||||
# - 박스 묶음을 위해 묶음키 3개 중 최소 하나는 있어야 한다.
|
||||
REQUIRED_COLUMNS: tuple[str, ...] = ("seller_sku",)
|
||||
|
||||
|
||||
def clean_text(value: Any) -> str:
|
||||
"""셀 값 → 안전한 문자열. None/NaN 은 빈 문자열. 양끝 공백 제거.
|
||||
|
||||
숫자는 과학적 표기/불필요한 .0 없이 보존한다(예: 송장번호가 float 로 읽혀도
|
||||
'12345678901' 로 보존). Tracking ID 보존 요구사항을 만족한다.
|
||||
"""
|
||||
if value is None:
|
||||
return ""
|
||||
# pandas/openpyxl NaN(float('nan')) 방어
|
||||
if isinstance(value, float):
|
||||
if value != value: # NaN
|
||||
return ""
|
||||
if value.is_integer():
|
||||
return str(int(value))
|
||||
return repr(value)
|
||||
if isinstance(value, int):
|
||||
return str(value)
|
||||
s = str(value).strip()
|
||||
if s.lower() in ("nan", "none"):
|
||||
return ""
|
||||
return s
|
||||
|
||||
|
||||
def normalize_header(value: Any) -> str:
|
||||
"""헤더 셀 → 비교용 키(공백 제거 + 소문자)."""
|
||||
return clean_text(value).lower()
|
||||
|
||||
|
||||
def resolve_columns(headers: Iterable[Any]) -> dict[str, int]:
|
||||
"""헤더 행 → {표준키: 열 인덱스}. 매핑 안 되는 열은 무시(개인정보 포함)."""
|
||||
norm = [normalize_header(h) for h in headers]
|
||||
mapping: dict[str, int] = {}
|
||||
for std_key, aliases in COLUMN_ALIASES.items():
|
||||
for idx, h in enumerate(norm):
|
||||
if h in aliases:
|
||||
mapping[std_key] = idx
|
||||
break
|
||||
return mapping
|
||||
|
||||
|
||||
def missing_required(mapping: dict[str, int]) -> list[str]:
|
||||
"""필수 컬럼 누락 목록(원문 헤더 형태로). 묶음키 전무도 누락으로 본다."""
|
||||
missing: list[str] = []
|
||||
for key in REQUIRED_COLUMNS:
|
||||
if key not in mapping:
|
||||
missing.append(COLUMN_DISPLAY[key])
|
||||
if not any(k in mapping for k in BOX_KEY_PRIORITY):
|
||||
missing.append("Package ID / Tracking ID / Order ID (최소 1개)")
|
||||
return missing
|
||||
|
||||
|
||||
def box_key(row: dict[str, str]) -> str:
|
||||
"""박스 묶음 키 — package_id > tracking_id > order_id 우선순위."""
|
||||
for key in BOX_KEY_PRIORITY:
|
||||
v = (row.get(key) or "").strip()
|
||||
if v:
|
||||
return f"{key}:{v}"
|
||||
return ""
|
||||
|
||||
|
||||
def group_parcels(rows: list[dict[str, str]]) -> list[dict[str, Any]]:
|
||||
"""정규화된 행 리스트 → 박스(parcel) 리스트.
|
||||
|
||||
rows 각 항목 키: order_id/package_id/tracking_id/shipping_provider/
|
||||
seller_sku/product_name/quantity(이미 문자열/int 정리됨).
|
||||
|
||||
반환(등장 순서 유지, seq 1..N 부여):
|
||||
[{seq, order_id, package_id, tracking_id, shipping_provider,
|
||||
items: [{seller_sku, product_name, quantity}, ...]}, ...]
|
||||
같은 박스 안 같은 SKU 는 수량 합산. 묶음키가 전혀 없는 행은 버린다.
|
||||
"""
|
||||
boxes: dict[str, dict[str, Any]] = {}
|
||||
for row in rows:
|
||||
key = box_key(row)
|
||||
if not key:
|
||||
continue
|
||||
box = boxes.get(key)
|
||||
if box is None:
|
||||
box = {
|
||||
"key": key,
|
||||
"order_id": (row.get("order_id") or "").strip(),
|
||||
"package_id": (row.get("package_id") or "").strip(),
|
||||
"tracking_id": (row.get("tracking_id") or "").strip(),
|
||||
"shipping_provider": (row.get("shipping_provider") or "").strip(),
|
||||
"_items": {}, # sku -> {product_name, quantity}
|
||||
}
|
||||
boxes[key] = box
|
||||
else:
|
||||
# 같은 박스인데 식별값이 빈 경우 뒤늦게 채운다(누락 보강).
|
||||
for field in ("order_id", "package_id", "tracking_id", "shipping_provider"):
|
||||
if not box[field] and (row.get(field) or "").strip():
|
||||
box[field] = (row.get(field) or "").strip()
|
||||
|
||||
sku = (row.get("seller_sku") or "").strip()
|
||||
if not sku:
|
||||
continue
|
||||
qty = normalize_quantity(row.get("quantity"))
|
||||
item = box["_items"].get(sku)
|
||||
if item is None:
|
||||
box["_items"][sku] = {
|
||||
"seller_sku": sku,
|
||||
"product_name": (row.get("product_name") or "").strip(),
|
||||
"quantity": qty,
|
||||
}
|
||||
else:
|
||||
item["quantity"] += qty
|
||||
if not item["product_name"] and (row.get("product_name") or "").strip():
|
||||
item["product_name"] = (row.get("product_name") or "").strip()
|
||||
|
||||
parcels: list[dict[str, Any]] = []
|
||||
for seq, box in enumerate(boxes.values(), start=1):
|
||||
items = sorted(box["_items"].values(), key=lambda it: it["seller_sku"])
|
||||
parcels.append(
|
||||
{
|
||||
"seq": seq,
|
||||
"order_id": box["order_id"],
|
||||
"package_id": box["package_id"],
|
||||
"tracking_id": box["tracking_id"],
|
||||
"shipping_provider": box["shipping_provider"],
|
||||
"items": items,
|
||||
}
|
||||
)
|
||||
return parcels
|
||||
|
||||
|
||||
def normalize_quantity(value: Any) -> int:
|
||||
"""수량 정규화. 비어있으면 1. 음수/0/파싱불가도 1 로 보정(요구사항 11)."""
|
||||
s = clean_text(value)
|
||||
if s == "":
|
||||
return 1
|
||||
try:
|
||||
n = int(float(s))
|
||||
except (TypeError, ValueError):
|
||||
return 1
|
||||
return n if n > 0 else 1
|
||||
|
||||
|
||||
def sku_summary(parcels: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
"""전체 박스 기준 SKU별 총 수량(피킹 요약). seller_sku 오름차순."""
|
||||
totals: dict[str, dict[str, Any]] = {}
|
||||
for p in parcels:
|
||||
for it in p.get("items", []):
|
||||
sku = it["seller_sku"]
|
||||
row = totals.get(sku)
|
||||
if row is None:
|
||||
totals[sku] = {
|
||||
"seller_sku": sku,
|
||||
"product_name": it.get("product_name", ""),
|
||||
"total_qty": int(it["quantity"]),
|
||||
}
|
||||
else:
|
||||
row["total_qty"] += int(it["quantity"])
|
||||
if not row["product_name"] and it.get("product_name"):
|
||||
row["product_name"] = it["product_name"]
|
||||
return sorted(totals.values(), key=lambda r: r["seller_sku"])
|
||||
Reference in New Issue
Block a user