erpnext/e_commerce/redisearch_utils.py

# Copyright (c) 2021, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt

import frappe
from frappe.utils.redis_wrapper import RedisWrapper
from redisearch import AutoCompleter, Client, IndexDefinition, Suggestion, TagField, TextField

WEBSITE_ITEM_INDEX = "website_items_index"
WEBSITE_ITEM_KEY_PREFIX = "website_item:"
WEBSITE_ITEM_NAME_AUTOCOMPLETE = "website_items_name_dict"
WEBSITE_ITEM_CATEGORY_AUTOCOMPLETE = "website_items_category_dict"


def get_indexable_web_fields():
	"Return valid fields from Website Item that can be searched for."
	web_item_meta = frappe.get_meta("Website Item", cached=True)
	valid_fields = filter(
		lambda df: df.fieldtype in ("Link", "Table MultiSelect", "Data", "Small Text", "Text Editor"),
		web_item_meta.fields,
	)

	return [df.fieldname for df in valid_fields]


def is_search_module_loaded():
	try:
		cache = frappe.cache()
		out = cache.execute_command("MODULE LIST")

		parsed_output = " ".join(
			(" ".join([s.decode() for s in o if not isinstance(s, int)]) for o in out)
		)
		return "search" in parsed_output
	except Exception:
		return False


def if_redisearch_loaded(function):
	"Decorator to check if Redisearch is loaded."

	def wrapper(*args, **kwargs):
		if is_search_module_loaded():
			func = function(*args, **kwargs)
			return func
		return

	return wrapper


def make_key(key):
	return "{0}|{1}".format(frappe.conf.db_name, key).encode("utf-8")


@if_redisearch_loaded
def create_website_items_index():
	"Creates Index Definition."

	# CREATE index
	client = Client(make_key(WEBSITE_ITEM_INDEX), conn=frappe.cache())

	# DROP if already exists
	try:
		client.drop_index()
	except Exception:
		pass

	idx_def = IndexDefinition([make_key(WEBSITE_ITEM_KEY_PREFIX)])

	# Based on e-commerce settings
	idx_fields = frappe.db.get_single_value("E Commerce Settings", "search_index_fields")
	idx_fields = idx_fields.split(",") if idx_fields else []

	if "web_item_name" in idx_fields:
		idx_fields.remove("web_item_name")

	idx_fields = list(map(to_search_field, idx_fields))

	client.create_index(
		[TextField("web_item_name", sortable=True)] + idx_fields,
		definition=idx_def,
	)

	reindex_all_web_items()
	define_autocomplete_dictionary()


def to_search_field(field):
	if field == "tags":
		return TagField("tags", separator=",")

	return TextField(field)


@if_redisearch_loaded
def insert_item_to_index(website_item_doc):
	# Insert item to index
	key = get_cache_key(website_item_doc.name)
	cache = frappe.cache()
	web_item = create_web_item_map(website_item_doc)

	for k, v in web_item.items():
		super(RedisWrapper, cache).hset(make_key(key), k, v)

	insert_to_name_ac(website_item_doc.web_item_name, website_item_doc.name)


@if_redisearch_loaded
def insert_to_name_ac(web_name, doc_name):
	ac = AutoCompleter(make_key(WEBSITE_ITEM_NAME_AUTOCOMPLETE), conn=frappe.cache())
	ac.add_suggestions(Suggestion(web_name, payload=doc_name))


def create_web_item_map(website_item_doc):
	fields_to_index = get_fields_indexed()
	web_item = {}

	for f in fields_to_index:
		web_item[f] = website_item_doc.get(f) or ""

	return web_item


@if_redisearch_loaded
def update_index_for_item(website_item_doc):
	# Reinsert to Cache
	insert_item_to_index(website_item_doc)
	define_autocomplete_dictionary()


@if_redisearch_loaded
def delete_item_from_index(website_item_doc):
	cache = frappe.cache()
	key = get_cache_key(website_item_doc.name)

	try:
		cache.delete(key)
	except Exception:
		return False

	delete_from_ac_dict(website_item_doc)
	return True


@if_redisearch_loaded
def delete_from_ac_dict(website_item_doc):
	"""Removes this items's name from autocomplete dictionary"""
	cache = frappe.cache()
	name_ac = AutoCompleter(make_key(WEBSITE_ITEM_NAME_AUTOCOMPLETE), conn=cache)
	name_ac.delete(website_item_doc.web_item_name)


@if_redisearch_loaded
def define_autocomplete_dictionary():
	"""Creates an autocomplete search dictionary for `name`.
	Also creats autocomplete dictionary for `categories` if
	checked in E Commerce Settings"""

	cache = frappe.cache()
	name_ac = AutoCompleter(make_key(WEBSITE_ITEM_NAME_AUTOCOMPLETE), conn=cache)
	cat_ac = AutoCompleter(make_key(WEBSITE_ITEM_CATEGORY_AUTOCOMPLETE), conn=cache)

	ac_categories = frappe.db.get_single_value(
		"E Commerce Settings", "show_categories_in_search_autocomplete"
	)

	# Delete both autocomplete dicts
	try:
		cache.delete(make_key(WEBSITE_ITEM_NAME_AUTOCOMPLETE))
		cache.delete(make_key(WEBSITE_ITEM_CATEGORY_AUTOCOMPLETE))
	except Exception:
		return False

	items = frappe.get_all(
		"Website Item", fields=["web_item_name", "item_group"], filters={"published": 1}
	)

	for item in items:
		name_ac.add_suggestions(Suggestion(item.web_item_name))
		if ac_categories and item.item_group:
			cat_ac.add_suggestions(Suggestion(item.item_group))

	return True


@if_redisearch_loaded
def reindex_all_web_items():
	items = frappe.get_all("Website Item", fields=get_fields_indexed(), filters={"published": True})

	cache = frappe.cache()
	for item in items:
		web_item = create_web_item_map(item)
		key = make_key(get_cache_key(item.name))

		for k, v in web_item.items():
			super(RedisWrapper, cache).hset(key, k, v)


def get_cache_key(name):
	name = frappe.scrub(name)
	return f"{WEBSITE_ITEM_KEY_PREFIX}{name}"


def get_fields_indexed():
	fields_to_index = frappe.db.get_single_value("E Commerce Settings", "search_index_fields")
	fields_to_index = fields_to_index.split(",") if fields_to_index else []

	mandatory_fields = ["name", "web_item_name", "route", "thumbnail", "ranking"]
	fields_to_index = fields_to_index + mandatory_fields

	return fields_to_index


# TODO: Remove later
# # Figure out a way to run this at startup
define_autocomplete_dictionary()
create_website_items_index()