Egor fureunoir Gorbunov
a91ba0c41a
Fixes: 1) Ensure correct handling of query results in category/brand auto-resolve methods; 2) Fix PO formatting issues across multiple locales. Extra: 1) Add `priority` field to Django admin for categories; 2) Suppress specific type-checking warnings in settings and utilities.
317 lines
9.8 KiB
Python
317 lines
9.8 KiB
Python
from django.conf import settings
|
||
from django.http import Http404
|
||
from django.shortcuts import get_object_or_404
|
||
from django.utils.text import slugify
|
||
from django.utils.translation import gettext_lazy as _
|
||
from django_elasticsearch_dsl import fields
|
||
from django_elasticsearch_dsl.registries import registry
|
||
from elasticsearch import NotFoundError
|
||
from elasticsearch.dsl import Q, Search
|
||
from rest_framework.request import Request
|
||
|
||
from core.models import Brand, Category, Product
|
||
|
||
SMART_FIELDS = [
|
||
"name^4",
|
||
"name.ngram^3",
|
||
"name.phonetic",
|
||
"description^2",
|
||
"description.ngram",
|
||
"description.phonetic",
|
||
"name.auto^4",
|
||
"description.auto^2",
|
||
"brand__name^2",
|
||
"brand__name.ngram",
|
||
"brand__name.auto",
|
||
"category__name^2",
|
||
"category__name.ngram",
|
||
"category__name.auto",
|
||
"title^4",
|
||
"title.ngram^3",
|
||
"title.phonetic",
|
||
"title.auto^4",
|
||
]
|
||
|
||
functions = [
|
||
# — product boost whenever hitting the products` index —
|
||
{
|
||
"filter": Q("term", **{"_index": "products"}),
|
||
"field_value_factor": {
|
||
"field": "brand_priority",
|
||
"modifier": "log1p",
|
||
"factor": 1,
|
||
"missing": 0,
|
||
},
|
||
},
|
||
{
|
||
"filter": Q("term", **{"_index": "products"}),
|
||
"field_value_factor": {
|
||
"field": "rating",
|
||
"modifier": "log1p",
|
||
"factor": 1,
|
||
"missing": 0,
|
||
},
|
||
},
|
||
{
|
||
"filter": Q("term", **{"_index": "products"}),
|
||
"field_value_factor": {
|
||
"field": "total_orders",
|
||
"modifier": "log1p",
|
||
"factor": 1,
|
||
"missing": 0,
|
||
},
|
||
},
|
||
{
|
||
"filter": Q("term", **{"_index": "products"}),
|
||
"field_value_factor": {
|
||
"field": "category_priority",
|
||
"modifier": "log1p",
|
||
"factor": 1,
|
||
"missing": 0,
|
||
},
|
||
},
|
||
|
||
# — category boost whenever hitting the categories` index —
|
||
{
|
||
"filter": Q("term", **{"_index": "categories"}),
|
||
"field_value_factor": {
|
||
"field": "priority",
|
||
"modifier": "log1p",
|
||
"factor": 2, # you can tweak
|
||
"missing": 0,
|
||
},
|
||
},
|
||
|
||
# — brand boost whenever hitting the brands` index —
|
||
{
|
||
"filter": Q("term", **{"_index": "brands"}),
|
||
"field_value_factor": {
|
||
"field": "priority",
|
||
"modifier": "log1p",
|
||
"factor": 2,
|
||
"missing": 0,
|
||
},
|
||
},
|
||
]
|
||
|
||
|
||
def process_query(query: str = "", request: Request | None = None) -> dict[str, list[dict]] | None:
|
||
"""
|
||
Perform a lenient, typo‑tolerant, multi‑index search.
|
||
|
||
* Full‑text with fuzziness for spelling mistakes
|
||
* `bool_prefix` for edge‑ngram autocomplete / “icontains”
|
||
"""
|
||
if not query:
|
||
raise ValueError(_("no search term provided."))
|
||
|
||
query = query.strip()
|
||
try:
|
||
query_base = Q(
|
||
"bool",
|
||
should=[
|
||
Q(
|
||
"multi_match",
|
||
query=query,
|
||
fields=SMART_FIELDS,
|
||
fuzziness="AUTO",
|
||
operator="and",
|
||
),
|
||
Q(
|
||
"multi_match",
|
||
query=query,
|
||
fields=[f for f in SMART_FIELDS if f.endswith(".auto")],
|
||
type="bool_prefix",
|
||
),
|
||
],
|
||
minimum_should_match=1,
|
||
)
|
||
|
||
function_score_query = Q(
|
||
"function_score",
|
||
query=query_base,
|
||
functions=functions,
|
||
boost_mode="multiply",
|
||
score_mode="first",
|
||
)
|
||
|
||
search = Search(index=["products", "categories", "brands", "posts"]).query(function_score_query).extra(size=100)
|
||
response = search.execute()
|
||
|
||
results: dict = {"products": [], "categories": [], "brands": [], "posts": []}
|
||
for hit in response.hits:
|
||
obj_uuid = getattr(hit, "uuid", None) or hit.meta.id
|
||
obj_name = getattr(hit, "name", None) or getattr(hit, "title", None) or "N/A"
|
||
obj_slug = ""
|
||
raw_slug = getattr(hit, "slug", None)
|
||
if raw_slug:
|
||
obj_slug = raw_slug
|
||
elif hit.meta.index == "brands":
|
||
obj_slug = slugify(obj_name)
|
||
elif hit.meta.index == "categories":
|
||
obj_slug = slugify(f"{obj_name}")
|
||
|
||
image_url = None
|
||
idx = hit.meta.index
|
||
if idx == "products" and request:
|
||
prod = get_object_or_404(Product, uuid=obj_uuid)
|
||
first = prod.images.order_by("priority").first()
|
||
if first and first.image:
|
||
image_url = request.build_absolute_uri(first.image.url)
|
||
elif idx == "brands" and request:
|
||
brand = get_object_or_404(Brand, uuid=obj_uuid)
|
||
if brand.small_logo:
|
||
image_url = request.build_absolute_uri(brand.small_logo.url)
|
||
elif idx == "categories" and request:
|
||
cat = get_object_or_404(Category, uuid=obj_uuid)
|
||
if cat.image:
|
||
image_url = request.build_absolute_uri(cat.image.url)
|
||
|
||
hit_result = {
|
||
"uuid": str(obj_uuid),
|
||
"name": obj_name,
|
||
"slug": obj_slug,
|
||
"image": image_url,
|
||
}
|
||
|
||
if settings.DEBUG:
|
||
if idx == "products":
|
||
hit_result["rating_debug"] = getattr(hit, "rating", 0)
|
||
hit_result["total_orders_debug"] = getattr(hit, "total_orders", 0)
|
||
hit_result["brand_priority_debug"] = getattr(hit, "brand_priority", 0)
|
||
hit_result["category_priority_debug"] = getattr(hit, "category_priority", 0)
|
||
if idx == "brands":
|
||
hit_result["priority_debug"] = getattr(hit, "priority", 0)
|
||
if idx == "categories":
|
||
hit_result["priority_debug"] = getattr(hit, "priority", 0)
|
||
if idx == "posts":
|
||
pass
|
||
|
||
results[idx].append(hit_result)
|
||
|
||
return results
|
||
except NotFoundError:
|
||
raise Http404
|
||
|
||
|
||
LANGUAGE_ANALYZER_MAP = {
|
||
"ar": "arabic",
|
||
"cs": "czech",
|
||
"da": "danish",
|
||
"de": "german",
|
||
"en": "english",
|
||
"es": "spanish",
|
||
"fr": "french",
|
||
"hi": "hindi",
|
||
"it": "italian",
|
||
"ja": "standard",
|
||
"kk": "standard",
|
||
"nl": "dutch",
|
||
"pl": "standard",
|
||
"pt": "portuguese",
|
||
"ro": "romanian",
|
||
"ru": "russian",
|
||
"zh": "standard",
|
||
}
|
||
|
||
|
||
def _lang_analyzer(lang_code: str) -> str:
|
||
"""Return the best‑guess ES analyzer for an ISO language code."""
|
||
base = lang_code.split("-")[0].lower()
|
||
return LANGUAGE_ANALYZER_MAP.get(base, "standard")
|
||
|
||
|
||
class ActiveOnlyMixin:
|
||
"""QuerySet & indexing helpers, so only *active* objects are indexed."""
|
||
|
||
def get_queryset(self):
|
||
return super().get_queryset().filter(is_active=True)
|
||
|
||
def should_index_object(self, obj):
|
||
return getattr(obj, "is_active", False)
|
||
|
||
|
||
COMMON_ANALYSIS = {
|
||
"filter": {
|
||
"edge_ngram_filter": {"type": "edge_ngram", "min_gram": 1, "max_gram": 20},
|
||
"ngram_filter": {"type": "ngram", "min_gram": 2, "max_gram": 20},
|
||
"double_metaphone": {
|
||
"type": "phonetic",
|
||
"encoder": "double_metaphone",
|
||
"replace": False,
|
||
},
|
||
},
|
||
"analyzer": {
|
||
"autocomplete": {
|
||
"tokenizer": "standard",
|
||
"filter": ["lowercase", "asciifolding", "edge_ngram_filter"],
|
||
},
|
||
"autocomplete_search": {
|
||
"tokenizer": "standard",
|
||
"filter": ["lowercase", "asciifolding"],
|
||
},
|
||
"name_ngram": {
|
||
"tokenizer": "standard",
|
||
"filter": ["lowercase", "asciifolding", "ngram_filter"],
|
||
},
|
||
"name_phonetic": {
|
||
"tokenizer": "standard",
|
||
"filter": ["lowercase", "asciifolding", "double_metaphone"],
|
||
},
|
||
"query_lc": {"tokenizer": "standard", "filter": ["lowercase", "asciifolding"]},
|
||
},
|
||
}
|
||
|
||
|
||
def _add_multilang_fields(cls):
|
||
"""
|
||
Dynamically add multilingual name/description fields and prepare methods to guard against None.
|
||
"""
|
||
for code, _lang in settings.LANGUAGES:
|
||
lc = code.replace("-", "_").lower()
|
||
# name_{lc}
|
||
name_field = f"name_{lc}"
|
||
setattr(
|
||
cls,
|
||
name_field,
|
||
fields.TextField(
|
||
attr=name_field,
|
||
analyzer=_lang_analyzer(code),
|
||
copy_to="name",
|
||
fields={
|
||
"raw": fields.KeywordField(ignore_above=256),
|
||
"ngram": fields.TextField(analyzer="name_ngram", search_analyzer="query_lc"),
|
||
"phonetic": fields.TextField(analyzer="name_phonetic"),
|
||
},
|
||
),
|
||
)
|
||
|
||
# prepare_name_{lc} to ensure no None values
|
||
def make_prepare(attr):
|
||
return lambda self, instance: getattr(instance, attr, "") or ""
|
||
|
||
setattr(cls, f"prepare_{name_field}", make_prepare(name_field))
|
||
|
||
# description_{lc}
|
||
desc_field = f"description_{lc}"
|
||
setattr(
|
||
cls,
|
||
desc_field,
|
||
fields.TextField(
|
||
attr=desc_field,
|
||
analyzer=_lang_analyzer(code),
|
||
copy_to="description",
|
||
fields={
|
||
"raw": fields.KeywordField(ignore_above=256),
|
||
"ngram": fields.TextField(analyzer="name_ngram", search_analyzer="query_lc"),
|
||
"phonetic": fields.TextField(analyzer="name_phonetic"),
|
||
},
|
||
),
|
||
)
|
||
setattr(cls, f"prepare_{desc_field}", make_prepare(desc_field))
|
||
|
||
|
||
def populate_index():
|
||
for doc in registry.get_documents(set(registry.get_models())):
|
||
qs = doc().get_indexing_queryset()
|
||
doc().update(qs, parallel=True, refresh=True)
|