125 lines
4.9 KiB
Python
125 lines
4.9 KiB
Python
import re
|
||
|
||
from django.db.models import Case, IntegerField, Q, Value, When
|
||
from django.http import JsonResponse
|
||
from django.views.decorators.http import require_GET
|
||
|
||
from apps.customers.models import CustomerProfile
|
||
|
||
from .models import PhoneContactApiKey
|
||
|
||
|
||
_PHONE_DIGITS = str.maketrans("۰۱۲۳۴۵۶۷۸۹٠١٢٣٤٥٦٧٨٩", "01234567890123456789")
|
||
|
||
|
||
def _dialable_number(value):
|
||
number = re.sub(r"[\s().-]", "", str(value or "").translate(_PHONE_DIGITS))
|
||
return number if 0 < len(number) <= 64 and re.fullmatch(r"[0-9+*#]+", number) else ""
|
||
|
||
|
||
def _canonical_number(value):
|
||
number = _dialable_number(value)
|
||
if not number or not re.fullmatch(r"\+[0-9]+|[0-9*#]+", number):
|
||
return ""
|
||
if number.isdigit() and number.startswith("00") and len(number) > 2:
|
||
return "+" + number[2:]
|
||
if number.isdigit() and number.startswith("98") and len(number) >= 12:
|
||
return "+" + number
|
||
if number.isdigit() and number.startswith("0") and len(number) > 1:
|
||
return "+98" + number[1:]
|
||
if number.isdigit() and len(number) == 10:
|
||
return "+98" + number
|
||
return number
|
||
|
||
|
||
def _phone_variants(value):
|
||
canonical = _canonical_number(value)
|
||
if not canonical:
|
||
return ()
|
||
variants = {canonical, _dialable_number(value)}
|
||
if canonical.startswith("+98"):
|
||
variants.update({"0" + canonical[3:], "00" + canonical[1:], canonical[1:]})
|
||
elif canonical.startswith("+"):
|
||
variants.add("00" + canonical[1:])
|
||
return variants
|
||
|
||
|
||
def _display_number(value):
|
||
canonical = _canonical_number(value)
|
||
return "0" + canonical[3:] if canonical.startswith("+98") else canonical
|
||
|
||
|
||
def _json(data, status=200):
|
||
response = JsonResponse(data, status=status, json_dumps_params={"ensure_ascii": False})
|
||
response["Cache-Control"] = "no-store"
|
||
return response
|
||
|
||
|
||
def _matching_contacts(contacts, query):
|
||
phone_query = query.translate(_PHONE_DIGITS)
|
||
if re.fullmatch(r"[+*#0-9().\s-]+", phone_query):
|
||
variants = _phone_variants(phone_query)
|
||
return contacts.filter(user__mobile__in=variants).order_by("pk") if variants else contacts.none()
|
||
|
||
for word in re.split(r"\s+", query):
|
||
contacts = contacts.filter(
|
||
Q(first_name__icontains=word) | Q(last_name__icontains=word) |
|
||
Q(organization_name__icontains=word) | Q(user__first_name__icontains=word) |
|
||
Q(user__last_name__icontains=word) | Q(user__mobile__icontains=word)
|
||
)
|
||
|
||
preferred = (
|
||
Q(first_name__iexact=query) | Q(last_name__iexact=query) |
|
||
Q(organization_name__iexact=query) | Q(user__first_name__iexact=query) |
|
||
Q(user__last_name__iexact=query)
|
||
)
|
||
starts_with = (
|
||
Q(first_name__istartswith=query) | Q(last_name__istartswith=query) |
|
||
Q(organization_name__istartswith=query) | Q(user__first_name__istartswith=query) |
|
||
Q(user__last_name__istartswith=query)
|
||
)
|
||
return contacts.annotate(_search_rank=Case(
|
||
When(preferred, then=Value(0)),
|
||
When(starts_with, then=Value(1)),
|
||
default=Value(2), output_field=IntegerField(),
|
||
)).order_by("_search_rank", "pk")
|
||
|
||
|
||
@require_GET
|
||
def phone_contacts(request):
|
||
authorization = request.headers.get("Authorization", "")
|
||
scheme, separator, token = authorization.partition(" ")
|
||
if not separator or scheme.lower() != "bearer" or not token or token != token.strip():
|
||
return _json({"error": "unauthorized"}, 401)
|
||
|
||
credential = PhoneContactApiKey.objects.filter(pk=1, is_active=True).first()
|
||
if not credential or not credential.accepts(token):
|
||
return _json({"error": "unauthorized"}, 401)
|
||
|
||
query = request.GET.get("search", "").strip()[:80]
|
||
if len(query) < 2:
|
||
return _json({"contacts": []})
|
||
|
||
try:
|
||
limit = min(50, max(1, int(request.GET.get("limit", "50"))))
|
||
except ValueError:
|
||
return _json({"error": "invalid limit"}, 400)
|
||
|
||
query = query.translate(str.maketrans({"ي": "ی", "ك": "ک", "۰": "0", "۱": "1", "۲": "2", "۳": "3", "۴": "4", "۵": "5", "۶": "6", "۷": "7", "۸": "8", "۹": "9"}))
|
||
contacts = CustomerProfile.objects.select_related("user").filter(
|
||
user__account_type="CUSTOMER", user__is_active=True, user__mobile__isnull=False
|
||
).exclude(user__mobile="").filter(
|
||
Q(first_name__gt="") | Q(last_name__gt="") | Q(organization_name__gt="") |
|
||
Q(user__first_name__gt="") | Q(user__last_name__gt="")
|
||
)
|
||
results = []
|
||
for profile in _matching_contacts(contacts, query)[:limit]:
|
||
name = (f"{profile.first_name} {profile.last_name}".strip() or
|
||
f"{profile.user.first_name} {profile.user.last_name}".strip() or
|
||
profile.organization_name.strip())[:120]
|
||
number = _display_number(profile.user.mobile)
|
||
if name and number:
|
||
label = "همراه" if re.fullmatch(r"09[0-9]{9}", number) else "تلفن"
|
||
results.append({"name": name, "mobile": number, "phones": [{"type": label, "number": number}]})
|
||
return _json({"contacts": results})
|