Autosave: 20260125-124556

This commit is contained in:
Flatlogic Bot 2026-01-25 12:45:56 +00:00
parent 1e836a1d9d
commit 917a89a262
13 changed files with 254 additions and 75 deletions

View File

@ -205,7 +205,7 @@ else:
WHATSAPP_API_KEY = os.getenv("WHATSAPP_API_KEY", "")
WHATSAPP_PHONE_ID = os.getenv("WHATSAPP_PHONE_ID", "")
WHATSAPP_BUSINESS_ACCOUNT_ID = os.getenv("WHATSAPP_BUSINESS_ACCOUNT_ID", "")
WHATSAPP_ENABLED = os.getenv("WHATSAPP_ENABLED", "false").lower() == "true"
WHATSAPP_ENABLED = os.getenv("WHATSAPP_ENABLED", "true").lower() == "true"
# Default primary key field type
# https://docs.djangoproject.com/en/5.2/ref/settings/#default-auto-field

View File

@ -1,57 +1,96 @@
from django.contrib import admin
from django.contrib.auth.admin import UserAdmin
from django.contrib.auth.models import User
from .models import Profile, Parcel, Country, Governate, City, PlatformProfile
from django.utils.translation import gettext_lazy as _
from django.urls import path, reverse
from django.shortcuts import render
from django.utils.html import format_html
from django.contrib import messages
from .whatsapp_utils import send_whatsapp_message_detailed
import logging
@admin.register(Country)
class CountryAdmin(admin.ModelAdmin):
list_display = ('name_en', 'name_ar')
search_fields = ('name_en', 'name_ar')
class ProfileInline(admin.StackedInline):
model = Profile
can_delete = False
verbose_name_plural = _('Profiles')
@admin.register(Governate)
class GovernateAdmin(admin.ModelAdmin):
list_display = ('name_en', 'name_ar', 'country')
list_filter = ('country',)
search_fields = ('name_en', 'name_ar')
class CustomUserAdmin(UserAdmin):
inlines = (ProfileInline,)
@admin.register(City)
class CityAdmin(admin.ModelAdmin):
list_display = ('name_en', 'name_ar', 'governate')
list_filter = ('governate__country', 'governate')
search_fields = ('name_en', 'name_ar')
@admin.register(Profile)
class ProfileAdmin(admin.ModelAdmin):
list_display = ('user', 'role', 'phone_number', 'country', 'governate', 'city')
list_filter = ('role', 'country', 'governate')
search_fields = ('user__username', 'phone_number')
@admin.register(Parcel)
class ParcelAdmin(admin.ModelAdmin):
list_display = ('tracking_number', 'shipper', 'carrier', 'status', 'payment_status', 'created_at')
list_filter = ('status', 'payment_status', 'pickup_country', 'delivery_country')
search_fields = ('tracking_number', 'shipper__username', 'carrier__username', 'receiver_name')
readonly_fields = ('tracking_number', 'created_at', 'updated_at')
list_display = ('tracking_number', 'shipper', 'status', 'created_at')
list_filter = ('status', 'created_at')
search_fields = ('tracking_number', 'shipper__username', 'receiver_name')
@admin.register(PlatformProfile)
class PlatformProfileAdmin(admin.ModelAdmin):
list_display = ('name', 'phone_number', 'registration_number')
fieldsets = (
(None, {
'fields': ('name', 'logo', 'slogan')
(_('General Info'), {
'fields': ('name', 'logo', 'slogan', 'address', 'phone_number', 'registration_number', 'vat_number')
}),
('Contact Information', {
'fields': ('address', 'phone_number', 'registration_number', 'vat_number')
}),
('Legal', {
(_('Policies'), {
'fields': ('privacy_policy', 'terms_conditions')
}),
('WhatsApp Configuration', {
'fields': ('whatsapp_access_token', 'whatsapp_business_phone_number_id'),
'description': 'Enter your Meta WhatsApp Business API credentials here. These will override the system defaults.'
(_('WhatsApp Configuration (Wablas Gateway)'), {
'fields': ('whatsapp_access_token', 'whatsapp_app_secret', 'whatsapp_business_phone_number_id'),
'description': _('Configure your Wablas API connection. Use "Test WhatsApp Configuration" to verify.')
}),
)
def has_add_permission(self, request):
# Allow adding only if no instance exists
# Allow only one instance
if self.model.objects.exists():
return False
return super().has_add_permission(request)
return super().has_add_permission(request)
def get_urls(self):
urls = super().get_urls()
custom_urls = [
path('test-whatsapp/', self.admin_site.admin_view(self.test_whatsapp_view), name='test-whatsapp'),
]
return custom_urls + urls
def test_whatsapp_view(self, request):
phone_number = ''
if request.method == 'POST':
phone_number = request.POST.get('phone_number')
if phone_number:
success, msg = send_whatsapp_message_detailed(phone_number, "This is a test message from your Platform.")
if success:
messages.success(request, f"Success: {msg}")
else:
messages.error(request, f"Error: {msg}")
else:
messages.warning(request, "Please enter a phone number.")
context = dict(
self.admin_site.each_context(request),
phone_number=phone_number,
)
return render(request, "admin/core/platformprofile/test_whatsapp.html", context)
def test_connection_link(self, obj):
return format_html(
'<a class="button" href="{}">{}</a>',
reverse('admin:test-whatsapp'),
_('Test WhatsApp Configuration')
)
test_connection_link.short_description = _("Actions")
test_connection_link.allow_tags = True
readonly_fields = ('test_connection_link',)
def get_fieldsets(self, request, obj=None):
fieldsets = super().get_fieldsets(request, obj)
# Add the test link to the first fieldset or a new one
if obj:
fieldsets += ((_('Tools'), {'fields': ('test_connection_link',)}),)
return fieldsets
admin.site.unregister(User)
admin.site.register(User, CustomUserAdmin)
admin.site.register(Parcel, ParcelAdmin)
admin.site.register(Country)
admin.site.register(Governate)
admin.site.register(City)
admin.site.register(PlatformProfile, PlatformProfileAdmin)

View File

@ -0,0 +1,18 @@
# Generated by Django 5.2.7 on 2026-01-25 12:08
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0010_platformprofile_whatsapp_access_token_and_more'),
]
operations = [
migrations.AddField(
model_name='platformprofile',
name='whatsapp_app_secret',
field=models.CharField(blank=True, help_text='App Secret or Verify Token for Webhooks.', max_length=255, verbose_name='WhatsApp App Secret (Security Key)'),
),
]

View File

@ -0,0 +1,28 @@
# Generated by Django 5.2.7 on 2026-01-25 12:33
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0011_platformprofile_whatsapp_app_secret'),
]
operations = [
migrations.AlterField(
model_name='platformprofile',
name='whatsapp_access_token',
field=models.TextField(blank=True, help_text='Your Wablas API Token.', verbose_name='Wablas API Token'),
),
migrations.AlterField(
model_name='platformprofile',
name='whatsapp_app_secret',
field=models.CharField(blank=True, help_text='Your Wablas API Secret Key (if required).', max_length=255, verbose_name='Wablas Secret Key'),
),
migrations.AlterField(
model_name='platformprofile',
name='whatsapp_business_phone_number_id',
field=models.CharField(blank=True, default='https://deu.wablas.com', help_text='The Wablas API domain (e.g., https://deu.wablas.com).', max_length=100, verbose_name='Wablas Domain'),
),
]

View File

@ -5,6 +5,7 @@ from django.utils.translation import get_language
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.utils import timezone
from django.core.exceptions import ValidationError
import uuid
class Country(models.Model):
@ -160,9 +161,21 @@ class PlatformProfile(models.Model):
privacy_policy = models.TextField(_('Privacy Policy'), blank=True)
terms_conditions = models.TextField(_('Terms and Conditions'), blank=True)
# WhatsApp Configuration
whatsapp_access_token = models.TextField(_('WhatsApp Access Token'), blank=True, help_text=_("Permanent or temporary access token from Meta Business."))
whatsapp_business_phone_number_id = models.CharField(_('WhatsApp Phone Number ID'), max_length=100, blank=True, help_text=_("The Phone Number ID from WhatsApp API setup."))
# WhatsApp Configuration (Wablas Gateway)
whatsapp_access_token = models.TextField(_('Wablas API Token'), blank=True, help_text=_("Your Wablas API Token."))
whatsapp_business_phone_number_id = models.CharField(_('Wablas Domain'), max_length=100, blank=True, default="https://deu.wablas.com", help_text=_("The Wablas API domain (e.g., https://deu.wablas.com)."))
whatsapp_app_secret = models.CharField(_('Wablas Secret Key'), max_length=255, blank=True, help_text=_("Your Wablas API Secret Key (if required)."))
def save(self, *args, **kwargs):
# Auto-clean whitespace from credentials
if self.whatsapp_access_token:
self.whatsapp_access_token = self.whatsapp_access_token.strip()
if self.whatsapp_business_phone_number_id:
self.whatsapp_business_phone_number_id = self.whatsapp_business_phone_number_id.strip()
if self.whatsapp_app_secret:
self.whatsapp_app_secret = self.whatsapp_app_secret.strip()
super().save(*args, **kwargs)
def __str__(self):
return self.name

View File

@ -0,0 +1,39 @@
{% extends "admin/base_site.html" %}
{% load i18n admin_urls static admin_modify %}
{% block extrahead %}{{ block.super }}
<script src="{% url 'admin:jsi18n' %}"></script>
{{ media }}
{% endblock %}
{% block breadcrumbs %}
<div class="breadcrumbs">
<a href="{% url 'admin:index' %}">{% translate 'Home' %}</a>
&rsaquo; <a href="{% url 'admin:app_list' app_label='core' %}">Core</a>
&rsaquo; <a href="{% url 'admin:core_platformprofile_changelist' %}">Platform Profiles</a>
&rsaquo; Test WhatsApp Configuration
</div>
{% endblock %}
{% block content %}
<div id="content-main">
<form method="post">
{% csrf_token %}
<div>
<fieldset class="module aligned">
<h2>Test WhatsApp Configuration</h2>
<div class="form-row">
<label for="id_phone_number" class="required">Target Phone Number:</label>
<input type="text" name="phone_number" id="id_phone_number" value="{{ phone_number }}" required>
<div class="help">Enter the phone number (with country code) to receive the test message.</div>
</div>
</fieldset>
<div class="submit-row">
<input type="submit" value="Send Test Message" class="default" name="_save">
<a href="../" class="button closelink" style="margin-left: 10px;">Go Back</a>
</div>
</div>
</form>
</div>
{% endblock %}

View File

@ -1,5 +1,6 @@
import requests
import logging
import json
from django.conf import settings
from .models import PlatformProfile
@ -7,71 +8,112 @@ logger = logging.getLogger(__name__)
def get_whatsapp_credentials():
"""
Retrieves WhatsApp credentials from PlatformProfile (preferred) or settings.
Returns tuple: (api_key, phone_id)
Retrieves Wablas WhatsApp credentials from PlatformProfile.
Returns tuple: (api_token, secret_key, domain, source_info)
"""
# Default to settings
api_key = settings.WHATSAPP_API_KEY
phone_id = settings.WHATSAPP_PHONE_ID
# Defaults
api_token = settings.WHATSAPP_API_KEY if hasattr(settings, 'WHATSAPP_API_KEY') else ""
# We repurpose Phone ID as Domain in settings if needed, or default to Wablas DEU
domain = "https://deu.wablas.com"
secret_key = "" # Add this to settings if you want env support, but for now mostly DB
source = "Settings/Env"
# Try to fetch from PlatformProfile
try:
profile = PlatformProfile.objects.first()
if profile:
# Check for token override
if profile.whatsapp_access_token:
api_key = profile.whatsapp_access_token
api_token = profile.whatsapp_access_token.strip()
source = "Database (PlatformProfile)"
# Check for secret key override
if profile.whatsapp_app_secret:
secret_key = profile.whatsapp_app_secret.strip()
# Check for domain override
if profile.whatsapp_business_phone_number_id:
phone_id = profile.whatsapp_business_phone_number_id
domain = profile.whatsapp_business_phone_number_id.strip()
# Ensure no trailing slash
if domain.endswith('/'):
domain = domain[:-1]
except Exception as e:
logger.warning(f"Failed to fetch PlatformProfile for WhatsApp config: {e}")
return api_key, phone_id
return api_token, secret_key, domain, source
def send_whatsapp_message(phone_number, message):
"""
Sends a WhatsApp message using the configured gateway.
This implementation assumes Meta WhatsApp Business API (Graph API).
Sends a WhatsApp message using the Wablas gateway.
Returns True if successful, False otherwise.
"""
if not settings.WHATSAPP_ENABLED:
logger.info("WhatsApp notifications are disabled by settings.")
return False
success, _ = send_whatsapp_message_detailed(phone_number, message)
return success
api_key, phone_id = get_whatsapp_credentials()
def send_whatsapp_message_detailed(phone_number, message):
"""
Sends a WhatsApp message via Wablas V2 API and returns detailed status.
Returns tuple: (success: bool, response_msg: str)
"""
if not getattr(settings, 'WHATSAPP_ENABLED', True):
msg = "WhatsApp notifications are disabled by settings (WHATSAPP_ENABLED=False)."
logger.info(msg)
return False, msg
if not api_key or not phone_id:
logger.warning("WhatsApp API configuration is missing (checked PlatformProfile and settings).")
return False
api_token, secret_key, domain, source = get_whatsapp_credentials()
# Normalize phone number (ensure it has country code and no +)
if not api_token:
msg = f"WhatsApp API configuration (Token) is missing. (Source: {source})"
logger.warning(msg)
return False, msg
# Normalize phone number (Wablas expects international format without +, e.g. 628123...)
# Remove all non-digits
clean_phone = "".join(filter(str.isdigit, str(phone_number)))
url = f"https://graph.facebook.com/v17.0/{phone_id}/messages"
# Construct Authorization Header
# Wablas V2: Authorization: {$token}.{$secret_key}
# Some Wablas servers just need Token, but docs say Token.Secret
auth_header = api_token
if secret_key:
auth_header = f"{api_token}.{secret_key}"
# Endpoint V2
url = f"{domain}/api/v2/send-message"
headers = {
"Authorization": f"Bearer {api_key}",
"Authorization": auth_header,
"Content-Type": "application/json",
}
payload = {
"messaging_product": "whatsapp",
"to": clean_phone,
"type": "text",
"text": {"body": message}
"data": [
{
"phone": clean_phone,
"message": message,
"isGroup": "false",
"flag": "instant" # Priority
}
]
}
try:
response = requests.post(url, headers=headers, json=payload, timeout=10)
response = requests.post(url, headers=headers, json=payload, timeout=15)
response_data = response.json()
if response.status_code == 200:
logger.info(f"WhatsApp message sent to {clean_phone}")
return True
# Wablas success usually has status: true
if response.status_code == 200 and response_data.get('status') is not False:
logger.info(f"WhatsApp message sent to {clean_phone} via Wablas")
return True, f"Message sent successfully via Wablas. (Source: {source})"
else:
logger.error(f"WhatsApp API error: {response.status_code} - {response_data}")
return False
error_msg = f"Wablas API error (Source: {source}): {response.status_code} - {response_data}"
logger.error(error_msg)
return False, error_msg
except Exception as e:
logger.error(f"Failed to send WhatsApp message: {str(e)}")
return False
error_msg = f"Failed to send WhatsApp message via Wablas (Source: {source}): {str(e)}"
logger.error(error_msg)
return False, error_msg
def notify_shipment_created(parcel):
"""Notifies the shipper that the shipment request was received."""