|
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419 |
- import traceback
-
- import frappe
- import taxjar
- from frappe import _
- from frappe.contacts.doctype.address.address import get_company_address
- from frappe.utils import cint, flt
-
- from erpnext import get_default_company, get_region
-
- SUPPORTED_COUNTRY_CODES = [
- "AT",
- "AU",
- "BE",
- "BG",
- "CA",
- "CY",
- "CZ",
- "DE",
- "DK",
- "EE",
- "ES",
- "FI",
- "FR",
- "GB",
- "GR",
- "HR",
- "HU",
- "IE",
- "IT",
- "LT",
- "LU",
- "LV",
- "MT",
- "NL",
- "PL",
- "PT",
- "RO",
- "SE",
- "SI",
- "SK",
- "US",
- ]
- SUPPORTED_STATE_CODES = [
- "AL",
- "AK",
- "AZ",
- "AR",
- "CA",
- "CO",
- "CT",
- "DE",
- "DC",
- "FL",
- "GA",
- "HI",
- "ID",
- "IL",
- "IN",
- "IA",
- "KS",
- "KY",
- "LA",
- "ME",
- "MD",
- "MA",
- "MI",
- "MN",
- "MS",
- "MO",
- "MT",
- "NE",
- "NV",
- "NH",
- "NJ",
- "NM",
- "NY",
- "NC",
- "ND",
- "OH",
- "OK",
- "OR",
- "PA",
- "RI",
- "SC",
- "SD",
- "TN",
- "TX",
- "UT",
- "VT",
- "VA",
- "WA",
- "WV",
- "WI",
- "WY",
- ]
-
-
- def get_client():
- taxjar_settings = frappe.get_single("TaxJar Settings")
-
- if not taxjar_settings.is_sandbox:
- api_key = taxjar_settings.api_key and taxjar_settings.get_password("api_key")
- api_url = taxjar.DEFAULT_API_URL
- else:
- api_key = taxjar_settings.sandbox_api_key and taxjar_settings.get_password("sandbox_api_key")
- api_url = taxjar.SANDBOX_API_URL
-
- if api_key and api_url:
- client = taxjar.Client(api_key=api_key, api_url=api_url)
- client.set_api_config("headers", {"x-api-version": "2022-01-24"})
- return client
-
-
- def create_transaction(doc, method):
- TAXJAR_CREATE_TRANSACTIONS = frappe.db.get_single_value(
- "TaxJar Settings", "taxjar_create_transactions"
- )
-
- """Create an order transaction in TaxJar"""
-
- if not TAXJAR_CREATE_TRANSACTIONS:
- return
-
- client = get_client()
-
- if not client:
- return
-
- TAX_ACCOUNT_HEAD = frappe.db.get_single_value("TaxJar Settings", "tax_account_head")
- sales_tax = sum([tax.tax_amount for tax in doc.taxes if tax.account_head == TAX_ACCOUNT_HEAD])
-
- if not sales_tax:
- return
-
- tax_dict = get_tax_data(doc)
-
- if not tax_dict:
- return
-
- tax_dict["transaction_id"] = doc.name
- tax_dict["transaction_date"] = frappe.utils.today()
- tax_dict["sales_tax"] = sales_tax
- tax_dict["amount"] = doc.total + tax_dict["shipping"]
-
- try:
- if doc.is_return:
- client.create_refund(tax_dict)
- else:
- client.create_order(tax_dict)
- except taxjar.exceptions.TaxJarResponseError as err:
- frappe.throw(_(sanitize_error_response(err)))
- except Exception as ex:
- print(traceback.format_exc(ex))
-
-
- def delete_transaction(doc, method):
- """Delete an existing TaxJar order transaction"""
- TAXJAR_CREATE_TRANSACTIONS = frappe.db.get_single_value(
- "TaxJar Settings", "taxjar_create_transactions"
- )
-
- if not TAXJAR_CREATE_TRANSACTIONS:
- return
-
- client = get_client()
-
- if not client:
- return
-
- client.delete_order(doc.name)
-
-
- def get_tax_data(doc):
- SHIP_ACCOUNT_HEAD = frappe.db.get_single_value("TaxJar Settings", "shipping_account_head")
-
- from_address = get_company_address_details(doc)
- from_shipping_state = from_address.get("state")
- from_country_code = frappe.db.get_value("Country", from_address.country, "code")
- from_country_code = from_country_code.upper()
-
- to_address = get_shipping_address_details(doc)
- to_shipping_state = to_address.get("state")
- to_country_code = frappe.db.get_value("Country", to_address.country, "code")
- to_country_code = to_country_code.upper()
-
- shipping = sum([tax.tax_amount for tax in doc.taxes if tax.account_head == SHIP_ACCOUNT_HEAD])
-
- line_items = [get_line_item_dict(item, doc.docstatus) for item in doc.items]
-
- if from_shipping_state not in SUPPORTED_STATE_CODES:
- from_shipping_state = get_state_code(from_address, "Company")
-
- if to_shipping_state not in SUPPORTED_STATE_CODES:
- to_shipping_state = get_state_code(to_address, "Shipping")
-
- tax_dict = {
- "from_country": from_country_code,
- "from_zip": from_address.pincode,
- "from_state": from_shipping_state,
- "from_city": from_address.city,
- "from_street": from_address.address_line1,
- "to_country": to_country_code,
- "to_zip": to_address.pincode,
- "to_city": to_address.city,
- "to_street": to_address.address_line1,
- "to_state": to_shipping_state,
- "shipping": shipping,
- "amount": doc.net_total,
- "plugin": "erpnext",
- "line_items": line_items,
- }
- return tax_dict
-
-
- def get_state_code(address, location):
- if address is not None:
- state_code = get_iso_3166_2_state_code(address)
- if state_code not in SUPPORTED_STATE_CODES:
- frappe.throw(_("Please enter a valid State in the {0} Address").format(location))
- else:
- frappe.throw(_("Please enter a valid State in the {0} Address").format(location))
-
- return state_code
-
-
- def get_line_item_dict(item, docstatus):
- tax_dict = dict(
- id=item.get("idx"),
- quantity=item.get("qty"),
- unit_price=item.get("rate"),
- product_tax_code=item.get("product_tax_category"),
- )
-
- if docstatus == 1:
- tax_dict.update({"sales_tax": item.get("tax_collectable")})
-
- return tax_dict
-
-
- def set_sales_tax(doc, method):
- TAX_ACCOUNT_HEAD = frappe.db.get_single_value("TaxJar Settings", "tax_account_head")
- TAXJAR_CALCULATE_TAX = frappe.db.get_single_value("TaxJar Settings", "taxjar_calculate_tax")
-
- if not TAXJAR_CALCULATE_TAX:
- return
-
- if get_region(doc.company) != "United States":
- return
-
- if not doc.items:
- return
-
- if check_sales_tax_exemption(doc):
- return
-
- tax_dict = get_tax_data(doc)
-
- if not tax_dict:
- # Remove existing tax rows if address is changed from a taxable state/country
- setattr(doc, "taxes", [tax for tax in doc.taxes if tax.account_head != TAX_ACCOUNT_HEAD])
- return
-
- # check if delivering within a nexus
- check_for_nexus(doc, tax_dict)
-
- tax_data = validate_tax_request(tax_dict)
- if tax_data is not None:
- if not tax_data.amount_to_collect:
- setattr(doc, "taxes", [tax for tax in doc.taxes if tax.account_head != TAX_ACCOUNT_HEAD])
- elif tax_data.amount_to_collect > 0:
- # Loop through tax rows for existing Sales Tax entry
- # If none are found, add a row with the tax amount
- for tax in doc.taxes:
- if tax.account_head == TAX_ACCOUNT_HEAD:
- tax.tax_amount = tax_data.amount_to_collect
-
- doc.run_method("calculate_taxes_and_totals")
- break
- else:
- doc.append(
- "taxes",
- {
- "charge_type": "Actual",
- "description": "Sales Tax",
- "account_head": TAX_ACCOUNT_HEAD,
- "tax_amount": tax_data.amount_to_collect,
- },
- )
- # Assigning values to tax_collectable and taxable_amount fields in sales item table
- for item in tax_data.breakdown.line_items:
- doc.get("items")[cint(item.id) - 1].tax_collectable = item.tax_collectable
- doc.get("items")[cint(item.id) - 1].taxable_amount = item.taxable_amount
-
- doc.run_method("calculate_taxes_and_totals")
-
-
- def check_for_nexus(doc, tax_dict):
- TAX_ACCOUNT_HEAD = frappe.db.get_single_value("TaxJar Settings", "tax_account_head")
- if not frappe.db.get_value("TaxJar Nexus", {"region_code": tax_dict["to_state"]}):
- for item in doc.get("items"):
- item.tax_collectable = flt(0)
- item.taxable_amount = flt(0)
-
- for tax in list(doc.taxes):
- if tax.account_head == TAX_ACCOUNT_HEAD:
- doc.taxes.remove(tax)
- return
-
-
- def check_sales_tax_exemption(doc):
- # if the party is exempt from sales tax, then set all tax account heads to zero
- TAX_ACCOUNT_HEAD = frappe.db.get_single_value("TaxJar Settings", "tax_account_head")
-
- sales_tax_exempted = (
- hasattr(doc, "exempt_from_sales_tax")
- and doc.exempt_from_sales_tax
- or frappe.db.has_column("Customer", "exempt_from_sales_tax")
- and frappe.db.get_value("Customer", doc.customer, "exempt_from_sales_tax")
- )
-
- if sales_tax_exempted:
- for tax in doc.taxes:
- if tax.account_head == TAX_ACCOUNT_HEAD:
- tax.tax_amount = 0
- break
- doc.run_method("calculate_taxes_and_totals")
- return True
- else:
- return False
-
-
- def validate_tax_request(tax_dict):
- """Return the sales tax that should be collected for a given order."""
-
- client = get_client()
-
- if not client:
- return
-
- try:
- tax_data = client.tax_for_order(tax_dict)
- except taxjar.exceptions.TaxJarResponseError as err:
- frappe.throw(_(sanitize_error_response(err)))
- else:
- return tax_data
-
-
- def get_company_address_details(doc):
- """Return default company address details"""
-
- company_address = get_company_address(get_default_company()).company_address
-
- if not company_address:
- frappe.throw(_("Please set a default company address"))
-
- company_address = frappe.get_doc("Address", company_address)
- return company_address
-
-
- def get_shipping_address_details(doc):
- """Return customer shipping address details"""
-
- if doc.shipping_address_name:
- shipping_address = frappe.get_doc("Address", doc.shipping_address_name)
- elif doc.customer_address:
- shipping_address = frappe.get_doc("Address", doc.customer_address)
- else:
- shipping_address = get_company_address_details(doc)
-
- return shipping_address
-
-
- def get_iso_3166_2_state_code(address):
- import pycountry
-
- country_code = frappe.db.get_value("Country", address.get("country"), "code")
-
- error_message = _(
- """{0} is not a valid state! Check for typos or enter the ISO code for your state."""
- ).format(address.get("state"))
- state = address.get("state").upper().strip()
-
- # The max length for ISO state codes is 3, excluding the country code
- if len(state) <= 3:
- # PyCountry returns state code as {country_code}-{state-code} (e.g. US-FL)
- address_state = (country_code + "-" + state).upper()
-
- states = pycountry.subdivisions.get(country_code=country_code.upper())
- states = [pystate.code for pystate in states]
-
- if address_state in states:
- return state
-
- frappe.throw(_(error_message))
- else:
- try:
- lookup_state = pycountry.subdivisions.lookup(state)
- except LookupError:
- frappe.throw(_(error_message))
- else:
- return lookup_state.code.split("-")[1]
-
-
- def sanitize_error_response(response):
- response = response.full_response.get("detail")
- response = response.replace("_", " ")
-
- sanitized_responses = {
- "to zip": "Zipcode",
- "to city": "City",
- "to state": "State",
- "to country": "Country",
- }
-
- for k, v in sanitized_responses.items():
- response = response.replace(k, v)
-
- return response
|