Files
Brunobrno 2213e115c6 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.
2026-01-11 16:32:51 +01:00

184 lines
6.8 KiB
Python

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'