Files
erpnext/erpnext/controllers/trends.py
Mihir Kandoi f03c1311cd fix(controllers): source trend report labels from the master (#57724)
* fix(controllers): source trend report labels from the master

item_name, customer_name, territory and supplier_name are stored on each
transaction and editable, so they are not functionally dependent on the grouped
key and historical documents can hold different values for the same item,
customer or supplier. Aggregating them with Max() is a text sort, and MariaDB
folds case while PostgreSQL orders by byte value, so the two engines can label
the same row differently.

Read each from its master instead. Those values ARE dependent on the grouped
key, so they can be grouped without splitting rows and agree on both engines by
construction rather than by an assumption about the data. Supplier needed no new
join -- the Supplier master was already joined as t3 for supplier_group.

A Quotation's party_name is a dynamic link to either a Customer or a Lead, so
neither master can be joined without dropping the other; there the values come
from correlated subqueries over both, keyed only on the grouped party_name.

Row counts and every numeric total are unchanged. What changes is that a
renamed record now shows its current name rather than whichever historical
snapshot happened to sort highest.

* test(selling): assert which label the trends report returns

The existing tests assert the customer stays one row but never which territory
or name comes back, so a divergence between engines passes unnoticed. Asserts
both equal the Customer master's values while an order stores a different
territory.

* fix(controllers): resolve a Quotation's party label through quotation_to

party_name is a dynamic link, so looking it up in Customer and Lead alone was
wrong twice over: a Quotation raised against a Prospect or a CRM Deal got a
blank label, and when a Lead shared its name with a Customer the Customer-first
lookup returned the wrong record's name and territory.

Resolve through the quotation_to discriminator instead, mirroring
Quotation.set_customer_name -- Customer, Lead (company_name falling back to
lead_name), Prospect, and CRM Deal. The CRM Deal branch is emitted only when its
table exists, since it ships with the CRM app.

quotation_to joins the GROUP BY as well: two parties of different types can
share a name, and merging them into one row was never right.

* style(controllers): name the quotation CASE branches

semgrep's string-concat-in-list flags adjacent string literals inside a list,
since that shape is usually a missing comma rather than deliberate. Bind each
branch to a name first so the concatenation is unambiguous.
2026-08-03 06:52:25 +00:00

644 lines
20 KiB
Python

# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
import frappe
from frappe import _
from frappe.utils import DateTimeLikeObject, getdate, today
import erpnext
from erpnext.accounts.utils import get_fiscal_year
def get_columns(filters, trans):
validate_filters(filters)
# get conditions for based_on filter cond
based_on_details = based_wise_columns_query(filters.get("based_on"), trans)
# get conditions for periodic filter cond
period_cols, period_select = period_wise_columns_query(filters, trans)
# get conditions for grouping filter cond
group_by_cols = group_wise_column(filters.get("group_by"))
columns = (
based_on_details["based_on_cols"]
+ period_cols
+ [_("Total(Qty)") + ":Float:120", _("Total(Amt)") + ":Currency/currency:120"]
)
if group_by_cols:
columns = (
based_on_details["based_on_cols"]
+ group_by_cols
+ period_cols
+ [_("Total(Qty)") + ":Float:120", _("Total(Amt)") + ":Currency/currency:120"]
)
conditions = {
"based_on_select": based_on_details["based_on_select"],
"period_wise_select": period_select,
"columns": columns,
"group_by": based_on_details["based_on_group_by"],
"grbc": group_by_cols,
"trans": trans,
"addl_tables": based_on_details["addl_tables"],
"addl_tables_relational_cond": based_on_details.get("addl_tables_relational_cond", ""),
}
conditions["company_currency"] = (
erpnext.get_company_currency(filters.get("company")) if filters.get("company") else None
)
return conditions
def validate_filters(filters):
if not filters.get("fiscal_year"):
filters["fiscal_year"] = get_fiscal_year(today())[0]
if not filters.get("company"):
filters["company"] = frappe.defaults.get_user_default("Company")
for f in ["Fiscal Year", "Based On", "Period", "Company"]:
if not filters.get(f.lower().replace(" ", "_")):
frappe.throw(_("{0} is mandatory").format(_(f)))
if not frappe.db.exists("Fiscal Year", filters.get("fiscal_year")):
frappe.throw(_("Fiscal Year {0} does not exist").format(filters.get("fiscal_year")))
if filters.get("based_on") == filters.get("group_by"):
frappe.throw(_("'Based On' and 'Group By' can not be the same"))
if filters.get("period_based_on") and filters.period_based_on not in ["bill_date", "posting_date"]:
frappe.throw(
msg=_("{0} can be either {1} or {2}.").format(
frappe.bold("Period based On"), frappe.bold("Posting Date"), frappe.bold("Billing Date")
),
title=_("Invalid Filter"),
)
def get_data(filters, conditions):
data = []
inc, cond = "", ""
query_details = conditions["based_on_select"] + conditions["period_wise_select"]
posting_date = "t1.transaction_date"
if conditions.get("trans") in [
"Sales Invoice",
"Purchase Invoice",
"Purchase Receipt",
"Delivery Note",
]:
posting_date = "t1.posting_date"
if filters.period_based_on and conditions.get("trans") in ["Sales Invoice", "Purchase Invoice"]:
posting_date = "t1." + filters.period_based_on
if conditions["based_on_select"] in ["t1.project,", "t2.project,"]:
cond = " and " + conditions["based_on_select"][:-1] + " IS Not NULL"
if not filters.get("include_closed_orders"):
if conditions.get("trans") in ["Sales Order", "Purchase Order"]:
cond += " and t1.status != 'Closed'"
if conditions.get("trans") == "Quotation" and filters.get("group_by") == "Customer":
cond += " and t1.quotation_to = 'Customer'"
year_start_date, year_end_date = frappe.get_cached_value(
"Fiscal Year", filters.get("fiscal_year"), ["year_start_date", "year_end_date"]
)
if filters.get("group_by"):
sel_col = ""
ind = conditions["columns"].index(conditions["grbc"][0])
if filters.get("group_by") == "Item":
sel_col = "t2.item_code"
elif filters.get("group_by") == "Customer":
sel_col = "t1.party_name" if conditions.get("trans") == "Quotation" else "t1.customer"
elif filters.get("group_by") == "Supplier":
sel_col = "t1.supplier"
# first column of the multi-column group_by = the based-on key the detail queries equate against
based_on_key = conditions["group_by"].split(",")[0].strip()
if filters.get("based_on") in ["Customer", "Supplier"]:
inc = 3
elif filters.get("based_on") in ["Item"]:
inc = 2
else:
inc = 1
data1 = frappe.db.sql(
""" select {} from `tab{}` t1, `tab{} Item` t2 {}
where t2.parent = t1.name and t1.company = {} and {} between {} and {} and
t1.docstatus = 1 {} {}
group by {}
""".format(
query_details,
conditions["trans"],
conditions["trans"],
conditions["addl_tables"],
"%s",
posting_date,
"%s",
"%s",
conditions.get("addl_tables_relational_cond"),
cond,
conditions["group_by"],
),
(filters.get("company"), year_start_date, year_end_date),
as_list=1,
)
for d in range(len(data1)):
# to add blanck column
dt = data1[d]
dt.insert(ind, "")
data.append(dt)
# to get distinct value of col specified by group_by in filter
row = frappe.db.sql(
"""select DISTINCT({}) from `tab{}` t1, `tab{} Item` t2 {}
where t2.parent = t1.name and t1.company = {} and {} between {} and {}
and t1.docstatus = 1 and {} = {} {} {}
""".format(
sel_col,
conditions["trans"],
conditions["trans"],
conditions["addl_tables"],
"%s",
posting_date,
"%s",
"%s",
based_on_key,
"%s",
conditions.get("addl_tables_relational_cond"),
cond,
),
(filters.get("company"), year_start_date, year_end_date, data1[d][0]),
as_list=1,
)
for i in range(len(row)):
des = ["" for q in range(len(conditions["columns"]))]
# get data for group_by filter
row1 = frappe.db.sql(
""" select t4.default_currency AS currency , {} , {} from `tab{}` t1, `tab{} Item` t2 {}
where t2.parent = t1.name and t1.company = {} and {} between {} and {}
and t1.docstatus = 1 and {} = {} and {} = {} {} {}
group by t4.default_currency, {}
""".format(
sel_col,
conditions["period_wise_select"],
conditions["trans"],
conditions["trans"],
conditions["addl_tables"],
"%s",
posting_date,
"%s",
"%s",
sel_col,
"%s",
based_on_key,
"%s",
conditions.get("addl_tables_relational_cond"),
cond,
sel_col,
),
(filters.get("company"), year_start_date, year_end_date, row[i][0], data1[d][0]),
as_list=1,
)
if not row1:
continue
des[ind] = row[i][0]
des[ind - 1] = row1[0][0]
for j in range(1, len(conditions["columns"]) - inc):
des[j + inc] = row1[0][j]
data.append(des)
total_row = calculate_total_row(data1, conditions["columns"], conditions.get("company_currency"))
data.append(total_row)
else:
data = frappe.db.sql(
""" select {} from `tab{}` t1, `tab{} Item` t2 {}
where t2.parent = t1.name and t1.company = {} and {} between {} and {} and
t1.docstatus = 1 {} {}
group by {}
""".format(
query_details,
conditions["trans"],
conditions["trans"],
conditions["addl_tables"],
"%s",
posting_date,
"%s",
"%s",
cond,
conditions.get("addl_tables_relational_cond", ""),
conditions["group_by"],
),
(filters.get("company"), year_start_date, year_end_date),
as_list=1,
)
total_row = calculate_total_row(data, conditions["columns"], conditions.get("company_currency"))
data.append(total_row)
return data
def calculate_total_row(data, columns, company_currency=None):
def wrap_in_quotes(label):
return f"'{label}'"
total_values = {}
currency_col_idx = None
for i, col in enumerate(columns):
if "Float" in col or "Currency/currency" in col:
total_values[i] = 0
if "Link/Currency" in col:
currency_col_idx = i
for row in data:
for i in total_values.keys():
total_values[i] += row[i] if row[i] is not None else 0
total_row = [wrap_in_quotes(_("Total"))]
for i in range(1, len(columns)):
total_row.append(total_values.get(i, None))
if currency_col_idx is not None:
total_row[currency_col_idx] = company_currency
return total_row
def get_mon(dt):
return getdate(dt).strftime("%b")
def period_wise_columns_query(filters, trans):
query_details = ""
pwc = []
bet_dates = get_period_date_ranges(filters.get("period"), filters.get("fiscal_year"))
if trans in ["Purchase Receipt", "Delivery Note", "Purchase Invoice", "Sales Invoice"]:
trans_date = "posting_date"
if filters.period_based_on and trans in ["Purchase Invoice", "Sales Invoice"]:
trans_date = filters.period_based_on
else:
trans_date = "transaction_date"
if filters.get("period") != "Yearly":
for dt in bet_dates:
get_period_wise_columns(dt, filters.get("period"), pwc)
query_details = get_period_wise_query(dt, trans_date, query_details)
else:
pwc = [
_(filters.get("fiscal_year")) + " (" + _("Qty") + "):Float:120",
_(filters.get("fiscal_year")) + " (" + _("Amt") + "):Currency/currency:120",
]
query_details = " SUM(t2.stock_qty), SUM(t2.base_net_amount),"
query_details += "SUM(t2.stock_qty), SUM(t2.base_net_amount)"
return pwc, query_details
def get_period_wise_columns(bet_dates, period, pwc):
if period == "Monthly":
pwc += [
_(get_mon(bet_dates[0])) + " (" + _("Qty") + "):Float:120",
_(get_mon(bet_dates[0])) + " (" + _("Amt") + "):Currency/currency:120",
]
else:
pwc += [
_(get_mon(bet_dates[0])) + "-" + _(get_mon(bet_dates[1])) + " (" + _("Qty") + "):Float:120",
_(get_mon(bet_dates[0]))
+ "-"
+ _(get_mon(bet_dates[1]))
+ " ("
+ _("Amt")
+ "):Currency/currency:120",
]
def get_period_wise_query(bet_dates, trans_date, query_details):
query_details += """SUM(CASE WHEN t1.{trans_date} BETWEEN '{sd}' AND '{ed}' THEN t2.stock_qty ELSE NULL END),
SUM(CASE WHEN t1.{trans_date} BETWEEN '{sd}' AND '{ed}' THEN t2.base_net_amount ELSE NULL END),
""".format(
trans_date=trans_date,
sd=bet_dates[0],
ed=bet_dates[1],
)
return query_details
@frappe.whitelist()
def get_period_date_ranges(
period: str, fiscal_year: str | None = None, year_start_date: DateTimeLikeObject | None = None
):
from dateutil.relativedelta import relativedelta
if not year_start_date:
year_start_date, year_end_date = frappe.get_cached_value(
"Fiscal Year", fiscal_year, ["year_start_date", "year_end_date"]
)
increment = {"Monthly": 1, "Quarterly": 3, "Half-Yearly": 6, "Yearly": 12}.get(period)
period_date_ranges = []
for _i in range(1, 13, increment):
period_end_date = getdate(year_start_date) + relativedelta(months=increment, days=-1)
if period_end_date > getdate(year_end_date):
period_end_date = year_end_date
period_date_ranges.append([year_start_date, period_end_date])
year_start_date = period_end_date + relativedelta(days=1)
if period_end_date == year_end_date:
break
return period_date_ranges
def get_period_month_ranges(period, fiscal_year):
from dateutil.relativedelta import relativedelta
period_month_ranges = []
for start_date, end_date in get_period_date_ranges(period, fiscal_year):
months_in_this_period = []
while start_date <= end_date:
months_in_this_period.append(start_date.strftime("%B"))
start_date += relativedelta(months=1)
period_month_ranges.append(months_in_this_period)
return period_month_ranges
def quotation_party_name_expr():
"""Resolve a Quotation's party label from its dynamic link, mirroring set_customer_name()."""
customer_branch = (
"when t1.quotation_to = 'Customer' then "
"(select c.customer_name from `tabCustomer` c where c.name = t1.party_name)"
)
lead_branch = (
"when t1.quotation_to = 'Lead' then "
"(select coalesce(nullif(l.company_name, ''), l.lead_name) from `tabLead` l "
"where l.name = t1.party_name)"
)
prospect_branch = "when t1.quotation_to = 'Prospect' then t1.party_name"
branches = [customer_branch, lead_branch, prospect_branch]
# CRM Deal ships with the CRM app; skip the branch when its table is absent
if frappe.db.table_exists("CRM Deal"):
branches.append(
"when t1.quotation_to = 'CRM Deal' then "
"(select d.organization from `tabCRM Deal` d where d.name = t1.party_name)"
)
return "case " + " ".join(branches) + " end"
def quotation_territory_expr():
"""Only Customer and Lead carry a territory; other party types have none."""
return (
"case "
"when t1.quotation_to = 'Customer' then "
"(select c.territory from `tabCustomer` c where c.name = t1.party_name) "
"when t1.quotation_to = 'Lead' then "
"(select l.territory from `tabLead` l where l.name = t1.party_name) "
"end"
)
def based_wise_columns_query(based_on, trans):
based_on_details = {}
# based_on_cols, based_on_select, based_on_group_by, addl_tables
if based_on == "Item":
based_on_details["based_on_cols"] = [
{"label": _("Item"), "fieldtype": "Link", "options": "Item", "width": 120, "fieldname": "item"},
{"label": _("Item Name"), "fieldtype": "Data", "width": 120, "fieldname": "item_name"},
]
# item_name is stored per line and editable, so it is not functionally dependent on item_code
# and Max() over it is a sort -- which MariaDB and PostgreSQL resolve differently. Read it
# from the Item master instead: that IS functionally dependent on the grouped item_code, so
# it can be grouped without splitting rows and is identical on both engines by construction.
based_on_details["based_on_select"] = "t2.item_code, item_master.item_name as item_name,"
based_on_details["based_on_group_by"] = "t2.item_code, item_master.item_name"
based_on_details["addl_tables"] = ",`tabItem` item_master"
based_on_details["addl_tables_relational_cond"] = " and t2.item_code = item_master.name"
elif based_on == "Item Group":
based_on_details["based_on_cols"] = [
{
"label": _("Item Group"),
"fieldtype": "Link",
"options": "Item Group",
"width": 120,
"fieldname": "item_group",
}
]
based_on_details["based_on_select"] = "t2.item_group,"
based_on_details["based_on_group_by"] = "t2.item_group"
based_on_details["addl_tables"] = ""
elif based_on == "Customer":
if trans == "Quotation":
based_on_details["based_on_cols"] = [
{
"label": _("Party"),
"fieldtype": "Link",
"options": "Customer",
"width": 120,
"fieldname": "party",
},
{"label": _("Party Name"), "fieldtype": "Data", "width": 120, "fieldname": "party_name"},
{
"label": _("Territory"),
"fieldtype": "Link",
"options": "Territory",
"width": 120,
"fieldname": "territory",
},
]
# a Quotation's party_name is a dynamic link, so no single master can be joined. Resolve
# it through the quotation_to discriminator, mirroring Quotation.set_customer_name, and
# group by it too: two parties of different types can share a name, and merging them
# under one row was never right. Correlated only on grouped columns, so the query stays
# valid under GROUP BY and free of any text sort.
based_on_details["based_on_select"] = (
f"t1.party_name, {quotation_party_name_expr()} as customer_name, "
f"{quotation_territory_expr()} as territory,"
)
based_on_details["based_on_group_by"] = "t1.party_name, t1.quotation_to"
based_on_details["addl_tables"] = ""
else:
based_on_details["based_on_cols"] = [
{
"label": _("Customer"),
"fieldtype": "Link",
"options": "Customer",
"width": 120,
"fieldname": "customer",
},
{
"label": _("Customer Name"),
"fieldtype": "Data",
"width": 120,
"fieldname": "customer_name",
},
{
"label": _("Territory"),
"fieldtype": "Link",
"options": "Territory",
"width": 120,
"fieldname": "territory",
},
]
# customer_name and territory are stored per transaction and editable, so they are not
# functionally dependent on the customer and Max() over them is a text sort, which the
# engines resolve differently. The Customer master's values ARE dependent on the grouped
# key, so they can be grouped without splitting rows and agree on both engines.
based_on_details["based_on_select"] = (
"t1.customer, customer_master.customer_name as customer_name, "
"customer_master.territory as territory,"
)
based_on_details[
"based_on_group_by"
] = "t1.customer, customer_master.customer_name, customer_master.territory"
based_on_details["addl_tables"] = ",`tabCustomer` customer_master"
based_on_details["addl_tables_relational_cond"] = " and t1.customer = customer_master.name"
elif based_on == "Customer Group":
based_on_details["based_on_cols"] = [
{
"label": _("Customer Group"),
"fieldtype": "Link",
"options": "Customer Group",
"fieldname": "customer_group",
}
]
based_on_details["based_on_select"] = "t1.customer_group,"
based_on_details["based_on_group_by"] = "t1.customer_group"
based_on_details["addl_tables"] = ""
elif based_on == "Supplier":
based_on_details["based_on_cols"] = [
{
"label": _("Supplier"),
"fieldtype": "Link",
"options": "Supplier",
"width": 120,
"fieldname": "supplier",
},
{"label": _("Supplier Name"), "fieldtype": "Data", "width": 120, "fieldname": "supplier_name"},
{
"label": _("Supplier Group"),
"fieldtype": "Link",
"options": "Supplier Group",
"width": 140,
"fieldname": "supplier_group",
},
]
# supplier_name is stored per transaction and editable, so Max() over it is a text sort that
# the engines resolve differently. The Supplier master is already joined here as t3 and its
# columns are functionally dependent on the grouped supplier, so both can simply be grouped:
# no row split, and identical on both engines by construction.
based_on_details["based_on_select"] = "t1.supplier, t3.supplier_name, t3.supplier_group,"
based_on_details["based_on_group_by"] = "t1.supplier, t3.supplier_name, t3.supplier_group"
based_on_details["addl_tables"] = ",`tabSupplier` t3"
based_on_details["addl_tables_relational_cond"] = " and t1.supplier = t3.name"
elif based_on == "Supplier Group":
based_on_details["based_on_cols"] = [
{
"label": _("Supplier Group"),
"fieldtype": "Link",
"options": "Supplier Group",
"width": 140,
"fieldname": "supplier_group",
}
]
based_on_details["based_on_select"] = "t3.supplier_group,"
based_on_details["based_on_group_by"] = "t3.supplier_group"
based_on_details["addl_tables"] = ",`tabSupplier` t3"
based_on_details["addl_tables_relational_cond"] = " and t1.supplier = t3.name"
elif based_on == "Territory":
based_on_details["based_on_cols"] = [
{
"label": _("Territory"),
"fieldtype": "Link",
"options": "Territory",
"width": 120,
"fieldname": "territory",
}
]
based_on_details["based_on_select"] = "t1.territory,"
based_on_details["based_on_group_by"] = "t1.territory"
based_on_details["addl_tables"] = ""
elif based_on == "Project":
if trans in ["Sales Invoice", "Delivery Note", "Sales Order"]:
based_on_details["based_on_cols"] = [
{
"label": _("Project"),
"fieldtype": "Link",
"options": "Project",
"width": 120,
"fieldname": "project",
}
]
based_on_details["based_on_select"] = "t1.project,"
based_on_details["based_on_group_by"] = "t1.project"
based_on_details["addl_tables"] = ""
elif trans in ["Purchase Order", "Purchase Invoice", "Purchase Receipt"]:
based_on_details["based_on_cols"] = [
{
"label": _("Project"),
"fieldtype": "Link",
"options": "Project",
"width": 120,
"fieldname": "project",
}
]
based_on_details["based_on_select"] = "t2.project,"
based_on_details["based_on_group_by"] = "t2.project"
based_on_details["addl_tables"] = ""
else:
frappe.throw(_("Project-wise data is not available for Quotation"))
based_on_details["based_on_select"] += "t4.default_currency as currency,"
based_on_details["based_on_group_by"] += ", t4.default_currency"
based_on_details["based_on_cols"].append(
{
"label": _("Currency"),
"fieldtype": "Link",
"options": "Currency",
"width": 120,
"fieldname": "currency",
}
)
based_on_details["addl_tables"] += ", `tabCompany` t4"
based_on_details["addl_tables_relational_cond"] = (
based_on_details.get("addl_tables_relational_cond", "") + " and t1.company = t4.name"
)
return based_on_details
def group_wise_column(group_by):
if group_by:
return [
{
"label": _(group_by),
"fieldtype": "Link",
"options": group_by,
"width": 120,
"fieldname": frappe.scrub(group_by),
}
]
else:
return []