109 lines
4.4 KiB
Python
109 lines
4.4 KiB
Python
import typesense
|
|
from django.conf import settings
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PRODUCTS_SCHEMA = {
|
|
'name': 'products',
|
|
'fields': [
|
|
{'name': 'id', 'type': 'string'},
|
|
{'name': 'title', 'type': 'string'},
|
|
{'name': 'description', 'type': 'string', 'optional': True},
|
|
{'name': 'category', 'type': 'string', 'facet': True},
|
|
{'name': 'supplier_id', 'type': 'int32', 'facet': True},
|
|
{'name': 'supplier_name', 'type': 'string', 'optional': True},
|
|
{'name': 'price', 'type': 'float', 'facet': True},
|
|
{'name': 'compare_at_price', 'type': 'float', 'optional': True},
|
|
{'name': 'stock', 'type': 'int32'},
|
|
{'name': 'sku', 'type': 'string'},
|
|
{'name': 'image', 'type': 'string', 'optional': True},
|
|
{'name': 'is_gi_tagged', 'type': 'bool', 'facet': True},
|
|
{'name': 'is_active', 'type': 'bool', 'facet': True},
|
|
{'name': 'status', 'type': 'string', 'facet': True},
|
|
{'name': 'created_at', 'type': 'int64', 'optional': True},
|
|
],
|
|
'default_sorting_field': 'created_at'
|
|
}
|
|
|
|
class TypesenseService:
|
|
_client = None
|
|
|
|
@classmethod
|
|
def get_client(cls):
|
|
if cls._client is None:
|
|
cfg = getattr(settings, 'TYPESENSE_CONFIG', {})
|
|
cls._client = typesense.Client({
|
|
'nodes': [{
|
|
'host': cfg.get('HOST', '127.0.0.1'),
|
|
'port': str(cfg.get('PORT', '8108')),
|
|
'protocol': cfg.get('PROTOCOL', 'http')
|
|
}],
|
|
'api_key': cfg.get('API_KEY', 'vignesh_typesense_master_key_2026'),
|
|
'connection_timeout_seconds': cfg.get('TIMEOUT', 5.0)
|
|
})
|
|
return cls._client
|
|
|
|
@classmethod
|
|
def ensure_schema(cls):
|
|
client = cls.get_client()
|
|
try:
|
|
client.collections['products'].retrieve()
|
|
logger.info("Typesense 'products' collection already exists.")
|
|
except Exception:
|
|
logger.info("Creating Typesense 'products' collection schema...")
|
|
client.collections.create(PRODUCTS_SCHEMA)
|
|
|
|
@classmethod
|
|
def index_product(cls, product_dict):
|
|
cls.ensure_schema()
|
|
client = cls.get_client()
|
|
doc = {
|
|
'id': str(product_dict.get('id', product_dict.get('sku'))),
|
|
'title': str(product_dict.get('title', '')),
|
|
'description': str(product_dict.get('description', '') or ''),
|
|
'category': str(product_dict.get('category', 'Uncategorized')),
|
|
'supplier_id': int(product_dict.get('supplier_id', 0)),
|
|
'supplier_name': str(product_dict.get('supplier_name', '') or ''),
|
|
'price': float(product_dict.get('price', 0.0)),
|
|
'compare_at_price': float(product_dict.get('compare_at_price', 0.0) or 0.0),
|
|
'stock': int(product_dict.get('stock', 0)),
|
|
'sku': str(product_dict.get('sku', '')),
|
|
'image': str(product_dict.get('image', '') or ''),
|
|
'is_gi_tagged': bool(product_dict.get('is_gi_tagged', False)),
|
|
'is_active': bool(product_dict.get('is_active', True)),
|
|
'status': str(product_dict.get('status', 'active')),
|
|
'created_at': int(product_dict.get('created_at', 0))
|
|
}
|
|
return client.collections['products'].documents.upsert(doc)
|
|
|
|
@classmethod
|
|
def delete_product(cls, product_id):
|
|
client = cls.get_client()
|
|
try:
|
|
return client.collections['products'].documents[str(product_id)].delete()
|
|
except Exception as e:
|
|
logger.warning(f"Could not delete document {product_id} from Typesense: {e}")
|
|
return None
|
|
|
|
@classmethod
|
|
def search_products(cls, query="*", filter_by="", sort_by="", page=1, per_page=20):
|
|
cls.ensure_schema()
|
|
client = cls.get_client()
|
|
search_parameters = {
|
|
'q': query if query else '*',
|
|
'query_by': 'title,description,category,sku',
|
|
'facet_by': 'category,is_gi_tagged,supplier_id',
|
|
'page': page,
|
|
'per_page': per_page,
|
|
'num_typos': 2,
|
|
'prefix': True
|
|
}
|
|
if filter_by:
|
|
search_parameters['filter_by'] = filter_by
|
|
if sort_by:
|
|
search_parameters['sort_by'] = sort_by
|
|
else:
|
|
search_parameters['sort_by'] = 'created_at:desc'
|
|
|
|
return client.collections['products'].documents.search(search_parameters)
|