madmin/Order-Manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

More Shipstation integrations

Browse Source
This commit is contained in:
madmin
2026-08-31 10:08:30 -05:00
parent ae74f78520
commit 4c172c1fd5
10 changed files with 527 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/base.py
+3
View File
@@ -22,6 +22,9 @@ class NormalizedOrder(TypedDict):
ticket_number: Optional[str]
company: str
skus: List[str]
line_items: List[dict]
shipping_info: dict
creator: Optional[str]
tracking_numbers: List[dict]
summary: str
status: str
app/services/jira_service.py
+56 -5
View File
@@ -60,6 +60,21 @@ class JiraService(OrderService):
settings["COMPANY_SKU_MAP"] or config.DEFAULT_COMPANY_SKU_MAP
)
# Recipient/shipping contact fields - shared across both companies'
# tickets (unlike the deliverable fields, these aren't split by
# company). Only used by the emergency "Send to ShipStation" action.
self.contact_field_ids = {
"name": settings["JIRA_FIELD_NAME"].strip(),
"phone": settings["JIRA_FIELD_PHONE"].strip(),
"email": settings["JIRA_FIELD_EMAIL"].strip(),
"address1": settings["JIRA_FIELD_ADDRESS1"].strip(),
"address2": settings["JIRA_FIELD_ADDRESS2"].strip(),
"city": settings["JIRA_FIELD_CITY"].strip(),
"state": settings["JIRA_FIELD_STATE"].strip(),
"zip": settings["JIRA_FIELD_ZIP"].strip(),
"npi": settings["JIRA_FIELD_NPI"].strip(),
}
# Statuses at which we stop re-checking a ticket for changes -
# see fetch_orders() for why we re-check at all.
self.terminal_statuses = parse_status_list(
@@ -129,10 +144,14 @@ class JiraService(OrderService):
auth = (self.email, self.api_token)
headers = {"Accept": "application/json"}
# Always pull summary/status/created, plus every configured deliverable field.
fields = "summary,status,created"
# Always pull summary/status/created/creator, plus every configured
# deliverable field and contact field.
fields = "summary,status,created,creator"
if self.all_sku_field_ids:
fields += "," + ",".join(self.all_sku_field_ids)
contact_field_ids = [v for v in self.contact_field_ids.values() if v]
if contact_field_ids:
fields += "," + ",".join(contact_field_ids)
all_issues: List[dict] = []
start_at = 0
@@ -221,26 +240,53 @@ class JiraService(OrderService):
return []
def _extract_skus_and_texts(self, fields: dict) -> Tuple[List[str], List[str]]:
def _extract_skus_and_texts(self, fields: dict) -> Tuple[List[str], List[str], List[dict]]:
"""
Scan every configured deliverable field and return:
- skus: just the code part (e.g. "SH011"), deduped, order preserved
- texts: the full "CODE: description" strings, for use as a
fallback summary when the ticket's actual Summary is blank
(which is the normal case for these order tickets)
- line_items: [{"sku": "SH011", "item_name": "Shipping - Return
Label..."}], one per deliverable - this is what the ShipStation
CSV template's SKU/Item Name columns come from
"""
skus: List[str] = []
texts: List[str] = []
line_items: List[dict] = []
for field_id in self.all_sku_field_ids:
for raw_value in self._raw_values_from_field(fields, field_id):
texts.append(raw_value)
match = DELIVERABLE_CODE_PATTERN.match(raw_value)
code = match.group(1) if match else raw_value
item_name = match.group(2).strip() if match else ""
if code not in skus:
skus.append(code)
line_items.append({"sku": code, "item_name": item_name})
return skus, texts
return skus, texts, line_items
def _extract_single_value(self, fields: dict, field_id: str) -> str:
"""One value from a contact field (name/phone/address/etc.), which -
same as the deliverable fields - can come back as plain text, a
single-select dict, or (rarely) a list. Reuses the same flexible
extraction and just takes the first value."""
if not field_id:
return ""
values = self._raw_values_from_field(fields, field_id)
return values[0] if values else ""
def _extract_shipping_info(self, fields: dict) -> dict:
return {
key: self._extract_single_value(fields, field_id)
for key, field_id in self.contact_field_ids.items()
}
@staticmethod
def _extract_creator(fields: dict) -> str:
creator = fields.get("creator") or {}
return creator.get("displayName") or creator.get("emailAddress") or ""
def _to_normalized_order(self, issue: dict) -> NormalizedOrder:
fields = issue.get("fields", {})
@@ -254,8 +300,10 @@ class JiraService(OrderService):
created_at = None
ticket_number = issue.get("key", "")
skus, deliverable_texts = self._extract_skus_and_texts(fields)
skus, deliverable_texts, line_items = self._extract_skus_and_texts(fields)
company = resolve_company_for_skus(skus, self.sku_map)
shipping_info = self._extract_shipping_info(fields)
creator = self._extract_creator(fields)
# These order tickets typically leave the JIRA Summary field
# blank - fall back to the deliverables so there's still
@@ -268,6 +316,9 @@ class JiraService(OrderService):
ticket_number=ticket_number,
company=company,
skus=skus,
line_items=line_items,
shipping_info=shipping_info,
creator=creator,
tracking_numbers=[],
summary=summary,
status=(fields.get("status") or {}).get("name", ""),
app/services/shipstation_send.py
+230
View File
@@ -0,0 +1,230 @@
"""
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.
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 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."
)
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,
"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