""" Order domain: persist Digistore IPN events and expose simple DB listing. """ import logging from typing import Any from django.core.exceptions import MultipleObjectsReturned from orders.models import Order logger = logging.getLogger(__name__) # Digistore purchases that should become a paid Shopify order (aligned with on_payment). _SYNCABLE_BILLING_STATUSES = frozenset( {"completed", "paying"} ) # Local / Digistore states that must not be pushed to Shopify. _SKIP_LOCAL_ORDER_STATUSES = frozenset( {"refunded", "partially refunded", "cancelled", "canceled"} ) _NON_SYNCABLE_BILLING_STATUSES = frozenset( { "cancelled", "canceled", "refunded", "aborted", "rejected", "chargeback", "refund", "stopped", "failed", } ) _NON_SYNCABLE_TRANSACTION_TYPES = frozenset( {"refund", "chargeback", "cancel", "cancellation"} ) _SYNC_ERROR_MAX_LEN = 8000 def _truncate_sync_error(message: str, max_len: int = _SYNC_ERROR_MAX_LEN) -> str: s = (message or "").strip() if len(s) <= max_len: return s return s[: max_len - 3] + "..." def _amount_str(value: Any) -> str: if value is None or value == "": return "" return str(value).replace(",", ".").strip() def _amount_float(value: Any) -> float: s = _amount_str(value) if not s: return 0.0 try: return float(s) except (TypeError, ValueError): return 0.0 def _ipn_has_line_pricing(ipn_data: dict[str, Any]) -> bool: try: item_count = max(1, int(str(ipn_data.get("item_count") or "1"))) except (TypeError, ValueError): item_count = 1 for idx in range(1, item_count + 1): suffix = "" if idx == 1 else f"_{idx}" for key in (f"product_txn_amount{suffix}", f"product_amount{suffix}"): if _amount_float(ipn_data.get(key)) > 0: return True return False def _purchase_item_pricing_fields(item: dict[str, Any]) -> dict[str, str]: first = (item.get("amounts") or {}).get("first") or {} fields: dict[str, str] = {} mapping = ( ("total_brutto_amount", "product_txn_amount"), ("total_netto_amount", "product_txn_netto_amount"), ("total_vat_amount", "product_txn_vat_amount"), ("unit_brutto_amount", "product_amount"), ("unit_netto_amount", "product_netto_amount"), ("unit_vat_amount", "product_vat_amount"), ) for src, dst in mapping: val = _amount_str(first.get(src)) if val: fields[dst] = val return fields class OrderService: """Business logic for orders (Digistore IPN → local DB).""" @staticmethod def list_from_db(limit: int = 100, offset: int = 0) -> list[dict[str, Any]]: """List orders from our DB (synced records).""" qs = Order.objects.order_by("-created_at")[offset : offset + limit] return [ { "shopify_order_id": o.shopify_order_id, "digistore_order_id": o.digistore_order_id or None, "status": o.status, "created_at": o.created_at.isoformat(), } for o in qs ] # --- Digistore IPN helpers ------------------------------------------- @staticmethod def record_digistore_payment(ipn_data: dict[str, Any]) -> dict[str, Any]: """ Persist a Digistore24 on_payment IPN into the local Order table. We treat the Digistore `order_id` as our primary identifier and store the full IPN payload in `raw_payload`. """ order_id = str(ipn_data.get("order_id") or "").strip() if not order_id: return {"ok": False, "error": "missing_order_id"} # Basic status; you can refine based on ipn_data["billing_type"], etc. status = "paid" try: order, created = Order.objects.update_or_create( digistore_order_id=order_id, defaults={ # Shopify order id will be set after we create the Shopify order "status": status, "raw_payload": ipn_data, "sync_error": "", }, ) except MultipleObjectsReturned: # `digistore_order_id` is not unique in our model yet, so # older duplicates can exist. Avoid 500s by collapsing to one row. qs = Order.objects.filter(digistore_order_id=order_id).order_by("pk") order = qs.first() if not order: return {"ok": False, "error": "order_missing_after_multiple"} created = False logger.warning( "Multiple Orders found for digistore_order_id=%s; updating first pk=%s", order_id, order.pk, ) order.status = status order.raw_payload = ipn_data order.sync_error = "" order.save( update_fields=["status", "raw_payload", "sync_error", "updated_at"] ) return { "ok": True, "created": created, "digistore_order_id": order.digistore_order_id, } @staticmethod def set_order_sync_error(digistore_order_id: str, message: str) -> None: """Persist last Shopify/Digistore sync failure for admin inspection.""" oid = (digistore_order_id or "").strip() if not oid: return err = _truncate_sync_error(message) updated = False for order in Order.objects.filter(digistore_order_id=oid).order_by("pk")[:5]: order.sync_error = err order.save(update_fields=["sync_error", "updated_at"]) updated = True if not updated: logger.warning( "set_order_sync_error: no Order row for digistore_order_id=%s", oid, ) @staticmethod def clear_order_sync_error(digistore_order_id: str) -> None: oid = (digistore_order_id or "").strip() if not oid: return for order in Order.objects.filter(digistore_order_id=oid).order_by("pk")[:5]: if order.sync_error: order.sync_error = "" order.save(update_fields=["sync_error", "updated_at"]) @staticmethod def purchase_detail_to_ipn_payload(purchase: dict[str, Any]) -> dict[str, Any]: """ Map a Digistore getPurchase-style dict to a flat payload compatible with ShopifyService.create_order_from_digistore_ipn (same shape as on_payment IPN). """ order_id = str( purchase.get("id") or purchase.get("purchase_id") or purchase.get("order_id") or "" ).strip() buyer = purchase.get("buyer") if not isinstance(buyer, dict): buyer = {} def pick(*keys: str) -> str: for k in keys: v = buyer.get(k) if v is None or v == "": v = purchase.get(k) if v is not None and str(v).strip() != "": return str(v).strip() return "" email = pick("email", "buyer_email") first_name = pick("first_name", "address_first_name", "firstname") last_name = pick("last_name", "address_last_name", "lastname") address_street = pick("address_street", "street", "street_name") address_street2 = pick("address_street2", "street2") address_city = pick("address_city", "city") address_state = pick("address_state", "state", "province") address_zipcode = pick("address_zipcode", "zipcode", "zip") address_country = pick("address_country", "country") or "DE" items = purchase.get("items") or purchase.get("purchase_items") or [] if not isinstance(items, list): items = [] normalized: list[dict[str, Any]] = [] for it in items: if not isinstance(it, dict): continue pid = it.get("product_id") or it.get("id") if pid is None: continue qty = it.get("quantity") or 1 try: q = max(1, int(str(qty))) except (TypeError, ValueError): q = 1 row: dict[str, Any] = { "product_id": str(pid).strip(), "quantity": q, } row.update(_purchase_item_pricing_fields(it)) normalized.append(row) if not normalized and purchase.get("product_id"): try: q = max(1, int(str(purchase.get("quantity") or 1))) except (TypeError, ValueError): q = 1 normalized.append( { "product_id": str(purchase.get("product_id")).strip(), "quantity": q, } ) item_count = max(1, len(normalized)) out: dict[str, Any] = { "event": "on_payment", "order_id": order_id, "email": email, "address_first_name": first_name, "address_last_name": last_name, "address_street": address_street, "address_street2": address_street2, "address_city": address_city, "address_state": address_state, "address_zipcode": address_zipcode, "address_country": address_country, "item_count": str(item_count), "currency": pick("currency") or "EUR", "transaction_currency": pick("currency") or "EUR", } amount = purchase.get("amount_brutto") if amount is None: amount = purchase.get("amount") if amount is not None: out["amount_brutto"] = str(amount).replace(",", ".") out["amount"] = out["amount_brutto"] for idx, row in enumerate(normalized, start=1): suffix = "" if idx == 1 else f"_{idx}" out[f"product_id{suffix}"] = row["product_id"] out[f"quantity{suffix}"] = str(row["quantity"]) for key, value in row.items(): if key in ("product_id", "quantity"): continue out[f"{key}{suffix}"] = value if item_count == 1 and "product_id" not in out: out["product_id"] = normalized[0]["product_id"] out["quantity"] = str(normalized[0]["quantity"]) return out @staticmethod def _get_order_row(digistore_order_id: str) -> Order | None: oid = (digistore_order_id or "").strip() if not oid: return None return Order.objects.filter(digistore_order_id=oid).order_by("pk").first() @staticmethod def _resolve_ipn_for_sync( digistore_order_id: str, *, ipn_data: dict[str, Any] | None, purchase: dict[str, Any] | None, local_order: Order | None, ) -> dict[str, Any]: """Pick the best IPN-shaped payload (priced IPN beats getPurchase rebuild).""" candidates: list[dict[str, Any]] = [] if ipn_data: candidates.append(ipn_data) if local_order and isinstance(local_order.raw_payload, dict): candidates.append(local_order.raw_payload) if purchase: candidates.append(OrderService.purchase_detail_to_ipn_payload(purchase)) for candidate in candidates: if _ipn_has_line_pricing(candidate): return candidate if ipn_data: return ipn_data if local_order and isinstance(local_order.raw_payload, dict): return local_order.raw_payload if purchase: return OrderService.purchase_detail_to_ipn_payload(purchase) return {"order_id": digistore_order_id, "event": "on_payment"} @staticmethod def _record_digistore_payment_preserving_priced_ipn( ipn_data: dict[str, Any], local_order: Order | None, ) -> None: """ Persist payment metadata without replacing a priced IPN with a stripped payload. """ if ( local_order and isinstance(local_order.raw_payload, dict) and _ipn_has_line_pricing(local_order.raw_payload) and not _ipn_has_line_pricing(ipn_data) ): local_order.status = "paid" local_order.sync_error = "" local_order.save( update_fields=["status", "sync_error", "updated_at"] ) return OrderService.record_digistore_payment(ipn_data) @staticmethod def _shopify_id_from_order_response(shopify_order: dict[str, Any]) -> str: raw_shopify_order_id = str(shopify_order.get("id") or "").strip() if raw_shopify_order_id.startswith("gid://"): return raw_shopify_order_id.split("/")[-1] return raw_shopify_order_id @staticmethod def sync_order_to_shopify( digistore_order_id: str, *, ipn_data: dict[str, Any] | None = None, dry_run: bool = False, fetch_purchase: bool = True, ) -> dict[str, Any]: """ Single import path for IPN, admin sync, and retry commands. Skips duplicates, refunded/cancelled purchases, and orders with only omitted Digistore product IDs. """ from products.services import DigistoreService, ShopifyService oid = (digistore_order_id or "").strip() if not oid: return { "ok": False, "result": "failed", "error": "missing_digistore_order_id", } local_order = OrderService._get_order_row(oid) existing_shopify_id = ( (local_order.shopify_order_id or "").strip() if local_order else "" ) if existing_shopify_id: OrderService.clear_order_sync_error(oid) return { "ok": True, "result": "skipped_already_synced", "shopify_order_id": existing_shopify_id, "digistore_order_id": oid, } purchase = None if fetch_purchase: purchase = DigistoreService.get_purchase(oid, log=False) if ( fetch_purchase and not purchase and not ipn_data and not ( local_order and isinstance(local_order.raw_payload, dict) and local_order.raw_payload ) ): msg = "getPurchase_empty" if not dry_run: OrderService.set_order_sync_error(oid, msg) return { "ok": False, "result": "failed", "error": msg, "digistore_order_id": oid, } skip, skip_reason = OrderService.should_skip_purchase_sync( local_order=local_order, purchase=purchase, ) if skip: if not dry_run: if purchase and not local_order: OrderService._record_digistore_payment_preserving_priced_ipn( OrderService.purchase_detail_to_ipn_payload(purchase), local_order, ) OrderService.set_order_sync_error(oid, f"skipped:{skip_reason}") return { "ok": True, "result": "skipped", "reason": skip_reason, "digistore_order_id": oid, } ipn = OrderService._resolve_ipn_for_sync( oid, ipn_data=ipn_data, purchase=purchase, local_order=local_order, ) if not str(ipn.get("order_id") or "").strip(): ipn["order_id"] = oid if OrderService._only_skip_products(ipn): reason = "only_skipped_products" if not dry_run: OrderService._record_digistore_payment_preserving_priced_ipn( ipn, local_order ) OrderService.set_order_sync_error(oid, f"skipped:{reason}") return { "ok": True, "result": "skipped", "reason": reason, "digistore_order_id": oid, } if dry_run: return { "ok": True, "result": "dry_run_would_create", "digistore_order_id": oid, } OrderService._record_digistore_payment_preserving_priced_ipn( ipn, local_order ) local_order = OrderService._get_order_row(oid) existing_shopify_id = ( (local_order.shopify_order_id or "").strip() if local_order else "" ) if existing_shopify_id: OrderService.clear_order_sync_error(oid) return { "ok": True, "result": "skipped_already_synced", "shopify_order_id": existing_shopify_id, "digistore_order_id": oid, } ipn_for_create = ( local_order.raw_payload if local_order and isinstance(local_order.raw_payload, dict) and _ipn_has_line_pricing(local_order.raw_payload) else ipn ) try: shopify_order = ShopifyService.create_order_from_digistore_ipn( ipn_for_create ) except Exception as exc: logger.exception( "sync_order_to_shopify failed digistore_order_id=%s", oid, ) OrderService.set_order_sync_error(oid, str(exc)) return { "ok": False, "result": "failed", "error": str(exc), "digistore_order_id": oid, } shopify_id = OrderService._shopify_id_from_order_response(shopify_order) if not shopify_id: msg = "shopify_order_missing_id" OrderService.set_order_sync_error(oid, msg) return { "ok": False, "result": "failed", "error": msg, "digistore_order_id": oid, } target = OrderService._get_order_row(oid) if not target: msg = "no_order_row_after_create" OrderService.set_order_sync_error(oid, msg) return { "ok": False, "result": "failed", "error": msg, "digistore_order_id": oid, } target.shopify_order_id = shopify_id target.sync_error = "" target.save(update_fields=["shopify_order_id", "sync_error", "updated_at"]) return { "ok": True, "result": "created", "shopify_order_id": shopify_id, "digistore_order_id": oid, } @staticmethod def sync_digistore_orders_to_shopify( from_time: str, to_time: str, *, dry_run: bool = False, page_size: int = 100, ) -> dict[str, Any]: """ List Digistore purchases in [from_time, to_time], then for each purchase id create a Shopify order when local Order has no shopify_order_id yet (same rules as the IPN handler). """ from products.services import DigistoreService summaries = DigistoreService.list_purchases( from_time, to_time, page_size=page_size, log=False, ) def purchase_id_from_row(row: dict[str, Any]) -> str: return str( row.get("id") or row.get("purchase_id") or row.get("order_id") or "" ).strip() seen: list[str] = [] for row in summaries: pid = purchase_id_from_row(row) if pid and pid not in seen: seen.append(pid) skipped_has_shopify = 0 skipped_status = 0 skipped_only_products = 0 skipped_missing_id = 0 created_shopify = 0 errors: list[dict[str, Any]] = [] dry_run_would = 0 would_create_purchase_ids: list[str] = [] for purchase_id in seen: if not purchase_id: skipped_missing_id += 1 continue result = OrderService.sync_order_to_shopify( purchase_id, dry_run=dry_run, fetch_purchase=True, ) outcome = result.get("result") if outcome == "skipped_already_synced": skipped_has_shopify += 1 continue if outcome == "skipped": if result.get("reason") == "only_skipped_products": skipped_only_products += 1 else: skipped_status += 1 continue if outcome == "dry_run_would_create": dry_run_would += 1 would_create_purchase_ids.append(purchase_id) continue if outcome == "created": created_shopify += 1 continue errors.append( { "purchase_id": purchase_id, "error": result.get("error") or outcome or "sync_failed", } ) return { "ok": True, "from": from_time, "to": to_time, "purchase_ids_seen": len(seen), "skipped_already_in_shopify": skipped_has_shopify, "skipped_billing_status": skipped_status, "skipped_only_skipped_products": skipped_only_products, "skipped_missing_purchase_id": skipped_missing_id, "shopify_orders_created": created_shopify, "dry_run_would_create": dry_run_would, "would_create_purchase_ids": would_create_purchase_ids, "errors": errors, } @staticmethod def _ipn_product_ids(ipn_data: dict[str, Any]) -> list[str]: try: item_count = max(1, int(str(ipn_data.get("item_count") or "1"))) except (TypeError, ValueError): item_count = 1 ids: list[str] = [] for idx in range(1, item_count + 1): suffix = "" if idx == 1 else f"_{idx}" pid = str(ipn_data.get(f"product_id{suffix}") or "").strip() if pid: ids.append(pid) return ids @staticmethod def _only_skip_products(ipn_data: dict[str, Any]) -> bool: from django.conf import settings skip_ids = { str(pid).strip() for pid in (getattr(settings, "SKIP_DIGISTORE_PRODUCT_IDS", None) or []) if str(pid).strip() } product_ids = OrderService._ipn_product_ids(ipn_data) return bool(product_ids) and all(pid in skip_ids for pid in product_ids) @staticmethod def should_skip_purchase_sync( *, local_order: Order | None = None, purchase: dict[str, Any] | None = None, ) -> tuple[bool, str]: """Return (True, reason) when an order must not be imported to Shopify.""" if local_order: local_status = (local_order.status or "").strip().lower() if local_status in _SKIP_LOCAL_ORDER_STATUSES: return True, f"local_status_{local_status.replace(' ', '_')}" if not purchase: return False, "" billing_status = (purchase.get("billing_status") or "").strip().lower() if billing_status in _NON_SYNCABLE_BILLING_STATUSES: return True, f"digistore_billing_{billing_status}" if billing_status and billing_status not in _SYNCABLE_BILLING_STATUSES: return True, f"digistore_billing_{billing_status}" for tx in purchase.get("transactions") or []: if not isinstance(tx, dict): continue tx_type = ( tx.get("type") or tx.get("transaction_type") or "" ).strip().lower() if tx_type in _NON_SYNCABLE_TRANSACTION_TYPES: return True, f"digistore_transaction_{tx_type}" return False, "" @staticmethod def retry_failed_orders_to_shopify( *, sync_error_contains: str = "", digistore_order_id: str = "", limit: int = 500, dry_run: bool = False, refresh_from_digistore: bool = True, ) -> dict[str, Any]: """ Retry local Order rows that have no shopify_order_id. Skips refunded/cancelled purchases (local status or live Digistore state). Omits configured SKIP_DIGISTORE_PRODUCT_IDS from Shopify line items. """ from django.db.models import Q qs = Order.objects.filter( Q(shopify_order_id="") | Q(shopify_order_id__isnull=True), ).order_by("pk") oid_filter = (digistore_order_id or "").strip() if oid_filter: qs = qs.filter(digistore_order_id=oid_filter) err_filter = (sync_error_contains or "").strip() if err_filter: qs = qs.filter(sync_error__icontains=err_filter) if limit > 0: qs = qs[:limit] created = 0 skipped = 0 failed = 0 dry_run_would = 0 details: list[dict[str, Any]] = [] for order in qs: oid = (order.digistore_order_id or "").strip() if not oid: skipped += 1 details.append( {"digistore_order_id": "", "result": "skipped_missing_digistore_id"} ) continue result = OrderService.sync_order_to_shopify( oid, ipn_data=order.raw_payload if isinstance(order.raw_payload, dict) else None, dry_run=dry_run, fetch_purchase=refresh_from_digistore, ) outcome = result.get("result") if outcome == "created": created += 1 details.append( { "digistore_order_id": oid, "result": "created", "shopify_order_id": result.get("shopify_order_id"), } ) elif outcome in {"skipped", "skipped_already_synced"}: skipped += 1 details.append( { "digistore_order_id": oid, "result": "skipped", "reason": result.get("reason") or outcome, } ) elif outcome == "dry_run_would_create": dry_run_would += 1 details.append( {"digistore_order_id": oid, "result": "dry_run_would_create"} ) else: failed += 1 details.append( { "digistore_order_id": oid, "result": "failed", "error": result.get("error") or outcome, } ) return { "ok": True, "examined": len(details), "created": created, "skipped": skipped, "failed": failed, "dry_run_would_create": dry_run_would, "details": details, } @staticmethod def refund_purchase( digistore_order_id: str, *, reason: str = "", force: bool = False, request_date: str = "", ) -> dict[str, Any]: """ Orders-domain wrapper for Digistore full refunds. """ from products.services import DigistoreService oid = str(digistore_order_id or "").strip() if not oid: raise ValueError("digistore_order_id_required") return DigistoreService.refund_purchase( oid, reason=reason, force=force, request_date=request_date, ) @staticmethod def refund_partially( digistore_order_id: str, amount: float, *, reason: str = "", force: bool = False, ) -> dict[str, Any]: """ Orders-domain wrapper for Digistore partial refunds. """ from products.services import DigistoreService oid = str(digistore_order_id or "").strip() if not oid: raise ValueError("digistore_order_id_required") if amount <= 0: raise ValueError("partial_refund_amount_must_be_positive") return DigistoreService.refund_partially( oid, amount, reason=reason, force=force, ) @staticmethod def _map_parcel_service(company: str) -> str: """Map Shopify carrier name to Digistore parcel_service key.""" raw = (company or "").strip().lower() if not raw: return "other" aliases = { "ups": "ups", "united parcel service": "ups", "dhl": "dhl", "dhl express": "dhl", "dpd": "dpd", "fedex": "fedex", "federal express": "fedex", "usps": "usps", "hermes": "hermes", "gls": "gls", "tnt": "tnt", "chronopost": "chronopost", "colissimo": "colissimo", "postnl": "postnl", "deutsche post": "deutsche-post", "dhl paket": "dhl", } if raw in aliases: return aliases[raw] for needle, key in aliases.items(): if needle in raw: return key # Digistore accepts free-text carriers (max 31 ascii), lowercased. cleaned = "".join(ch if ch.isalnum() or ch in "-_" else "-" for ch in raw) cleaned = cleaned.strip("-_")[:31] or "other" return cleaned @staticmethod def sync_tracking_to_digistore( *, digistore_order_id: str = "", shopify_order_id: str = "", tracking_id: str = "", parcel_service: str = "", tracking_url: str = "", quantity: int = 0, notify_via_email: bool = False, dry_run: bool = False, ) -> dict[str, Any]: """ Push Shopify fulfillment tracking for one mapped order to Digistore updateDelivery. Looks up Order by digistore_order_id and/or shopify_order_id, reads tracking from Shopify fulfillments (unless CLI overrides are provided), resolves Digistore delivery via listDeliveries, then calls updateDelivery. """ from products.services import DigistoreService, ShopifyService ds_id = str(digistore_order_id or "").strip() sh_id = str(shopify_order_id or "").strip() if not ds_id and not sh_id: raise ValueError("digistore_order_id_or_shopify_order_id_required") order: Order | None = None if ds_id and sh_id: order = ( Order.objects.filter( digistore_order_id=ds_id, shopify_order_id=sh_id ) .order_by("pk") .first() ) elif ds_id: order = ( Order.objects.filter(digistore_order_id=ds_id).order_by("pk").first() ) else: order = ( Order.objects.filter(shopify_order_id=sh_id).order_by("pk").first() ) if not order: raise ValueError( f"order_mapping_not_found digistore={ds_id or '-'} shopify={sh_id or '-'}" ) ds_id = str(order.digistore_order_id or "").strip() sh_id = str(order.shopify_order_id or "").strip() if not ds_id: raise ValueError("order_missing_digistore_order_id") if not sh_id and not str(tracking_id or "").strip(): raise ValueError("order_missing_shopify_order_id") has_tracking_override = bool(str(tracking_id or "").strip()) shopify_tracking: dict[str, Any] | None = None shopify_info: dict[str, Any] = { "shopify_order_id": sh_id, "tracking": None, } can_fetch_shopify = bool(sh_id) and ( sh_id.isdigit() or sh_id.startswith("gid://shopify/Order/") ) if has_tracking_override: # Caller already supplied tracking (webhook payload or CLI override): # skip the Shopify GraphQL round-trip entirely. shopify_info = { "shopify_order_id": sh_id, "tracking": None, "source": "override", } elif can_fetch_shopify: try: shopify_info = ShopifyService.get_order_fulfillment_tracking(sh_id) shopify_tracking = shopify_info.get("tracking") except Exception as exc: raise RuntimeError(f"shopify_tracking_fetch_failed: {exc}") from exc elif sh_id: raise ValueError( f"shopify_order_id_not_numeric={sh_id!r}; " "pass --tracking-id or fix Order.shopify_order_id" ) override_tid = str(tracking_id or "").strip() override_carrier = str(parcel_service or "").strip() override_url = str(tracking_url or "").strip() st = shopify_tracking if isinstance(shopify_tracking, dict) else None tid = override_tid or (str((st or {}).get("number") or "").strip()) company = override_carrier or (str((st or {}).get("company") or "").strip()) url = override_url or (str((st or {}).get("url") or "").strip()) carrier = OrderService._map_parcel_service(company) if not tid: raise ValueError( f"no_tracking_on_shopify_order shopify_order_id={sh_id} " f"fulfillment_status={shopify_info.get('display_fulfillment_status')}" ) deliveries = DigistoreService.list_deliveries_for_purchase(ds_id) delivery = DigistoreService.pick_delivery_for_update(deliveries) if not delivery: raise ValueError(f"no_digistore_deliveries purchase_id={ds_id}") delivery_id = str(delivery.get("id") or "").strip() if not delivery_id: raise ValueError(f"delivery_missing_id purchase_id={ds_id}") try: total = int(delivery.get("quantity_total") or 0) already = int(delivery.get("quantity_delivered") or 0) except (TypeError, ValueError): total, already = 0, 0 add_qty = int(quantity or 0) if add_qty <= 0: remaining = total - already add_qty = remaining if remaining > 0 else 1 quantity_delivered = already + add_qty payload_preview = { "delivery_id": int(delivery_id) if delivery_id.isdigit() else delivery_id, "notify_via_email": "Y" if notify_via_email else "N", "data": { "type": "delivery", "quantity_delivered": quantity_delivered, }, "tracking": { "0": { "parcel_service": carrier, "tracking_id": tid, **({"tracking_url": url} if url else {}), } }, } result: dict[str, Any] = { "ok": True, "dry_run": bool(dry_run), "order_id": order.pk, "digistore_order_id": ds_id, "shopify_order_id": sh_id, "shopify_order_name": shopify_info.get("order_name") or "", "shopify_fulfillment_status": shopify_info.get( "display_fulfillment_status" ) or "", "delivery_id": delivery_id, "delivery_type": delivery.get("type"), "quantity_delivered": quantity_delivered, "tracking": { "number": tid, "company": company, "parcel_service": carrier, "url": url, }, "payload": payload_preview, } if dry_run: result["skipped"] = "dry_run" return result api_res = DigistoreService.update_delivery( delivery_id, quantity_delivered=quantity_delivered, tracking_id=tid, parcel_service=carrier, tracking_url=url, notify_via_email=notify_via_email, ) result["digistore_response"] = api_res if isinstance(api_res, dict) and api_res.get("result") == "error": result["ok"] = False result["error"] = ( f"API error {api_res.get('code')}: {api_res.get('message')}" ) return result