from appservices.common.util import * def encrypt_worldline(plain_text, key, iv): print("encryption key used", key) print("encryption iv used", iv) key = key.encode() iv = iv.encode() cipher = AES.new(key, AES.MODE_CBC, iv) padded_text = pad(plain_text.encode(), AES.block_size) encrypted_data = cipher.encrypt(padded_text) return binascii.hexlify(encrypted_data).decode() def decrypt_worldline(decrypt_data, key, iv): try: key = key.encode() iv = iv.encode() encrypted_data = binascii.unhexlify(decrypt_data) cipher = AES.new(key, AES.MODE_CBC, iv) decrypted_data = unpad(cipher.decrypt(encrypted_data), AES.block_size) return decrypted_data.decode('utf-8') except (ValueError, KeyError) as e: return f"Decryption error: {str(e)}" def generate_transaction_identifier(): prefix = "TPSL" random_part = ''.join(random.choices(string.digits, k=10)) return f"{prefix}{random_part}" def worldline_create_payment_link(merchantIdentifier,consumerIdentifier,get_base_url,callback_url,encryption_key,encryption_iv,pgOrderId,amount,cartIdentifier,transactionDeviceIdentifier,transactionType,name,mobileNumber,emailID,payInPaymentGatewayId,client_ip,note,apiType,userId): try: paymentgatewayresponseDict = {} request_body = { "merchant": { "webhookEndpointURL": "", "responseType": "", "responseEndpointURL": callback_url, "identifier": merchantIdentifier, "webhookType": "" }, "cart": { "item": [{ "description": note, "providerIdentifier": "", "surchargeOrDiscountAmount": "", "amount": amount, "reference": "", # Need to check "identifier": cartIdentifier }], "reference": "",# Need to check "identifier": cartIdentifier, "description": note }, "transaction": { "deviceIdentifier": transactionDeviceIdentifier, "smsSending": "N", "amount": amount, "forced3DSCall": "Y", "type": transactionType, "currency": "INR", "isRegistration": "Y", "identifier": pgOrderId, "dateTime": datetime.datetime.now().strftime("%d-%m-%Y"), "requestType": "T" }, "consumer": { "accountHolderName":name, "mobileNumber": mobileNumber, "emailID": emailID, "identifier": consumerIdentifier } } plain_text = json.dumps(request_body) encrypted_request = encrypt_worldline(plain_text, encryption_key, encryption_iv) try: requestData = [request_body] url = f"{get_base_url}/{merchantIdentifier}" print(request_body,"((((((((((((((((((( before encryption body )))))))))))))))))))") print(url,"((((((((((((((((((( url )))))))))))))))))))") print(encrypted_request,"(((((((((((((((((((encrypted_request)))))))))))))))))))") response = requests.post(url, data=encrypted_request) decrypt_data = response.text print(decrypt_data,"(((((((((((((((((((decrypt_data)))))))))))))))))))") decrypted_data = decrypt_worldline(decrypt_data, encryption_key, encryption_iv) print(decrypted_data,"(((((((((((((((((((decrypted_data)))))))))))))))))))") worldLineCreatePaymentLinkResponseData = json.loads(decrypted_data) transactionData = [worldLineCreatePaymentLinkResponseData] transactionState = worldLineCreatePaymentLinkResponseData.get('transactionState') if transactionState == "I": acs_details = worldLineCreatePaymentLinkResponseData.get('paymentMethod', {}).get('aCS', {}) worldline_create_payment_url = acs_details.get("bankAcsUrl","") if worldline_create_payment_url != None: paymentgatewayresponseDict = { "responseStatus": 1, "pgOrderId": pgOrderId, "payment_request_id": pgOrderId, "upi_intent_url": worldline_create_payment_url, "paymentChannel": "Worldline", "transactionData":transactionData, "result": "Success" } return paymentgatewayresponseDict else: paymentgatewayresponseDict["responseStatus"]=0 paymentgatewayresponseDict["responseStatus"]="Payment link not found in the response!!" return paymentgatewayresponseDict else: paymentgatewayresponseDict["responseStatus"]=0 paymentgatewayresponseDict["responseStatus"]="Invalid request data!!" return paymentgatewayresponseDict except requests.RequestException as e: app.logger.error(traceback.format_exc()) paymentgatewayresponseDict["responseStatus"]=0 paymentgatewayresponseDict["responseStatus"]="Our banking partner server is down, please try after some time!!" return paymentgatewayresponseDict except Exception as e: app.logger.error(traceback.format_exc()) paymentgatewayresponseDict["responseStatus"]=0 paymentgatewayresponseDict["responseStatus"]="Our banking partner server is down, please try after some time!!" return paymentgatewayresponseDict def worldline_status_check(transactionIdentifier, dateTime, merchantIdentifier, encryption_key, encryption_iv, get_base_url): paymentgatewayresponseDict = {} try: request_body = { "merchant": { "identifier": merchantIdentifier }, "transaction": { "identifier": transactionIdentifier, "dateTime": dateTime, "requestType": "O" } } status_check_request_body = json.dumps(request_body) encrypted_request = encrypt_worldline(status_check_request_body, encryption_key, encryption_iv) url = get_base_url+"/"+str(merchantIdentifier) try: response = requests.post(url, data=encrypted_request) decrypt_data = response.text decrypted_data = decrypt_worldline(decrypt_data, encryption_key, encryption_iv) worldline_response_data = json.loads(decrypted_data) print(worldline_response_data,"WORLD LINE STATUS CHECK RESPONSE JSON IN CAMPUS") except requests.RequestException as e: app.logger.error(traceback.format_exc()) paymentgatewayresponseDict["responseStatus"]=0 paymentgatewayresponseDict["result"]= "Our banking partner server is down, please try after some time!!" return paymentgatewayresponseDict bankRefNo="" transactionData = [worldline_response_data] status_code = worldline_response_data.get("paymentMethod", {}).get("paymentTransaction", {}).get("statusCode") status_message = worldline_response_data.get("paymentMethod", {}).get("paymentTransaction", {}).get("statusMessage") amount = worldline_response_data.get("paymentMethod", {}).get("paymentTransaction", {}).get("amount") if worldline_response_data: paymentStatus=3 status="pending" if status_code == "0300": paymentStatus=1 status = "success" elif status_code == "9999": status = "failed" paymentStatus=0 else: paymentStatus = 3 status ="pending" paymentgatewayresponseDict = { "responseStatus":1, "paymentStatus":paymentStatus, "status":status, "currency":"", "message":status_message, "bankRefNo":bankRefNo, 'payment_mode':"", 'response_code':"", 'orderAmount':amount, "transactionData":transactionData, "pgOrderId":transactionIdentifier, ####new # "customerVpa": "", # "cardmasked": "" } return paymentgatewayresponseDict else: paymentgatewayresponseDict = { "responseStatus":0, "status":"failed", "message":"failed" } return paymentgatewayresponseDict except Exception as e: app.logger.error(traceback.format_exc()) paymentgatewayresponseDict["responseStatus"]=0 paymentgatewayresponseDict["result"]= "Our banking partner server is down, please try after some time!!" return paymentgatewayresponseDict def getbankCodeByType(cardCategory): data_status={"responseStatus":0,"result":""} print("in getbankCodeByType") try: bankCode="" if cardCategory == "VISA" or cardCategory == "MAST" or cardCategory == "CREDIT CARD" or cardCategory == "HDFC_MAST" or cardCategory == "HDFC_VISA": bankCode = 1060 data_status["responseStatus"]=1 elif cardCategory == "CE_VISA" or cardCategory == "CE_VISA" or cardCategory == "CREDIT CARD": bankCode = 21100 data_status["responseStatus"]=1 elif cardCategory == "CE_RUPAY": bankCode = 21070 data_status["responseStatus"]=1 elif cardCategory == "RUPAY": bankCode = 2040 data_status["responseStatus"]=1 else: data_status["result"] = "Invalid Card" data_status["bankCode"] = bankCode print("in getbankCodeByType if else") return data_status except Exception as e: print("in getbankCodeByType Exception") app.logger.error(traceback.format_exc()) data_status["result"]="Unable To Verify Card" return data_status def worldline_create_seamless_payment_link(merchantIdentifier,get_base_url,callback_url,encryption_key,encryption_iv,pgOrderId,amount,cartIdentifier,card_number,cardCategory,exp_year,exp_month,note,transactionType,requestType,cvv): data_status={"responseStatus":0,"result":""} try: paymentgatewayresponseDict = {} comAmt = "0" ### in sample sheet value exp_year = int(exp_year) + 2000 print("cardCategory",cardCategory) bankCodeResp = getbankCodeByType(cardCategory) print("bankCodeResp",bankCodeResp) if bankCodeResp.get("responseStatus") == 1: bankCodeId = bankCodeResp.get("bankCode") else: data_status["result"] = bankCodeResp.get("result") return data_status request_body = { "merchant": { "webhookEndpointURL": "", "responseType": "", "responseEndpointURL": callback_url, "description": "", "identifier": merchantIdentifier, "webhookType": "" }, "cart": { "item": [ { "description": note, "providerIdentifier": "", "surchargeOrDiscountAmount": "", "amount": amount, "comAmt": comAmt, "sKU": "", "reference": "", "identifier": cartIdentifier } ], "description": "" }, "payment": { "method": { "token": bankCodeId ###1060 bankCode for each card type }, "instrument": { "expiry": { "year": exp_year, "month": exp_month, "dateTime": "" }, "provider": "", "iFSC": "", "holder": { "name": "Rudra" }, "verificationCode": cvv, "token": card_number, # "CardSuffix": "", # "Card_Category": "CRT", #CONDITIONAL ###type CRT/DBT # "Card_Scheme": "VISA", #CONDITIONAL # "CardScope": "DOM", #CONDITIONAL } }, "transaction": { "amount": amount, "type": transactionType, "currency": "INR", "identifier": pgOrderId, "dateTime": datetime.datetime.now().strftime("%d-%m-%Y"), "requestType": requestType } } print("((((((((request_body))))))))",request_body) plain_text = json.dumps(request_body) encrypted_request = encrypt_worldline(plain_text, encryption_key, encryption_iv) try: requestData = [request_body] url = f"{get_base_url}/{merchantIdentifier}" print(request_body,"((((((((((((((((((( before encryption body )))))))))))))))))))") print(url,"((((((((((((((((((( url )))))))))))))))))))") print(encrypted_request,"(((((((((((((((((((encrypted_request)))))))))))))))))))") response = requests.post(url, data=encrypted_request) decrypt_data1 = response.text print(decrypt_data1,"(((((((((((((((((((Encrypted Response )))))))))))))))))))") decrypted_data = decrypt_worldline(decrypt_data1, encryption_key, encryption_iv) print(decrypted_data,"(((((((((((((((((((decrypted_data)))))))))))))))))))") worldLineCreatePaymentLinkResponseData = json.loads(decrypted_data) transactionData = [worldLineCreatePaymentLinkResponseData] transactionState = worldLineCreatePaymentLinkResponseData.get('transactionState') if transactionState == "I": acs_details = worldLineCreatePaymentLinkResponseData.get('paymentMethod', {}).get('aCS', {}) worldline_create_payment_url = acs_details.get("bankAcsUrl","") if worldline_create_payment_url != None: paymentgatewayresponseDict = { "responseStatus": 1, "pgOrderId": pgOrderId, "payment_request_id": pgOrderId, "upi_intent_url": worldline_create_payment_url, "paymentChannel": "Worldline", "transactionData":transactionData, "result": "Success" } return paymentgatewayresponseDict else: paymentgatewayresponseDict["responseStatus"]=0 paymentgatewayresponseDict["responseStatus"]="Payment link not found in the response!!" return paymentgatewayresponseDict else: paymentgatewayresponseDict["responseStatus"]=0 paymentgatewayresponseDict["responseStatus"]="Invalid request data!!" return paymentgatewayresponseDict except requests.RequestException as e: app.logger.error(traceback.format_exc()) paymentgatewayresponseDict["responseStatus"]=0 paymentgatewayresponseDict["responseStatus"]="Our banking partner server is down, please try after some time!!" return paymentgatewayresponseDict except Exception as e: app.logger.error(traceback.format_exc()) paymentgatewayresponseDict["responseStatus"]=0 paymentgatewayresponseDict["responseStatus"]="Our banking partner server is down, please try after some time!!" return paymentgatewayresponseDict