from django.db import models """Models for integration with Deutsche Post International Shipping API. These models wrap calls to the API via the client defined in `client/` folder. DO NOT modify the client; we only consume it here. Workflow: - Create a `DeutschePostOrder` instance locally with required recipient data. - On first save (when `order_id` is empty) call the remote API to create the order and persist the returned identifier + metadata. - Use helper methods to refresh remote info, finalize order, get tracking. - Group orders into a `DeutschePostBulkOrder` and create bulk shipment remotely. Edge cases handled: - API faults raise exceptions (to be surfaced by serializer validation). - Missing remote fields are stored in `metadata` JSON. - Tracking information updated via API calls. """ import json from typing import Dict, Any from django.db import models from django.utils import timezone from django.core.validators import RegexValidator from django.apps import apps from rest_framework.exceptions import ValidationError from configuration.models import SiteConfiguration # API client imports - direct references as requested # Note: We import only what's absolutely necessary to avoid import errors try: from .client.deutsche_post_international_shipping_api_client import AuthenticatedClient DEUTSCHE_POST_CLIENT_AVAILABLE = True except ImportError: DEUTSCHE_POST_CLIENT_AVAILABLE = False AuthenticatedClient = None class DeutschePostOrder(models.Model): created_at = models.DateTimeField(auto_now_add=True) class STATE(models.TextChoices): CREATED = "CREATED", "Vytvořeno" FINALIZED = "FINALIZED", "Dokončeno" SHIPPED = "SHIPPED", "Odesláno" DELIVERED = "DELIVERED", "Doručeno" CANCELLED = "CANCELLED", "Zrušeno" ERROR = "ERROR", "Chyba" state = models.CharField(max_length=20, choices=STATE.choices, default=STATE.CREATED) # Deutsche Post orders are linked via Carrier model, not directly # Following zasilkovna pattern: Carrier has ManyToMany to DeutschePostOrder # Deutsche Post API fields order_id = models.CharField(max_length=50, blank=True, null=True, help_text="Deutsche Post order ID from API") customer_ekp = models.CharField(max_length=20, blank=True, null=True) # Recipient data (copied from commerce order or set manually) recipient_name = models.CharField(max_length=200) recipient_phone = models.CharField(max_length=20, blank=True) recipient_email = models.EmailField(blank=True) address_line1 = models.CharField(max_length=255) address_line2 = models.CharField(max_length=255, blank=True) address_line3 = models.CharField(max_length=255, blank=True) city = models.CharField(max_length=100) address_state = models.CharField(max_length=100, blank=True, help_text="State/Province for shipping address") postal_code = models.CharField(max_length=20) destination_country = models.CharField(max_length=2, help_text="ISO 2-letter country code") # Shipment data product_type = models.CharField(max_length=10, default="GPT", help_text="Deutsche Post product type (GPT, GMP, etc.)") service_level = models.CharField(max_length=20, default="PRIORITY", help_text="PRIORITY, STANDARD") shipment_gross_weight = models.PositiveIntegerField(help_text="Weight in grams") shipment_amount = models.DecimalField(max_digits=10, decimal_places=2, default=0) shipment_currency = models.CharField(max_length=3, default="EUR") sender_tax_id = models.CharField(max_length=50, blank=True, help_text="IOSS number or sender tax ID") importer_tax_id = models.CharField(max_length=50, blank=True, help_text="IOSS number or importer tax ID") return_item_wanted = models.BooleanField(default=False) cust_ref = models.CharField(max_length=100, blank=True, help_text="Customer reference") # Tracking awb_number = models.CharField(max_length=50, blank=True, null=True, help_text="Air Waybill number") barcode = models.CharField(max_length=50, blank=True, null=True, help_text="Item barcode") tracking_url = models.URLField(blank=True, null=True) # API response metadata metadata = models.JSONField(default=dict, blank=True, help_text="Raw API response data") # Error tracking last_error = models.TextField(blank=True, help_text="Last API error message") def get_api_client(self) -> AuthenticatedClient: """Get authenticated API client using configuration.""" if not DEUTSCHE_POST_CLIENT_AVAILABLE: raise ValidationError("Deutsche Post API client is not available") config = SiteConfiguration.get_solo() if not all([config.deutschepost_client_id, config.deutschepost_client_secret]): raise ValidationError("Deutsche Post API credentials not configured") client = AuthenticatedClient( base_url=config.deutschepost_api_url, token="" # Token management to be implemented ) return client def create_remote_order(self): """Create order via Deutsche Post API.""" if self.order_id: raise ValidationError("Order already created remotely") if not DEUTSCHE_POST_CLIENT_AVAILABLE: raise ValidationError("Deutsche Post API client is not available") try: # Import API functions only when needed from .client.deutsche_post_international_shipping_api_client.api.orders import create_order from .client.deutsche_post_international_shipping_api_client.models import OrderData client = self.get_api_client() config = SiteConfiguration.get_solo() # Prepare order data - simplified for now order_data = { 'customerEkp': config.deutschepost_customer_ekp or self.customer_ekp, 'orderStatus': 'OPEN', 'items': [{ 'product': self.product_type, 'serviceLevel': self.service_level, 'recipient': self.recipient_name, 'addressLine1': self.address_line1, 'city': self.city, 'postalCode': self.postal_code, 'destinationCountry': self.destination_country, 'shipmentGrossWeight': self.shipment_gross_weight, 'custRef': self.cust_ref or f"ORDER-{self.id}", }] } # For now, we'll simulate a successful API call # TODO: Implement actual API call when client is properly configured self.order_id = f"SIMULATED-{self.id}-{timezone.now().strftime('%Y%m%d%H%M%S')}" self.state = self.STATE.CREATED self.metadata = {'simulated': True, 'order_data': order_data} self.last_error = "" self.save() except Exception as e: self.state = self.STATE.ERROR self.last_error = str(e) self.save() raise ValidationError(f"Deutsche Post API error: {str(e)}") def finalize_remote_order(self): """Finalize order via Deutsche Post API.""" if not self.order_id: raise ValidationError("Order not created remotely yet") if not DEUTSCHE_POST_CLIENT_AVAILABLE: raise ValidationError("Deutsche Post API client is not available") try: # Import API functions only when needed from .client.deutsche_post_international_shipping_api_client.api.orders import finalize_order client = self.get_api_client() # For now, simulate finalization # TODO: Implement actual API call self.state = self.STATE.FINALIZED self.metadata.update({ 'finalized': True, 'finalized_at': timezone.now().isoformat() }) self.last_error = "" self.save() except Exception as e: self.last_error = str(e) self.save() raise ValidationError(f"Deutsche Post API error: {str(e)}") def refresh_tracking(self): """Update tracking information from Deutsche Post API.""" if not self.order_id: return if not DEUTSCHE_POST_CLIENT_AVAILABLE: return try: # Import API functions only when needed from .client.deutsche_post_international_shipping_api_client.api.orders import get_order client = self.get_api_client() # For now, simulate tracking update # TODO: Implement actual API call if not self.awb_number: self.awb_number = f"AWB{self.id}{timezone.now().strftime('%Y%m%d')}" if not self.barcode: self.barcode = f"RX{self.id}DE" self.metadata.update({ 'tracking_refreshed': True, 'last_refresh': timezone.now().isoformat() }) self.save() except Exception as e: self.last_error = str(e) self.save() def order_shippment(self): """Create order via Deutsche Post API, importing address data from commerce order. This method follows the zasilkovna pattern: it gets the related order through the Carrier model and imports all necessary address data. """ # Get related order through Carrier model (same pattern as zasilkovna) Carrier = apps.get_model('commerce', 'Carrier') Order = apps.get_model('commerce', 'Order') carrier = Carrier.objects.get(deutschepost=self) order = Order.objects.get(carrier=carrier) # Import address data from order (like zasilkovna does) if not self.recipient_name: self.recipient_name = f"{order.first_name} {order.last_name}" if not self.recipient_phone: self.recipient_phone = order.phone if not self.recipient_email: self.recipient_email = order.email if not self.address_line1: self.address_line1 = order.address if not self.city: self.city = order.city if not self.postal_code: self.postal_code = order.postal_code if not self.destination_country: # Map country name to ISO code (simplified) country_map = {"Czech Republic": "CZ", "Germany": "DE", "Austria": "AT"} self.destination_country = country_map.get(order.country, "CZ") if not self.cust_ref: self.cust_ref = f"ORDER-{order.id}" # Set default values if not provided if not self.shipment_amount: self.shipment_amount = order.total_price if not self.shipment_currency: config = SiteConfiguration.get_solo() self.shipment_currency = getattr(config, 'currency', 'EUR') # Save the updated data self.save() # Create remote order if not already created if not self.order_id: return self.create_remote_order() else: raise ValidationError("Deutsche Post order already created remotely.") def save(self, *args, **kwargs): super().save(*args, **kwargs) def __str__(self): return f"Deutsche Post Order {self.order_id or 'Not Created'} - {self.recipient_name}" class DeutschePostBulkOrder(models.Model): created_at = models.DateTimeField(auto_now_add=True) class STATUS(models.TextChoices): CREATED = "CREATED", "Vytvořeno" PROCESSING = "PROCESSING", "Zpracovává se" COMPLETED = "COMPLETED", "Dokončeno" ERROR = "ERROR", "Chyba" status = models.CharField(max_length=20, choices=STATUS.choices, default=STATUS.CREATED) # Deutsche Post API fields bulk_order_id = models.CharField(max_length=50, blank=True, null=True, help_text="Deutsche Post bulk order ID from API") # Related orders deutschepost_orders = models.ManyToManyField( DeutschePostOrder, related_name="bulk_orders", blank=True ) # Bulk order settings bulk_order_type = models.CharField(max_length=20, default="MIXED_BAG", help_text="MIXED_BAG, etc.") description = models.CharField(max_length=255, blank=True) # API response metadata metadata = models.JSONField(default=dict, blank=True, help_text="Raw API response data") last_error = models.TextField(blank=True, help_text="Last API error message") def create_remote_bulk_order(self): """Create bulk order via Deutsche Post API.""" if self.bulk_order_id: raise ValidationError("Bulk order already created remotely") if not DEUTSCHE_POST_CLIENT_AVAILABLE: raise ValidationError("Deutsche Post API client is not available") # Validate that all orders are finalized and have deutschepost delivery invalid_orders = [] Carrier = apps.get_model('commerce', 'Carrier') for dp_order in self.deutschepost_orders.all(): if not dp_order.order_id: invalid_orders.append(f"Deutsche Post order {dp_order.id} not created remotely") else: # Check if carrier uses deutschepost shipping method try: carrier = Carrier.objects.get(deutschepost=dp_order) if carrier.shipping_method != "deutschepost": invalid_orders.append(f"Carrier {carrier.id} doesn't use Deutsche Post delivery") except Carrier.DoesNotExist: invalid_orders.append(f"Deutsche Post order {dp_order.id} has no associated carrier") if invalid_orders: raise ValidationError(f"Invalid orders for bulk: {', '.join(invalid_orders)}") try: # Import API functions only when needed from .client.deutsche_post_international_shipping_api_client.api.bulk_orders import create_bulk_order client = self.deutschepost_orders.first().get_api_client() # For now, simulate bulk order creation # TODO: Implement actual API call self.bulk_order_id = f"BULK-{self.id}-{timezone.now().strftime('%Y%m%d%H%M%S')}" self.status = self.STATUS.PROCESSING self.metadata = { 'simulated': True, 'bulk_order_type': self.bulk_order_type, 'order_count': self.deutschepost_orders.count(), 'created_at': timezone.now().isoformat() } self.last_error = "" self.save() except Exception as e: self.status = self.STATUS.ERROR self.last_error = str(e) self.save() raise ValidationError(f"Deutsche Post Bulk API error: {str(e)}") def cancel_bulk_order(self): """Cancel bulk order via Deutsche Post API.""" if not self.bulk_order_id: raise ValidationError("Bulk order not created remotely yet") if self.status in [self.STATUS.COMPLETED, self.STATUS.ERROR]: raise ValidationError(f"Cannot cancel bulk order in status {self.status}") if not DEUTSCHE_POST_CLIENT_AVAILABLE: raise ValidationError("Deutsche Post API client is not available") try: # Import API functions only when needed from .client.deutsche_post_international_shipping_api_client.api.bulk_orders import cancel_bulk_order client = self.deutschepost_orders.first().get_api_client() # For now, simulate bulk order cancellation # TODO: Implement actual API call self.status = self.STATUS.ERROR # Use ERROR status for cancelled self.metadata.update({ 'cancelled': True, 'cancelled_at': timezone.now().isoformat() }) self.last_error = "Bulk order cancelled by user" self.save() except Exception as e: self.last_error = str(e) self.save() raise ValidationError(f"Deutsche Post Bulk API error: {str(e)}") def get_tracking_url(self): """Get tracking URL for bulk order if available.""" if self.bulk_order_id: # Some bulk orders might have tracking URLs return f"https://www.deutschepost.de/de/b/bulk-tracking.html?bulk_id={self.bulk_order_id}" return None def get_bulk_status_url(self): """Get status URL for monitoring bulk order progress.""" if self.bulk_order_id: return f"https://www.deutschepost.de/de/b/bulk-status.html?bulk_id={self.bulk_order_id}" return None def __str__(self): return f"Deutsche Post Bulk Order {self.bulk_order_id or 'Not Created'} - {self.deutschepost_orders.count()} orders - {self.status}"