168 lines
8.2 KiB
Python
168 lines
8.2 KiB
Python
from datetime import datetime
|
|
from decimal import Decimal, InvalidOperation
|
|
import re
|
|
|
|
from django.db import connection
|
|
from django.db.models import CharField, Q, QuerySet
|
|
from django.db.models.functions import Coalesce
|
|
|
|
from .models import Device, FieldDefinition
|
|
|
|
|
|
SORT_FIELDS = {
|
|
"number": "number", "date": "date", "shelf": "shelf", "status": "status__label", "stage": "stage",
|
|
"device_type": "device_type__label", "price": "price", "ram_gb": "_ram_gb", "ssd_gb": "_ssd_gb",
|
|
"hdd_gb": "_hdd_gb", "display_inches": "_display_inches", "cpu": "_cpu", "model": "model",
|
|
"employee": "employee__code", "recipient": "recipient__label", "ticket": "ticket",
|
|
}
|
|
COMPUTER_RELATIONS = ("laptop_specification", "desktop_specification", "aio_specification", "computer_specification")
|
|
SPECIFICATION_RELATIONS = (*COMPUTER_RELATIONS, "monitor_specification", "printer_specification")
|
|
|
|
|
|
def computer_field_query(field, lookup, value):
|
|
query = Q()
|
|
for relation in COMPUTER_RELATIONS:
|
|
query |= Q(**{f"{relation}__{field}__{lookup}": value})
|
|
return query
|
|
|
|
|
|
def cpu_patterns(*patterns):
|
|
query = Q()
|
|
for lookup, value in patterns:
|
|
query |= computer_field_query("cpu", lookup, value)
|
|
return query
|
|
|
|
CPU_VENDOR_QUERIES = {
|
|
"intel": cpu_patterns(("icontains", "intel"), ("istartswith", "core "), ("istartswith", "i3"), ("istartswith", "i5"), ("istartswith", "i7"), ("istartswith", "i9")),
|
|
"amd": cpu_patterns(("icontains", "amd"), ("icontains", "ryzen"), ("icontains", "threadripper"), ("istartswith", "fx-")),
|
|
}
|
|
CPU_FAMILY_QUERIES = {
|
|
"intel_i3": cpu_patterns(("icontains", "core i3"), ("istartswith", "i3")),
|
|
"intel_i5": cpu_patterns(("icontains", "core i5"), ("istartswith", "i5")),
|
|
"intel_i7": cpu_patterns(("icontains", "core i7"), ("istartswith", "i7")),
|
|
"intel_i9": cpu_patterns(("icontains", "core i9"), ("istartswith", "i9")),
|
|
"intel_ultra_5": cpu_patterns(("icontains", "core ultra 5")),
|
|
"intel_ultra_7": cpu_patterns(("icontains", "core ultra 7")),
|
|
"intel_ultra_9": cpu_patterns(("icontains", "core ultra 9")),
|
|
"amd_ryzen_3": cpu_patterns(("icontains", "ryzen 3 ")),
|
|
"amd_ryzen_5": cpu_patterns(("icontains", "ryzen 5 ")),
|
|
"amd_ryzen_7": cpu_patterns(("icontains", "ryzen 7 ")),
|
|
"amd_ryzen_9": cpu_patterns(("icontains", "ryzen 9 ")),
|
|
"amd_ryzen_ai": cpu_patterns(("icontains", "ryzen ai")),
|
|
"amd_threadripper": cpu_patterns(("icontains", "threadripper")),
|
|
"amd_fx": cpu_patterns(("icontains", "amd fx-"), ("istartswith", "fx-")),
|
|
}
|
|
|
|
|
|
def parse_filter_date(value):
|
|
for date_format in ("%d.%m.%Y", "%Y-%m-%d"):
|
|
try:
|
|
return datetime.strptime(str(value), date_format).date()
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def device_queryset(params, archived: bool = False) -> QuerySet[Device]:
|
|
qs = Device.objects.select_related("status", "recipient", "location", "device_type", "employee", *SPECIFICATION_RELATIONS).annotate(
|
|
_display_inches=Coalesce("laptop_specification__display_inches", "aio_specification__display_inches", "monitor_specification__display_inches", "computer_specification__display_inches"),
|
|
_ram_gb=Coalesce(*(f"{name}__ram_gb" for name in COMPUTER_RELATIONS)),
|
|
_ssd_gb=Coalesce(*(f"{name}__ssd_gb" for name in COMPUTER_RELATIONS)),
|
|
_hdd_gb=Coalesce(*(f"{name}__hdd_gb" for name in COMPUTER_RELATIONS)),
|
|
_cpu=Coalesce(*(f"{name}__cpu" for name in COMPUTER_RELATIONS), output_field=CharField()),
|
|
)
|
|
qs = qs.filter(archived_at__isnull=not archived)
|
|
search = params.get("q", "").strip()
|
|
if search:
|
|
if search.isdigit() and not search.startswith("0") and len(search) <= 10:
|
|
prefix = int(search)
|
|
if params.get("exact") == "1":
|
|
qs = qs.filter(number=prefix)
|
|
else:
|
|
number_query = Q(number=prefix)
|
|
for total_digits in range(len(search) + 1, 11):
|
|
factor = 10 ** (total_digits - len(search))
|
|
number_query |= Q(number__gte=prefix * factor, number__lt=(prefix + 1) * factor)
|
|
qs = qs.filter(number_query)
|
|
else:
|
|
qs = qs.none()
|
|
for param, field in [("status", "status_id"), ("type", "device_type_id"), ("employee", "employee_id"), ("recipient", "recipient_id")]:
|
|
values = [v for v in params.getlist(param) if v]
|
|
if values:
|
|
if any(not v.isascii() or not v.isdigit() or len(v) > 19 or int(v) > 9223372036854775807 for v in values):
|
|
return qs.none()
|
|
qs = qs.filter(**{f"{field}__in": values})
|
|
if stage := params.get("stage"):
|
|
if stage not in Device.Stage.values:
|
|
return qs.none()
|
|
qs = qs.filter(stage=stage)
|
|
for param, field in [("ram", "_ram_gb"), ("ssd", "_ssd_gb"), ("hdd", "_hdd_gb")]:
|
|
for suffix, lookup in [("min", "gte"), ("max", "lte"), ("exact", "exact")]:
|
|
value = params.get(f"{param}_{suffix}")
|
|
if value not in (None, ""):
|
|
if not value.isascii() or not value.isdigit() or len(value) > 10 or int(value) > 2147483647:
|
|
return qs.none()
|
|
qs = qs.filter(**{f"{field}__{lookup}": int(value)})
|
|
for param, field in [("price", "price"), ("display", "_display_inches")]:
|
|
for suffix, lookup in [("min", "gte"), ("max", "lte")]:
|
|
value = params.get(f"{param}_{suffix}")
|
|
if value not in (None, ""):
|
|
try:
|
|
numeric = Decimal(value.replace(",", "."))
|
|
if not numeric.is_finite() or abs(numeric) > Decimal("99999999.99"):
|
|
return qs.none()
|
|
except InvalidOperation:
|
|
return qs.none()
|
|
qs = qs.filter(**{f"{field}__{lookup}": numeric})
|
|
for name in ["battery", "wifi", "lan"]:
|
|
value = params.get(name)
|
|
if value:
|
|
if name == "battery":
|
|
qs = qs.filter(Q(laptop_specification__battery=value) | Q(computer_specification__battery=value))
|
|
else:
|
|
qs = qs.filter(computer_field_query(name, "exact", value) | Q(**{f"printer_specification__{name}": value}))
|
|
vendor_query = Q()
|
|
for value in params.getlist("cpu_vendor"):
|
|
if value in CPU_VENDOR_QUERIES:
|
|
vendor_query |= CPU_VENDOR_QUERIES[value]
|
|
if vendor_query:
|
|
qs = qs.filter(vendor_query)
|
|
family_query = Q()
|
|
for value in params.getlist("cpu_family"):
|
|
if value in CPU_FAMILY_QUERIES:
|
|
family_query |= CPU_FAMILY_QUERIES[value]
|
|
if family_query:
|
|
qs = qs.filter(family_query)
|
|
date_from = parse_filter_date(params.get("date_from")) if params.get("date_from") else None
|
|
date_to = parse_filter_date(params.get("date_to")) if params.get("date_to") else None
|
|
if date_from:
|
|
qs = qs.filter(date__gte=date_from)
|
|
if date_to:
|
|
qs = qs.filter(date__lte=date_to)
|
|
for name in ["model", "ticket"]:
|
|
value = params.get(f"{name}_contains", "").strip()
|
|
if value:
|
|
qs = qs.filter(**{f"{name}__icontains": value})
|
|
cpu_value = params.get("cpu_contains", "").strip()
|
|
for token in re.findall(r"[a-z]+\d*|\d+[a-z]*", cpu_value.casefold()):
|
|
qs = qs.filter(computer_field_query("cpu", "icontains", token))
|
|
for definition in FieldDefinition.objects.filter(active=True, filterable=True):
|
|
value = params.get(f"custom_{definition.key}", "").strip()
|
|
if not value:
|
|
continue
|
|
lookup = "exact" if definition.field_type in {"integer", "decimal", "boolean", "date", "select"} else "icontains"
|
|
if definition.field_type == "integer":
|
|
try:
|
|
value = int(value)
|
|
except ValueError:
|
|
continue
|
|
elif definition.field_type == "boolean":
|
|
value = value.lower() in {"1", "true", "yes", "ja"}
|
|
if lookup == "exact" and connection.vendor == "postgresql":
|
|
qs = qs.filter(custom_values__contains={definition.key: value})
|
|
else:
|
|
qs = qs.filter(**{f"custom_values__{definition.key}__{lookup}": value})
|
|
sort = params.get("sort", "number")
|
|
direction = "-" if params.get("dir", "desc") == "desc" else ""
|
|
return qs.order_by(f"{direction}{SORT_FIELDS.get(sort, 'number')}", "-number")
|