iranian bank gateway added to apps

schedule job for updeing bank status and az iranian bank gateway admin style
This commit is contained in:
Parsa Nazer
2025-03-18 18:18:09 +03:30
parent 062001ef0e
commit 5cbe093400
52 changed files with 3008 additions and 34 deletions
+2
View File
@@ -0,0 +1,2 @@
__version__ = "v2.0.5"
default_app_config = "azbankgateways.apps.AZIranianBankGatewaysConfig"
+72
View File
@@ -0,0 +1,72 @@
from django.contrib import admin
from utils.admin import ModelAdmin
from .models import Bank
class BankAdmin(ModelAdmin):
fields = [
"pk",
"status",
"bank_type",
"tracking_code",
"amount",
"reference_number",
"response_result",
"callback_url",
"extra_information",
"bank_choose_identifier",
"created_at",
"update_at",
'order'
]
list_display = [
"pk",
"status",
"bank_type",
"tracking_code",
"amount",
"reference_number",
"response_result",
"callback_url",
"extra_information",
"bank_choose_identifier",
"created_at",
"update_at",
'order'
]
list_filter = [
"status",
"bank_type",
"created_at",
"update_at",
]
search_fields = [
"status",
"bank_type",
"tracking_code",
"amount",
"reference_number",
"response_result",
"callback_url",
"extra_information",
"created_at",
"update_at",
]
exclude = []
dynamic_raw_id_fields = []
readonly_fields = [
"pk",
"status",
"bank_type",
"tracking_code",
"amount",
"reference_number",
"response_result",
"callback_url",
"extra_information",
"created_at",
"update_at",
]
admin.site.register(Bank, BankAdmin)
+11
View File
@@ -0,0 +1,11 @@
# -*- coding: utf-8 -*-
from django.apps import AppConfig
from django.utils.translation import gettext_lazy as _
class AZIranianBankGatewaysConfig(AppConfig):
name = "azbankgateways"
verbose_name = _("Iranian bank gateway")
verbose_name_plural = _("Iranian bank gateways")
# compatible with django >= 3.2
default_auto_field = "django.db.models.AutoField"
+63
View File
@@ -0,0 +1,63 @@
from __future__ import absolute_import, unicode_literals
import importlib
import logging
from . import default_settings as settings
from .banks import BaseBank
from .exceptions.exceptions import BankGatewayAutoConnectionFailed
from .models import BankType
class BankFactory:
def __init__(self):
logging.debug("Create bank factory")
self._secret_value_reader = self._import(settings.SETTING_VALUE_READER_CLASS)()
@staticmethod
def _import(path):
package, attr = path.rsplit(".", 1)
klass = getattr(importlib.import_module(package), attr)
return klass
def _import_bank(self, bank_type: BankType, identifier: str):
"""
helper to import bank aliases from string paths.
raises an AttributeError if a bank can't be found by it's alias
"""
bank_class = self._import(self._secret_value_reader.klass(bank_type=bank_type, identifier=identifier))
logging.debug("Import bank class")
return bank_class, self._secret_value_reader.read(bank_type=bank_type, identifier=identifier)
def create(self, bank_type: BankType = None, identifier: str = "1") -> BaseBank:
"""Build bank class"""
if not bank_type:
bank_type = self._secret_value_reader.default(identifier)
logging.debug("Request create bank", extra={"bank_type": bank_type})
bank_klass, bank_settings = self._import_bank(bank_type, identifier)
bank = bank_klass(**bank_settings, identifier=identifier)
bank.set_currency(self._secret_value_reader.currency(identifier))
logging.debug("Create bank")
return bank
def auto_create(self, identifier: str = "1", amount=None) -> BaseBank:
logging.debug("Request create bank automatically")
bank_list = self._secret_value_reader.get_bank_priorities(identifier)
errors = []
for bank_type in bank_list:
try:
bank = self.create(bank_type, identifier)
bank.check_gateway(amount)
return bank
except Exception as e:
logging.debug(str(e))
logging.debug("Try to connect another bank...")
errors.append(e)
continue
logging.debug("All banks failed to connect")
errors_msg = "\n".join([str(e) for e in errors])
raise BankGatewayAutoConnectionFailed(errors_msg)
+8
View File
@@ -0,0 +1,8 @@
from .bahamta import Bahamta # noqa
from .banks import BaseBank # noqa
from .bmi import BMI # noqa
from .idpay import IDPay # noqa
from .mellat import Mellat # noqa
from .sep import SEP # noqa
from .zarinpal import Zarinpal # noqa
from .zibal import Zibal # noqa
+134
View File
@@ -0,0 +1,134 @@
import json
import logging
import requests
from azbankgateways.exceptions import BankGatewayConnectionError, SettingDoesNotExist
from azbankgateways.exceptions.exceptions import BankGatewayRejectPayment
from azbankgateways.models import BankType, CurrencyEnum, PaymentStatus
from azbankgateways.utils import append_querystring, get_json, split_to_dict_querystring
from .banks import BaseBank
class Bahamta(BaseBank):
_merchant_code = None
_params = {}
def __init__(self, **kwargs):
super(Bahamta, self).__init__(**kwargs)
self.set_gateway_currency(CurrencyEnum.IRR)
self._token_api_url = "https://webpay.bahamta.com/api/create_request"
self._payment_url = None
self._verify_api_url = "https://webpay.bahamta.com/api/confirm_payment"
def get_bank_type(self):
return BankType.BAHAMTA
def set_default_settings(self):
for item in ["MERCHANT_CODE"]:
if item not in self.default_setting_kwargs:
raise SettingDoesNotExist()
setattr(self, f"_{item.lower()}", self.default_setting_kwargs[item])
"""
Gateway
"""
def _get_gateway_payment_url_parameter(self):
return self._payment_url
def _get_gateway_payment_parameter(self):
params = {}
params.update(self._params)
return params
def _get_gateway_payment_method_parameter(self):
return "GET"
"""
pay
"""
def get_pay_data(self):
data = {
"api_key": self._merchant_code,
"reference": self.get_tracking_code(),
"amount_irr": self.get_gateway_amount(),
"payer_mobile": self.get_mobile_number(),
"callback_url": self._get_gateway_callback_url(),
}
return data
def prepare_pay(self):
super(Bahamta, self).prepare_pay()
def pay(self):
super(Bahamta, self).pay()
data = self.get_pay_data()
response_json = self._send_data(self._token_api_url, data)
if response_json["ok"]:
# در این سیستم رفرنس برای ذخیره سازی بر نمی گردد!
token = self.get_tracking_code()
self._payment_url, self._params = split_to_dict_querystring(response_json["result"]["payment_url"])
self._set_reference_number(token)
else:
logging.critical("Bahamta gateway reject payment")
raise BankGatewayRejectPayment(self.get_transaction_status_text())
"""
verify gateway
"""
def prepare_verify_from_gateway(self):
super(Bahamta, self).prepare_verify_from_gateway()
token = self.get_request().GET.get("reference", None)
self._set_reference_number(token)
self._set_bank_record()
def verify_from_gateway(self, request):
super(Bahamta, self).verify_from_gateway(request)
"""
verify
"""
def get_verify_data(self):
super(Bahamta, self).get_verify_data()
data = {
"api_key": self._merchant_code,
"reference": self.get_reference_number(),
"amount_irr": self.get_gateway_amount(),
}
return data
def prepare_verify(self, tracking_code):
super(Bahamta, self).prepare_verify(tracking_code)
def verify(self, transaction_code):
super(Bahamta, self).verify(transaction_code)
data = self.get_verify_data()
response_json = self._send_data(self._verify_api_url, data)
if response_json.get("ok", False) and response_json.get("result", {}).get("state", None) == "paid":
self._set_payment_status(PaymentStatus.COMPLETE)
extra_information = json.dumps(response_json.get("result", {}))
self._bank.extra_information = extra_information
self._bank.save()
else:
self._set_payment_status(PaymentStatus.CANCEL_BY_USER)
logging.debug("Bahamta gateway unapprove payment")
def _send_data(self, api, data):
try:
url = append_querystring(api, data)
response = requests.get(url, timeout=5)
except requests.Timeout:
logging.exception("Bahamta time out gateway {}".format(data))
raise BankGatewayConnectionError()
except requests.ConnectionError:
logging.exception("Bahamta time out gateway {}".format(data))
raise BankGatewayConnectionError()
response_json = get_json(response)
self._set_transaction_status_text(response_json.get("error"))
return response_json
+370
View File
@@ -0,0 +1,370 @@
import abc
import logging
import uuid
from urllib import parse
import six
from django.db.models import Q
from django.shortcuts import redirect
from django.urls import reverse
from django.utils import timezone
from .. import default_settings as settings
from ..exceptions import (
AmountDoesNotSupport,
BankGatewayStateInvalid,
BankGatewayTokenExpired,
CurrencyDoesNotSupport,
SafeSettingsEnabled,
)
from ..models import Bank, CurrencyEnum, PaymentStatus
from ..utils import append_querystring
# TODO: handle and expire record after 15 minutes
@six.add_metaclass(abc.ABCMeta)
class BaseBank:
"""Base bank for sending to gateway."""
_gateway_currency: str = CurrencyEnum.IRR
_currency: str = CurrencyEnum.IRR
_amount: int = 0
_gateway_amount: int = 0
_mobile_number: str = None
_tracking_code: int = None
_reference_number: str = ""
_transaction_status_text: str = ""
_client_callback_url: str = ""
_bank: Bank = None
_request = None
def __init__(self, identifier: str, **kwargs):
self.identifier = identifier
self.default_setting_kwargs = kwargs
self.set_default_settings()
@abc.abstractmethod
def set_default_settings(self):
"""default setting, like fetch merchant code, terminal id and etc"""
pass
def prepare_amount(self):
"""prepare amount"""
if self._currency == self._gateway_currency:
self._gateway_amount = self._amount
elif self._currency == CurrencyEnum.IRR and self._gateway_currency == CurrencyEnum.IRT:
self._gateway_amount = CurrencyEnum.rial_to_toman(self._amount)
elif self._currency == CurrencyEnum.IRT and self._gateway_currency == CurrencyEnum.IRR:
self._gateway_amount = CurrencyEnum.toman_to_rial(self._amount)
else:
self._gateway_amount = self._amount
if not self.check_amount():
raise AmountDoesNotSupport()
def check_amount(self):
return self.get_gateway_amount() >= self.get_minimum_amount()
@classmethod
def get_minimum_amount(cls):
return 1000
@abc.abstractmethod
def get_bank_type(self):
pass
def get_amount(self):
"""get the amount"""
return self._amount
def set_amount(self, amount):
"""set amount"""
if int(amount) <= 0:
raise AmountDoesNotSupport()
self._amount = int(amount)
@abc.abstractmethod
def prepare_pay(self):
logging.debug("Prepare pay method")
self.prepare_amount()
tracking_code = int(str(uuid.uuid4().int)[-1 * settings.TRACKING_CODE_LENGTH :])
self._set_tracking_code(tracking_code)
@abc.abstractmethod
def get_pay_data(self):
pass
@abc.abstractmethod
def pay(self):
logging.debug("Pay method")
self.prepare_pay()
@abc.abstractmethod
def get_verify_data(self):
pass
@abc.abstractmethod
def prepare_verify(self, tracking_code):
logging.debug("Prepare verify method")
self._set_tracking_code(tracking_code)
self._set_bank_record()
self.prepare_amount()
@abc.abstractmethod
def verify(self, tracking_code):
logging.debug("Verify method")
self.prepare_verify(tracking_code)
def ready(self) -> Bank:
self.pay()
bank = Bank.objects.create(
bank_choose_identifier=self.identifier,
bank_type=self.get_bank_type(),
amount=self.get_amount(),
reference_number=self.get_reference_number(),
response_result=self.get_transaction_status_text(),
tracking_code=self.get_tracking_code(),
)
self._bank = bank
self._set_payment_status(PaymentStatus.WAITING)
if self._client_callback_url:
self._bank.callback_url = self._client_callback_url
return bank
@abc.abstractmethod
def prepare_verify_from_gateway(self):
pass
def verify_from_gateway(self, request):
"""زمانی که کاربر از گیت وی بانک باز میگردد این متد فراخوانی می شود."""
self.set_request(request)
self.prepare_verify_from_gateway()
self._set_payment_status(PaymentStatus.RETURN_FROM_BANK)
self.verify(self.get_tracking_code())
def get_client_callback_url(self):
"""این متد پس از وریفای شدن استفاده خواهد شد. لینک برگشت را بر میگرداند.حال چه وریفای موفقیت آمیز باشد چه با
لغو کاربر مواجه شده باشد"""
return append_querystring(
self._bank.callback_url,
{settings.TRACKING_CODE_QUERY_PARAM: self.get_tracking_code()},
)
def redirect_client_callback(self):
""" "این متد کاربر را به مسیری که نرم افزار میخواهد هدایت خواهد کرد و پس از وریفای شدن استفاده می شود."""
logging.debug("Redirect to client")
return redirect(self.get_client_callback_url())
def set_mobile_number(self, mobile_number):
"""شماره موبایل کاربر را جهت ارسال به درگاه برای فتچ کردن شماره کارت ها و ... ارسال خواهد کرد."""
self._mobile_number = mobile_number
def get_mobile_number(self):
return self._mobile_number
def set_client_callback_url(self, callback_url):
"""ذخیره کال بک از طریق نرم افزار برای بازگردانی کاربر پس از بازگشت درگاه بانک به پکیج و سپس از پکیج به نرم
افزار."""
if not self._bank:
self._client_callback_url = callback_url
else:
logging.critical(
"You are change the call back url in invalid situation.",
extra={
"bank_id": self._bank.pk,
"status": self._bank.status,
},
)
raise BankGatewayStateInvalid(
"Bank state not equal to waiting. Probably finish "
f"or redirect to bank gateway. status is {self._bank.status}"
)
def _set_reference_number(self, reference_number):
"""reference number get from bank"""
self._reference_number = reference_number
def _set_bank_record(self):
try:
self._bank = Bank.objects.get(
Q(Q(reference_number=self.get_reference_number()) | Q(tracking_code=self.get_tracking_code())),
Q(bank_type=self.get_bank_type()),
)
logging.debug("Set reference find bank object.")
except Bank.DoesNotExist:
logging.debug("Cant find bank record object.")
raise BankGatewayStateInvalid(
"Cant find bank record with reference number reference number is {}".format(
self.get_reference_number()
)
)
self._set_tracking_code(self._bank.tracking_code)
self._set_reference_number(self._bank.reference_number)
self.set_amount(self._bank.amount)
def get_reference_number(self):
return self._reference_number
"""
ترنزکشن تکست متنی است که از طرف درگاه بانک به عنوان پیام باز میگردد.
"""
def _set_transaction_status_text(self, txt):
self._transaction_status_text = txt
def get_transaction_status_text(self):
return self._transaction_status_text
def _set_payment_status(self, payment_status):
if payment_status == PaymentStatus.RETURN_FROM_BANK and self._bank.status != PaymentStatus.REDIRECT_TO_BANK:
logging.debug(
"Payment status is not status suitable.",
extra={"status": self._bank.status},
)
raise BankGatewayStateInvalid(
"You change the status bank record before/after this record change status from redirect to bank. "
"current status is {}".format(self._bank.status)
)
self._bank.status = payment_status
self._bank.save()
logging.debug("Change bank payment status", extra={"status": payment_status})
def set_gateway_currency(self, currency: CurrencyEnum):
"""واحد پولی درگاه بانک"""
if currency not in [CurrencyEnum.IRR, CurrencyEnum.IRT]:
raise CurrencyDoesNotSupport()
self._gateway_currency = currency
def get_gateway_currency(self):
return self._gateway_currency
def set_currency(self, currency: CurrencyEnum):
""" "واحد پولی نرم افزار"""
if currency not in [CurrencyEnum.IRR, CurrencyEnum.IRT]:
raise CurrencyDoesNotSupport()
self._currency = currency
def get_currency(self):
return self._currency
def get_gateway_amount(self):
return self._gateway_amount
"""
ترکینگ کد توسط برنامه تولید شده و برای استفاده های بعدی کاربرد خواهد داشت.
"""
def _set_tracking_code(self, tracking_code):
self._tracking_code = tracking_code
def get_tracking_code(self):
return self._tracking_code
"""ًRequest"""
def set_request(self, request):
self._request = request
def get_request(self):
return self._request
"""gateway"""
def _prepare_check_gateway(self, amount=None):
"""ست کردن داده های اولیه"""
if amount:
self.set_amount(amount)
else:
self.set_amount(10000)
self.set_client_callback_url("/")
def check_gateway(self, amount=None):
"""با این متد از صحت و سلامت گیت وی برای اتصال اطمینان حاصل می کنیم."""
self._prepare_check_gateway(amount)
self.pay()
@abc.abstractmethod
def _get_gateway_payment_url_parameter(self):
"""این متد بسته به بانک متفاوت پر می شود."""
"""
:return
url: str
"""
pass
@abc.abstractmethod
def _get_gateway_payment_parameter(self):
"""این متد بسته به بانک متفاوت پر می شود."""
"""
:return
params: dict
"""
pass
@abc.abstractmethod
def _get_gateway_payment_method_parameter(self):
"""این متد بسته به بانک متفاوت پر می شود."""
"""
:return
method: POST, GET
"""
pass
def _verify_payment_expiry(self):
"""برسی میکند درگاه ساخته شده اعتبار دارد یا خیر"""
if (timezone.now() - self._bank.created_at).seconds > 120:
self._set_payment_status(PaymentStatus.EXPIRE_GATEWAY_TOKEN)
logging.debug("Redirect to bank expire!")
raise BankGatewayTokenExpired()
def redirect_gateway(self):
"""کاربر را به درگاه بانک هدایت می کند"""
self._verify_payment_expiry()
if settings.IS_SAFE_GET_GATEWAY_PAYMENT:
raise SafeSettingsEnabled()
logging.debug("Redirect to bank")
self._set_payment_status(PaymentStatus.REDIRECT_TO_BANK)
return redirect(self.get_gateway_payment_url())
def get_gateway(self):
"""اطلاعات درگاه پرداخت را برمیگرداند"""
self._verify_payment_expiry()
logging.debug("Redirect to bank")
self._set_payment_status(PaymentStatus.REDIRECT_TO_BANK)
return self.safe_get_gateway_payment_url()
def safe_get_gateway_payment_url(self):
url = self._get_gateway_payment_url_parameter()
params = self._get_gateway_payment_parameter()
method = self._get_gateway_payment_method_parameter()
context = {"params": params, "url": url, "method": method}
return context
def get_gateway_payment_url(self):
redirect_url = reverse(settings.GO_TO_BANK_GATEWAY_NAMESPACE)
url = self._get_gateway_payment_url_parameter()
params = self._get_gateway_payment_parameter()
method = self._get_gateway_payment_method_parameter()
params.update(
{
"url": url,
"method": method,
}
)
redirect_url = append_querystring(redirect_url, params)
if self.get_request():
redirect_url = self.get_request().build_absolute_uri(redirect_url)
return redirect_url
def _get_gateway_callback_url(self):
url = reverse(settings.CALLBACK_NAMESPACE)
if self.get_request():
url_parts = list(parse.urlparse(url))
if not (url_parts[0] and url_parts[1]):
url = self.get_request().build_absolute_uri(url)
query = dict(parse.parse_qsl(self.get_request().GET.urlencode()))
query.update({"bank_type": self.get_bank_type()})
query.update({"identifier": self.identifier})
url = append_querystring(url, query)
return url
+155
View File
@@ -0,0 +1,155 @@
import base64
import datetime
import logging
import requests
from Crypto.Cipher import DES3
from azbankgateways.banks import BaseBank
from azbankgateways.exceptions import BankGatewayConnectionError, SettingDoesNotExist
from azbankgateways.exceptions.exceptions import (
BankGatewayRejectPayment,
BankGatewayStateInvalid,
)
from azbankgateways.models import BankType, CurrencyEnum, PaymentStatus
from azbankgateways.utils import get_json
class BMI(BaseBank):
_merchant_code = None
_terminal_code = None
_secret_key = None
def __init__(self, **kwargs):
super(BMI, self).__init__(**kwargs)
self.set_gateway_currency(CurrencyEnum.IRR)
self._token_api_url = "https://sadad.shaparak.ir/vpg/api/v0/Request/PaymentRequest"
self._payment_url = "https://sadad.shaparak.ir/VPG/Purchase"
self._verify_api_url = "https://sadad.shaparak.ir/vpg/api/v0/Advice/Verify"
def get_bank_type(self):
return BankType.BMI
def set_default_settings(self):
for item in ["MERCHANT_CODE", "TERMINAL_CODE", "SECRET_KEY"]:
if item not in self.default_setting_kwargs:
raise SettingDoesNotExist()
setattr(self, f"_{item.lower()}", self.default_setting_kwargs[item])
def get_pay_data(self):
time_now = datetime.datetime.now().strftime("%m/%d/%Y %H:%M:%S %p")
data = {
"TerminalId": self._terminal_code,
"MerchantId": self._merchant_code,
"Amount": self.get_gateway_amount(),
"SignData": self._encrypt_des3(
"{};{};{}".format(
self._terminal_code,
self.get_tracking_code(),
self.get_gateway_amount(),
)
),
"ReturnUrl": self._get_gateway_callback_url(),
"LocalDateTime": time_now,
"OrderId": self.get_tracking_code(),
"AdditionalData": "oi:%s-ou:%s" % (self.get_tracking_code(), self.get_mobile_number()),
}
return data
def prepare_pay(self):
super(BMI, self).prepare_pay()
def pay(self):
super(BMI, self).pay()
data = self.get_pay_data()
response_json = self._send_data(self._token_api_url, data)
if response_json["ResCode"] == "0":
token = response_json["Token"]
self._set_reference_number(token)
else:
logging.critical("BMI gateway reject payment")
raise BankGatewayRejectPayment(self.get_transaction_status_text())
"""
: gateway
"""
def _get_gateway_payment_method_parameter(self):
return "GET"
def _get_gateway_payment_url_parameter(self):
return self._payment_url
def _get_gateway_payment_parameter(self):
params = {"Token": self.get_reference_number()}
return params
def get_verify_data(self):
super(BMI, self).get_verify_data()
data = {
"Token": self.get_reference_number(),
"SignData": self._encrypt_des3(self.get_reference_number()),
}
return data
def prepare_verify(self, tracking_code):
super(BMI, self).prepare_verify(tracking_code)
def verify(self, transaction_code):
super(BMI, self).verify(transaction_code)
data = self.get_verify_data()
response_json = self._send_data(self._verify_api_url, data)
if response_json["ResCode"] == "0":
self._set_payment_status(PaymentStatus.COMPLETE)
extra_information = (
f"RetrivalRefNo={response_json['RetrivalRefNo']},SystemTraceNo={response_json['SystemTraceNo']}"
)
self._bank.extra_information = extra_information
self._bank.save()
else:
self._set_payment_status(PaymentStatus.CANCEL_BY_USER)
logging.debug("BMI gateway unapprove payment")
def prepare_verify_from_gateway(self):
super(BMI, self).prepare_verify_from_gateway()
request = self.get_request()
for method in ["POST", "GET", "data", "PUT"]:
token = getattr(request, method, {}).get("token", None)
if token:
break
if not token:
raise BankGatewayStateInvalid
self._set_reference_number(token)
self._set_bank_record()
def verify_from_gateway(self, request):
super(BMI, self).verify_from_gateway(request)
@classmethod
def _pad(cls, text, pad_size=16):
text_length = len(text)
last_block_size = text_length % pad_size
remaining_space = pad_size - last_block_size
text = text + (remaining_space * chr(remaining_space))
return text
def _encrypt_des3(self, text):
secret_key_bytes = base64.b64decode(self._secret_key)
text = self._pad(text, 8)
cipher = DES3.new(secret_key_bytes, DES3.MODE_ECB)
cipher_text = cipher.encrypt(str.encode(text))
return base64.b64encode(cipher_text).decode("utf-8")
def _send_data(self, api, data):
try:
response = requests.post(api, json=data, timeout=5)
except requests.Timeout:
logging.exception("BMI time out gateway {}".format(data))
raise BankGatewayConnectionError()
except requests.ConnectionError:
logging.exception("BMI time out gateway {}".format(data))
raise BankGatewayConnectionError()
response_json = get_json(response)
self._set_transaction_status_text(response_json["Description"])
return response_json
+141
View File
@@ -0,0 +1,141 @@
import json
import logging
import requests
from azbankgateways.banks import BaseBank
from azbankgateways.exceptions import BankGatewayConnectionError, SettingDoesNotExist
from azbankgateways.exceptions.exceptions import BankGatewayRejectPayment
from azbankgateways.models import BankType, CurrencyEnum, PaymentStatus
from azbankgateways.utils import get_json, split_to_dict_querystring
class IDPay(BaseBank):
_merchant_code = None
_method = None
_x_sandbox = None
_payment_url = None
_params = {}
def __init__(self, **kwargs):
super(IDPay, self).__init__(**kwargs)
self.set_gateway_currency(CurrencyEnum.IRR)
self._token_api_url = "https://api.idpay.ir/v1.1/payment"
self._verify_api_url = "https://api.idpay.ir/v1.1/payment/verify"
def get_bank_type(self):
return BankType.IDPAY
def set_default_settings(self):
for item in ["MERCHANT_CODE", "METHOD", "X_SANDBOX"]:
if item not in self.default_setting_kwargs:
raise SettingDoesNotExist()
setattr(self, f"_{item.lower()}", self.default_setting_kwargs[item])
self._x_sandbox = str(self._x_sandbox)
"""
gateway
"""
def _get_gateway_payment_url_parameter(self):
return self._payment_url
def _get_gateway_payment_parameter(self):
params = {}
params.update(self._params)
return params
def _get_gateway_payment_method_parameter(self):
return "GET"
"""
pay
"""
def get_pay_data(self):
data = {
"order_id": self.get_tracking_code(),
"amount": self.get_gateway_amount(),
"phone": self.get_mobile_number(),
"callback": self._get_gateway_callback_url(),
}
return data
def prepare_pay(self):
super(IDPay, self).prepare_pay()
def pay(self):
super(IDPay, self).pay()
data = self.get_pay_data()
response_json = self._send_data(self._token_api_url, data)
if "id" in response_json and "link" in response_json and response_json["link"] and response_json["id"]:
token = response_json["id"]
self._payment_url, self._params = split_to_dict_querystring(response_json["link"])
self._set_reference_number(token)
else:
logging.critical("IDPay gateway reject payment")
raise BankGatewayRejectPayment(self.get_transaction_status_text())
"""
verify gateway
"""
def prepare_verify_from_gateway(self):
super(IDPay, self).prepare_verify_from_gateway()
for method in ["GET", "POST", "data"]:
token = getattr(self.get_request(), method).get("id", None)
if token:
self._set_reference_number(token)
self._set_bank_record()
break
def verify_from_gateway(self, request):
super(IDPay, self).verify_from_gateway(request)
"""
verify
"""
def get_verify_data(self):
super(IDPay, self).get_verify_data()
data = {
"id": self.get_reference_number(),
"order_id": self.get_tracking_code(),
}
return data
def prepare_verify(self, tracking_code):
super(IDPay, self).prepare_verify(tracking_code)
def verify(self, transaction_code):
super(IDPay, self).verify(transaction_code)
data = self.get_verify_data()
response_json = self._send_data(self._verify_api_url, data, timeout=10)
if response_json.get("verify", {}).get("date", None):
self._set_payment_status(PaymentStatus.COMPLETE)
extra_information = json.dumps(response_json)
self._bank.extra_information = extra_information
self._bank.save()
else:
self._set_payment_status(PaymentStatus.CANCEL_BY_USER)
logging.debug("IDPay gateway unapprove payment")
def _send_data(self, api, data, timeout=5):
headers = {
"X-API-KEY": self._merchant_code,
"X-SANDBOX": self._x_sandbox,
}
try:
response = requests.post(api, headers=headers, json=data, timeout=timeout)
except requests.Timeout:
logging.exception("IDPay time out gateway {}".format(data))
raise BankGatewayConnectionError()
except requests.ConnectionError:
logging.exception("IDPay time out gateway {}".format(data))
raise BankGatewayConnectionError()
response_json = get_json(response)
if "error_message" in response_json:
self._set_transaction_status_text(response_json["error_message"])
return response_json
+267
View File
@@ -0,0 +1,267 @@
import logging
from json import dumps, loads
from time import gmtime, strftime
from zeep import Client, Transport
from azbankgateways.banks import BaseBank
from azbankgateways.exceptions import SettingDoesNotExist
from azbankgateways.exceptions.exceptions import BankGatewayRejectPayment
from azbankgateways.models import BankType, CurrencyEnum, PaymentStatus
class Mellat(BaseBank):
_terminal_code = None
_username = None
_password = None
def __init__(self, **kwargs):
super(Mellat, self).__init__(**kwargs)
self.set_gateway_currency(CurrencyEnum.IRR)
self._payment_url = "https://bpm.shaparak.ir/pgwchannel/startpay.mellat"
def get_bank_type(self):
return BankType.MELLAT
def set_default_settings(self):
for item in ["TERMINAL_CODE", "USERNAME", "PASSWORD"]:
if item not in self.default_setting_kwargs:
raise SettingDoesNotExist()
setattr(self, f"_{item.lower()}", self.default_setting_kwargs[item])
"""
gateway
"""
@classmethod
def get_minimum_amount(cls):
return 1000
def _get_gateway_payment_url_parameter(self):
return self._payment_url
def _get_gateway_payment_parameter(self):
params = {
"RefId": self.get_reference_number(),
"MobileNo": self.get_mobile_number(),
}
return params
def _get_gateway_payment_method_parameter(self):
return "GET"
"""
pay
"""
def get_pay_data(self):
description = "خرید با شماره پیگیری - {}".format(self.get_tracking_code())
data = {
"terminalId": int(self._terminal_code),
"userName": self._username,
"userPassword": self._password,
"orderId": int(self.get_tracking_code()),
"amount": int(self.get_gateway_amount()),
"localDate": self._get_current_date(),
"localTime": self._get_current_time(),
"additionalData": description,
"callBackUrl": self._get_gateway_callback_url(),
"payerId": 0,
}
return data
def prepare_pay(self):
super(Mellat, self).prepare_pay()
def pay(self):
super(Mellat, self).pay()
data = self.get_pay_data()
client = self._get_client()
response = client.service.bpPayRequest(**data)
try:
status, token = response.split(",")
if status == "0":
self._set_reference_number(token)
except ValueError:
status_text = "Unknown error"
if response == "11":
status_text = "Card number is invalid"
elif response == "12":
status_text = "Insufficient inventory"
elif response == "13":
status_text = "Password is incorrect"
elif response == "14":
status_text = "Max try reached"
elif response == "15":
status_text = "Card is invalid"
elif response == "16":
status_text = "The number of withdrawals is more than allowed"
elif response == "17":
status_text = "The user has abandoned the transaction"
elif response == "18":
status_text = "The card has expired"
elif response == "19":
status_text = "The withdrawal amount is over the limit"
elif response == "21":
status_text = "Invalid service"
elif response == "23":
status_text = "A security error has occurred"
elif response == "24":
status_text = "The recipient's user information is invalid"
elif response == "25":
status_text = "The amount is invalid"
elif response == "31":
status_text = "The response is invalid"
elif response == "32":
status_text = "The format of the entered information is not correct"
elif response == "33":
status_text = "The account is invalid"
elif response == "34":
status_text = "System error"
elif response == "35":
status_text = "Date is invalid"
elif response == "41":
status_text = "The request number is duplicate"
elif response == "42":
status_text = "Sale transaction not found"
elif response == "43":
status_text = "Verify has already been requested"
elif response == "44":
status_text = "Verify request not found"
elif response == "45":
status_text = "The transaction has been settled"
elif response == "46":
status_text = "The transaction has not been settled"
elif response == "47":
status_text = "Settle transaction not found"
elif response == "48":
status_text = "The transaction has been reversed"
elif response == "49":
status_text = "Refund transaction not found"
elif response == "51":
status_text = "The transaction is repeated"
elif response == "54":
status_text = "The reference transaction does not exist"
elif response == "55":
status_text = "The transaction is invalid"
elif response == "61":
status_text = "Error in deposit"
elif response == "111":
status_text = "Card issuer is invalid"
elif response == "112":
status_text = "Card issuing switch error"
elif response == "113":
status_text = "No response was received from the card issuer"
elif response == "114":
status_text = "The cardholder is not authorized to perform this transaction"
elif response == "113":
status_text = "No response was received from the card issuer"
elif response == "412":
status_text = "The invoice ID is incorrect"
elif response == "413":
status_text = "Payment ID is incorrect"
elif response == "414":
status_text = "The organization issuing the bill is invalid"
elif response == "415":
status_text = "The working session has ended"
elif response == "416":
status_text = "The working session has ended"
elif response == "417":
status_text = "Payer ID is invalid"
elif response == "418":
status_text = "Problems in defining customer information"
elif response == "419":
status_text = "The number of data entries has exceeded the limit"
elif response == "421":
status_text = "Invalid IP address"
self._set_transaction_status_text(status_text)
logging.critical(status_text)
raise BankGatewayRejectPayment(self.get_transaction_status_text())
"""
verify from gateway
"""
def prepare_verify_from_gateway(self):
super(Mellat, self).prepare_verify_from_gateway()
post = self.get_request().POST
token = post.get("RefId", None)
if not token:
return
self._set_reference_number(token)
self._set_bank_record()
self._bank.extra_information = dumps(dict(zip(post.keys(), post.values())))
self._bank.save()
def verify_from_gateway(self, request):
super(Mellat, self).verify_from_gateway(request)
"""
verify
"""
def get_verify_data(self):
super(Mellat, self).get_verify_data()
data = {
"terminalId": self._terminal_code,
"userName": self._username,
"userPassword": self._password,
"orderId": self.get_tracking_code(),
"saleOrderId": self.get_tracking_code(),
"saleReferenceId": self._get_sale_reference_id(),
}
return data
def prepare_verify(self, tracking_code):
super(Mellat, self).prepare_verify(tracking_code)
def verify(self, transaction_code):
super(Mellat, self).verify(transaction_code)
data = self.get_verify_data()
client = self._get_client()
verify_result = client.service.bpVerifyRequest(**data)
if verify_result == "0":
self._settle_transaction()
else:
verify_result = client.service.bpInquiryRequest(**data)
if verify_result == "0":
self._settle_transaction()
else:
logging.debug("Not able to verify the transaction, Making reversal request")
reversal_result = client.service.bpReversalRequest(**data)
if reversal_result != "0":
logging.debug("Reversal request was not successfull")
self._set_payment_status(PaymentStatus.CANCEL_BY_USER)
logging.debug("Mellat gateway unapproved the payment")
def _settle_transaction(self):
data = self.get_verify_data()
client = self._get_client()
settle_result = client.service.bpSettleRequest(**data)
if settle_result == "0":
self._set_payment_status(PaymentStatus.COMPLETE)
else:
logging.debug("Mellat gateway did not settle the payment")
@staticmethod
def _get_client():
transport = Transport(timeout=5, operation_timeout=5)
client = Client("https://bpm.shaparak.ir/pgwchannel/services/pgw?wsdl", transport=transport)
return client
@staticmethod
def _get_current_time():
return strftime("%H%M%S")
@staticmethod
def _get_current_date():
return strftime("%Y%m%d", gmtime())
def _get_sale_reference_id(self):
extra_information = loads(getattr(self._bank, "extra_information", "{}"))
return extra_information.get("SaleReferenceId", "1")
+152
View File
@@ -0,0 +1,152 @@
import json
import logging
import requests
from azbankgateways.banks import BaseBank
from azbankgateways.default_settings import TRACKING_CODE_QUERY_PARAM
from azbankgateways.exceptions import BankGatewayConnectionError, SettingDoesNotExist
from azbankgateways.exceptions.exceptions import (
BankGatewayRejectPayment,
BankGatewayStateInvalid,
)
from azbankgateways.models import BankType, CurrencyEnum, PaymentStatus
class PayV1(BaseBank):
_merchant_code = None
_x_sandbox = None
def __init__(self, **kwargs):
super(PayV1, self).__init__(**kwargs)
self.set_gateway_currency(CurrencyEnum.IRR)
self._token_api_url = "https://pay.ir/pg/send"
self._payment_url = "https://pay.ir/pg/{}"
self._verify_api_url = "https://pay.ir/pg/verify"
def get_bank_type(self):
return BankType.PAYV1
def set_default_settings(self):
for item in ["MERCHANT_CODE", "X_SANDBOX"]:
if item not in self.default_setting_kwargs:
raise SettingDoesNotExist()
setattr(self, f"_{item.lower()}", self.default_setting_kwargs[item])
self._merchant_code = self._merchant_code if not self._x_sandbox else "test"
"""
gateway
"""
def _get_gateway_payment_url_parameter(self):
return self._payment_url.format(self._reference_number)
def _get_gateway_payment_parameter(self):
return {}
def _get_gateway_payment_method_parameter(self):
return "GET"
"""
pay
"""
def get_pay_data(self):
data = {
"api": self._merchant_code,
"amount": self.get_gateway_amount(),
"redirect": self._get_gateway_callback_url(),
"mobile": self.get_mobile_number(),
"factorNumber": self.get_tracking_code(),
}
return data
def prepare_pay(self):
super(PayV1, self).prepare_pay()
def pay(self):
super(PayV1, self).pay()
data = self.get_pay_data()
response = self._send_data(self._token_api_url, data)
response_json = response.json()
if response.status_code == 200 and int(response_json["status"]) == 1:
token = response_json["token"]
self._set_reference_number(token)
else:
logging.critical(
"PayV1 gateway reject payment with error code {0} and status code {1}".format(
response_json["errorCode"], response.status_code
)
)
raise BankGatewayRejectPayment(self.get_transaction_status_text())
"""
verify gateway
"""
def prepare_verify_from_gateway(self):
super(PayV1, self).prepare_verify_from_gateway()
for method in ["GET", "POST", "data"]:
token = getattr(self.get_request(), method).get(TRACKING_CODE_QUERY_PARAM, None)
if token:
self._set_reference_number(token)
self._set_bank_record()
break
else:
raise BankGatewayStateInvalid
def verify_from_gateway(self, request):
super(PayV1, self).verify_from_gateway(request)
"""
verify
"""
def get_verify_data(self):
super(PayV1, self).get_verify_data()
data = {
"api": self._merchant_code(),
"token": self.get_reference_number(),
}
return data
def prepare_verify(self, tracking_code):
super(PayV1, self).prepare_verify(tracking_code)
def verify(self, tracking_code):
super(PayV1, self).verify(tracking_code)
data = self.get_verify_data()
response = self._send_data(self._verify_api_url, data, timeout=10)
response_json = response.json()
status = PaymentStatus.COMPLETE
if int(response_json["status"]) != 1:
if int(response_json["errorCode"]) == -5:
status = PaymentStatus.ERROR
elif int(response_json["errorCode"]) == -9:
status = PaymentStatus.EXPIRE_VERIFY_PAYMENT
elif int(response_json["errorCode"]) == -15:
status = PaymentStatus.CANCEL_BY_USER
elif int(response_json["errorCode"]) == -27:
status = PaymentStatus.RETURN_FROM_BANK
else:
status = PaymentStatus.ERROR
self._set_payment_status(status)
extra_information = json.dumps(response_json)
self._bank.extra_information = extra_information
self._bank.save()
def _send_data(self, url, data, timeout=5) -> requests.post:
try:
logging.debug("Sending POST request to {} with data {}".format(url, data))
response = requests.post(url, json=data, timeout=timeout)
except requests.Timeout:
logging.exception("PayV1 time out gateway {}".format(data))
raise BankGatewayConnectionError()
except requests.ConnectionError:
logging.exception("PayV1 time out gateway {}".format(data))
raise BankGatewayConnectionError()
return response
+146
View File
@@ -0,0 +1,146 @@
import logging
import requests
from zeep import Client, Transport
from azbankgateways.banks import BaseBank
from azbankgateways.exceptions import BankGatewayConnectionError, SettingDoesNotExist
from azbankgateways.exceptions.exceptions import BankGatewayRejectPayment
from azbankgateways.models import BankType, CurrencyEnum, PaymentStatus
from azbankgateways.utils import get_json
class SEP(BaseBank):
_merchant_code = None
_terminal_code = None
def __init__(self, **kwargs):
super(SEP, self).__init__(**kwargs)
self.set_gateway_currency(CurrencyEnum.IRR)
self._token_api_url = "https://sep.shaparak.ir/MobilePG/MobilePayment"
self._payment_url = "https://sep.shaparak.ir/OnlinePG/OnlinePG"
self._verify_api_url = "https://verify.sep.ir/Payments/ReferencePayment.asmx?WSDL"
def get_bank_type(self):
return BankType.SEP
def set_default_settings(self):
for item in ["MERCHANT_CODE", "TERMINAL_CODE"]:
if item not in self.default_setting_kwargs:
raise SettingDoesNotExist()
setattr(self, f"_{item.lower()}", self.default_setting_kwargs[item])
def get_pay_data(self):
data = {
"Action": "Token",
"Amount": self.get_gateway_amount(),
"Wage": 0,
"TerminalId": self._merchant_code,
"ResNum": self.get_tracking_code(),
"RedirectURL": self._get_gateway_callback_url(),
"CellNumber": self.get_mobile_number(),
}
return data
def prepare_pay(self):
super(SEP, self).prepare_pay()
def pay(self):
super(SEP, self).pay()
data = self.get_pay_data()
response_json = self._send_data(self._token_api_url, data)
if str(response_json["status"]) == "1":
token = response_json["token"]
self._set_reference_number(token)
else:
logging.critical("SEP gateway reject payment")
raise BankGatewayRejectPayment(self.get_transaction_status_text())
"""
: gateway
"""
def _get_gateway_payment_url_parameter(self):
return self._payment_url
def _get_gateway_payment_method_parameter(self):
return "POST"
def _get_gateway_payment_parameter(self):
params = {
"Token": self.get_reference_number(),
"GetMethod": "true",
}
return params
"""
verify from gateway
"""
def prepare_verify_from_gateway(self):
super(SEP, self).prepare_verify_from_gateway()
request = self.get_request()
tracking_code = request.GET.get("ResNum", None)
token = request.GET.get("Token", None)
self._set_tracking_code(tracking_code)
self._set_bank_record()
ref_num = request.GET.get("RefNum", None)
if request.GET.get("State", "NOK") == "OK" and ref_num:
self._set_reference_number(ref_num)
self._bank.reference_number = ref_num
extra_information = f"TRACENO={request.GET.get('TRACENO', None)}, RefNum={ref_num}, Token={token}"
self._bank.extra_information = extra_information
self._bank.save()
def verify_from_gateway(self, request):
super(SEP, self).verify_from_gateway(request)
"""
verify
"""
def get_verify_data(self):
super(SEP, self).get_verify_data()
data = self.get_reference_number(), self._merchant_code
return data
def prepare_verify(self, tracking_code):
super(SEP, self).prepare_verify(tracking_code)
def verify(self, transaction_code):
super(SEP, self).verify(transaction_code)
data = self.get_verify_data()
client = self._get_client(self._verify_api_url)
result = client.service.verifyTransaction(*data)
if result == self.get_gateway_amount():
self._set_payment_status(PaymentStatus.COMPLETE)
else:
self._set_payment_status(PaymentStatus.CANCEL_BY_USER)
logging.debug("SEP gateway unapprove payment")
def _send_data(self, api, data):
try:
response = requests.post(api, json=data, timeout=5)
except requests.Timeout:
logging.exception("SEP time out gateway {}".format(data))
raise BankGatewayConnectionError()
except requests.ConnectionError:
logging.exception("SEP time out gateway {}".format(data))
raise BankGatewayConnectionError()
response_json = get_json(response)
self._set_transaction_status_text(response_json.get("errorDesc"))
return response_json
@staticmethod
def _get_client(url):
headers = {
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate, br",
"Connection": "keep-alive",
"User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:12.0) Gecko/20100101 Firefox/12.0",
}
transport = Transport(timeout=5, operation_timeout=5)
transport.session.headers = headers
client = Client(url, transport=transport)
return client
+131
View File
@@ -0,0 +1,131 @@
import logging
from zeep import Client, Transport
from azbankgateways.banks import BaseBank
from azbankgateways.exceptions import SettingDoesNotExist
from azbankgateways.exceptions.exceptions import BankGatewayRejectPayment
from azbankgateways.models import BankType, CurrencyEnum, PaymentStatus
class Zarinpal(BaseBank):
_merchant_code = None
_sandbox = None
def __init__(self, **kwargs):
kwargs.setdefault("SANDBOX", 0)
super(Zarinpal, self).__init__(**kwargs)
self.set_gateway_currency(CurrencyEnum.IRT)
self._payment_url = "https://www.zarinpal.com/pg/StartPay/{}/ZarinGate"
self._sandbox_url = "https://sandbox.zarinpal.com/pg/StartPay/{}/ZarinGate"
def get_bank_type(self):
return BankType.ZARINPAL
def set_default_settings(self):
for item in ["MERCHANT_CODE", "SANDBOX"]:
if item not in self.default_setting_kwargs:
raise SettingDoesNotExist(f"{item} does not exist in default_setting_kwargs")
setattr(self, f"_{item.lower()}", self.default_setting_kwargs[item])
"""
gateway
"""
@classmethod
def get_minimum_amount(cls):
return 1000
def _get_gateway_payment_url_parameter(self):
if self._sandbox:
return self._sandbox_url.format(self.get_reference_number())
return self._payment_url.format(self.get_reference_number())
def _get_gateway_payment_parameter(self):
return {}
def _get_gateway_payment_method_parameter(self):
return "GET"
"""
pay
"""
def get_pay_data(self):
description = "خرید با شماره پیگیری - {}".format(self.get_tracking_code())
return {
"Description": description,
"MerchantID": self._merchant_code,
"Amount": self.get_gateway_amount(),
"Email": None,
"Mobile": self.get_mobile_number(),
"CallbackURL": self._get_gateway_callback_url(),
}
def prepare_pay(self):
super(Zarinpal, self).prepare_pay()
def pay(self):
super(Zarinpal, self).pay()
data = self.get_pay_data()
client = self._get_client()
result = client.service.PaymentRequest(**data)
if result.Status == 100:
token = result.Authority
self._set_reference_number(token)
else:
logging.critical("Zarinpal gateway reject payment")
raise BankGatewayRejectPayment(self.get_transaction_status_text())
"""
verify from gateway
"""
def prepare_verify_from_gateway(self):
super(Zarinpal, self).prepare_verify_from_gateway()
token = self.get_request().GET.get("Authority", None)
self._set_reference_number(token)
self._set_bank_record()
def verify_from_gateway(self, request):
super(Zarinpal, self).verify_from_gateway(request)
"""
verify
"""
def get_verify_data(self):
super(Zarinpal, self).get_verify_data()
return {
"MerchantID": self._merchant_code,
"Authority": self.get_reference_number(),
"Amount": self.get_gateway_amount(),
}
def prepare_verify(self, tracking_code):
super(Zarinpal, self).prepare_verify(tracking_code)
def verify(self, transaction_code):
super(Zarinpal, self).verify(transaction_code)
data = self.get_verify_data()
client = self._get_client(timeout=10)
result = client.service.PaymentVerification(**data)
if result.Status in [100, 101]:
self._set_payment_status(PaymentStatus.COMPLETE)
else:
self._set_payment_status(PaymentStatus.CANCEL_BY_USER)
logging.debug("Zarinpal gateway unapprove payment")
def _get_client(self, timeout=5):
transport = Transport(timeout=timeout, operation_timeout=timeout)
if self._sandbox:
return Client(
"https://sandbox.zarinpal.com/pg/services/WebGate/wsdl",
transport=transport,
)
return Client(
"https://www.zarinpal.com/pg/services/WebGate/wsdl",
transport=transport,
)
+127
View File
@@ -0,0 +1,127 @@
import json
import logging
import requests
from azbankgateways.banks import BaseBank
from azbankgateways.exceptions import BankGatewayConnectionError, SettingDoesNotExist
from azbankgateways.exceptions.exceptions import BankGatewayRejectPayment
from azbankgateways.models import BankType, CurrencyEnum, PaymentStatus
from azbankgateways.utils import get_json
class Zibal(BaseBank):
_merchant_code = None
def __init__(self, **kwargs):
super(Zibal, self).__init__(**kwargs)
self.set_gateway_currency(CurrencyEnum.IRR)
self._token_api_url = "https://gateway.zibal.ir/v1/request"
self._payment_url = "https://gateway.zibal.ir/start/{}"
self._verify_api_url = "https://gateway.zibal.ir/v1/verify"
def get_bank_type(self):
return BankType.ZIBAL
def set_default_settings(self):
for item in ["MERCHANT_CODE"]:
if item not in self.default_setting_kwargs:
raise SettingDoesNotExist()
setattr(self, f"_{item.lower()}", self.default_setting_kwargs[item])
"""
gateway
"""
def _get_gateway_payment_url_parameter(self):
return self._payment_url.format(self.get_reference_number())
def _get_gateway_payment_parameter(self):
params = {}
return params
def _get_gateway_payment_method_parameter(self):
return "GET"
"""
pay
"""
def get_pay_data(self):
data = {
"merchant": self._merchant_code,
"amount": self.get_gateway_amount(),
"callbackUrl": self._get_gateway_callback_url(),
"orderId": self.get_tracking_code(),
"mobile": self.get_mobile_number(),
}
return data
def prepare_pay(self):
super(Zibal, self).prepare_pay()
def pay(self):
super(Zibal, self).pay()
data = self.get_pay_data()
response_json = self._send_data(self._token_api_url, data)
if response_json["result"] == 100:
token = response_json["trackId"]
self._set_reference_number(token)
else:
logging.critical("Zibal gateway reject payment")
raise BankGatewayRejectPayment(self.get_transaction_status_text())
"""
verify from gateway
"""
def prepare_verify_from_gateway(self):
super(Zibal, self).prepare_verify_from_gateway()
token = self.get_request().GET.get("trackId", None)
self._set_reference_number(token)
self._set_bank_record()
def verify_from_gateway(self, request):
super(Zibal, self).verify_from_gateway(request)
"""
verify
"""
def get_verify_data(self):
super(Zibal, self).get_verify_data()
data = {
"trackId": self.get_reference_number(),
"merchant": self._merchant_code,
}
return data
def prepare_verify(self, tracking_code):
super(Zibal, self).prepare_verify(tracking_code)
def verify(self, transaction_code):
super(Zibal, self).verify(transaction_code)
data = self.get_verify_data()
response_json = self._send_data(self._verify_api_url, data)
if response_json["result"] == 100 and response_json["status"] == 1:
self._set_payment_status(PaymentStatus.COMPLETE)
extra_information = json.dumps(response_json)
self._bank.extra_information = extra_information
self._bank.save()
else:
self._set_payment_status(PaymentStatus.CANCEL_BY_USER)
logging.debug("Zibal gateway unapprove payment")
def _send_data(self, api, data):
try:
response = requests.post(api, json=data, timeout=5)
except requests.Timeout:
logging.exception("Zibal time out gateway {}".format(data))
raise BankGatewayConnectionError()
except requests.ConnectionError:
logging.exception("Zibal time out gateway {}".format(data))
raise BankGatewayConnectionError()
response_json = get_json(response)
self._set_transaction_status_text(response_json["message"])
return response_json
@@ -0,0 +1,48 @@
"""Default settings for messaging."""
from django.conf import settings
from azbankgateways.apps import AZIranianBankGatewaysConfig
BANK_CLASS = getattr(
settings,
"CLASS",
{
"BMI": "azbankgateways.banks.BMI",
"SEP": "azbankgateways.banks.SEP",
"ZARINPAL": "azbankgateways.banks.Zarinpal",
"IDPAY": "azbankgateways.banks.IDPay",
"ZIBAL": "azbankgateways.banks.Zibal",
"BAHAMTA": "azbankgateways.banks.Bahamta",
"MELLAT": "azbankgateways.banks.Mellat",
"PAYV1": "azbankgateways.banks.PayV1",
},
)
_AZ_IRANIAN_BANK_GATEWAYS = getattr(settings, "AZ_IRANIAN_BANK_GATEWAYS", {})
BANK_PRIORITIES = _AZ_IRANIAN_BANK_GATEWAYS.get("BANK_PRIORITIES", [])
BANK_GATEWAYS = _AZ_IRANIAN_BANK_GATEWAYS.get("GATEWAYS", {})
BANK_DEFAULT = _AZ_IRANIAN_BANK_GATEWAYS.get("DEFAULT", "BMI")
SETTING_VALUE_READER_CLASS = _AZ_IRANIAN_BANK_GATEWAYS.get(
"SETTING_VALUE_READER_CLASS", "azbankgateways.readers.DefaultReader"
)
CURRENCY = _AZ_IRANIAN_BANK_GATEWAYS.get("CURRENCY", "IRR")
TRACKING_CODE_QUERY_PARAM = _AZ_IRANIAN_BANK_GATEWAYS.get("TRACKING_CODE_QUERY_PARAM", "tc")
TRACKING_CODE_LENGTH = _AZ_IRANIAN_BANK_GATEWAYS.get("TRACKING_CODE_LENGTH", 16)
IS_SAMPLE_FORM_ENABLE = _AZ_IRANIAN_BANK_GATEWAYS.get("IS_SAMPLE_FORM_ENABLE", False)
IS_SAFE_GET_GATEWAY_PAYMENT = _AZ_IRANIAN_BANK_GATEWAYS.get("IS_SAFE_GET_GATEWAY_PAYMENT", False)
CUSTOM_APP = _AZ_IRANIAN_BANK_GATEWAYS.get("CUSTOM_APP")
if CUSTOM_APP:
CALLBACK_NAMESPACE = f"{CUSTOM_APP}:{AZIranianBankGatewaysConfig.name}:callback"
GO_TO_BANK_GATEWAY_NAMESPACE = f"{CUSTOM_APP}:{AZIranianBankGatewaysConfig.name}:go-to-bank-gateway"
SAMPLE_RESULT_NAMESPACE = f"{CUSTOM_APP}:{AZIranianBankGatewaysConfig.name}:sample-result"
else:
CALLBACK_NAMESPACE = _AZ_IRANIAN_BANK_GATEWAYS.get(
"CALLBACK_NAMESPACE", f"{AZIranianBankGatewaysConfig.name}:callback"
)
GO_TO_BANK_GATEWAY_NAMESPACE = _AZ_IRANIAN_BANK_GATEWAYS.get(
"GO_TO_BANK_GATEWAY_NAMESPACE", f"{AZIranianBankGatewaysConfig.name}:go-to-bank-gateway"
)
SAMPLE_RESULT_NAMESPACE = _AZ_IRANIAN_BANK_GATEWAYS.get(
"SAMPLE_RESULT_NAMESPACE", f"{AZIranianBankGatewaysConfig.name}:sample-result"
)
@@ -0,0 +1,11 @@
from .exceptions import ( # noqa
AmountDoesNotSupport,
AZBankGatewaysException,
BankGatewayConnectionError,
BankGatewayStateInvalid,
BankGatewayTokenExpired,
BankGatewayUnclear,
CurrencyDoesNotSupport,
SettingDoesNotExist,
SafeSettingsEnabled,
)
@@ -0,0 +1,42 @@
class AZBankGatewaysException(Exception):
"""AZ bank gateways exception"""
class SettingDoesNotExist(AZBankGatewaysException):
"""The requested setting does not exist"""
class CurrencyDoesNotSupport(AZBankGatewaysException):
"""The requested currency does not support"""
class AmountDoesNotSupport(AZBankGatewaysException):
"""The requested amount does not support"""
class BankGatewayConnectionError(AZBankGatewaysException):
"""The requested gateway connection error"""
class BankGatewayRejectPayment(AZBankGatewaysException):
"""The requested bank reject payment"""
class BankGatewayTokenExpired(AZBankGatewaysException):
"""The requested bank token expire"""
class BankGatewayUnclear(AZBankGatewaysException):
"""The requested bank unclear"""
class BankGatewayStateInvalid(AZBankGatewaysException):
"""The requested bank unclear"""
class BankGatewayAutoConnectionFailed(AZBankGatewaysException):
"""The auto connection cant find bank"""
class SafeSettingsEnabled(AZBankGatewaysException):
"""This feature is disabled when the safe gateway is active"""
+6
View File
@@ -0,0 +1,6 @@
from django import forms
class PaymentSampleForm(forms.Form):
amount = forms.IntegerField(label="Amount", initial=10000)
mobile_number = forms.CharField(label="Mobile", max_length=13, initial="+989112223344")
@@ -0,0 +1,135 @@
# SOME DESCRIPTIVE TITLE.
# Copyright (C) YEAR THE PACKAGE'S COPYRIGHT HOLDER
# This file is distributed under the same license as the PACKAGE package.
# FIRST AUTHOR <EMAIL@ADDRESS>, YEAR.
#
#, fuzzy
msgid ""
msgstr ""
"Project-Id-Version: PACKAGE VERSION\n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2022-06-11 23:03+0300\n"
"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language-Team: LANGUAGE <LL@li.org>\n"
"Language: \n"
"MIME-Version: 1.0\n"
"Content-Type: text/plain; charset=UTF-8\n"
"Content-Transfer-Encoding: 8bit\n"
"Plural-Forms: nplurals=2; plural=(n != 1);\n"
#: apps.py:8
msgid "Iranian bank gateway"
msgstr ""
#: apps.py:9
msgid "Iranian bank gateways"
msgstr ""
#: models/banks.py:50
msgid "Status"
msgstr ""
#: models/banks.py:55
msgid "Bank"
msgstr ""
#: models/banks.py:62
msgid "Tracking code"
msgstr ""
#: models/banks.py:68
msgid "Amount"
msgstr ""
#: models/banks.py:76
msgid "Reference number"
msgstr ""
#: models/banks.py:81
msgid "Bank result"
msgstr ""
#: models/banks.py:86
msgid "Callback url"
msgstr ""
#: models/banks.py:91
msgid "Extra information"
msgstr ""
#: models/banks.py:97
msgid "Bank choose identifier"
msgstr ""
#: models/banks.py:112
msgid "Bank gateway"
msgstr ""
#: models/banks.py:113
msgid "Bank gateways"
msgstr ""
#: models/enum.py:7
msgid "BMI"
msgstr ""
#: models/enum.py:8
msgid "SEP"
msgstr ""
#: models/enum.py:9
msgid "Zarinpal"
msgstr ""
#: models/enum.py:10
msgid "IDPay"
msgstr ""
#: models/enum.py:11
msgid "Zibal"
msgstr ""
#: models/enum.py:12
msgid "Bahamta"
msgstr ""
#: models/enum.py:13
msgid "Mellat"
msgstr ""
#: models/enum.py:17
msgid "Rial"
msgstr ""
#: models/enum.py:18
msgid "Toman"
msgstr ""
#: models/enum.py:30
msgid "Waiting"
msgstr ""
#: models/enum.py:31
msgid "Redirect to bank"
msgstr ""
#: models/enum.py:32
msgid "Return from bank"
msgstr ""
#: models/enum.py:33
msgid "Cancel by user"
msgstr ""
#: models/enum.py:34
msgid "Expire gateway token"
msgstr ""
#: models/enum.py:35
msgid "Expire verify payment"
msgstr ""
#: models/enum.py:36
msgid "Complete"
msgstr ""
@@ -0,0 +1,147 @@
# SOME DESCRIPTIVE TITLE.
# Copyright (C) YEAR THE PACKAGE'S COPYRIGHT HOLDER
# This file is distributed under the same license as the PACKAGE package.
# FIRST AUTHOR <EMAIL@ADDRESS>, YEAR.
#
#, fuzzy
msgid ""
msgstr ""
"Project-Id-Version: PACKAGE VERSION\n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2022-06-11 23:04+0300\n"
"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language-Team: LANGUAGE <LL@li.org>\n"
"Language: \n"
"MIME-Version: 1.0\n"
"Content-Type: text/plain; charset=UTF-8\n"
"Content-Transfer-Encoding: 8bit\n"
"Plural-Forms: nplurals=2; plural=(n > 1);\n"
#: apps.py:8
msgid "Iranian bank gateway"
msgstr "ماژول درگاه پرداخت"
#: apps.py:9
msgid "Iranian bank gateways"
msgstr "ماژول درگاه های پرداخت"
#: models/banks.py:50
msgid "Status"
msgstr "وضعیت"
#: models/banks.py:55
msgid "Bank"
msgstr "بانک"
#: models/banks.py:62
msgid "Tracking code"
msgstr "کد پیگیری"
#: models/banks.py:68
msgid "Amount"
msgstr "مبلغ"
#: models/banks.py:76
msgid "Reference number"
msgstr "رفرنس"
#: models/banks.py:81
msgid "Bank result"
msgstr "نتیجه بانک"
#: models/banks.py:86
msgid "Callback url"
msgstr "آدرس کال بک"
#: models/banks.py:91
msgid "Extra information"
msgstr "توضیحات"
#: models/banks.py:97
msgid "Bank choose identifier"
msgstr "مشخصه درگاه"
#: models/banks.py:112
msgid "Bank gateway"
msgstr "پرداخت"
#: models/banks.py:113
msgid "Bank gateways"
msgstr "پرداخت ها"
#: models/banks.py:75
msgid "Created at"
msgstr "تاریخ ایجاد"
#: models/banks.py:76
msgid "Updated at"
msgstr "تاریخ بروزرسانی"
#: models/enum.py:7
msgid "BMI"
msgstr "بانک ملی ایران"
#: models/enum.py:8
msgid "SEP"
msgstr "بانک سامان"
#: models/enum.py:9
msgid "Zarinpal"
msgstr "زرین پال"
#: models/enum.py:10
msgid "IDPay"
msgstr "آی دی پی"
#: models/enum.py:11
msgid "Zibal"
msgstr "زیبال"
#: models/enum.py:12
msgid "Bahamta"
msgstr "باهمتا"
#: models/enum.py:13
msgid "Mellat"
msgstr "بانک ملت"
#: models/enum.py:17
msgid "Rial"
msgstr "ریال"
#: models/enum.py:18
msgid "Toman"
msgstr "تومان"
#: models/enum.py:30
msgid "Waiting"
msgstr "در انتظار"
#: models/enum.py:31
msgid "Redirect to bank"
msgstr "هدایت شده به بانک"
#: models/enum.py:32
msgid "Return from bank"
msgstr "بازگشته از بانک"
#: models/enum.py:33
msgid "Cancel by user"
msgstr "لغو توسط کاربر"
#: models/enum.py:34
msgid "Expire gateway token"
msgstr "توکن منقضی شده"
#: models/enum.py:35
msgid "Expire verify payment"
msgstr "مهلت پرداخت به اتمام رسیده"
#: models/enum.py:36
msgid "Complete"
msgstr "تکمیل شده"
#: models/enum.py:38
msgid "Unknown error acquired"
msgstr "مشکلی پیش آمده"
@@ -0,0 +1,74 @@
# Generated by Django 3.1.4 on 2020-12-06 13:35
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = []
operations = [
migrations.CreateModel(
name="Bank",
fields=[
(
"id",
models.AutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
(
"status",
models.CharField(
choices=[
("Waiting", "Waiting"),
("Redirect to bank", "Redirect To Bank"),
("Return from bank", "Return From Bank"),
("Cancel by user", "Cancel By User"),
("Expire gateway token", "Expire Gateway Token"),
("Complete", "Complete"),
],
max_length=50,
verbose_name="Status",
),
),
(
"bank_type",
models.CharField(
choices=[("BMI", "BMI"), ("ZARINPAL", "Zarinpal")],
max_length=50,
verbose_name="Bank",
),
),
(
"tracking_code",
models.CharField(max_length=255, verbose_name="Tracking code"),
),
("amount", models.CharField(max_length=10, verbose_name="Amount")),
(
"reference_number",
models.CharField(max_length=255, unique=True, verbose_name="Reference number"),
),
(
"response_result",
models.TextField(blank=True, null=True, verbose_name="Bank result"),
),
("callback_url", models.TextField(verbose_name="Callback url")),
(
"extra_information",
models.TextField(blank=True, null=True, verbose_name="Extra information"),
),
("created_at", models.DateTimeField(auto_now_add=True)),
("update_at", models.DateTimeField(auto_now=True)),
],
options={
"verbose_name": "Bank gateway",
"verbose_name_plural": "Bank gateways",
},
),
]
@@ -0,0 +1,46 @@
# Generated by Django 3.1.4 on 2021-01-02 07:21
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("azbankgateways", "0001_initial"),
]
operations = [
migrations.AlterField(
model_name="bank",
name="bank_type",
field=models.CharField(
choices=[
("BMI", "BMI"),
("SEP", "SEP"),
("ZARINPAL", "Zarinpal"),
("IDPAY", "IDPay"),
("ZIBAL", "Zibal"),
("BAHAMTA", "Bahamta"),
],
max_length=50,
verbose_name="Bank",
),
),
migrations.AlterField(
model_name="bank",
name="status",
field=models.CharField(
choices=[
("Waiting", "Waiting"),
("Redirect to bank", "Redirect To Bank"),
("Return from bank", "Return From Bank"),
("Cancel by user", "Cancel By User"),
("Expire gateway token", "Expire Gateway Token"),
("Expire verify payment", "Expire Verify Payment"),
("Complete", "Complete"),
],
max_length=50,
verbose_name="Status",
),
),
]
@@ -0,0 +1,23 @@
# Generated by Django 3.1.4 on 2021-01-04 03:14
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("azbankgateways", "0002_auto_20210102_0721"),
]
operations = [
migrations.AddField(
model_name="bank",
name="bank_choose_identifier",
field=models.CharField(
blank=True,
max_length=255,
null=True,
verbose_name="Bank choose identifier",
),
),
]
@@ -0,0 +1,30 @@
# Generated by Django 3.2 on 2021-11-15 15:00
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("azbankgateways", "0003_bank_bank_choose_identifier"),
]
operations = [
migrations.AlterField(
model_name="bank",
name="bank_type",
field=models.CharField(
choices=[
("BMI", "BMI"),
("SEP", "SEP"),
("ZARINPAL", "Zarinpal"),
("IDPAY", "IDPay"),
("ZIBAL", "Zibal"),
("BAHAMTA", "Bahamta"),
("MELLAT", "Mellat"),
],
max_length=50,
verbose_name="Bank",
),
),
]
@@ -0,0 +1,58 @@
# Generated by Django 5.0.3 on 2024-03-28 13:42
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('azbankgateways', '0004_auto_20211115_1500'),
]
operations = [
migrations.AlterField(
model_name='bank',
name='bank_type',
field=models.CharField(
choices=[
('BMI', 'BMI'),
('SEP', 'SEP'),
('ZARINPAL', 'Zarinpal'),
('IDPAY', 'IDPay'),
('ZIBAL', 'Zibal'),
('BAHAMTA', 'Bahamta'),
('MELLAT', 'Mellat'),
('PAYV1', 'PayV1'),
],
max_length=50,
verbose_name='Bank',
),
),
migrations.AlterField(
model_name='bank',
name='created_at',
field=models.DateTimeField(auto_now_add=True, verbose_name='Created at'),
),
migrations.AlterField(
model_name='bank',
name='status',
field=models.CharField(
choices=[
('WAITING', 'Waiting'),
('REDIRECT_TO_BANK', 'Redirect to bank'),
('RETURN_FROM_BANK', 'Return from bank'),
('CANCEL_BY_USER', 'Cancel by user'),
('EXPIRE_GATEWAY_TOKEN', 'Expire gateway token'),
('EXPIRE_VERIFY_PAYMENT', 'Expire verify payment'),
('COMPLETE', 'Complete'),
('ERROR', 'Unknown error acquired'),
],
max_length=50,
verbose_name='Status',
),
),
migrations.AlterField(
model_name='bank',
name='update_at',
field=models.DateTimeField(auto_now=True, verbose_name='Updated at'),
),
]
@@ -0,0 +1,20 @@
# Generated by Django 5.1.2 on 2025-03-18 13:30
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('azbankgateways', '0005_alter_bank_bank_type_alter_bank_created_at_and_more'),
('order', '0023_remove_ordermodel_bank_records'),
]
operations = [
migrations.AddField(
model_name='bank',
name='order',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to='order.ordermodel'),
),
]
@@ -0,0 +1,2 @@
from .banks import Bank # noqa
from .enum import BankType, CurrencyEnum, PaymentStatus # noqa
+92
View File
@@ -0,0 +1,92 @@
import datetime
from django.db import models
from django.utils.translation import gettext_lazy as _
from .enum import BankType, PaymentStatus
from order.models import OrderModel
class BankQuerySet(models.QuerySet):
def __init__(self, *args, **kwargs):
super(BankQuerySet, self).__init__(*args, **kwargs)
def active(self):
return self.filter()
class BankManager(models.Manager):
def get_queryset(self):
return BankQuerySet(self.model, using=self._db)
def active(self):
return self.get_queryset().active()
def update_expire_records(self):
count = (
self.active()
.filter(
status=PaymentStatus.RETURN_FROM_BANK,
update_at__lte=datetime.datetime.now() - datetime.timedelta(minutes=15),
)
.update(status=PaymentStatus.EXPIRE_VERIFY_PAYMENT)
)
count = count + self.active().filter(
status=PaymentStatus.REDIRECT_TO_BANK,
update_at__lt=datetime.datetime.now() - datetime.timedelta(minutes=15),
).update(status=PaymentStatus.EXPIRE_GATEWAY_TOKEN)
return count
def filter_return_from_bank(self):
return self.active().filter(status=PaymentStatus.RETURN_FROM_BANK)
class Bank(models.Model):
status = models.CharField(
max_length=50,
null=False,
blank=False,
choices=PaymentStatus.choices,
verbose_name=_("Status"),
)
bank_type = models.CharField(
max_length=50,
choices=BankType.choices,
verbose_name=_("Bank"),
)
# It's local and generate locally
tracking_code = models.CharField(max_length=255, null=False, blank=False, verbose_name=_("Tracking code"))
amount = models.CharField(max_length=10, null=False, blank=False, verbose_name=_("Amount"))
# Reference number return from bank
reference_number = models.CharField(
unique=True,
max_length=255,
null=False,
blank=False,
verbose_name=_("Reference number"),
)
response_result = models.TextField(null=True, blank=True, verbose_name=_("Bank result"))
callback_url = models.TextField(null=False, blank=False, verbose_name=_("Callback url"))
extra_information = models.TextField(null=True, blank=True, verbose_name=_("Extra information"))
bank_choose_identifier = models.CharField(
max_length=255, blank=True, null=True, verbose_name=_("Bank choose identifier")
)
order = models.ForeignKey(OrderModel, on_delete=models.SET_NULL, null=True, blank=True ,related_name='bank_records')
created_at = models.DateTimeField(auto_now_add=True, editable=False, verbose_name=_("Created at"))
update_at = models.DateTimeField(auto_now=True, editable=False, verbose_name=_("Updated at"))
objects = BankManager()
class Meta:
verbose_name = _("Bank gateway")
verbose_name_plural = _("Bank gateways")
def __str__(self):
return "{}-{}".format(self.pk, self.tracking_code)
@property
def is_success(self):
return self.status == PaymentStatus.COMPLETE
+37
View File
@@ -0,0 +1,37 @@
from django.db import models
from django.utils.translation import gettext_lazy as _
class BankType(models.TextChoices):
BMI = "BMI", _("BMI")
SEP = "SEP", _("SEP")
ZARINPAL = "ZARINPAL", _("Zarinpal")
IDPAY = "IDPAY", _("IDPay")
ZIBAL = "ZIBAL", _("Zibal")
BAHAMTA = "BAHAMTA", _("Bahamta")
MELLAT = "MELLAT", _("Mellat")
PAYV1 = "PAYV1", _("PayV1")
class CurrencyEnum(models.TextChoices):
IRR = "IRR", _("Rial")
IRT = "IRT", _("Toman")
@classmethod
def rial_to_toman(cls, amount):
return amount / 10
@classmethod
def toman_to_rial(cls, amount):
return amount * 10
class PaymentStatus(models.TextChoices):
WAITING = "WAITING", _("Waiting")
REDIRECT_TO_BANK = "REDIRECT_TO_BANK", _("Redirect to bank")
RETURN_FROM_BANK = "RETURN_FROM_BANK", _("Return from bank")
CANCEL_BY_USER = "CANCEL_BY_USER", _("Cancel by user")
EXPIRE_GATEWAY_TOKEN = "EXPIRE_GATEWAY_TOKEN", _("Expire gateway token")
EXPIRE_VERIFY_PAYMENT = "EXPIRE_VERIFY_PAYMENT", _("Expire verify payment")
COMPLETE = "COMPLETE", _("Complete")
ERROR = "ERROR", _("Unknown error acquired")
@@ -0,0 +1,2 @@
from .bases import Reader # noqa
from .defaults import DefaultReader # noqa
+40
View File
@@ -0,0 +1,40 @@
import abc
import six
from azbankgateways import default_settings as settings
from azbankgateways.models import BankType
@six.add_metaclass(abc.ABCMeta)
class Reader:
@abc.abstractmethod
def read(self, bank_type: BankType, identifier: str) -> dict:
"""
:param bank_type:
:param identifier:
:return:
base on bank type for example for BMI:
{
'MERCHANT_CODE': '<YOUR INFO>',
'TERMINAL_CODE': '<YOUR INFO>',
'SECRET_KEY': '<YOUR INFO>',
}
"""
pass
def klass(self, bank_type: BankType, identifier: str) -> dict:
return settings.BANK_CLASS[bank_type]
@abc.abstractmethod
def get_bank_priorities(self, identifier: str) -> list:
pass
@abc.abstractmethod
def default(self, identifier: str):
pass
@abc.abstractmethod
def currency(self, identifier: str):
pass
@@ -0,0 +1,32 @@
from azbankgateways import default_settings as settings
from azbankgateways.models import BankType
from .bases import Reader
class DefaultReader(Reader):
def read(self, bank_type: BankType, identifier: str) -> dict:
"""
:param bank_type:
:param identifier:
:return:
base on bank type for example for BMI:
{
'MERCHANT_CODE': '<YOUR INFO>',
'TERMINAL_CODE': '<YOUR INFO>',
'SECRET_KEY': '<YOUR INFO>',
}
"""
return settings.BANK_GATEWAYS[bank_type]
def default(self, identifier: str):
return settings.BANK_DEFAULT
def currency(self, identifier: str):
return settings.CURRENCY
def get_bank_priorities(self, identifier: str) -> list:
priorities = [self.default(identifier)]
priorities = list(dict.fromkeys(priorities + settings.BANK_PRIORITIES))
return priorities
@@ -0,0 +1,28 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title></title>
</head>
<body>
<form id='id_form' action="{{ url }}" method="{{ method }}">
{% csrf_token %}
{% for key, value in params.items %}
<input type="hidden" name="{{ key }}" value="{{ value }}">
{% endfor %}
</form>
<script type="text/javascript">
window.onload = function () {
function submitForm() {
document.forms['id_form'].submit();
}
submitForm();
}
</script>
</body>
</html>
@@ -0,0 +1,30 @@
{% load i18n static %}
<!DOCTYPE html>
{% get_current_language as LANGUAGE_CODE %}{% get_current_language_bidi as LANGUAGE_BIDI %}
<html lang="{{ LANGUAGE_CODE|default:"en-us" }}" {% if LANGUAGE_BIDI %}dir="rtl"{% endif %}>
<head>
<title>{% block title %}{% endblock %}</title>
{% block extrastyle %}{% endblock %}
{% if LANGUAGE_BIDI %}
{% endif %}
{% block extrahead %}{% endblock %}
{% block extrameta %}{% endblock %}
{% block blockbots %}
<meta name="robots" content="index, follow">
{% endblock %}
<meta charset="utf-8">
<link
rel="stylesheet"
href="https://stackpath.bootstrapcdn.com/bootstrap/4.3.1/css/bootstrap.min.css"
integrity="sha384-ggOyR0iXCbMQv3Xipma34MD+dH/1fQ784/j6cY/iJTQUOhcWr7x9JvoRxT2MZw1T"
crossorigin="anonymous"
/>
</head>
<body>
<div class="container">
{% block content %} {% endblock %}
</div>
</body>
</html>
@@ -0,0 +1,7 @@
{% extends "azbankgateways/samples/base.html" %}
{% load static %}
{% block extrameta %}
{{ block.super }}
{% endblock %}
@@ -0,0 +1,19 @@
{% extends "azbankgateways/samples/base_site.html" %}
{% load static %}
{% block title %}پرداخت{% endblock %}
{% block content %}
<div class="mt-5">
<form method="post">
{% csrf_token %}
{% for field in form %}
<div class="form-group">
<label>{{ field.label_tag }}</label>
<input type="{{ field.type }}" class="form-control" id="{{ field.name }}" name="{{ field.name }}"
placeholder="{{ field.label_name }}" value="{{ field.initial }}">
</div>
{% endfor %}
<button type="submit" class="btn btn-primary">Submit</button>
</form>
</div>
{% endblock %}
@@ -0,0 +1,21 @@
{% extends "azbankgateways/samples/base_site.html" %}
{% load static %}
{% block title %}نتیجه پرداخت{% endblock %}
{% block content %}
<div class="mt-5">
<div class="card text-center">
<div class="card-header">
{{ bank_record.bank_type }} - نتیجه پرداخت
</div>
<div class="card-body">
<h5 class="card-title {% if bank_record.is_success %}bg-success{% else %}bg-danger{% endif %}">{% if bank_record.is_success %}پرداخت موفق{% else %}پرداخت نا موفق{% endif %}</h5>
<p class="card-text">{{ bank_record.tracking_code }} - {{ bank_record.reference_number }} - {{ bank_record.response_result }} - {{ bank_record.extra_information }}</p>
<a href="{{ bank_record.callback_url|cut:'/sample-result/'|add:'/sample-payment/' }}" class="btn btn-primary">Go somewhere</a>
</div>
<div class="card-footer text-muted">
{{ bank_record.created_at }}
</div>
</div>
</div>
{% endblock %}
+3
View File
@@ -0,0 +1,3 @@
import typing
DictQuerystring = typing.Tuple[str, dict]
+31
View File
@@ -0,0 +1,31 @@
from django.urls import path
from . import default_settings as settings
from .apps import AZIranianBankGatewaysConfig
from .views import (
callback_view,
go_to_bank_gateway,
sample_payment_view,
sample_result_view,
)
app_name = AZIranianBankGatewaysConfig.name
_urlpatterns = [
path("callback/", callback_view, name="callback"),
]
if not settings.IS_SAFE_GET_GATEWAY_PAYMENT:
_urlpatterns += [
path("go-to-bank-gateway/", go_to_bank_gateway, name="go-to-bank-gateway"),
]
if settings.IS_SAMPLE_FORM_ENABLE:
_urlpatterns += [
path("sample-payment/", sample_payment_view, name="sample-payment"),
path("sample-result/", sample_result_view, name="sample-result"),
]
def az_bank_gateways_urls():
return _urlpatterns, app_name, app_name
+35
View File
@@ -0,0 +1,35 @@
import json
from urllib import parse
from azbankgateways.types import DictQuerystring
def get_json(resp):
"""
:param response:returned response as json when sending a request
using 'requests' module.
:return:response's content with json format
"""
return json.loads(resp.content.decode("utf-8"))
def append_querystring(url: str, params: dict) -> str:
url_parts = list(parse.urlparse(url))
query = dict(parse.parse_qsl(url_parts[4]))
query.update(params)
url_parts[4] = parse.urlencode(query)
return parse.urlunparse(url_parts)
def split_to_dict_querystring(url: str) -> DictQuerystring:
url_parts = list(parse.urlparse(url))
query = dict(parse.parse_qsl(url_parts[4]))
url_parts[4] = ""
url_parts[5] = ""
return parse.urlunparse(url_parts), query
+2
View File
@@ -0,0 +1,2 @@
from .banks import callback_view, go_to_bank_gateway # noqa
from .samples import sample_payment_view, sample_result_view # noqa
+39
View File
@@ -0,0 +1,39 @@
import logging
from urllib.parse import unquote
from django.http import Http404
from django.shortcuts import render
from django.views.decorators.csrf import csrf_exempt
from azbankgateways.bankfactories import BankFactory
from azbankgateways.exceptions import AZBankGatewaysException
@csrf_exempt
def callback_view(request):
bank_type = request.GET.get("bank_type", None)
identifier = request.GET.get("identifier", None)
if not bank_type:
logging.critical("Bank type is required. but it doesnt send.")
raise Http404
factory = BankFactory()
bank = factory.create(bank_type, identifier=identifier)
try:
bank.verify_from_gateway(request)
except AZBankGatewaysException:
logging.exception("Verify from gateway failed.", stack_info=True)
return bank.redirect_client_callback()
@csrf_exempt
def go_to_bank_gateway(request):
context = {"params": {}}
for key, value in request.GET.items():
if key == "url" or key == "method":
context[key] = unquote(value)
else:
context["params"][key] = unquote(value)
return render(request, "azbankgateways/redirect_to_bank.html", context=context)
+68
View File
@@ -0,0 +1,68 @@
import logging
from django.http import Http404
from django.shortcuts import render
from django.urls import reverse
from azbankgateways import bankfactories
from azbankgateways import default_settings as settings
from azbankgateways import models as bank_models
from azbankgateways.apps import AZIranianBankGatewaysConfig
from azbankgateways.exceptions import AZBankGatewaysException
from ..forms import PaymentSampleForm
def sample_payment_view(request):
# if this is a POST request we need to process the form data
if request.method == "POST":
# create a form instance and populate it with data from the request:
form = PaymentSampleForm(request.POST)
# check whether it's valid:
if form.is_valid():
amount = form.cleaned_data["amount"]
mobile_number = form.cleaned_data["mobile_number"]
factory = bankfactories.BankFactory()
try:
bank = factory.auto_create()
bank.set_request(request)
bank.set_amount(amount)
# یو آر ال بازگشت به نرم افزار برای ادامه فرآیند
bank.set_client_callback_url(reverse(settings.SAMPLE_RESULT_NAMESPACE))
bank.set_mobile_number(mobile_number) # اختیاری
# در صورت تمایل اتصال این رکورد به رکورد فاکتور یا هر چیزی که
# بعدا بتوانید ارتباط بین محصول یا خدمات را با این
# پرداخت برقرار کنید.
bank_record = bank.ready() # noqa
# هدایت کاربر به درگاه بانک
if settings.IS_SAMPLE_FORM_ENABLE:
return render(request, 'azbankgateways/redirect_to_bank.html', context=bank.get_gateway())
return bank.redirect_gateway()
except AZBankGatewaysException as e:
logging.critical(e)
# TODO: redirect to failed result.
raise e
# if a GET (or any other method) we'll create a blank form
else:
form = PaymentSampleForm()
return render(request, "azbankgateways/samples/gateway.html", {"form": form})
def sample_result_view(request):
tracking_code = request.GET.get(settings.TRACKING_CODE_QUERY_PARAM, None)
if not tracking_code:
logging.debug("این لینک معتبر نیست.")
raise Http404
try:
bank_record = bank_models.Bank.objects.get(tracking_code=tracking_code)
except bank_models.Bank.DoesNotExist:
logging.debug("این لینک معتبر نیست.")
raise Http404
return render(request, "azbankgateways/samples/result.html", {"bank_record": bank_record})
+4
View File
@@ -65,4 +65,8 @@ CELERY_BEAT_SCHEDULE = {
'task': 'product.tasks.update_product_prices',
'schedule': crontab(minute='*'),
},
'update-bank-record-every-minute': {
'task': 'order.tasks.udpate_bank_status',
'schedule': crontab(minute='*'),
},
}
+3 -2
View File
@@ -5,9 +5,10 @@ def main():
settings_module = "core.settings.production"
if "--develop" in sys.argv:
if "--develop" in sys.argv or '-d' in sys.argv:
settings_module = "core.settings.development"
sys.argv.remove("--develop")
dev_flag = '--develop' if "--develop" in sys.argv else '-d'
sys.argv.remove(dev_flag)
os.environ.setdefault("DJANGO_SETTINGS_MODULE", settings_module)
+55 -16
View File
@@ -1,4 +1,4 @@
from django.contrib import admin
from django.contrib import admin, messages
from .models import *
from unfold.admin import TabularInline, StackedInline
@@ -8,6 +8,11 @@ from unfold.contrib.forms.widgets import ArrayWidget, WysiwygWidget
from django.contrib.postgres.fields import ArrayField
from utils.admin import ModelAdmin
from django.utils.html import format_html, format_html_join
from azbankgateways.models.banks import Bank
from unfold.decorators import action
from django.shortcuts import redirect
class OrderItemModelInline(StackedInline):
model = OrderItemModel
extra = 0
@@ -26,34 +31,68 @@ class DiscountCodeAdmin(ModelAdmin, ImportExportModelAdmin):
list_display = ['code', 'expiration_date', 'percent', 'quantity']
class BankRecordInline(StackedInline):
model = Bank
extra = 0
max_num = 0
def has_delete_permission(self, request, obj=None):
return False
def get_readonly_fields(self, request, obj=None):
return [field.name for field in self.model._meta.fields]
@admin.register(OrderModel)
class OrderAdmin(ModelAdmin, ImportExportModelAdmin):
import_form_class = ImportForm
export_form_class = ExportForm
list_filter = ['is_paid', 'status']
list_display = ['user', 'is_paid', 'status', 'discount_code', 'address', ]
readonly_fields = ('created_at', 'bank_links')
actions_list = ['redirect_to_learn', 'udpate_bank_status']
list_display = ['user', 'is_paid', 'status', 'discount_code', 'address',]
readonly_fields = ('created_at', )
compressed_fields = True
warn_unsaved_form = True
exclude = ('bank_records',)
# exclude = ('bank_records',)
formfield_overrides = {
ArrayField: {
"widget": ArrayWidget,
}
}
inlines = [OrderItemModelInline]
def bank_links(self, obj):
banks = obj.bank_records.all()
inlines = [OrderItemModelInline, BankRecordInline]
# def bank_links(self, obj):
# banks = obj.bank_records.all()
if not banks.exists():
return "-"
# if not banks.exists():
# return "-"
return format_html_join(
"",
'<a style="padding-bottom:10px;display:block;" href="/secret-admin/azbankgateways/bank/{}/change/" class="text-primary-600 dark:text-primary-500">{}</a>',
[(bank.id, bank.tracking_code) for bank in banks]
) or "-"
# return format_html_join(
# "",
# '<a style="padding-bottom:10px;display:block;" href="/secret-admin/azbankgateways/bank/{}/change/" class="text-primary-600 dark:text-primary-500">{}</a>',
# [(bank.id, bank.tracking_code) for bank in banks]
# ) or "-"
bank_links.short_description = "Bank Records"
# bank_links.short_description = "Bank Records"
@action(description='اپدیت وضعیت رکورد های بانکی')
def udpate_bank_status(self, request):
import logging
from azbankgateways import (
bankfactories,
models as bank_models,
default_settings as settings,
)
factory = bankfactories.BankFactory()
bank_models.Bank.objects.update_expire_records()
for item in bank_models.Bank.objects.filter_return_from_bank():
bank = factory.create(
bank_type=item.bank_type, identifier=item.bank_choose_identifier
)
bank.verify(item.tracking_code)
bank_record = bank_models.Bank.objects.get(tracking_code=item.tracking_code)
if bank_record.is_success:
logging.debug("This record is verify now.", extra={"pk": bank_record.pk})
messages.success(request, f"با موفقیت اپدیت شد")
return redirect("admin:order_ordermodel_changelist")
@@ -0,0 +1,17 @@
# Generated by Django 5.1.2 on 2025-03-18 13:30
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('order', '0022_alter_orderitemmodel_price'),
]
operations = [
migrations.RemoveField(
model_name='ordermodel',
name='bank_records',
),
]
+1 -2
View File
@@ -4,7 +4,6 @@ from product.models import ProductModel, ProductVariant, ProductImageModel
from django.utils import timezone
from .execptions import DiscountNotAvailableError
from django_jalali.db import models as jmodels
from azbankgateways.models.banks import Bank
class DiscountCode(models.Model):
@@ -56,7 +55,7 @@ class OrderModel(models.Model):
tax = models.BigIntegerField(null=True, blank=True, verbose_name='مالیات')
final_price = models.BigIntegerField(null=True, blank=True, verbose_name='قیمت نهایی')
cart_total = models.BigIntegerField(null=True, blank=True, verbose_name='کل سبد خرید')
bank_records = models.ManyToManyField(Bank, max_length=100, verbose_name='رکورد بانکی', null=True, blank=True)
# bank_records = models.ManyToManyField(Bank, max_length=100, verbose_name='رکورد بانکی', null=True, blank=True)
def __str__(self):
return f'سفارش: {self.id + 1000}'
+17 -10
View File
@@ -5,15 +5,22 @@ from azbankgateways import (
default_settings as settings,
)
# factory = bankfactories.BankFactory()
# bank_models.Bank.objects.update_expire_records()
from celery import shared_task
# for item in bank_models.Bank.objects.filter_return_from_bank():
# bank = factory.create(
# bank_type=item.bank_type, identifier=item.bank_choose_identifier
# )
# bank.verify(item.tracking_code)
# bank_record = bank_models.Bank.objects.get(tracking_code=item.tracking_code)
# if bank_record.is_success:
# logging.debug("This record is verify now.", extra={"pk": bank_record.pk})
@shared_task
def udpate_bank_status():
factory = bankfactories.BankFactory()
bank_models.Bank.objects.update_expire_records()
for item in bank_models.Bank.objects.filter_return_from_bank():
bank = factory.create(
bank_type=item.bank_type, identifier=item.bank_choose_identifier
)
bank.verify(item.tracking_code)
bank_record = bank_models.Bank.objects.get(tracking_code=item.tracking_code)
if bank_record.is_success:
logging.debug("This record is verify now.", extra={"pk": bank_record.pk})
print('update bank record is done')
+4 -3
View File
@@ -205,9 +205,10 @@ class PaymentView(APIView):
bank.set_mobile_number(user_mobile_number)
bank_record = bank.ready()
cart_order.bank_records.add(bank_record)
cart_order.save()
print(bank.redirect_gateway().url)
# cart_order.bank_records.add(bank_record)
# cart_order.save()
bank_record.order = cart_order
bank_record.save()
return Response(bank.redirect_gateway().url)
except AZBankGatewaysException as e:
print(e)
-1
View File
@@ -6,7 +6,6 @@ annotated-types==0.7.0
anyio==4.6.0
asgiref==3.8.1
attrs==24.2.0
az-iranian-bank-gateways==2.0.5
beautifulsoup4==4.12.3
billiard==4.2.1
boto3==1.36.26