""" ShipStation tracking-number puller (API V2). Performance note: the previous version did one GET /v2/shipments/{id} call per unique shipment referenced by today's labels - an N+1 pattern that meant a busy day (100+ tickets) meant 100+ sequential network round-trips just for shipment lookups. That was almost certainly the actual bottleneck. This version instead: 1. Bulk-fetches ALL of today's labels and ALL recent shipments (a wider window - see SHIPMENT_LOOKBACK_DAYS - since a shipment can sit "pending" for a day or two before its label gets generated), each as a small number of paginated list calls. 2. Fetches both endpoints, and any additional pages either needs, concurrently using a PyQt QThreadPool - "multiple workers" pulling pages at once instead of one request waiting on the last. 3. Only once both full batches are in memory does it sort/correlate labels to shipments to ticket numbers - no more network calls interleaved with processing. 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. 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 import threading from typing import Callable, Dict, List, Optional import requests from PyQt6.QtCore import QRunnable, QThreadPool from app import config from app.companies import parse_mapping 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 # How far back to look for shipments, beyond just "today". Labels are # fetched strictly for today (that's the definition of "today's tracking # numbers"), but a shipment can be created a day or two before ShipStation's # automation actually gets around to producing its label - if that window # is too narrow, otherwise-matchable labels start showing up as unmatched. SHIPMENT_LOOKBACK_DAYS = 7 # Cap on simultaneous ShipStation requests. Kept modest to stay well clear # of any API rate limit rather than firing dozens of requests at once. MAX_CONCURRENT_REQUESTS = 5 def _resolve_shipping_method(service_code: str) -> str: """Maps a ShipStation service_code (e.g. 'ups_ground') to your team's JIRA-facing term (e.g. 'Ground') via SHIPPING_METHOD_LABELS. Falls back to a prettified version of the raw code for anything not mapped, so an unmapped service still shows something readable rather than nothing.""" if not service_code: return "" settings = config.load_settings() mapping = parse_mapping( settings.get("SHIPPING_METHOD_LABELS", "") or config.DEFAULT_SHIPPING_METHOD_LABELS ) if service_code in mapping: return mapping[service_code] return service_code.replace("_", " ").title() class ShipStationServiceError(Exception): """Raised for any ShipStation fetch failure, with a message safe to show in the UI.""" class _CallableTask(QRunnable): """Runs a zero-arg callable on a QThreadPool worker thread and stashes its result (or exception) into a shared, lock-protected dict/list.""" def __init__(self, key: str, fn: Callable[[], dict], results: dict, lock: threading.Lock, errors: list): super().__init__() self.key = key self.fn = fn self.results = results self.lock = lock self.errors = errors def run(self) -> None: try: value = self.fn() except Exception as exc: # noqa: BLE001 - surfaced to the caller via `errors` with self.lock: self.errors.append(exc) return with self.lock: self.results[self.key] = value 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, shipments = self._fetch_labels_and_shipments() usable_labels = [ label for label in labels if not label.get("voided") and label.get("tracking_number") ] shipments_by_id = {s["shipment_id"]: s for s in shipments if s.get("shipment_id")} # --- Sort/correlate now, entirely in memory, after both batches landed --- tracking_by_ticket: Dict[str, List[dict]] = {} raw_by_ticket: Dict[str, dict] = {} shipping_method_by_ticket: Dict[str, str] = {} self.unmatched_labels = [] 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) # Shipping Method reflects the OUTBOUND label specifically (matches # the Ship Sheet's usage) - only set from the first non-return label # seen per ticket, so a return label's service doesn't overwrite it. if not label.get("is_return_label") and ticket_number not in shipping_method_by_ticket: shipping_method_by_ticket[ticket_number] = _resolve_shipping_method( label.get("service_code", "") ) return [ self._to_normalized_order( ticket_number, tracking_by_ticket[ticket_number], raw_by_ticket[ticket_number], shipping_method_by_ticket.get(ticket_number, ""), ) for ticket_number in tracking_by_ticket ] # -- batch fetching (concurrent) ------------------------------------ def _fetch_labels_and_shipments(self) -> tuple[List[dict], List[dict]]: today_start = dt.datetime.combine(dt.date.today(), dt.time.min) today_end = today_start + dt.timedelta(days=1) shipment_start = today_start - dt.timedelta(days=SHIPMENT_LOOKBACK_DAYS) label_params = { "created_at_start": today_start.isoformat() + "Z", "created_at_end": today_end.isoformat() + "Z", "page_size": PAGE_SIZE, "sort_by": "created_at", "sort_dir": "desc", } shipment_params = { "created_at_start": shipment_start.isoformat() + "Z", "created_at_end": today_end.isoformat() + "Z", "page_size": PAGE_SIZE, "sort_by": "created_at", "sort_dir": "desc", } # Step 1: page 1 of each, fetched concurrently, to learn total page counts. first_pages = self._run_concurrently( { "labels": lambda: self._get("/labels", {**label_params, "page": 1}), "shipments": lambda: self._get("/shipments", {**shipment_params, "page": 1}), } ) labels = list(first_pages["labels"].get("labels", [])) shipments = list(first_pages["shipments"].get("shipments", [])) labels_total_pages = first_pages["labels"].get("pages", 1) shipments_total_pages = first_pages["shipments"].get("pages", 1) # Step 2: every remaining page across BOTH endpoints, fetched # concurrently together (not one endpoint at a time). remaining_jobs: Dict[str, Callable[[], dict]] = {} for page in range(2, labels_total_pages + 1): remaining_jobs[f"labels:{page}"] = ( lambda page=page: self._get("/labels", {**label_params, "page": page}) ) for page in range(2, shipments_total_pages + 1): remaining_jobs[f"shipments:{page}"] = ( lambda page=page: self._get("/shipments", {**shipment_params, "page": page}) ) if remaining_jobs: remaining = self._run_concurrently(remaining_jobs) for key, data in remaining.items(): if key.startswith("labels:"): labels.extend(data.get("labels", [])) else: shipments.extend(data.get("shipments", [])) return labels, shipments def _run_concurrently(self, jobs: Dict[str, Callable[[], dict]]) -> Dict[str, dict]: """Runs each zero-arg callable on its own QThreadPool worker thread, blocks until all finish, and returns {key: result}. If anything failed, raises the first error encountered (with its job key).""" pool = QThreadPool() pool.setMaxThreadCount(min(MAX_CONCURRENT_REQUESTS, max(1, len(jobs)))) results: Dict[str, dict] = {} errors: List[Exception] = [] lock = threading.Lock() for key, fn in jobs.items(): pool.start(_CallableTask(key, fn, results, lock, errors)) pool.waitForDone() if errors: raise errors[0] return results # -- single-request helper ------------------------------------------ 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 ({path}): {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 # -- correlation ------------------------------------------------------ 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, shipping_method: str ) -> 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=[], line_items=[], shipping_info={}, creator=None, assignee=None, description=None, tracking_numbers=tracking_numbers, shipping_method=shipping_method, summary=numbers_display, status=suggested_status, source_created_at=None, raw_data=raw, )