"""말레이시아 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"])