178 lines
6.7 KiB
Python
178 lines
6.7 KiB
Python
import frappe
|
|
from frappe import _
|
|
from frappe.utils import cint, flt, fmt_money
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_taxes_from_vendor_invoices(vendor_invoices, company):
|
|
"""Return service-item rows from the given Purchase Invoices, ready to
|
|
be appended to the Landed Cost Voucher `taxes` table.
|
|
|
|
Service items are identified by Item.is_stock_item = 0. Rows are grouped
|
|
by (item, expense_account) and amounts are summed in company currency.
|
|
expense_account falls back to Item.default_landed_cost_account when the
|
|
Purchase Invoice Item does not specify one.
|
|
"""
|
|
if isinstance(vendor_invoices, str):
|
|
import json
|
|
vendor_invoices = json.loads(vendor_invoices)
|
|
|
|
vendor_invoices = [v for v in (vendor_invoices or []) if v]
|
|
if not vendor_invoices:
|
|
return []
|
|
|
|
pii = frappe.qb.DocType("Purchase Invoice Item")
|
|
item = frappe.qb.DocType("Item")
|
|
rows = (
|
|
frappe.qb.from_(pii)
|
|
.left_join(item).on(item.name == pii.item_code)
|
|
.select(
|
|
pii.item_code,
|
|
pii.item_name,
|
|
pii.description,
|
|
pii.base_amount,
|
|
item.default_landed_cost_account,
|
|
)
|
|
.where(pii.parent.isin(vendor_invoices))
|
|
.where(pii.parenttype == "Purchase Invoice")
|
|
.where(item.is_stock_item == 0)
|
|
).run(as_dict=True)
|
|
|
|
company_currency = frappe.get_cached_value("Company", company, "default_currency")
|
|
|
|
# expense_account is taken strictly from Item.default_landed_cost_account.
|
|
# We deliberately ignore Purchase Invoice Item's expense_account so the
|
|
# account in LCV is always controlled from the Item master.
|
|
#
|
|
# base_amount/exchange_rate/account_currency are pre-filled so the
|
|
# standard client-side distribution (set_total_taxes_and_charges →
|
|
# set_applicable_charges_for_item) works immediately. PI base_amount is
|
|
# already in company currency, so amount == base_amount and rate == 1.
|
|
aggregated: dict[tuple, dict] = {}
|
|
for r in rows:
|
|
expense_account = r.default_landed_cost_account or None
|
|
key = (r.item_code, expense_account or "")
|
|
if key not in aggregated:
|
|
aggregated[key] = {
|
|
"item": r.item_code,
|
|
"description": r.item_name or r.description or r.item_code,
|
|
"expense_account": expense_account,
|
|
"amount": 0.0,
|
|
"base_amount": 0.0,
|
|
"exchange_rate": 1,
|
|
"account_currency": company_currency,
|
|
}
|
|
amt = flt(r.base_amount)
|
|
aggregated[key]["amount"] = flt(aggregated[key]["amount"]) + amt
|
|
aggregated[key]["base_amount"] = flt(aggregated[key]["base_amount"]) + amt
|
|
|
|
return list(aggregated.values())
|
|
|
|
|
|
def validate_taxes_rows(doc, method=None):
|
|
"""Per-row sanity checks on the Landed Cost Taxes and Charges table.
|
|
Runs on before_submit so drafts can still be saved with incomplete data.
|
|
|
|
Catches:
|
|
- item missing (defence in depth — also enforced by reqd=1);
|
|
- item not found / disabled;
|
|
- item is not a service (is_stock_item == 1);
|
|
- amount not positive;
|
|
- expense_account missing (means Default Landed Cost Account was not
|
|
set on the Item — submit must not silently leave the account empty);
|
|
- duplicate (item, expense_account) rows.
|
|
"""
|
|
rows = doc.get("taxes") or []
|
|
if not rows:
|
|
frappe.throw(
|
|
_("Landed Cost table cannot be empty on submit."),
|
|
title=_("Landed Cost Validation"),
|
|
)
|
|
|
|
seen_keys: set[tuple] = set()
|
|
item_cache: dict[str, dict] = {}
|
|
|
|
for r in rows:
|
|
idx = r.idx
|
|
|
|
if not r.item:
|
|
frappe.throw(
|
|
_("Row {0}: Item is required.").format(idx),
|
|
title=_("Landed Cost Validation"),
|
|
)
|
|
|
|
if r.item not in item_cache:
|
|
item_cache[r.item] = frappe.db.get_value(
|
|
"Item", r.item,
|
|
["is_stock_item", "disabled", "item_name"],
|
|
as_dict=True,
|
|
) or {}
|
|
item_data = item_cache[r.item]
|
|
|
|
if not item_data:
|
|
frappe.throw(
|
|
_("Row {0}: Item {1} does not exist.").format(idx, frappe.bold(r.item)),
|
|
title=_("Landed Cost Validation"),
|
|
)
|
|
if cint(item_data.get("is_stock_item")) == 1:
|
|
frappe.throw(
|
|
_("Row {0}: Item {1} is not a service. Only items with Maintain Stock disabled are allowed in Landed Cost.").format(
|
|
idx, frappe.bold(r.item),
|
|
),
|
|
title=_("Landed Cost Validation"),
|
|
)
|
|
if cint(item_data.get("disabled")) == 1:
|
|
frappe.throw(
|
|
_("Row {0}: Item {1} is disabled.").format(idx, frappe.bold(r.item)),
|
|
title=_("Landed Cost Validation"),
|
|
)
|
|
|
|
if flt(r.amount) <= 0:
|
|
frappe.throw(
|
|
_("Row {0}: Amount must be greater than zero.").format(idx),
|
|
title=_("Landed Cost Validation"),
|
|
)
|
|
|
|
if not r.expense_account:
|
|
frappe.throw(
|
|
_("Row {0}: Expense Account is missing for item {1}. Set Default Landed Cost Account on the Item or pick the account manually.").format(
|
|
idx, frappe.bold(r.item),
|
|
),
|
|
title=_("Landed Cost Validation"),
|
|
)
|
|
|
|
key = (r.item, r.expense_account)
|
|
if key in seen_keys:
|
|
frappe.throw(
|
|
_("Row {0}: Duplicate combination of Item {1} and Expense Account {2}. Merge the rows or change one of the values.").format(
|
|
idx, frappe.bold(r.item), frappe.bold(r.expense_account),
|
|
),
|
|
title=_("Landed Cost Validation"),
|
|
)
|
|
seen_keys.add(key)
|
|
|
|
|
|
def validate_total_services_against_invoices(doc, method=None):
|
|
"""Block submit when the sum of services in the taxes table exceeds
|
|
the total of vendor invoices. Skipped when no vendor invoices are
|
|
attached (LCV may be built only from purchase_receipts in that case).
|
|
"""
|
|
total_invoices = flt(doc.get("total_vendor_invoices_cost") or 0)
|
|
if not total_invoices:
|
|
return
|
|
|
|
total_services = sum(flt(r.base_amount) for r in (doc.get("taxes") or []))
|
|
|
|
# Tolerance for rounding noise in company currency precision.
|
|
if total_services <= total_invoices + 0.01:
|
|
return
|
|
|
|
company_currency = frappe.get_cached_value("Company", doc.company, "default_currency")
|
|
frappe.throw(
|
|
_("Total services in Landed Cost ({0}) cannot exceed Total Vendor Invoices Cost ({1}).").format(
|
|
fmt_money(total_services, currency=company_currency),
|
|
fmt_money(total_invoices, currency=company_currency),
|
|
),
|
|
title=_("Landed Cost Validation"),
|
|
)
|