Brunobrno
2213e115c6
Added Deutsche Post as a shipping carrier, including new models, admin, serializers, and API client integration. Updated Carrier and SiteConfiguration models to support Deutsche Post, including shipping price and API credentials. Added requirements for the Deutsche Post API client and dependencies.
407 lines
17 KiB
Python
407 lines
17 KiB
Python
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", "cz#Vytvořeno"
|
|
FINALIZED = "FINALIZED", "cz#Dokončeno"
|
|
SHIPPED = "SHIPPED", "cz#Odesláno"
|
|
DELIVERED = "DELIVERED", "cz#Doručeno"
|
|
CANCELLED = "CANCELLED", "cz#Zrušeno"
|
|
ERROR = "ERROR", "cz#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", "cz#Vytvořeno"
|
|
PROCESSING = "PROCESSING", "cz#Zpracovává se"
|
|
COMPLETED = "COMPLETED", "cz#Dokončeno"
|
|
ERROR = "ERROR", "cz#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}"
|