Jonah
332 lines
12 KiB
Python
332 lines
12 KiB
Python
"""ShipEngine API client — USPS label creation.
|
|
|
|
Authentication uses a simple API key header (no OAuth / token refresh).
|
|
Keys prefixed with TEST_ create non-mailable test labels with no postage
|
|
charged — same endpoint, same code path.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
from dataclasses import dataclass
|
|
|
|
import requests
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
_BASE_URL = "https://api.shipengine.com"
|
|
|
|
# Our internal package codes → ShipEngine package codes
|
|
_PACKAGE_CODES: dict[str, str] = {
|
|
"package": "package",
|
|
"large_envelope": "large_envelope_or_flat",
|
|
"letter": "letter",
|
|
"thick_envelope": "thick_envelope",
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class LabelResult:
|
|
label_id: str
|
|
tracking_number: str
|
|
label_pdf_path: str
|
|
|
|
|
|
class ShipEngineClient:
|
|
"""Thin wrapper around the ShipEngine REST API."""
|
|
|
|
def __init__(self, api_key: str) -> None:
|
|
self._key = api_key
|
|
self._headers = {
|
|
"API-Key": api_key,
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
# ── Address validation ─────────────────────────────────────────────────
|
|
|
|
def validate_address(self, addr: dict) -> dict:
|
|
"""Validate and normalise a structured address dict.
|
|
|
|
Input keys: name, company_name, address_line1, address_line2,
|
|
city, state_province, postal_code
|
|
Returns the same key set with corrected values, plus a
|
|
'validation_status' key ("verified" | "warning" | "unverified" | "error").
|
|
"""
|
|
payload = [{
|
|
"name": addr.get("name", ""),
|
|
"company_name": addr.get("company_name", ""),
|
|
"address_line1": addr.get("address_line1", ""),
|
|
"address_line2": addr.get("address_line2", ""),
|
|
"city_locality": addr.get("city", ""),
|
|
"state_province": addr.get("state_province", ""),
|
|
"postal_code": addr.get("postal_code", ""),
|
|
"country_code": "US",
|
|
}]
|
|
resp = requests.post(
|
|
f"{_BASE_URL}/v1/addresses/validate",
|
|
json=payload,
|
|
headers=self._headers,
|
|
timeout=15,
|
|
)
|
|
if not resp.ok:
|
|
raise RuntimeError(_format_error(resp, "Address validation"))
|
|
|
|
result = resp.json()
|
|
if not result:
|
|
return addr
|
|
|
|
item = result[0]
|
|
status = item.get("status", "unverified")
|
|
matched = item.get("matched_address") or {}
|
|
|
|
return {
|
|
"name": matched.get("name", addr.get("name", "")),
|
|
"company_name": matched.get("company_name", addr.get("company_name", "")),
|
|
"address_line1": matched.get("address_line1", addr.get("address_line1", "")),
|
|
"address_line2": matched.get("address_line2", addr.get("address_line2", "")),
|
|
"city": matched.get("city_locality", addr.get("city", "")),
|
|
"state_province": matched.get("state_province", addr.get("state_province", "")),
|
|
"postal_code": matched.get("postal_code", addr.get("postal_code", "")),
|
|
"validation_status": status,
|
|
}
|
|
|
|
# ── Rate fetch ─────────────────────────────────────────────────────────
|
|
|
|
def get_rate(
|
|
self,
|
|
from_zip: str,
|
|
to_zip: str,
|
|
weight_oz: int,
|
|
service_code: str,
|
|
package_code: str,
|
|
dim_l: float = 0,
|
|
dim_w: float = 0,
|
|
dim_h: float = 0,
|
|
) -> float | None:
|
|
"""Return total shipment cost in USD, or None on any failure."""
|
|
package: dict = {
|
|
"weight": {"value": max(1, weight_oz), "unit": "ounce"},
|
|
}
|
|
if dim_l and dim_w and dim_h:
|
|
package["dimensions"] = {
|
|
"length": dim_l, "width": dim_w, "height": dim_h,
|
|
"unit": "inch",
|
|
}
|
|
if package_code and package_code != "package":
|
|
package["package_code"] = _PACKAGE_CODES.get(package_code, package_code)
|
|
|
|
payload = {
|
|
"rate_options": {"service_codes": [service_code]},
|
|
"shipment": {
|
|
"service_code": service_code,
|
|
"ship_from": {"postal_code": from_zip, "country_code": "US"},
|
|
"ship_to": {"postal_code": to_zip, "country_code": "US"},
|
|
"packages": [package],
|
|
},
|
|
}
|
|
try:
|
|
resp = requests.post(
|
|
f"{_BASE_URL}/v1/rates",
|
|
json=payload,
|
|
headers=self._headers,
|
|
timeout=12,
|
|
)
|
|
if not resp.ok:
|
|
return None
|
|
data = resp.json()
|
|
rate_list = (data.get("rate_response") or {}).get("rates") or []
|
|
if not rate_list:
|
|
return None
|
|
total = rate_list[0].get("shipping_amount", {}).get("amount")
|
|
return float(total) if total is not None else None
|
|
except Exception as exc:
|
|
log.debug("Rate fetch failed: %s", exc)
|
|
return None
|
|
|
|
# ── Label creation ─────────────────────────────────────────────────────
|
|
|
|
def create_label(self, data: dict, parsed_addr: dict) -> LabelResult:
|
|
"""Create a USPS label and save the PDF locally.
|
|
|
|
`data` is the dict from _UspsPanel._on_create().
|
|
`parsed_addr` is a structured address dict (from parse_address_text
|
|
+ validate_address).
|
|
"""
|
|
pkg_code = _PACKAGE_CODES.get(data["package_code"], data["package_code"])
|
|
|
|
package: dict = {
|
|
"weight": {"value": max(1, data["weight_oz"]), "unit": "ounce"},
|
|
"package_code": pkg_code,
|
|
}
|
|
if data.get("dim_l") and data.get("dim_w") and data.get("dim_h"):
|
|
package["dimensions"] = {
|
|
"length": data["dim_l"],
|
|
"width": data["dim_w"],
|
|
"height": data["dim_h"],
|
|
"unit": "inch",
|
|
}
|
|
if data.get("print_message"):
|
|
package["label_messages"] = {"reference1": data["print_message"][:60]}
|
|
|
|
ship_to: dict = {
|
|
"name": parsed_addr.get("name", ""),
|
|
"company_name": parsed_addr.get("company_name", ""),
|
|
"address_line1": parsed_addr.get("address_line1", ""),
|
|
"address_line2": parsed_addr.get("address_line2", ""),
|
|
"city_locality": parsed_addr.get("city", ""),
|
|
"state_province": parsed_addr.get("state_province", ""),
|
|
"postal_code": parsed_addr.get("postal_code", ""),
|
|
"country_code": "US",
|
|
}
|
|
if data.get("email"):
|
|
ship_to["email"] = data["email"]
|
|
|
|
payload = {
|
|
"label_format": "pdf",
|
|
"label_layout": "4x6",
|
|
"shipment": {
|
|
"service_code": data["service_code"],
|
|
"ship_from": self._shipper,
|
|
"ship_to": ship_to,
|
|
"packages": [package],
|
|
},
|
|
}
|
|
|
|
resp = requests.post(
|
|
f"{_BASE_URL}/v1/labels",
|
|
json=payload,
|
|
headers=self._headers,
|
|
timeout=30,
|
|
)
|
|
if not resp.ok:
|
|
raise RuntimeError(_format_error(resp, "Label creation"))
|
|
|
|
result = resp.json()
|
|
label_id = result.get("label_id", "")
|
|
tracking = result.get("tracking_number", "")
|
|
pdf_url = (result.get("label_download") or {}).get("pdf", "")
|
|
|
|
if not pdf_url:
|
|
raise RuntimeError(
|
|
f"Label created (tracking: {tracking}) but ShipEngine returned no PDF URL."
|
|
)
|
|
|
|
pdf_resp = requests.get(pdf_url, timeout=20)
|
|
pdf_resp.raise_for_status()
|
|
|
|
from datetime import datetime
|
|
from .paths import appdata_dir
|
|
labels_dir = appdata_dir() / "labels"
|
|
labels_dir.mkdir(exist_ok=True)
|
|
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
label_path = labels_dir / f"USPS_{tracking or 'label'}_{stamp}.pdf"
|
|
label_path.write_bytes(pdf_resp.content)
|
|
|
|
log.info("USPS label created: tracking=%s label=%s", tracking, label_path)
|
|
return LabelResult(
|
|
label_id = label_id,
|
|
tracking_number = tracking,
|
|
label_pdf_path = str(label_path),
|
|
)
|
|
|
|
def set_shipper(self, cfg) -> None:
|
|
"""Build the ship_from block from config and cache it."""
|
|
self._shipper = {
|
|
"name": cfg.usps_shipper_name,
|
|
"company_name": cfg.usps_shipper_company,
|
|
"phone": cfg.usps_shipper_phone,
|
|
"address_line1": cfg.usps_shipper_address1,
|
|
"city_locality": cfg.usps_shipper_city,
|
|
"state_province": cfg.usps_shipper_state,
|
|
"postal_code": cfg.usps_shipper_zip,
|
|
"country_code": "US",
|
|
}
|
|
|
|
|
|
# ── Error formatting ────────────────────────────────────────────────────────
|
|
|
|
def _format_error(resp: requests.Response, context: str) -> str:
|
|
status = resp.status_code
|
|
try:
|
|
body = resp.json()
|
|
errors = body.get("errors") or []
|
|
msgs = [e.get("message", "") for e in errors if e.get("message")]
|
|
detail = " | ".join(msgs) if msgs else str(body)
|
|
except Exception:
|
|
detail = resp.text or "(no body)"
|
|
return f"{context} failed — ShipEngine HTTP {status}: {detail}"
|
|
|
|
|
|
# ── Freeform address parser ─────────────────────────────────────────────────
|
|
|
|
_CSZ_RE = re.compile(
|
|
r'^(.+?),?\s+([A-Za-z]{2})\s+(\d{5}(?:-\d{4})?)$'
|
|
)
|
|
_STREET_RE = re.compile(r'^\d')
|
|
|
|
|
|
def parse_address_text(text: str) -> dict:
|
|
"""Parse a freeform US address block into a structured dict.
|
|
|
|
Handles the common copy-paste format:
|
|
Name
|
|
[Company]
|
|
Street Address [Apt/Suite]
|
|
City, ST ZIP
|
|
"""
|
|
lines = [l.strip() for l in text.strip().splitlines() if l.strip()]
|
|
if not lines:
|
|
return {}
|
|
|
|
# Find the city/state/zip line (search from bottom)
|
|
csz_idx = None
|
|
city = state = zip_ = ""
|
|
for i in range(len(lines) - 1, -1, -1):
|
|
m = _CSZ_RE.match(lines[i])
|
|
if m:
|
|
csz_idx = i
|
|
city = m.group(1).strip()
|
|
state = m.group(2).upper()
|
|
zip_ = m.group(3)
|
|
break
|
|
|
|
if csz_idx is None:
|
|
# Can't find city/state/zip — return what we have
|
|
return {"address_line1": lines[0]}
|
|
|
|
before = lines[:csz_idx]
|
|
name = company = addr1 = addr2 = ""
|
|
|
|
if len(before) == 0:
|
|
pass
|
|
elif len(before) == 1:
|
|
addr1 = before[0]
|
|
elif len(before) == 2:
|
|
name = before[0]
|
|
addr1 = before[1]
|
|
elif len(before) == 3:
|
|
name = before[0]
|
|
# Middle line: company or address depending on whether last line is street
|
|
if _STREET_RE.match(before[2]):
|
|
company = before[1]
|
|
addr1 = before[2]
|
|
else:
|
|
addr1 = before[1]
|
|
addr2 = before[2]
|
|
else:
|
|
name = before[0]
|
|
company = before[1]
|
|
addr1 = before[2]
|
|
addr2 = before[3]
|
|
|
|
return {
|
|
"name": name,
|
|
"company_name": company,
|
|
"address_line1": addr1,
|
|
"address_line2": addr2,
|
|
"city": city,
|
|
"state_province": state,
|
|
"postal_code": zip_,
|
|
}
|