"""라벨 PDF → 주문별 받는 사람(이름/전화/주소) 추출 (pdfplumber). xlsx 는 박스/SKU/수량/주문/송장을 담당하고, 받는 사람 개인정보는 가려져 있거나 컬럼이 없어 라벨 PDF 에서 가져온다. 추출 결과는 Order ID 로 박스와 조인한다. - Shopee(SPX 라벨): 1주문 = 라벨 1p + 패킹리스트 1p. 라벨 페이지에 'Recipient Details (Penerima)' 블록이 있다. 2단 레이아웃이라 왼쪽 열만 본다. 전화는 표시되지 않는다(빈 값). - TikTok: 1주문 = 1p. 'To <이름> (+60)<전화>' 줄 + 그 아래 주소. 전화는 일부 마스킹될 수 있다(보이는 그대로 보존). 좌표 의존 파서라 라벨 양식이 바뀌면 조정이 필요하다(상단 상수만 손보면 됨). """ from __future__ import annotations import logging import re from typing import Any logger = logging.getLogger("dispatch.pdf") # Shopee 라벨 2단 중 왼쪽 열 경계(x0 < 값). 오른쪽 열의 바코드/코드(SPXMY…, # Seller Details, KV6-…)를 배제한다. _SHOPEE_LEFT_MAX = 185 # 같은 줄로 묶을 top 좌표 허용 오차(px). _LINE_TOL = 3 # 전화번호 토큰: (+60)193819581 / (+60)11******36 등. _PHONE_RE = re.compile(r"\(\+?\d[\d()*\- ]*") class PdfParseError(Exception): """PDF 파싱 실패. message 는 사용자에게 보여줄 한국어 메시지.""" def extract_recipients(path: str, platform: str) -> dict[str, dict[str, str]]: """라벨 PDF → {order_id: {recipient_name, recipient_phone, recipient_address}}. 실패해도 빈 dict 를 반환할 수 있다(호출부는 PII 없이 진행). 열기 자체 실패는 PdfParseError. """ try: import pdfplumber # noqa: WPS433 — 지연 import except ImportError as exc: # pragma: no cover raise PdfParseError("pdfplumber 가 설치되어 있지 않습니다.") from exc try: pdf = pdfplumber.open(path) except Exception as exc: # noqa: BLE001 raise PdfParseError(f"PDF 를 열 수 없습니다: {type(exc).__name__}") from exc out: dict[str, dict[str, str]] = {} try: for page in pdf.pages: try: words = page.extract_words() except Exception: # noqa: BLE001 continue if not words: continue if platform == "Shopee": rec = _shopee_page(words) else: rec = _tiktok_page(words) if rec and rec.get("order_id"): oid = rec.pop("order_id") # 이미 있으면 빈 값만 보강(중복 페이지 방어). cur = out.setdefault(oid, {"recipient_name": "", "recipient_phone": "", "recipient_address": ""}) for k, v in rec.items(): if v and not cur.get(k): cur[k] = v finally: pdf.close() return out # ──────────────────────────────────────────────────────────── # 줄 단위 헬퍼 # ──────────────────────────────────────────────────────────── def _lines(words: list[dict[str, Any]], *, x_max: float | None = None) -> list[tuple[float, list[dict[str, Any]]]]: """단어들을 top 좌표로 묶어 줄 리스트로. 각 줄: (top, [word,...]) x0 오름차순.""" ws = [w for w in words if x_max is None or w["x0"] < x_max] ws.sort(key=lambda w: (round(w["top"]), w["x0"])) lines: list[tuple[float, list[dict[str, Any]]]] = [] for w in ws: if lines and abs(w["top"] - lines[-1][0]) <= _LINE_TOL: lines[-1][1].append(w) else: lines.append((w["top"], [w])) return lines def _text(line_words: list[dict[str, Any]]) -> str: return " ".join(w["text"] for w in sorted(line_words, key=lambda w: w["x0"])).strip() # ──────────────────────────────────────────────────────────── # Shopee 라벨 페이지 # ──────────────────────────────────────────────────────────── def _shopee_page(words: list[dict[str, Any]]) -> dict[str, str] | None: lines = _lines(words, x_max=_SHOPEE_LEFT_MAX) texts = [_text(lw) for _, lw in lines] joined = " ".join(texts) if "Recipient Details" not in joined and "Penerima" not in joined: return None # 라벨 페이지 아님(패킹리스트 등) order_id = "" rec_idx = None for i, t in enumerate(texts): if not order_id: m = re.search(r"Order\s*ID\s*:\s*([A-Za-z0-9]+)", t) if m: order_id = m.group(1) if rec_idx is None and ("Recipient Details" in t or "Penerima" in t): rec_idx = i if rec_idx is None: return None name = "" addr_parts: list[str] = [] postcode = "" collecting = False for t in texts[rec_idx + 1:]: if t.startswith("Name:"): name = t[len("Name:"):].strip() continue if t.startswith("Address:"): collecting = True addr_parts.append(t[len("Address:"):].strip()) continue if t.startswith("Postcode:"): postcode = t[len("Postcode:"):].strip() break # 받는 사람 우편번호에서 주소 끝 if collecting: # 프로모/푸터 시작 전까지 주소 줄로 본다. if t.startswith(("Select", "Scan", "Seller Details")): break addr_parts.append(t) address = ", ".join(p for p in addr_parts if p) if postcode and postcode not in address: address = f"{address}, {postcode}" if address else postcode address = _tidy(address) return {"order_id": order_id, "recipient_name": name, "recipient_phone": "", "recipient_address": address} # ──────────────────────────────────────────────────────────── # TikTok 라벨 페이지 # ──────────────────────────────────────────────────────────── # 주소 수집을 멈추는 키워드(섹션 경계). TikTok 라벨은 배송사마다 양식이 달라 # 넉넉히 둔다. _TIKTOK_STOP = ("CASHLESS", "CASH ON", "COD", "Order ID", "Shipping Date", "Product Name", "Estimated", "In transit", "DROP-OFF", "PICK-UP", "Self Collect", "Order Created", "Return for", "Scan me", "Sender") # 받는 사람 앵커: 'To <이름> (+60)…'(Ninja) 또는 'Receiver <이름>'(NDD 등). _TIKTOK_ANCHOR = re.compile(r"^(To|Receiver)\b\s*(.*)$") def _tiktok_page(words: list[dict[str, Any]]) -> dict[str, str] | None: lines = _lines(words) texts = [_text(lw) for _, lw in lines] order_id = "" for t in texts: m = re.search(r"Order\s*ID\s*:\s*(\d+)", t) if m: order_id = m.group(1) break if not order_id: return None anchor_idx = None for i, t in enumerate(texts): m = _TIKTOK_ANCHOR.match(t) if m and (m.group(1) == "Receiver" or _PHONE_RE.search(t)): anchor_idx = i break if anchor_idx is None: return {"order_id": order_id, "recipient_name": "", "recipient_phone": "", "recipient_address": ""} rest = _TIKTOK_ANCHOR.match(texts[anchor_idx]).group(2).strip() phone = "" name = rest pm = _PHONE_RE.search(rest) if pm: # 'To <이름> (+60)<전화>' 형식 phone = rest[pm.start():].strip() name = rest[:pm.start()].strip() addr_parts: list[str] = [] for t in texts[anchor_idx + 1:]: if any(k in t for k in _TIKTOK_STOP): break if t.strip(): addr_parts.append(t.strip()) # 일부 배송사 라벨은 바코드/송장 숫자가 주소 영역에 섞인다. 10자리 이상 # 숫자열은 주소가 아니라 바코드이므로 제거(우편번호 5자리는 보존). address = re.sub(r"\b\d{10,}\b", "", ", ".join(addr_parts)) address = _tidy(address) return {"order_id": order_id, "recipient_name": name, "recipient_phone": phone, "recipient_address": address} def _tidy(s: str) -> str: """중복 공백/콤마 정리.""" s = re.sub(r"\s+", " ", s) s = re.sub(r"\s*,\s*", ", ", s) s = re.sub(r"(,\s*){2,}", ", ", s) return s.strip().strip(",").strip()