king
8523a620ad
jtexpress.my/tracking/<송장번호> 형태면 자동 조회됨(쿼리 파라미터는 무시). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
304 lines
12 KiB
Python
304 lines
12 KiB
Python
"""말레이시아 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}",
|
|
# J&T MY 는 경로(path)에 송장번호를 붙이면 자동 조회된다(쿼리 파라미터는 무시).
|
|
"jnt": "https://www.jtexpress.my/tracking/{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"])
|