Integrate Deutsche Post shipping API and models

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.
This commit is contained in:
2026-01-11 16:32:51 +01:00
parent 7ebc83dd8c
commit 2213e115c6
131 changed files with 14686 additions and 5 deletions

183
backend/thirdparty/deutschepost/admin.py vendored Normal file
View File

@@ -0,0 +1,183 @@
from django.contrib import admin
from django.contrib import admin
from django.utils.html import format_html
from django.urls import reverse
from django.http import HttpResponseRedirect
from django.contrib import messages
from .models import DeutschePostOrder, DeutschePostBulkOrder
@admin.register(DeutschePostOrder)
class DeutschePostOrderAdmin(admin.ModelAdmin):
list_display = [
'id', 'order_id', 'recipient_name', 'state', 'commerce_order_link',
'awb_number', 'created_at'
]
list_filter = ['state', 'product_type', 'service_level', 'destination_country', 'created_at']
search_fields = ['order_id', 'recipient_name', 'recipient_email', 'awb_number', 'barcode']
readonly_fields = [
'order_id', 'awb_number', 'barcode', 'tracking_url',
'metadata', 'last_error', 'created_at'
]
fieldsets = (
('Basic Information', {
'fields': ('state', 'commerce_order', 'order_id', 'customer_ekp')
}),
('Recipient Information', {
'fields': (
'recipient_name', 'recipient_phone', 'recipient_email',
'address_line1', 'address_line2', 'address_line3',
'city', 'address_state', 'postal_code', 'destination_country'
)
}),
('Shipment Details', {
'fields': (
'product_type', 'service_level', 'shipment_gross_weight',
'shipment_amount', 'shipment_currency',
'sender_tax_id', 'importer_tax_id', 'return_item_wanted',
'cust_ref'
)
}),
('Tracking Information', {
'fields': ('awb_number', 'barcode', 'tracking_url'),
'classes': ['collapse']
}),
('API Data', {
'fields': ('metadata', 'last_error', 'created_at'),
'classes': ['collapse']
})
)
actions = ['create_remote_order', 'finalize_remote_order', 'refresh_tracking']
def commerce_order_link(self, obj):
"""Link to related commerce order."""
if obj.commerce_order:
url = reverse('admin:commerce_order_change', args=[obj.commerce_order.pk])
return format_html('<a href="{}">{}</a>', url, obj.commerce_order)
return '-'
commerce_order_link.short_description = 'Commerce Order'
def create_remote_order(self, request, queryset):
"""Admin action to create orders remotely."""
created_count = 0
error_count = 0
for order in queryset:
try:
if not order.order_id:
order.create_remote_order()
created_count += 1
else:
error_count += 1
except Exception as e:
messages.error(request, f'Error creating order {order.id}: {str(e)}')
error_count += 1
if created_count:
messages.success(request, f'{created_count} orders created remotely')
if error_count:
messages.warning(request, f'{error_count} orders could not be created')
create_remote_order.short_description = 'Create selected orders remotely'
def finalize_remote_order(self, request, queryset):
"""Admin action to finalize orders remotely."""
finalized_count = 0
error_count = 0
for order in queryset:
try:
if order.order_id and order.state != DeutschePostOrder.STATE.FINALIZED:
order.finalize_remote_order()
finalized_count += 1
else:
error_count += 1
except Exception as e:
messages.error(request, f'Error finalizing order {order.id}: {str(e)}')
error_count += 1
if finalized_count:
messages.success(request, f'{finalized_count} orders finalized')
if error_count:
messages.warning(request, f'{error_count} orders could not be finalized')
finalize_remote_order.short_description = 'Finalize selected orders'
def refresh_tracking(self, request, queryset):
"""Admin action to refresh tracking information."""
updated_count = 0
error_count = 0
for order in queryset:
try:
if order.order_id:
order.refresh_tracking()
updated_count += 1
else:
error_count += 1
except Exception as e:
messages.error(request, f'Error refreshing tracking for order {order.id}: {str(e)}')
error_count += 1
if updated_count:
messages.success(request, f'{updated_count} orders tracking updated')
if error_count:
messages.warning(request, f'{error_count} orders could not be updated')
refresh_tracking.short_description = 'Refresh tracking for selected orders'
@admin.register(DeutschePostBulkOrder)
class DeutschePostBulkOrderAdmin(admin.ModelAdmin):
list_display = ['id', 'bulk_order_id', 'status', 'orders_count', 'created_at']
list_filter = ['status', 'bulk_order_type', 'created_at']
search_fields = ['bulk_order_id', 'description']
readonly_fields = ['bulk_order_id', 'orders_count', 'metadata', 'last_error', 'created_at']
fieldsets = (
('Basic Information', {
'fields': ('status', 'bulk_order_id', 'bulk_order_type', 'description')
}),
('Orders', {
'fields': ('deutschepost_orders', 'orders_count')
}),
('API Data', {
'fields': ('metadata', 'last_error', 'created_at'),
'classes': ['collapse']
})
)
filter_horizontal = ['deutschepost_orders']
actions = ['create_remote_bulk_order']
def orders_count(self, obj):
"""Count of orders in this bulk order."""
return obj.deutschepost_orders.count()
orders_count.short_description = 'Orders Count'
def create_remote_bulk_order(self, request, queryset):
"""Admin action to create bulk orders remotely."""
created_count = 0
error_count = 0
for bulk_order in queryset:
try:
if not bulk_order.bulk_order_id:
bulk_order.create_remote_bulk_order()
created_count += 1
else:
error_count += 1
except Exception as e:
messages.error(request, f'Error creating bulk order {bulk_order.id}: {str(e)}')
error_count += 1
if created_count:
messages.success(request, f'{created_count} bulk orders created remotely')
if error_count:
messages.warning(request, f'{error_count} bulk orders could not be created')
create_remote_bulk_order.short_description = 'Create selected bulk orders remotely'