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 import base64 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, Client from .client.deutsche_post_international_shipping_api_client.api.authentication import get_access_token from .client.deutsche_post_international_shipping_api_client.api.orders import ( create_order, finalize_order, get_order ) from .client.deutsche_post_international_shipping_api_client.api.bulk_orders import ( create_mixed_order, get_bulk_order ) from .client.deutsche_post_international_shipping_api_client.models import ( OrderData, OrderDataOrderStatus, ItemData, ItemDataServiceLevel, ItemDataShipmentNaturetype, Paperwork, PaperworkPickupType, MixedBagOrderDTO, BulkOrderDto, Content ) from .client.deutsche_post_international_shipping_api_client.errors import UnexpectedStatus 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") # Label/Document storage label_pdf = models.FileField(upload_to='deutschepost/labels/', blank=True, null=True, help_text="Shipping label PDF") class LABEL_SIZE(models.TextChoices): A4 = "A4", "A4 (210x297mm)" A5 = "A5", "A5 (148x210mm)" A6 = "A6", "A6 (105x148mm)" label_size = models.CharField(max_length=10, choices=LABEL_SIZE.choices, default=LABEL_SIZE.A4) 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") # First get an access token using basic auth basic_client = Client( base_url=config.deutschepost_api_url or "https://api-sandbox.deutschepost.com", raise_on_unexpected_status=True ) # Create basic auth header credentials = f"{config.deutschepost_client_id}:{config.deutschepost_client_secret}" encoded_credentials = base64.b64encode(credentials.encode()).decode() try: # Get access token response = get_access_token.sync_detailed( client=basic_client, authorization=f"Basic {encoded_credentials}", accept="application/json" ) if response.parsed is None or not hasattr(response.parsed, 'access_token'): raise ValidationError("Failed to obtain access token from Deutsche Post API") access_token = response.parsed.access_token # Create authenticated client with the token client = AuthenticatedClient( base_url=config.deutschepost_api_url or "https://api-sandbox.deutschepost.com", token=access_token, raise_on_unexpected_status=True ) return client except Exception as e: raise ValidationError(f"Deutsche Post authentication error: {str(e)}") def create_remote_order(self): """Create order via Deutsche Post API.""" if self.order_id: raise ValidationError(detail="Order already created remotely") if not DEUTSCHE_POST_CLIENT_AVAILABLE: raise ValidationError(detail="Deutsche Post API client is not available") try: client = self.get_api_client() config = SiteConfiguration.get_solo() # Create content pieces if we have shipment data contents = [] if self.shipment_amount > 0: contents.append(Content( content_piece_hs_code=1234567890, # Default HS code, should be configurable content_piece_description="E-commerce goods", content_piece_value=str(self.shipment_amount), content_piece_netweight=max(100, self.shipment_gross_weight - 50), # Estimate net weight content_piece_origin=config.deutschepost_origin_country or "DE", content_piece_amount=1 )) # Create item data item_data = ItemData( product=self.product_type, service_level=ItemDataServiceLevel(self.service_level), recipient=self.recipient_name, address_line_1=self.address_line1, address_line_2=self.address_line2 or None, address_line_3=self.address_line3 or None, city=self.city, state=self.address_state or None, postal_code=self.postal_code, destination_country=self.destination_country, shipment_gross_weight=self.shipment_gross_weight, recipient_phone=self.recipient_phone or None, recipient_email=self.recipient_email or None, sender_tax_id=self.sender_tax_id or None, importer_tax_id=self.importer_tax_id or None, shipment_amount=float(self.shipment_amount) if self.shipment_amount else None, shipment_currency=self.shipment_currency or None, return_item_wanted=self.return_item_wanted, shipment_naturetype=ItemDataShipmentNaturetype.SALE_GOODS, cust_ref=self.cust_ref or f"ORDER-{self.id}", contents=contents if contents else None ) # Create paperwork data paperwork = Paperwork( contact_name=config.deutschepost_contact_name or "Contact", awb_copy_count=1, job_reference=f"JOB-{self.id}", pickup_type=PaperworkPickupType.CUSTOMER_DROP_OFF, telephone_number=config.deutschepost_contact_phone or "+490000000000" ) # Create order data order_data = OrderData( customer_ekp=config.deutschepost_customer_ekp or self.customer_ekp or "1234567890", order_status=OrderDataOrderStatus.OPEN, paperwork=paperwork, items=[item_data] ) # Make API call response = create_order.sync_detailed( client=client, body=order_data ) if response.parsed is None: raise ValidationError(f"Failed to create order. Status: {response.status_code}") # Handle different response types if hasattr(response.parsed, 'order_id'): # Successful response order_response = response.parsed self.order_id = order_response.order_id self.state = self.STATE.CREATED self.metadata = { 'api_response': order_response.to_dict() if hasattr(order_response, 'to_dict') else str(order_response), 'created_at': timezone.now().isoformat() } self.last_error = "" else: # Error response error_info = response.parsed.to_dict() if hasattr(response.parsed, 'to_dict') else str(response.parsed) self.state = self.STATE.ERROR self.last_error = f"API Error: {error_info}" self.metadata = {'api_error': error_info} raise ValidationError(detail=f"Deutsche Post API error: {error_info}") self.save() except UnexpectedStatus as e: self.state = self.STATE.ERROR self.last_error = f"API Error {e.status_code}: {e.content}" self.save() raise ValidationError(detail=f"Deutsche Post API error {e.status_code}: {e.content}") except Exception as e: self.state = self.STATE.ERROR self.last_error = str(e) self.save() raise ValidationError(detail=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(detail="Order not created remotely yet") if not DEUTSCHE_POST_CLIENT_AVAILABLE: raise ValidationError(detail="Deutsche Post API client is not available") try: client = self.get_api_client() config = SiteConfiguration.get_solo() # Create paperwork for finalization paperwork = Paperwork( contact_name=config.deutschepost_contact_name or "Contact", awb_copy_count=1, job_reference=f"JOB-{self.id}-FINAL", pickup_type=PaperworkPickupType.CUSTOMER_DROP_OFF, telephone_number=config.deutschepost_contact_phone or "+420000000000" ) # Make API call to finalize order response = finalize_order.sync_detailed( client=client, order_id=self.order_id, body=paperwork ) if response.parsed is None: raise ValidationError(f"Failed to finalize order. Status: {response.status_code}") # Handle different response types if hasattr(response.parsed, 'order_id'): # Successful response order_response = response.parsed self.state = self.STATE.FINALIZED # Extract tracking information if available if hasattr(order_response, 'items') and order_response.items: for item in order_response.items: if hasattr(item, 'awb') and item.awb: self.awb_number = item.awb if hasattr(item, 'barcode') and item.barcode: self.barcode = item.barcode self.metadata.update({ 'finalized': True, 'finalized_at': timezone.now().isoformat(), 'api_response': order_response.to_dict() if hasattr(order_response, 'to_dict') else str(order_response) }) self.last_error = "" else: # Error response error_info = response.parsed.to_dict() if hasattr(response.parsed, 'to_dict') else str(response.parsed) self.last_error = f"Finalization failed: {error_info}" raise ValidationError(detail=f"Deutsche Post API finalization error: {error_info}") self.save() except UnexpectedStatus as e: self.last_error = f"API Error {e.status_code}: {e.content}" self.save() raise ValidationError(f"Deutsche Post API error {e.status_code}: {e.content}") 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: client = self.get_api_client() # Make API call to get current order status response = get_order.sync_detailed( client=client, order_id=self.order_id ) if response.parsed is None: self.last_error = f"Failed to refresh tracking. Status: {response.status_code}" self.save() return # Handle different response types if hasattr(response.parsed, 'order_id'): # Successful response order_response = response.parsed # Update order status based on API response if hasattr(order_response, 'order_status'): api_status = str(order_response.order_status) if api_status == 'FINALIZED': self.state = self.STATE.FINALIZED elif api_status in ['SHIPPED', 'IN_TRANSIT']: self.state = self.STATE.SHIPPED elif api_status == 'DELIVERED': self.state = self.STATE.DELIVERED elif api_status in ['CANCELLED', 'RETURNED']: self.state = self.STATE.CANCELLED # Extract tracking information from items if hasattr(order_response, 'items') and order_response.items: for item in order_response.items: if hasattr(item, 'awb') and item.awb and not self.awb_number: self.awb_number = item.awb if hasattr(item, 'barcode') and item.barcode and not self.barcode: self.barcode = item.barcode # Update tracking URL if AWB number is available if self.awb_number: self.tracking_url = f"https://www.deutschepost.de/de/sendungsverfolgung.html?piececode={self.awb_number}" self.metadata.update({ 'tracking_refreshed': True, 'last_refresh': timezone.now().isoformat(), 'api_response': order_response.to_dict() if hasattr(order_response, 'to_dict') else str(order_response) }) self.last_error = "" else: # Error response error_info = response.parsed.to_dict() if hasattr(response.parsed, 'to_dict') else str(response.parsed) self.last_error = f"Tracking refresh failed: {error_info}" self.save() except UnexpectedStatus as e: self.last_error = f"API Error {e.status_code}: {e.content}" self.save() except Exception as e: self.last_error = str(e) self.save() def validate_for_shipping(self): """Validate that all required fields are present before API calls.""" errors = [] # Required recipient fields if not self.recipient_name: errors.append("Recipient name is required") if not self.address_line1: errors.append("Address line 1 is required") if not self.city: errors.append("City is required") if not self.postal_code: errors.append("Postal code is required") if not self.destination_country: errors.append("Destination country is required") # Required shipment fields if not self.shipment_gross_weight or self.shipment_gross_weight <= 0: errors.append("Valid shipment weight is required") # Contact information (email or phone required for notifications) if not self.recipient_email and not self.recipient_phone: errors.append("Either recipient email or phone is required for delivery notifications") if errors: raise ValidationError("; ".join(errors)) return True def get_shipping_cost_estimate(self): "Get estimated shipping cost based on service level and destination." config = SiteConfiguration.get_solo() base_price = config.deutschepost_shipping_price # Adjust price based on service level if self.service_level == "PRIORITY": multiplier = 1.2 # 20% premium for priority else: # STANDARD multiplier = 1.0 # Adjust price based on destination if self.destination_country not in ["DE", "AT", "FR", "NL", "BE", "LU", "SK", "PL"]: # Non-EU countries cost more multiplier *= 1.5 return base_price * multiplier def generate_shipping_label(self): """Generate and download shipping label PDF from Deutsche Post API.""" if not self.order_id: raise ValidationError("Order must be created remotely first") if not self.awb_number: raise ValidationError("AWB number required for label generation") if not DEUTSCHE_POST_CLIENT_AVAILABLE: raise ValidationError("Deutsche Post API client is not available") try: # Note: Deutsche Post API endpoints for labels vary # This is a placeholder for the actual label generation logic # You'll need to check Deutsche Post API documentation for exact endpoints client = self.get_api_client() # Try to get label via items endpoint or specific label endpoint # This is conceptual - adjust based on actual API from .client.deutsche_post_international_shipping_api_client.api.orders import get_order response = get_order.sync_detailed( client=client, order_id=self.order_id ) if response.parsed and hasattr(response.parsed, 'items') and response.parsed.items: for item in response.parsed.items: if hasattr(item, 'label_data') and item.label_data: # Save label PDF from django.core.files.base import ContentFile import base64 if isinstance(item.label_data, str): # Base64 encoded PDF pdf_content = base64.b64decode(item.label_data) else: # Binary data pdf_content = item.label_data filename = f"deutschepost_label_{self.order_id}_{timezone.now().strftime('%Y%m%d_%H%M%S')}.pdf" self.label_pdf.save( filename, ContentFile(pdf_content), save=True ) self.metadata.update({ 'label_generated': True, 'label_generated_at': timezone.now().isoformat() }) return True # Alternative: Generate label using AWB number # This is a fallback approach - create a simple label with AWB info self._generate_simple_label() return True except Exception as e: self.last_error = f"Label generation failed: {str(e)}" self.save() raise ValidationError(f"Deutsche Post label generation error: {str(e)}") def _generate_simple_label(self): """Generate a simple label PDF with basic shipping information using HTML template.""" from django.template.loader import render_to_string from django.core.files.base import ContentFile from weasyprint import HTML import io # Prepare template context context = { 'order': self, 'now': timezone.now(), 'estimated_delivery_days': self.get_estimated_delivery_days() } # Render HTML template html_string = render_to_string('deutschepost/shipping_label.html', context) # Generate PDF from HTML pdf_buffer = io.BytesIO() HTML(string=html_string).write_pdf(pdf_buffer) pdf_data = pdf_buffer.getvalue() pdf_buffer.close() # Save to model filename = f"deutschepost_label_{self.order_id or self.id}_{timezone.now().strftime('%Y%m%d_%H%M%S')}.pdf" self.label_pdf.save( filename, ContentFile(pdf_data), save=True ) self.metadata.update({ 'simple_label_generated': True, 'label_generated_at': timezone.now().isoformat() }) def get_tracking_url(self): """Get tracking URL for this order if tracking number is available.""" if self.awb_number: return f"https://www.deutschepost.de/de/sendungsverfolgung.html?piececode={self.awb_number}" elif self.barcode: return f"https://www.deutschepost.de/de/sendungsverfolgung.html?piececode={self.barcode}" return None def can_be_finalized(self): """Check if order can be finalized.""" return ( self.order_id and self.state == self.STATE.CREATED and not self.last_error ) def is_trackable(self): """Check if order has tracking information.""" return bool(self.awb_number or self.barcode) def get_estimated_delivery_days(self): """Get estimated delivery days based on service level and destination.""" if self.service_level == "PRIORITY": # EU countries usually 2-4 days, others 4-7 days if self.destination_country in ["DE", "AT", "FR", "NL", "BE", "LU", "SK", "PL"]: return "2-4" else: return "4-7" else: # STANDARD # EU countries usually 4-6 days, others 7-14 days if self.destination_country in ["DE", "AT", "FR", "NL", "BE", "LU", "SK", "PL"]: return "4-6" else: return "7-14" 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(app_label='commerce', model_name='Carrier') Order = apps.get_model(app_label='commerce', model_name='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 (comprehensive mapping) country_map = { "Czech Republic": "CZ", "Czechia": "CZ", "Germany": "DE", "Deutschland": "DE", "Austria": "AT", "Österreich": "AT", "Slovakia": "SK", "Slovak Republic": "SK", "Poland": "PL", "Polska": "PL", "Hungary": "HU", "Magyarország": "HU", "France": "FR", "République française": "FR", "Netherlands": "NL", "Nederland": "NL", "Belgium": "BE", "België": "BE", "Belgique": "BE", "Luxembourg": "LU", "Lëtzebuerg": "LU", "Switzerland": "CH", "Schweiz": "CH", "Italy": "IT", "Italia": "IT", "Spain": "ES", "España": "ES", "United Kingdom": "GB", "UK": "GB", "United States": "US", "USA": "US", "Canada": "CA" } 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): # Auto-create remote order when first saved (similar to zasilkovna pattern) is_creating = not self.pk super().save(*args, **kwargs) if is_creating and not self.order_id: # Only auto-create if we have minimum required data try: self.validate_for_shipping() self.create_remote_order() except ValidationError: # Don't auto-create if validation fails - admin can fix and retry manually pass def delete(self, *args, **kwargs): """Override delete to cancel remote order if possible.""" if self.order_id and self.can_be_cancelled(): try: self.cancel_remote_order() except ValidationError: # Log error but don't prevent deletion self.last_error = "Failed to cancel remote order during deletion" self.save() super().delete(*args, **kwargs) def can_be_cancelled(self): """Check if order can be cancelled (not yet shipped).""" return ( self.order_id and self.state in [self.STATE.CREATED, self.STATE.FINALIZED] and not self.awb_number # No AWB means not yet shipped ) def cancel_remote_order(self): """Cancel order via Deutsche Post API (if supported).""" if not self.order_id: raise ValidationError("Order not created remotely yet") # Note: Deutsche Post API might not have explicit cancel endpoint # We mark as cancelled locally and track in metadata self.state = self.STATE.CANCELLED self.metadata.update({ 'cancelled': True, 'cancelled_at': timezone.now().isoformat(), 'cancellation_note': 'Cancelled locally - API cancel might not be supported' }) self.last_error = "Order cancelled by user" self.save() 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") # Bulk shipment documents bulk_label_pdf = models.FileField(upload_to='deutschepost/bulk_labels/', blank=True, null=True, help_text="Bulk shipment label PDF") paperwork_pdf = models.FileField(upload_to='deutschepost/paperwork/', blank=True, null=True, help_text="Bulk shipment paperwork PDF") 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: client = self.deutschepost_orders.first().get_api_client() config = SiteConfiguration.get_solo() # Calculate totals from all orders total_count = self.deutschepost_orders.count() total_weight = sum(order.shipment_gross_weight / 1000.0 for order in self.deutschepost_orders.all()) # Convert grams to kg if total_count == 0: raise ValidationError("No orders found for bulk order") # Use first order's product type and service level (assuming all are same type) first_order = self.deutschepost_orders.first() # Create mixed bag order data mixed_bag_order = MixedBagOrderDTO( contact_name=config.deutschepost_contact_name or "Contact", product=first_order.product_type, service_level=first_order.service_level, items_count=total_count, items_weight_in_kilogram=max(0.1, total_weight), # Minimum 0.1kg total_count_receptacles=total_count, # Each order is one receptacle format_="MIXED", telephone_number=config.deutschepost_contact_phone, job_reference=self.description or f"Bulk-{self.id}", customer_ekp=config.deutschepost_customer_ekp or "1234567890" ) # Get customer EKP customer_ekp = config.deutschepost_customer_ekp or "1234567890" # Make API call using mixed order endpoint response = create_mixed_order.sync_detailed( client=client, customer_ekp=customer_ekp, body=mixed_bag_order ) if response.parsed is None: raise ValidationError(f"Failed to create bulk order. Status: {response.status_code}") # Handle different response types if hasattr(response.parsed, 'order_id'): # Successful response (BulkOrderDto) bulk_response = response.parsed self.bulk_order_id = str(bulk_response.order_id) self.status = self.STATUS.PROCESSING self.metadata = { 'api_response': bulk_response.to_dict() if hasattr(bulk_response, 'to_dict') else str(bulk_response), 'bulk_order_type': self.bulk_order_type, 'order_count': total_count, 'total_weight_kg': total_weight, 'created_at': timezone.now().isoformat() } self.last_error = "" else: # Error response error_info = response.parsed.to_dict() if hasattr(response.parsed, 'to_dict') else str(response.parsed) self.status = self.STATUS.ERROR self.last_error = f"Bulk order creation failed: {error_info}" self.metadata = {'api_error': error_info} raise ValidationError(f"Deutsche Post Bulk API error: {error_info}") self.save() except UnexpectedStatus as e: self.status = self.STATUS.ERROR self.last_error = f"API Error {e.status_code}: {e.content}" self.save() raise ValidationError(f"Deutsche Post Bulk API error {e.status_code}: {e.content}") 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: client = self.deutschepost_orders.first().get_api_client() # Note: Deutsche Post API might not have a direct cancel endpoint for bulk orders # In that case, we check the current status and mark as cancelled locally try: # Try to get current bulk order status response = get_bulk_order.sync_detailed( client=client, bulk_order_id=self.bulk_order_id ) if response.parsed and hasattr(response.parsed, 'order_status'): api_status = str(response.parsed.order_status) if api_status in ['COMPLETED', 'FINALIZED']: raise ValidationError("Cannot cancel bulk order that is already completed") except UnexpectedStatus: # If we can't get status, proceed with local cancellation pass # Since there might not be a cancel API endpoint, mark as cancelled locally self.status = self.STATUS.ERROR # Use ERROR status for cancelled self.metadata.update({ 'cancelled': True, 'cancelled_at': timezone.now().isoformat(), 'cancellation_note': 'Cancelled locally - API might not support bulk order cancellation' }) self.last_error = "Bulk order cancelled by user" self.save() except UnexpectedStatus as e: self.last_error = f"API Error {e.status_code}: {e.content}" self.save() raise ValidationError(f"Deutsche Post Bulk API error {e.status_code}: {e.content}") except Exception as e: self.last_error = str(e) self.save() raise ValidationError(f"Deutsche Post Bulk API error: {str(e)}") def refresh_bulk_status(self): """Refresh bulk order status from Deutsche Post API.""" if not self.bulk_order_id: return if not DEUTSCHE_POST_CLIENT_AVAILABLE: return try: client = self.deutschepost_orders.first().get_api_client() # Make API call to get current bulk order status response = get_bulk_order.sync_detailed( client=client, bulk_order_id=self.bulk_order_id ) if response.parsed is None: self.last_error = f"Failed to refresh bulk status. Status: {response.status_code}" self.save() return # Handle successful response if hasattr(response.parsed, 'order_status'): api_status = str(response.parsed.order_status) if api_status in ['COMPLETED', 'FINALIZED']: self.status = self.STATUS.COMPLETED elif api_status in ['PROCESSING', 'IN_PROGRESS']: self.status = self.STATUS.PROCESSING elif api_status in ['CANCELLED', 'ERROR']: self.status = self.STATUS.ERROR self.metadata.update({ 'status_refreshed': True, 'last_refresh': timezone.now().isoformat(), 'api_response': response.parsed.to_dict() if hasattr(response.parsed, 'to_dict') else str(response.parsed) }) self.last_error = "" else: # Error response error_info = response.parsed.to_dict() if hasattr(response.parsed, 'to_dict') else str(response.parsed) self.last_error = f"Status refresh failed: {error_info}" self.save() except UnexpectedStatus as e: self.last_error = f"API Error {e.status_code}: {e.content}" self.save() except Exception as e: self.last_error = str(e) self.save() def get_total_weight_kg(self): """Get total weight of all orders in kg.""" return sum(order.shipment_gross_weight / 1000.0 for order in self.deutschepost_orders.all()) def can_be_cancelled(self): """Check if bulk order can be cancelled.""" return ( self.bulk_order_id and self.status in [self.STATUS.CREATED, self.STATUS.PROCESSING] and not self.last_error ) 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 save(self, *args, **kwargs): # Auto-create remote bulk order when first saved with orders is_creating = not self.pk super().save(*args, **kwargs) if is_creating and not self.bulk_order_id and self.deutschepost_orders.exists(): try: self.create_remote_bulk_order() except ValidationError: # Don't auto-create if validation fails - admin can fix and retry manually pass def delete(self, *args, **kwargs): """Override delete to cancel remote bulk order if possible.""" if self.bulk_order_id and self.can_be_cancelled(): try: self.cancel_bulk_order() except ValidationError: # Log error but don't prevent deletion self.last_error = "Failed to cancel remote bulk order during deletion" self.save() super().delete(*args, **kwargs) def generate_bulk_labels(self): """Generate combined PDF with all order labels using HTML template.""" if not self.deutschepost_orders.exists(): raise ValidationError("No orders in bulk shipment") from django.template.loader import render_to_string from django.core.files.base import ContentFile from weasyprint import HTML import io orders = self.deutschepost_orders.all() total_weight_kg = self.get_total_weight_kg() # Prepare template context context = { 'bulk_order': self, 'orders': orders, 'total_weight_kg': total_weight_kg, 'now': timezone.now() } # Render HTML template html_string = render_to_string('deutschepost/bulk_labels.html', context) # Generate PDF from HTML pdf_buffer = io.BytesIO() HTML(string=html_string).write_pdf(pdf_buffer) pdf_data = pdf_buffer.getvalue() pdf_buffer.close() # Save bulk label PDF filename = f"deutschepost_bulk_labels_{self.bulk_order_id or self.id}_{timezone.now().strftime('%Y%m%d_%H%M%S')}.pdf" self.bulk_label_pdf.save( filename, ContentFile(pdf_data), save=True ) self.metadata.update({ 'bulk_labels_generated': True, 'labels_generated_at': timezone.now().isoformat() }) return True def __str__(self): return f"Deutsche Post Bulk Order {self.bulk_order_id or 'Not Created'} - {self.deutschepost_orders.count()} orders - {self.status}"