brotherton-erpnext/erpnext/erpnext_integrations/utils.py

76 lines
2.2 KiB
Python

from __future__ import unicode_literals
import frappe
from frappe import _
import base64, hashlib, hmac
from six.moves.urllib.parse import urlparse
from erpnext import get_default_company
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()
)
if frappe.request.data and \
frappe.get_request_header(hmac_key) 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)
if exclude_uri:
return endpoint
try:
url = frappe.request.url
except RuntimeError:
url = "http://localhost:8000"
url_data = urlparse(url)
scheme = "https" if force_https else url_data.scheme
netloc = url_data.netloc
server_url = f"{scheme}://{netloc}/api/method/{endpoint}"
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'])
if not frappe.db.exists("Mode of Payment", gateway) 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.insert(ignore_permissions=True)
def get_tracking_url(carrier, tracking_number):
# Return the formatted Tracking URL.
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})
return tracking_url