mirror of
https://github.com/frappe/erpnext.git
synced 2026-06-03 12:19:12 +00:00
chore: remove unwanted fuctions
This commit is contained in:
Saqib Ansari
@@ -47,229 +47,6 @@ def validate_einvoice_fields(doc):
|
||||
elif doc.docstatus == 2 and doc._action == 'cancel' and not doc.irn_cancelled:
|
||||
frappe.throw(_('You must cancel IRN before cancelling the document.'), title=_('Cancel Not Allowed'))
|
||||
|
||||
def get_credentials():
|
||||
doc = frappe.get_doc('E Invoice Settings')
|
||||
if not doc.enable:
|
||||
frappe.throw(_("To setup E Invoicing you need to enable E Invoice Settings first."), title=_("E Invoicing Disabled"))
|
||||
|
||||
if not doc.token_expiry or time_diff_in_seconds(now_datetime(), doc.token_expiry) > 5.0:
|
||||
fetch_token(doc)
|
||||
doc.load_from_db()
|
||||
|
||||
return doc
|
||||
|
||||
def rsa_encrypt(msg, key):
|
||||
if not (isinstance(msg, bytes) or isinstance(msg, bytearray)):
|
||||
msg = str.encode(msg)
|
||||
|
||||
rsa_pub_key = RSA.import_key(key)
|
||||
cipher = PKCS1_v1_5.new(rsa_pub_key)
|
||||
enc_msg = cipher.encrypt(msg)
|
||||
b64_enc_msg = base64.b64encode(enc_msg)
|
||||
return b64_enc_msg.decode()
|
||||
|
||||
def aes_decrypt(enc_msg, key):
|
||||
encode_as_b64 = True
|
||||
if not (isinstance(key, bytes) or isinstance(key, bytearray)):
|
||||
key = base64.b64decode(key)
|
||||
encode_as_b64 = False
|
||||
|
||||
cipher = AES.new(key, AES.MODE_ECB)
|
||||
b64_enc_msg = base64.b64decode(enc_msg)
|
||||
msg_bytes = cipher.decrypt(b64_enc_msg)
|
||||
msg_bytes = unpad(msg_bytes, AES.block_size) # due to ECB/PKCS5Padding
|
||||
if encode_as_b64:
|
||||
msg_bytes = base64.b64encode(msg_bytes)
|
||||
return msg_bytes.decode()
|
||||
|
||||
def aes_encrypt(msg, key):
|
||||
if not (isinstance(key, bytes) or isinstance(key, bytearray)):
|
||||
key = base64.b64decode(key)
|
||||
|
||||
cipher = AES.new(key, AES.MODE_ECB)
|
||||
bytes_msg = str.encode(msg)
|
||||
padded_bytes_msg = pad(bytes_msg, AES.block_size)
|
||||
enc_msg = cipher.encrypt(padded_bytes_msg)
|
||||
b64_enc_msg = base64.b64encode(enc_msg)
|
||||
return b64_enc_msg.decode()
|
||||
|
||||
def jwt_decrypt(token):
|
||||
return jwt.decode(token, verify=False)
|
||||
|
||||
def get_header(creds):
|
||||
headers = { 'content-type': 'application/json' }
|
||||
headers.update(dict(client_id=creds.client_id, client_secret=creds.client_secret, user_name=creds.username))
|
||||
headers.update(dict(Gstin=creds.gstin, AuthToken=creds.auth_token))
|
||||
return headers
|
||||
|
||||
@frappe.whitelist()
|
||||
def fetch_token(credentials=None):
|
||||
if not credentials:
|
||||
credentials = frappe.get_doc('E Invoice Settings')
|
||||
|
||||
endpoint = 'https://einv-apisandbox.nic.in/eivital/v1.03/auth'
|
||||
headers = { 'content-type': 'application/json' }
|
||||
headers.update(dict(client_id=credentials.client_id, client_secret=credentials.client_secret))
|
||||
payload = dict(UserName=credentials.username, ForceRefreshAccessToken=bool(credentials.auto_refresh_token))
|
||||
|
||||
appkey = bytearray(os.urandom(32))
|
||||
enc_appkey = rsa_encrypt(appkey, credentials.public_key)
|
||||
|
||||
password = credentials.get_password(fieldname='password')
|
||||
enc_password = rsa_encrypt(password, credentials.public_key)
|
||||
|
||||
payload.update(dict(Password=enc_password, AppKey=enc_appkey))
|
||||
|
||||
res = make_post_request(endpoint, headers=headers, data=json.dumps({ 'data': payload }))
|
||||
handle_err_response(res)
|
||||
|
||||
auth_token, token_expiry, sek = extract_token_and_sek(res, appkey)
|
||||
|
||||
credentials.auth_token = auth_token
|
||||
credentials.token_expiry = get_datetime(token_expiry)
|
||||
credentials.sek = sek
|
||||
credentials.save()
|
||||
|
||||
def extract_token_and_sek(response, appkey):
|
||||
data = response.get('Data')
|
||||
auth_token = data.get('AuthToken')
|
||||
token_expiry = data.get('TokenExpiry')
|
||||
enc_sek = data.get('Sek')
|
||||
sek = aes_decrypt(enc_sek, appkey)
|
||||
return auth_token, token_expiry, sek
|
||||
|
||||
def attach_signed_invoice(doctype, name, data):
|
||||
f = frappe.get_doc({
|
||||
'doctype': 'File',
|
||||
'file_name': 'E-INV--{}.json'.format(name),
|
||||
'attached_to_doctype': doctype,
|
||||
'attached_to_name': name,
|
||||
'content': json.dumps(data),
|
||||
'is_private': True
|
||||
}).insert()
|
||||
|
||||
def get_gstin_details(gstin):
|
||||
credentials = get_credentials()
|
||||
|
||||
endpoint = 'https://einv-apisandbox.nic.in/eivital/v1.03/Master/gstin/{gstin}'.format(gstin=gstin)
|
||||
headers = get_header(credentials)
|
||||
|
||||
res = make_get_request(endpoint, headers=headers)
|
||||
handle_err_response(res)
|
||||
|
||||
enc_details = res.get('Data')
|
||||
json_str = aes_decrypt(enc_details, credentials.sek)
|
||||
details = json.loads(json_str)
|
||||
|
||||
return details
|
||||
|
||||
@frappe.whitelist()
|
||||
def generate_irn(doctype, name):
|
||||
endpoint = 'https://einv-apisandbox.nic.in/eicore/v1.03/Invoice'
|
||||
credentials = get_credentials()
|
||||
headers = get_header(credentials)
|
||||
|
||||
einvoice = make_einvoice(doctype, name)
|
||||
einvoice = json.dumps(einvoice)
|
||||
|
||||
enc_einvoice_json = aes_encrypt(einvoice, credentials.sek)
|
||||
payload = dict(Data=enc_einvoice_json)
|
||||
|
||||
res = make_post_request(endpoint, headers=headers, data=json.dumps(payload))
|
||||
res = handle_err_response(res)
|
||||
|
||||
enc_json = res.get('Data')
|
||||
json_str = aes_decrypt(enc_json, credentials.sek)
|
||||
|
||||
signed_einvoice = json.loads(json_str)
|
||||
decrypt_irn_response(signed_einvoice)
|
||||
|
||||
update_einvoice_fields(doctype, name, signed_einvoice)
|
||||
|
||||
attach_qrcode_image(doctype, name)
|
||||
attach_signed_invoice(doctype, name, signed_einvoice['DecryptedSignedInvoice'])
|
||||
|
||||
return signed_einvoice
|
||||
|
||||
def get_irn_details(irn):
|
||||
credentials = get_credentials()
|
||||
|
||||
endpoint = 'https://einv-apisandbox.nic.in/eicore/v1.03/Invoice/irn/{irn}'.format(irn=irn)
|
||||
headers = get_header(credentials)
|
||||
|
||||
res = make_get_request(endpoint, headers=headers)
|
||||
handle_err_response(res)
|
||||
|
||||
return res
|
||||
|
||||
@frappe.whitelist()
|
||||
def cancel_irn(doctype, name, irn, reason, remark=''):
|
||||
credentials = get_credentials()
|
||||
|
||||
endpoint = 'https://einv-apisandbox.nic.in/eicore/v1.03/Invoice/Cancel'
|
||||
headers = get_header(credentials)
|
||||
|
||||
cancel_einv = json.dumps(dict(Irn=irn, CnlRsn=reason, CnlRem=remark))
|
||||
enc_json = aes_encrypt(cancel_einv, credentials.sek)
|
||||
payload = dict(Data=enc_json)
|
||||
|
||||
res = make_post_request(endpoint, headers=headers, data=json.dumps(payload))
|
||||
handle_err_response(res)
|
||||
|
||||
frappe.db.set_value(doctype, name, 'irn_cancelled', 1)
|
||||
|
||||
return res
|
||||
|
||||
@frappe.whitelist()
|
||||
def cancel_eway_bill(doctype, name, eway_bill, reason, remark=''):
|
||||
credentials = get_credentials()
|
||||
endpoint = 'https://einv-apisandbox.nic.in/ewaybillapi/v1.03/ewayapi'
|
||||
headers = get_header(credentials)
|
||||
|
||||
cancel_eway_bill_json = json.dumps(dict(ewbNo=eway_bill, cancelRsnCode=reason, cancelRmrk=remark))
|
||||
enc_json = aes_encrypt(cancel_eway_bill_json, credentials.sek)
|
||||
payload = dict(action='CANEWB', Data=enc_json)
|
||||
|
||||
res = make_post_request(endpoint, headers=headers, data=json.dumps(payload))
|
||||
handle_err_response(res)
|
||||
|
||||
frappe.db.set_value(doctype, name, 'ewaybill', '')
|
||||
frappe.db.set_value(doctype, name, 'eway_bill_cancelled', 1)
|
||||
|
||||
return res
|
||||
|
||||
def decrypt_irn_response(data):
|
||||
enc_signed_invoice = data['SignedInvoice']
|
||||
enc_signed_qr_code = data['SignedQRCode']
|
||||
signed_invoice = jwt_decrypt(enc_signed_invoice)['data']
|
||||
signed_qr_code = jwt_decrypt(enc_signed_qr_code)['data']
|
||||
data['DecryptedSignedInvoice'] = json.loads(signed_invoice)
|
||||
data['DecryptedSignedQRCode'] = json.loads(signed_qr_code)
|
||||
|
||||
def handle_err_response(response):
|
||||
if response.get('Status') == 0:
|
||||
err_details = response.get('ErrorDetails')
|
||||
errors = []
|
||||
for d in err_details:
|
||||
err_code = d.get('ErrorCode')
|
||||
|
||||
if err_code == '2150':
|
||||
irn = [d['Desc']['Irn'] for d in response.get('InfoDtls') if d['InfCd'] == 'DUPIRN']
|
||||
response = get_irn_details(irn[0])
|
||||
return response
|
||||
|
||||
errors.append(d.get('ErrorMessage'))
|
||||
|
||||
if errors:
|
||||
frappe.log_error(title="E Invoice API Request Failed", message=json.dumps(errors, default=str, indent=4))
|
||||
if len(errors) > 1:
|
||||
li = ['<li>'+ d +'</li>' for d in errors]
|
||||
frappe.throw(_("""<ul style='padding-left: 20px'>{}</ul>""").format(''.join(li)), title=_('API Request Failed'))
|
||||
else:
|
||||
frappe.throw(errors[0], title=_('API Request Failed'))
|
||||
|
||||
return response
|
||||
|
||||
def read_json(name):
|
||||
file_path = os.path.join(os.path.dirname(__file__), '{name}.json'.format(name=name))
|
||||
with open(file_path, 'r') as f:
|
||||
@@ -568,50 +345,6 @@ def update_invoice(invoice, res):
|
||||
frappe.db.set_value(doctype, name, 'signed_invoice', dec_signed_invoice)
|
||||
frappe.db.set_value(doctype, name, 'signed_qr_code', res.get('SignedQRCode'))
|
||||
|
||||
@frappe.whitelist()
|
||||
def download_einvoice():
|
||||
data = frappe._dict(frappe.local.form_dict)
|
||||
einvoice = data['einvoice']
|
||||
name = data['name']
|
||||
|
||||
frappe.response['filename'] = 'E-Invoice-' + name + '.json'
|
||||
frappe.response['filecontent'] = einvoice
|
||||
frappe.response['content_type'] = 'application/json'
|
||||
frappe.response['type'] = 'download'
|
||||
|
||||
@frappe.whitelist()
|
||||
def upload_einvoice():
|
||||
signed_einvoice = json.loads(frappe.local.uploaded_file)
|
||||
data = frappe._dict(frappe.local.form_dict)
|
||||
doctype = data['doctype']
|
||||
name = data['docname']
|
||||
|
||||
update_einvoice_fields(doctype, name, signed_einvoice)
|
||||
attach_qrcode_image(doctype, name)
|
||||
|
||||
@frappe.whitelist()
|
||||
def download_cancel_einvoice():
|
||||
data = frappe._dict(frappe.local.form_dict)
|
||||
name = data['name']
|
||||
irn = data['irn']
|
||||
reason = data['reason']
|
||||
remark = data['remark']
|
||||
|
||||
cancel_einvoice = json.dumps([dict(Irn=irn, CnlRsn=reason, CnlRem=remark)])
|
||||
|
||||
frappe.response['filename'] = 'Cancel E-Invoice ' + name + '.json'
|
||||
frappe.response['filecontent'] = cancel_einvoice
|
||||
frappe.response['content_type'] = 'application/json'
|
||||
frappe.response['type'] = 'download'
|
||||
|
||||
@frappe.whitelist()
|
||||
def upload_cancel_ack():
|
||||
cancel_ack = json.loads(frappe.local.uploaded_file)
|
||||
data = frappe._dict(frappe.local.form_dict)
|
||||
doctype = data['doctype']
|
||||
name = data['docname']
|
||||
|
||||
frappe.db.set_value(doctype, name, 'irn_cancelled', 1)
|
||||
|
||||
def attach_qrcode_image(doctype, name):
|
||||
qrcode = frappe.db.get_value(doctype, name, 'signed_qr_code')
|
||||
|
||||
Reference in New Issue
Block a user