queries.py

import frappe
from frappe import _
from frappe.desk.reportview import get_filters_cond, get_match_cond
from frappe.utils import unique


@frappe.whitelist()
@frappe.validate_and_sanitize_search_inputs
def customer_query(
    doctype: str,
    txt: str,
    searchfield: str,
    start: int,
    page_len: int,
    filters: dict,
    as_dict: bool = False,
) -> list:
    doctype = "Customer"
    conditions = []
    cust_master_name = frappe.defaults.get_user_default("cust_master_name")

    fields = ["name"]
    if cust_master_name != "Customer Name":
        fields.append("customer_name")

    fields = get_fields(doctype, fields)
    searchfields = frappe.get_meta(doctype).get_search_fields()
    searchfields = " or ".join(field + " like %(txt)s" for field in searchfields)

    return frappe.db.sql(  # nosemgrep
        """select {fields} from `tabCustomer`
		where docstatus < 2
			and ({scond}) and disabled=0
			{fcond} {mcond}
		order by
			(case when locate(%(_txt)s, name) > 0 then locate(%(_txt)s, name) else 99999 end),
			(case when locate(%(_txt)s, customer_name) > 0 then locate(%(_txt)s, customer_name) else 99999 end),
			idx desc,
			name, customer_name
		limit %(page_len)s offset %(start)s""".format(
            **{
                "fields": ", ".join(fields),
                "scond": searchfields,
                "mcond": get_match_cond(doctype),
                "fcond": get_filters_cond(doctype, filters, conditions).replace(
                    "%", "%%"
                ),
            }
        ),
        {
            "txt": "%%%s%%" % txt,
            "_txt": txt.replace("%", ""),
            "start": start,
            "page_len": page_len,
        },
        as_dict=as_dict,
    )


def get_fields(doctype, fields=None):
    if fields is None:
        fields = []
    meta = frappe.get_meta(doctype)
    fields.extend(meta.get_search_fields())

    if meta.title_field and meta.title_field.strip() not in fields:
        fields.insert(1, meta.title_field.strip())

    return unique(fields)


@frappe.whitelist()
def get_ledger_balance(customer: str) -> float:
    if not customer:
        frappe.throw(_("Customer ID is required."))

    # Fetch receivable balance for the customer
    balance = frappe.db.sql(
        """
        SELECT SUM(debit - credit) AS receivable
        FROM `tabGL Entry`
        WHERE party_type = 'Customer'
        AND party = %s
        AND account IN (
            SELECT name FROM `tabAccount` WHERE account_type = 'Receivable'
        )
        AND is_cancelled = 0
    """,
        (customer,),
        as_dict=True,
    )

    # Return the balance
    return balance[0].get("receivable", 0) if balance else 0