searchengine/api/typesense_client.py

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'
}

def safe_str(val, default=""):
    if val is None:
        return default
    return str(val)

def safe_int(val, default=0):
    if val is None or val == "":
        return default
    try:
        return int(val)
    except (ValueError, TypeError):
        return default

def safe_float(val, default=0.0):
    if val is None or val == "":
        return default
    try:
        return float(val)
    except (ValueError, TypeError):
        return default

def safe_bool(val, default=False):
    if val is None:
        return default
    if isinstance(val, bool):
        return val
    if isinstance(val, (int, float)):
        return bool(val)
    if isinstance(val, str):
        return val.lower() in ("true", "1", "yes")
    return default


class TypesenseService:
    _client = None
    _schema_initialized = False

    @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, force=False):
        if cls._schema_initialized and not force:
            return
        client = cls.get_client()
        try:
            client.collections['products'].retrieve()
            logger.info("Typesense 'products' collection already exists.")
            cls._schema_initialized = True
        except Exception:
            logger.info("Creating Typesense 'products' collection schema...")
            client.collections.create(PRODUCTS_SCHEMA)
            cls._schema_initialized = True

    @classmethod
    def prepare_product_doc(cls, product_dict):
        doc_id = product_dict.get('id')
        if doc_id is None:
            doc_id = product_dict.get('sku')

        return {
            'id': safe_str(doc_id),
            'title': safe_str(product_dict.get('title')),
            'description': safe_str(product_dict.get('description')),
            'category': safe_str(product_dict.get('category'), 'Uncategorized'),
            'supplier_id': safe_int(product_dict.get('supplier_id'), 0),
            'supplier_name': safe_str(product_dict.get('supplier_name')),
            'price': safe_float(product_dict.get('price'), 0.0),
            'compare_at_price': safe_float(product_dict.get('compare_at_price'), 0.0),
            'stock': safe_int(product_dict.get('stock'), 0),
            'sku': safe_str(product_dict.get('sku')),
            'image': safe_str(product_dict.get('image')),
            'is_gi_tagged': safe_bool(product_dict.get('is_gi_tagged'), False),
            'is_active': safe_bool(product_dict.get('is_active'), True),
            'status': safe_str(product_dict.get('status'), 'active'),
            'created_at': safe_int(product_dict.get('created_at'), 0)
        }

    @classmethod
    def index_product(cls, product_dict):
        cls.ensure_schema()
        client = cls.get_client()
        doc = cls.prepare_product_doc(product_dict)
        return client.collections['products'].documents.upsert(doc)

    @classmethod
    def index_products_batch(cls, products_list):
        if not products_list:
            return []
        cls.ensure_schema()
        client = cls.get_client()
        docs = [cls.prepare_product_doc(p) for p in products_list]
        return client.collections['products'].documents.import_(docs, {'action': 'upsert'})

    @classmethod
    def delete_product(cls, product_id):
        client = cls.get_client()
        pid_str = str(product_id).strip()
        try:
            return client.collections['products'].documents[pid_str].delete()
        except Exception as e:
            logger.info(f"Direct delete by ID {pid_str} failed ({e}), attempting delete by SKU filter...")
            try:
                return client.collections['products'].documents.delete({'filter_by': f'sku:="{pid_str}"'})
            except Exception as e_filter:
                logger.warning(f"Could not delete document with identifier {pid_str} from Typesense: {e_filter}")
                return None

    @classmethod
    def search_products(cls, query="*", filter_by="", sort_by="", page=1, per_page=20):
        cls.ensure_schema()
        client = cls.get_client()
        q_clean = query.strip() if query else "*"

        search_parameters = {
            'q': q_clean if q_clean 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:
            if q_clean and q_clean != '*':
                search_parameters['sort_by'] = '_text_match:desc,created_at:desc'
            else:
                search_parameters['sort_by'] = 'created_at:desc'

        return client.collections['products'].documents.search(search_parameters)