jcrm/crm/integrations/api.py

import jingrow
from jingrow.query_builder import Order
from pypika.functions import Replace

from crm.utils import are_same_phone_number, parse_phone_number


@jingrow.whitelist()
def is_call_integration_enabled():
	twilio_enabled = jingrow.db.get_single_value("CRM Twilio Settings", "enabled")
	exotel_enabled = jingrow.db.get_single_value("CRM Exotel Settings", "enabled")

	return {
		"twilio_enabled": twilio_enabled,
		"exotel_enabled": exotel_enabled,
		"default_calling_medium": get_user_default_calling_medium(),
	}


def get_user_default_calling_medium():
	if not jingrow.db.exists("CRM Telephony Agent", jingrow.session.user):
		return None

	default_medium = jingrow.db.get_value("CRM Telephony Agent", jingrow.session.user, "default_medium")

	if not default_medium:
		return None

	return default_medium


@jingrow.whitelist()
def set_default_calling_medium(medium):
	if not jingrow.db.exists("CRM Telephony Agent", jingrow.session.user):
		jingrow.get_pg(
			{
				"pagetype": "CRM Telephony Agent",
				"user": jingrow.session.user,
				"default_medium": medium,
			}
		).insert(ignore_permissions=True)
	else:
		jingrow.db.set_value("CRM Telephony Agent", jingrow.session.user, "default_medium", medium)

	return get_user_default_calling_medium()


@jingrow.whitelist()
def add_note_to_call_log(call_sid, note):
	"""Add/Update note to call log based on call sid."""
	_note = None
	if not note.get("name"):
		_note = jingrow.get_pg(
			{
				"pagetype": "FCRM Note",
				"title": note.get("title", "Call Note"),
				"content": note.get("content"),
			}
		).insert(ignore_permissions=True)
	else:
		_note = jingrow.set_value("FCRM Note", note.get("name"), "content", note.get("content"))

	call_log = jingrow.get_cached_pg("CRM Call Log", call_sid)
	call_log.link_with_reference_pg("FCRM Note", _note.name)
	call_log.save(ignore_permissions=True)

	return _note


@jingrow.whitelist()
def add_task_to_call_log(call_sid, task):
	"""Add/Update task to call log based on call sid."""
	_task = None
	if not task.get("name"):
		_task = jingrow.get_pg(
			{
				"pagetype": "CRM Task",
				"title": task.get("title"),
				"description": task.get("description"),
				"assigned_to": task.get("assigned_to"),
				"due_date": task.get("due_date"),
				"status": task.get("status"),
				"priority": task.get("priority"),
			}
		).insert(ignore_permissions=True)
	else:
		_task = jingrow.get_pg("CRM Task", task.get("name"))
		_task.update(
			{
				"title": task.get("title"),
				"description": task.get("description"),
				"assigned_to": task.get("assigned_to"),
				"due_date": task.get("due_date"),
				"status": task.get("status"),
				"priority": task.get("priority"),
			}
		)
		_task.save(ignore_permissions=True)

	call_log = jingrow.get_pg("CRM Call Log", call_sid)
	call_log.link_with_reference_pg("CRM Task", _task.name)
	call_log.save(ignore_permissions=True)

	return _task


@jingrow.whitelist()
def get_contact_by_phone_number(phone_number):
	"""Get contact by phone number."""
	number = parse_phone_number(phone_number)

	if number.get("is_valid"):
		return get_contact(number.get("national_number"), number.get("country"))
	else:
		return get_contact(phone_number, number.get("country"), exact_match=True)


def get_contact(phone_number, country="IN", exact_match=False):
	if not phone_number:
		return {"mobile_no": phone_number}

	cleaned_number = (
		phone_number.strip()
		.replace(" ", "")
		.replace("-", "")
		.replace("(", "")
		.replace(")", "")
		.replace("+", "")
	)

	# Check if the number is associated with a contact
	Contact = jingrow.qb.PageType("Contact")
	normalized_phone = Replace(
		Replace(Replace(Replace(Replace(Contact.mobile_no, " ", ""), "-", ""), "(", ""), ")", ""), "+", ""
	)

	query = (
		jingrow.qb.from_(Contact)
		.select(Contact.name, Contact.full_name, Contact.image, Contact.mobile_no)
		.where(normalized_phone.like(f"%{cleaned_number}%"))
		.orderby("modified", order=Order.desc)
	)
	contacts = query.run(as_dict=True)

	if len(contacts):
		# Check if the contact is associated with a deal
		for contact in contacts:
			if jingrow.db.exists("CRM Contacts", {"contact": contact.name, "is_primary": 1}):
				deal = jingrow.db.get_value(
					"CRM Contacts", {"contact": contact.name, "is_primary": 1}, "parent"
				)
				if are_same_phone_number(contact.mobile_no, phone_number, country, validate=not exact_match):
					contact["deal"] = deal
					return contact
		# Else, return the first contact
		if are_same_phone_number(contacts[0].mobile_no, phone_number, country, validate=not exact_match):
			return contacts[0]

	# Else, Check if the number is associated with a lead
	Lead = jingrow.qb.PageType("CRM Lead")
	normalized_phone = Replace(
		Replace(Replace(Replace(Replace(Lead.mobile_no, " ", ""), "-", ""), "(", ""), ")", ""), "+", ""
	)

	query = (
		jingrow.qb.from_(Lead)
		.select(Lead.name, Lead.lead_name, Lead.image, Lead.mobile_no)
		.where(Lead.converted == 0)
		.where(normalized_phone.like(f"%{cleaned_number}%"))
		.orderby("modified", order=Order.desc)
	)
	leads = query.run(as_dict=True)

	if len(leads):
		for lead in leads:
			if are_same_phone_number(lead.mobile_no, phone_number, country, validate=not exact_match):
				lead["lead"] = lead.name
				lead["full_name"] = lead.lead_name
				return lead

	return {"mobile_no": phone_number}