style: format code with black
diff --git a/erpnext/erpnext_integrations/connectors/woocommerce_connection.py b/erpnext/erpnext_integrations/connectors/woocommerce_connection.py
index 9409485..46a4d3c 100644
--- a/erpnext/erpnext_integrations/connectors/woocommerce_connection.py
+++ b/erpnext/erpnext_integrations/connectors/woocommerce_connection.py
@@ -12,26 +12,30 @@
woocommerce_settings = frappe.get_doc("Woocommerce Settings")
sig = base64.b64encode(
hmac.new(
- woocommerce_settings.secret.encode('utf8'),
- frappe.request.data,
- hashlib.sha256
+ woocommerce_settings.secret.encode("utf8"), frappe.request.data, hashlib.sha256
).digest()
)
- if frappe.request.data and \
- not sig == frappe.get_request_header("X-Wc-Webhook-Signature", "").encode():
- frappe.throw(_("Unverified Webhook Data"))
+ if (
+ frappe.request.data
+ and not sig == frappe.get_request_header("X-Wc-Webhook-Signature", "").encode()
+ ):
+ frappe.throw(_("Unverified Webhook Data"))
frappe.set_user(woocommerce_settings.creation_user)
+
@frappe.whitelist(allow_guest=True)
def order(*args, **kwargs):
try:
_order(*args, **kwargs)
except Exception:
- error_message = frappe.get_traceback()+"\n\n Request Data: \n"+json.loads(frappe.request.data).__str__()
+ error_message = (
+ frappe.get_traceback() + "\n\n Request Data: \n" + json.loads(frappe.request.data).__str__()
+ )
frappe.log_error(error_message, "WooCommerce Error")
raise
+
def _order(*args, **kwargs):
woocommerce_settings = frappe.get_doc("Woocommerce Settings")
if frappe.flags.woocomm_test_order_data:
@@ -43,7 +47,7 @@
try:
order = json.loads(frappe.request.data)
except ValueError:
- #woocommerce returns 'webhook_id=value' for the first request which is not JSON
+ # woocommerce returns 'webhook_id=value' for the first request which is not JSON
order = frappe.request.data
event = frappe.get_request_header("X-Wc-Webhook-Event")
@@ -51,7 +55,7 @@
return "success"
if event == "created":
- sys_lang = frappe.get_single("System Settings").language or 'en'
+ sys_lang = frappe.get_single("System Settings").language or "en"
raw_billing_data = order.get("billing")
raw_shipping_data = order.get("shipping")
customer_name = raw_billing_data.get("first_name") + " " + raw_billing_data.get("last_name")
@@ -59,6 +63,7 @@
link_items(order.get("line_items"), woocommerce_settings, sys_lang)
create_sales_order(order, woocommerce_settings, customer_name, sys_lang)
+
def link_customer_and_address(raw_billing_data, raw_shipping_data, customer_name):
customer_woo_com_email = raw_billing_data.get("email")
customer_exists = frappe.get_value("Customer", {"woocommerce_email": customer_woo_com_email})
@@ -77,9 +82,14 @@
if customer_exists:
frappe.rename_doc("Customer", old_name, customer_name)
- for address_type in ("Billing", "Shipping",):
+ for address_type in (
+ "Billing",
+ "Shipping",
+ ):
try:
- address = frappe.get_doc("Address", {"woocommerce_email": customer_woo_com_email, "address_type": address_type})
+ address = frappe.get_doc(
+ "Address", {"woocommerce_email": customer_woo_com_email, "address_type": address_type}
+ )
rename_address(address, customer)
except (
frappe.DoesNotExistError,
@@ -92,6 +102,7 @@
create_address(raw_shipping_data, customer, "Shipping")
create_contact(raw_billing_data, customer)
+
def create_contact(data, customer):
email = data.get("email", None)
phone = data.get("phone", None)
@@ -111,14 +122,12 @@
if email:
contact.add_email(email, is_primary=1)
- contact.append("links", {
- "link_doctype": "Customer",
- "link_name": customer.name
- })
+ contact.append("links", {"link_doctype": "Customer", "link_name": customer.name})
contact.flags.ignore_mandatory = True
contact.save()
+
def create_address(raw_data, customer, address_type):
address = frappe.new_doc("Address")
@@ -132,14 +141,12 @@
address.pincode = raw_data.get("postcode")
address.phone = raw_data.get("phone")
address.email_id = customer.woocommerce_email
- address.append("links", {
- "link_doctype": "Customer",
- "link_name": customer.name
- })
+ address.append("links", {"link_doctype": "Customer", "link_name": customer.name})
address.flags.ignore_mandatory = True
address.save()
+
def rename_address(address, customer):
old_address_title = address.name
new_address_title = customer.name + "-" + address.address_type
@@ -148,12 +155,13 @@
frappe.rename_doc("Address", old_address_title, new_address_title)
+
def link_items(items_list, woocommerce_settings, sys_lang):
for item_data in items_list:
item_woo_com_id = cstr(item_data.get("product_id"))
- if not frappe.db.get_value("Item", {"woocommerce_id": item_woo_com_id}, 'name'):
- #Create Item
+ if not frappe.db.get_value("Item", {"woocommerce_id": item_woo_com_id}, "name"):
+ # Create Item
item = frappe.new_doc("Item")
item.item_code = _("woocommerce - {0}", sys_lang).format(item_woo_com_id)
item.stock_uom = woocommerce_settings.uom or _("Nos", sys_lang)
@@ -164,6 +172,7 @@
item.flags.ignore_mandatory = True
item.save()
+
def create_sales_order(order, woocommerce_settings, customer_name, sys_lang):
new_sales_order = frappe.new_doc("Sales Order")
new_sales_order.customer = customer_name
@@ -185,12 +194,12 @@
frappe.db.commit()
+
def set_items_in_sales_order(new_sales_order, woocommerce_settings, order, sys_lang):
- company_abbr = frappe.db.get_value('Company', woocommerce_settings.company, 'abbr')
+ company_abbr = frappe.db.get_value("Company", woocommerce_settings.company, "abbr")
default_warehouse = _("Stores - {0}", sys_lang).format(company_abbr)
- if not frappe.db.exists("Warehouse", default_warehouse) \
- and not woocommerce_settings.warehouse:
+ if not frappe.db.exists("Warehouse", default_warehouse) and not woocommerce_settings.warehouse:
frappe.throw(_("Please set Warehouse in Woocommerce Settings"))
for item in order.get("line_items"):
@@ -199,28 +208,44 @@
ordered_items_tax = item.get("total_tax")
- new_sales_order.append("items", {
- "item_code": found_item.name,
- "item_name": found_item.item_name,
- "description": found_item.item_name,
- "delivery_date": new_sales_order.delivery_date,
- "uom": woocommerce_settings.uom or _("Nos", sys_lang),
- "qty": item.get("quantity"),
- "rate": item.get("price"),
- "warehouse": woocommerce_settings.warehouse or default_warehouse
- })
+ new_sales_order.append(
+ "items",
+ {
+ "item_code": found_item.name,
+ "item_name": found_item.item_name,
+ "description": found_item.item_name,
+ "delivery_date": new_sales_order.delivery_date,
+ "uom": woocommerce_settings.uom or _("Nos", sys_lang),
+ "qty": item.get("quantity"),
+ "rate": item.get("price"),
+ "warehouse": woocommerce_settings.warehouse or default_warehouse,
+ },
+ )
- add_tax_details(new_sales_order, ordered_items_tax, "Ordered Item tax", woocommerce_settings.tax_account)
+ add_tax_details(
+ new_sales_order, ordered_items_tax, "Ordered Item tax", woocommerce_settings.tax_account
+ )
# shipping_details = order.get("shipping_lines") # used for detailed order
- add_tax_details(new_sales_order, order.get("shipping_tax"), "Shipping Tax", woocommerce_settings.f_n_f_account)
- add_tax_details(new_sales_order, order.get("shipping_total"), "Shipping Total", woocommerce_settings.f_n_f_account)
+ add_tax_details(
+ new_sales_order, order.get("shipping_tax"), "Shipping Tax", woocommerce_settings.f_n_f_account
+ )
+ add_tax_details(
+ new_sales_order,
+ order.get("shipping_total"),
+ "Shipping Total",
+ woocommerce_settings.f_n_f_account,
+ )
+
def add_tax_details(sales_order, price, desc, tax_account_head):
- sales_order.append("taxes", {
- "charge_type":"Actual",
- "account_head": tax_account_head,
- "tax_amount": price,
- "description": desc
- })
+ sales_order.append(
+ "taxes",
+ {
+ "charge_type": "Actual",
+ "account_head": tax_account_head,
+ "tax_amount": price,
+ "description": desc,
+ },
+ )
diff --git a/erpnext/erpnext_integrations/data_migration_mapping/issue_to_task/__init__.py b/erpnext/erpnext_integrations/data_migration_mapping/issue_to_task/__init__.py
index 1d0dfa5..616ecfb 100644
--- a/erpnext/erpnext_integrations/data_migration_mapping/issue_to_task/__init__.py
+++ b/erpnext/erpnext_integrations/data_migration_mapping/issue_to_task/__init__.py
@@ -3,10 +3,10 @@
def pre_process(issue):
- project = frappe.db.get_value('Project', filters={'project_name': issue.milestone})
+ project = frappe.db.get_value("Project", filters={"project_name": issue.milestone})
return {
- 'title': issue.title,
- 'body': frappe.utils.md_to_html(issue.body or ''),
- 'state': issue.state.title(),
- 'project': project or ''
+ "title": issue.title,
+ "body": frappe.utils.md_to_html(issue.body or ""),
+ "state": issue.state.title(),
+ "project": project or "",
}
diff --git a/erpnext/erpnext_integrations/data_migration_mapping/milestone_to_project/__init__.py b/erpnext/erpnext_integrations/data_migration_mapping/milestone_to_project/__init__.py
index 212f81b..d44fc04 100644
--- a/erpnext/erpnext_integrations/data_migration_mapping/milestone_to_project/__init__.py
+++ b/erpnext/erpnext_integrations/data_migration_mapping/milestone_to_project/__init__.py
@@ -1,6 +1,6 @@
def pre_process(milestone):
return {
- 'title': milestone.title,
- 'description': milestone.description,
- 'state': milestone.state.title()
+ "title": milestone.title,
+ "description": milestone.description,
+ "state": milestone.state.title(),
}
diff --git a/erpnext/erpnext_integrations/doctype/exotel_settings/exotel_settings.py b/erpnext/erpnext_integrations/doctype/exotel_settings/exotel_settings.py
index e84093c..4879cb5 100644
--- a/erpnext/erpnext_integrations/doctype/exotel_settings/exotel_settings.py
+++ b/erpnext/erpnext_integrations/doctype/exotel_settings/exotel_settings.py
@@ -14,7 +14,9 @@
def verify_credentials(self):
if self.enabled:
- response = requests.get('https://api.exotel.com/v1/Accounts/{sid}'
- .format(sid = self.account_sid), auth=(self.api_key, self.api_token))
+ response = requests.get(
+ "https://api.exotel.com/v1/Accounts/{sid}".format(sid=self.account_sid),
+ auth=(self.api_key, self.api_token),
+ )
if response.status_code != 200:
frappe.throw(_("Invalid credentials"))
diff --git a/erpnext/erpnext_integrations/doctype/gocardless_settings/__init__.py b/erpnext/erpnext_integrations/doctype/gocardless_settings/__init__.py
index bb62c39..65be599 100644
--- a/erpnext/erpnext_integrations/doctype/gocardless_settings/__init__.py
+++ b/erpnext/erpnext_integrations/doctype/gocardless_settings/__init__.py
@@ -23,12 +23,15 @@
set_status(event)
return 200
+
+
def set_status(event):
resource_type = event.get("resource_type", {})
if resource_type == "mandates":
set_mandate_status(event)
+
def set_mandate_status(event):
mandates = []
if isinstance(event["links"], (list,)):
@@ -37,7 +40,12 @@
else:
mandates.append(event["links"]["mandate"])
- if event["action"] == "pending_customer_approval" or event["action"] == "pending_submission" or event["action"] == "submitted" or event["action"] == "active":
+ if (
+ event["action"] == "pending_customer_approval"
+ or event["action"] == "pending_submission"
+ or event["action"] == "submitted"
+ or event["action"] == "active"
+ ):
disabled = 0
else:
disabled = 1
@@ -45,6 +53,7 @@
for mandate in mandates:
frappe.db.set_value("GoCardless Mandate", mandate, "disabled", disabled)
+
def authenticate_signature(r):
"""Returns True if the received signature matches the generated signature"""
received_signature = frappe.get_request_header("Webhook-Signature")
@@ -59,13 +68,22 @@
return False
+
def get_webhook_keys():
def _get_webhook_keys():
- webhook_keys = [d.webhooks_secret for d in frappe.get_all("GoCardless Settings", fields=["webhooks_secret"],) if d.webhooks_secret]
+ webhook_keys = [
+ d.webhooks_secret
+ for d in frappe.get_all(
+ "GoCardless Settings",
+ fields=["webhooks_secret"],
+ )
+ if d.webhooks_secret
+ ]
return webhook_keys
return frappe.cache().get_value("gocardless_webhooks_secret", _get_webhook_keys)
+
def clear_cache():
frappe.cache().delete_value("gocardless_webhooks_secret")
diff --git a/erpnext/erpnext_integrations/doctype/gocardless_settings/gocardless_settings.py b/erpnext/erpnext_integrations/doctype/gocardless_settings/gocardless_settings.py
index f02f76e..e1972ae 100644
--- a/erpnext/erpnext_integrations/doctype/gocardless_settings/gocardless_settings.py
+++ b/erpnext/erpnext_integrations/doctype/gocardless_settings/gocardless_settings.py
@@ -22,32 +22,35 @@
self.environment = self.get_environment()
try:
self.client = gocardless_pro.Client(
- access_token=self.access_token,
- environment=self.environment
- )
+ access_token=self.access_token, environment=self.environment
+ )
return self.client
except Exception as e:
frappe.throw(e)
def on_update(self):
- create_payment_gateway('GoCardless-' + self.gateway_name, settings='GoCardLess Settings', controller=self.gateway_name)
- call_hook_method('payment_gateway_enabled', gateway='GoCardless-' + self.gateway_name)
+ create_payment_gateway(
+ "GoCardless-" + self.gateway_name, settings="GoCardLess Settings", controller=self.gateway_name
+ )
+ call_hook_method("payment_gateway_enabled", gateway="GoCardless-" + self.gateway_name)
def on_payment_request_submission(self, data):
if data.reference_doctype != "Fees":
- customer_data = frappe.db.get_value(data.reference_doctype, data.reference_name, ["company", "customer_name"], as_dict=1)
+ customer_data = frappe.db.get_value(
+ data.reference_doctype, data.reference_name, ["company", "customer_name"], as_dict=1
+ )
data = {
- "amount": flt(data.grand_total, data.precision("grand_total")),
- "title": customer_data.company.encode("utf-8"),
- "description": data.subject.encode("utf-8"),
- "reference_doctype": data.doctype,
- "reference_docname": data.name,
- "payer_email": data.email_to or frappe.session.user,
- "payer_name": customer_data.customer_name,
- "order_id": data.name,
- "currency": data.currency
- }
+ "amount": flt(data.grand_total, data.precision("grand_total")),
+ "title": customer_data.company.encode("utf-8"),
+ "description": data.subject.encode("utf-8"),
+ "reference_doctype": data.doctype,
+ "reference_docname": data.name,
+ "payer_email": data.email_to or frappe.session.user,
+ "payer_name": customer_data.customer_name,
+ "order_id": data.name,
+ "currency": data.currency,
+ }
valid_mandate = self.check_mandate_validity(data)
if valid_mandate is not None:
@@ -60,12 +63,19 @@
def check_mandate_validity(self, data):
- if frappe.db.exists("GoCardless Mandate", dict(customer=data.get('payer_name'), disabled=0)):
- registered_mandate = frappe.db.get_value("GoCardless Mandate", dict(customer=data.get('payer_name'), disabled=0), 'mandate')
+ if frappe.db.exists("GoCardless Mandate", dict(customer=data.get("payer_name"), disabled=0)):
+ registered_mandate = frappe.db.get_value(
+ "GoCardless Mandate", dict(customer=data.get("payer_name"), disabled=0), "mandate"
+ )
self.initialize_client()
mandate = self.client.mandates.get(registered_mandate)
- if mandate.status=="pending_customer_approval" or mandate.status=="pending_submission" or mandate.status=="submitted" or mandate.status=="active":
+ if (
+ mandate.status == "pending_customer_approval"
+ or mandate.status == "pending_submission"
+ or mandate.status == "submitted"
+ or mandate.status == "active"
+ ):
return {"mandate": registered_mandate}
else:
return None
@@ -74,13 +84,17 @@
def get_environment(self):
if self.use_sandbox:
- return 'sandbox'
+ return "sandbox"
else:
- return 'live'
+ return "live"
def validate_transaction_currency(self, currency):
if currency not in self.supported_currencies:
- frappe.throw(_("Please select another payment method. Go Cardless does not support transactions in currency '{0}'").format(currency))
+ frappe.throw(
+ _(
+ "Please select another payment method. Go Cardless does not support transactions in currency '{0}'"
+ ).format(currency)
+ )
def get_payment_url(self, **kwargs):
return get_url("./integrations/gocardless_checkout?{0}".format(urlencode(kwargs)))
@@ -94,63 +108,85 @@
except Exception:
frappe.log_error(frappe.get_traceback())
- return{
- "redirect_to": frappe.redirect_to_message(_('Server Error'), _("There seems to be an issue with the server's GoCardless configuration. Don't worry, in case of failure, the amount will get refunded to your account.")),
- "status": 401
+ return {
+ "redirect_to": frappe.redirect_to_message(
+ _("Server Error"),
+ _(
+ "There seems to be an issue with the server's GoCardless configuration. Don't worry, in case of failure, the amount will get refunded to your account."
+ ),
+ ),
+ "status": 401,
}
def create_charge_on_gocardless(self):
- redirect_to = self.data.get('redirect_to') or None
- redirect_message = self.data.get('redirect_message') or None
+ redirect_to = self.data.get("redirect_to") or None
+ redirect_message = self.data.get("redirect_message") or None
- reference_doc = frappe.get_doc(self.data.get('reference_doctype'), self.data.get('reference_docname'))
+ reference_doc = frappe.get_doc(
+ self.data.get("reference_doctype"), self.data.get("reference_docname")
+ )
self.initialize_client()
try:
payment = self.client.payments.create(
params={
- "amount" : cint(reference_doc.grand_total * 100),
- "currency" : reference_doc.currency,
- "links" : {
- "mandate": self.data.get('mandate')
- },
+ "amount": cint(reference_doc.grand_total * 100),
+ "currency": reference_doc.currency,
+ "links": {"mandate": self.data.get("mandate")},
"metadata": {
- "reference_doctype": reference_doc.doctype,
- "reference_document": reference_doc.name
- }
- }, headers={
- 'Idempotency-Key' : self.data.get('reference_docname'),
- })
+ "reference_doctype": reference_doc.doctype,
+ "reference_document": reference_doc.name,
+ },
+ },
+ headers={
+ "Idempotency-Key": self.data.get("reference_docname"),
+ },
+ )
- if payment.status=="pending_submission" or payment.status=="pending_customer_approval" or payment.status=="submitted":
- self.integration_request.db_set('status', 'Authorized', update_modified=False)
+ if (
+ payment.status == "pending_submission"
+ or payment.status == "pending_customer_approval"
+ or payment.status == "submitted"
+ ):
+ self.integration_request.db_set("status", "Authorized", update_modified=False)
self.flags.status_changed_to = "Completed"
- self.integration_request.db_set('output', payment.status, update_modified=False)
+ self.integration_request.db_set("output", payment.status, update_modified=False)
- elif payment.status=="confirmed" or payment.status=="paid_out":
- self.integration_request.db_set('status', 'Completed', update_modified=False)
+ elif payment.status == "confirmed" or payment.status == "paid_out":
+ self.integration_request.db_set("status", "Completed", update_modified=False)
self.flags.status_changed_to = "Completed"
- self.integration_request.db_set('output', payment.status, update_modified=False)
+ self.integration_request.db_set("output", payment.status, update_modified=False)
- elif payment.status=="cancelled" or payment.status=="customer_approval_denied" or payment.status=="charged_back":
- self.integration_request.db_set('status', 'Cancelled', update_modified=False)
- frappe.log_error(_("Payment Cancelled. Please check your GoCardless Account for more details"), "GoCardless Payment Error")
- self.integration_request.db_set('error', payment.status, update_modified=False)
+ elif (
+ payment.status == "cancelled"
+ or payment.status == "customer_approval_denied"
+ or payment.status == "charged_back"
+ ):
+ self.integration_request.db_set("status", "Cancelled", update_modified=False)
+ frappe.log_error(
+ _("Payment Cancelled. Please check your GoCardless Account for more details"),
+ "GoCardless Payment Error",
+ )
+ self.integration_request.db_set("error", payment.status, update_modified=False)
else:
- self.integration_request.db_set('status', 'Failed', update_modified=False)
- frappe.log_error(_("Payment Failed. Please check your GoCardless Account for more details"), "GoCardless Payment Error")
- self.integration_request.db_set('error', payment.status, update_modified=False)
+ self.integration_request.db_set("status", "Failed", update_modified=False)
+ frappe.log_error(
+ _("Payment Failed. Please check your GoCardless Account for more details"),
+ "GoCardless Payment Error",
+ )
+ self.integration_request.db_set("error", payment.status, update_modified=False)
except Exception as e:
frappe.log_error(e, "GoCardless Payment Error")
if self.flags.status_changed_to == "Completed":
- status = 'Completed'
- if 'reference_doctype' in self.data and 'reference_docname' in self.data:
+ status = "Completed"
+ if "reference_doctype" in self.data and "reference_docname" in self.data:
custom_redirect_to = None
try:
- custom_redirect_to = frappe.get_doc(self.data.get('reference_doctype'),
- self.data.get('reference_docname')).run_method("on_payment_authorized", self.flags.status_changed_to)
+ custom_redirect_to = frappe.get_doc(
+ self.data.get("reference_doctype"), self.data.get("reference_docname")
+ ).run_method("on_payment_authorized", self.flags.status_changed_to)
except Exception:
frappe.log_error(frappe.get_traceback())
@@ -159,24 +195,25 @@
redirect_url = redirect_to
else:
- status = 'Error'
- redirect_url = 'payment-failed'
+ status = "Error"
+ redirect_url = "payment-failed"
if redirect_message:
- redirect_url += '&' + urlencode({'redirect_message': redirect_message})
+ redirect_url += "&" + urlencode({"redirect_message": redirect_message})
redirect_url = get_url(redirect_url)
- return {
- "redirect_to": redirect_url,
- "status": status
- }
+ return {"redirect_to": redirect_url, "status": status}
+
def get_gateway_controller(doc):
payment_request = frappe.get_doc("Payment Request", doc)
- gateway_controller = frappe.db.get_value("Payment Gateway", payment_request.payment_gateway, "gateway_controller")
+ gateway_controller = frappe.db.get_value(
+ "Payment Gateway", payment_request.payment_gateway, "gateway_controller"
+ )
return gateway_controller
+
def gocardless_initialization(doc):
gateway_controller = get_gateway_controller(doc)
settings = frappe.get_doc("GoCardless Settings", gateway_controller)
diff --git a/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_connector.py b/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_connector.py
index 6d46a1c..a577e7f 100644
--- a/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_connector.py
+++ b/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_connector.py
@@ -5,9 +5,15 @@
from requests.auth import HTTPBasicAuth
-class MpesaConnector():
- def __init__(self, env="sandbox", app_key=None, app_secret=None, sandbox_url="https://sandbox.safaricom.co.ke",
- live_url="https://api.safaricom.co.ke"):
+class MpesaConnector:
+ def __init__(
+ self,
+ env="sandbox",
+ app_key=None,
+ app_secret=None,
+ sandbox_url="https://sandbox.safaricom.co.ke",
+ live_url="https://api.safaricom.co.ke",
+ ):
"""Setup configuration for Mpesa connector and generate new access token."""
self.env = env
self.app_key = app_key
@@ -23,36 +29,41 @@
This method is used to fetch the access token required by Mpesa.
Returns:
- access_token (str): This token is to be used with the Bearer header for further API calls to Mpesa.
+ access_token (str): This token is to be used with the Bearer header for further API calls to Mpesa.
"""
authenticate_uri = "/oauth/v1/generate?grant_type=client_credentials"
authenticate_url = "{0}{1}".format(self.base_url, authenticate_uri)
- r = requests.get(
- authenticate_url,
- auth=HTTPBasicAuth(self.app_key, self.app_secret)
- )
- self.authentication_token = r.json()['access_token']
- return r.json()['access_token']
+ r = requests.get(authenticate_url, auth=HTTPBasicAuth(self.app_key, self.app_secret))
+ self.authentication_token = r.json()["access_token"]
+ return r.json()["access_token"]
- def get_balance(self, initiator=None, security_credential=None, party_a=None, identifier_type=None,
- remarks=None, queue_timeout_url=None,result_url=None):
+ def get_balance(
+ self,
+ initiator=None,
+ security_credential=None,
+ party_a=None,
+ identifier_type=None,
+ remarks=None,
+ queue_timeout_url=None,
+ result_url=None,
+ ):
"""
This method uses Mpesa's Account Balance API to to enquire the balance on a M-Pesa BuyGoods (Till Number).
Args:
- initiator (str): Username used to authenticate the transaction.
- security_credential (str): Generate from developer portal.
- command_id (str): AccountBalance.
- party_a (int): Till number being queried.
- identifier_type (int): Type of organization receiving the transaction. (MSISDN/Till Number/Organization short code)
- remarks (str): Comments that are sent along with the transaction(maximum 100 characters).
- queue_timeout_url (str): The url that handles information of timed out transactions.
- result_url (str): The url that receives results from M-Pesa api call.
+ initiator (str): Username used to authenticate the transaction.
+ security_credential (str): Generate from developer portal.
+ command_id (str): AccountBalance.
+ party_a (int): Till number being queried.
+ identifier_type (int): Type of organization receiving the transaction. (MSISDN/Till Number/Organization short code)
+ remarks (str): Comments that are sent along with the transaction(maximum 100 characters).
+ queue_timeout_url (str): The url that handles information of timed out transactions.
+ result_url (str): The url that receives results from M-Pesa api call.
Returns:
- OriginatorConverstionID (str): The unique request ID for tracking a transaction.
- ConversationID (str): The unique request ID returned by mpesa for each request made
- ResponseDescription (str): Response Description message
+ OriginatorConverstionID (str): The unique request ID for tracking a transaction.
+ ConversationID (str): The unique request ID returned by mpesa for each request made
+ ResponseDescription (str): Response Description message
"""
payload = {
@@ -63,43 +74,56 @@
"IdentifierType": identifier_type,
"Remarks": remarks,
"QueueTimeOutURL": queue_timeout_url,
- "ResultURL": result_url
+ "ResultURL": result_url,
}
- headers = {'Authorization': 'Bearer {0}'.format(self.authentication_token), 'Content-Type': "application/json"}
+ headers = {
+ "Authorization": "Bearer {0}".format(self.authentication_token),
+ "Content-Type": "application/json",
+ }
saf_url = "{0}{1}".format(self.base_url, "/mpesa/accountbalance/v1/query")
r = requests.post(saf_url, headers=headers, json=payload)
return r.json()
- def stk_push(self, business_shortcode=None, passcode=None, amount=None, callback_url=None, reference_code=None,
- phone_number=None, description=None):
+ def stk_push(
+ self,
+ business_shortcode=None,
+ passcode=None,
+ amount=None,
+ callback_url=None,
+ reference_code=None,
+ phone_number=None,
+ description=None,
+ ):
"""
This method uses Mpesa's Express API to initiate online payment on behalf of a customer.
Args:
- business_shortcode (int): The short code of the organization.
- passcode (str): Get from developer portal
- amount (int): The amount being transacted
- callback_url (str): A CallBack URL is a valid secure URL that is used to receive notifications from M-Pesa API.
- reference_code(str): Account Reference: This is an Alpha-Numeric parameter that is defined by your system as an Identifier of the transaction for CustomerPayBillOnline transaction type.
- phone_number(int): The Mobile Number to receive the STK Pin Prompt.
- description(str): This is any additional information/comment that can be sent along with the request from your system. MAX 13 characters
+ business_shortcode (int): The short code of the organization.
+ passcode (str): Get from developer portal
+ amount (int): The amount being transacted
+ callback_url (str): A CallBack URL is a valid secure URL that is used to receive notifications from M-Pesa API.
+ reference_code(str): Account Reference: This is an Alpha-Numeric parameter that is defined by your system as an Identifier of the transaction for CustomerPayBillOnline transaction type.
+ phone_number(int): The Mobile Number to receive the STK Pin Prompt.
+ description(str): This is any additional information/comment that can be sent along with the request from your system. MAX 13 characters
Success Response:
- CustomerMessage(str): Messages that customers can understand.
- CheckoutRequestID(str): This is a global unique identifier of the processed checkout transaction request.
- ResponseDescription(str): Describes Success or failure
- MerchantRequestID(str): This is a global unique Identifier for any submitted payment request.
- ResponseCode(int): 0 means success all others are error codes. e.g.404.001.03
+ CustomerMessage(str): Messages that customers can understand.
+ CheckoutRequestID(str): This is a global unique identifier of the processed checkout transaction request.
+ ResponseDescription(str): Describes Success or failure
+ MerchantRequestID(str): This is a global unique Identifier for any submitted payment request.
+ ResponseCode(int): 0 means success all others are error codes. e.g.404.001.03
Error Reponse:
- requestId(str): This is a unique requestID for the payment request
- errorCode(str): This is a predefined code that indicates the reason for request failure.
- errorMessage(str): This is a predefined code that indicates the reason for request failure.
+ requestId(str): This is a unique requestID for the payment request
+ errorCode(str): This is a predefined code that indicates the reason for request failure.
+ errorMessage(str): This is a predefined code that indicates the reason for request failure.
"""
- time = str(datetime.datetime.now()).split(".")[0].replace("-", "").replace(" ", "").replace(":", "")
+ time = (
+ str(datetime.datetime.now()).split(".")[0].replace("-", "").replace(" ", "").replace(":", "")
+ )
password = "{0}{1}{2}".format(str(business_shortcode), str(passcode), time)
- encoded = base64.b64encode(bytes(password, encoding='utf8'))
+ encoded = base64.b64encode(bytes(password, encoding="utf8"))
payload = {
"BusinessShortCode": business_shortcode,
"Password": encoded.decode("utf-8"),
@@ -111,9 +135,14 @@
"CallBackURL": callback_url,
"AccountReference": reference_code,
"TransactionDesc": description,
- "TransactionType": "CustomerPayBillOnline" if self.env == "sandbox" else "CustomerBuyGoodsOnline"
+ "TransactionType": "CustomerPayBillOnline"
+ if self.env == "sandbox"
+ else "CustomerBuyGoodsOnline",
}
- headers = {'Authorization': 'Bearer {0}'.format(self.authentication_token), 'Content-Type': "application/json"}
+ headers = {
+ "Authorization": "Bearer {0}".format(self.authentication_token),
+ "Content-Type": "application/json",
+ }
saf_url = "{0}{1}".format(self.base_url, "/mpesa/stkpush/v1/processrequest")
r = requests.post(saf_url, headers=headers, json=payload)
diff --git a/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_custom_fields.py b/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_custom_fields.py
index 368139b..c92edc5 100644
--- a/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_custom_fields.py
+++ b/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_custom_fields.py
@@ -11,21 +11,22 @@
"label": "Request for Payment",
"fieldtype": "Button",
"hidden": 1,
- "insert_after": "contact_email"
+ "insert_after": "contact_email",
},
{
"fieldname": "mpesa_receipt_number",
"label": "Mpesa Receipt Number",
"fieldtype": "Data",
"read_only": 1,
- "insert_after": "company"
- }
+ "insert_after": "company",
+ },
]
}
if not frappe.get_meta("POS Invoice").has_field("request_for_payment"):
create_custom_fields(pos_field)
- record_dict = [{
+ record_dict = [
+ {
"doctype": "POS Field",
"fieldname": "contact_mobile",
"label": "Mobile No",
@@ -33,7 +34,7 @@
"options": "Phone",
"parenttype": "POS Settings",
"parent": "POS Settings",
- "parentfield": "invoice_fields"
+ "parentfield": "invoice_fields",
},
{
"doctype": "POS Field",
@@ -42,11 +43,12 @@
"fieldtype": "Button",
"parenttype": "POS Settings",
"parent": "POS Settings",
- "parentfield": "invoice_fields"
- }
+ "parentfield": "invoice_fields",
+ },
]
create_pos_settings(record_dict)
+
def create_pos_settings(record_dict):
for record in record_dict:
if frappe.db.exists("POS Field", {"fieldname": record.get("fieldname")}):
diff --git a/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_settings.py b/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_settings.py
index e7b4a30..78a598c 100644
--- a/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_settings.py
+++ b/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_settings.py
@@ -2,7 +2,6 @@
# For license information, please see license.txt
-
from json import dumps, loads
import frappe
@@ -23,16 +22,26 @@
def validate_transaction_currency(self, currency):
if currency not in self.supported_currencies:
- frappe.throw(_("Please select another payment method. Mpesa does not support transactions in currency '{0}'").format(currency))
+ frappe.throw(
+ _(
+ "Please select another payment method. Mpesa does not support transactions in currency '{0}'"
+ ).format(currency)
+ )
def on_update(self):
create_custom_pos_fields()
- create_payment_gateway('Mpesa-' + self.payment_gateway_name, settings='Mpesa Settings', controller=self.payment_gateway_name)
- call_hook_method('payment_gateway_enabled', gateway='Mpesa-' + self.payment_gateway_name, payment_channel="Phone")
+ create_payment_gateway(
+ "Mpesa-" + self.payment_gateway_name,
+ settings="Mpesa Settings",
+ controller=self.payment_gateway_name,
+ )
+ call_hook_method(
+ "payment_gateway_enabled", gateway="Mpesa-" + self.payment_gateway_name, payment_channel="Phone"
+ )
# required to fetch the bank account details from the payment gateway account
frappe.db.commit()
- create_mode_of_payment('Mpesa-' + self.payment_gateway_name, payment_type="Phone")
+ create_mode_of_payment("Mpesa-" + self.payment_gateway_name, payment_type="Phone")
def request_for_payment(self, **kwargs):
args = frappe._dict(kwargs)
@@ -44,6 +53,7 @@
from erpnext.erpnext_integrations.doctype.mpesa_settings.test_mpesa_settings import (
get_payment_request_response_payload,
)
+
response = frappe._dict(get_payment_request_response_payload(amount))
else:
response = frappe._dict(generate_stk_push(**args))
@@ -55,11 +65,15 @@
if request_amount > self.transaction_limit:
# make multiple requests
request_amounts = []
- requests_to_be_made = frappe.utils.ceil(request_amount / self.transaction_limit) # 480/150 = ceil(3.2) = 4
+ requests_to_be_made = frappe.utils.ceil(
+ request_amount / self.transaction_limit
+ ) # 480/150 = ceil(3.2) = 4
for i in range(requests_to_be_made):
amount = self.transaction_limit
if i == requests_to_be_made - 1:
- amount = request_amount - (self.transaction_limit * i) # for 4th request, 480 - (150 * 3) = 30
+ amount = request_amount - (
+ self.transaction_limit * i
+ ) # for 4th request, 480 - (150 * 3) = 30
request_amounts.append(amount)
else:
request_amounts = [request_amount]
@@ -69,15 +83,14 @@
@frappe.whitelist()
def get_account_balance_info(self):
payload = dict(
- reference_doctype="Mpesa Settings",
- reference_docname=self.name,
- doc_details=vars(self)
+ reference_doctype="Mpesa Settings", reference_docname=self.name, doc_details=vars(self)
)
if frappe.flags.in_test:
from erpnext.erpnext_integrations.doctype.mpesa_settings.test_mpesa_settings import (
get_test_account_balance_response,
)
+
response = frappe._dict(get_test_account_balance_response())
else:
response = frappe._dict(get_account_balance(payload))
@@ -95,46 +108,62 @@
req_name = getattr(response, global_id)
error = None
- if not frappe.db.exists('Integration Request', req_name):
+ if not frappe.db.exists("Integration Request", req_name):
create_request_log(request_dict, "Host", "Mpesa", req_name, error)
if error:
frappe.throw(_(getattr(response, "errorMessage")), title=_("Transaction Error"))
+
def generate_stk_push(**kwargs):
"""Generate stk push by making a API call to the stk push API."""
args = frappe._dict(kwargs)
try:
- callback_url = get_request_site_address(True) + "/api/method/erpnext.erpnext_integrations.doctype.mpesa_settings.mpesa_settings.verify_transaction"
+ callback_url = (
+ get_request_site_address(True)
+ + "/api/method/erpnext.erpnext_integrations.doctype.mpesa_settings.mpesa_settings.verify_transaction"
+ )
mpesa_settings = frappe.get_doc("Mpesa Settings", args.payment_gateway[6:])
env = "production" if not mpesa_settings.sandbox else "sandbox"
# for sandbox, business shortcode is same as till number
- business_shortcode = mpesa_settings.business_shortcode if env == "production" else mpesa_settings.till_number
+ business_shortcode = (
+ mpesa_settings.business_shortcode if env == "production" else mpesa_settings.till_number
+ )
- connector = MpesaConnector(env=env,
+ connector = MpesaConnector(
+ env=env,
app_key=mpesa_settings.consumer_key,
- app_secret=mpesa_settings.get_password("consumer_secret"))
+ app_secret=mpesa_settings.get_password("consumer_secret"),
+ )
mobile_number = sanitize_mobile_number(args.sender)
response = connector.stk_push(
- business_shortcode=business_shortcode, amount=args.request_amount,
+ business_shortcode=business_shortcode,
+ amount=args.request_amount,
passcode=mpesa_settings.get_password("online_passkey"),
- callback_url=callback_url, reference_code=mpesa_settings.till_number,
- phone_number=mobile_number, description="POS Payment"
+ callback_url=callback_url,
+ reference_code=mpesa_settings.till_number,
+ phone_number=mobile_number,
+ description="POS Payment",
)
return response
except Exception:
frappe.log_error(title=_("Mpesa Express Transaction Error"))
- frappe.throw(_("Issue detected with Mpesa configuration, check the error logs for more details"), title=_("Mpesa Express Error"))
+ frappe.throw(
+ _("Issue detected with Mpesa configuration, check the error logs for more details"),
+ title=_("Mpesa Express Error"),
+ )
+
def sanitize_mobile_number(number):
"""Add country code and strip leading zeroes from the phone number."""
return "254" + str(number).lstrip("0")
+
@frappe.whitelist(allow_guest=True)
def verify_transaction(**kwargs):
"""Verify the transaction result received via callback from stk."""
@@ -146,28 +175,28 @@
integration_request = frappe.get_doc("Integration Request", checkout_id)
transaction_data = frappe._dict(loads(integration_request.data))
- total_paid = 0 # for multiple integration request made against a pos invoice
- success = False # for reporting successfull callback to point of sale ui
+ total_paid = 0 # for multiple integration request made against a pos invoice
+ success = False # for reporting successfull callback to point of sale ui
- if transaction_response['ResultCode'] == 0:
+ if transaction_response["ResultCode"] == 0:
if integration_request.reference_doctype and integration_request.reference_docname:
try:
item_response = transaction_response["CallbackMetadata"]["Item"]
amount = fetch_param_value(item_response, "Amount", "Name")
mpesa_receipt = fetch_param_value(item_response, "MpesaReceiptNumber", "Name")
- pr = frappe.get_doc(integration_request.reference_doctype, integration_request.reference_docname)
+ pr = frappe.get_doc(
+ integration_request.reference_doctype, integration_request.reference_docname
+ )
mpesa_receipts, completed_payments = get_completed_integration_requests_info(
- integration_request.reference_doctype,
- integration_request.reference_docname,
- checkout_id
+ integration_request.reference_doctype, integration_request.reference_docname, checkout_id
)
total_paid = amount + sum(completed_payments)
- mpesa_receipts = ', '.join(mpesa_receipts + [mpesa_receipt])
+ mpesa_receipts = ", ".join(mpesa_receipts + [mpesa_receipt])
if total_paid >= pr.grand_total:
- pr.run_method("on_payment_authorized", 'Completed')
+ pr.run_method("on_payment_authorized", "Completed")
success = True
frappe.db.set_value("POS Invoice", pr.reference_name, "mpesa_receipt_number", mpesa_receipts)
@@ -180,24 +209,31 @@
integration_request.handle_failure(transaction_response)
frappe.publish_realtime(
- event='process_phone_payment',
+ event="process_phone_payment",
doctype="POS Invoice",
docname=transaction_data.payment_reference,
user=integration_request.owner,
message={
- 'amount': total_paid,
- 'success': success,
- 'failure_message': transaction_response["ResultDesc"] if transaction_response['ResultCode'] != 0 else ''
+ "amount": total_paid,
+ "success": success,
+ "failure_message": transaction_response["ResultDesc"]
+ if transaction_response["ResultCode"] != 0
+ else "",
},
)
+
def get_completed_integration_requests_info(reference_doctype, reference_docname, checkout_id):
- output_of_other_completed_requests = frappe.get_all("Integration Request", filters={
- 'name': ['!=', checkout_id],
- 'reference_doctype': reference_doctype,
- 'reference_docname': reference_docname,
- 'status': 'Completed'
- }, pluck="output")
+ output_of_other_completed_requests = frappe.get_all(
+ "Integration Request",
+ filters={
+ "name": ["!=", checkout_id],
+ "reference_doctype": reference_doctype,
+ "reference_docname": reference_docname,
+ "status": "Completed",
+ },
+ pluck="output",
+ )
mpesa_receipts, completed_payments = [], []
@@ -211,23 +247,38 @@
return mpesa_receipts, completed_payments
+
def get_account_balance(request_payload):
"""Call account balance API to send the request to the Mpesa Servers."""
try:
mpesa_settings = frappe.get_doc("Mpesa Settings", request_payload.get("reference_docname"))
env = "production" if not mpesa_settings.sandbox else "sandbox"
- connector = MpesaConnector(env=env,
+ connector = MpesaConnector(
+ env=env,
app_key=mpesa_settings.consumer_key,
- app_secret=mpesa_settings.get_password("consumer_secret"))
+ app_secret=mpesa_settings.get_password("consumer_secret"),
+ )
- callback_url = get_request_site_address(True) + "/api/method/erpnext.erpnext_integrations.doctype.mpesa_settings.mpesa_settings.process_balance_info"
+ callback_url = (
+ get_request_site_address(True)
+ + "/api/method/erpnext.erpnext_integrations.doctype.mpesa_settings.mpesa_settings.process_balance_info"
+ )
- response = connector.get_balance(mpesa_settings.initiator_name, mpesa_settings.security_credential, mpesa_settings.till_number, 4, mpesa_settings.name, callback_url, callback_url)
+ response = connector.get_balance(
+ mpesa_settings.initiator_name,
+ mpesa_settings.security_credential,
+ mpesa_settings.till_number,
+ 4,
+ mpesa_settings.name,
+ callback_url,
+ callback_url,
+ )
return response
except Exception:
frappe.log_error(title=_("Account Balance Processing Error"))
frappe.throw(_("Please check your configuration and try again"), title=_("Error"))
+
@frappe.whitelist(allow_guest=True)
def process_balance_info(**kwargs):
"""Process and store account balance information received via callback from the account balance API call."""
@@ -255,35 +306,43 @@
ref_doc.db_set("account_balance", balance_info)
request.handle_success(account_balance_response)
- frappe.publish_realtime("refresh_mpesa_dashboard", doctype="Mpesa Settings",
- docname=transaction_data.reference_docname, user=transaction_data.owner)
+ frappe.publish_realtime(
+ "refresh_mpesa_dashboard",
+ doctype="Mpesa Settings",
+ docname=transaction_data.reference_docname,
+ user=transaction_data.owner,
+ )
except Exception:
request.handle_failure(account_balance_response)
- frappe.log_error(title=_("Mpesa Account Balance Processing Error"), message=account_balance_response)
+ frappe.log_error(
+ title=_("Mpesa Account Balance Processing Error"), message=account_balance_response
+ )
else:
request.handle_failure(account_balance_response)
+
def format_string_to_json(balance_info):
"""
Format string to json.
e.g: '''Working Account|KES|481000.00|481000.00|0.00|0.00'''
=> {'Working Account': {'current_balance': '481000.00',
- 'available_balance': '481000.00',
- 'reserved_balance': '0.00',
- 'uncleared_balance': '0.00'}}
+ 'available_balance': '481000.00',
+ 'reserved_balance': '0.00',
+ 'uncleared_balance': '0.00'}}
"""
balance_dict = frappe._dict()
for account_info in balance_info.split("&"):
- account_info = account_info.split('|')
+ account_info = account_info.split("|")
balance_dict[account_info[0]] = dict(
current_balance=fmt_money(account_info[2], currency="KES"),
available_balance=fmt_money(account_info[3], currency="KES"),
reserved_balance=fmt_money(account_info[4], currency="KES"),
- uncleared_balance=fmt_money(account_info[5], currency="KES")
+ uncleared_balance=fmt_money(account_info[5], currency="KES"),
)
return dumps(balance_dict)
+
def fetch_param_value(response, key, key_field):
"""Fetch the specified key from list of dictionary. Key is identified via the key field."""
for param in response:
diff --git a/erpnext/erpnext_integrations/doctype/mpesa_settings/test_mpesa_settings.py b/erpnext/erpnext_integrations/doctype/mpesa_settings/test_mpesa_settings.py
index 3945afa..17e332c 100644
--- a/erpnext/erpnext_integrations/doctype/mpesa_settings/test_mpesa_settings.py
+++ b/erpnext/erpnext_integrations/doctype/mpesa_settings/test_mpesa_settings.py
@@ -22,12 +22,12 @@
create_mpesa_settings(payment_gateway_name="Payment")
def tearDown(self):
- frappe.db.sql('delete from `tabMpesa Settings`')
+ frappe.db.sql("delete from `tabMpesa Settings`")
frappe.db.sql('delete from `tabIntegration Request` where integration_request_service = "Mpesa"')
def test_creation_of_payment_gateway(self):
- mode_of_payment = create_mode_of_payment('Mpesa-_Test', payment_type="Phone")
- self.assertTrue(frappe.db.exists("Payment Gateway Account", {'payment_gateway': "Mpesa-_Test"}))
+ mode_of_payment = create_mode_of_payment("Mpesa-_Test", payment_type="Phone")
+ self.assertTrue(frappe.db.exists("Payment Gateway Account", {"payment_gateway": "Mpesa-_Test"}))
self.assertTrue(mode_of_payment.name)
self.assertEqual(mode_of_payment.type, "Phone")
@@ -45,24 +45,33 @@
# test formatting of account balance received as string to json with appropriate currency symbol
mpesa_doc.reload()
- self.assertEqual(mpesa_doc.account_balance, dumps({
- "Working Account": {
- "current_balance": "Sh 481,000.00",
- "available_balance": "Sh 481,000.00",
- "reserved_balance": "Sh 0.00",
- "uncleared_balance": "Sh 0.00"
- }
- }))
+ self.assertEqual(
+ mpesa_doc.account_balance,
+ dumps(
+ {
+ "Working Account": {
+ "current_balance": "Sh 481,000.00",
+ "available_balance": "Sh 481,000.00",
+ "reserved_balance": "Sh 0.00",
+ "uncleared_balance": "Sh 0.00",
+ }
+ }
+ ),
+ )
integration_request.delete()
def test_processing_of_callback_payload(self):
- mpesa_account = frappe.db.get_value("Payment Gateway Account", {"payment_gateway": 'Mpesa-Payment'}, "payment_account")
+ mpesa_account = frappe.db.get_value(
+ "Payment Gateway Account", {"payment_gateway": "Mpesa-Payment"}, "payment_account"
+ )
frappe.db.set_value("Account", mpesa_account, "account_currency", "KES")
frappe.db.set_value("Customer", "_Test Customer", "default_currency", "KES")
pos_invoice = create_pos_invoice(do_not_submit=1)
- pos_invoice.append("payments", {'mode_of_payment': 'Mpesa-Payment', 'account': mpesa_account, 'amount': 500})
+ pos_invoice.append(
+ "payments", {"mode_of_payment": "Mpesa-Payment", "account": mpesa_account, "amount": 500}
+ )
pos_invoice.contact_mobile = "093456543894"
pos_invoice.currency = "KES"
pos_invoice.save()
@@ -72,12 +81,18 @@
self.assertEqual(pr.payment_gateway, "Mpesa-Payment")
# submitting payment request creates integration requests with random id
- integration_req_ids = frappe.get_all("Integration Request", filters={
- 'reference_doctype': pr.doctype,
- 'reference_docname': pr.name,
- }, pluck="name")
+ integration_req_ids = frappe.get_all(
+ "Integration Request",
+ filters={
+ "reference_doctype": pr.doctype,
+ "reference_docname": pr.name,
+ },
+ pluck="name",
+ )
- callback_response = get_payment_callback_payload(Amount=500, CheckoutRequestID=integration_req_ids[0])
+ callback_response = get_payment_callback_payload(
+ Amount=500, CheckoutRequestID=integration_req_ids[0]
+ )
verify_transaction(**callback_response)
# test creation of integration request
integration_request = frappe.get_doc("Integration Request", integration_req_ids[0])
@@ -99,13 +114,17 @@
pos_invoice.delete()
def test_processing_of_multiple_callback_payload(self):
- mpesa_account = frappe.db.get_value("Payment Gateway Account", {"payment_gateway": 'Mpesa-Payment'}, "payment_account")
+ mpesa_account = frappe.db.get_value(
+ "Payment Gateway Account", {"payment_gateway": "Mpesa-Payment"}, "payment_account"
+ )
frappe.db.set_value("Account", mpesa_account, "account_currency", "KES")
frappe.db.set_value("Mpesa Settings", "Payment", "transaction_limit", "500")
frappe.db.set_value("Customer", "_Test Customer", "default_currency", "KES")
pos_invoice = create_pos_invoice(do_not_submit=1)
- pos_invoice.append("payments", {'mode_of_payment': 'Mpesa-Payment', 'account': mpesa_account, 'amount': 1000})
+ pos_invoice.append(
+ "payments", {"mode_of_payment": "Mpesa-Payment", "account": mpesa_account, "amount": 1000}
+ )
pos_invoice.contact_mobile = "093456543894"
pos_invoice.currency = "KES"
pos_invoice.save()
@@ -115,10 +134,14 @@
self.assertEqual(pr.payment_gateway, "Mpesa-Payment")
# submitting payment request creates integration requests with random id
- integration_req_ids = frappe.get_all("Integration Request", filters={
- 'reference_doctype': pr.doctype,
- 'reference_docname': pr.name,
- }, pluck="name")
+ integration_req_ids = frappe.get_all(
+ "Integration Request",
+ filters={
+ "reference_doctype": pr.doctype,
+ "reference_docname": pr.name,
+ },
+ pluck="name",
+ )
# create random receipt nos and send it as response to callback handler
mpesa_receipt_numbers = [frappe.utils.random_string(5) for d in integration_req_ids]
@@ -128,7 +151,7 @@
callback_response = get_payment_callback_payload(
Amount=500,
CheckoutRequestID=integration_req_ids[i],
- MpesaReceiptNumber=mpesa_receipt_numbers[i]
+ MpesaReceiptNumber=mpesa_receipt_numbers[i],
)
# handle response manually
verify_transaction(**callback_response)
@@ -139,7 +162,7 @@
# check receipt number once all the integration requests are completed
pos_invoice.reload()
- self.assertEqual(pos_invoice.mpesa_receipt_number, ', '.join(mpesa_receipt_numbers))
+ self.assertEqual(pos_invoice.mpesa_receipt_number, ", ".join(mpesa_receipt_numbers))
frappe.db.set_value("Customer", "_Test Customer", "default_currency", "")
[d.delete() for d in integration_requests]
@@ -149,13 +172,17 @@
pos_invoice.delete()
def test_processing_of_only_one_succes_callback_payload(self):
- mpesa_account = frappe.db.get_value("Payment Gateway Account", {"payment_gateway": 'Mpesa-Payment'}, "payment_account")
+ mpesa_account = frappe.db.get_value(
+ "Payment Gateway Account", {"payment_gateway": "Mpesa-Payment"}, "payment_account"
+ )
frappe.db.set_value("Account", mpesa_account, "account_currency", "KES")
frappe.db.set_value("Mpesa Settings", "Payment", "transaction_limit", "500")
frappe.db.set_value("Customer", "_Test Customer", "default_currency", "KES")
pos_invoice = create_pos_invoice(do_not_submit=1)
- pos_invoice.append("payments", {'mode_of_payment': 'Mpesa-Payment', 'account': mpesa_account, 'amount': 1000})
+ pos_invoice.append(
+ "payments", {"mode_of_payment": "Mpesa-Payment", "account": mpesa_account, "amount": 1000}
+ )
pos_invoice.contact_mobile = "093456543894"
pos_invoice.currency = "KES"
pos_invoice.save()
@@ -165,10 +192,14 @@
self.assertEqual(pr.payment_gateway, "Mpesa-Payment")
# submitting payment request creates integration requests with random id
- integration_req_ids = frappe.get_all("Integration Request", filters={
- 'reference_doctype': pr.doctype,
- 'reference_docname': pr.name,
- }, pluck="name")
+ integration_req_ids = frappe.get_all(
+ "Integration Request",
+ filters={
+ "reference_doctype": pr.doctype,
+ "reference_docname": pr.name,
+ },
+ pluck="name",
+ )
# create random receipt nos and send it as response to callback handler
mpesa_receipt_numbers = [frappe.utils.random_string(5) for d in integration_req_ids]
@@ -176,7 +207,7 @@
callback_response = get_payment_callback_payload(
Amount=500,
CheckoutRequestID=integration_req_ids[0],
- MpesaReceiptNumber=mpesa_receipt_numbers[0]
+ MpesaReceiptNumber=mpesa_receipt_numbers[0],
)
# handle response manually
verify_transaction(**callback_response)
@@ -188,11 +219,15 @@
# second integration request fails
# now retrying payment request should make only one integration request again
pr = pos_invoice.create_payment_request()
- new_integration_req_ids = frappe.get_all("Integration Request", filters={
- 'reference_doctype': pr.doctype,
- 'reference_docname': pr.name,
- 'name': ['not in', integration_req_ids]
- }, pluck="name")
+ new_integration_req_ids = frappe.get_all(
+ "Integration Request",
+ filters={
+ "reference_doctype": pr.doctype,
+ "reference_docname": pr.name,
+ "name": ["not in", integration_req_ids],
+ },
+ pluck="name",
+ )
self.assertEqual(len(new_integration_req_ids), 1)
@@ -203,94 +238,56 @@
pr.delete()
pos_invoice.delete()
+
def create_mpesa_settings(payment_gateway_name="Express"):
if frappe.db.exists("Mpesa Settings", payment_gateway_name):
return frappe.get_doc("Mpesa Settings", payment_gateway_name)
- doc = frappe.get_doc(dict( #nosec
- doctype="Mpesa Settings",
- sandbox=1,
- payment_gateway_name=payment_gateway_name,
- consumer_key="5sMu9LVI1oS3oBGPJfh3JyvLHwZOdTKn",
- consumer_secret="VI1oS3oBGPJfh3JyvLHw",
- online_passkey="LVI1oS3oBGPJfh3JyvLHwZOd",
- till_number="174379"
- ))
+ doc = frappe.get_doc(
+ dict( # nosec
+ doctype="Mpesa Settings",
+ sandbox=1,
+ payment_gateway_name=payment_gateway_name,
+ consumer_key="5sMu9LVI1oS3oBGPJfh3JyvLHwZOdTKn",
+ consumer_secret="VI1oS3oBGPJfh3JyvLHw",
+ online_passkey="LVI1oS3oBGPJfh3JyvLHwZOd",
+ till_number="174379",
+ )
+ )
doc.insert(ignore_permissions=True)
return doc
+
def get_test_account_balance_response():
"""Response received after calling the account balance API."""
return {
- "ResultType":0,
- "ResultCode":0,
- "ResultDesc":"The service request has been accepted successfully.",
- "OriginatorConversationID":"10816-694520-2",
- "ConversationID":"AG_20200927_00007cdb1f9fb6494315",
- "TransactionID":"LGR0000000",
- "ResultParameters":{
- "ResultParameter":[
- {
- "Key":"ReceiptNo",
- "Value":"LGR919G2AV"
- },
- {
- "Key":"Conversation ID",
- "Value":"AG_20170727_00004492b1b6d0078fbe"
- },
- {
- "Key":"FinalisedTime",
- "Value":20170727101415
- },
- {
- "Key":"Amount",
- "Value":10
- },
- {
- "Key":"TransactionStatus",
- "Value":"Completed"
- },
- {
- "Key":"ReasonType",
- "Value":"Salary Payment via API"
- },
- {
- "Key":"TransactionReason"
- },
- {
- "Key":"DebitPartyCharges",
- "Value":"Fee For B2C Payment|KES|33.00"
- },
- {
- "Key":"DebitAccountType",
- "Value":"Utility Account"
- },
- {
- "Key":"InitiatedTime",
- "Value":20170727101415
- },
- {
- "Key":"Originator Conversation ID",
- "Value":"19455-773836-1"
- },
- {
- "Key":"CreditPartyName",
- "Value":"254708374149 - John Doe"
- },
- {
- "Key":"DebitPartyName",
- "Value":"600134 - Safaricom157"
- }
- ]
- },
- "ReferenceData":{
- "ReferenceItem":{
- "Key":"Occasion",
- "Value":"aaaa"
+ "ResultType": 0,
+ "ResultCode": 0,
+ "ResultDesc": "The service request has been accepted successfully.",
+ "OriginatorConversationID": "10816-694520-2",
+ "ConversationID": "AG_20200927_00007cdb1f9fb6494315",
+ "TransactionID": "LGR0000000",
+ "ResultParameters": {
+ "ResultParameter": [
+ {"Key": "ReceiptNo", "Value": "LGR919G2AV"},
+ {"Key": "Conversation ID", "Value": "AG_20170727_00004492b1b6d0078fbe"},
+ {"Key": "FinalisedTime", "Value": 20170727101415},
+ {"Key": "Amount", "Value": 10},
+ {"Key": "TransactionStatus", "Value": "Completed"},
+ {"Key": "ReasonType", "Value": "Salary Payment via API"},
+ {"Key": "TransactionReason"},
+ {"Key": "DebitPartyCharges", "Value": "Fee For B2C Payment|KES|33.00"},
+ {"Key": "DebitAccountType", "Value": "Utility Account"},
+ {"Key": "InitiatedTime", "Value": 20170727101415},
+ {"Key": "Originator Conversation ID", "Value": "19455-773836-1"},
+ {"Key": "CreditPartyName", "Value": "254708374149 - John Doe"},
+ {"Key": "DebitPartyName", "Value": "600134 - Safaricom157"},
+ ]
+ },
+ "ReferenceData": {"ReferenceItem": {"Key": "Occasion", "Value": "aaaa"}},
}
- }
- }
+
def get_payment_request_response_payload(Amount=500):
"""Response received after successfully calling the stk push process request API."""
@@ -304,40 +301,44 @@
"ResultDesc": "The service request is processed successfully.",
"CallbackMetadata": {
"Item": [
- { "Name": "Amount", "Value": Amount },
- { "Name": "MpesaReceiptNumber", "Value": "LGR7OWQX0R" },
- { "Name": "TransactionDate", "Value": 20201006113336 },
- { "Name": "PhoneNumber", "Value": 254723575670 }
+ {"Name": "Amount", "Value": Amount},
+ {"Name": "MpesaReceiptNumber", "Value": "LGR7OWQX0R"},
+ {"Name": "TransactionDate", "Value": 20201006113336},
+ {"Name": "PhoneNumber", "Value": 254723575670},
]
- }
+ },
}
-def get_payment_callback_payload(Amount=500, CheckoutRequestID="ws_CO_061020201133231972", MpesaReceiptNumber="LGR7OWQX0R"):
+
+def get_payment_callback_payload(
+ Amount=500, CheckoutRequestID="ws_CO_061020201133231972", MpesaReceiptNumber="LGR7OWQX0R"
+):
"""Response received from the server as callback after calling the stkpush process request API."""
return {
- "Body":{
- "stkCallback":{
- "MerchantRequestID":"19465-780693-1",
- "CheckoutRequestID":CheckoutRequestID,
- "ResultCode":0,
- "ResultDesc":"The service request is processed successfully.",
- "CallbackMetadata":{
- "Item":[
- { "Name":"Amount", "Value":Amount },
- { "Name":"MpesaReceiptNumber", "Value":MpesaReceiptNumber },
- { "Name":"Balance" },
- { "Name":"TransactionDate", "Value":20170727154800 },
- { "Name":"PhoneNumber", "Value":254721566839 }
+ "Body": {
+ "stkCallback": {
+ "MerchantRequestID": "19465-780693-1",
+ "CheckoutRequestID": CheckoutRequestID,
+ "ResultCode": 0,
+ "ResultDesc": "The service request is processed successfully.",
+ "CallbackMetadata": {
+ "Item": [
+ {"Name": "Amount", "Value": Amount},
+ {"Name": "MpesaReceiptNumber", "Value": MpesaReceiptNumber},
+ {"Name": "Balance"},
+ {"Name": "TransactionDate", "Value": 20170727154800},
+ {"Name": "PhoneNumber", "Value": 254721566839},
]
- }
+ },
}
}
}
+
def get_account_balance_callback_payload():
"""Response received from the server as callback after calling the account balance API."""
return {
- "Result":{
+ "Result": {
"ResultType": 0,
"ResultCode": 0,
"ResultDesc": "The service request is processed successfully.",
@@ -346,18 +347,15 @@
"TransactionID": "OIR0000000",
"ResultParameters": {
"ResultParameter": [
- {
- "Key": "AccountBalance",
- "Value": "Working Account|KES|481000.00|481000.00|0.00|0.00"
- },
- { "Key": "BOCompletedTime", "Value": 20200927234123 }
+ {"Key": "AccountBalance", "Value": "Working Account|KES|481000.00|481000.00|0.00|0.00"},
+ {"Key": "BOCompletedTime", "Value": 20200927234123},
]
},
"ReferenceData": {
"ReferenceItem": {
"Key": "QueueTimeoutURL",
- "Value": "https://internalsandbox.safaricom.co.ke/mpesa/abresults/v1/submit"
+ "Value": "https://internalsandbox.safaricom.co.ke/mpesa/abresults/v1/submit",
}
- }
+ },
}
}
diff --git a/erpnext/erpnext_integrations/doctype/plaid_settings/plaid_connector.py b/erpnext/erpnext_integrations/doctype/plaid_settings/plaid_connector.py
index 0b552f9..625dd31 100644
--- a/erpnext/erpnext_integrations/doctype/plaid_settings/plaid_connector.py
+++ b/erpnext/erpnext_integrations/doctype/plaid_settings/plaid_connector.py
@@ -8,7 +8,7 @@
from plaid.errors import APIError, InvalidRequestError, ItemError
-class PlaidConnector():
+class PlaidConnector:
def __init__(self, access_token=None):
self.access_token = access_token
self.settings = frappe.get_single("Plaid Settings")
@@ -18,7 +18,7 @@
client_id=self.settings.plaid_client_id,
secret=self.settings.get_password("plaid_secret"),
environment=self.settings.plaid_env,
- api_version="2020-09-14"
+ api_version="2020-09-14",
)
def get_access_token(self, public_token):
@@ -29,25 +29,29 @@
return access_token
def get_token_request(self, update_mode=False):
- country_codes = ["US", "CA", "FR", "IE", "NL", "ES", "GB"] if self.settings.enable_european_access else ["US", "CA"]
+ country_codes = (
+ ["US", "CA", "FR", "IE", "NL", "ES", "GB"]
+ if self.settings.enable_european_access
+ else ["US", "CA"]
+ )
args = {
"client_name": self.client_name,
# only allow Plaid-supported languages and countries (LAST: Sep-19-2020)
"language": frappe.local.lang if frappe.local.lang in ["en", "fr", "es", "nl"] else "en",
"country_codes": country_codes,
- "user": {
- "client_user_id": frappe.generate_hash(frappe.session.user, length=32)
- }
+ "user": {"client_user_id": frappe.generate_hash(frappe.session.user, length=32)},
}
if update_mode:
args["access_token"] = self.access_token
else:
- args.update({
- "client_id": self.settings.plaid_client_id,
- "secret": self.settings.plaid_secret,
- "products": self.products,
- })
+ args.update(
+ {
+ "client_id": self.settings.plaid_client_id,
+ "secret": self.settings.plaid_secret,
+ "products": self.products,
+ }
+ )
return args
@@ -82,11 +86,7 @@
def get_transactions(self, start_date, end_date, account_id=None):
self.auth()
- kwargs = dict(
- access_token=self.access_token,
- start_date=start_date,
- end_date=end_date
- )
+ kwargs = dict(access_token=self.access_token, start_date=start_date, end_date=end_date)
if account_id:
kwargs.update(dict(account_ids=[account_id]))
@@ -94,7 +94,9 @@
response = self.client.Transactions.get(**kwargs)
transactions = response["transactions"]
while len(transactions) < response["total_transactions"]:
- response = self.client.Transactions.get(self.access_token, start_date=start_date, end_date=end_date, offset=len(transactions))
+ response = self.client.Transactions.get(
+ self.access_token, start_date=start_date, end_date=end_date, offset=len(transactions)
+ )
transactions.extend(response["transactions"])
return transactions
except ItemError as e:
diff --git a/erpnext/erpnext_integrations/doctype/plaid_settings/plaid_settings.py b/erpnext/erpnext_integrations/doctype/plaid_settings/plaid_settings.py
index 7e6f146..ce65f6c 100644
--- a/erpnext/erpnext_integrations/doctype/plaid_settings/plaid_settings.py
+++ b/erpnext/erpnext_integrations/doctype/plaid_settings/plaid_settings.py
@@ -29,7 +29,7 @@
return {
"plaid_env": plaid_settings.plaid_env,
"link_token": plaid_settings.get_link_token(),
- "client_name": frappe.local.site
+ "client_name": frappe.local.site,
}
return "disabled"
@@ -45,14 +45,16 @@
if not frappe.db.exists("Bank", response["institution"]["name"]):
try:
- bank = frappe.get_doc({
- "doctype": "Bank",
- "bank_name": response["institution"]["name"],
- "plaid_access_token": access_token
- })
+ bank = frappe.get_doc(
+ {
+ "doctype": "Bank",
+ "bank_name": response["institution"]["name"],
+ "plaid_access_token": access_token,
+ }
+ )
bank.insert()
except Exception:
- frappe.log_error(frappe.get_traceback(), title=_('Plaid Link Error'))
+ frappe.log_error(frappe.get_traceback(), title=_("Plaid Link Error"))
else:
bank = frappe.get_doc("Bank", response["institution"]["name"])
bank.plaid_access_token = access_token
@@ -89,65 +91,71 @@
if not existing_bank_account:
try:
- new_account = frappe.get_doc({
- "doctype": "Bank Account",
- "bank": bank["bank_name"],
- "account": default_gl_account.account,
- "account_name": account["name"],
- "account_type": account.get("type", ""),
- "account_subtype": account.get("subtype", ""),
- "mask": account.get("mask", ""),
- "integration_id": account["id"],
- "is_company_account": 1,
- "company": company
- })
+ new_account = frappe.get_doc(
+ {
+ "doctype": "Bank Account",
+ "bank": bank["bank_name"],
+ "account": default_gl_account.account,
+ "account_name": account["name"],
+ "account_type": account.get("type", ""),
+ "account_subtype": account.get("subtype", ""),
+ "mask": account.get("mask", ""),
+ "integration_id": account["id"],
+ "is_company_account": 1,
+ "company": company,
+ }
+ )
new_account.insert()
result.append(new_account.name)
except frappe.UniqueValidationError:
- frappe.msgprint(_("Bank account {0} already exists and could not be created again").format(account["name"]))
+ frappe.msgprint(
+ _("Bank account {0} already exists and could not be created again").format(account["name"])
+ )
except Exception:
frappe.log_error(frappe.get_traceback(), title=_("Plaid Link Error"))
- frappe.throw(_("There was an error creating Bank Account while linking with Plaid."),
- title=_("Plaid Link Failed"))
+ frappe.throw(
+ _("There was an error creating Bank Account while linking with Plaid."),
+ title=_("Plaid Link Failed"),
+ )
else:
try:
- existing_account = frappe.get_doc('Bank Account', existing_bank_account)
- existing_account.update({
- "bank": bank["bank_name"],
- "account_name": account["name"],
- "account_type": account.get("type", ""),
- "account_subtype": account.get("subtype", ""),
- "mask": account.get("mask", ""),
- "integration_id": account["id"]
- })
+ existing_account = frappe.get_doc("Bank Account", existing_bank_account)
+ existing_account.update(
+ {
+ "bank": bank["bank_name"],
+ "account_name": account["name"],
+ "account_type": account.get("type", ""),
+ "account_subtype": account.get("subtype", ""),
+ "mask": account.get("mask", ""),
+ "integration_id": account["id"],
+ }
+ )
existing_account.save()
result.append(existing_bank_account)
except Exception:
frappe.log_error(frappe.get_traceback(), title=_("Plaid Link Error"))
- frappe.throw(_("There was an error updating Bank Account {} while linking with Plaid.").format(
- existing_bank_account), title=_("Plaid Link Failed"))
+ frappe.throw(
+ _("There was an error updating Bank Account {} while linking with Plaid.").format(
+ existing_bank_account
+ ),
+ title=_("Plaid Link Failed"),
+ )
return result
def add_account_type(account_type):
try:
- frappe.get_doc({
- "doctype": "Bank Account Type",
- "account_type": account_type
- }).insert()
+ frappe.get_doc({"doctype": "Bank Account Type", "account_type": account_type}).insert()
except Exception:
frappe.throw(frappe.get_traceback())
def add_account_subtype(account_subtype):
try:
- frappe.get_doc({
- "doctype": "Bank Account Subtype",
- "account_subtype": account_subtype
- }).insert()
+ frappe.get_doc({"doctype": "Bank Account Subtype", "account_subtype": account_subtype}).insert()
except Exception:
frappe.throw(frappe.get_traceback())
@@ -164,19 +172,26 @@
end_date = formatdate(today(), "YYYY-MM-dd")
try:
- transactions = get_transactions(bank=bank, bank_account=bank_account, start_date=start_date, end_date=end_date)
+ transactions = get_transactions(
+ bank=bank, bank_account=bank_account, start_date=start_date, end_date=end_date
+ )
result = []
for transaction in reversed(transactions):
result += new_bank_transaction(transaction)
if result:
- last_transaction_date = frappe.db.get_value('Bank Transaction', result.pop(), 'date')
+ last_transaction_date = frappe.db.get_value("Bank Transaction", result.pop(), "date")
- frappe.logger().info("Plaid added {} new Bank Transactions from '{}' between {} and {}".format(
- len(result), bank_account, start_date, end_date))
+ frappe.logger().info(
+ "Plaid added {} new Bank Transactions from '{}' between {} and {}".format(
+ len(result), bank_account, start_date, end_date
+ )
+ )
- frappe.db.set_value("Bank Account", bank_account, "last_integration_date", last_transaction_date)
+ frappe.db.set_value(
+ "Bank Account", bank_account, "last_integration_date", last_transaction_date
+ )
except Exception:
frappe.log_error(frappe.get_traceback(), _("Plaid transactions sync error"))
@@ -185,7 +200,9 @@
access_token = None
if bank_account:
- related_bank = frappe.db.get_values("Bank Account", bank_account, ["bank", "integration_id"], as_dict=True)
+ related_bank = frappe.db.get_values(
+ "Bank Account", bank_account, ["bank", "integration_id"], as_dict=True
+ )
access_token = frappe.db.get_value("Bank", related_bank[0].bank, "plaid_access_token")
account_id = related_bank[0].integration_id
else:
@@ -196,7 +213,9 @@
transactions = []
try:
- transactions = plaid.get_transactions(start_date=start_date, end_date=end_date, account_id=account_id)
+ transactions = plaid.get_transactions(
+ start_date=start_date, end_date=end_date, account_id=account_id
+ )
except ItemError as e:
if e.code == "ITEM_LOGIN_REQUIRED":
msg = _("There was an error syncing transactions.") + " "
@@ -229,18 +248,20 @@
if not frappe.db.exists("Bank Transaction", dict(transaction_id=transaction["transaction_id"])):
try:
- new_transaction = frappe.get_doc({
- "doctype": "Bank Transaction",
- "date": getdate(transaction["date"]),
- "status": status,
- "bank_account": bank_account,
- "deposit": debit,
- "withdrawal": credit,
- "currency": transaction["iso_currency_code"],
- "transaction_id": transaction["transaction_id"],
- "reference_number": transaction["payment_meta"]["reference_number"],
- "description": transaction["name"]
- })
+ new_transaction = frappe.get_doc(
+ {
+ "doctype": "Bank Transaction",
+ "date": getdate(transaction["date"]),
+ "status": status,
+ "bank_account": bank_account,
+ "deposit": debit,
+ "withdrawal": credit,
+ "currency": transaction["iso_currency_code"],
+ "transaction_id": transaction["transaction_id"],
+ "reference_number": transaction["payment_meta"]["reference_number"],
+ "description": transaction["name"],
+ }
+ )
new_transaction.insert()
new_transaction.submit()
@@ -250,7 +271,7 @@
result.append(new_transaction.name)
except Exception:
- frappe.throw(title=_('Bank transaction creation error'))
+ frappe.throw(title=_("Bank transaction creation error"))
return result
@@ -260,19 +281,21 @@
if settings.enabled == 1 and settings.automatic_sync == 1:
enqueue_synchronization()
+
@frappe.whitelist()
def enqueue_synchronization():
- plaid_accounts = frappe.get_all("Bank Account",
- filters={"integration_id": ["!=", ""]},
- fields=["name", "bank"])
+ plaid_accounts = frappe.get_all(
+ "Bank Account", filters={"integration_id": ["!=", ""]}, fields=["name", "bank"]
+ )
for plaid_account in plaid_accounts:
frappe.enqueue(
"erpnext.erpnext_integrations.doctype.plaid_settings.plaid_settings.sync_transactions",
bank=plaid_account.bank,
- bank_account=plaid_account.name
+ bank_account=plaid_account.name,
)
+
@frappe.whitelist()
def get_link_token_for_update(access_token):
plaid = PlaidConnector(access_token)
diff --git a/erpnext/erpnext_integrations/doctype/plaid_settings/test_plaid_settings.py b/erpnext/erpnext_integrations/doctype/plaid_settings/test_plaid_settings.py
index 535d7fa..e8dc3e2 100644
--- a/erpnext/erpnext_integrations/doctype/plaid_settings/test_plaid_settings.py
+++ b/erpnext/erpnext_integrations/doctype/plaid_settings/test_plaid_settings.py
@@ -45,111 +45,110 @@
def test_default_bank_account(self):
if not frappe.db.exists("Bank", "Citi"):
- frappe.get_doc({
- "doctype": "Bank",
- "bank_name": "Citi"
- }).insert()
+ frappe.get_doc({"doctype": "Bank", "bank_name": "Citi"}).insert()
bank_accounts = {
- 'account': {
- 'subtype': 'checking',
- 'mask': '0000',
- 'type': 'depository',
- 'id': '6GbM6RRQgdfy3lAqGz4JUnpmR948WZFg8DjQK',
- 'name': 'Plaid Checking'
+ "account": {
+ "subtype": "checking",
+ "mask": "0000",
+ "type": "depository",
+ "id": "6GbM6RRQgdfy3lAqGz4JUnpmR948WZFg8DjQK",
+ "name": "Plaid Checking",
},
- 'account_id': '6GbM6RRQgdfy3lAqGz4JUnpmR948WZFg8DjQK',
- 'link_session_id': 'db673d75-61aa-442a-864f-9b3f174f3725',
- 'accounts': [{
- 'type': 'depository',
- 'subtype': 'checking',
- 'mask': '0000',
- 'id': '6GbM6RRQgdfy3lAqGz4JUnpmR948WZFg8DjQK',
- 'name': 'Plaid Checking'
- }],
- 'institution': {
- 'institution_id': 'ins_6',
- 'name': 'Citi'
- }
+ "account_id": "6GbM6RRQgdfy3lAqGz4JUnpmR948WZFg8DjQK",
+ "link_session_id": "db673d75-61aa-442a-864f-9b3f174f3725",
+ "accounts": [
+ {
+ "type": "depository",
+ "subtype": "checking",
+ "mask": "0000",
+ "id": "6GbM6RRQgdfy3lAqGz4JUnpmR948WZFg8DjQK",
+ "name": "Plaid Checking",
+ }
+ ],
+ "institution": {"institution_id": "ins_6", "name": "Citi"},
}
bank = json.dumps(frappe.get_doc("Bank", "Citi").as_dict(), default=json_handler)
- company = frappe.db.get_single_value('Global Defaults', 'default_company')
+ company = frappe.db.get_single_value("Global Defaults", "default_company")
frappe.db.set_value("Company", company, "default_bank_account", None)
- self.assertRaises(frappe.ValidationError, add_bank_accounts, response=bank_accounts, bank=bank, company=company)
+ self.assertRaises(
+ frappe.ValidationError, add_bank_accounts, response=bank_accounts, bank=bank, company=company
+ )
def test_new_transaction(self):
if not frappe.db.exists("Bank", "Citi"):
- frappe.get_doc({
- "doctype": "Bank",
- "bank_name": "Citi"
- }).insert()
+ frappe.get_doc({"doctype": "Bank", "bank_name": "Citi"}).insert()
bank_accounts = {
- 'account': {
- 'subtype': 'checking',
- 'mask': '0000',
- 'type': 'depository',
- 'id': '6GbM6RRQgdfy3lAqGz4JUnpmR948WZFg8DjQK',
- 'name': 'Plaid Checking'
+ "account": {
+ "subtype": "checking",
+ "mask": "0000",
+ "type": "depository",
+ "id": "6GbM6RRQgdfy3lAqGz4JUnpmR948WZFg8DjQK",
+ "name": "Plaid Checking",
},
- 'account_id': '6GbM6RRQgdfy3lAqGz4JUnpmR948WZFg8DjQK',
- 'link_session_id': 'db673d75-61aa-442a-864f-9b3f174f3725',
- 'accounts': [{
- 'type': 'depository',
- 'subtype': 'checking',
- 'mask': '0000',
- 'id': '6GbM6RRQgdfy3lAqGz4JUnpmR948WZFg8DjQK',
- 'name': 'Plaid Checking'
- }],
- 'institution': {
- 'institution_id': 'ins_6',
- 'name': 'Citi'
- }
+ "account_id": "6GbM6RRQgdfy3lAqGz4JUnpmR948WZFg8DjQK",
+ "link_session_id": "db673d75-61aa-442a-864f-9b3f174f3725",
+ "accounts": [
+ {
+ "type": "depository",
+ "subtype": "checking",
+ "mask": "0000",
+ "id": "6GbM6RRQgdfy3lAqGz4JUnpmR948WZFg8DjQK",
+ "name": "Plaid Checking",
+ }
+ ],
+ "institution": {"institution_id": "ins_6", "name": "Citi"},
}
bank = json.dumps(frappe.get_doc("Bank", "Citi").as_dict(), default=json_handler)
- company = frappe.db.get_single_value('Global Defaults', 'default_company')
+ company = frappe.db.get_single_value("Global Defaults", "default_company")
if frappe.db.get_value("Company", company, "default_bank_account") is None:
- frappe.db.set_value("Company", company, "default_bank_account", get_default_bank_cash_account(company, "Cash").get("account"))
+ frappe.db.set_value(
+ "Company",
+ company,
+ "default_bank_account",
+ get_default_bank_cash_account(company, "Cash").get("account"),
+ )
add_bank_accounts(bank_accounts, bank, company)
transactions = {
- 'account_owner': None,
- 'category': ['Food and Drink', 'Restaurants'],
- 'account_id': 'b4Jkp1LJDZiPgojpr1ansXJrj5Q6w9fVmv6ov',
- 'pending_transaction_id': None,
- 'transaction_id': 'x374xPa7DvUewqlR5mjNIeGK8r8rl3Sn647LM',
- 'unofficial_currency_code': None,
- 'name': 'INTRST PYMNT',
- 'transaction_type': 'place',
- 'amount': -4.22,
- 'location': {
- 'city': None,
- 'zip': None,
- 'store_number': None,
- 'lon': None,
- 'state': None,
- 'address': None,
- 'lat': None
+ "account_owner": None,
+ "category": ["Food and Drink", "Restaurants"],
+ "account_id": "b4Jkp1LJDZiPgojpr1ansXJrj5Q6w9fVmv6ov",
+ "pending_transaction_id": None,
+ "transaction_id": "x374xPa7DvUewqlR5mjNIeGK8r8rl3Sn647LM",
+ "unofficial_currency_code": None,
+ "name": "INTRST PYMNT",
+ "transaction_type": "place",
+ "amount": -4.22,
+ "location": {
+ "city": None,
+ "zip": None,
+ "store_number": None,
+ "lon": None,
+ "state": None,
+ "address": None,
+ "lat": None,
},
- 'payment_meta': {
- 'reference_number': None,
- 'payer': None,
- 'payment_method': None,
- 'reason': None,
- 'payee': None,
- 'ppd_id': None,
- 'payment_processor': None,
- 'by_order_of': None
+ "payment_meta": {
+ "reference_number": None,
+ "payer": None,
+ "payment_method": None,
+ "reason": None,
+ "payee": None,
+ "ppd_id": None,
+ "payment_processor": None,
+ "by_order_of": None,
},
- 'date': '2017-12-22',
- 'category_id': '13005000',
- 'pending': False,
- 'iso_currency_code': 'USD'
+ "date": "2017-12-22",
+ "category_id": "13005000",
+ "pending": False,
+ "iso_currency_code": "USD",
}
new_bank_transaction(transactions)
diff --git a/erpnext/erpnext_integrations/doctype/quickbooks_migrator/quickbooks_migrator.py b/erpnext/erpnext_integrations/doctype/quickbooks_migrator/quickbooks_migrator.py
index 5de5682..b93c5c4 100644
--- a/erpnext/erpnext_integrations/doctype/quickbooks_migrator/quickbooks_migrator.py
+++ b/erpnext/erpnext_integrations/doctype/quickbooks_migrator/quickbooks_migrator.py
@@ -37,30 +37,27 @@
def __init__(self, *args, **kwargs):
super(QuickBooksMigrator, self).__init__(*args, **kwargs)
self.oauth = OAuth2Session(
- client_id=self.client_id,
- redirect_uri=self.redirect_url,
- scope=self.scope
+ client_id=self.client_id, redirect_uri=self.redirect_url, scope=self.scope
)
if not self.authorization_url and self.authorization_endpoint:
self.authorization_url = self.oauth.authorization_url(self.authorization_endpoint)[0]
-
def on_update(self):
if self.company:
# We need a Cost Center corresponding to the selected erpnext Company
- self.default_cost_center = frappe.db.get_value('Company', self.company, 'cost_center')
- company_warehouses = frappe.get_all('Warehouse', filters={"company": self.company, "is_group": 0})
+ self.default_cost_center = frappe.db.get_value("Company", self.company, "cost_center")
+ company_warehouses = frappe.get_all(
+ "Warehouse", filters={"company": self.company, "is_group": 0}
+ )
if company_warehouses:
self.default_warehouse = company_warehouses[0].name
if self.authorization_endpoint:
self.authorization_url = self.oauth.authorization_url(self.authorization_endpoint)[0]
-
@frappe.whitelist()
def migrate(self):
frappe.enqueue_doc("QuickBooks Migrator", "QuickBooks Migrator", "_migrate", queue="long")
-
def _migrate(self):
try:
self.set_indicator("In Progress")
@@ -86,18 +83,33 @@
# Following entities are directly available from API
# Invoice can be an exception sometimes though (as explained above).
entities_for_normal_transform = [
- "Customer", "Item", "Vendor",
+ "Customer",
+ "Item",
+ "Vendor",
"Preferences",
- "JournalEntry", "Purchase", "Deposit",
- "Invoice", "CreditMemo", "SalesReceipt", "RefundReceipt",
- "Bill", "VendorCredit",
- "Payment", "BillPayment",
+ "JournalEntry",
+ "Purchase",
+ "Deposit",
+ "Invoice",
+ "CreditMemo",
+ "SalesReceipt",
+ "RefundReceipt",
+ "Bill",
+ "VendorCredit",
+ "Payment",
+ "BillPayment",
]
for entity in entities_for_normal_transform:
self._migrate_entries(entity)
# Following entries are not available directly from API, Need to be regenrated from GeneralLedger Report
- entities_for_gl_transform = ["Advance Payment", "Tax Payment", "Sales Tax Payment", "Purchase Tax Payment", "Inventory Qty Adjust"]
+ entities_for_gl_transform = [
+ "Advance Payment",
+ "Tax Payment",
+ "Sales Tax Payment",
+ "Purchase Tax Payment",
+ "Inventory Qty Adjust",
+ ]
for entity in entities_for_gl_transform:
self._migrate_entries_from_gl(entity)
self.set_indicator("Complete")
@@ -107,33 +119,37 @@
frappe.db.commit()
-
def get_tokens(self):
token = self.oauth.fetch_token(
- token_url=self.token_endpoint,
- client_secret=self.client_secret,
- code=self.code
+ token_url=self.token_endpoint, client_secret=self.client_secret, code=self.code
)
self.access_token = token["access_token"]
self.refresh_token = token["refresh_token"]
self.save()
-
def _refresh_tokens(self):
token = self.oauth.refresh_token(
token_url=self.token_endpoint,
client_id=self.client_id,
refresh_token=self.refresh_token,
client_secret=self.client_secret,
- code=self.code
+ code=self.code,
)
self.access_token = token["access_token"]
self.refresh_token = token["refresh_token"]
self.save()
-
def _make_custom_fields(self):
- doctypes_for_quickbooks_id_field = ["Account", "Customer", "Address", "Item", "Supplier", "Sales Invoice", "Journal Entry", "Purchase Invoice"]
+ doctypes_for_quickbooks_id_field = [
+ "Account",
+ "Customer",
+ "Address",
+ "Item",
+ "Supplier",
+ "Sales Invoice",
+ "Journal Entry",
+ "Purchase Invoice",
+ ]
for doctype in doctypes_for_quickbooks_id_field:
self._make_custom_quickbooks_id_field(doctype)
@@ -143,53 +159,60 @@
frappe.db.commit()
-
def _make_custom_quickbooks_id_field(self, doctype):
if not frappe.get_meta(doctype).has_field("quickbooks_id"):
- frappe.get_doc({
- "doctype": "Custom Field",
- "label": "QuickBooks ID",
- "dt": doctype,
- "fieldname": "quickbooks_id",
- "fieldtype": "Data",
- }).insert()
-
+ frappe.get_doc(
+ {
+ "doctype": "Custom Field",
+ "label": "QuickBooks ID",
+ "dt": doctype,
+ "fieldname": "quickbooks_id",
+ "fieldtype": "Data",
+ }
+ ).insert()
def _make_custom_company_field(self, doctype):
if not frappe.get_meta(doctype).has_field("company"):
- frappe.get_doc({
- "doctype": "Custom Field",
- "label": "Company",
- "dt": doctype,
- "fieldname": "company",
- "fieldtype": "Link",
- "options": "Company",
- }).insert()
-
+ frappe.get_doc(
+ {
+ "doctype": "Custom Field",
+ "label": "Company",
+ "dt": doctype,
+ "fieldname": "company",
+ "fieldtype": "Link",
+ "options": "Company",
+ }
+ ).insert()
def _migrate_accounts(self):
self._make_root_accounts()
for entity in ["Account", "TaxRate", "TaxCode"]:
self._migrate_entries(entity)
-
def _make_root_accounts(self):
roots = ["Asset", "Equity", "Expense", "Liability", "Income"]
for root in roots:
try:
- if not frappe.db.exists({"doctype": "Account", "name": encode_company_abbr("{} - QB".format(root), self.company), "company": self.company}):
- frappe.get_doc({
+ if not frappe.db.exists(
+ {
"doctype": "Account",
- "account_name": "{} - QB".format(root),
- "root_type": root,
- "is_group": "1",
+ "name": encode_company_abbr("{} - QB".format(root), self.company),
"company": self.company,
- }).insert(ignore_mandatory=True)
+ }
+ ):
+ frappe.get_doc(
+ {
+ "doctype": "Account",
+ "account_name": "{} - QB".format(root),
+ "root_type": root,
+ "is_group": "1",
+ "company": self.company,
+ }
+ ).insert(ignore_mandatory=True)
except Exception as e:
self._log_error(e, root)
frappe.db.commit()
-
def _migrate_entries(self, entity):
try:
query_uri = "{}/company/{}/query".format(
@@ -198,22 +221,19 @@
)
max_result_count = 1000
# Count number of entries
- response = self._get(query_uri,
- params={
- "query": """SELECT COUNT(*) FROM {}""".format(entity)
- }
- )
+ response = self._get(query_uri, params={"query": """SELECT COUNT(*) FROM {}""".format(entity)})
entry_count = response.json()["QueryResponse"]["totalCount"]
# fetch pages and accumulate
entries = []
for start_position in range(1, entry_count + 1, max_result_count):
- response = self._get(query_uri,
+ response = self._get(
+ query_uri,
params={
"query": """SELECT * FROM {} STARTPOSITION {} MAXRESULTS {}""".format(
entity, start_position, max_result_count
)
- }
+ },
)
entries.extend(response.json()["QueryResponse"][entity])
entries = self._preprocess_entries(entity, entries)
@@ -221,16 +241,18 @@
except Exception as e:
self._log_error(e, response.text)
-
def _fetch_general_ledger(self):
try:
- query_uri = "{}/company/{}/reports/GeneralLedger".format(self.api_endpoint, self.quickbooks_company_id)
- response = self._get(query_uri,
+ query_uri = "{}/company/{}/reports/GeneralLedger".format(
+ self.api_endpoint, self.quickbooks_company_id
+ )
+ response = self._get(
+ query_uri,
params={
"columns": ",".join(["tx_date", "txn_type", "credit_amt", "debt_amt"]),
"date_macro": "All",
"minorversion": 3,
- }
+ },
)
self.gl_entries = {}
for section in response.json()["Rows"]["Row"]:
@@ -250,7 +272,6 @@
except Exception as e:
self._log_error(e, response.text)
-
def _create_fiscal_years(self):
try:
# Assumes that exactly one fiscal year has been created so far
@@ -258,10 +279,12 @@
from itertools import chain
from frappe.utils.data import add_years, getdate
- smallest_ledger_entry_date = getdate(min(entry["date"] for entry in chain(*self.gl_entries.values()) if entry["date"]))
- oldest_fiscal_year = frappe.get_all("Fiscal Year",
- fields=["year_start_date", "year_end_date"],
- order_by="year_start_date"
+
+ smallest_ledger_entry_date = getdate(
+ min(entry["date"] for entry in chain(*self.gl_entries.values()) if entry["date"])
+ )
+ oldest_fiscal_year = frappe.get_all(
+ "Fiscal Year", fields=["year_start_date", "year_end_date"], order_by="year_start_date"
)[0]
# Keep on creating fiscal years
# until smallest_ledger_entry_date is no longer smaller than the oldest fiscal year's start date
@@ -272,7 +295,9 @@
if new_fiscal_year.year_start_date.year == new_fiscal_year.year_end_date.year:
new_fiscal_year.year = new_fiscal_year.year_start_date.year
else:
- new_fiscal_year.year = "{}-{}".format(new_fiscal_year.year_start_date.year, new_fiscal_year.year_end_date.year)
+ new_fiscal_year.year = "{}-{}".format(
+ new_fiscal_year.year_start_date.year, new_fiscal_year.year_end_date.year
+ )
new_fiscal_year.save()
oldest_fiscal_year = new_fiscal_year
@@ -280,40 +305,30 @@
except Exception as e:
self._log_error(e)
-
def _migrate_entries_from_gl(self, entity):
if entity in self.general_ledger:
self._save_entries(entity, self.general_ledger[entity].values())
-
def _save_entries(self, entity, entries):
entity_method_map = {
"Account": self._save_account,
"TaxRate": self._save_tax_rate,
"TaxCode": self._save_tax_code,
-
"Preferences": self._save_preference,
-
"Customer": self._save_customer,
"Item": self._save_item,
"Vendor": self._save_vendor,
-
"Invoice": self._save_invoice,
"CreditMemo": self._save_credit_memo,
"SalesReceipt": self._save_sales_receipt,
"RefundReceipt": self._save_refund_receipt,
-
"JournalEntry": self._save_journal_entry,
-
"Bill": self._save_bill,
"VendorCredit": self._save_vendor_credit,
-
"Payment": self._save_payment,
"BillPayment": self._save_bill_payment,
-
"Purchase": self._save_purchase,
"Deposit": self._save_deposit,
-
"Advance Payment": self._save_advance_payment,
"Tax Payment": self._save_tax_payment,
"Sales Tax Payment": self._save_tax_payment,
@@ -322,11 +337,17 @@
}
total = len(entries)
for index, entry in enumerate(entries, start=1):
- self._publish({"event": "progress", "message": _("Saving {0}").format(entity), "count": index, "total": total})
+ self._publish(
+ {
+ "event": "progress",
+ "message": _("Saving {0}").format(entity),
+ "count": index,
+ "total": total,
+ }
+ )
entity_method_map[entity](entry)
frappe.db.commit()
-
def _preprocess_entries(self, entity, entries):
entity_method_map = {
"Account": self._preprocess_accounts,
@@ -338,7 +359,6 @@
entries = preprocessor(entries)
return entries
-
def _get_gl_entries_from_section(self, section, account=None):
if "Header" in section:
if "id" in section["Header"]["ColData"][0]:
@@ -359,19 +379,20 @@
for row in section["Rows"]["Row"]:
if row["type"] == "Data":
data = row["ColData"]
- entries.append({
- "account": account,
- "date": data[0]["value"],
- "type": data[1]["value"],
- "id": data[1].get("id"),
- "credit": frappe.utils.flt(data[2]["value"]),
- "debit": frappe.utils.flt(data[3]["value"]),
- })
+ entries.append(
+ {
+ "account": account,
+ "date": data[0]["value"],
+ "type": data[1]["value"],
+ "id": data[1].get("id"),
+ "credit": frappe.utils.flt(data[2]["value"]),
+ "debit": frappe.utils.flt(data[3]["value"]),
+ }
+ )
if row["type"] == "Section":
self._get_gl_entries_from_section(row, account)
self.gl_entries.setdefault(account, []).extend(entries)
-
def _preprocess_accounts(self, accounts):
self.accounts = {account["Name"]: account for account in accounts}
for account in accounts:
@@ -381,7 +402,6 @@
account["is_group"] = 0
return sorted(accounts, key=lambda account: int(account["Id"]))
-
def _save_account(self, account):
mapping = {
"Bank": "Asset",
@@ -389,24 +409,22 @@
"Fixed Asset": "Asset",
"Other Asset": "Asset",
"Accounts Receivable": "Asset",
-
"Equity": "Equity",
-
"Expense": "Expense",
"Other Expense": "Expense",
"Cost of Goods Sold": "Expense",
-
"Accounts Payable": "Liability",
"Credit Card": "Liability",
"Long Term Liability": "Liability",
"Other Current Liability": "Liability",
-
"Income": "Income",
"Other Income": "Income",
}
# Map Quickbooks Account Types to ERPNext root_accunts and and root_type
try:
- if not frappe.db.exists({"doctype": "Account", "quickbooks_id": account["Id"], "company": self.company}):
+ if not frappe.db.exists(
+ {"doctype": "Account", "quickbooks_id": account["Id"], "company": self.company}
+ ):
is_child = account["SubAccount"]
is_group = account["is_group"]
# Create Two Accounts for every Group Account
@@ -416,103 +434,125 @@
account_id = account["Id"]
if is_child:
- parent_account = self._get_account_name_by_id("Group - {}".format(account["ParentRef"]["value"]))
+ parent_account = self._get_account_name_by_id(
+ "Group - {}".format(account["ParentRef"]["value"])
+ )
else:
- parent_account = encode_company_abbr("{} - QB".format(mapping[account["AccountType"]]), self.company)
+ parent_account = encode_company_abbr(
+ "{} - QB".format(mapping[account["AccountType"]]), self.company
+ )
- frappe.get_doc({
- "doctype": "Account",
- "quickbooks_id": account_id,
- "account_name": self._get_unique_account_name(account["Name"]),
- "root_type": mapping[account["AccountType"]],
- "account_type": self._get_account_type(account),
- "account_currency": account["CurrencyRef"]["value"],
- "parent_account": parent_account,
- "is_group": is_group,
- "company": self.company,
- }).insert()
-
- if is_group:
- # Create a Leaf account corresponding to the group account
- frappe.get_doc({
+ frappe.get_doc(
+ {
"doctype": "Account",
- "quickbooks_id": account["Id"],
+ "quickbooks_id": account_id,
"account_name": self._get_unique_account_name(account["Name"]),
"root_type": mapping[account["AccountType"]],
"account_type": self._get_account_type(account),
"account_currency": account["CurrencyRef"]["value"],
- "parent_account": self._get_account_name_by_id(account_id),
- "is_group": 0,
+ "parent_account": parent_account,
+ "is_group": is_group,
"company": self.company,
- }).insert()
+ }
+ ).insert()
+
+ if is_group:
+ # Create a Leaf account corresponding to the group account
+ frappe.get_doc(
+ {
+ "doctype": "Account",
+ "quickbooks_id": account["Id"],
+ "account_name": self._get_unique_account_name(account["Name"]),
+ "root_type": mapping[account["AccountType"]],
+ "account_type": self._get_account_type(account),
+ "account_currency": account["CurrencyRef"]["value"],
+ "parent_account": self._get_account_name_by_id(account_id),
+ "is_group": 0,
+ "company": self.company,
+ }
+ ).insert()
if account.get("AccountSubType") == "UndepositedFunds":
self.undeposited_funds_account = self._get_account_name_by_id(account["Id"])
self.save()
except Exception as e:
self._log_error(e, account)
-
def _get_account_type(self, account):
account_subtype_mapping = {"UndepositedFunds": "Cash"}
account_type = account_subtype_mapping.get(account.get("AccountSubType"))
if account_type is None:
- account_type_mapping = {"Accounts Payable": "Payable", "Accounts Receivable": "Receivable", "Bank": "Bank", "Credit Card": "Bank"}
+ account_type_mapping = {
+ "Accounts Payable": "Payable",
+ "Accounts Receivable": "Receivable",
+ "Bank": "Bank",
+ "Credit Card": "Bank",
+ }
account_type = account_type_mapping.get(account["AccountType"])
return account_type
-
def _preprocess_tax_rates(self, tax_rates):
self.tax_rates = {tax_rate["Id"]: tax_rate for tax_rate in tax_rates}
return tax_rates
-
def _save_tax_rate(self, tax_rate):
try:
- if not frappe.db.exists({"doctype": "Account", "quickbooks_id": "TaxRate - {}".format(tax_rate["Id"]), "company": self.company}):
- frappe.get_doc({
+ if not frappe.db.exists(
+ {
"doctype": "Account",
"quickbooks_id": "TaxRate - {}".format(tax_rate["Id"]),
- "account_name": "{} - QB".format(tax_rate["Name"]),
- "root_type": "Liability",
- "parent_account": encode_company_abbr("{} - QB".format("Liability"), self.company),
- "is_group": "0",
"company": self.company,
- }).insert()
+ }
+ ):
+ frappe.get_doc(
+ {
+ "doctype": "Account",
+ "quickbooks_id": "TaxRate - {}".format(tax_rate["Id"]),
+ "account_name": "{} - QB".format(tax_rate["Name"]),
+ "root_type": "Liability",
+ "parent_account": encode_company_abbr("{} - QB".format("Liability"), self.company),
+ "is_group": "0",
+ "company": self.company,
+ }
+ ).insert()
except Exception as e:
self._log_error(e, tax_rate)
-
def _preprocess_tax_codes(self, tax_codes):
self.tax_codes = {tax_code["Id"]: tax_code for tax_code in tax_codes}
return tax_codes
-
def _save_tax_code(self, tax_code):
pass
-
def _save_customer(self, customer):
try:
- if not frappe.db.exists({"doctype": "Customer", "quickbooks_id": customer["Id"], "company": self.company}):
+ if not frappe.db.exists(
+ {"doctype": "Customer", "quickbooks_id": customer["Id"], "company": self.company}
+ ):
try:
- receivable_account = frappe.get_all("Account", filters={
- "account_type": "Receivable",
- "account_currency": customer["CurrencyRef"]["value"],
- "company": self.company,
- })[0]["name"]
+ receivable_account = frappe.get_all(
+ "Account",
+ filters={
+ "account_type": "Receivable",
+ "account_currency": customer["CurrencyRef"]["value"],
+ "company": self.company,
+ },
+ )[0]["name"]
except Exception:
receivable_account = None
- erpcustomer = frappe.get_doc({
- "doctype": "Customer",
- "quickbooks_id": customer["Id"],
- "customer_name": encode_company_abbr(customer["DisplayName"], self.company),
- "customer_type": "Individual",
- "customer_group": "Commercial",
- "default_currency": customer["CurrencyRef"]["value"],
- "accounts": [{"company": self.company, "account": receivable_account}],
- "territory": "All Territories",
- "company": self.company,
- }).insert()
+ erpcustomer = frappe.get_doc(
+ {
+ "doctype": "Customer",
+ "quickbooks_id": customer["Id"],
+ "customer_name": encode_company_abbr(customer["DisplayName"], self.company),
+ "customer_type": "Individual",
+ "customer_group": "Commercial",
+ "default_currency": customer["CurrencyRef"]["value"],
+ "accounts": [{"company": self.company, "account": receivable_account}],
+ "territory": "All Territories",
+ "company": self.company,
+ }
+ ).insert()
if "BillAddr" in customer:
self._create_address(erpcustomer, "Customer", customer["BillAddr"], "Billing")
if "ShipAddr" in customer:
@@ -520,10 +560,11 @@
except Exception as e:
self._log_error(e, customer)
-
def _save_item(self, item):
try:
- if not frappe.db.exists({"doctype": "Item", "quickbooks_id": item["Id"], "company": self.company}):
+ if not frappe.db.exists(
+ {"doctype": "Item", "quickbooks_id": item["Id"], "company": self.company}
+ ):
if item["Type"] in ("Service", "Inventory"):
item_dict = {
"doctype": "Item",
@@ -533,7 +574,7 @@
"is_stock_item": 0,
"item_group": "All Item Groups",
"company": self.company,
- "item_defaults": [{"company": self.company, "default_warehouse": self.default_warehouse}]
+ "item_defaults": [{"company": self.company, "default_warehouse": self.default_warehouse}],
}
if "ExpenseAccountRef" in item:
expense_account = self._get_account_name_by_id(item["ExpenseAccountRef"]["value"])
@@ -545,21 +586,23 @@
except Exception as e:
self._log_error(e, item)
-
def _allow_fraction_in_unit(self):
frappe.db.set_value("UOM", "Unit", "must_be_whole_number", 0)
-
def _save_vendor(self, vendor):
try:
- if not frappe.db.exists({"doctype": "Supplier", "quickbooks_id": vendor["Id"], "company": self.company}):
- erpsupplier = frappe.get_doc({
- "doctype": "Supplier",
- "quickbooks_id": vendor["Id"],
- "supplier_name": encode_company_abbr(vendor["DisplayName"], self.company),
- "supplier_group": "All Supplier Groups",
- "company": self.company,
- }).insert()
+ if not frappe.db.exists(
+ {"doctype": "Supplier", "quickbooks_id": vendor["Id"], "company": self.company}
+ ):
+ erpsupplier = frappe.get_doc(
+ {
+ "doctype": "Supplier",
+ "quickbooks_id": vendor["Id"],
+ "supplier_name": encode_company_abbr(vendor["DisplayName"], self.company),
+ "supplier_group": "All Supplier Groups",
+ "company": self.company,
+ }
+ ).insert()
if "BillAddr" in vendor:
self._create_address(erpsupplier, "Supplier", vendor["BillAddr"], "Billing")
if "ShipAddr" in vendor:
@@ -567,7 +610,6 @@
except Exception as e:
self._log_error(e)
-
def _save_preference(self, preference):
try:
if preference["SalesFormsPrefs"]["AllowShipping"]:
@@ -577,7 +619,6 @@
except Exception as e:
self._log_error(e, preference)
-
def _save_invoice(self, invoice):
# Invoice can be Linked with Another Transactions
# If any of these transactions is a "StatementCharge" or "ReimburseCharge" then in the UI
@@ -585,59 +626,56 @@
# Also as of now there is no way of fetching the corresponding transaction from api
# We in order to correctly reflect account balance make an equivalent Journal Entry
quickbooks_id = "Invoice - {}".format(invoice["Id"])
- if any(linked["TxnType"] in ("StatementCharge", "ReimburseCharge") for linked in invoice["LinkedTxn"]):
+ if any(
+ linked["TxnType"] in ("StatementCharge", "ReimburseCharge") for linked in invoice["LinkedTxn"]
+ ):
self._save_invoice_as_journal_entry(invoice, quickbooks_id)
else:
self._save_sales_invoice(invoice, quickbooks_id)
-
def _save_credit_memo(self, credit_memo):
# Credit Memo is equivalent to a return Sales Invoice
quickbooks_id = "Credit Memo - {}".format(credit_memo["Id"])
self._save_sales_invoice(credit_memo, quickbooks_id, is_return=True)
-
def _save_sales_receipt(self, sales_receipt):
# Sales Receipt is equivalent to a POS Sales Invoice
quickbooks_id = "Sales Receipt - {}".format(sales_receipt["Id"])
self._save_sales_invoice(sales_receipt, quickbooks_id, is_pos=True)
-
def _save_refund_receipt(self, refund_receipt):
# Refund Receipt is equivalent to a return POS Sales Invoice
quickbooks_id = "Refund Receipt - {}".format(refund_receipt["Id"])
self._save_sales_invoice(refund_receipt, quickbooks_id, is_return=True, is_pos=True)
-
def _save_sales_invoice(self, invoice, quickbooks_id, is_return=False, is_pos=False):
try:
- if not frappe.db.exists({"doctype": "Sales Invoice", "quickbooks_id": quickbooks_id, "company": self.company}):
+ if not frappe.db.exists(
+ {"doctype": "Sales Invoice", "quickbooks_id": quickbooks_id, "company": self.company}
+ ):
invoice_dict = {
"doctype": "Sales Invoice",
"quickbooks_id": quickbooks_id,
-
# Quickbooks uses ISO 4217 Code
# of course this gonna come back to bite me
"currency": invoice["CurrencyRef"]["value"],
-
# Exchange Rate is provided if multicurrency is enabled
# It is not provided if multicurrency is not enabled
"conversion_rate": invoice.get("ExchangeRate", 1),
"posting_date": invoice["TxnDate"],
-
# QuickBooks doesn't make Due Date a mandatory field this is a hack
"due_date": invoice.get("DueDate", invoice["TxnDate"]),
- "customer": frappe.get_all("Customer",
+ "customer": frappe.get_all(
+ "Customer",
filters={
"quickbooks_id": invoice["CustomerRef"]["value"],
"company": self.company,
- })[0]["name"],
+ },
+ )[0]["name"],
"items": self._get_si_items(invoice, is_return=is_return),
"taxes": self._get_taxes(invoice),
-
# Do not change posting_date upon submission
"set_posting_time": 1,
-
# QuickBooks doesn't round total
"disable_rounded_total": 1,
"is_return": is_return,
@@ -659,7 +697,6 @@
except Exception as e:
self._log_error(e, [invoice, invoice_dict, json.loads(invoice_doc.as_json())])
-
def _get_si_items(self, invoice, is_return=False):
items = []
for line in invoice["Line"]:
@@ -672,48 +709,56 @@
else:
tax_code = "NON"
if line["SalesItemLineDetail"]["ItemRef"]["value"] != "SHIPPING_ITEM_ID":
- item = frappe.db.get_all("Item",
+ item = frappe.db.get_all(
+ "Item",
filters={
"quickbooks_id": line["SalesItemLineDetail"]["ItemRef"]["value"],
"company": self.company,
},
- fields=["name", "stock_uom"]
+ fields=["name", "stock_uom"],
)[0]
- items.append({
- "item_code": item["name"],
- "conversion_factor": 1,
- "uom": item["stock_uom"],
- "description": line.get("Description", line["SalesItemLineDetail"]["ItemRef"]["name"]),
- "qty": line["SalesItemLineDetail"]["Qty"],
- "price_list_rate": line["SalesItemLineDetail"]["UnitPrice"],
- "cost_center": self.default_cost_center,
- "warehouse": self.default_warehouse,
- "item_tax_rate": json.dumps(self._get_item_taxes(tax_code))
- })
+ items.append(
+ {
+ "item_code": item["name"],
+ "conversion_factor": 1,
+ "uom": item["stock_uom"],
+ "description": line.get("Description", line["SalesItemLineDetail"]["ItemRef"]["name"]),
+ "qty": line["SalesItemLineDetail"]["Qty"],
+ "price_list_rate": line["SalesItemLineDetail"]["UnitPrice"],
+ "cost_center": self.default_cost_center,
+ "warehouse": self.default_warehouse,
+ "item_tax_rate": json.dumps(self._get_item_taxes(tax_code)),
+ }
+ )
else:
- items.append({
- "item_name": "Shipping",
- "conversion_factor": 1,
- "expense_account": self._get_account_name_by_id("TaxRate - {}".format(line["SalesItemLineDetail"]["TaxCodeRef"]["value"])),
- "uom": "Unit",
- "description": "Shipping",
- "income_account": self.default_shipping_account,
- "qty": 1,
- "price_list_rate": line["Amount"],
- "cost_center": self.default_cost_center,
- "warehouse": self.default_warehouse,
- "item_tax_rate": json.dumps(self._get_item_taxes(tax_code))
- })
+ items.append(
+ {
+ "item_name": "Shipping",
+ "conversion_factor": 1,
+ "expense_account": self._get_account_name_by_id(
+ "TaxRate - {}".format(line["SalesItemLineDetail"]["TaxCodeRef"]["value"])
+ ),
+ "uom": "Unit",
+ "description": "Shipping",
+ "income_account": self.default_shipping_account,
+ "qty": 1,
+ "price_list_rate": line["Amount"],
+ "cost_center": self.default_cost_center,
+ "warehouse": self.default_warehouse,
+ "item_tax_rate": json.dumps(self._get_item_taxes(tax_code)),
+ }
+ )
if is_return:
items[-1]["qty"] *= -1
elif line["DetailType"] == "DescriptionOnly":
- items[-1].update({
- "margin_type": "Percentage",
- "margin_rate_or_amount": int(line["Description"].split("%")[0]),
- })
+ items[-1].update(
+ {
+ "margin_type": "Percentage",
+ "margin_rate_or_amount": int(line["Description"].split("%")[0]),
+ }
+ )
return items
-
def _get_item_taxes(self, tax_code):
tax_rates = self.tax_rates
item_taxes = {}
@@ -723,30 +768,31 @@
if rate_list_type in tax_code:
for tax_rate_detail in tax_code[rate_list_type]["TaxRateDetail"]:
if tax_rate_detail["TaxTypeApplicable"] == "TaxOnAmount":
- tax_head = self._get_account_name_by_id("TaxRate - {}".format(tax_rate_detail["TaxRateRef"]["value"]))
+ tax_head = self._get_account_name_by_id(
+ "TaxRate - {}".format(tax_rate_detail["TaxRateRef"]["value"])
+ )
tax_rate = tax_rates[tax_rate_detail["TaxRateRef"]["value"]]
item_taxes[tax_head] = tax_rate["RateValue"]
return item_taxes
-
def _get_invoice_payments(self, invoice, is_return=False, is_pos=False):
if is_pos:
amount = invoice["TotalAmt"]
if is_return:
amount = -amount
- return [{
- "mode_of_payment": "Cash",
- "account": self._get_account_name_by_id(invoice["DepositToAccountRef"]["value"]),
- "amount": amount,
- }]
-
+ return [
+ {
+ "mode_of_payment": "Cash",
+ "account": self._get_account_name_by_id(invoice["DepositToAccountRef"]["value"]),
+ "amount": amount,
+ }
+ ]
def _get_discount(self, lines):
for line in lines:
if line["DetailType"] == "DiscountLineDetail" and "Amount" in line["DiscountLineDetail"]:
return line
-
def _save_invoice_as_journal_entry(self, invoice, quickbooks_id):
try:
accounts = []
@@ -758,8 +804,9 @@
account_line["credit_in_account_currency"] = line["credit"]
if frappe.db.get_value("Account", line["account"], "account_type") == "Receivable":
account_line["party_type"] = "Customer"
- account_line["party"] = frappe.get_all("Customer",
- filters={"quickbooks_id": invoice["CustomerRef"]["value"], "company": self.company}
+ account_line["party"] = frappe.get_all(
+ "Customer",
+ filters={"quickbooks_id": invoice["CustomerRef"]["value"], "company": self.company},
)[0]["name"]
accounts.append(account_line)
@@ -769,7 +816,6 @@
except Exception as e:
self._log_error(e, [invoice, accounts])
-
def _save_journal_entry(self, journal_entry):
# JournalEntry is equivalent to a Journal Entry
@@ -782,13 +828,17 @@
accounts = []
for line in lines:
if line["DetailType"] == "JournalEntryLineDetail":
- account_name = self._get_account_name_by_id(line["JournalEntryLineDetail"]["AccountRef"]["value"])
+ account_name = self._get_account_name_by_id(
+ line["JournalEntryLineDetail"]["AccountRef"]["value"]
+ )
posting_type = line["JournalEntryLineDetail"]["PostingType"]
- accounts.append({
- "account": account_name,
- posting_type_field_mapping[posting_type]: line["Amount"],
- "cost_center": self.default_cost_center,
- })
+ accounts.append(
+ {
+ "account": account_name,
+ posting_type_field_mapping[posting_type]: line["Amount"],
+ "cost_center": self.default_cost_center,
+ }
+ )
return accounts
quickbooks_id = "Journal Entry - {}".format(journal_entry["Id"])
@@ -796,39 +846,41 @@
posting_date = journal_entry["TxnDate"]
self.__save_journal_entry(quickbooks_id, accounts, posting_date)
-
def __save_journal_entry(self, quickbooks_id, accounts, posting_date):
try:
- if not frappe.db.exists({"doctype": "Journal Entry", "quickbooks_id": quickbooks_id, "company": self.company}):
- je = frappe.get_doc({
- "doctype": "Journal Entry",
- "quickbooks_id": quickbooks_id,
- "company": self.company,
- "posting_date": posting_date,
- "accounts": accounts,
- "multi_currency": 1,
- })
+ if not frappe.db.exists(
+ {"doctype": "Journal Entry", "quickbooks_id": quickbooks_id, "company": self.company}
+ ):
+ je = frappe.get_doc(
+ {
+ "doctype": "Journal Entry",
+ "quickbooks_id": quickbooks_id,
+ "company": self.company,
+ "posting_date": posting_date,
+ "accounts": accounts,
+ "multi_currency": 1,
+ }
+ )
je.insert()
je.submit()
except Exception as e:
self._log_error(e, [accounts, json.loads(je.as_json())])
-
def _save_bill(self, bill):
# Bill is equivalent to a Purchase Invoice
quickbooks_id = "Bill - {}".format(bill["Id"])
self.__save_purchase_invoice(bill, quickbooks_id)
-
def _save_vendor_credit(self, vendor_credit):
# Vendor Credit is equivalent to a return Purchase Invoice
quickbooks_id = "Vendor Credit - {}".format(vendor_credit["Id"])
self.__save_purchase_invoice(vendor_credit, quickbooks_id, is_return=True)
-
def __save_purchase_invoice(self, invoice, quickbooks_id, is_return=False):
try:
- if not frappe.db.exists({"doctype": "Purchase Invoice", "quickbooks_id": quickbooks_id, "company": self.company}):
+ if not frappe.db.exists(
+ {"doctype": "Purchase Invoice", "quickbooks_id": quickbooks_id, "company": self.company}
+ ):
credit_to_account = self._get_account_name_by_id(invoice["APAccountRef"]["value"])
invoice_dict = {
"doctype": "Purchase Invoice",
@@ -838,11 +890,13 @@
"posting_date": invoice["TxnDate"],
"due_date": invoice.get("DueDate", invoice["TxnDate"]),
"credit_to": credit_to_account,
- "supplier": frappe.get_all("Supplier",
+ "supplier": frappe.get_all(
+ "Supplier",
filters={
"quickbooks_id": invoice["VendorRef"]["value"],
"company": self.company,
- })[0]["name"],
+ },
+ )[0]["name"],
"items": self._get_pi_items(invoice, is_return=is_return),
"taxes": self._get_taxes(invoice),
"set_posting_time": 1,
@@ -857,7 +911,6 @@
except Exception as e:
self._log_error(e, [invoice, invoice_dict, json.loads(invoice_doc.as_json())])
-
def _get_pi_items(self, purchase_invoice, is_return=False):
items = []
for line in purchase_invoice["Line"]:
@@ -869,24 +922,29 @@
tax_code = purchase_invoice["TxnTaxDetail"]["TxnTaxCodeRef"]["value"]
else:
tax_code = "NON"
- item = frappe.db.get_all("Item",
+ item = frappe.db.get_all(
+ "Item",
filters={
"quickbooks_id": line["ItemBasedExpenseLineDetail"]["ItemRef"]["value"],
- "company": self.company
+ "company": self.company,
},
- fields=["name", "stock_uom"]
+ fields=["name", "stock_uom"],
)[0]
- items.append({
- "item_code": item["name"],
- "conversion_factor": 1,
- "uom": item["stock_uom"],
- "description": line.get("Description", line["ItemBasedExpenseLineDetail"]["ItemRef"]["name"]),
- "qty": line["ItemBasedExpenseLineDetail"]["Qty"],
- "price_list_rate": line["ItemBasedExpenseLineDetail"]["UnitPrice"],
- "warehouse": self.default_warehouse,
- "cost_center": self.default_cost_center,
- "item_tax_rate": json.dumps(self._get_item_taxes(tax_code)),
- })
+ items.append(
+ {
+ "item_code": item["name"],
+ "conversion_factor": 1,
+ "uom": item["stock_uom"],
+ "description": line.get(
+ "Description", line["ItemBasedExpenseLineDetail"]["ItemRef"]["name"]
+ ),
+ "qty": line["ItemBasedExpenseLineDetail"]["Qty"],
+ "price_list_rate": line["ItemBasedExpenseLineDetail"]["UnitPrice"],
+ "warehouse": self.default_warehouse,
+ "cost_center": self.default_cost_center,
+ "item_tax_rate": json.dumps(self._get_item_taxes(tax_code)),
+ }
+ )
elif line["DetailType"] == "AccountBasedExpenseLineDetail":
if line["AccountBasedExpenseLineDetail"]["TaxCodeRef"]["value"] != "TAX":
tax_code = line["AccountBasedExpenseLineDetail"]["TaxCodeRef"]["value"]
@@ -895,23 +953,30 @@
tax_code = purchase_invoice["TxnTaxDetail"]["TxnTaxCodeRef"]["value"]
else:
tax_code = "NON"
- items.append({
- "item_name": line.get("Description", line["AccountBasedExpenseLineDetail"]["AccountRef"]["name"]),
- "conversion_factor": 1,
- "expense_account": self._get_account_name_by_id(line["AccountBasedExpenseLineDetail"]["AccountRef"]["value"]),
- "uom": "Unit",
- "description": line.get("Description", line["AccountBasedExpenseLineDetail"]["AccountRef"]["name"]),
- "qty": 1,
- "price_list_rate": line["Amount"],
- "warehouse": self.default_warehouse,
- "cost_center": self.default_cost_center,
- "item_tax_rate": json.dumps(self._get_item_taxes(tax_code)),
- })
+ items.append(
+ {
+ "item_name": line.get(
+ "Description", line["AccountBasedExpenseLineDetail"]["AccountRef"]["name"]
+ ),
+ "conversion_factor": 1,
+ "expense_account": self._get_account_name_by_id(
+ line["AccountBasedExpenseLineDetail"]["AccountRef"]["value"]
+ ),
+ "uom": "Unit",
+ "description": line.get(
+ "Description", line["AccountBasedExpenseLineDetail"]["AccountRef"]["name"]
+ ),
+ "qty": 1,
+ "price_list_rate": line["Amount"],
+ "warehouse": self.default_warehouse,
+ "cost_center": self.default_cost_center,
+ "item_tax_rate": json.dumps(self._get_item_taxes(tax_code)),
+ }
+ )
if is_return:
items[-1]["qty"] *= -1
return items
-
def _save_payment(self, payment):
try:
quickbooks_id = "Payment - {}".format(payment["Id"])
@@ -928,8 +993,11 @@
if linked_transaction["TxnType"] == "Invoice":
si_quickbooks_id = "Invoice - {}".format(linked_transaction["TxnId"])
# Invoice could have been saved as a Sales Invoice or a Journal Entry
- if frappe.db.exists({"doctype": "Sales Invoice", "quickbooks_id": si_quickbooks_id, "company": self.company}):
- sales_invoice = frappe.get_all("Sales Invoice",
+ if frappe.db.exists(
+ {"doctype": "Sales Invoice", "quickbooks_id": si_quickbooks_id, "company": self.company}
+ ):
+ sales_invoice = frappe.get_all(
+ "Sales Invoice",
filters={
"quickbooks_id": si_quickbooks_id,
"company": self.company,
@@ -941,43 +1009,51 @@
party = sales_invoice["customer"]
party_account = sales_invoice["debit_to"]
- if frappe.db.exists({"doctype": "Journal Entry", "quickbooks_id": si_quickbooks_id, "company": self.company}):
- journal_entry = frappe.get_doc("Journal Entry",
+ if frappe.db.exists(
+ {"doctype": "Journal Entry", "quickbooks_id": si_quickbooks_id, "company": self.company}
+ ):
+ journal_entry = frappe.get_doc(
+ "Journal Entry",
{
"quickbooks_id": si_quickbooks_id,
"company": self.company,
- }
+ },
)
# Invoice saved as a Journal Entry must have party and party_type set on line containing Receivable Account
- customer_account_line = list(filter(lambda acc: acc.party_type == "Customer", journal_entry.accounts))[0]
+ customer_account_line = list(
+ filter(lambda acc: acc.party_type == "Customer", journal_entry.accounts)
+ )[0]
reference_type = "Journal Entry"
reference_name = journal_entry.name
party = customer_account_line.party
party_account = customer_account_line.account
- accounts.append({
- "party_type": "Customer",
- "party": party,
- "reference_type": reference_type,
- "reference_name": reference_name,
- "account": party_account,
- "credit_in_account_currency": line["Amount"],
- "cost_center": self.default_cost_center,
- })
+ accounts.append(
+ {
+ "party_type": "Customer",
+ "party": party,
+ "reference_type": reference_type,
+ "reference_name": reference_name,
+ "account": party_account,
+ "credit_in_account_currency": line["Amount"],
+ "cost_center": self.default_cost_center,
+ }
+ )
deposit_account = self._get_account_name_by_id(payment["DepositToAccountRef"]["value"])
- accounts.append({
- "account": deposit_account,
- "debit_in_account_currency": payment["TotalAmt"],
- "cost_center": self.default_cost_center,
- })
+ accounts.append(
+ {
+ "account": deposit_account,
+ "debit_in_account_currency": payment["TotalAmt"],
+ "cost_center": self.default_cost_center,
+ }
+ )
posting_date = payment["TxnDate"]
self.__save_journal_entry(quickbooks_id, accounts, posting_date)
except Exception as e:
self._log_error(e, [payment, accounts])
-
def _save_bill_payment(self, bill_payment):
try:
quickbooks_id = "BillPayment - {}".format(bill_payment["Id"])
@@ -987,8 +1063,11 @@
linked_transaction = line["LinkedTxn"][0]
if linked_transaction["TxnType"] == "Bill":
pi_quickbooks_id = "Bill - {}".format(linked_transaction["TxnId"])
- if frappe.db.exists({"doctype": "Purchase Invoice", "quickbooks_id": pi_quickbooks_id, "company": self.company}):
- purchase_invoice = frappe.get_all("Purchase Invoice",
+ if frappe.db.exists(
+ {"doctype": "Purchase Invoice", "quickbooks_id": pi_quickbooks_id, "company": self.company}
+ ):
+ purchase_invoice = frappe.get_all(
+ "Purchase Invoice",
filters={
"quickbooks_id": pi_quickbooks_id,
"company": self.company,
@@ -999,15 +1078,17 @@
reference_name = purchase_invoice["name"]
party = purchase_invoice["supplier"]
party_account = purchase_invoice["credit_to"]
- accounts.append({
- "party_type": "Supplier",
- "party": party,
- "reference_type": reference_type,
- "reference_name": reference_name,
- "account": party_account,
- "debit_in_account_currency": line["Amount"],
- "cost_center": self.default_cost_center,
- })
+ accounts.append(
+ {
+ "party_type": "Supplier",
+ "party": party,
+ "reference_type": reference_type,
+ "reference_name": reference_name,
+ "account": party_account,
+ "debit_in_account_currency": line["Amount"],
+ "cost_center": self.default_cost_center,
+ }
+ )
if bill_payment["PayType"] == "Check":
bank_account_id = bill_payment["CheckPayment"]["BankAccountRef"]["value"]
@@ -1015,49 +1096,68 @@
bank_account_id = bill_payment["CreditCardPayment"]["CCAccountRef"]["value"]
bank_account = self._get_account_name_by_id(bank_account_id)
- accounts.append({
- "account": bank_account,
- "credit_in_account_currency": bill_payment["TotalAmt"],
- "cost_center": self.default_cost_center,
- })
+ accounts.append(
+ {
+ "account": bank_account,
+ "credit_in_account_currency": bill_payment["TotalAmt"],
+ "cost_center": self.default_cost_center,
+ }
+ )
posting_date = bill_payment["TxnDate"]
self.__save_journal_entry(quickbooks_id, accounts, posting_date)
except Exception as e:
self._log_error(e, [bill_payment, accounts])
-
def _save_purchase(self, purchase):
try:
quickbooks_id = "Purchase - {}".format(purchase["Id"])
# Credit Bank Account
- accounts = [{
- "account": self._get_account_name_by_id(purchase["AccountRef"]["value"]),
- "credit_in_account_currency": purchase["TotalAmt"],
- "cost_center": self.default_cost_center,
- }]
+ accounts = [
+ {
+ "account": self._get_account_name_by_id(purchase["AccountRef"]["value"]),
+ "credit_in_account_currency": purchase["TotalAmt"],
+ "cost_center": self.default_cost_center,
+ }
+ ]
# Debit Mentioned Accounts
for line in purchase["Line"]:
if line["DetailType"] == "AccountBasedExpenseLineDetail":
- account = self._get_account_name_by_id(line["AccountBasedExpenseLineDetail"]["AccountRef"]["value"])
+ account = self._get_account_name_by_id(
+ line["AccountBasedExpenseLineDetail"]["AccountRef"]["value"]
+ )
elif line["DetailType"] == "ItemBasedExpenseLineDetail":
- account = frappe.get_doc("Item",
- {"quickbooks_id": line["ItemBasedExpenseLineDetail"]["ItemRef"]["value"], "company": self.company}
- ).item_defaults[0].expense_account
- accounts.append({
- "account": account,
- "debit_in_account_currency": line["Amount"],
- "cost_center": self.default_cost_center,
- })
+ account = (
+ frappe.get_doc(
+ "Item",
+ {
+ "quickbooks_id": line["ItemBasedExpenseLineDetail"]["ItemRef"]["value"],
+ "company": self.company,
+ },
+ )
+ .item_defaults[0]
+ .expense_account
+ )
+ accounts.append(
+ {
+ "account": account,
+ "debit_in_account_currency": line["Amount"],
+ "cost_center": self.default_cost_center,
+ }
+ )
# Debit Tax Accounts
if "TxnTaxDetail" in purchase:
for line in purchase["TxnTaxDetail"]["TaxLine"]:
- accounts.append({
- "account": self._get_account_name_by_id("TaxRate - {}".format(line["TaxLineDetail"]["TaxRateRef"]["value"])),
- "debit_in_account_currency": line["Amount"],
- "cost_center": self.default_cost_center,
- })
+ accounts.append(
+ {
+ "account": self._get_account_name_by_id(
+ "TaxRate - {}".format(line["TaxLineDetail"]["TaxRateRef"]["value"])
+ ),
+ "debit_in_account_currency": line["Amount"],
+ "cost_center": self.default_cost_center,
+ }
+ )
# If purchase["Credit"] is set to be True then it represents a refund
if purchase.get("Credit"):
@@ -1074,61 +1174,64 @@
except Exception as e:
self._log_error(e, [purchase, accounts])
-
def _save_deposit(self, deposit):
try:
quickbooks_id = "Deposit - {}".format(deposit["Id"])
# Debit Bank Account
- accounts = [{
- "account": self._get_account_name_by_id(deposit["DepositToAccountRef"]["value"]),
- "debit_in_account_currency": deposit["TotalAmt"],
- "cost_center": self.default_cost_center,
- }]
+ accounts = [
+ {
+ "account": self._get_account_name_by_id(deposit["DepositToAccountRef"]["value"]),
+ "debit_in_account_currency": deposit["TotalAmt"],
+ "cost_center": self.default_cost_center,
+ }
+ ]
# Credit Mentioned Accounts
for line in deposit["Line"]:
if "LinkedTxn" in line:
- accounts.append({
- "account": self.undeposited_funds_account,
- "credit_in_account_currency": line["Amount"],
- "cost_center": self.default_cost_center,
- })
+ accounts.append(
+ {
+ "account": self.undeposited_funds_account,
+ "credit_in_account_currency": line["Amount"],
+ "cost_center": self.default_cost_center,
+ }
+ )
else:
- accounts.append({
- "account": self._get_account_name_by_id(line["DepositLineDetail"]["AccountRef"]["value"]),
- "credit_in_account_currency": line["Amount"],
- "cost_center": self.default_cost_center,
- })
+ accounts.append(
+ {
+ "account": self._get_account_name_by_id(line["DepositLineDetail"]["AccountRef"]["value"]),
+ "credit_in_account_currency": line["Amount"],
+ "cost_center": self.default_cost_center,
+ }
+ )
# Debit Cashback if mentioned
if "CashBack" in deposit:
- accounts.append({
- "account": self._get_account_name_by_id(deposit["CashBack"]["AccountRef"]["value"]),
- "debit_in_account_currency": deposit["CashBack"]["Amount"],
- "cost_center": self.default_cost_center,
- })
+ accounts.append(
+ {
+ "account": self._get_account_name_by_id(deposit["CashBack"]["AccountRef"]["value"]),
+ "debit_in_account_currency": deposit["CashBack"]["Amount"],
+ "cost_center": self.default_cost_center,
+ }
+ )
posting_date = deposit["TxnDate"]
self.__save_journal_entry(quickbooks_id, accounts, posting_date)
except Exception as e:
self._log_error(e, [deposit, accounts])
-
def _save_advance_payment(self, advance_payment):
quickbooks_id = "Advance Payment - {}".format(advance_payment["id"])
self.__save_ledger_entry_as_je(advance_payment, quickbooks_id)
-
def _save_tax_payment(self, tax_payment):
quickbooks_id = "Tax Payment - {}".format(tax_payment["id"])
self.__save_ledger_entry_as_je(tax_payment, quickbooks_id)
-
def _save_inventory_qty_adjust(self, inventory_qty_adjust):
quickbooks_id = "Inventory Qty Adjust - {}".format(inventory_qty_adjust["id"])
self.__save_ledger_entry_as_je(inventory_qty_adjust, quickbooks_id)
-
def __save_ledger_entry_as_je(self, ledger_entry, quickbooks_id):
try:
accounts = []
@@ -1145,7 +1248,6 @@
except Exception as e:
self._log_error(e, ledger_entry)
-
def _get_taxes(self, entry):
taxes = []
if "TxnTaxDetail" not in entry or "TaxLine" not in entry["TxnTaxDetail"]:
@@ -1155,27 +1257,30 @@
account_head = self._get_account_name_by_id("TaxRate - {}".format(tax_rate))
tax_type_applicable = self._get_tax_type(tax_rate)
if tax_type_applicable == "TaxOnAmount":
- taxes.append({
- "charge_type": "On Net Total",
- "account_head": account_head,
- "description": account_head,
- "cost_center": self.default_cost_center,
- "rate": 0,
- })
+ taxes.append(
+ {
+ "charge_type": "On Net Total",
+ "account_head": account_head,
+ "description": account_head,
+ "cost_center": self.default_cost_center,
+ "rate": 0,
+ }
+ )
else:
parent_tax_rate = self._get_parent_tax_rate(tax_rate)
parent_row_id = self._get_parent_row_id(parent_tax_rate, taxes)
- taxes.append({
- "charge_type": "On Previous Row Amount",
- "row_id": parent_row_id,
- "account_head": account_head,
- "description": account_head,
- "cost_center": self.default_cost_center,
- "rate": line["TaxLineDetail"]["TaxPercent"],
- })
+ taxes.append(
+ {
+ "charge_type": "On Previous Row Amount",
+ "row_id": parent_row_id,
+ "account_head": account_head,
+ "description": account_head,
+ "cost_center": self.default_cost_center,
+ "rate": line["TaxLineDetail"]["TaxPercent"],
+ }
+ )
return taxes
-
def _get_tax_type(self, tax_rate):
for tax_code in self.tax_codes.values():
for rate_list_type in ("SalesTaxRateList", "PurchaseTaxRateList"):
@@ -1184,7 +1289,6 @@
if tax_rate_detail["TaxRateRef"]["value"] == tax_rate:
return tax_rate_detail["TaxTypeApplicable"]
-
def _get_parent_tax_rate(self, tax_rate):
parent = None
for tax_code in self.tax_codes.values():
@@ -1198,34 +1302,33 @@
if tax_rate_detail["TaxOrder"] == parent:
return tax_rate_detail["TaxRateRef"]["value"]
-
def _get_parent_row_id(self, tax_rate, taxes):
tax_account = self._get_account_name_by_id("TaxRate - {}".format(tax_rate))
for index, tax in enumerate(taxes):
if tax["account_head"] == tax_account:
return index + 1
-
def _create_address(self, entity, doctype, address, address_type):
try:
if not frappe.db.exists({"doctype": "Address", "quickbooks_id": address["Id"]}):
- frappe.get_doc({
- "doctype": "Address",
- "quickbooks_address_id": address["Id"],
- "address_title": entity.name,
- "address_type": address_type,
- "address_line1": address["Line1"],
- "city": address["City"],
- "links": [{"link_doctype": doctype, "link_name": entity.name}]
- }).insert()
+ frappe.get_doc(
+ {
+ "doctype": "Address",
+ "quickbooks_address_id": address["Id"],
+ "address_title": entity.name,
+ "address_type": address_type,
+ "address_line1": address["Line1"],
+ "city": address["City"],
+ "links": [{"link_doctype": doctype, "link_name": entity.name}],
+ }
+ ).insert()
except Exception as e:
self._log_error(e, address)
-
def _get(self, *args, **kwargs):
kwargs["headers"] = {
"Accept": "application/json",
- "Authorization": "Bearer {}".format(self.access_token)
+ "Authorization": "Bearer {}".format(self.access_token),
}
response = requests.get(*args, **kwargs)
# HTTP Status code 401 here means that the access_token is expired
@@ -1236,43 +1339,41 @@
response = self._get(*args, **kwargs)
return response
-
def _get_account_name_by_id(self, quickbooks_id):
- return frappe.get_all("Account", filters={"quickbooks_id": quickbooks_id, "company": self.company})[0]["name"]
-
+ return frappe.get_all(
+ "Account", filters={"quickbooks_id": quickbooks_id, "company": self.company}
+ )[0]["name"]
def _publish(self, *args, **kwargs):
frappe.publish_realtime("quickbooks_progress_update", *args, **kwargs)
-
def _get_unique_account_name(self, quickbooks_name, number=0):
if number:
quickbooks_account_name = "{} - {} - QB".format(quickbooks_name, number)
else:
quickbooks_account_name = "{} - QB".format(quickbooks_name)
company_encoded_account_name = encode_company_abbr(quickbooks_account_name, self.company)
- if frappe.db.exists({"doctype": "Account", "name": company_encoded_account_name, "company": self.company}):
+ if frappe.db.exists(
+ {"doctype": "Account", "name": company_encoded_account_name, "company": self.company}
+ ):
unique_account_name = self._get_unique_account_name(quickbooks_name, number + 1)
else:
unique_account_name = quickbooks_account_name
return unique_account_name
-
def _log_error(self, execption, data=""):
- frappe.log_error(title="QuickBooks Migration Error",
- message="\n".join([
- "Data",
- json.dumps(data,
- sort_keys=True,
- indent=4,
- separators=(',', ': ')
- ),
- "Exception",
- traceback.format_exc()
- ])
+ frappe.log_error(
+ title="QuickBooks Migration Error",
+ message="\n".join(
+ [
+ "Data",
+ json.dumps(data, sort_keys=True, indent=4, separators=(",", ": ")),
+ "Exception",
+ traceback.format_exc(),
+ ]
+ ),
)
-
def set_indicator(self, status):
self.status = status
self.save()
diff --git a/erpnext/erpnext_integrations/doctype/tally_migration/tally_migration.py b/erpnext/erpnext_integrations/doctype/tally_migration/tally_migration.py
index 26bd19f..7d676e4 100644
--- a/erpnext/erpnext_integrations/doctype/tally_migration/tally_migration.py
+++ b/erpnext/erpnext_integrations/doctype/tally_migration/tally_migration.py
@@ -36,6 +36,7 @@
return doc
+
class TallyMigration(Document):
def validate(self):
failed_import_log = json.loads(self.failed_import_log)
@@ -73,14 +74,16 @@
def dump_processed_data(self, data):
for key, value in data.items():
- f = frappe.get_doc({
- "doctype": "File",
- "file_name": key + ".json",
- "attached_to_doctype": self.doctype,
- "attached_to_name": self.name,
- "content": json.dumps(value),
- "is_private": True
- })
+ f = frappe.get_doc(
+ {
+ "doctype": "File",
+ "file_name": key + ".json",
+ "attached_to_doctype": self.doctype,
+ "attached_to_name": self.name,
+ "content": json.dumps(value),
+ "is_private": True,
+ }
+ )
try:
f.insert(ignore_if_duplicate=True)
except frappe.DuplicateEntryError:
@@ -88,8 +91,12 @@
setattr(self, key, f.file_url)
def set_account_defaults(self):
- self.default_cost_center, self.default_round_off_account = frappe.db.get_value("Company", self.erpnext_company, ["cost_center", "round_off_account"])
- self.default_warehouse = frappe.db.get_value("Stock Settings", "Stock Settings", "default_warehouse")
+ self.default_cost_center, self.default_round_off_account = frappe.db.get_value(
+ "Company", self.erpnext_company, ["cost_center", "round_off_account"]
+ )
+ self.default_warehouse = frappe.db.get_value(
+ "Stock Settings", "Stock Settings", "default_warehouse"
+ )
def _process_master_data(self):
def get_company_name(collection):
@@ -100,18 +107,24 @@
"Application of Funds (Assets)": "Asset",
"Expenses": "Expense",
"Income": "Income",
- "Source of Funds (Liabilities)": "Liability"
+ "Source of Funds (Liabilities)": "Liability",
}
roots = set(root_type_map.keys())
- accounts = list(get_groups(collection.find_all("GROUP"))) + list(get_ledgers(collection.find_all("LEDGER")))
+ accounts = list(get_groups(collection.find_all("GROUP"))) + list(
+ get_ledgers(collection.find_all("LEDGER"))
+ )
children, parents = get_children_and_parent_dict(accounts)
- group_set = [acc[1] for acc in accounts if acc[2]]
+ group_set = [acc[1] for acc in accounts if acc[2]]
children, customers, suppliers = remove_parties(parents, children, group_set)
try:
coa = traverse({}, children, roots, roots, group_set)
except RecursionError:
- self.log(_("Error occured while parsing Chart of Accounts: Please make sure that no two accounts have the same name"))
+ self.log(
+ _(
+ "Error occured while parsing Chart of Accounts: Please make sure that no two accounts have the same name"
+ )
+ )
for account in coa:
coa[account]["root_type"] = root_type_map[account]
@@ -185,42 +198,48 @@
links = []
if account.NAME.string.strip() in customers:
party_type = "Customer"
- parties.append({
- "doctype": party_type,
- "customer_name": account.NAME.string.strip(),
- "tax_id": account.INCOMETAXNUMBER.string.strip() if account.INCOMETAXNUMBER else None,
- "customer_group": "All Customer Groups",
- "territory": "All Territories",
- "customer_type": "Individual",
- })
+ parties.append(
+ {
+ "doctype": party_type,
+ "customer_name": account.NAME.string.strip(),
+ "tax_id": account.INCOMETAXNUMBER.string.strip() if account.INCOMETAXNUMBER else None,
+ "customer_group": "All Customer Groups",
+ "territory": "All Territories",
+ "customer_type": "Individual",
+ }
+ )
links.append({"link_doctype": party_type, "link_name": account["NAME"]})
if account.NAME.string.strip() in suppliers:
party_type = "Supplier"
- parties.append({
- "doctype": party_type,
- "supplier_name": account.NAME.string.strip(),
- "pan": account.INCOMETAXNUMBER.string.strip() if account.INCOMETAXNUMBER else None,
- "supplier_group": "All Supplier Groups",
- "supplier_type": "Individual",
- })
+ parties.append(
+ {
+ "doctype": party_type,
+ "supplier_name": account.NAME.string.strip(),
+ "pan": account.INCOMETAXNUMBER.string.strip() if account.INCOMETAXNUMBER else None,
+ "supplier_group": "All Supplier Groups",
+ "supplier_type": "Individual",
+ }
+ )
links.append({"link_doctype": party_type, "link_name": account["NAME"]})
if party_type:
address = "\n".join([a.string.strip() for a in account.find_all("ADDRESS")])
- addresses.append({
- "doctype": "Address",
- "address_line1": address[:140].strip(),
- "address_line2": address[140:].strip(),
- "country": account.COUNTRYNAME.string.strip() if account.COUNTRYNAME else None,
- "state": account.LEDSTATENAME.string.strip() if account.LEDSTATENAME else None,
- "gst_state": account.LEDSTATENAME.string.strip() if account.LEDSTATENAME else None,
- "pin_code": account.PINCODE.string.strip() if account.PINCODE else None,
- "mobile": account.LEDGERPHONE.string.strip() if account.LEDGERPHONE else None,
- "phone": account.LEDGERPHONE.string.strip() if account.LEDGERPHONE else None,
- "gstin": account.PARTYGSTIN.string.strip() if account.PARTYGSTIN else None,
- "links": links
- })
+ addresses.append(
+ {
+ "doctype": "Address",
+ "address_line1": address[:140].strip(),
+ "address_line2": address[140:].strip(),
+ "country": account.COUNTRYNAME.string.strip() if account.COUNTRYNAME else None,
+ "state": account.LEDSTATENAME.string.strip() if account.LEDSTATENAME else None,
+ "gst_state": account.LEDSTATENAME.string.strip() if account.LEDSTATENAME else None,
+ "pin_code": account.PINCODE.string.strip() if account.PINCODE else None,
+ "mobile": account.LEDGERPHONE.string.strip() if account.LEDGERPHONE else None,
+ "phone": account.LEDGERPHONE.string.strip() if account.LEDGERPHONE else None,
+ "gstin": account.PARTYGSTIN.string.strip() if account.PARTYGSTIN else None,
+ "links": links,
+ }
+ )
return parties, addresses
def get_stock_items_uoms(collection):
@@ -231,14 +250,16 @@
items = []
for item in collection.find_all("STOCKITEM"):
stock_uom = item.BASEUNITS.string.strip() if item.BASEUNITS else self.default_uom
- items.append({
- "doctype": "Item",
- "item_code" : item.NAME.string.strip(),
- "stock_uom": stock_uom.strip(),
- "is_stock_item": 0,
- "item_group": "All Item Groups",
- "item_defaults": [{"company": self.erpnext_company}]
- })
+ items.append(
+ {
+ "doctype": "Item",
+ "item_code": item.NAME.string.strip(),
+ "stock_uom": stock_uom.strip(),
+ "is_stock_item": 0,
+ "item_group": "All Item Groups",
+ "item_defaults": [{"company": self.erpnext_company}],
+ }
+ )
return items, uoms
@@ -257,7 +278,13 @@
self.publish("Process Master Data", _("Processing Items and UOMs"), 4, 5)
items, uoms = get_stock_items_uoms(collection)
- data = {"chart_of_accounts": chart_of_accounts, "parties": parties, "addresses": addresses, "items": items, "uoms": uoms}
+ data = {
+ "chart_of_accounts": chart_of_accounts,
+ "parties": parties,
+ "addresses": addresses,
+ "items": items,
+ "uoms": uoms,
+ }
self.publish("Process Master Data", _("Done"), 5, 5)
self.dump_processed_data(data)
@@ -272,7 +299,10 @@
self.set_status()
def publish(self, title, message, count, total):
- frappe.publish_realtime("tally_migration_progress_update", {"title": title, "message": message, "count": count, "total": total})
+ frappe.publish_realtime(
+ "tally_migration_progress_update",
+ {"title": title, "message": message, "count": count, "total": total},
+ )
def _import_master_data(self):
def create_company_and_coa(coa_file_url):
@@ -280,12 +310,14 @@
frappe.local.flags.ignore_chart_of_accounts = True
try:
- company = frappe.get_doc({
- "doctype": "Company",
- "company_name": self.erpnext_company,
- "default_currency": "INR",
- "enable_perpetual_inventory": 0,
- }).insert()
+ company = frappe.get_doc(
+ {
+ "doctype": "Company",
+ "company_name": self.erpnext_company,
+ "default_currency": "INR",
+ "enable_perpetual_inventory": 0,
+ }
+ ).insert()
except frappe.DuplicateEntryError:
company = frappe.get_doc("Company", self.erpnext_company)
unset_existing_data(self.erpnext_company)
@@ -358,8 +390,16 @@
for voucher in collection.find_all("VOUCHER"):
if voucher.ISCANCELLED.string.strip() == "Yes":
continue
- inventory_entries = voucher.find_all("INVENTORYENTRIES.LIST") + voucher.find_all("ALLINVENTORYENTRIES.LIST") + voucher.find_all("INVENTORYENTRIESIN.LIST") + voucher.find_all("INVENTORYENTRIESOUT.LIST")
- if voucher.VOUCHERTYPENAME.string.strip() not in ["Journal", "Receipt", "Payment", "Contra"] and inventory_entries:
+ inventory_entries = (
+ voucher.find_all("INVENTORYENTRIES.LIST")
+ + voucher.find_all("ALLINVENTORYENTRIES.LIST")
+ + voucher.find_all("INVENTORYENTRIESIN.LIST")
+ + voucher.find_all("INVENTORYENTRIESOUT.LIST")
+ )
+ if (
+ voucher.VOUCHERTYPENAME.string.strip() not in ["Journal", "Receipt", "Payment", "Contra"]
+ and inventory_entries
+ ):
function = voucher_to_invoice
else:
function = voucher_to_journal_entry
@@ -375,9 +415,14 @@
def voucher_to_journal_entry(voucher):
accounts = []
- ledger_entries = voucher.find_all("ALLLEDGERENTRIES.LIST") + voucher.find_all("LEDGERENTRIES.LIST")
+ ledger_entries = voucher.find_all("ALLLEDGERENTRIES.LIST") + voucher.find_all(
+ "LEDGERENTRIES.LIST"
+ )
for entry in ledger_entries:
- account = {"account": encode_company_abbr(entry.LEDGERNAME.string.strip(), self.erpnext_company), "cost_center": self.default_cost_center}
+ account = {
+ "account": encode_company_abbr(entry.LEDGERNAME.string.strip(), self.erpnext_company),
+ "cost_center": self.default_cost_center,
+ }
if entry.ISPARTYLEDGER.string.strip() == "Yes":
party_details = get_party(entry.LEDGERNAME.string.strip())
if party_details:
@@ -438,7 +483,12 @@
return invoice
def get_voucher_items(voucher, doctype):
- inventory_entries = voucher.find_all("INVENTORYENTRIES.LIST") + voucher.find_all("ALLINVENTORYENTRIES.LIST") + voucher.find_all("INVENTORYENTRIESIN.LIST") + voucher.find_all("INVENTORYENTRIESOUT.LIST")
+ inventory_entries = (
+ voucher.find_all("INVENTORYENTRIES.LIST")
+ + voucher.find_all("ALLINVENTORYENTRIES.LIST")
+ + voucher.find_all("INVENTORYENTRIESIN.LIST")
+ + voucher.find_all("INVENTORYENTRIESOUT.LIST")
+ )
if doctype == "Sales Invoice":
account_field = "income_account"
elif doctype == "Purchase Invoice":
@@ -446,32 +496,41 @@
items = []
for entry in inventory_entries:
qty, uom = entry.ACTUALQTY.string.strip().split()
- items.append({
- "item_code": entry.STOCKITEMNAME.string.strip(),
- "description": entry.STOCKITEMNAME.string.strip(),
- "qty": qty.strip(),
- "uom": uom.strip(),
- "conversion_factor": 1,
- "price_list_rate": entry.RATE.string.strip().split("/")[0],
- "cost_center": self.default_cost_center,
- "warehouse": self.default_warehouse,
- account_field: encode_company_abbr(entry.find_all("ACCOUNTINGALLOCATIONS.LIST")[0].LEDGERNAME.string.strip(), self.erpnext_company),
- })
+ items.append(
+ {
+ "item_code": entry.STOCKITEMNAME.string.strip(),
+ "description": entry.STOCKITEMNAME.string.strip(),
+ "qty": qty.strip(),
+ "uom": uom.strip(),
+ "conversion_factor": 1,
+ "price_list_rate": entry.RATE.string.strip().split("/")[0],
+ "cost_center": self.default_cost_center,
+ "warehouse": self.default_warehouse,
+ account_field: encode_company_abbr(
+ entry.find_all("ACCOUNTINGALLOCATIONS.LIST")[0].LEDGERNAME.string.strip(),
+ self.erpnext_company,
+ ),
+ }
+ )
return items
def get_voucher_taxes(voucher):
- ledger_entries = voucher.find_all("ALLLEDGERENTRIES.LIST") + voucher.find_all("LEDGERENTRIES.LIST")
+ ledger_entries = voucher.find_all("ALLLEDGERENTRIES.LIST") + voucher.find_all(
+ "LEDGERENTRIES.LIST"
+ )
taxes = []
for entry in ledger_entries:
if entry.ISPARTYLEDGER.string.strip() == "No":
tax_account = encode_company_abbr(entry.LEDGERNAME.string.strip(), self.erpnext_company)
- taxes.append({
- "charge_type": "Actual",
- "account_head": tax_account,
- "description": tax_account,
- "tax_amount": entry.AMOUNT.string.strip(),
- "cost_center": self.default_cost_center,
- })
+ taxes.append(
+ {
+ "charge_type": "Actual",
+ "account_head": tax_account,
+ "description": tax_account,
+ "tax_amount": entry.AMOUNT.string.strip(),
+ "cost_center": self.default_cost_center,
+ }
+ )
return taxes
def get_party(party):
@@ -502,8 +561,11 @@
def _import_day_book_data(self):
def create_fiscal_years(vouchers):
from frappe.utils.data import add_years, getdate
+
earliest_date = getdate(min(voucher["posting_date"] for voucher in vouchers))
- oldest_year = frappe.get_all("Fiscal Year", fields=["year_start_date", "year_end_date"], order_by="year_start_date")[0]
+ oldest_year = frappe.get_all(
+ "Fiscal Year", fields=["year_start_date", "year_end_date"], order_by="year_start_date"
+ )[0]
while earliest_date < oldest_year.year_start_date:
new_year = frappe.get_doc({"doctype": "Fiscal Year"})
new_year.year_start_date = add_years(oldest_year.year_start_date, -1)
@@ -520,32 +582,46 @@
"fieldtype": "Data",
"fieldname": "tally_guid",
"read_only": 1,
- "label": "Tally GUID"
+ "label": "Tally GUID",
}
tally_voucher_no_df = {
"fieldtype": "Data",
"fieldname": "tally_voucher_no",
"read_only": 1,
- "label": "Tally Voucher Number"
+ "label": "Tally Voucher Number",
}
for df in [tally_guid_df, tally_voucher_no_df]:
for doctype in doctypes:
create_custom_field(doctype, df)
def create_price_list():
- frappe.get_doc({
- "doctype": "Price List",
- "price_list_name": "Tally Price List",
- "selling": 1,
- "buying": 1,
- "enabled": 1,
- "currency": "INR"
- }).insert()
+ frappe.get_doc(
+ {
+ "doctype": "Price List",
+ "price_list_name": "Tally Price List",
+ "selling": 1,
+ "buying": 1,
+ "enabled": 1,
+ "currency": "INR",
+ }
+ ).insert()
try:
- frappe.db.set_value("Account", encode_company_abbr(self.tally_creditors_account, self.erpnext_company), "account_type", "Payable")
- frappe.db.set_value("Account", encode_company_abbr(self.tally_debtors_account, self.erpnext_company), "account_type", "Receivable")
- frappe.db.set_value("Company", self.erpnext_company, "round_off_account", self.default_round_off_account)
+ frappe.db.set_value(
+ "Account",
+ encode_company_abbr(self.tally_creditors_account, self.erpnext_company),
+ "account_type",
+ "Payable",
+ )
+ frappe.db.set_value(
+ "Account",
+ encode_company_abbr(self.tally_debtors_account, self.erpnext_company),
+ "account_type",
+ "Receivable",
+ )
+ frappe.db.set_value(
+ "Company", self.erpnext_company, "round_off_account", self.default_round_off_account
+ )
vouchers_file = frappe.get_doc("File", {"file_url": self.vouchers})
vouchers = json.loads(vouchers_file.get_content())
@@ -560,7 +636,16 @@
for index in range(0, total, VOUCHER_CHUNK_SIZE):
if index + VOUCHER_CHUNK_SIZE >= total:
is_last = True
- frappe.enqueue_doc(self.doctype, self.name, "_import_vouchers", queue="long", timeout=3600, start=index+1, total=total, is_last=is_last)
+ frappe.enqueue_doc(
+ self.doctype,
+ self.name,
+ "_import_vouchers",
+ queue="long",
+ timeout=3600,
+ start=index + 1,
+ total=total,
+ is_last=is_last,
+ )
except Exception:
self.log()
@@ -572,7 +657,7 @@
frappe.flags.in_migrate = True
vouchers_file = frappe.get_doc("File", {"file_url": self.vouchers})
vouchers = json.loads(vouchers_file.get_content())
- chunk = vouchers[start: start + VOUCHER_CHUNK_SIZE]
+ chunk = vouchers[start : start + VOUCHER_CHUNK_SIZE]
for index, voucher in enumerate(chunk, start=start):
try:
@@ -617,17 +702,22 @@
if sys.exc_info()[1].__class__ != frappe.DuplicateEntryError:
failed_import_log = json.loads(self.failed_import_log)
doc = data.as_dict()
- failed_import_log.append({
- "doc": doc,
- "exc": traceback.format_exc()
- })
- self.failed_import_log = json.dumps(failed_import_log, separators=(',', ':'))
+ failed_import_log.append({"doc": doc, "exc": traceback.format_exc()})
+ self.failed_import_log = json.dumps(failed_import_log, separators=(",", ":"))
self.save()
frappe.db.commit()
else:
data = data or self.status
- message = "\n".join(["Data:", json.dumps(data, default=str, indent=4), "--" * 50, "\nException:", traceback.format_exc()])
+ message = "\n".join(
+ [
+ "Data:",
+ json.dumps(data, default=str, indent=4),
+ "--" * 50,
+ "\nException:",
+ traceback.format_exc(),
+ ]
+ )
return frappe.log_error(title="Tally Migration Error", message=message)
def set_status(self, status=""):
diff --git a/erpnext/erpnext_integrations/doctype/taxjar_settings/taxjar_settings.py b/erpnext/erpnext_integrations/doctype/taxjar_settings/taxjar_settings.py
index d4bbe88..2148863 100644
--- a/erpnext/erpnext_integrations/doctype/taxjar_settings/taxjar_settings.py
+++ b/erpnext/erpnext_integrations/doctype/taxjar_settings/taxjar_settings.py
@@ -14,27 +14,31 @@
class TaxJarSettings(Document):
-
def on_update(self):
TAXJAR_CREATE_TRANSACTIONS = self.taxjar_create_transactions
TAXJAR_CALCULATE_TAX = self.taxjar_calculate_tax
TAXJAR_SANDBOX_MODE = self.is_sandbox
- fields_already_exist = frappe.db.exists('Custom Field', {'dt': ('in', ['Item','Sales Invoice Item']), 'fieldname':'product_tax_category'})
- fields_hidden = frappe.get_value('Custom Field', {'dt': ('in', ['Sales Invoice Item'])}, 'hidden')
+ fields_already_exist = frappe.db.exists(
+ "Custom Field",
+ {"dt": ("in", ["Item", "Sales Invoice Item"]), "fieldname": "product_tax_category"},
+ )
+ fields_hidden = frappe.get_value(
+ "Custom Field", {"dt": ("in", ["Sales Invoice Item"])}, "hidden"
+ )
- if (TAXJAR_CREATE_TRANSACTIONS or TAXJAR_CALCULATE_TAX or TAXJAR_SANDBOX_MODE):
+ if TAXJAR_CREATE_TRANSACTIONS or TAXJAR_CALCULATE_TAX or TAXJAR_SANDBOX_MODE:
if not fields_already_exist:
add_product_tax_categories()
make_custom_fields()
add_permissions()
- frappe.enqueue('erpnext.regional.united_states.setup.add_product_tax_categories', now=False)
+ frappe.enqueue("erpnext.regional.united_states.setup.add_product_tax_categories", now=False)
elif fields_already_exist and fields_hidden:
- toggle_tax_category_fields(hidden='0')
+ toggle_tax_category_fields(hidden="0")
elif fields_already_exist:
- toggle_tax_category_fields(hidden='1')
+ toggle_tax_category_fields(hidden="1")
def validate(self):
self.calculate_taxes_validation_for_create_transactions()
@@ -46,54 +50,97 @@
new_nexus_list = [frappe._dict(address) for address in nexus]
- self.set('nexus', [])
- self.set('nexus', new_nexus_list)
+ self.set("nexus", [])
+ self.set("nexus", new_nexus_list)
self.save()
def calculate_taxes_validation_for_create_transactions(self):
if not self.taxjar_calculate_tax and (self.taxjar_create_transactions or self.is_sandbox):
- frappe.throw(frappe._('Before enabling <b>Create Transaction</b> or <b>Sandbox Mode</b>, you need to check the <b>Enable Tax Calculation</b> box'))
+ frappe.throw(
+ frappe._(
+ "Before enabling <b>Create Transaction</b> or <b>Sandbox Mode</b>, you need to check the <b>Enable Tax Calculation</b> box"
+ )
+ )
def toggle_tax_category_fields(hidden):
- frappe.set_value('Custom Field', {'dt':'Sales Invoice Item', 'fieldname':'product_tax_category'}, 'hidden', hidden)
- frappe.set_value('Custom Field', {'dt':'Item', 'fieldname':'product_tax_category'}, 'hidden', hidden)
+ frappe.set_value(
+ "Custom Field",
+ {"dt": "Sales Invoice Item", "fieldname": "product_tax_category"},
+ "hidden",
+ hidden,
+ )
+ frappe.set_value(
+ "Custom Field", {"dt": "Item", "fieldname": "product_tax_category"}, "hidden", hidden
+ )
def add_product_tax_categories():
- with open(os.path.join(os.path.dirname(__file__), 'product_tax_category_data.json'), 'r') as f:
+ with open(os.path.join(os.path.dirname(__file__), "product_tax_category_data.json"), "r") as f:
tax_categories = json.loads(f.read())
- create_tax_categories(tax_categories['categories'])
+ create_tax_categories(tax_categories["categories"])
+
def create_tax_categories(data):
for d in data:
- if not frappe.db.exists('Product Tax Category',{'product_tax_code':d.get('product_tax_code')}):
- tax_category = frappe.new_doc('Product Tax Category')
+ if not frappe.db.exists("Product Tax Category", {"product_tax_code": d.get("product_tax_code")}):
+ tax_category = frappe.new_doc("Product Tax Category")
tax_category.description = d.get("description")
tax_category.product_tax_code = d.get("product_tax_code")
tax_category.category_name = d.get("name")
tax_category.db_insert()
+
def make_custom_fields(update=True):
custom_fields = {
- 'Sales Invoice Item': [
- dict(fieldname='product_tax_category', fieldtype='Link', insert_after='description', options='Product Tax Category',
- label='Product Tax Category', fetch_from='item_code.product_tax_category'),
- dict(fieldname='tax_collectable', fieldtype='Currency', insert_after='net_amount',
- label='Tax Collectable', read_only=1, options='currency'),
- dict(fieldname='taxable_amount', fieldtype='Currency', insert_after='tax_collectable',
- label='Taxable Amount', read_only=1, options='currency')
+ "Sales Invoice Item": [
+ dict(
+ fieldname="product_tax_category",
+ fieldtype="Link",
+ insert_after="description",
+ options="Product Tax Category",
+ label="Product Tax Category",
+ fetch_from="item_code.product_tax_category",
+ ),
+ dict(
+ fieldname="tax_collectable",
+ fieldtype="Currency",
+ insert_after="net_amount",
+ label="Tax Collectable",
+ read_only=1,
+ options="currency",
+ ),
+ dict(
+ fieldname="taxable_amount",
+ fieldtype="Currency",
+ insert_after="tax_collectable",
+ label="Taxable Amount",
+ read_only=1,
+ options="currency",
+ ),
],
- 'Item': [
- dict(fieldname='product_tax_category', fieldtype='Link', insert_after='item_group', options='Product Tax Category',
- label='Product Tax Category')
- ]
+ "Item": [
+ dict(
+ fieldname="product_tax_category",
+ fieldtype="Link",
+ insert_after="item_group",
+ options="Product Tax Category",
+ label="Product Tax Category",
+ )
+ ],
}
create_custom_fields(custom_fields, update=update)
+
def add_permissions():
doctype = "Product Tax Category"
- for role in ('Accounts Manager', 'Accounts User', 'System Manager','Item Manager', 'Stock Manager'):
+ for role in (
+ "Accounts Manager",
+ "Accounts User",
+ "System Manager",
+ "Item Manager",
+ "Stock Manager",
+ ):
add_permission(doctype, role, 0)
- update_permission_property(doctype, role, 0, 'write', 1)
- update_permission_property(doctype, role, 0, 'create', 1)
+ update_permission_property(doctype, role, 0, "write", 1)
+ update_permission_property(doctype, role, 0, "create", 1)
diff --git a/erpnext/erpnext_integrations/doctype/woocommerce_settings/woocommerce_settings.py b/erpnext/erpnext_integrations/doctype/woocommerce_settings/woocommerce_settings.py
index 309d2cb..2e18776 100644
--- a/erpnext/erpnext_integrations/doctype/woocommerce_settings/woocommerce_settings.py
+++ b/erpnext/erpnext_integrations/doctype/woocommerce_settings/woocommerce_settings.py
@@ -22,11 +22,23 @@
custom_fields = {}
# create
for doctype in ["Customer", "Sales Order", "Item", "Address"]:
- df = dict(fieldname='woocommerce_id', label='Woocommerce ID', fieldtype='Data', read_only=1, print_hide=1)
+ df = dict(
+ fieldname="woocommerce_id",
+ label="Woocommerce ID",
+ fieldtype="Data",
+ read_only=1,
+ print_hide=1,
+ )
create_custom_field(doctype, df)
for doctype in ["Customer", "Address"]:
- df = dict(fieldname='woocommerce_email', label='Woocommerce Email', fieldtype='Data', read_only=1, print_hide=1)
+ df = dict(
+ fieldname="woocommerce_email",
+ label="Woocommerce Email",
+ fieldtype="Data",
+ read_only=1,
+ print_hide=1,
+ )
create_custom_field(doctype, df)
if not frappe.get_value("Item Group", {"name": _("WooCommerce Products")}):
@@ -58,21 +70,21 @@
# for CI Test to work
url = "http://localhost:8000"
- server_url = '{uri.scheme}://{uri.netloc}'.format(
- uri=urlparse(url)
- )
+ server_url = "{uri.scheme}://{uri.netloc}".format(uri=urlparse(url))
delivery_url = server_url + endpoint
self.endpoint = delivery_url
+
@frappe.whitelist()
def generate_secret():
woocommerce_settings = frappe.get_doc("Woocommerce Settings")
woocommerce_settings.secret = frappe.generate_hash()
woocommerce_settings.save()
+
@frappe.whitelist()
def get_series():
return {
- "sales_order_series" : frappe.get_meta("Sales Order").get_options("naming_series") or "SO-WOO-",
+ "sales_order_series": frappe.get_meta("Sales Order").get_options("naming_series") or "SO-WOO-",
}
diff --git a/erpnext/erpnext_integrations/exotel_integration.py b/erpnext/erpnext_integrations/exotel_integration.py
index 167fcb7..1f5df67 100644
--- a/erpnext/erpnext_integrations/exotel_integration.py
+++ b/erpnext/erpnext_integrations/exotel_integration.py
@@ -6,15 +6,17 @@
# api/method/erpnext.erpnext_integrations.exotel_integration.handle_end_call
# api/method/erpnext.erpnext_integrations.exotel_integration.handle_missed_call
+
@frappe.whitelist(allow_guest=True)
def handle_incoming_call(**kwargs):
try:
exotel_settings = get_exotel_settings()
- if not exotel_settings.enabled: return
+ if not exotel_settings.enabled:
+ return
call_payload = kwargs
- status = call_payload.get('Status')
- if status == 'free':
+ status = call_payload.get("Status")
+ if status == "free":
return
call_log = get_call_log(call_payload)
@@ -24,87 +26,100 @@
update_call_log(call_payload, call_log=call_log)
except Exception as e:
frappe.db.rollback()
- frappe.log_error(title=_('Error in Exotel incoming call'))
+ frappe.log_error(title=_("Error in Exotel incoming call"))
frappe.db.commit()
+
@frappe.whitelist(allow_guest=True)
def handle_end_call(**kwargs):
- update_call_log(kwargs, 'Completed')
+ update_call_log(kwargs, "Completed")
+
@frappe.whitelist(allow_guest=True)
def handle_missed_call(**kwargs):
- update_call_log(kwargs, 'Missed')
+ update_call_log(kwargs, "Missed")
-def update_call_log(call_payload, status='Ringing', call_log=None):
+
+def update_call_log(call_payload, status="Ringing", call_log=None):
call_log = call_log or get_call_log(call_payload)
if call_log:
call_log.status = status
- call_log.to = call_payload.get('DialWhomNumber')
- call_log.duration = call_payload.get('DialCallDuration') or 0
- call_log.recording_url = call_payload.get('RecordingUrl')
+ call_log.to = call_payload.get("DialWhomNumber")
+ call_log.duration = call_payload.get("DialCallDuration") or 0
+ call_log.recording_url = call_payload.get("RecordingUrl")
call_log.save(ignore_permissions=True)
frappe.db.commit()
return call_log
+
def get_call_log(call_payload):
- call_log = frappe.get_all('Call Log', {
- 'id': call_payload.get('CallSid'),
- }, limit=1)
+ call_log = frappe.get_all(
+ "Call Log",
+ {
+ "id": call_payload.get("CallSid"),
+ },
+ limit=1,
+ )
if call_log:
- return frappe.get_doc('Call Log', call_log[0].name)
+ return frappe.get_doc("Call Log", call_log[0].name)
+
def create_call_log(call_payload):
- call_log = frappe.new_doc('Call Log')
- call_log.id = call_payload.get('CallSid')
- call_log.to = call_payload.get('DialWhomNumber')
- call_log.medium = call_payload.get('To')
- call_log.status = 'Ringing'
- setattr(call_log, 'from', call_payload.get('CallFrom'))
+ call_log = frappe.new_doc("Call Log")
+ call_log.id = call_payload.get("CallSid")
+ call_log.to = call_payload.get("DialWhomNumber")
+ call_log.medium = call_payload.get("To")
+ call_log.status = "Ringing"
+ setattr(call_log, "from", call_payload.get("CallFrom"))
call_log.save(ignore_permissions=True)
frappe.db.commit()
return call_log
+
@frappe.whitelist()
def get_call_status(call_id):
- endpoint = get_exotel_endpoint('Calls/{call_id}.json'.format(call_id=call_id))
+ endpoint = get_exotel_endpoint("Calls/{call_id}.json".format(call_id=call_id))
response = requests.get(endpoint)
- status = response.json().get('Call', {}).get('Status')
+ status = response.json().get("Call", {}).get("Status")
return status
+
@frappe.whitelist()
def make_a_call(from_number, to_number, caller_id):
- endpoint = get_exotel_endpoint('Calls/connect.json?details=true')
- response = requests.post(endpoint, data={
- 'From': from_number,
- 'To': to_number,
- 'CallerId': caller_id
- })
+ endpoint = get_exotel_endpoint("Calls/connect.json?details=true")
+ response = requests.post(
+ endpoint, data={"From": from_number, "To": to_number, "CallerId": caller_id}
+ )
return response.json()
+
def get_exotel_settings():
- return frappe.get_single('Exotel Settings')
+ return frappe.get_single("Exotel Settings")
+
def whitelist_numbers(numbers, caller_id):
- endpoint = get_exotel_endpoint('CustomerWhitelist')
- response = requests.post(endpoint, data={
- 'VirtualNumber': caller_id,
- 'Number': numbers,
- })
+ endpoint = get_exotel_endpoint("CustomerWhitelist")
+ response = requests.post(
+ endpoint,
+ data={
+ "VirtualNumber": caller_id,
+ "Number": numbers,
+ },
+ )
return response
+
def get_all_exophones():
- endpoint = get_exotel_endpoint('IncomingPhoneNumbers')
+ endpoint = get_exotel_endpoint("IncomingPhoneNumbers")
response = requests.post(endpoint)
return response
+
def get_exotel_endpoint(action):
settings = get_exotel_settings()
- return 'https://{api_key}:{api_token}@api.exotel.com/v1/Accounts/{sid}/{action}'.format(
- api_key=settings.api_key,
- api_token=settings.api_token,
- sid=settings.account_sid,
- action=action
+ return "https://{api_key}:{api_token}@api.exotel.com/v1/Accounts/{sid}/{action}".format(
+ api_key=settings.api_key, api_token=settings.api_token, sid=settings.account_sid, action=action
)
diff --git a/erpnext/erpnext_integrations/stripe_integration.py b/erpnext/erpnext_integrations/stripe_integration.py
index 502cb5f..b12adc1 100644
--- a/erpnext/erpnext_integrations/stripe_integration.py
+++ b/erpnext/erpnext_integrations/stripe_integration.py
@@ -16,17 +16,21 @@
try:
stripe_settings.integration_request = create_request_log(stripe_settings.data, "Host", "Stripe")
- stripe_settings.payment_plans = frappe.get_doc("Payment Request", stripe_settings.data.reference_docname).subscription_plans
+ stripe_settings.payment_plans = frappe.get_doc(
+ "Payment Request", stripe_settings.data.reference_docname
+ ).subscription_plans
return create_subscription_on_stripe(stripe_settings)
except Exception:
frappe.log_error(frappe.get_traceback())
- return{
+ return {
"redirect_to": frappe.redirect_to_message(
- _('Server Error'),
- _("It seems that there is an issue with the server's stripe configuration. In case of failure, the amount will get refunded to your account.")
+ _("Server Error"),
+ _(
+ "It seems that there is an issue with the server's stripe configuration. In case of failure, the amount will get refunded to your account."
+ ),
),
- "status": 401
+ "status": 401,
}
@@ -40,20 +44,20 @@
customer = stripe.Customer.create(
source=stripe_settings.data.stripe_token_id,
description=stripe_settings.data.payer_name,
- email=stripe_settings.data.payer_email
+ email=stripe_settings.data.payer_email,
)
subscription = stripe.Subscription.create(customer=customer, items=items)
if subscription.status == "active":
- stripe_settings.integration_request.db_set('status', 'Completed', update_modified=False)
+ stripe_settings.integration_request.db_set("status", "Completed", update_modified=False)
stripe_settings.flags.status_changed_to = "Completed"
else:
- stripe_settings.integration_request.db_set('status', 'Failed', update_modified=False)
- frappe.log_error('Subscription N°: ' + subscription.id, 'Stripe Payment not completed')
+ stripe_settings.integration_request.db_set("status", "Failed", update_modified=False)
+ frappe.log_error("Subscription N°: " + subscription.id, "Stripe Payment not completed")
except Exception:
- stripe_settings.integration_request.db_set('status', 'Failed', update_modified=False)
+ stripe_settings.integration_request.db_set("status", "Failed", update_modified=False)
frappe.log_error(frappe.get_traceback())
return stripe_settings.finalize_request()
diff --git a/erpnext/erpnext_integrations/taxjar_integration.py b/erpnext/erpnext_integrations/taxjar_integration.py
index 14c86d5..b8893aa 100644
--- a/erpnext/erpnext_integrations/taxjar_integration.py
+++ b/erpnext/erpnext_integrations/taxjar_integration.py
@@ -8,14 +8,92 @@
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']
-
+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():
@@ -30,14 +108,14 @@
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'
- })
+ 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")
+ TAXJAR_CREATE_TRANSACTIONS = frappe.db.get_single_value(
+ "TaxJar Settings", "taxjar_create_transactions"
+ )
"""Create an order transaction in TaxJar"""
@@ -60,10 +138,10 @@
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']
+ 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:
@@ -78,7 +156,9 @@
def delete_transaction(doc, method):
"""Delete an existing TaxJar order transaction"""
- TAXJAR_CREATE_TRANSACTIONS = frappe.db.get_single_value("TaxJar Settings", "taxjar_create_transactions")
+ TAXJAR_CREATE_TRANSACTIONS = frappe.db.get_single_value(
+ "TaxJar Settings", "taxjar_create_transactions"
+ )
if not TAXJAR_CREATE_TRANSACTIONS:
return
@@ -109,10 +189,10 @@
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')
+ 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')
+ to_shipping_state = get_state_code(to_address, "Shipping")
tax_dict = {
"from_country": from_country_code,
@@ -128,10 +208,11 @@
"shipping": shipping,
"amount": doc.net_total,
"plugin": "erpnext",
- "line_items": line_items
+ "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)
@@ -142,21 +223,21 @@
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')
+ 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')
- })
+ 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")
@@ -164,7 +245,7 @@
if not TAXJAR_CALCULATE_TAX:
return
- if get_region(doc.company) != 'United States':
+ if get_region(doc.company) != "United States":
return
if not doc.items:
@@ -197,22 +278,26 @@
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
- })
+ 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.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"]}):
+ 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)
@@ -222,13 +307,17 @@
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") \
+ 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:
@@ -240,6 +329,7 @@
else:
return False
+
def validate_tax_request(tax_dict):
"""Return the sales tax that should be collected for a given order."""
@@ -283,9 +373,12 @@
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"))
+ 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
@@ -306,7 +399,7 @@
except LookupError:
frappe.throw(_(error_message))
else:
- return lookup_state.code.split('-')[1]
+ return lookup_state.code.split("-")[1]
def sanitize_error_response(response):
@@ -317,7 +410,7 @@
"to zip": "Zipcode",
"to city": "City",
"to state": "State",
- "to country": "Country"
+ "to country": "Country",
}
for k, v in sanitized_responses.items():
diff --git a/erpnext/erpnext_integrations/utils.py b/erpnext/erpnext_integrations/utils.py
index 30d3948..981486e 100644
--- a/erpnext/erpnext_integrations/utils.py
+++ b/erpnext/erpnext_integrations/utils.py
@@ -9,28 +9,24 @@
from erpnext import get_default_company
-def validate_webhooks_request(doctype, hmac_key, secret_key='secret'):
+def validate_webhooks_request(doctype, hmac_key, secret_key="secret"):
def innerfn(fn):
settings = frappe.get_doc(doctype)
if frappe.request and settings and settings.get(secret_key) and not frappe.flags.in_test:
sig = base64.b64encode(
- hmac.new(
- settings.get(secret_key).encode('utf8'),
- frappe.request.data,
- hashlib.sha256
- ).digest()
+ hmac.new(settings.get(secret_key).encode("utf8"), frappe.request.data, hashlib.sha256).digest()
)
- if frappe.request.data and \
- not sig == bytes(frappe.get_request_header(hmac_key).encode()):
- frappe.throw(_("Unverified Webhook Data"))
+ if frappe.request.data and not sig == bytes(frappe.get_request_header(hmac_key).encode()):
+ frappe.throw(_("Unverified Webhook Data"))
frappe.set_user(settings.modified_by)
return fn
return innerfn
+
def get_webhook_address(connector_name, method, exclude_uri=False, force_https=False):
endpoint = "erpnext.erpnext_integrations.connectors.{0}.{1}".format(connector_name, method)
@@ -50,34 +46,40 @@
return server_url
+
def create_mode_of_payment(gateway, payment_type="General"):
- payment_gateway_account = frappe.db.get_value("Payment Gateway Account", {
- "payment_gateway": gateway
- }, ['payment_account'])
+ payment_gateway_account = frappe.db.get_value(
+ "Payment Gateway Account", {"payment_gateway": gateway}, ["payment_account"]
+ )
mode_of_payment = frappe.db.exists("Mode of Payment", gateway)
if not mode_of_payment and payment_gateway_account:
- mode_of_payment = frappe.get_doc({
- "doctype": "Mode of Payment",
- "mode_of_payment": gateway,
- "enabled": 1,
- "type": payment_type,
- "accounts": [{
- "doctype": "Mode of Payment Account",
- "company": get_default_company(),
- "default_account": payment_gateway_account
- }]
- })
+ mode_of_payment = frappe.get_doc(
+ {
+ "doctype": "Mode of Payment",
+ "mode_of_payment": gateway,
+ "enabled": 1,
+ "type": payment_type,
+ "accounts": [
+ {
+ "doctype": "Mode of Payment Account",
+ "company": get_default_company(),
+ "default_account": payment_gateway_account,
+ }
+ ],
+ }
+ )
mode_of_payment.insert(ignore_permissions=True)
return mode_of_payment
elif mode_of_payment:
return frappe.get_doc("Mode of Payment", mode_of_payment)
+
def get_tracking_url(carrier, tracking_number):
# Return the formatted Tracking URL.
- tracking_url = ''
- url_reference = frappe.get_value('Parcel Service', carrier, 'url_reference')
+ tracking_url = ""
+ url_reference = frappe.get_value("Parcel Service", carrier, "url_reference")
if url_reference:
- tracking_url = frappe.render_template(url_reference, {'tracking_number': tracking_number})
+ tracking_url = frappe.render_template(url_reference, {"tracking_number": tracking_number})
return tracking_url