feat(selling): show the latest quotation revision as latest (#59485)

erpnext/accounts/doctype/purchase_invoice/purchase_invoice.py

@@ -40,7 +40,6 @@ from erpnext.accounts.utils import (
 )
 from erpnext.assets.doctype.asset.asset import is_cwip_accounting_enabled
 from erpnext.controllers.buying_controller import BuyingController
-from erpnext.stock.doctype.purchase_receipt.services.billing_status import is_billed_by_qty
 
 
 class WarehouseMissingError(frappe.ValidationError):
@@ -297,12 +296,7 @@ class PurchaseInvoice(BuyingController):
 
 		from erpnext.accounts.services.billing_validation import BillingValidationService
 
-		billing_validation = BillingValidationService(self)
-		if is_billed_by_qty():
-			billing_validation.validate_multiple_billing("Purchase Receipt", "pr_detail", "qty")
-			billing_validation.validate_multiple_billing("Purchase Order", "po_detail", "qty")
-		else:
-			billing_validation.validate_multiple_billing("Purchase Receipt", "pr_detail", "amount")
+		BillingValidationService(self).validate_multiple_billing("Purchase Receipt", "pr_detail", "amount")
 		self.set_status()
 		self.validate_purchase_receipt_if_update_stock()
 		self.validate_exchange_rate_with_purchase_receipt()
@@ -595,9 +589,6 @@ class PurchaseInvoice(BuyingController):
 					frappe.throw(_("Purchase Receipt {0} is not submitted").format(d.purchase_receipt))
 
 	def update_status_updater_args(self):
-		if is_billed_by_qty():
-			self.set_purchase_order_billing_by_qty()
-
 		if cint(self.update_stock):
 			self.status_updater.append(
 				{
@@ -648,13 +639,6 @@ class PurchaseInvoice(BuyingController):
 					}
 				)
 
-	def set_purchase_order_billing_by_qty(self):
-		"""The status updater keeps billed_amt current; billing % and over-billing follow invoiced qty instead."""
-		for args in self.status_updater:
-			if args.get("overflow_type") == "billing":
-				args.pop("percent_join_field", None)
-				args["validate_overflow"] = False
-
 	def validate_purchase_receipt_if_update_stock(self):
 		if self.update_stock:
 			for item in self.get("items"):
@@ -688,7 +672,6 @@ class PurchaseInvoice(BuyingController):
 
 		self.update_status_updater_args()
 		self.update_prevdoc_status()
-		BillingStatusService(self).update_billing_status_in_po()
 
 		frappe.get_cached_doc("Authorization Control").validate_approving_authority(
 			self.doctype, self.company, self.base_grand_total
@@ -803,7 +786,6 @@ class PurchaseInvoice(BuyingController):
 
 		self.update_status_updater_args()
 		self.update_prevdoc_status()
-		BillingStatusService(self).update_billing_status_in_po()
 
 		if not self.is_return:
 			self.update_billing_status_for_zero_amount_refdoc("Purchase Receipt")

erpnext/accounts/doctype/purchase_invoice/services/billing_status.py

@@ -1,7 +1,7 @@
 # Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
 # License: GNU General Public License v3. See license.txt
 
-"""Purchase Receipt and Purchase Order billing sync and provisional-entry cancellation for Purchase Invoice."""
+"""Purchase Receipt billing sync and provisional-entry cancellation for Purchase Invoice."""
 
 import frappe
 from frappe import qb
@@ -9,9 +9,6 @@ from frappe.query_builder.functions import Sum
 from frappe.utils import flt
 
 from erpnext.stock.doctype.purchase_receipt.services.billing_status import (
-	get_invoiced_qty_and_amount,
-	get_purchase_receipts_against_po_details,
-	is_billed_by_qty,
 	update_billed_amount_based_on_po,
 	update_billing_percentage,
 )
@@ -47,52 +44,16 @@ class BillingStatusService:
 		if po_details:
 			updated_pr += update_billed_amount_based_on_po(po_details, update_modified)
 
-		if not is_billed_by_qty():
-			for pr in set(updated_pr):
-				pr_doc = frappe.get_lazy_doc("Purchase Receipt", pr)
-				update_billing_percentage(pr_doc, update_modified=update_modified)
-			return
-
-		self.update_billing_status_in_receipts_on_po_lines(set(updated_pr), update_modified)
-
-	def update_billing_status_in_receipts_on_po_lines(self, updated_pr: set, update_modified: bool) -> None:
-		"""Order invoices are spread over every receipt on the line, so all of them are refreshed from one split."""
-		pr_docs = [
-			frappe.get_lazy_doc("Purchase Receipt", pr) for pr in updated_pr | self.get_receipts_on_po_lines()
-		]
-
-		pr_items = []
-		for pr_doc in pr_docs:
-			pr_items.extend(pr_doc.items)
-
-		bill_for_rejected = frappe.db.get_single_value(
-			"Buying Settings", "bill_for_rejected_quantity_in_purchase_invoice"
+		adjust_incoming_rate = frappe.db.get_single_value(
+			"Buying Settings", "set_landed_cost_based_on_purchase_invoice_rate"
 		)
-		invoiced = get_invoiced_qty_and_amount(pr_items, bill_for_rejected)
 
-		for pr_doc in pr_docs:
+		for pr in set(updated_pr):
+			pr_doc = frappe.get_lazy_doc("Purchase Receipt", pr)
 			update_billing_percentage(
-				pr_doc,
-				update_modified=update_modified,
-				adjust_incoming_rate=True,
-				invoiced=invoiced,
+				pr_doc, update_modified=update_modified, adjust_incoming_rate=adjust_incoming_rate
 			)
 
-	def get_receipts_on_po_lines(self) -> set:
-		po_details = list({d.po_detail for d in self.doc.get("items") if d.po_detail})
-		if not po_details:
-			return set()
-
-		return {pr_item.parent for pr_item in get_purchase_receipts_against_po_details(po_details)}
-
-	def update_billing_status_in_po(self) -> None:
-		doc = self.doc
-		if not is_billed_by_qty() or (doc.is_return and not doc.update_billed_amount_in_purchase_order):
-			return
-
-		for purchase_order in {item.purchase_order for item in doc.items if item.purchase_order}:
-			frappe.get_doc("Purchase Order", purchase_order).update_billing_percentage()
-
 	def get_pr_details_billed_amt(self) -> dict:
 		# Get billed amount based on purchase receipt item reference (pr_detail) in purchase invoice
 

erpnext/accounts/doctype/purchase_invoice/test_purchase_invoice.py

@@ -3787,30 +3787,6 @@ class TestPurchaseInvoice(ERPNextTestSuite, StockTestMixin):
 		# Test 4 - Since this PI is overbilled by 130% and only 120% is allowed, it will fail
 		self.assertRaises(frappe.ValidationError, pi.submit)
 
-	@ERPNextTestSuite.change_settings("Accounts Settings", {"over_billing_allowance": 0})
-	@ERPNextTestSuite.change_settings(
-		"Buying Settings",
-		{
-			"maintain_same_rate": 0,
-			"set_landed_cost_based_on_purchase_invoice_rate": 1,
-			"bill_for_rejected_quantity_in_purchase_invoice": 0,
-		},
-	)
-	def test_receipt_over_billing_by_qty_when_landed_cost_follows_invoice_rate(self):
-		pr = make_purchase_receipt(qty=100, rate=50)
-		for qty in (25, 75):
-			pi = create_purchase_invoice_from_receipt(pr.name)
-			pi.items[0].qty = qty
-			pi.items[0].rate = 200
-			pi.submit()
-
-		pr.reload()
-		self.assertEqual(pr.status, "Completed")
-
-		extra_invoice = frappe.copy_doc(pi)
-		extra_invoice.items[0].qty = 100
-		self.assertRaisesRegex(frappe.ValidationError, "Cannot overbill", extra_invoice.submit)
-
 	@ERPNextTestSuite.change_settings("Accounts Settings", {"over_billing_allowance": 0})
 	def test_non_stock_item_over_billing_against_po_is_blocked(self):
 		service_item = create_item(

erpnext/accounts/services/billing_validation.py

@@ -49,7 +49,7 @@ class BillingValidationService:
 					overbilled_items.append(row)
 
 		if overbilled_items:
-			self.throw_overbill_exception(overbilled_items, precision, based_on)
+			self.throw_overbill_exception(overbilled_items, precision)
 
 		if is_overbilling_allowed and total_overbilled_amt > 0.1:
 			frappe.msgprint(
@@ -92,9 +92,7 @@ class BillingValidationService:
 
 			ref_wise_billed_amount.setdefault(
 				key,
-				frappe._dict(
-					item_code=item.item_code, uom=item.get("uom"), billed_amt=0.0, ref_amt=ref_amt, rows=[]
-				),
+				frappe._dict(item_code=item.item_code, billed_amt=0.0, ref_amt=ref_amt, rows=[]),
 			)
 			ref_wise_billed_amount[key]["rows"].append(item.idx)
 			ref_wise_billed_amount[key]["ref_amt"] = ref_amt
@@ -133,7 +131,7 @@ class BillingValidationService:
 			).run()
 		)
 
-	def throw_overbill_exception(self, overbilled_items: list, precision: int, based_on: str) -> None:
+	def throw_overbill_exception(self, overbilled_items: list, precision: int) -> None:
 		message = (
 			_("<p>Cannot overbill for the following Items:</p>")
 			+ "<ul>"
@@ -141,7 +139,9 @@ class BillingValidationService:
 				_("<li>Item {0} in row(s) {1} billed more than {2}</li>").format(
 					frappe.bold(item.item_code),
 					", ".join(str(x) for x in item.rows),
-					frappe.bold(self.get_formatted_limit(item, precision, based_on)),
+					frappe.bold(
+						fmt_money(item.max_allowed_amt, precision=precision, currency=self.doc.currency)
+					),
 				)
 				for item in overbilled_items
 			)
@@ -149,9 +149,3 @@ class BillingValidationService:
 		)
 		message += _("<p>To allow over-billing, please set allowance in Accounts Settings.</p>")
 		frappe.throw(message)
-
-	def get_formatted_limit(self, item: frappe._dict, precision: int, based_on: str) -> str:
-		if based_on == "qty":
-			return f"{flt(item.max_allowed_amt, precision)} {item.uom}"
-
-		return fmt_money(item.max_allowed_amt, precision=precision, currency=self.doc.currency)

erpnext/buying/doctype/purchase_order/services/status.py

@@ -10,10 +10,6 @@ from frappe.utils import cstr, flt
 
 from erpnext.buying.doctype.purchase_order.services.subcontracting import SubcontractingService
 from erpnext.controllers.item_close import validate_parent_reopen
-from erpnext.stock.doctype.purchase_receipt.services.billing_status import (
-	get_invoiced_qty_against_po_items,
-	get_qty_based_percent_billed,
-)
 
 
 class StatusService:
@@ -65,9 +61,3 @@ class StatusService:
 
 		per_received = flt(received_qty / total_qty) * 100 if total_qty else 0
 		doc.db_set("per_received", per_received, update_modified=False)
-
-	def get_percent_billed_by_qty(self) -> float:
-		items = [item for item in self.doc.items if not item.closed] or self.doc.items
-		billable_qty = {item.name: flt(item.qty) for item in items}
-		invoiced_qty = get_invoiced_qty_against_po_items(list(billable_qty))
-		return get_qty_based_percent_billed(items, billable_qty, invoiced_qty)

erpnext/buying/doctype/purchase_order/test_purchase_order.py

@@ -1521,49 +1521,6 @@ class TestPurchaseOrder(ERPNextTestSuite):
 		po.reload()
 		self.assertEqual(po.per_billed, 100)
 
-	@ERPNextTestSuite.change_settings(
-		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
-	)
-	def test_per_billed_by_qty_when_landed_cost_follows_invoice_rate(self):
-		po = create_purchase_order(qty=100, rate=50)
-		pi = make_pi_from_po(po.name)
-		pi.items[0].qty = 25
-		pi.items[0].rate = 200
-		pi.submit()
-
-		po.reload()
-		self.assertEqual(po.per_billed, 25)
-		self.assertEqual(po.status, "To Receive and Bill")
-
-		pi.reload()
-		pi.cancel()
-		po.reload()
-		self.assertEqual(po.per_billed, 0)
-
-	@ERPNextTestSuite.change_settings("Accounts Settings", {"over_billing_allowance": 0})
-	@ERPNextTestSuite.change_settings(
-		"Buying Settings",
-		{
-			"maintain_same_rate": 0,
-			"set_landed_cost_based_on_purchase_invoice_rate": 1,
-			"bill_for_rejected_quantity_in_purchase_invoice": 0,
-		},
-	)
-	def test_over_billing_by_qty_when_landed_cost_follows_invoice_rate(self):
-		po = create_purchase_order(qty=100, rate=50)
-		for qty in (25, 75):
-			pi = make_pi_from_po(po.name)
-			pi.items[0].qty = qty
-			pi.items[0].rate = 200
-			pi.submit()
-
-		po.reload()
-		self.assertEqual(po.per_billed, 100)
-
-		extra_invoice = frappe.copy_doc(pi)
-		extra_invoice.items[0].qty = 100
-		self.assertRaisesRegex(frappe.ValidationError, "Cannot overbill", extra_invoice.submit)
-
 	@ERPNextTestSuite.change_settings("Buying Settings", {"allow_zero_qty_in_purchase_order": 1})
 	def test_receive_zero_qty_purchase_order(self):
 		"""

erpnext/controllers/status_updater.py

@@ -394,7 +394,7 @@ class StatusUpdater(Document):
 						)
 					)
 
-			if items_to_validate and args.get("validate_overflow") is not False:
+			if items_to_validate:
 				pp_sub_assembly_items = [
 					item.production_plan_sub_assembly_item
 					for item in items_to_validate

erpnext/controllers/stock_controller.py

@@ -33,7 +33,6 @@ from erpnext.setup.doctype.brand.brand import get_brand_defaults
 from erpnext.setup.doctype.item_group.item_group import get_item_group_defaults
 from erpnext.stock import get_warehouse_account, get_warehouse_account_map
 from erpnext.stock.doctype.item.item import get_item_defaults
-from erpnext.stock.doctype.purchase_receipt.services.billing_status import is_billed_by_qty
 from erpnext.stock.services.internal_transfer import StockInternalTransferService
 from erpnext.stock.stock_ledger import get_items_to_be_repost
 
@@ -341,10 +340,6 @@ class StockController(AccountsController):
 		if self.doctype == "Delivery Note":
 			# Bill by amount, falling back to qty when the invoiced amount is short (e.g. rate drop).
 			args["billing_percentage"] = self.get_delivery_note_billing_percentage()
-		elif self.doctype == "Purchase Order" and is_billed_by_qty():
-			from erpnext.buying.doctype.purchase_order.services.status import StatusService
-
-			args["billing_percentage"] = StatusService(self).get_percent_billed_by_qty()
 
 		self._update_percent_field(args, update_modified)
 

erpnext/selling/doctype/quotation/quotation.json

@@ -25,6 +25,7 @@
   "has_unit_price_items",
   "amended_from",
   "revision_of",
+  "is_latest_revision",
   "currency_and_price_list",
   "currency",
   "conversion_rate",
@@ -227,6 +228,16 @@
    "read_only": 1,
    "search_index": 1
   },
+  {
+   "default": "0",
+   "fieldname": "is_latest_revision",
+   "fieldtype": "Check",
+   "hidden": 1,
+   "label": "Is Latest Revision",
+   "no_copy": 1,
+   "print_hide": 1,
+   "read_only": 1
+  },
   {
    "fieldname": "company",
    "fieldtype": "Link",
@@ -1169,7 +1180,7 @@
  "idx": 82,
  "is_submittable": 1,
  "links": [],
- "modified": "2026-09-24 14:30:00.000000",
+ "modified": "2026-09-26 12:00:00.000000",
  "modified_by": "Administrator",
  "module": "Selling",
  "name": "Quotation",

erpnext/selling/doctype/quotation/quotation.py

@@ -79,6 +79,7 @@ class Quotation(SellingController):
 		in_words: DF.Data | None
 		incoterm: DF.Link | None
 		is_active: DF.Check
+		is_latest_revision: DF.Check
 		item_wise_tax_details: DF.Table[ItemWiseTaxDetail]
 		items: DF.Table[QuotationItem]
 		language: DF.Link | None
@@ -384,6 +385,7 @@ class Quotation(SellingController):
 		self.update_opportunity("Quotation")
 		self.update_lead()
 		self.deactivate_other_versions()
+		self.update_latest_revision()
 
 	def deactivate_other_versions(self):
 		if not (self.revision_of and self.is_active):
@@ -403,9 +405,31 @@ class Quotation(SellingController):
 		return bool(self.get_other_versions(VERSIONS_TO_SET_AS_LOST))
 
 	def update_other_versions(self, filters: dict, values: dict):
-		names = [version.name for version in self.get_other_versions(filters)]
-		frappe.db.bulk_update("Quotation", {name: values for name in names})
-		for name in names:
+		self.update_versions({version.name: values for version in self.get_other_versions(filters)})
+
+	def update_latest_revision(self):
+		versions = self.get_other_versions({})
+		if not (versions or self.is_latest_revision):
+			return
+
+		if self.docstatus == 1:
+			versions.append(self)
+
+		latest = max(versions, key=get_version_order).name if len(versions) > 1 else None
+		self.update_versions(
+			{
+				version.name: {"is_latest_revision": int(version.name == latest)}
+				for version in versions
+				if version.name != self.name
+			},
+			update_modified=False,
+		)
+		self.db_set("is_latest_revision", int(self.name == latest), update_modified=False)
+
+	@staticmethod
+	def update_versions(updates: dict[str, dict], update_modified: bool = True):
+		frappe.db.bulk_update("Quotation", updates, update_modified=update_modified)
+		for name in updates:
 			frappe.clear_document_cache("Quotation", name)
 
 	@property
@@ -413,12 +437,8 @@ class Quotation(SellingController):
 		return not self.get_newer_versions()
 
 	def get_newer_versions(self) -> list[frappe._dict]:
-		own_order = (getdate(self.transaction_date), get_datetime(self.creation))
-		return [
-			version
-			for version in self.get_other_versions({})
-			if (version.transaction_date, version.creation) > own_order
-		]
+		own_order = get_version_order(self)
+		return [version for version in self.get_other_versions({}) if get_version_order(version) > own_order]
 
 	def validate_can_be_revised(self):
 		if self.status in ("Lost", "Ordered"):
@@ -443,6 +463,7 @@ class Quotation(SellingController):
 		self.set_status(update=True)
 		self.update_opportunity("Open")
 		self.update_lead()
+		self.update_latest_revision()
 
 	def carry_forward_communication(self):
 		from erpnext.crm.utils import copy_comments, link_communications
@@ -485,6 +506,10 @@ class Quotation(SellingController):
 		return rows_with_alternatives
 
 
+def get_version_order(version) -> tuple:
+	return (getdate(version.transaction_date), get_datetime(version.creation))
+
+
 def get_list_context(context=None):
 	from erpnext.controllers.website_list_for_contact import get_list_context
 

erpnext/selling/doctype/quotation/quotation_list.js

@@ -7,6 +7,7 @@ frappe.listview_settings["Quotation"] = {
 		"currency",
 		"valid_till",
 		"is_active",
+		"is_latest_revision",
 	],
 
 	onload: function (listview) {
@@ -38,6 +39,8 @@ frappe.listview_settings["Quotation"] = {
 			return [__("Lost"), "gray", "status,=,Lost"];
 		} else if (doc.docstatus === 1 && !doc.is_active) {
 			return [__("Inactive"), "red", "is_active,=,0"];
+		} else if (doc.status === "Open" && doc.is_latest_revision) {
+			return [__("Latest"), "orange", "is_latest_revision,=,1"];
 		} else if (doc.status === "Open") {
 			return [__("Open"), "orange", "status,=,Open"];
 		} else if (doc.status === "Partially Ordered") {

erpnext/selling/doctype/quotation/test_quotation.py

@@ -539,6 +539,29 @@ class TestQuotation(ERPNextTestSuite):
 		self.assertEqual(revision.items[0].rate, 250)
 		self.assertEqual(revision.items[0].prevdoc_docname, opportunity.name)
 
+	def test_latest_revision_is_flagged(self):
+		quotation = make_quotation()
+		self.assertEqual(quotation.is_latest_revision, 0)
+
+		first_revision = make_revision(quotation.name)
+		first_revision.insert()
+		first_revision.submit()
+		second_revision = make_revision(first_revision.name)
+		second_revision.insert()
+		second_revision.submit()
+
+		self.assertEqual(self.get_latest_revision_flags(quotation), [0, 0, 1])
+
+		second_revision.cancel()
+
+		self.assertEqual(self.get_latest_revision_flags(quotation), [0, 1, 0])
+
+	def get_latest_revision_flags(self, quotation):
+		return [
+			frappe.db.get_value("Quotation", name, "is_latest_revision")
+			for name in (quotation.name, f"{quotation.name}-R1", f"{quotation.name}-R2")
+		]
+
 	def test_submitting_a_revision_deactivates_other_versions(self):
 		quotation = make_quotation()
 		first_revision = make_revision(quotation.name)

erpnext/stock/doctype/purchase_receipt/services/billing_status.py

@@ -115,7 +115,6 @@ def get_purchase_receipts_against_po_details(po_details: list) -> list[dict]:
 		.select(
 			purchase_receipt_item.name,
 			purchase_receipt_item.qty,
-			purchase_receipt_item.rejected_qty,
 			purchase_receipt_item.parent,
 			purchase_receipt_item.amount,
 			purchase_receipt_item.billed_amt,
@@ -181,46 +180,30 @@ def get_billed_amount_against_po(po_items: list) -> dict:
 
 
 def update_billing_percentage(
-	pr_doc, update_modified: bool = True, adjust_incoming_rate: bool = False, invoiced: dict | None = None
+	pr_doc, update_modified: bool = True, adjust_incoming_rate: bool = False
 ) -> None:
-	"""`invoiced` from get_invoiced_qty_and_amount lets receipts on one order line share its invoice split."""
+	# Update Billing % based on pending accepted qty
 	buying_settings = frappe.get_single("Buying Settings")
-	bill_for_rejected = buying_settings.bill_for_rejected_quantity_in_purchase_invoice
-	items = [item for item in pr_doc.items if not item.closed] or pr_doc.items
-
-	if buying_settings.set_landed_cost_based_on_purchase_invoice_rate:
-		if invoiced is None:
-			invoiced = get_invoiced_qty_and_amount(pr_doc.items, bill_for_rejected)
-		percent_billed = get_percent_billed_by_qty(pr_doc, items, bill_for_rejected, invoiced)
-	else:
-		percent_billed = get_percent_billed_by_amount(pr_doc, items, bill_for_rejected)
-
-	pr_doc.db_set("per_billed", percent_billed)
-
-	if update_modified:
-		pr_doc.set_status(update=True)
-		pr_doc.notify_update()
-
-	if adjust_incoming_rate and set_amount_difference_with_purchase_invoice(items, invoiced):
-		adjust_incoming_rate_for_pr(pr_doc)
-
-
-def get_percent_billed_by_amount(pr_doc, items: list, bill_for_rejected: bool) -> float:
 	over_billing_allowance, role_allowed_to_over_bill = frappe.get_single_value(
 		"Accounts Settings", ["over_billing_allowance", "role_allowed_to_over_bill"]
 	)
 
-	total_amount, total_billed_amount = 0, 0
-	item_wise_returned_qty = get_item_wise_returned_qty([item.name for item in pr_doc.items])
+	total_amount, total_billed_amount, pi_landed_cost_amount = 0, 0, 0
+	item_wise_returned_qty = get_item_wise_returned_qty(pr_doc)
+	billed_qty_amt = frappe._dict()
 
-	for item in items:
+	if adjust_incoming_rate:
+		billed_qty_amt = get_billed_qty_amount_against_purchase_receipt(pr_doc)
+		billed_qty_amt_based_on_po = get_billed_qty_amount_against_purchase_order(pr_doc)
+
+	for item in [item for item in pr_doc.items if not item.closed] or pr_doc.items:
 		returned_qty = flt(item_wise_returned_qty.get(item.name))
 		returned_amount = flt(returned_qty) * flt(item.rate)
 		pending_amount = flt(item.amount) - returned_amount
 
 		# When rejected qty is billable, its value is part of the billable base too
 		rejected_amount = 0.0
-		if bill_for_rejected:
+		if buying_settings.bill_for_rejected_quantity_in_purchase_invoice:
 			rejected_amount = flt(item.rejected_qty * item.rate, item.precision("amount"))
 			pending_amount = flt(item.amount) + rejected_amount
 
@@ -236,7 +219,54 @@ def get_percent_billed_by_amount(pr_doc, items: list, bill_for_rejected: bool) -
 
 		amount = flt(item.amount) + rejected_amount
 
-		if amount and item.billed_amt > amount:
+		if adjust_incoming_rate:
+			adjusted_amt = 0.0
+
+			if (
+				item.billed_amt is not None
+				and item.amount is not None
+				and (
+					billed_qty_amt.get(item.name) or billed_qty_amt_based_on_po.get(item.purchase_order_item)
+				)
+			):
+				qty = None
+				if billed_qty_amt.get(item.name):
+					qty = billed_qty_amt.get(item.name).get("qty")
+
+				if not qty and billed_qty_amt_based_on_po.get(item.purchase_order_item):
+					if item.qty < billed_qty_amt_based_on_po.get(item.purchase_order_item)["qty"]:
+						qty = item.qty
+					else:
+						qty = billed_qty_amt_based_on_po.get(item.purchase_order_item)["qty"]
+
+					billed_qty_amt_based_on_po[item.purchase_order_item]["qty"] -= qty
+
+				billed_amt = item.billed_amt
+				if billed_qty_amt.get(item.name):
+					billed_amt = flt(billed_qty_amt.get(item.name).get("amount"))
+				elif billed_qty_amt_based_on_po.get(item.purchase_order_item):
+					total_billed_qty = (
+						billed_qty_amt_based_on_po.get(item.purchase_order_item).get("qty") + qty
+					)
+
+					if total_billed_qty:
+						billed_amt = flt(
+							flt(billed_qty_amt_based_on_po.get(item.purchase_order_item).get("amount"))
+							* (qty / total_billed_qty)
+						)
+					else:
+						billed_amt = 0.0
+
+					# Reduce billed amount based on PO for next iterations
+					billed_qty_amt_based_on_po[item.purchase_order_item]["amount"] -= billed_amt
+
+				if qty:
+					adjusted_amt = flt(billed_amt / qty) * item.qty - flt(item.base_net_amount)
+
+			adjusted_amt = flt(adjusted_amt, item.precision("amount"))
+			pi_landed_cost_amount += adjusted_amt
+			item.db_set("amount_difference_with_purchase_invoice", adjusted_amt, update_modified=False)
+		elif amount and item.billed_amt > amount:
 			per_over_billed = (flt(item.billed_amt / amount, 2) * 100) - 100
 			if (
 				per_over_billed > over_billing_allowance
@@ -248,212 +278,22 @@ def get_percent_billed_by_amount(pr_doc, items: list, bill_for_rejected: bool) -
 					)
 				)
 
-	return round(100 * (total_billed_amount / (total_amount or 1)), 6)
+	if pi_landed_cost_amount < 0:
+		total_billed_amount += abs(pi_landed_cost_amount)
+
+	percent_billed = round(100 * (total_billed_amount / (total_amount or 1)), 6)
+	pr_doc.db_set("per_billed", percent_billed)
+
+	if update_modified:
+		pr_doc.set_status(update=True)
+		pr_doc.notify_update()
+
+	if adjust_incoming_rate:
+		adjust_incoming_rate_for_pr(pr_doc)
 
 
-def is_billed_by_qty() -> bool:
-	"""Invoice-rate landed cost leaves qty as the only stable measure of billing."""
-	return bool(
-		frappe.db.get_single_value("Buying Settings", "set_landed_cost_based_on_purchase_invoice_rate")
-	)
-
-
-def get_percent_billed_by_qty(pr_doc, items: list, bill_for_rejected: bool, invoiced: dict) -> float:
-	billable_qty = get_billable_qty_by_row(pr_doc, items, bill_for_rejected)
-	return get_qty_based_percent_billed(items, billable_qty, get_invoiced_qty(pr_doc, invoiced))
-
-
-def get_qty_based_percent_billed(items: list, billable_qty: dict, invoiced_qty: dict) -> float:
-	"""Share of each row's billable qty that is invoiced, weighted by the row's value, or by qty when no row has one."""
-	weigh_by_value = any(flt(item.rate) for item in items)
-
-	total_weight, billed_weight = 0.0, 0.0
-	for item in items:
-		qty = flt(billable_qty.get(item.name))
-		if not qty:
-			continue
-
-		weight = abs(qty * flt(item.rate)) if weigh_by_value else abs(qty)
-		total_weight += weight
-		billed_weight += weight * min(flt(invoiced_qty.get(item.name)) / qty, 1)
-
-	return round(100 * (billed_weight / (total_weight or 1)), 6)
-
-
-def get_billable_qty_by_row(pr_doc, items: list, bill_for_rejected: bool) -> dict:
-	"""Qty left to bill per row; a receipt returned in full is measured against what it received."""
-	returned_qty = get_item_wise_returned_qty([item.name for item in pr_doc.items])
-	billable_qty = {
-		item.name: get_billable_qty(item, returned_qty.get(item.name), bill_for_rejected) for item in items
-	}
-	if any(qty > 0 for qty in billable_qty.values()):
-		return billable_qty
-
-	return {item.name: flt(item.qty) for item in items}
-
-
-def get_billable_qty(item, returned_qty: float | None, bill_for_rejected: bool) -> float:
-	if bill_for_rejected:
-		return flt(item.qty) + flt(item.rejected_qty)
-
-	return flt(item.qty) - flt(returned_qty)
-
-
-def get_invoiced_qty(pr_doc, invoiced: dict) -> dict:
-	invoiced_qty = {name: row.qty for name, row in invoiced.items()}
-
-	for item in pr_doc.items:
-		if item.purchase_invoice_item:
-			invoiced_qty[item.name] = flt(item.qty)
-
-	return invoiced_qty
-
-
-def get_invoiced_qty_and_amount(pr_items: list, bill_for_rejected: bool) -> dict:
-	"""Invoiced qty and base amount per Purchase Receipt Item, direct and through the Purchase Order."""
-	billed = get_billed_qty_amount_against_purchase_receipt([item.name for item in pr_items])
-	invoiced = {
-		pr_detail: frappe._dict(qty=flt(row["qty"]), amount=flt(row["amount"]))
-		for pr_detail, row in billed.items()
-	}
-
-	po_details = list({item.purchase_order_item for item in pr_items if item.purchase_order_item})
-	po_invoice_share = get_po_invoice_share(po_details, bill_for_rejected) if po_details else {}
-
-	for item in pr_items:
-		share = po_invoice_share.get(item.name)
-		if not share or item.purchase_invoice_item:
-			continue
-
-		row = invoiced.setdefault(item.name, frappe._dict(qty=0.0, amount=0.0))
-		row.qty += share.qty
-		row.amount += share.amount
-
-	return invoiced
-
-
-def get_po_invoice_share(po_details: list, bill_for_rejected: bool) -> dict:
-	"""Split invoices made against the Purchase Order over its receipts, oldest invoice to oldest receipt."""
-	po_invoices = get_po_invoices(po_details)
-
-	pr_items = get_purchase_receipts_against_po_details(po_details)
-	pr_item_names = [pr_item.name for pr_item in pr_items]
-	billed_against_pr = get_billed_qty_amount_against_purchase_receipt(pr_item_names)
-	returned_qty = get_item_wise_returned_qty(pr_item_names)
-
-	share = {}
-	for pr_item in pr_items:
-		direct_qty = flt(billed_against_pr.get(pr_item.name, {}).get("qty"))
-		billable_qty = get_billable_qty(pr_item, returned_qty.get(pr_item.name), bill_for_rejected)
-		invoices = po_invoices.get(pr_item.purchase_order_item, [])
-		share[pr_item.name] = take_from_invoices(invoices, billable_qty - direct_qty)
-
-	return share
-
-
-def take_from_invoices(invoices: list, pending_qty: float) -> frappe._dict:
-	"""Take qty from the oldest invoices first, each at its own rate."""
-	taken = frappe._dict(qty=0.0, amount=0.0)
-	for invoice in invoices:
-		qty = min(invoice.qty, pending_qty - taken.qty)
-		if qty <= 0:
-			continue
-
-		amount = invoice.amount * qty / invoice.qty
-		invoice.qty -= qty
-		invoice.amount -= amount
-		taken.qty += qty
-		taken.amount += amount
-
-	return taken
-
-
-def get_po_invoices(po_details: list) -> dict:
-	"""Net qty and base amount of each invoice made against the Purchase Order, oldest first."""
-	purchase_invoice = frappe.qb.DocType("Purchase Invoice")
-	purchase_invoice_item = frappe.qb.DocType("Purchase Invoice Item")
-
-	rows = (
-		frappe.qb.from_(purchase_invoice_item)
-		.inner_join(purchase_invoice)
-		.on(purchase_invoice_item.parent == purchase_invoice.name)
-		.select(
-			purchase_invoice_item.po_detail,
-			purchase_invoice_item.qty,
-			purchase_invoice_item.base_net_amount,
-			purchase_invoice.name,
-			purchase_invoice.is_return,
-			purchase_invoice.return_against,
-		)
-		.where(
-			(purchase_invoice_item.po_detail.isin(po_details))
-			& ((purchase_invoice_item.pr_detail.isnull()) | (purchase_invoice_item.pr_detail == ""))
-			& (purchase_invoice.docstatus == 1)
-			& (purchase_invoice.update_stock == 0)
-		)
-		.orderby(CombineDatetime(purchase_invoice.posting_date, purchase_invoice.posting_time))
-		.orderby(purchase_invoice.name)
-		.orderby(purchase_invoice_item.idx)
-	).run(as_dict=True)
-
-	po_invoices = {}
-	for row in rows:
-		invoice_name = row.return_against if row.is_return else row.name
-		invoices = po_invoices.setdefault(row.po_detail, {})
-		invoice = invoices.setdefault(invoice_name, frappe._dict(qty=0.0, amount=0.0))
-		invoice.qty += flt(row.qty)
-		invoice.amount += flt(row.base_net_amount)
-
-	return {po_detail: list(invoices.values()) for po_detail, invoices in po_invoices.items()}
-
-
-def get_invoiced_qty_against_po_items(po_items: list) -> dict:
-	"""Invoiced qty per Purchase Order Item, leaving out returns that do not touch the order."""
-	purchase_invoice = frappe.qb.DocType("Purchase Invoice")
-	purchase_invoice_item = frappe.qb.DocType("Purchase Invoice Item")
-
-	query = (
-		frappe.qb.from_(purchase_invoice_item)
-		.inner_join(purchase_invoice)
-		.on(purchase_invoice_item.parent == purchase_invoice.name)
-		.select(purchase_invoice_item.po_detail, fn.Sum(purchase_invoice_item.qty))
-		.where(
-			(purchase_invoice_item.po_detail.isin(po_items))
-			& (purchase_invoice.docstatus == 1)
-			& (
-				(purchase_invoice.is_return == 0)
-				| (purchase_invoice.update_billed_amount_in_purchase_order == 1)
-			)
-		)
-		.groupby(purchase_invoice_item.po_detail)
-	)
-
-	return frappe._dict(query.run())
-
-
-def set_amount_difference_with_purchase_invoice(items: list, invoiced: dict) -> bool:
-	"""Store each row's gap to its invoiced value; returns True when any row moved."""
-	has_changed = False
-	for item in items:
-		adjusted_amt = 0.0
-		row = invoiced.get(item.name)
-		if row and row.qty:
-			adjusted_amt = flt(row.amount / row.qty) * flt(item.qty) - flt(item.base_net_amount)
-
-		adjusted_amt = flt(adjusted_amt, item.precision("amount"))
-		if adjusted_amt == flt(item.amount_difference_with_purchase_invoice, item.precision("amount")):
-			continue
-
-		item.db_set("amount_difference_with_purchase_invoice", adjusted_amt, update_modified=False)
-		has_changed = True
-
-	return has_changed
-
-
-def get_billed_qty_amount_against_purchase_receipt(pr_names: list) -> dict:
-	if not pr_names:
-		return frappe._dict()
-
+def get_billed_qty_amount_against_purchase_receipt(pr_doc) -> dict:
+	pr_names = [d.name for d in pr_doc.items]
 	parent_table = frappe.qb.DocType("Purchase Invoice")
 	table = frappe.qb.DocType("Purchase Invoice Item")
 	query = (
@@ -465,11 +305,7 @@ def get_billed_qty_amount_against_purchase_receipt(pr_names: list) -> dict:
 			fn.Sum(table.base_net_amount).as_("amount"),
 			fn.Sum(table.qty).as_("qty"),
 		)
-		.where(
-			(table.pr_detail.isin(pr_names))
-			& (table.docstatus == 1)
-			& ((parent_table.is_return == 0) | (parent_table.update_billed_amount_in_purchase_receipt == 1))
-		)
+		.where((table.pr_detail.isin(pr_names)) & (table.docstatus == 1))
 		.groupby(table.pr_detail)
 	)
 	invoice_data = query.run(as_dict=1)
@@ -489,6 +325,49 @@ def get_billed_qty_amount_against_purchase_receipt(pr_names: list) -> dict:
 	return billed_qty_amt
 
 
+def get_billed_qty_amount_against_purchase_order(pr_doc) -> dict:
+	po_names = list(
+		set(
+			[
+				d.purchase_order_item
+				for d in pr_doc.items
+				if d.purchase_order_item and not d.purchase_invoice_item
+			]
+		)
+	)
+
+	invoice_data_po_based = frappe._dict()
+	if po_names:
+		parent_table = frappe.qb.DocType("Purchase Invoice")
+		table = frappe.qb.DocType("Purchase Invoice Item")
+
+		query = (
+			frappe.qb.from_(parent_table)
+			.inner_join(table)
+			.on(parent_table.name == table.parent)
+			.select(
+				table.po_detail,
+				fn.Sum(table.qty).as_("qty"),
+				fn.Sum(table.base_net_amount).as_("amount"),
+			)
+			.where((table.po_detail.isin(po_names)) & (table.docstatus == 1) & (table.pr_detail.isnull()))
+			.groupby(table.po_detail)
+		)
+
+		invoice_data = query.run(as_dict=1)
+		if not invoice_data:
+			return frappe._dict()
+
+		for row in invoice_data:
+			if row.po_detail not in invoice_data_po_based:
+				invoice_data_po_based[row.po_detail] = {"amount": 0, "qty": 0}
+
+			invoice_data_po_based[row.po_detail]["amount"] += flt(row.amount)
+			invoice_data_po_based[row.po_detail]["qty"] += flt(row.qty)
+
+	return invoice_data_po_based
+
+
 def adjust_incoming_rate_for_pr(doc) -> None:
 	doc.update_valuation_rate(reset_outgoing_rate=False)
 
@@ -501,7 +380,9 @@ def adjust_incoming_rate_for_pr(doc) -> None:
 	doc.repost_future_sle_and_gle(force=True)
 
 
-def get_item_wise_returned_qty(items: list) -> dict:
+def get_item_wise_returned_qty(pr_doc) -> dict:
+	items = [d.name for d in pr_doc.items]
+
 	return frappe._dict(
 		frappe.get_all(
 			"Purchase Receipt",

erpnext/stock/doctype/purchase_receipt/test_purchase_receipt.py

@@ -1117,169 +1117,6 @@ class TestPurchaseReceipt(ERPNextTestSuite):
 		po.reload()
 		po.cancel()
 
-	@ERPNextTestSuite.change_settings(
-		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
-	)
-	def test_per_billed_by_qty_when_landed_cost_follows_invoice_rate(self):
-		pr = make_purchase_receipt(qty=100, rate=50)
-		pi = make_purchase_invoice(pr.name)
-		pi.items[0].qty = 25
-		pi.items[0].rate = 200
-		pi.submit()
-
-		pr.reload()
-		self.assertEqual(pr.per_billed, 25)
-		self.assertEqual(pr.status, "Partly Billed")
-
-	@ERPNextTestSuite.change_settings(
-		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
-	)
-	def test_po_invoice_qty_spread_fifo_when_landed_cost_follows_invoice_rate(self):
-		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
-
-		po = create_purchase_order(qty=100, rate=50)
-		receipts = make_receipts_against_order(po.name, ((60, "08:00"), (40, "10:00")))
-		make_invoice_against_order(po.name, qty=70, rate=40)
-
-		for pr in receipts:
-			pr.reload()
-
-		self.assertEqual(receipts[0].per_billed, 100)
-		self.assertEqual(receipts[1].per_billed, 25)
-
-	@ERPNextTestSuite.change_settings(
-		"Buying Settings",
-		{
-			"maintain_same_rate": 0,
-			"set_landed_cost_based_on_purchase_invoice_rate": 1,
-			"bill_for_rejected_quantity_in_purchase_invoice": 0,
-		},
-	)
-	def test_fully_returned_receipt_skipped_in_po_invoice_split(self):
-		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
-		from erpnext.stock.doctype.purchase_receipt.mapper import make_purchase_return
-
-		po = create_purchase_order(qty=100, rate=50)
-		receipts = make_receipts_against_order(po.name, ((60, "08:00"), (40, "10:00")))
-		make_purchase_return(receipts[0].name).submit()
-		make_invoice_against_order(po.name, qty=40, rate=50)
-
-		for pr in receipts:
-			pr.reload()
-
-		self.assertEqual(receipts[0].per_billed, 0)
-		self.assertEqual(receipts[1].per_billed, 100)
-
-	@ERPNextTestSuite.change_settings(
-		"Buying Settings",
-		{
-			"maintain_same_rate": 0,
-			"set_landed_cost_based_on_purchase_invoice_rate": 1,
-			"bill_for_rejected_quantity_in_purchase_invoice": 0,
-		},
-	)
-	def test_fully_returned_row_left_out_of_qty_billing(self):
-		from erpnext.stock.doctype.purchase_receipt.mapper import make_purchase_return
-
-		pr = make_purchase_receipt(qty=10, rate=50, do_not_save=True)
-		pr.append("items", pr.items[0].as_dict(no_default_fields=True))
-		pr.submit()
-		returned_row, invoiced_row = pr.items
-
-		pr_return = make_purchase_return(pr.name)
-		pr_return.set(
-			"items", [row for row in pr_return.items if row.purchase_receipt_item == returned_row.name]
-		)
-		pr_return.submit()
-
-		pi = make_purchase_invoice(pr.name)
-		pi.set("items", [row for row in pi.items if row.pr_detail == invoiced_row.name])
-		pi.submit()
-
-		pr.reload()
-		self.assertEqual(pr.per_billed, 100)
-
-	@ERPNextTestSuite.change_settings(
-		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
-	)
-	def test_zero_rate_receipt_billed_by_qty(self):
-		pr = make_purchase_receipt(item_code="_Test Non Stock Item", qty=10, rate=0)
-		pi = make_purchase_invoice(pr.name)
-		pi.items[0].rate = 5
-		pi.submit()
-
-		pr.reload()
-		self.assertEqual(pr.per_billed, 100)
-
-	@ERPNextTestSuite.change_settings(
-		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
-	)
-	def test_non_updating_debit_note_kept_out_of_qty_billing(self):
-		from erpnext.accounts.doctype.purchase_invoice.mapper import make_debit_note
-
-		pr = make_purchase_receipt(qty=100, rate=50)
-		pi = make_purchase_invoice(pr.name)
-		pi.items[0].qty = 50
-		pi.submit()
-
-		debit_note = make_debit_note(pi.name)
-		debit_note.items[0].qty = -20
-		debit_note.update_billed_amount_in_purchase_receipt = 0
-		debit_note.submit()
-
-		pi = make_purchase_invoice(pr.name)
-		pi.items[0].qty = 50
-		pi.submit()
-
-		pr.reload()
-		self.assertEqual(pr.per_billed, 100)
-
-	@ERPNextTestSuite.change_settings(
-		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
-	)
-	def test_landed_cost_takes_order_invoices_oldest_first(self):
-		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
-
-		def get_amount_differences():
-			return [
-				frappe.db.get_value(
-					"Purchase Receipt Item", pr.items[0].name, "amount_difference_with_purchase_invoice"
-				)
-				for pr in receipts
-			]
-
-		po = create_purchase_order(qty=100, rate=50)
-		receipts = make_receipts_against_order(po.name, ((60, "08:00"), (40, "10:00")))
-
-		make_invoice_against_order(po.name, qty=60, rate=70)
-		self.assertEqual(get_amount_differences(), [1200, 0])
-
-		make_invoice_against_order(po.name, qty=40, rate=55)
-		self.assertEqual(get_amount_differences(), [1200, 200])
-
-	@ERPNextTestSuite.change_settings(
-		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
-	)
-	def test_landed_cost_refreshed_on_receipt_whose_share_moved(self):
-		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
-
-		po = create_purchase_order(qty=100, rate=50)
-		receipts = make_receipts_against_order(po.name, ((60, "08:00"), (40, "10:00")))
-
-		pi = make_purchase_invoice(receipts[0].name)
-		pi.items[0].qty = 1
-		pi.submit()
-		make_invoice_against_order(po.name, qty=69, rate=40)
-
-		second_row = receipts[1].items[0].name
-		self.assertEqual(frappe.db.get_value("Purchase Receipt Item", second_row, "billed_amt"), 0)
-		self.assertEqual(
-			frappe.db.get_value(
-				"Purchase Receipt Item", second_row, "amount_difference_with_purchase_invoice"
-			),
-			-400,
-		)
-
 	def test_serial_no_against_purchase_receipt(self):
 		item_code = "Test Manual Created Serial No"
 		if not frappe.db.exists("Item", item_code):
@@ -7934,31 +7771,6 @@ def get_items(**args):
 	]
 
 
-def make_receipts_against_order(purchase_order: str, receipts: tuple) -> list:
-	from erpnext.buying.doctype.purchase_order.mapper import make_purchase_receipt as make_receipt_from_order
-
-	receipt_docs = []
-	for qty, posting_time in receipts:
-		pr = make_receipt_from_order(purchase_order)
-		pr.set_posting_time = 1
-		pr.posting_time = posting_time
-		pr.items[0].received_qty = qty
-		pr.items[0].qty = qty
-		pr.submit()
-		receipt_docs.append(pr)
-
-	return receipt_docs
-
-
-def make_invoice_against_order(purchase_order: str, qty: float, rate: float) -> None:
-	from erpnext.buying.doctype.purchase_order.mapper import make_purchase_invoice as make_invoice_from_order
-
-	pi = make_invoice_from_order(purchase_order)
-	pi.items[0].qty = qty
-	pi.items[0].rate = rate
-	pi.submit()
-
-
 def make_purchase_receipt(**args):
 	frappe.db.set_single_value("Buying Settings", "allow_multiple_items", 1)
 	pr = frappe.new_doc("Purchase Receipt")