mirror of
https://github.com/bringout/oca-financial.git
synced 2026-04-27 13:42:00 +02:00
938 lines
34 KiB
Python
938 lines
34 KiB
Python
import requests
|
|
|
|
from odoo import _, fields, models
|
|
from odoo.exceptions import UserError
|
|
|
|
from odoo.addons.account_avatax_oca.models.avatax_rest_api import AvaTaxRESTService
|
|
from odoo.addons.queue_job.exception import FailedJobError
|
|
|
|
|
|
class AvalaraSalestax(models.Model):
|
|
_inherit = "avalara.salestax"
|
|
|
|
avatax_company_id = fields.Char(
|
|
"Company ID",
|
|
help="The company ID as defined in the Admin Console of AvaTax",
|
|
)
|
|
tax_item_export = fields.Boolean()
|
|
exemption_export = fields.Boolean()
|
|
exemption_rule_export = fields.Boolean()
|
|
use_commercial_entity = fields.Boolean(default=True)
|
|
|
|
def create_transaction(
|
|
self,
|
|
doc_date,
|
|
doc_code,
|
|
doc_type,
|
|
partner,
|
|
ship_from_address,
|
|
shipping_address,
|
|
lines,
|
|
user=None,
|
|
exemption_number=None,
|
|
exemption_code_name=None,
|
|
commit=False,
|
|
invoice_date=None,
|
|
reference_code=None,
|
|
location_code=None,
|
|
is_override=None,
|
|
currency_id=None,
|
|
ignore_error=None,
|
|
log_to_record=False,
|
|
):
|
|
if self.use_commercial_entity and partner.commercial_partner_id:
|
|
partner = partner.commercial_partner_id
|
|
return super().create_transaction(
|
|
doc_date,
|
|
doc_code,
|
|
doc_type,
|
|
partner,
|
|
ship_from_address,
|
|
shipping_address,
|
|
lines,
|
|
user=user,
|
|
exemption_number=exemption_number,
|
|
exemption_code_name=exemption_code_name,
|
|
commit=commit,
|
|
invoice_date=invoice_date,
|
|
reference_code=reference_code,
|
|
location_code=location_code,
|
|
is_override=is_override,
|
|
currency_id=currency_id,
|
|
ignore_error=ignore_error,
|
|
log_to_record=log_to_record,
|
|
)
|
|
|
|
def set_tax_item_info_to_product(self, record, product):
|
|
vals = {}
|
|
product_tax_codes = self.env["product.tax.code"].search([])
|
|
if product:
|
|
tax_code = product_tax_codes.filtered(lambda x: x.name == record["taxCode"])
|
|
if not tax_code:
|
|
tax_code = product_tax_codes.create(
|
|
{
|
|
"type": "product",
|
|
"name": record["taxCode"],
|
|
}
|
|
)
|
|
vals["tax_code_id"] = tax_code.id
|
|
vals["avatax_item_id"] = record["id"]
|
|
product.with_context(skip_job_creation=True).write(vals)
|
|
|
|
def import_exemption_activity_type(self):
|
|
self.ensure_one()
|
|
business_type_obj = self.env["res.partner.exemption.business.type"]
|
|
avatax_restpoint = AvaTaxRESTService(config=self)
|
|
r = avatax_restpoint.client.list_certificate_exempt_reasons()
|
|
result = r.json()
|
|
if "error" in result:
|
|
error = result["error"]
|
|
error_message = "Code: {}\nMessage: {}\nTarget: {}\nDetails;{}".format(
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
raise FailedJobError(error_message)
|
|
for record in result["value"]:
|
|
business_type = business_type_obj.search(
|
|
["|", ("name", "=", record["name"]), ("avatax_id", "=", record["id"])],
|
|
limit=1,
|
|
)
|
|
if not business_type:
|
|
business_type_obj.create(
|
|
{
|
|
"name": record["name"],
|
|
"avatax_id": record["id"],
|
|
}
|
|
)
|
|
|
|
def import_exemption_country_state_code(self):
|
|
self.ensure_one()
|
|
state_obj = self.env["res.country.state"]
|
|
avatax_restpoint = AvaTaxRESTService(config=self)
|
|
r = avatax_restpoint.client.list_jurisdictions()
|
|
result = r.json()
|
|
if "error" in result:
|
|
error = result["error"]
|
|
error_message = "Code: {}\nMessage: {}\nTarget: {}\nDetails;{}".format(
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
raise FailedJobError(error_message)
|
|
for record in result["value"]:
|
|
if record["type"] != "State":
|
|
continue
|
|
state = state_obj.search(
|
|
[
|
|
("code", "=", record["region"]),
|
|
("country_id.code", "=", record["country"]),
|
|
("avatax_code", "=", False),
|
|
],
|
|
limit=1,
|
|
)
|
|
if state:
|
|
state.write(
|
|
{
|
|
"avatax_code": record["code"],
|
|
"avatax_name": record["name"],
|
|
}
|
|
)
|
|
|
|
r2 = avatax_restpoint.client.list_nexus_by_company(self.avatax_company_id)
|
|
result2 = r2.json()
|
|
if "error" in result2:
|
|
error = result2["error"]
|
|
error_message = "Code: {}\nMessage: {}\nTarget: {}\nDetails;{}".format(
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
raise FailedJobError(error_message)
|
|
for record in result2["value"]:
|
|
if record["jurisdictionTypeId"] != "State":
|
|
continue
|
|
state = state_obj.search(
|
|
[
|
|
("code", "=", record["region"]),
|
|
("country_id.code", "=", record["country"]),
|
|
],
|
|
limit=1,
|
|
)
|
|
if state:
|
|
state.write(
|
|
{
|
|
"avatax_nexus": True,
|
|
}
|
|
)
|
|
|
|
exemption_rule_obj = self.env["exemption.code.rule"]
|
|
states = state_obj.search([("avatax_nexus", "=", True)])
|
|
entity_use_codes = self.env["exemption.code"].search([])
|
|
for state in states:
|
|
for use_code in entity_use_codes.filtered(lambda x: x.flag):
|
|
exemption_rule = exemption_rule_obj.search(
|
|
[
|
|
("exemption_code_id", "=", use_code.id),
|
|
("state_id", "=", state.id),
|
|
("taxable", "=", True),
|
|
],
|
|
limit=1,
|
|
)
|
|
if exemption_rule:
|
|
continue
|
|
else:
|
|
exemption_rule_obj.create(
|
|
{
|
|
"exemption_code_id": use_code.id,
|
|
"state_id": state.id,
|
|
"taxable": True,
|
|
"state": "draft",
|
|
}
|
|
)
|
|
|
|
def import_tax_items(self):
|
|
self.ensure_one()
|
|
products = self.env["product.product"].search(
|
|
[("default_code", "!=", False), ("avatax_item_id", "=", False)]
|
|
)
|
|
|
|
avatax_restpoint = AvaTaxRESTService(config=self)
|
|
client = avatax_restpoint.client
|
|
|
|
result_vals = []
|
|
main_url = url = "{}/api/v2/companies/{}/items".format(
|
|
client.base_url, self.avatax_company_id
|
|
)
|
|
count = 0
|
|
while True:
|
|
r = requests.get(
|
|
url,
|
|
auth=client.auth,
|
|
headers=client.client_header,
|
|
timeout=client.timeout_limit if client.timeout_limit else 1200,
|
|
)
|
|
result = r.json()
|
|
count += 1000
|
|
if "error" in result:
|
|
error = result["error"]
|
|
error_message = "Code: {}\nMessage: {}\nTarget: {}\nDetails;{}".format(
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
raise FailedJobError(error_message)
|
|
result_vals += result["value"]
|
|
if result["@recordsetCount"] <= count:
|
|
break
|
|
else:
|
|
url = main_url + "?%24skip=" + str(count)
|
|
|
|
for product in products:
|
|
for record in result_vals:
|
|
if product.default_code == record["itemCode"]:
|
|
self.set_tax_item_info_to_product(record, product)
|
|
break
|
|
|
|
def export_new_tax_items(self):
|
|
if not self.ids:
|
|
self = self.search([("tax_item_export", "=", True)], limit=1)
|
|
if not self.tax_item_export:
|
|
return
|
|
products = self.env["product.product"].search(
|
|
[
|
|
("default_code", "!=", False),
|
|
("avatax_item_id", "=", False),
|
|
"|",
|
|
("tax_code_id", "!=", False),
|
|
("categ_id.tax_code_id", "!=", False),
|
|
],
|
|
)
|
|
|
|
for product in products:
|
|
self.with_delay(
|
|
description="Export Tax Item %s" % (product.display_name)
|
|
)._export_tax_item(product)
|
|
|
|
def export_new_exemption_rules(self, rules=None):
|
|
if not self.ids:
|
|
self = self.search([("exemption_rule_export", "=", True)], limit=1)
|
|
if not self.exemption_rule_export:
|
|
return
|
|
if not rules:
|
|
rules = self.env["exemption.code.rule"].search(
|
|
[("avatax_id", "=", False), ("state", "=", "progress")],
|
|
)
|
|
|
|
queue_job_sudo = self.env["queue.job"].sudo()
|
|
for rule in rules:
|
|
job = queue_job_sudo.search(
|
|
[
|
|
("method_name", "=", "_export_base_rule_based_on_type"),
|
|
("state", "!=", "done"),
|
|
("args", "ilike", "%[" + str(rule.id) + "]%"),
|
|
],
|
|
limit=1,
|
|
)
|
|
if not job:
|
|
self.with_delay(
|
|
priority=5,
|
|
max_retries=2,
|
|
description="Export Rule %s" % (rule.name),
|
|
)._export_base_rule_based_on_type(rule)
|
|
|
|
def map_rule_vals_to_fields(self, json_data):
|
|
avatax_rule_data = {
|
|
"avatax_id": json_data["id"],
|
|
"avatax_rate": json_data.get("value", ""),
|
|
"exemption_code_id": json_data.get("customerUsageType", ""),
|
|
"is_all_juris": json_data.get("isAllJuris", False),
|
|
"state": "done",
|
|
}
|
|
# Search for tax code and exemption code in Odoo models
|
|
tax_code_id = (
|
|
self.env["product.tax.code"].search(
|
|
[("name", "=", json_data.get("taxCode", "") or "")], limit=1
|
|
)
|
|
or None
|
|
)
|
|
if tax_code_id:
|
|
avatax_rule_data["avatax_tax_code"] = tax_code_id.id
|
|
exemption_code_id = (
|
|
self.env["exemption.code"].search(
|
|
[("code", "=", json_data.get("customerUsageType", "") or "")], limit=1
|
|
)
|
|
or None
|
|
)
|
|
if exemption_code_id:
|
|
avatax_rule_data["exemption_code_id"] = exemption_code_id.id
|
|
# Search for country and state in Odoo models
|
|
state_id = (
|
|
self.env["res.country.state"].search(
|
|
[
|
|
("country_id.code", "=", json_data.get("country", "") or ""),
|
|
("code", "=", json_data.get("region", "") or ""),
|
|
],
|
|
limit=1,
|
|
)
|
|
or None
|
|
)
|
|
if state_id:
|
|
avatax_rule_data["state_id"] = state_id.id
|
|
avatax_rule_data["taxable"] = state_id.avatax_nexus
|
|
return avatax_rule_data
|
|
|
|
def import_tax_rules(self):
|
|
avatax_custom_rule_model = self.env["exemption.code.rule"]
|
|
avatax_restpoint = AvaTaxRESTService(config=self)
|
|
rules = avatax_custom_rule_model.search(
|
|
[("avatax_id", "!=", False), ("state", "=", "done")],
|
|
)
|
|
include_option = None
|
|
if rules:
|
|
filter_rules = "id not in ("
|
|
filter_rules += ", ".join(map(str, rules.mapped("avatax_id")))
|
|
filter_rules += ")"
|
|
include_option = "$filter=" + filter_rules
|
|
r = avatax_restpoint.client.list_tax_rules(
|
|
self.avatax_company_id, include_option
|
|
)
|
|
result = r.json()
|
|
if "error" in result:
|
|
error = result["error"]
|
|
error_message = "Code: {}\nMessage: {}\nTarget: {}\nDetails;{}".format(
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
raise UserError(error_message)
|
|
mapped_data = []
|
|
for rule_vals in result["value"]:
|
|
if rule_vals.get("customerUsageType", "") and rule_vals.get("region", ""):
|
|
mapped_data.append(self.map_rule_vals_to_fields(rule_vals))
|
|
avatax_custom_rule_model.create(mapped_data)
|
|
|
|
def download_exemptions(self):
|
|
if not self.ids:
|
|
self = self.search([("exemption_export", "=", True)], limit=1)
|
|
if not self.exemption_export:
|
|
raise UserError(
|
|
_("Avatax Exemption export is disabled in Avatax configuration")
|
|
)
|
|
|
|
avatax_restpoint = AvaTaxRESTService(config=self)
|
|
count = 0
|
|
result_vals = []
|
|
include_option = None
|
|
while True:
|
|
r = avatax_restpoint.client.query_certificates(
|
|
self.avatax_company_id, include_option
|
|
)
|
|
result = r.json()
|
|
count += 100
|
|
if "error" in result:
|
|
error = result["error"]
|
|
error_message = "Code: {}\nMessage: {}\nTarget: {}\nDetails;{}".format(
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
raise UserError(error_message)
|
|
result_vals += result["value"]
|
|
if result["@recordsetCount"] <= count:
|
|
break
|
|
else:
|
|
include_option = "$skip=" + str(count)
|
|
|
|
exemptions = (
|
|
self.env["res.partner.exemption.line"]
|
|
.sudo()
|
|
.search([("avatax_id", "!=", False)])
|
|
)
|
|
for exemption in result_vals:
|
|
avatax_id = exemption["id"]
|
|
if avatax_id not in exemptions.mapped("avatax_id"):
|
|
self.with_delay(
|
|
description="Download Exemption: %s" % (avatax_id)
|
|
)._search_create_exemption_line(avatax_id)
|
|
|
|
def _export_base_rule_based_on_type(self, rule):
|
|
error_message = False
|
|
if not rule.state_id.avatax_code:
|
|
raise FailedJobError("Avatax code for State not setup")
|
|
if not rule.exemption_code_id.flag:
|
|
raise FailedJobError("Taxed by Default is disabled in Exemption Code")
|
|
avatax_restpoint = AvaTaxRESTService(config=self)
|
|
|
|
avatax_value = 0
|
|
rule_type = "ExemptEntityRule"
|
|
if rule.taxable:
|
|
avatax_value = 1
|
|
elif rule.avatax_rate == 100.0:
|
|
avatax_value = 1
|
|
elif rule.avatax_rate:
|
|
rule_type = "RateOverrideRule"
|
|
avatax_value = rule.avatax_rate / 100
|
|
tax_rule_info = {
|
|
"companyId": self.avatax_company_id,
|
|
"taxCode": rule.avatax_tax_code.name or None,
|
|
"taxTypeId": "BothSalesAndUseTax",
|
|
"taxRuleTypeId": rule_type,
|
|
"jurisCode": rule.state_id.avatax_code,
|
|
"jurisName": rule.state_id.avatax_name,
|
|
"jurisTypeId": "STA",
|
|
"jurisdictionTypeId": "State",
|
|
"isAllJuris": rule.is_all_juris,
|
|
"value": avatax_value,
|
|
"cap": 0,
|
|
"threshold": 0,
|
|
"effectiveDate": fields.Datetime.to_string(fields.Date.today()),
|
|
"description": "%s - %s - %s"
|
|
% (rule.state_id.avatax_name, rule.exemption_code_id.code, rule.name),
|
|
"country": rule.state_id.country_id.code,
|
|
"region": rule.state_id.code,
|
|
"stateFIPS": rule.state_id.avatax_code,
|
|
"taxTypeGroup": "SalesAndUse",
|
|
"customerUsageType": rule.exemption_code_id.code,
|
|
"taxSubType": "ALL",
|
|
}
|
|
r = avatax_restpoint.client.create_tax_rules(
|
|
self.avatax_company_id, [tax_rule_info]
|
|
)
|
|
result = r.json()
|
|
if "error" in result:
|
|
error = result["error"]
|
|
error_message = (
|
|
"Rule: %s\nCode: %s\nMessage: %s\nTarget: %s\nDetails;%s"
|
|
% (
|
|
rule.name,
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
)
|
|
raise FailedJobError(error_message)
|
|
rule.write(
|
|
{
|
|
"avatax_id": result[0]["id"],
|
|
"state": "done",
|
|
}
|
|
)
|
|
|
|
return result
|
|
|
|
def _cancel_custom_rule(self, rule):
|
|
error_message = False
|
|
if not rule.avatax_id:
|
|
raise FailedJobError("Avatax Custom Rule ID not available")
|
|
avatax_restpoint = AvaTaxRESTService(config=self)
|
|
|
|
r = avatax_restpoint.client.delete_tax_rule(
|
|
self.avatax_company_id, rule.avatax_id
|
|
)
|
|
result = r.json()
|
|
if "error" in result:
|
|
error = result["error"]
|
|
error_message = (
|
|
"Rule: %s\nCode: %s\nMessage: %s\nTarget: %s\nDetails;%s"
|
|
% (
|
|
rule.name,
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
)
|
|
raise FailedJobError(error_message)
|
|
rule.write(
|
|
{
|
|
"avatax_id": False,
|
|
"state": "cancel",
|
|
}
|
|
)
|
|
|
|
return result
|
|
|
|
def _export_tax_item(self, product):
|
|
error_message = False
|
|
if not self.tax_item_export:
|
|
raise FailedJobError("Tax Item Export is disabled in Avatax configuration")
|
|
if product.avatax_item_id:
|
|
return "Product exported with Avatax ID: %s" % (product.avatax_item_id)
|
|
avatax_restpoint = AvaTaxRESTService(config=self)
|
|
|
|
item_info = {
|
|
"itemCode": product.default_code,
|
|
"taxCode": product.tax_code_id.name or product.categ_id.tax_code_id.name,
|
|
"description": product.name,
|
|
}
|
|
r = avatax_restpoint.client.create_items(self.avatax_company_id, item_info)
|
|
result = r.json()
|
|
if "error" in result:
|
|
error = result["error"]
|
|
error_message = (
|
|
"Product: %s\nCode: %s\nMessage: %s\nTarget: %s\nDetails;%s"
|
|
% (
|
|
product.display_name,
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
)
|
|
raise FailedJobError(error_message)
|
|
product.with_context(skip_job_creation=True).write(
|
|
{
|
|
"avatax_item_id": result[0]["id"],
|
|
}
|
|
)
|
|
|
|
return result
|
|
|
|
def _delete_tax_item(self, product):
|
|
error_message = False
|
|
if not self.tax_item_export:
|
|
raise FailedJobError("Tax Item Export is disabled in Avatax configuration")
|
|
if not product.avatax_item_id:
|
|
return "Avatax ID not available in Product: %s" % (product.display_name)
|
|
avatax_restpoint = AvaTaxRESTService(config=self)
|
|
|
|
r = avatax_restpoint.client.delete_item(
|
|
self.avatax_company_id, product.avatax_item_id
|
|
)
|
|
result = r.json()
|
|
if "error" in result:
|
|
error = result["error"]
|
|
error_message = (
|
|
"Product: %s\nCode: %s\nMessage: %s\nTarget: %s\nDetails;%s"
|
|
% (
|
|
product.display_name,
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
)
|
|
raise FailedJobError(error_message)
|
|
product.with_context(skip_job_creation=True).write(
|
|
{
|
|
"avatax_item_id": False,
|
|
}
|
|
)
|
|
|
|
return result
|
|
|
|
def _update_tax_item(self, tax_item_id, product):
|
|
if not self.tax_item_export:
|
|
raise FailedJobError("Tax Item Export is disabled in Avatax configuration")
|
|
error_message = False
|
|
avatax_restpoint = AvaTaxRESTService(config=self)
|
|
|
|
item_info = {
|
|
"itemCode": product.default_code,
|
|
"taxCode": product.tax_code_id.name or product.categ_id.tax_code_id.name,
|
|
"description": product.name,
|
|
}
|
|
r = avatax_restpoint.client.update_item(
|
|
self.avatax_company_id, tax_item_id, item_info
|
|
)
|
|
result = r.json()
|
|
if "error" in result:
|
|
error = result["error"]
|
|
error_message = (
|
|
"Product: %s\nCode: %s\nMessage: %s\nTarget: %s\nDetails;%s"
|
|
% (
|
|
product.display_name,
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
)
|
|
raise FailedJobError(error_message)
|
|
|
|
return result
|
|
|
|
def _export_avatax_customer(self, partner):
|
|
error_message = False
|
|
if not self.exemption_export:
|
|
raise FailedJobError(
|
|
"Avatax Exemption export is disabled in Avatax configuration"
|
|
)
|
|
|
|
avatax_restpoint = AvaTaxRESTService(config=self)
|
|
if partner.avatax_id:
|
|
return "Avatax Customer ID: %s" % (partner.avatax_id)
|
|
customer_info = [
|
|
{
|
|
"customerCode": partner.customer_code,
|
|
"alternateId": partner.id,
|
|
"name": partner.name,
|
|
"line1": partner.street,
|
|
"city": partner.city,
|
|
"postalCode": partner.zip,
|
|
"phoneNumber": partner.phone,
|
|
"emailAddress": partner.email,
|
|
"contactName": partner.name,
|
|
"country": partner.country_id.code,
|
|
"region": partner.state_id.code,
|
|
}
|
|
]
|
|
r = avatax_restpoint.client.create_customers(
|
|
self.avatax_company_id, customer_info
|
|
)
|
|
result = r.json()
|
|
if "error" in result:
|
|
error = result["error"]
|
|
error_message = (
|
|
"Partner: %s\nCode: %s\nMessage: %s\nTarget: %s\nDetails;%s"
|
|
% (
|
|
partner.display_name,
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
)
|
|
raise FailedJobError(error_message)
|
|
partner.with_context(skip_job_creation=True).write(
|
|
{
|
|
"avatax_id": result[0]["id"],
|
|
}
|
|
)
|
|
|
|
return result
|
|
|
|
def _export_avatax_exemption_line(self, exemption_line):
|
|
error_message = False
|
|
if not self.exemption_export:
|
|
raise FailedJobError(
|
|
"Avatax Exemption export is disabled in Avatax configuration"
|
|
)
|
|
|
|
avatax_restpoint = AvaTaxRESTService(config=self)
|
|
if exemption_line.avatax_id:
|
|
return "Avatax Customer ID: %s" % (exemption_line.avatax_id)
|
|
exemption_line_info = [
|
|
{
|
|
"signedDate": fields.Datetime.to_string(
|
|
exemption_line.exemption_id.effective_date
|
|
),
|
|
"expirationDate": fields.Datetime.to_string(
|
|
exemption_line.exemption_id.expiry_date
|
|
),
|
|
"filename": exemption_line.name,
|
|
"valid": True,
|
|
"exemptionNumber": exemption_line.exemption_number
|
|
if exemption_line.add_exemption_number
|
|
else exemption_line.exemption_id.exemption_number,
|
|
"exemptPercentage": 100.0,
|
|
"validatedExemptionReason": {
|
|
"name": exemption_line.exemption_id.business_type.name,
|
|
},
|
|
"exemptionReason": {
|
|
"name": exemption_line.exemption_id.business_type.name,
|
|
},
|
|
"exposureZone": {
|
|
"name": exemption_line.state_id.name,
|
|
},
|
|
"pages": [None],
|
|
}
|
|
]
|
|
r = avatax_restpoint.client.create_certificates(
|
|
self.avatax_company_id, exemption_line_info
|
|
)
|
|
result = r.json()
|
|
if "error" in result:
|
|
error = result["error"]
|
|
error_message = (
|
|
"Exemption: %s\nCode: %s\nMessage: %s\nTarget: %s\nDetails;%s"
|
|
% (
|
|
exemption_line.display_name,
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
)
|
|
raise FailedJobError(error_message)
|
|
exemption_line.write(
|
|
{
|
|
"avatax_id": result[0]["id"],
|
|
}
|
|
)
|
|
|
|
self.with_delay(
|
|
priority=6,
|
|
max_retries=2,
|
|
description="Link Customer %s with Exemption %s"
|
|
% (exemption_line.partner_id.display_name, exemption_line.name),
|
|
).link_certificates_to_customer(exemption_line)
|
|
|
|
return result
|
|
|
|
def link_certificates_to_customer(self, exemption_line):
|
|
error_message = False
|
|
if not self.exemption_export:
|
|
raise FailedJobError(
|
|
"Avatax Exemption export is disabled in Avatax configuration"
|
|
)
|
|
if not exemption_line.exemption_id.partner_id.avatax_id:
|
|
raise FailedJobError("Avatax Customer export has failed")
|
|
|
|
avatax_restpoint = AvaTaxRESTService(config=self)
|
|
r = avatax_restpoint.client.link_certificates_to_customer(
|
|
self.avatax_company_id,
|
|
exemption_line.exemption_id.partner_id.customer_code,
|
|
{"certificates": [exemption_line.avatax_id]},
|
|
)
|
|
result = r.json()
|
|
if "error" in result:
|
|
error = result["error"]
|
|
error_message = (
|
|
"Exemption: %s\nCode: %s\nMessage: %s\nTarget: %s\nDetails;%s"
|
|
% (
|
|
exemption_line.display_name,
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
)
|
|
raise FailedJobError(error_message)
|
|
exemption_line.write(
|
|
{
|
|
"linked_to_customer": True,
|
|
}
|
|
)
|
|
if all(
|
|
exemption_line.exemption_id.exemption_line_ids.mapped("linked_to_customer")
|
|
):
|
|
exemption_line.exemption_id.write(
|
|
{
|
|
"state": "done",
|
|
}
|
|
)
|
|
return result
|
|
|
|
def _update_avatax_exemption_line_status(self, exemption_line, exemption_status):
|
|
error_message = False
|
|
if not self.exemption_export:
|
|
raise FailedJobError(
|
|
"Avatax Exemption export is disabled in Avatax configuration"
|
|
)
|
|
|
|
avatax_restpoint = AvaTaxRESTService(config=self)
|
|
if not exemption_line.avatax_id:
|
|
raise FailedJobError("Avatax Exemption ID is not found")
|
|
|
|
r1 = avatax_restpoint.client.get_certificate(
|
|
self.avatax_company_id, exemption_line.avatax_id
|
|
)
|
|
result1 = r1.json()
|
|
if "error" in result1:
|
|
error = result1["error"]
|
|
error_message = (
|
|
"Exemption: %s\nCode: %s\nMessage: %s\nTarget: %s\nDetails;%s"
|
|
% (
|
|
exemption_line.display_name,
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
)
|
|
raise FailedJobError(error_message)
|
|
exemption_line_info = dict(result1)
|
|
exemption_line_info["valid"] = exemption_status
|
|
r2 = avatax_restpoint.client.update_certificate(
|
|
self.avatax_company_id, exemption_line.avatax_id, exemption_line_info
|
|
)
|
|
result2 = r2.json()
|
|
if "error" in result2:
|
|
error = result2["error"]
|
|
error_message = (
|
|
"Exemption: %s\nCode: %s\nMessage: %s\nTarget: %s\nDetails;%s"
|
|
% (
|
|
exemption_line.display_name,
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
)
|
|
raise FailedJobError(error_message)
|
|
exemption_line.write(
|
|
{
|
|
"avatax_status": exemption_status,
|
|
}
|
|
)
|
|
exemption_line.exemption_id.write(
|
|
{
|
|
"state": "done" if exemption_status else "cancel",
|
|
}
|
|
)
|
|
|
|
return result2
|
|
|
|
def _search_create_exemption_line(self, avatax_id):
|
|
exemption_sudo = self.env["res.partner.exemption"].sudo()
|
|
partner_sudo = self.env["res.partner"].sudo()
|
|
error_message = False
|
|
if not self.exemption_export:
|
|
raise FailedJobError(
|
|
"Avatax Exemption export is disabled in Avatax configuration"
|
|
)
|
|
|
|
avatax_restpoint = AvaTaxRESTService(config=self)
|
|
r = avatax_restpoint.client.get_certificate(
|
|
self.avatax_company_id, avatax_id, "$include=customers"
|
|
)
|
|
result = r.json()
|
|
if "error" in result:
|
|
error = result["error"]
|
|
error_message = "Code: {}\nMessage: {}\nTarget: {}\nDetails;{}".format(
|
|
error.get("code", False),
|
|
error.get("message", False),
|
|
error.get("target", False),
|
|
error.get("details", False),
|
|
)
|
|
raise FailedJobError(error_message)
|
|
if result.get("customers", []):
|
|
customer_info = result["customers"][0]
|
|
partner = partner_sudo.search(
|
|
[("avatax_id", "=", customer_info["id"])], limit=1
|
|
)
|
|
if partner:
|
|
partner.customer_code = customer_info["customerCode"]
|
|
if not partner:
|
|
partner = partner_sudo.search(
|
|
[("customer_code", "=", customer_info["customerCode"])], limit=1
|
|
)
|
|
if not partner:
|
|
partner = partner_sudo.search(
|
|
[("customer_code", "=", "%s:0" % (customer_info["customerCode"]))],
|
|
limit=1,
|
|
)
|
|
if not partner:
|
|
state = self.env["res.country.state"]
|
|
if "region" in customer_info:
|
|
state = (
|
|
self.env["res.country.state"]
|
|
.sudo()
|
|
.search(
|
|
[
|
|
("code", "=", customer_info["region"]),
|
|
("country_id.code", "=", customer_info["country"]),
|
|
],
|
|
limit=1,
|
|
)
|
|
)
|
|
partner_vals = {
|
|
"name": customer_info["name"],
|
|
"street": customer_info["line1"],
|
|
"city": customer_info["city"],
|
|
"zip": customer_info["postalCode"],
|
|
"state_id": state.id,
|
|
"country_id": state.country_id.id,
|
|
"email": customer_info.get("emailAddress", False),
|
|
"phone": customer_info.get("phoneNumber", False),
|
|
"avatax_id": customer_info["id"],
|
|
"customer_code": customer_info["customerCode"],
|
|
}
|
|
partner = partner_sudo.create(partner_vals)
|
|
|
|
# Check if exemption is already available in system
|
|
exemption_line = (
|
|
self.env["res.partner.exemption.line"]
|
|
.sudo()
|
|
.search([("avatax_id", "=", result["id"])], limit=1)
|
|
)
|
|
if exemption_line:
|
|
return "Exemption Already Downloaded\nSearch Response: %s" % (result)
|
|
exposure_zone_info = result["exposureZone"]
|
|
exposure_state = (
|
|
self.env["res.country.state"]
|
|
.sudo()
|
|
.search(
|
|
[
|
|
("code", "=", exposure_zone_info["region"]),
|
|
("country_id.code", "=", exposure_zone_info["country"]),
|
|
],
|
|
limit=1,
|
|
)
|
|
)
|
|
business_type = (
|
|
self.env["res.partner.exemption.business.type"]
|
|
.sudo()
|
|
.search([("avatax_id", "=", result["exemptionReason"]["id"])], limit=1)
|
|
)
|
|
exemption_line_vals = {
|
|
"state_id": exposure_state.id,
|
|
"avatax_id": result["id"],
|
|
"avatax_status": result["valid"],
|
|
"linked_to_customer": True,
|
|
}
|
|
exemption_sudo.create(
|
|
{
|
|
"partner_id": partner.id,
|
|
"business_type": business_type.id,
|
|
"exemption_code_id": business_type.exemption_code_id.id,
|
|
"state_ids": [(6, 0, [exposure_state.id])],
|
|
"exemption_number": result["exemptionNumber"],
|
|
"effective_date": result["signedDate"],
|
|
"expiry_date": result["expirationDate"],
|
|
"state": "done" if result["valid"] else "cancel",
|
|
"exemption_line_ids": [(0, 0, exemption_line_vals)],
|
|
}
|
|
)
|
|
return result
|
|
else:
|
|
raise FailedJobError("Exemption ID is not linked with a customer in Avatax")
|