430 lines
16 KiB
Python
430 lines
16 KiB
Python
"""
|
|
Emergency "Send to ShipStation" - for the rare case a ticket needs to
|
|
skip the normal daily batch and get to ShipStation right away.
|
|
|
|
Two independent paths, on purpose (per "in case the API goes down we
|
|
still have an option for CSV uploads"):
|
|
|
|
- export_order_to_shipstation_csv(): writes rows matching your real
|
|
upload template exactly (columns confirmed against OAK.csv) - one
|
|
row per line item, customer/address info repeated on each row.
|
|
- send_order_to_shipstation_api(): calls ShipStation's V2 API
|
|
directly (POST /v2/shipments with create_sales_order: true) to
|
|
create the order without leaving the app. Needs a Warehouse ID per
|
|
company (SHIPSTATION_SIGNIFY_WAREHOUSE_ID /
|
|
SHIPSTATION_OAKSTREET_WAREHOUSE_ID) - ShipStation requires knowing
|
|
where the package ships FROM, either via a warehouse or an explicit
|
|
ship_from address; only the warehouse path is wired up today.
|
|
|
|
IMPORTANT - please verify the first real send: ShipStation's docs say
|
|
automation rules apply tags to orders "when they import based on any
|
|
criteria you set" - meaning your 90 box-packing rules should fire
|
|
automatically off the SKU/item data here, same as your CSV import, with
|
|
no manual tagging needed. That's the best read of the docs, but it's
|
|
not something I can verify without your actual account, so treat the
|
|
first live send (API or CSV) as a test: confirm ShipStation packs and
|
|
prices it the way a normal order would before trusting it in a real
|
|
emergency.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import datetime as dt
|
|
from pathlib import Path
|
|
from typing import List
|
|
|
|
import requests
|
|
|
|
from app import config
|
|
from app.models import Order
|
|
|
|
API_BASE = "https://api.shipstation.com/v2"
|
|
REQUEST_TIMEOUT_SECONDS = 30
|
|
|
|
# Column order confirmed against the real ShipStation upload template
|
|
# (OAK.csv) - "COPY ME ALREADY" is a spreadsheet-only helper column and
|
|
# is intentionally left out here.
|
|
CSV_COLUMNS = [
|
|
"Custom field (Email Address)",
|
|
"Summary",
|
|
"Custom field (Name)",
|
|
"Custom field (NPI Number)",
|
|
"Custom field (Address 1)",
|
|
"Custom field (Address 2)",
|
|
"Custom field (City)",
|
|
"Custom field (State)",
|
|
"Custom field (Zip Code)",
|
|
"Issue key",
|
|
"Issue id",
|
|
"Status",
|
|
"deliverables",
|
|
"SKU",
|
|
"Item Name",
|
|
"Quantity",
|
|
]
|
|
|
|
|
|
class ShipStationSendError(Exception):
|
|
"""Raised for any emergency-send failure, with a message safe to show in the UI."""
|
|
|
|
|
|
def _deliverable_text(sku: str, item_name: str) -> str:
|
|
return f"{sku}: {item_name}" if item_name else sku
|
|
|
|
|
|
def build_csv_rows(order: Order) -> List[List[str]]:
|
|
"""One row per line item, matching the real template column-for-column.
|
|
Falls back to a single row (blank SKU/Item Name) if there are no line
|
|
items yet, so the customer/address info is still exportable."""
|
|
info = order.shipping_info or {}
|
|
line_items = order.line_items or [{"sku": s, "item_name": ""} for s in (order.skus or [])]
|
|
if not line_items:
|
|
line_items = [{"sku": "", "item_name": ""}]
|
|
|
|
issue_id = ""
|
|
if isinstance(order.raw_data, dict):
|
|
issue_id = order.raw_data.get("id", "")
|
|
|
|
rows = []
|
|
for item in line_items:
|
|
rows.append(
|
|
[
|
|
info.get("email", ""),
|
|
order.summary,
|
|
info.get("name", ""),
|
|
info.get("npi", ""),
|
|
info.get("address1", ""),
|
|
info.get("address2", ""),
|
|
info.get("city", ""),
|
|
info.get("state", ""),
|
|
info.get("zip", ""),
|
|
order.ticket_number or order.external_id,
|
|
issue_id,
|
|
order.status,
|
|
_deliverable_text(item.get("sku", ""), item.get("item_name", "")),
|
|
item.get("sku", ""),
|
|
item.get("item_name", ""),
|
|
1, # quantity - not tracked per-item today, defaults to 1 per line
|
|
]
|
|
)
|
|
return rows
|
|
|
|
|
|
def export_order_to_shipstation_csv(orders: List[Order], filepath: str) -> int:
|
|
"""Write one or more orders to a CSV in the real upload template shape.
|
|
Returns the number of line-item rows written (not the number of orders,
|
|
since one order can produce several rows)."""
|
|
path = Path(filepath)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
row_count = 0
|
|
with path.open("w", newline="", encoding="utf-8") as f:
|
|
writer = csv.writer(f)
|
|
writer.writerow(CSV_COLUMNS)
|
|
for order in orders:
|
|
for row in build_csv_rows(order):
|
|
writer.writerow(row)
|
|
row_count += 1
|
|
|
|
return row_count
|
|
|
|
|
|
def _store_id_for_company(company: str) -> str:
|
|
settings = config.load_settings()
|
|
if company == "Signify Health":
|
|
return settings.get("SHIPSTATION_SIGNIFY_STORE_ID", "")
|
|
if company == "Oak Street Health":
|
|
return settings.get("SHIPSTATION_OAKSTREET_STORE_ID", "")
|
|
return ""
|
|
|
|
|
|
def _warehouse_id_for_company(company: str) -> str:
|
|
settings = config.load_settings()
|
|
if company == "Signify Health":
|
|
return settings.get("SHIPSTATION_SIGNIFY_WAREHOUSE_ID", "")
|
|
if company == "Oak Street Health":
|
|
return settings.get("SHIPSTATION_OAKSTREET_WAREHOUSE_ID", "")
|
|
return ""
|
|
|
|
|
|
def send_order_to_shipstation_api(order: Order) -> dict:
|
|
"""
|
|
Creates the order directly in ShipStation via POST /v2/shipments with
|
|
create_sales_order: true, so it lands in the Orders tab and (per
|
|
ShipStation's own docs on import automation) should pick up your
|
|
box-packing rules the same as a CSV-imported order. Returns the
|
|
created shipment's JSON on success.
|
|
"""
|
|
settings = config.load_settings()
|
|
api_key = settings.get("SHIPSTATION_API_KEY", "")
|
|
if not api_key:
|
|
raise ShipStationSendError("ShipStation API Key is not set. Add it in Settings.")
|
|
|
|
store_id = _store_id_for_company(order.company)
|
|
if not store_id:
|
|
raise ShipStationSendError(
|
|
f"No ShipStation Store ID configured for '{order.company}'. "
|
|
"Add it in Settings under ShipStation."
|
|
)
|
|
|
|
# ShipStation needs to know where the package ships FROM - either a
|
|
# configured warehouse, or an explicit ship_from address. Only the
|
|
# warehouse path is wired up today; fail clearly here rather than
|
|
# sending an incomplete request and getting ShipStation's less
|
|
# actionable "ship_from is required when warehouse_id is not present".
|
|
warehouse_id = _warehouse_id_for_company(order.company)
|
|
if not warehouse_id:
|
|
raise ShipStationSendError(
|
|
f"No ShipStation Warehouse ID configured for '{order.company}'. "
|
|
"Add it in Settings under ShipStation - find it in ShipStation under "
|
|
"Settings > Shipping > Warehouses/Ship From Locations."
|
|
)
|
|
|
|
info = order.shipping_info or {}
|
|
if not (info.get("address1") and info.get("city") and info.get("state") and info.get("zip")):
|
|
raise ShipStationSendError(
|
|
"This ticket is missing address information (address/city/state/zip) - "
|
|
"can't create a shipment without a destination."
|
|
)
|
|
|
|
line_items = order.line_items or [{"sku": s, "item_name": ""} for s in (order.skus or [])]
|
|
if not line_items:
|
|
raise ShipStationSendError("This ticket has no SKUs/line items to send.")
|
|
|
|
ticket_number = order.ticket_number or order.external_id
|
|
|
|
payload = {
|
|
"shipments": [
|
|
{
|
|
"create_sales_order": True,
|
|
"store_id": store_id,
|
|
"warehouse_id": warehouse_id,
|
|
"external_shipment_id": ticket_number,
|
|
"shipment_number": ticket_number,
|
|
"ship_to": {
|
|
"name": info.get("name", ""),
|
|
"phone": info.get("phone", ""),
|
|
"company_name": order.company,
|
|
"address_line1": info.get("address1", ""),
|
|
"address_line2": info.get("address2", "") or None,
|
|
"city_locality": info.get("city", ""),
|
|
"state_province": info.get("state", ""),
|
|
"postal_code": info.get("zip", ""),
|
|
"country_code": "US",
|
|
},
|
|
"items": [
|
|
{
|
|
"sku": item.get("sku", ""),
|
|
"name": item.get("item_name") or item.get("sku", ""),
|
|
"quantity": 1,
|
|
}
|
|
for item in line_items
|
|
],
|
|
}
|
|
]
|
|
}
|
|
|
|
try:
|
|
response = requests.post(
|
|
f"{API_BASE}/shipments",
|
|
json=payload,
|
|
headers={"API-Key": api_key, "Accept": "application/json"},
|
|
timeout=REQUEST_TIMEOUT_SECONDS,
|
|
)
|
|
except requests.RequestException as exc:
|
|
raise ShipStationSendError(f"Could not reach ShipStation: {exc}") from exc
|
|
|
|
if response.status_code == 401:
|
|
raise ShipStationSendError("ShipStation rejected the API key (401). Check it in Settings.")
|
|
if not response.ok:
|
|
raise ShipStationSendError(
|
|
f"ShipStation returned an error ({response.status_code}): {response.text[:400]}"
|
|
)
|
|
|
|
try:
|
|
data = response.json()
|
|
except ValueError as exc:
|
|
raise ShipStationSendError("ShipStation returned a response that wasn't valid JSON.") from exc
|
|
|
|
shipments = data.get("shipments", [])
|
|
if not shipments:
|
|
raise ShipStationSendError(f"ShipStation didn't return a created shipment: {data}")
|
|
|
|
result = shipments[0]
|
|
if result.get("errors"):
|
|
raise ShipStationSendError(f"ShipStation reported errors: {result['errors']}")
|
|
|
|
return result
|
|
|
|
|
|
# --- Emailed return labels (SH007 / OK012) --------------------------------
|
|
#
|
|
# A genuinely different shape from the outbound send above: for a return
|
|
# label, ship_from is the CUSTOMER and ship_to is YOUR warehouse/return
|
|
# center (confirmed against ShipStation's own return-labels docs - this is
|
|
# reversed from every other label this app creates). This also doesn't go
|
|
# through create_sales_order + automation, since there's no outbound side
|
|
# to it - carrier/service are specified directly.
|
|
#
|
|
# After creation, this app does NOT email the label - per your workflow,
|
|
# that's done from ShipStation itself (Returns tab -> Other Actions ->
|
|
# Send Return Label) so it goes out through your branded email template.
|
|
|
|
VALID_CHARGE_EVENTS = {"on_creation", "on_carrier_acceptance", "carrier_default"}
|
|
|
|
|
|
def _return_config_prefix(order: Order) -> str:
|
|
"""
|
|
Normally keyed off company, but RMD-prefixed SKUs (RubiconMD, an Oak
|
|
Street subsidiary) use their own UPS account and return address even
|
|
though the ticket is still classified as Oak Street Health for every
|
|
other purpose - so this checks the ticket's actual SKUs, not just
|
|
order.company. (The IE prefix seen in Signify's catalog is deprecated
|
|
and deliberately not handled here - not expected on current tickets.)
|
|
"""
|
|
if any((sku or "").strip().upper().startswith("RMD") for sku in (order.skus or [])):
|
|
return "RUBICONMD_RETURN_"
|
|
if order.company == "Signify Health":
|
|
return "SIGNIFY_RETURN_"
|
|
return "OAKSTREET_RETURN_"
|
|
|
|
|
|
def _return_address_for_order(order: Order) -> dict:
|
|
settings = config.load_settings()
|
|
prefix = _return_config_prefix(order)
|
|
return {
|
|
"name": settings.get(f"{prefix}NAME", ""),
|
|
"phone": settings.get(f"{prefix}PHONE", ""),
|
|
"address_line1": settings.get(f"{prefix}ADDRESS1", ""),
|
|
"address_line2": settings.get(f"{prefix}ADDRESS2", "") or None,
|
|
"city_locality": settings.get(f"{prefix}CITY", ""),
|
|
"state_province": settings.get(f"{prefix}STATE", ""),
|
|
"postal_code": settings.get(f"{prefix}ZIP", ""),
|
|
"country_code": "US",
|
|
}
|
|
|
|
|
|
def _return_carrier_for_order(order: Order) -> tuple[str, str]:
|
|
"""Each shipping account (Signify, Oak Street, and RubiconMD as its own
|
|
subsidiary account) has its own UPS carrier_id, even where the physical
|
|
warehouse is shared."""
|
|
settings = config.load_settings()
|
|
prefix = _return_config_prefix(order)
|
|
return settings.get(f"{prefix}CARRIER_ID", ""), settings.get(f"{prefix}SERVICE_CODE", "")
|
|
|
|
|
|
def _build_return_package(weight_oz: float, length: float, width: float, height: float) -> dict:
|
|
package = {"weight": {"value": weight_oz, "unit": "ounce"}}
|
|
if length > 0 and width > 0 and height > 0:
|
|
package["dimensions"] = {
|
|
"length": length,
|
|
"width": width,
|
|
"height": height,
|
|
"unit": "inch",
|
|
}
|
|
return package
|
|
|
|
|
|
def create_return_label(order: Order, packages: List[dict], charge_event: str | None = None) -> dict:
|
|
"""
|
|
packages: [{"weight_oz": 1.0, "length": 0, "width": 0, "height": 0}, ...]
|
|
- one entry per physical box the customer will use, matching your
|
|
current "dummy ticket" approach but with real, per-box control
|
|
instead of always defaulting to 1x1x1/1oz.
|
|
"""
|
|
settings = config.load_settings()
|
|
api_key = settings.get("SHIPSTATION_API_KEY", "")
|
|
if not api_key:
|
|
raise ShipStationSendError("ShipStation API Key is not set. Add it in Settings.")
|
|
|
|
carrier_id, service_code = _return_carrier_for_order(order)
|
|
if not carrier_id or not service_code:
|
|
raise ShipStationSendError(
|
|
f"No return-label Carrier ID/Service Code configured for '{order.company}'. "
|
|
"Add them in Settings under Return Labels."
|
|
)
|
|
|
|
return_address = _return_address_for_order(order)
|
|
if not (
|
|
return_address["address_line1"]
|
|
and return_address["city_locality"]
|
|
and return_address["state_province"]
|
|
and return_address["postal_code"]
|
|
):
|
|
raise ShipStationSendError(
|
|
f"No return warehouse address configured for '{order.company}'. "
|
|
"Add it in Settings under Return Labels."
|
|
)
|
|
|
|
info = order.shipping_info or {}
|
|
if not (info.get("address1") and info.get("city") and info.get("state") and info.get("zip")):
|
|
raise ShipStationSendError(
|
|
"This ticket is missing the customer's address information - "
|
|
"can't create a return label without knowing where it ships from."
|
|
)
|
|
|
|
if not packages:
|
|
raise ShipStationSendError("At least one package is required.")
|
|
|
|
resolved_charge_event = charge_event or settings.get(
|
|
"SHIPSTATION_RETURN_CHARGE_EVENT", config.DEFAULT_RETURN_CHARGE_EVENT
|
|
)
|
|
if resolved_charge_event not in VALID_CHARGE_EVENTS:
|
|
raise ShipStationSendError(
|
|
f"charge_event must be one of {sorted(VALID_CHARGE_EVENTS)}, got "
|
|
f"'{resolved_charge_event}'."
|
|
)
|
|
|
|
payload = {
|
|
"is_return_label": True,
|
|
"charge_event": resolved_charge_event,
|
|
"shipment": {
|
|
"carrier_id": carrier_id,
|
|
"service_code": service_code,
|
|
"ship_to": return_address,
|
|
"ship_from": {
|
|
"name": info.get("name", ""),
|
|
"phone": info.get("phone", ""),
|
|
"address_line1": info.get("address1", ""),
|
|
"address_line2": info.get("address2", "") or None,
|
|
"city_locality": info.get("city", ""),
|
|
"state_province": info.get("state", ""),
|
|
"postal_code": info.get("zip", ""),
|
|
"country_code": "US",
|
|
},
|
|
"packages": [
|
|
_build_return_package(
|
|
p.get("weight_oz", 1.0), p.get("length", 0), p.get("width", 0), p.get("height", 0)
|
|
)
|
|
for p in packages
|
|
],
|
|
},
|
|
}
|
|
|
|
try:
|
|
response = requests.post(
|
|
f"{API_BASE}/labels",
|
|
json=payload,
|
|
headers={"API-Key": api_key, "Accept": "application/json"},
|
|
timeout=REQUEST_TIMEOUT_SECONDS,
|
|
)
|
|
except requests.RequestException as exc:
|
|
raise ShipStationSendError(f"Could not reach ShipStation: {exc}") from exc
|
|
|
|
if response.status_code == 401:
|
|
raise ShipStationSendError("ShipStation rejected the API key (401). Check it in Settings.")
|
|
if not response.ok:
|
|
raise ShipStationSendError(
|
|
f"ShipStation returned an error ({response.status_code}): {response.text[:400]}"
|
|
)
|
|
|
|
try:
|
|
data = response.json()
|
|
except ValueError as exc:
|
|
raise ShipStationSendError("ShipStation returned a response that wasn't valid JSON.") from exc
|
|
|
|
if data.get("errors"):
|
|
raise ShipStationSendError(f"ShipStation reported errors: {data['errors']}")
|
|
|
|
return data
|