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")