""" ShipStation tracking-number puller (API V2). ShipStation is used for exactly one thing here: generating shipping labels for JIRA tickets. It's not an independent order source, so this doesn't produce its own order rows - it produces tracking numbers keyed by ticket number, which app.workers merges directly onto the matching JIRA-sourced Order row. See app/tracking.py for the "what JIRA status should this become" logic. How this maps to the real API (confirmed against a live payload, not guessed): - GET /v2/labels gives tracking_number + is_return_label directly - exactly what's needed to tell "Waiting For Return" apart from "Device Return Not Needed". - A label only carries a shipment_id, not the ticket number, so for each label we fetch its shipment via GET /v2/shipments/{id}. A real shipment payload showed the ticket number in BOTH shipment_number and external_shipment_id (e.g. both were "AR-160269") - we check both, then fall back to scanning the whole payload with TICKET_NUMBER_REGEX as a last resort. - Shipments don't reliably carry store_id (a still-"pending" shipment has none), which is fine - company comes from the JIRA side, this never needs to know it. """ from __future__ import annotations import datetime as dt import json import re from typing import Dict, List, Optional import requests from app import config from app.services.base import OrderService, NormalizedOrder from app.tracking import suggest_jira_status API_BASE = "https://api.shipstation.com/v2" PAGE_SIZE = 100 REQUEST_TIMEOUT_SECONDS = 30 class ShipStationServiceError(Exception): """Raised for any ShipStation fetch failure, with a message safe to show in the UI.""" class ShipStationService(OrderService): name = "shipstation" def __init__(self) -> None: settings = config.load_settings() self.api_key = settings["SHIPSTATION_API_KEY"] self.ticket_pattern = re.compile( settings["TICKET_NUMBER_REGEX"] or config.DEFAULT_TICKET_NUMBER_REGEX ) self._headers = {"API-Key": self.api_key, "Accept": "application/json"} self.unmatched_labels: List[dict] = [] def is_configured(self) -> bool: return bool(self.api_key) def fetch_orders(self) -> List[NormalizedOrder]: if not self.is_configured(): raise ShipStationServiceError( "ShipStation is not configured yet. Open Settings and fill in the API Key." ) labels = self._fetch_todays_labels() # Only labels that actually produced a usable tracking number matter. usable_labels = [ label for label in labels if not label.get("voided") and label.get("tracking_number") ] # One shipment lookup per unique shipment_id referenced, not per # label (an outgoing + return label pair share the same shipment). shipment_ids = { label["shipment_id"] for label in usable_labels if label.get("shipment_id") } shipments_by_id = {sid: self._fetch_shipment(sid) for sid in shipment_ids} tracking_by_ticket: Dict[str, List[dict]] = {} raw_by_ticket: Dict[str, dict] = {} self.unmatched_labels = [] # labels we couldn't tie to a ticket number for label in usable_labels: shipment = shipments_by_id.get(label.get("shipment_id")) ticket_number = self._extract_ticket_number(shipment) if shipment else None if not ticket_number: self.unmatched_labels.append(label) continue tracking_by_ticket.setdefault(ticket_number, []).append( { "number": label.get("tracking_number"), "carrier": label.get("carrier_code"), "is_return": bool(label.get("is_return_label")), } ) raw_by_ticket.setdefault(ticket_number, {"labels": [], "shipment": shipment}) raw_by_ticket[ticket_number]["labels"].append(label) return [ self._to_normalized_order( ticket_number, tracking_by_ticket[ticket_number], raw_by_ticket[ticket_number] ) for ticket_number in tracking_by_ticket ] # -- internals ----------------------------------------------------- def _fetch_todays_labels(self) -> List[dict]: today_start = dt.datetime.combine(dt.date.today(), dt.time.min) today_end = today_start + dt.timedelta(days=1) all_labels: List[dict] = [] page = 1 while True: params = { "created_at_start": today_start.isoformat() + "Z", "created_at_end": today_end.isoformat() + "Z", "page": page, "page_size": PAGE_SIZE, "sort_by": "created_at", "sort_dir": "desc", } data = self._get("/labels", params) batch = data.get("labels", []) all_labels.extend(batch) total_pages = data.get("pages", 1) if page >= total_pages or not batch: break page += 1 return all_labels def _fetch_shipment(self, shipment_id: str) -> Optional[dict]: try: return self._get(f"/shipments/{shipment_id}") except ShipStationServiceError: # Don't let one bad lookup fail the whole import - this label's # tracking number just won't get matched to a ticket this run. return None def _get(self, path: str, params: Optional[dict] = None) -> dict: try: response = requests.get( f"{API_BASE}{path}", params=params, headers=self._headers, timeout=REQUEST_TIMEOUT_SECONDS, ) except requests.RequestException as exc: raise ShipStationServiceError(f"Could not reach ShipStation: {exc}") from exc if response.status_code == 401: raise ShipStationServiceError( "ShipStation rejected the API key (401). Check it in Settings." ) if not response.ok: raise ShipStationServiceError( f"ShipStation returned an error ({response.status_code}) for {path}: " f"{response.text[:300]}" ) try: return response.json() except ValueError as exc: raise ShipStationServiceError( f"ShipStation returned a response that wasn't valid JSON for {path}." ) from exc def _extract_ticket_number(self, shipment: dict) -> Optional[str]: # Confirmed against a real payload: both of these can carry the # ticket number directly. Check the more purpose-built field first. for field in ("shipment_number", "external_shipment_id"): value = shipment.get(field) if value and self.ticket_pattern.fullmatch(str(value).strip()): return str(value).strip() # Fall back to scanning the whole payload in case it shows up # somewhere else (a tag, a note, etc.) on a differently-shaped # shipment. try: blob = json.dumps(shipment) except (TypeError, ValueError): return None match = self.ticket_pattern.search(blob) return match.group(0) if match else None @staticmethod def _to_normalized_order( ticket_number: str, tracking_numbers: List[dict], raw: dict ) -> NormalizedOrder: suggested_status = suggest_jira_status(tracking_numbers) or "Tracking Pulled" numbers_display = ", ".join( f"{t['number']} ({'return' if t['is_return'] else 'outgoing'})" for t in tracking_numbers ) return NormalizedOrder( source="shipstation", external_id=ticket_number, ticket_number=ticket_number, company="", # not used - the JIRA row this merges onto already has one skus=[], tracking_numbers=tracking_numbers, summary=numbers_display, status=suggested_status, source_created_at=None, raw_data=raw, )