Merge branch 'fix/landed-cost-po-invoice-fifo' into fix/refresh-order-line-receipts

The real reason the extra invoice went through in CI: an earlier test leaves "Bill for Rejected Quantity" cached on, and with it on the Purchase Invoice skips the over-billing error entirely. The test now pins that setting off, and the fresh item from the previous attempt is dropped because the allowance was never the cause.

erpnext/accounts/doctype/purchase_invoice/purchase_invoice.py

@@ -40,6 +40,7 @@ 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):
@@ -296,7 +297,12 @@ class PurchaseInvoice(BuyingController):
 
 		from erpnext.accounts.services.billing_validation import BillingValidationService
 
-		BillingValidationService(self).validate_multiple_billing("Purchase Receipt", "pr_detail", "amount")
+		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")
 		self.set_status()
 		self.validate_purchase_receipt_if_update_stock()
 		self.validate_exchange_rate_with_purchase_receipt()
@@ -589,6 +595,9 @@ 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(
 				{
@@ -639,6 +648,13 @@ 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"):
@@ -672,6 +688,7 @@ 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
@@ -786,6 +803,7 @@ 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 billing sync and provisional-entry cancellation for Purchase Invoice."""
+"""Purchase Receipt and Purchase Order billing sync and provisional-entry cancellation for Purchase Invoice."""
 
 import frappe
 from frappe import qb
@@ -9,6 +9,9 @@ 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,
 )
@@ -44,16 +47,52 @@ class BillingStatusService:
 		if po_details:
 			updated_pr += update_billed_amount_based_on_po(po_details, update_modified)
 
-		adjust_incoming_rate = frappe.db.get_single_value(
-			"Buying Settings", "set_landed_cost_based_on_purchase_invoice_rate"
-		)
+		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
 
-		for pr in set(updated_pr):
-			pr_doc = frappe.get_lazy_doc("Purchase Receipt", pr)
+		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"
+		)
+		invoiced = get_invoiced_qty_and_amount(pr_items, bill_for_rejected)
+
+		for pr_doc in pr_docs:
 			update_billing_percentage(
-				pr_doc, update_modified=update_modified, adjust_incoming_rate=adjust_incoming_rate
+				pr_doc,
+				update_modified=update_modified,
+				adjust_incoming_rate=True,
+				invoiced=invoiced,
 			)
 
+	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,6 +3787,30 @@ 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)
+			self.throw_overbill_exception(overbilled_items, precision, based_on)
 
 		if is_overbilling_allowed and total_overbilled_amt > 0.1:
 			frappe.msgprint(
@@ -92,7 +92,9 @@ class BillingValidationService:
 
 			ref_wise_billed_amount.setdefault(
 				key,
-				frappe._dict(item_code=item.item_code, billed_amt=0.0, ref_amt=ref_amt, rows=[]),
+				frappe._dict(
+					item_code=item.item_code, uom=item.get("uom"), 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
@@ -131,7 +133,7 @@ class BillingValidationService:
 			).run()
 		)
 
-	def throw_overbill_exception(self, overbilled_items: list, precision: int) -> None:
+	def throw_overbill_exception(self, overbilled_items: list, precision: int, based_on: str) -> None:
 		message = (
 			_("<p>Cannot overbill for the following Items:</p>")
 			+ "<ul>"
@@ -139,9 +141,7 @@ 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(
-						fmt_money(item.max_allowed_amt, precision=precision, currency=self.doc.currency)
-					),
+					frappe.bold(self.get_formatted_limit(item, precision, based_on)),
 				)
 				for item in overbilled_items
 			)
@@ -149,3 +149,9 @@ 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,6 +10,10 @@ 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:
@@ -61,3 +65,9 @@ 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,6 +1521,49 @@ 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:
+			if items_to_validate and args.get("validate_overflow") is not False:
 				pp_sub_assembly_items = [
 					item.production_plan_sub_assembly_item
 					for item in items_to_validate

erpnext/controllers/stock_controller.py

@@ -33,6 +33,7 @@ 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
 
@@ -340,6 +341,10 @@ 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/stock/doctype/purchase_receipt/services/billing_status.py

@@ -115,6 +115,7 @@ 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,
@@ -180,30 +181,46 @@ 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
+	pr_doc, update_modified: bool = True, adjust_incoming_rate: bool = False, invoiced: dict | None = None
 ) -> None:
-	# Update Billing % based on pending accepted qty
+	"""`invoiced` from get_invoiced_qty_and_amount lets receipts on one order line share its invoice split."""
 	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, pi_landed_cost_amount = 0, 0, 0
-	item_wise_returned_qty = get_item_wise_returned_qty(pr_doc)
-	billed_qty_amt = frappe._dict()
+	total_amount, total_billed_amount = 0, 0
+	item_wise_returned_qty = get_item_wise_returned_qty([item.name for item in pr_doc.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:
+	for item in 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 buying_settings.bill_for_rejected_quantity_in_purchase_invoice:
+		if bill_for_rejected:
 			rejected_amount = flt(item.rejected_qty * item.rate, item.precision("amount"))
 			pending_amount = flt(item.amount) + rejected_amount
 
@@ -219,54 +236,7 @@ def update_billing_percentage(
 
 		amount = flt(item.amount) + rejected_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:
+		if amount and item.billed_amt > amount:
 			per_over_billed = (flt(item.billed_amt / amount, 2) * 100) - 100
 			if (
 				per_over_billed > over_billing_allowance
@@ -278,22 +248,212 @@ def update_billing_percentage(
 					)
 				)
 
-	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)
+	return round(100 * (total_billed_amount / (total_amount or 1)), 6)
 
 
-def get_billed_qty_amount_against_purchase_receipt(pr_doc) -> dict:
-	pr_names = [d.name for d in pr_doc.items]
+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()
+
 	parent_table = frappe.qb.DocType("Purchase Invoice")
 	table = frappe.qb.DocType("Purchase Invoice Item")
 	query = (
@@ -305,7 +465,11 @@ def get_billed_qty_amount_against_purchase_receipt(pr_doc) -> 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))
+		.where(
+			(table.pr_detail.isin(pr_names))
+			& (table.docstatus == 1)
+			& ((parent_table.is_return == 0) | (parent_table.update_billed_amount_in_purchase_receipt == 1))
+		)
 		.groupby(table.pr_detail)
 	)
 	invoice_data = query.run(as_dict=1)
@@ -325,49 +489,6 @@ def get_billed_qty_amount_against_purchase_receipt(pr_doc) -> 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)
 
@@ -380,9 +501,7 @@ def adjust_incoming_rate_for_pr(doc) -> None:
 	doc.repost_future_sle_and_gle(force=True)
 
 
-def get_item_wise_returned_qty(pr_doc) -> dict:
-	items = [d.name for d in pr_doc.items]
-
+def get_item_wise_returned_qty(items: list) -> dict:
 	return frappe._dict(
 		frappe.get_all(
 			"Purchase Receipt",

erpnext/stock/doctype/purchase_receipt/test_purchase_receipt.py

@@ -1117,6 +1117,169 @@ 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):
@@ -7771,6 +7934,31 @@ 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")