"""말레이시아 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 functools import lru_cache from pathlib import Path from typing import Any, Iterable from urllib.parse import quote # ── 작업 상태 필드(토글 대상) ── # 순서 = 작업 진행 순서. label_ko/label_en 은 버튼 표기. # 상품은 미리 제작·박스 포장까지 끝난 상태로 들어오므로 product_ready/packed # 단계는 화면에서 쓰지 않는다(DB 컬럼은 보존 — 과거 데이터/확장 대비). # 실제 작업: 송장 라벨 부착 → Kagayaku 전달 → 택배 스캔 확인. STATUS_FIELDS: tuple[str, ...] = ( "label_attached", "handed_to_kagayaku", ) 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 # ── 배송사 로고 ── # 배송사명(부분 일치, 소문자) → 슬러그. 슬러그 이름의 이미지 파일이 실제로 # 있으면 이미지로 렌더하고, 없으면 텍스트 배지로 폴백한다(깨진 이미지 방지). # 새 배송사 로고 추가: 파일 1개 + 키워드 1줄. # 1) app/static/dispatch/courier/<슬러그>.png (또는 .svg) 추가 # 2) 아래 _COURIER_KEYWORDS 에 (배송사명키워드, 슬러그) 추가 _COURIER_KEYWORDS: tuple[tuple[str, str], ...] = ( ("ninja", "ninjavan"), ("j&t", "jnt"), ("jnt", "jnt"), ("flash", "flash"), ("city", "citylink"), ("pos laju", "poslaju"), ("poslaju", "poslaju"), ("gdex", "gdex"), ("gd express", "gdex"), ("shopee", "spx"), ("spx", "spx"), ) # 로고 파일 디렉토리(app/static/dispatch/courier). png 를 svg 보다 우선한다. _COURIER_DIR = Path(__file__).resolve().parents[2] / "static" / "dispatch" / "courier" _COURIER_EXTS: tuple[str, ...] = (".png", ".svg") def courier_slug(provider: Any) -> str: """배송사명 → 로고 슬러그(부분 일치). 매칭 없으면 "".""" name = clean_text(provider).lower() if not name: return "" for keyword, slug in _COURIER_KEYWORDS: if keyword in name: return slug return "" @lru_cache(maxsize=128) def _logo_url_for_slug(slug: str) -> str: """슬러그에 해당하는 실제 로고 파일 URL(png 우선). 없으면 "".""" if not slug: return "" for ext in _COURIER_EXTS: if (_COURIER_DIR / f"{slug}{ext}").exists(): return f"/static/dispatch/courier/{slug}{ext}" return "" def courier_logo_url(provider: Any) -> str: """배송사명 → 로고 이미지 URL. 파일 없으면 ""(템플릿이 텍스트로 폴백).""" return _logo_url_for_slug(courier_slug(provider)) # ── 배송사별 송장 조회 딥링크 ── # {t} 자리에 송장번호를 넣어 조회 페이지로 바로 이동(자동 입력 효과). # ⚠️ 배송사 사이트 URL 은 바뀔 수 있다. 안 열리면 여기만 고치면 된다. # 매핑 없는 배송사는 링크 없이 텍스트로 표시(courier_tracking_url 가 "" 반환). _COURIER_TRACK_URL: dict[str, str] = { "ninjavan": "https://www.ninjavan.co/en-my/tracking?id={t}", "jnt": "https://www.jtexpress.my/index/query/gzquery.html?bills={t}", "flash": "https://www.flashexpress.my/fle/tracking?se={t}", "citylink": "https://www.citylinkexpress.com/tracking-result/?track_no={t}", "poslaju": "https://track.pos.com.my/postal-services/quick-access/?track-trace&trackingNo03={t}", "gdex": "https://web.gdexpress.com/official/web/ConsignmentSearch.html?capcheck=true&input_search={t}", "spx": "https://spx.com.my/track?tracking_number={t}", } def courier_tracking_url(tracking: Any, provider: Any = "") -> str: """송장번호 + 배송사명 → 조회 딥링크. 매핑/번호 없으면 "".""" t = clean_text(tracking) tmpl = _COURIER_TRACK_URL.get(courier_slug(provider)) if not t or not tmpl: return "" return tmpl.replace("{t}", quote(t, safe="")) 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"])