mirror of
https://github.com/frappe/erpnext.git
synced 2026-08-19 09:28:46 +00:00
refactor(sales_invoice): extract non-GL services (Phase 7)
Split the sales_invoice.py monolith into focused service modules under sales_invoice/services/: - fixed_assets.py — FixedAssetService (depreciation, disposal, split) - inter_company.py — validate/link/unlink inter-company docs - loyalty.py — LoyaltyService (earn, redeem, delete points) - pos.py — POSService + POS free functions - status.py — StatusService + is_overdue / get_discounting_status - timesheet_billing.py — TimesheetBillingService Lifecycle hooks (validate/on_submit/on_cancel) call services directly; no thin shims. The 7 methods POS Invoice calls via self.* are kept on the class with an explicit comment. @frappe.whitelist() doc-methods and framework hooks (set_status, set_indicator) stay on the class. sales_invoice.py: 2156 → 1205 lines. All 29 snapshot + 121 SI tests green. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Nabin Hait
File diff suppressed because it is too large
Load Diff
173
erpnext/accounts/doctype/sales_invoice/services/fixed_assets.py
Normal file
173
erpnext/accounts/doctype/sales_invoice/services/fixed_assets.py
Normal file
@@ -0,0 +1,173 @@
|
||||
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# License: GNU General Public License v3. See license.txt
|
||||
|
||||
"""Fixed asset lifecycle helpers for Sales Invoice."""
|
||||
|
||||
import frappe
|
||||
from frappe import _
|
||||
from frappe.utils import flt, get_link_to_form
|
||||
|
||||
from erpnext.assets.doctype.asset.asset import split_asset
|
||||
from erpnext.assets.doctype.asset.depreciation import (
|
||||
depreciate_asset,
|
||||
reset_depreciation_schedule,
|
||||
reverse_depreciation_entry_made_on_disposal,
|
||||
)
|
||||
from erpnext.assets.doctype.asset_activity.asset_activity import add_asset_activity
|
||||
|
||||
|
||||
class FixedAssetService:
|
||||
def __init__(self, doc):
|
||||
self.doc = doc
|
||||
|
||||
def validate_fixed_asset(self) -> None:
|
||||
doc = self.doc
|
||||
if doc.doctype != "Sales Invoice":
|
||||
return
|
||||
|
||||
for d in doc.get("items"):
|
||||
if not d.is_fixed_asset:
|
||||
continue
|
||||
|
||||
if d.asset:
|
||||
if not doc.is_return:
|
||||
asset_status = frappe.db.get_value("Asset", d.asset, "status")
|
||||
if doc.update_stock:
|
||||
frappe.throw(_("'Update Stock' cannot be checked for fixed asset sale"))
|
||||
elif asset_status in ("Scrapped", "Cancelled", "Capitalized"):
|
||||
frappe.throw(
|
||||
_("Row #{0}: Asset {1} cannot be sold, it is already {2}").format(
|
||||
d.idx, d.asset, asset_status
|
||||
)
|
||||
)
|
||||
elif asset_status == "Sold" and not doc.is_return:
|
||||
frappe.throw(_("Row #{0}: Asset {1} is already sold").format(d.idx, d.asset))
|
||||
elif not doc.return_against:
|
||||
frappe.throw(_("Row #{0}: Return Against is required for returning asset").format(d.idx))
|
||||
else:
|
||||
frappe.throw(
|
||||
_("Row #{0}: You must select an Asset for Item {1}.").format(d.idx, d.item_code),
|
||||
title=_("Missing Asset"),
|
||||
)
|
||||
|
||||
def set_income_account_for_fixed_assets(self) -> None:
|
||||
for item in self.doc.items:
|
||||
item.set_income_account_for_fixed_asset(self.doc.company)
|
||||
|
||||
def process_asset_depreciation(self) -> None:
|
||||
doc = self.doc
|
||||
if doc.is_internal_transfer():
|
||||
return
|
||||
|
||||
if (doc.is_return and doc.docstatus == 2) or (not doc.is_return and doc.docstatus == 1):
|
||||
self._depreciate_asset_on_sale()
|
||||
else:
|
||||
self._restore_asset()
|
||||
|
||||
self._update_asset()
|
||||
|
||||
def split_asset_based_on_sale_qty(self) -> None:
|
||||
asset_qty_map = self._get_asset_qty()
|
||||
for asset, qty in asset_qty_map.items():
|
||||
if qty["actual_qty"] < qty["sale_qty"]:
|
||||
frappe.throw(
|
||||
_(
|
||||
"Sell quantity cannot exceed the asset quantity. Asset {0} has only {1} item(s)."
|
||||
).format(asset, qty["actual_qty"])
|
||||
)
|
||||
|
||||
remaining_qty = qty["actual_qty"] - qty["sale_qty"]
|
||||
if remaining_qty > 0:
|
||||
split_asset(asset, remaining_qty)
|
||||
|
||||
def get_disposal_date(self) -> str:
|
||||
doc = self.doc
|
||||
if doc.is_return:
|
||||
return frappe.db.get_value("Sales Invoice", doc.return_against, "posting_date")
|
||||
return doc.posting_date
|
||||
|
||||
def _depreciate_asset_on_sale(self) -> None:
|
||||
disposal_date = self.get_disposal_date()
|
||||
for d in self.doc.get("items"):
|
||||
if d.asset:
|
||||
asset = frappe.get_doc("Asset", d.asset)
|
||||
if asset.calculate_depreciation and asset.status != "Fully Depreciated":
|
||||
depreciate_asset(asset, disposal_date, self._get_note_for_asset_sale(asset))
|
||||
|
||||
def _restore_asset(self) -> None:
|
||||
for d in self.doc.get("items"):
|
||||
if d.asset:
|
||||
asset = frappe.get_cached_doc("Asset", d.asset)
|
||||
if asset.calculate_depreciation:
|
||||
reverse_depreciation_entry_made_on_disposal(asset)
|
||||
reset_depreciation_schedule(asset, self._get_note_for_asset_return(asset))
|
||||
|
||||
def _update_asset(self) -> None:
|
||||
doc = self.doc
|
||||
disposal_date = self.get_disposal_date()
|
||||
|
||||
for d in doc.get("items"):
|
||||
if not d.asset:
|
||||
continue
|
||||
|
||||
asset = frappe.get_cached_doc("Asset", d.asset)
|
||||
|
||||
if (doc.is_return and doc.docstatus == 1) or (not doc.is_return and doc.docstatus == 2):
|
||||
note = _("Asset returned") if doc.is_return else _("Asset sold")
|
||||
asset_status, disposal_date = None, None
|
||||
else:
|
||||
note = _("Asset sold") if not doc.is_return else _("Return invoice of asset cancelled")
|
||||
asset_status = "Sold"
|
||||
|
||||
frappe.db.set_value("Asset", d.asset, "disposal_date", disposal_date)
|
||||
add_asset_activity(asset.name, note)
|
||||
asset.set_status(asset_status)
|
||||
|
||||
def _get_asset_qty(self) -> dict:
|
||||
doc = self.doc
|
||||
asset_qty_map = {}
|
||||
|
||||
assets = {row.asset for row in doc.items if row.is_fixed_asset and row.asset}
|
||||
if not assets or doc.is_return:
|
||||
return asset_qty_map
|
||||
|
||||
asset_actual_qty = dict(
|
||||
frappe.db.get_all(
|
||||
"Asset",
|
||||
{"name": ["in", list(assets)]},
|
||||
["name", "asset_quantity"],
|
||||
as_list=True,
|
||||
)
|
||||
)
|
||||
for row in doc.items:
|
||||
if row.is_fixed_asset and row.asset:
|
||||
actual_qty = asset_actual_qty.get(row.asset)
|
||||
if row.asset in asset_qty_map:
|
||||
asset_qty_map[row.asset]["sale_qty"] += flt(row.qty)
|
||||
else:
|
||||
asset_qty_map[row.asset] = {
|
||||
"sale_qty": flt(row.qty),
|
||||
"actual_qty": flt(actual_qty),
|
||||
}
|
||||
|
||||
return asset_qty_map
|
||||
|
||||
def _get_note_for_asset_sale(self, asset) -> str:
|
||||
doc = self.doc
|
||||
return _("This schedule was created when Asset {0} was {1} through Sales Invoice {2}.").format(
|
||||
get_link_to_form(asset.doctype, asset.name),
|
||||
_("returned") if doc.is_return else _("sold"),
|
||||
get_link_to_form(doc.doctype, doc.get("name")),
|
||||
)
|
||||
|
||||
def _get_note_for_asset_return(self, asset) -> str:
|
||||
doc = self.doc
|
||||
asset_link = get_link_to_form(asset.doctype, asset.name)
|
||||
invoice_link = get_link_to_form(doc.doctype, doc.get("name"))
|
||||
if doc.is_return:
|
||||
return _(
|
||||
"This schedule was created when Asset {0} was returned through Sales Invoice {1}."
|
||||
).format(asset_link, invoice_link)
|
||||
return _(
|
||||
"This schedule was created when Asset {0} was restored due to Sales Invoice {1} cancellation."
|
||||
).format(asset_link, invoice_link)
|
||||
@@ -0,0 +1,68 @@
|
||||
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# License: GNU General Public License v3. See license.txt
|
||||
|
||||
"""Inter-company transaction helpers for Sales Invoice."""
|
||||
|
||||
import frappe
|
||||
from frappe import _
|
||||
|
||||
|
||||
def validate_inter_company_party(
|
||||
doctype: str, party: str, company: str, inter_company_reference: str | None
|
||||
) -> None:
|
||||
if not party:
|
||||
return
|
||||
|
||||
if doctype in ["Sales Invoice", "Sales Order"]:
|
||||
partytype, ref_partytype, internal = "Customer", "Supplier", "is_internal_customer"
|
||||
ref_doc = "Purchase Invoice" if doctype == "Sales Invoice" else "Purchase Order"
|
||||
else:
|
||||
partytype, ref_partytype, internal = "Supplier", "Customer", "is_internal_supplier"
|
||||
ref_doc = "Sales Invoice" if doctype == "Purchase Invoice" else "Sales Order"
|
||||
|
||||
if inter_company_reference:
|
||||
doc = frappe.get_doc(ref_doc, inter_company_reference)
|
||||
ref_party = doc.supplier if doctype in ["Sales Invoice", "Sales Order"] else doc.customer
|
||||
if frappe.db.get_value(partytype, {"represents_company": doc.company}, "name") != party:
|
||||
frappe.throw(_("Invalid {0} for Inter Company Transaction.").format(_(partytype)))
|
||||
if frappe.get_cached_value(ref_partytype, ref_party, "represents_company") != company:
|
||||
frappe.throw(_("Invalid Company for Inter Company Transaction."))
|
||||
|
||||
elif frappe.db.get_value(partytype, {"name": party, internal: 1}, "name") == party:
|
||||
companies = [
|
||||
d.company
|
||||
for d in frappe.get_all(
|
||||
"Allowed To Transact With",
|
||||
fields=["company"],
|
||||
filters={"parenttype": partytype, "parent": party},
|
||||
)
|
||||
]
|
||||
if company not in companies:
|
||||
frappe.throw(
|
||||
_(
|
||||
"{0} not allowed to transact with {1}. Please change the Company or add the Company in the 'Allowed To Transact With'-Section in the Customer record."
|
||||
).format(_(partytype), company)
|
||||
)
|
||||
|
||||
|
||||
def update_linked_doc(doctype: str, name: str, inter_company_reference: str | None) -> None:
|
||||
ref_field = (
|
||||
"inter_company_invoice_reference"
|
||||
if doctype in ["Sales Invoice", "Purchase Invoice"]
|
||||
else "inter_company_order_reference"
|
||||
)
|
||||
if inter_company_reference:
|
||||
frappe.db.set_value(doctype, inter_company_reference, ref_field, name)
|
||||
|
||||
|
||||
def unlink_inter_company_doc(doctype: str, name: str, inter_company_reference: str | None) -> None:
|
||||
if doctype in ["Sales Invoice", "Purchase Invoice"]:
|
||||
ref_doc = "Purchase Invoice" if doctype == "Sales Invoice" else "Sales Invoice"
|
||||
ref_field = "inter_company_invoice_reference"
|
||||
else:
|
||||
ref_doc = "Purchase Order" if doctype == "Sales Order" else "Sales Order"
|
||||
ref_field = "inter_company_order_reference"
|
||||
|
||||
if inter_company_reference:
|
||||
frappe.db.set_value(doctype, name, ref_field, "")
|
||||
frappe.db.set_value(ref_doc, inter_company_reference, ref_field, "")
|
||||
162
erpnext/accounts/doctype/sales_invoice/services/loyalty.py
Normal file
162
erpnext/accounts/doctype/sales_invoice/services/loyalty.py
Normal file
@@ -0,0 +1,162 @@
|
||||
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# License: GNU General Public License v3. See license.txt
|
||||
|
||||
"""Loyalty program helpers for Sales Invoice."""
|
||||
|
||||
import frappe
|
||||
from frappe import _
|
||||
from frappe.utils import add_days, cint, flt, getdate
|
||||
|
||||
from erpnext.accounts.doctype.loyalty_program.loyalty_program import (
|
||||
get_loyalty_program_details_with_points,
|
||||
)
|
||||
|
||||
|
||||
class LoyaltyService:
|
||||
def __init__(self, doc):
|
||||
self.doc = doc
|
||||
|
||||
def make_loyalty_point_entry(self) -> None:
|
||||
doc = self.doc
|
||||
returned_amount = self._get_returned_amount()
|
||||
current_amount = flt(doc.grand_total) - cint(doc.loyalty_amount)
|
||||
eligible_amount = current_amount - returned_amount
|
||||
lp_details = get_loyalty_program_details_with_points(
|
||||
doc.customer,
|
||||
company=doc.company,
|
||||
current_transaction_amount=current_amount,
|
||||
loyalty_program=doc.loyalty_program,
|
||||
expiry_date=doc.posting_date,
|
||||
include_expired_entry=True,
|
||||
)
|
||||
if (
|
||||
lp_details
|
||||
and getdate(lp_details.from_date) <= getdate(doc.posting_date)
|
||||
and (not lp_details.to_date or getdate(lp_details.to_date) >= getdate(doc.posting_date))
|
||||
):
|
||||
collection_factor = lp_details.collection_factor if lp_details.collection_factor else 1.0
|
||||
points_earned = cint(eligible_amount / collection_factor)
|
||||
|
||||
entry = frappe.get_doc(
|
||||
{
|
||||
"doctype": "Loyalty Point Entry",
|
||||
"company": doc.company,
|
||||
"loyalty_program": lp_details.loyalty_program,
|
||||
"loyalty_program_tier": lp_details.tier_name,
|
||||
"customer": doc.customer,
|
||||
"invoice_type": doc.doctype,
|
||||
"invoice": doc.name,
|
||||
"loyalty_points": points_earned,
|
||||
"purchase_amount": eligible_amount,
|
||||
"expiry_date": add_days(doc.posting_date, lp_details.expiry_duration),
|
||||
"posting_date": doc.posting_date,
|
||||
}
|
||||
)
|
||||
entry.flags.ignore_permissions = 1
|
||||
entry.save()
|
||||
self._set_loyalty_program_tier()
|
||||
|
||||
def delete_loyalty_point_entry(self) -> None:
|
||||
doc = self.doc
|
||||
lp_entry = frappe.db.sql(
|
||||
"select name from `tabLoyalty Point Entry` where invoice=%s", (doc.name), as_dict=1
|
||||
)
|
||||
|
||||
if not lp_entry:
|
||||
return
|
||||
|
||||
against_lp_entry = frappe.db.sql(
|
||||
"""select name, invoice from `tabLoyalty Point Entry`
|
||||
where redeem_against=%s""",
|
||||
(lp_entry[0].name),
|
||||
as_dict=1,
|
||||
)
|
||||
if against_lp_entry:
|
||||
invoice_list = ", ".join([d.invoice for d in against_lp_entry])
|
||||
frappe.throw(
|
||||
_(
|
||||
"""{} can't be cancelled since the Loyalty Points earned has been redeemed. First cancel the {} No {}"""
|
||||
).format(doc.doctype, doc.doctype, invoice_list)
|
||||
)
|
||||
else:
|
||||
frappe.db.sql("""delete from `tabLoyalty Point Entry` where invoice=%s""", (doc.name))
|
||||
self._set_loyalty_program_tier()
|
||||
|
||||
def apply_loyalty_points(self) -> None:
|
||||
from erpnext.accounts.doctype.loyalty_point_entry.loyalty_point_entry import (
|
||||
get_loyalty_point_entries,
|
||||
get_redemption_details,
|
||||
)
|
||||
|
||||
doc = self.doc
|
||||
loyalty_point_entries = get_loyalty_point_entries(
|
||||
doc.customer, doc.loyalty_program, doc.company, doc.posting_date
|
||||
)
|
||||
redemption_details = get_redemption_details(doc.customer, doc.loyalty_program, doc.company)
|
||||
|
||||
points_to_redeem = doc.loyalty_points
|
||||
for lp_entry in loyalty_point_entries:
|
||||
if lp_entry.invoice_type != doc.doctype or lp_entry.invoice == doc.name:
|
||||
continue
|
||||
available_points = lp_entry.loyalty_points - flt(redemption_details.get(lp_entry.name))
|
||||
redeemed_points = min(available_points, points_to_redeem)
|
||||
entry = frappe.get_doc(
|
||||
{
|
||||
"doctype": "Loyalty Point Entry",
|
||||
"company": doc.company,
|
||||
"loyalty_program": doc.loyalty_program,
|
||||
"loyalty_program_tier": lp_entry.loyalty_program_tier,
|
||||
"customer": doc.customer,
|
||||
"invoice_type": doc.doctype,
|
||||
"invoice": doc.name,
|
||||
"redeem_against": lp_entry.name,
|
||||
"loyalty_points": -1 * redeemed_points,
|
||||
"purchase_amount": doc.grand_total,
|
||||
"expiry_date": lp_entry.expiry_date,
|
||||
"posting_date": doc.posting_date,
|
||||
}
|
||||
)
|
||||
entry.flags.ignore_permissions = 1
|
||||
entry.save()
|
||||
points_to_redeem -= redeemed_points
|
||||
if points_to_redeem < 1:
|
||||
break
|
||||
|
||||
def _set_loyalty_program_tier(self) -> None:
|
||||
doc = self.doc
|
||||
lp_details = get_loyalty_program_details_with_points(
|
||||
doc.customer,
|
||||
company=doc.company,
|
||||
loyalty_program=doc.loyalty_program,
|
||||
include_expired_entry=True,
|
||||
)
|
||||
customer = frappe.get_doc("Customer", doc.customer)
|
||||
customer.db_set("loyalty_program_tier", lp_details.tier_name)
|
||||
|
||||
def _get_returned_amount(self) -> float:
|
||||
from frappe.query_builder.functions import Sum
|
||||
|
||||
doc = frappe.qb.DocType(self.doc.doctype)
|
||||
returned_amount = (
|
||||
frappe.qb.from_(doc)
|
||||
.select(Sum(doc.grand_total))
|
||||
.where((doc.docstatus == 1) & (doc.is_return == 1) & (doc.return_against == self.doc.name))
|
||||
).run()
|
||||
|
||||
return abs(returned_amount[0][0]) if returned_amount[0][0] else 0
|
||||
|
||||
|
||||
def get_loyalty_programs(customer: str) -> list:
|
||||
"""Return applicable loyalty programs for the customer."""
|
||||
from erpnext.selling.doctype.customer.customer import get_loyalty_programs as _get
|
||||
|
||||
customer_doc = frappe.get_doc("Customer", customer)
|
||||
if customer_doc.loyalty_program:
|
||||
return [customer_doc.loyalty_program]
|
||||
|
||||
lp_details = _get(customer_doc)
|
||||
|
||||
if len(lp_details) == 1:
|
||||
customer_doc.db_set("loyalty_program", lp_details[0])
|
||||
|
||||
return lp_details
|
||||
396
erpnext/accounts/doctype/sales_invoice/services/pos.py
Normal file
396
erpnext/accounts/doctype/sales_invoice/services/pos.py
Normal file
@@ -0,0 +1,396 @@
|
||||
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# License: GNU General Public License v3. See license.txt
|
||||
|
||||
"""POS helpers for Sales Invoice."""
|
||||
|
||||
import frappe
|
||||
from frappe import _, msgprint
|
||||
from frappe.utils import cint, flt, get_link_to_form
|
||||
|
||||
|
||||
class PartialPaymentValidationError(frappe.ValidationError):
|
||||
pass
|
||||
|
||||
|
||||
class POSService:
|
||||
def __init__(self, doc):
|
||||
self.doc = doc
|
||||
|
||||
def set_pos_fields(self, for_validate: bool = False) -> frappe.Document | None:
|
||||
"""Populate POS-profile fields on the invoice; return the profile or None."""
|
||||
doc = self.doc
|
||||
if cint(doc.is_pos) != 1:
|
||||
return None
|
||||
|
||||
if not doc.account_for_change_amount:
|
||||
doc.account_for_change_amount = frappe.get_cached_value(
|
||||
"Company", doc.company, "default_cash_account"
|
||||
)
|
||||
|
||||
from erpnext.stock.get_item_details import (
|
||||
ItemDetailsCtx,
|
||||
get_pos_profile,
|
||||
get_pos_profile_item_details_,
|
||||
)
|
||||
|
||||
if not doc.pos_profile and not doc.flags.ignore_pos_profile:
|
||||
pos_profile = get_pos_profile(doc.company) or {}
|
||||
if not pos_profile:
|
||||
return None
|
||||
doc.pos_profile = pos_profile.get("name")
|
||||
|
||||
pos = {}
|
||||
if doc.pos_profile:
|
||||
pos = frappe.get_doc("POS Profile", doc.pos_profile)
|
||||
|
||||
if pos:
|
||||
if not for_validate:
|
||||
update_multi_mode_option(doc, pos)
|
||||
doc.tax_category = pos.get("tax_category")
|
||||
|
||||
if not for_validate and not doc.customer:
|
||||
doc.customer = pos.customer
|
||||
|
||||
if not for_validate:
|
||||
doc.ignore_pricing_rule = pos.ignore_pricing_rule
|
||||
|
||||
if pos.get("account_for_change_amount"):
|
||||
doc.account_for_change_amount = pos.get("account_for_change_amount")
|
||||
|
||||
for fieldname in (
|
||||
"currency",
|
||||
"letter_head",
|
||||
"tc_name",
|
||||
"company",
|
||||
"select_print_heading",
|
||||
"write_off_account",
|
||||
"taxes_and_charges",
|
||||
"write_off_cost_center",
|
||||
"apply_discount_on",
|
||||
"cost_center",
|
||||
):
|
||||
if (not for_validate) or (for_validate and not doc.get(fieldname)):
|
||||
doc.set(fieldname, pos.get(fieldname))
|
||||
|
||||
if pos.get("company_address"):
|
||||
doc.company_address = pos.get("company_address")
|
||||
|
||||
if doc.customer:
|
||||
customer_price_list, customer_group = frappe.get_value(
|
||||
"Customer", doc.customer, ["default_price_list", "customer_group"]
|
||||
)
|
||||
customer_group_price_list = frappe.get_value(
|
||||
"Customer Group", customer_group, "default_price_list"
|
||||
)
|
||||
selling_price_list = (
|
||||
customer_price_list or customer_group_price_list or pos.get("selling_price_list")
|
||||
)
|
||||
else:
|
||||
selling_price_list = pos.get("selling_price_list")
|
||||
|
||||
if selling_price_list:
|
||||
doc.set("selling_price_list", selling_price_list)
|
||||
|
||||
if not for_validate:
|
||||
doc.update_stock = cint(pos.get("update_stock"))
|
||||
|
||||
for item in doc.get("items"):
|
||||
if item.get("item_code"):
|
||||
profile_details = get_pos_profile_item_details_(
|
||||
ItemDetailsCtx(item.as_dict()), pos, pos, update_data=True
|
||||
)
|
||||
for fname, val in profile_details.items():
|
||||
if (not for_validate) or (for_validate and not item.get(fname)):
|
||||
item.set(fname, val)
|
||||
|
||||
if doc.tc_name and not doc.terms:
|
||||
doc.terms = frappe.db.get_value("Terms and Conditions", doc.tc_name, "terms")
|
||||
|
||||
if doc.taxes_and_charges and not len(doc.get("taxes")):
|
||||
from erpnext.accounts.services.taxes import TaxService
|
||||
|
||||
TaxService(doc).set_taxes()
|
||||
|
||||
return pos
|
||||
|
||||
def set_paid_amount(self) -> None:
|
||||
doc = self.doc
|
||||
paid_amount = 0.0
|
||||
base_paid_amount = 0.0
|
||||
for data in doc.payments:
|
||||
data.base_amount = flt(data.amount * doc.conversion_rate, doc.precision("base_paid_amount"))
|
||||
paid_amount += data.amount
|
||||
base_paid_amount += data.base_amount
|
||||
doc.paid_amount = paid_amount
|
||||
doc.base_paid_amount = base_paid_amount
|
||||
|
||||
def set_account_for_mode_of_payment(self) -> None:
|
||||
for payment in self.doc.payments:
|
||||
payment.account = get_bank_cash_account(payment.mode_of_payment, self.doc.company).get("account")
|
||||
|
||||
def reset_mode_of_payments(self) -> None:
|
||||
doc = self.doc
|
||||
if doc.pos_profile:
|
||||
pos_profile = frappe.get_cached_doc("POS Profile", doc.pos_profile)
|
||||
update_multi_mode_option(doc, pos_profile)
|
||||
doc.paid_amount = 0
|
||||
|
||||
def validate_pos_return(self) -> None:
|
||||
doc = self.doc
|
||||
if doc.is_consolidated:
|
||||
return
|
||||
|
||||
if doc.is_pos and doc.is_return:
|
||||
total_amount_in_payments = sum(payment.amount for payment in doc.payments)
|
||||
invoice_total = doc.rounded_total or doc.grand_total
|
||||
if total_amount_in_payments < invoice_total:
|
||||
frappe.throw(_("Total payments amount can't be greater than {}").format(-invoice_total))
|
||||
|
||||
def validate_pos_paid_amount(self) -> None:
|
||||
doc = self.doc
|
||||
if len(doc.payments) == 0 and doc.is_pos and flt(doc.grand_total) > 0:
|
||||
frappe.throw(_("At least one mode of payment is required for POS invoice."))
|
||||
|
||||
def validate_pos(self) -> None:
|
||||
doc = self.doc
|
||||
if doc.is_return:
|
||||
invoice_total = doc.rounded_total or doc.grand_total
|
||||
if abs(flt(doc.paid_amount)) + abs(flt(doc.write_off_amount)) - abs(flt(invoice_total)) > 1.0 / (
|
||||
10.0 ** (doc.precision("grand_total") + 1.0)
|
||||
):
|
||||
frappe.throw(_("Paid amount + Write Off Amount can not be greater than Grand Total"))
|
||||
|
||||
def validate_created_using_pos(self) -> None:
|
||||
doc = self.doc
|
||||
if doc.is_created_using_pos and not doc.pos_profile:
|
||||
frappe.throw(_("POS Profile is mandatory to mark this invoice as POS Transaction."))
|
||||
|
||||
doc.invoice_type_in_pos = frappe.db.get_single_value("POS Settings", "invoice_type")
|
||||
if doc.invoice_type_in_pos == "POS Invoice" and not doc.is_return:
|
||||
frappe.throw(_("Transactions using Sales Invoice in POS are disabled."))
|
||||
|
||||
self.validate_pos_opening_entry()
|
||||
|
||||
def validate_full_payment(self) -> None:
|
||||
doc = self.doc
|
||||
allow_partial_payment = frappe.db.get_value("POS Profile", doc.pos_profile, "allow_partial_payment")
|
||||
invoice_total = flt(doc.rounded_total) or flt(doc.grand_total)
|
||||
|
||||
if (
|
||||
doc.docstatus == 1
|
||||
and not doc.is_return
|
||||
and not allow_partial_payment
|
||||
and doc.paid_amount < invoice_total
|
||||
):
|
||||
frappe.throw(
|
||||
msg=_("Partial Payment in POS Transactions are not allowed."),
|
||||
exc=PartialPaymentValidationError,
|
||||
)
|
||||
|
||||
def validate_pos_opening_entry(self) -> None:
|
||||
doc = self.doc
|
||||
opening_entries = frappe.get_all(
|
||||
"POS Opening Entry",
|
||||
fields=["name", "period_start_date"],
|
||||
filters={"pos_profile": doc.pos_profile, "status": "Open"},
|
||||
order_by="period_start_date desc",
|
||||
)
|
||||
if not opening_entries:
|
||||
frappe.throw(
|
||||
title=_("POS Opening Entry Missing"),
|
||||
msg=_("No open POS Opening Entry found for POS Profile {0}.").format(
|
||||
frappe.bold(doc.pos_profile)
|
||||
),
|
||||
)
|
||||
if len(opening_entries) > 1:
|
||||
frappe.throw(
|
||||
title=_("Multiple POS Opening Entry"),
|
||||
msg=_(
|
||||
"POS Profile - {0} has multiple open POS Opening Entries. Please close or cancel the existing entries before proceeding."
|
||||
).format(doc.pos_profile),
|
||||
)
|
||||
if frappe.utils.get_date_str(opening_entries[0].get("period_start_date")) != frappe.utils.today():
|
||||
frappe.throw(
|
||||
title=_("Outdated POS Opening Entry"),
|
||||
msg=_(
|
||||
"POS Opening Entry - {0} is outdated. Please close the POS and create a new POS Opening Entry."
|
||||
).format(opening_entries[0].get("name")),
|
||||
)
|
||||
|
||||
def check_if_consolidated_invoice(self) -> None:
|
||||
doc = self.doc
|
||||
if doc.doctype == "Sales Invoice" and doc.is_consolidated:
|
||||
invoice_or_credit_note = "consolidated_credit_note" if doc.is_return else "consolidated_invoice"
|
||||
pos_closing_entry = frappe.get_all(
|
||||
"POS Invoice Merge Log",
|
||||
filters={invoice_or_credit_note: doc.name},
|
||||
pluck="pos_closing_entry",
|
||||
)
|
||||
if pos_closing_entry and pos_closing_entry[0]:
|
||||
msg = _("To cancel a {} you need to cancel the POS Closing Entry {}.").format(
|
||||
frappe.bold(_("Consolidated Sales Invoice")),
|
||||
get_link_to_form("POS Closing Entry", pos_closing_entry[0]),
|
||||
)
|
||||
frappe.throw(msg, title=_("Not Allowed"))
|
||||
|
||||
def check_if_created_using_pos_and_pos_closing_entry_generated(self) -> None:
|
||||
doc = self.doc
|
||||
if doc.doctype == "Sales Invoice" and doc.is_created_using_pos and doc.pos_closing_entry:
|
||||
pos_closing_entry_docstatus = frappe.db.get_value(
|
||||
"POS Closing Entry", doc.pos_closing_entry, "docstatus"
|
||||
)
|
||||
if pos_closing_entry_docstatus == 1:
|
||||
frappe.throw(
|
||||
msg=_(
|
||||
"To cancel this Sales Invoice you need to cancel the POS Closing Entry {0}."
|
||||
).format(get_link_to_form("POS Closing Entry", doc.pos_closing_entry)),
|
||||
title=_("Not Allowed"),
|
||||
)
|
||||
|
||||
def cancel_pos_invoice_credit_note_generated_during_sales_invoice_mode(self) -> None:
|
||||
pos_invoices = frappe.get_all(
|
||||
"POS Invoice", filters={"consolidated_invoice": self.doc.name}, pluck="name"
|
||||
)
|
||||
for pos_invoice in pos_invoices:
|
||||
frappe.get_doc("POS Invoice", pos_invoice).cancel()
|
||||
|
||||
def clear_unallocated_mode_of_payments(self) -> None:
|
||||
doc = self.doc
|
||||
doc.set("payments", doc.get("payments", {"amount": ["not in", [0, None, ""]]}))
|
||||
frappe.db.sql(
|
||||
"""delete from `tabSales Invoice Payment` where parent = %s and amount = 0""",
|
||||
doc.name,
|
||||
)
|
||||
|
||||
def allow_write_off_only_on_pos(self) -> None:
|
||||
if not self.doc.is_pos and self.doc.write_off_account:
|
||||
self.doc.write_off_account = None
|
||||
|
||||
def verify_payment_amount_is_positive(self) -> None:
|
||||
for entry in self.doc.payments:
|
||||
if entry.amount < 0:
|
||||
frappe.throw(_("Row #{0} (Payment Table): Amount must be positive").format(entry.idx))
|
||||
|
||||
def verify_payment_amount_is_negative(self) -> None:
|
||||
for entry in self.doc.payments:
|
||||
if entry.amount > 0:
|
||||
frappe.throw(_("Row #{0} (Payment Table): Amount must be negative").format(entry.idx))
|
||||
|
||||
def get_warehouse(self) -> str | None:
|
||||
doc = self.doc
|
||||
user_pos_profile = frappe.db.sql(
|
||||
"""select name, warehouse from `tabPOS Profile`
|
||||
where ifnull(user,'') = %s and company = %s""",
|
||||
(frappe.session["user"], doc.company),
|
||||
)
|
||||
warehouse = user_pos_profile[0][1] if user_pos_profile else None
|
||||
|
||||
if not warehouse:
|
||||
global_pos_profile = frappe.db.sql(
|
||||
"""select name, warehouse from `tabPOS Profile`
|
||||
where (user is null or user = '') and company = %s""",
|
||||
doc.company,
|
||||
)
|
||||
if global_pos_profile:
|
||||
warehouse = global_pos_profile[0][1]
|
||||
elif not user_pos_profile:
|
||||
msgprint(_("POS Profile required to make POS Entry"), raise_exception=True)
|
||||
|
||||
return warehouse
|
||||
|
||||
|
||||
def get_bank_cash_account(mode_of_payment: str, company: str) -> dict:
|
||||
account = frappe.db.get_value(
|
||||
"Mode of Payment Account",
|
||||
{"parent": mode_of_payment, "company": company},
|
||||
"default_account",
|
||||
)
|
||||
if not account:
|
||||
frappe.throw(
|
||||
_("Please set default Cash or Bank account in Mode of Payment {0}").format(
|
||||
get_link_to_form("Mode of Payment", mode_of_payment)
|
||||
),
|
||||
title=_("Missing Account"),
|
||||
)
|
||||
return {"account": account}
|
||||
|
||||
|
||||
def update_multi_mode_option(doc, pos_profile) -> None:
|
||||
def append_payment(payment_mode):
|
||||
payment = doc.append("payments", {})
|
||||
payment.default = payment_mode.default
|
||||
payment.mode_of_payment = payment_mode.mop
|
||||
payment.account = payment_mode.default_account
|
||||
payment.type = payment_mode.type
|
||||
|
||||
mop_refetched = bool(doc.payments) and not doc.is_created_using_pos
|
||||
|
||||
doc.set("payments", [])
|
||||
invalid_modes = []
|
||||
mode_of_payments = [d.mode_of_payment for d in pos_profile.get("payments")]
|
||||
mode_of_payments_info = get_mode_of_payments_info(mode_of_payments, doc.company)
|
||||
|
||||
for row in pos_profile.get("payments"):
|
||||
payment_mode = mode_of_payments_info.get(row.mode_of_payment)
|
||||
if not payment_mode:
|
||||
invalid_modes.append(get_link_to_form("Mode of Payment", row.mode_of_payment))
|
||||
continue
|
||||
|
||||
payment_mode.default = row.default
|
||||
append_payment(payment_mode)
|
||||
|
||||
if invalid_modes:
|
||||
if invalid_modes == 1:
|
||||
msg = _("Please set default Cash or Bank account in Mode of Payment {}")
|
||||
else:
|
||||
msg = _("Please set default Cash or Bank account in Mode of Payments {}")
|
||||
frappe.throw(msg.format(", ".join(invalid_modes)), title=_("Missing Account"))
|
||||
|
||||
if mop_refetched:
|
||||
frappe.toast(
|
||||
_("Payment methods refreshed. Please review before proceeding."),
|
||||
indicator="orange",
|
||||
)
|
||||
|
||||
|
||||
def get_all_mode_of_payments(doc) -> list:
|
||||
return frappe.db.sql(
|
||||
"""
|
||||
select mpa.default_account, mpa.parent, mp.type as type
|
||||
from `tabMode of Payment Account` mpa,`tabMode of Payment` mp
|
||||
where mpa.parent = mp.name and mpa.company = %(company)s and mp.enabled = 1""",
|
||||
{"company": doc.company},
|
||||
as_dict=1,
|
||||
)
|
||||
|
||||
|
||||
def get_mode_of_payments_info(mode_of_payments: list, company: str) -> dict:
|
||||
data = frappe.db.sql(
|
||||
"""
|
||||
select
|
||||
mpa.default_account, mpa.parent as mop, mp.type as type
|
||||
from
|
||||
`tabMode of Payment Account` mpa,`tabMode of Payment` mp
|
||||
where
|
||||
mpa.parent = mp.name and
|
||||
mpa.company = %s and
|
||||
mp.enabled = 1 and
|
||||
mp.name in %s
|
||||
group by
|
||||
mp.name
|
||||
""",
|
||||
(company, mode_of_payments),
|
||||
as_dict=1,
|
||||
)
|
||||
return {row.get("mop"): row for row in data}
|
||||
|
||||
|
||||
def get_mode_of_payment_info(mode_of_payment: str, company: str) -> list:
|
||||
return frappe.db.sql(
|
||||
"""
|
||||
select mpa.default_account, mpa.parent, mp.type as type
|
||||
from `tabMode of Payment Account` mpa,`tabMode of Payment` mp
|
||||
where mpa.parent = mp.name and mpa.company = %s and mp.enabled = 1 and mp.name = %s""",
|
||||
(company, mode_of_payment),
|
||||
as_dict=1,
|
||||
)
|
||||
130
erpnext/accounts/doctype/sales_invoice/services/status.py
Normal file
130
erpnext/accounts/doctype/sales_invoice/services/status.py
Normal file
@@ -0,0 +1,130 @@
|
||||
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# License: GNU General Public License v3. See license.txt
|
||||
|
||||
"""Status computation and display helpers for Sales Invoice."""
|
||||
|
||||
import frappe
|
||||
from frappe import _
|
||||
from frappe.utils import cint, flt, getdate, nowdate
|
||||
|
||||
|
||||
class StatusService:
|
||||
def __init__(self, doc):
|
||||
self.doc = doc
|
||||
|
||||
def set_status(
|
||||
self, update: bool = False, status: str | None = None, update_modified: bool = True
|
||||
) -> None:
|
||||
doc = self.doc
|
||||
if doc.is_new():
|
||||
if doc.get("amended_from"):
|
||||
doc.status = "Draft"
|
||||
return
|
||||
|
||||
outstanding_amount = flt(doc.outstanding_amount, doc.precision("outstanding_amount"))
|
||||
total = get_total_in_party_account_currency(doc)
|
||||
|
||||
if not status:
|
||||
if doc.docstatus == 2:
|
||||
status = "Cancelled"
|
||||
elif doc.docstatus == 1:
|
||||
if doc.is_internal_transfer():
|
||||
doc.status = "Internal Transfer"
|
||||
elif is_overdue(doc, total):
|
||||
doc.status = "Overdue"
|
||||
elif 0 < outstanding_amount < total:
|
||||
doc.status = "Partly Paid"
|
||||
elif outstanding_amount > 0 and getdate(doc.due_date) >= getdate():
|
||||
doc.status = "Unpaid"
|
||||
elif doc.is_return == 0 and frappe.db.get_value(
|
||||
"Sales Invoice", {"is_return": 1, "return_against": doc.name, "docstatus": 1}
|
||||
):
|
||||
doc.status = "Credit Note Issued"
|
||||
elif doc.is_return == 1:
|
||||
doc.status = "Return"
|
||||
elif outstanding_amount <= 0:
|
||||
doc.status = "Paid"
|
||||
else:
|
||||
doc.status = "Submitted"
|
||||
|
||||
if (
|
||||
doc.status in ("Unpaid", "Partly Paid", "Overdue")
|
||||
and doc.is_discounted
|
||||
and get_discounting_status(doc.name) == "Disbursed"
|
||||
):
|
||||
doc.status += " and Discounted"
|
||||
|
||||
else:
|
||||
doc.status = "Draft"
|
||||
|
||||
if update:
|
||||
doc.db_set("status", doc.status, update_modified=update_modified)
|
||||
|
||||
def set_indicator(self) -> None:
|
||||
doc = self.doc
|
||||
if doc.outstanding_amount < 0:
|
||||
doc.indicator_title = _("Credit Note Issued")
|
||||
doc.indicator_color = "gray"
|
||||
elif doc.outstanding_amount > 0 and getdate(doc.due_date) >= getdate(nowdate()):
|
||||
doc.indicator_color = "orange"
|
||||
doc.indicator_title = _("Unpaid")
|
||||
elif doc.outstanding_amount > 0 and getdate(doc.due_date) < getdate(nowdate()):
|
||||
doc.indicator_color = "red"
|
||||
doc.indicator_title = _("Overdue")
|
||||
elif cint(doc.is_return) == 1:
|
||||
doc.indicator_title = _("Return")
|
||||
doc.indicator_color = "gray"
|
||||
else:
|
||||
doc.indicator_color = "green"
|
||||
doc.indicator_title = _("Paid")
|
||||
|
||||
|
||||
def get_total_in_party_account_currency(doc) -> float:
|
||||
total_fieldname = "grand_total" if doc.disable_rounded_total else "rounded_total"
|
||||
if doc.party_account_currency != doc.currency:
|
||||
total_fieldname = "base_" + total_fieldname
|
||||
return flt(doc.get(total_fieldname), doc.precision(total_fieldname))
|
||||
|
||||
|
||||
def is_overdue(doc, total: float) -> bool | None:
|
||||
outstanding_amount = flt(doc.outstanding_amount, doc.precision("outstanding_amount"))
|
||||
if outstanding_amount <= 0:
|
||||
return
|
||||
|
||||
today = getdate()
|
||||
if doc.get("is_pos") or not doc.get("payment_schedule"):
|
||||
return getdate(doc.due_date) < today
|
||||
|
||||
payment_amount_field = (
|
||||
"base_payment_amount" if doc.party_account_currency != doc.currency else "payment_amount"
|
||||
)
|
||||
payable_amount = flt(
|
||||
sum(
|
||||
payment.get(payment_amount_field)
|
||||
for payment in doc.payment_schedule
|
||||
if getdate(payment.due_date) < today
|
||||
),
|
||||
doc.precision("outstanding_amount"),
|
||||
)
|
||||
return flt(total - outstanding_amount, doc.precision("outstanding_amount")) < payable_amount
|
||||
|
||||
|
||||
def get_discounting_status(sales_invoice: str) -> str | None:
|
||||
status = None
|
||||
invoice_discounting_list = frappe.db.sql(
|
||||
"""
|
||||
select status
|
||||
from `tabInvoice Discounting` id, `tabDiscounted Invoice` d
|
||||
where
|
||||
id.name = d.parent
|
||||
and d.sales_invoice=%s
|
||||
and id.docstatus=1
|
||||
and status in ('Disbursed', 'Settled')
|
||||
""",
|
||||
sales_invoice,
|
||||
)
|
||||
for d in invoice_discounting_list:
|
||||
status = d[0]
|
||||
if status == "Disbursed":
|
||||
break
|
||||
return status
|
||||
@@ -0,0 +1,121 @@
|
||||
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# License: GNU General Public License v3. See license.txt
|
||||
|
||||
"""Timesheet billing helpers for Sales Invoice."""
|
||||
|
||||
import frappe
|
||||
from frappe import _
|
||||
from frappe.utils import flt
|
||||
|
||||
from erpnext.projects.doctype.timesheet.timesheet import get_projectwise_timesheet_data
|
||||
|
||||
|
||||
class TimesheetBillingService:
|
||||
def __init__(self, doc):
|
||||
self.doc = doc
|
||||
|
||||
def validate_time_sheets_are_submitted(self) -> None:
|
||||
for data in self.doc.timesheets:
|
||||
if data.time_sheet and data.timesheet_detail:
|
||||
if sales_invoice := frappe.db.get_value(
|
||||
"Timesheet Detail", data.timesheet_detail, "sales_invoice"
|
||||
):
|
||||
frappe.throw(
|
||||
_("Row {0}: Sales Invoice {1} is already created for {2}").format(
|
||||
data.idx, frappe.bold(sales_invoice), frappe.bold(data.time_sheet)
|
||||
)
|
||||
)
|
||||
|
||||
if data.time_sheet:
|
||||
status = frappe.db.get_value("Timesheet", data.time_sheet, "status")
|
||||
if status not in ["Submitted", "Payslip", "Partially Billed"]:
|
||||
frappe.throw(
|
||||
_("Timesheet {0} cannot be invoiced in its current state").format(data.time_sheet)
|
||||
)
|
||||
|
||||
def update_time_sheet(self, sales_invoice: str | None) -> None:
|
||||
for d in self.doc.timesheets:
|
||||
if d.time_sheet:
|
||||
timesheet = frappe.get_doc("Timesheet", d.time_sheet)
|
||||
self._update_time_sheet_detail(timesheet, d, sales_invoice)
|
||||
timesheet.calculate_total_amounts()
|
||||
timesheet.calculate_percentage_billed()
|
||||
timesheet.flags.ignore_validate_update_after_submit = True
|
||||
timesheet.set_status()
|
||||
timesheet.db_update_all()
|
||||
|
||||
def unlink_sales_invoice_from_timesheets(self) -> None:
|
||||
for row in self.doc.timesheets:
|
||||
timesheet = frappe.get_doc("Timesheet", row.time_sheet)
|
||||
timesheet.unlink_sales_invoice(self.doc.name)
|
||||
timesheet.flags.ignore_validate_update_after_submit = True
|
||||
timesheet.db_update_all()
|
||||
|
||||
def set_billing_hours_and_amount(self) -> None:
|
||||
doc = self.doc
|
||||
if doc.project:
|
||||
return
|
||||
|
||||
for timesheet in doc.timesheets:
|
||||
ts_doc = frappe.get_doc("Timesheet", timesheet.time_sheet)
|
||||
if not timesheet.billing_hours and ts_doc.total_billable_hours:
|
||||
timesheet.billing_hours = ts_doc.total_billable_hours
|
||||
if not timesheet.billing_amount and ts_doc.total_billable_amount:
|
||||
timesheet.billing_amount = ts_doc.total_billable_amount
|
||||
|
||||
def update_timesheet_billing_for_project(self) -> None:
|
||||
doc = self.doc
|
||||
if (
|
||||
not doc.is_return
|
||||
and not doc.timesheets
|
||||
and doc.project
|
||||
and frappe.db.get_single_value("Projects Settings", "fetch_timesheet_in_sales_invoice")
|
||||
):
|
||||
self.add_timesheet_data()
|
||||
else:
|
||||
self.calculate_billing_amount_for_timesheet()
|
||||
|
||||
def add_timesheet_data(self) -> None:
|
||||
doc = self.doc
|
||||
doc.set("timesheets", [])
|
||||
if doc.project:
|
||||
for data in get_projectwise_timesheet_data(doc.project):
|
||||
doc.append(
|
||||
"timesheets",
|
||||
{
|
||||
"time_sheet": data.time_sheet,
|
||||
"billing_hours": data.billing_hours,
|
||||
"billing_amount": data.billing_amount,
|
||||
"timesheet_detail": data.name,
|
||||
"activity_type": data.activity_type,
|
||||
"description": data.description,
|
||||
},
|
||||
)
|
||||
self.calculate_billing_amount_for_timesheet()
|
||||
|
||||
def calculate_billing_amount_for_timesheet(self) -> None:
|
||||
doc = self.doc
|
||||
doc.total_billing_amount = sum(flt(ts.billing_amount) for ts in doc.timesheets)
|
||||
doc.total_billing_hours = sum(flt(ts.billing_hours) for ts in doc.timesheets)
|
||||
|
||||
def _update_time_sheet_detail(self, timesheet, args, sales_invoice: str | None) -> None:
|
||||
doc = self.doc
|
||||
for data in timesheet.time_logs:
|
||||
if (
|
||||
(doc.project and args.timesheet_detail == data.name)
|
||||
or (not doc.project and not data.sales_invoice and args.timesheet_detail == data.name)
|
||||
or (
|
||||
not sales_invoice
|
||||
and data.sales_invoice == doc.name
|
||||
and args.timesheet_detail == data.name
|
||||
)
|
||||
or (
|
||||
doc.is_return
|
||||
and doc.return_against
|
||||
and data.sales_invoice
|
||||
and data.sales_invoice == doc.return_against
|
||||
and not sales_invoice
|
||||
and args.timesheet_detail == data.name
|
||||
)
|
||||
):
|
||||
data.sales_invoice = sales_invoice
|
||||
Reference in New Issue
Block a user