Untitled
unknown
plain_text
a year ago
27 kB
5
Indexable
import re
import json
import os
from django.conf import settings
from rest_framework import serializers
from rest_framework.fields import empty
from datetime import datetime, timedelta
from AdminApp.models import KycCustomerDetails
from django.core.validators import (
RegexValidator,
MaxLengthValidator,
MinLengthValidator,
)
from .choices import (
PURPOSE_CHOICES,
OCCUPATION_CHOICES,
SOURCE_OF_FUND_CHOICES,
MODE_OF_OPERATION_CHOICES,
EDUCATION_CHOICES,
MONTHLY_INCOME_CHOICES,
MONTHLY_WITHDRAWAL_CHOICES,
EXPECTED_REMITTANCE_CHOICES,
LATEST_INCOME_CHOICES,
RELATIONSHIP_WITH_APPLICANT_CHOICES,
ADDRESS_TYPE_NRO_CHOICES
)
class NonAssistantSerializer(serializers.Serializer):
senderId = serializers.CharField(max_length=10)
requestId = serializers.CharField(max_length=15)
mashreqCIFId = serializers.CharField(max_length=60)
emiratesId = serializers.CharField(max_length=30)
passportNum = serializers.CharField(max_length=10)
passportExpDt = serializers.CharField(max_length=10)
passportIssueDt = serializers.CharField(max_length=10)
overseasAddress = serializers.CharField(max_length=500)
name = serializers.CharField(max_length=60)
dateofbirth = serializers.CharField(max_length=10)
Gender = serializers.CharField(max_length=10)
email = serializers.EmailField()
mobilenum = serializers.CharField(max_length=20)
# created = serializers.DateTimeField()
def validate_senderId(self, value):
if not re.match("^[a-zA-Z]+$", value):
raise serializers.ValidationError("No special characters permitted.")
return value
def validate_requestId(self, value):
if not re.match("^[0-9]*$", value):
raise serializers.ValidationError("Only numbers permitted.")
return value
def name_validator(value):
pattern = r"^[a-zA-Z]+$"
if not re.match(pattern, value):
raise serializers.ValidationError("Error")
class RevertSerializers(serializers.ModelSerializer):
validator_path = os.path.join(
settings.BASE_DIR,
"nonassistant",
"json",
"customer_revert_field_validators.json",
)
class Meta:
model = KycCustomerDetails
fields = "__all__"
read_only_fields = ("uniqueid",)
def validate(self, attrs):
json_validators = []
with open(self.validator_path, "rb") as f:
json_validators = json.load(f)["fields"]
for key, value in attrs.items():
try:
field = next(
filter(lambda d: d.get("name", "") == key, json_validators), None
)
# if not field:
# continue
try:
validators = field["validators"]
except KeyError:
raise serializers.ValidationError(
"Missing validators for %s in json struct" % field["name"]
)
except TypeError:
raise serializers.ValidationError(
'Key "%s" not found in json struct' % field["name"]
)
for validator in validators:
try:
if validator["type"] == "length":
max = validator["params"].get("max", None)
min = validator["params"].get("min", None)
if max and len(value) > int(max["value"]):
raise serializers.ValidationError(max["message"])
if min and len(value) < int(min["value"]):
raise serializers.ValidationError(min["message"])
elif validator["type"] == "regex":
regex_validator = validator["params"]["regex"]
regex = regex_validator["value"]
if re.match(rf"{regex}", value) is None:
raise serializers.ValidationError(
regex_validator["message"]
)
except KeyError:
# skip this validation; as invalid validator format in json
# log this and correct json file
print(f"Failed... {value}")
continue
except (IndexError, KeyError) as e:
# skip this validation; as invalid validator format in json
# log this and correct json file
print("json loading failed")
raise e
return super().validate(attrs)
class DynamicFieldSerializer(serializers.Serializer):
validator_path = os.path.join(
settings.BASE_DIR,
"nonassistant",
"json",
"customer_revert_field_validators.json",
)
def __init__(self, instance=None, data=..., **kwargs):
super().__init__(instance, data, **kwargs)
self.raw_json_validators = []
with open(self.validator_path, "rb") as f:
self.raw_json_validators = json.load(f)["fields"]
for _field_name, ser_field in self.fields.items():
field = next(
filter(
lambda d: d.get("name", "") == _field_name, self.raw_json_validators
),
None,
)
if field:
for validator in field.get("validators", {}):
if validator["type"].lower() == "regex":
regex_param = validator["params"]["regex"]
regex = rf'{regex_param["value"]}'
err_msg = regex_param.get("message", "Invalid characters")
if regex:
ser_field.validators = [RegexValidator(regex, err_msg)]
def validate(self, attrs):
for key, value in attrs.items():
try:
field = next(
filter(
lambda d: d.get("name", "") == key, self.raw_json_validators
),
None,
)
if not field:
continue
try:
validators = field["validators"]
data_type = field.get("data_type", "string")
except KeyError:
raise serializers.ValidationError(
"Missing validators for %s in json struct" % field["name"]
)
except TypeError:
raise serializers.ValidationError(
'Key "%s" not found in json struct' % field["name"]
)
# Convert value to the specified data type
if data_type == "number":
if value:
try:
value = int(value)
except ValueError:
raise serializers.ValidationError(
{
key: [
"Invalid data type for %s. Expected number."
% key
]
}
)
else:
pass
if data_type == "date":
if value:
try:
# Convert value to date format (assuming mm/dd/yyyy)
value = datetime.strptime(value, "%d/%m/%Y").date()
except ValueError:
raise serializers.ValidationError(
{
key: [
"Invalid date format for %s. Expected dd/mm/yyyy."
% key
]
}
)
else:
pass
for validator in validators:
if value:
try:
if validator["type"] == "length":
max_value = (
validator["params"].get("max", {}).get("value")
)
min_value = (
validator["params"].get("min", {}).get("value")
)
if max_value is not None and len(str(value)) > int(
max_value
):
raise serializers.ValidationError(
{key: validator["params"]["max"]["message"]}
)
if min_value is not None and len(str(value)) < int(
min_value
):
raise serializers.ValidationError(
{key: validator["params"]["min"]["message"]}
)
if validator["type"] == "date":
max_value = (
validator["params"].get("max", {}).get("value")
)
min_value = (
validator["params"].get("min", {}).get("value")
)
if max_value is not None:
if int(max_value) == 0:
# If max is 0, allow up to today's date
today_date = datetime.now().date()
if value > today_date:
raise serializers.ValidationError(
{
key: validator["params"]["max"][
"message"
]
}
)
if int(max_value) < 0:
# If max is negative, go backward by the absolute value of max_value
max_timedelta = timedelta(
days=abs(int(max_value))
)
# Calculate the maximum allowed date
max_allowed_date = (
datetime.now().date() - max_timedelta
)
# Check if the provided date is on or before the calculated date
if value > max_allowed_date:
raise serializers.ValidationError(
{
key: validator["params"]["max"][
"message"
]
}
)
if min_value is not None:
# Convert the min_value to a timedelta
min_timedelta = timedelta(days=int(min_value))
# Calculate the minimum allowed date
min_allowed_date = (
datetime.now().date() + min_timedelta
)
# Check if the provided date is at least 6 months in the future
if value < min_allowed_date:
raise serializers.ValidationError(
{key: validator["params"]["min"]["message"]}
)
# Additional validation logic
if key == "nominee_name":
applicant_name = attrs.get("name", "")
co_applicant_name = attrs.get("co_applicant_name", "")
guardian_name = attrs.get("guardian_name", "")
# Capitalize the values and handle None
value = value.capitalize() if value else None
applicant_name = (
applicant_name.capitalize()
if applicant_name
else None
)
co_applicant_name = (
co_applicant_name.capitalize()
if co_applicant_name
else None
)
guardian_name = (
guardian_name.capitalize()
if guardian_name
else None
)
if (
value == applicant_name
or value == co_applicant_name
):
raise serializers.ValidationError(
{
key: "Nominee Name must be different from the applicant or co-applicant names."
}
)
if value == guardian_name:
raise serializers.ValidationError(
{
key: "Guardian's Name must be different from the Nominee Name."
}
)
if key == "guardian_name":
nominee_name = attrs.get("nominee_name", "")
applicant_name = attrs.get("name", "")
co_applicant_name = attrs.get("co_applicant_name", "")
# Capitalize the values and handle None
value = value.capitalize() if value else None
nominee_name = (
nominee_name.capitalize() if nominee_name else None
)
applicant_name = (
applicant_name.capitalize()
if applicant_name
else None
)
co_applicant_name = (
co_applicant_name.capitalize()
if co_applicant_name
else None
)
if any(
[
value == nominee_name,
value == applicant_name,
value == co_applicant_name,
]
):
raise serializers.ValidationError(
{
key: "Guardian's Name must be different from the Nominee Name/Applicant Name/ Co Applicant Name."
}
)
if key == "passport_expiry_date":
issue_date_str = attrs.get("passport_issue_date", "")
issue_date = datetime.strptime(
issue_date_str, "%m/%d/%Y"
).date()
# Assuming `value` is a date object
expiry_date = value
# Check if the expiry date is at least 10 years from the issue date
if expiry_date > issue_date + timedelta(days=3652):
raise serializers.ValidationError(
{
key: "The passport must have a validity of 10 years or less."
}
)
if key == "co_passport_expiry_date":
issue_date_str = attrs.get("co_passport_issue_date", "")
issue_date = datetime.strptime(
issue_date_str, "%m/%d/%Y"
).date()
# Assuming `value` is a date object
expiry_date = value
# Check if the expiry date is at least 10 years from the issue date
if expiry_date > issue_date + timedelta(days=3652):
raise serializers.ValidationError(
{
key: "The passport must have a validity of 10 years or less."
}
)
except KeyError:
# skip this validation; as an invalid validator format in json
# log this and correct the json file
print(f"Failed... {value}")
continue
except (IndexError, KeyError) as e:
# skip this validation; as an invalid validator format in json
# log this and correct the json file
print("json loading failed")
raise e
tax_residents_list = attrs.get("tax_residentsCountryList", [])
for tax_resident in tax_residents_list:
self.validate_tax_resident(tax_resident)
return super().validate(attrs)
def validate_tax_resident(self, tax_resident):
# Add validation logic for values inside tax_residentsCountryList
# You can access individual fields within tax_resident dictionary
country = tax_resident.get("country", "")
tax_id_number = tax_resident.get("Tax_Identification_Number", "")
identification_type = tax_resident.get("identification_type", "")
description = tax_resident.get("Tax_Identification_description", "")
# Validate country
if not all(char.isalpha() or char == '@' for char in country) or any(char.isdigit() for char in country):
raise serializers.ValidationError(
{"tax_residentsCountryList": "Invalid characters in the country field"}
)
if len(country) > 50:
raise serializers.ValidationError(
{"tax_residentsCountryList": "Country length exceeds 50 characters limit"}
)
# Validate Tax_Identification_Number
if not tax_id_number.isdigit() or len(tax_id_number) > 20:
raise serializers.ValidationError(
{"tax_residentsCountryList": "Invalid Tax_Identification_Number"}
)
# Validate identification_type
if not identification_type.isalpha() or len(identification_type) > 10:
raise serializers.ValidationError(
{"tax_residentsCountryList": "Invalid identification_type"}
)
# Validate Tax_Identification_description
if not description.isalpha() or len(description) > 30:
raise serializers.ValidationError(
{"tax_residentsCountryList": "Invalid Tax_Identification_description"}
)
class Stepper1Serializer(DynamicFieldSerializer):
account_type = serializers.CharField(required=True)
purpose_of_account_opening = serializers.ChoiceField(choices=PURPOSE_CHOICES)
monthly_withdrawal = serializers.ChoiceField(choices=MONTHLY_WITHDRAWAL_CHOICES)
monthly_remittance = serializers.ChoiceField(choices=EXPECTED_REMITTANCE_CHOICES)
monthly_income = serializers.ChoiceField(choices=LATEST_INCOME_CHOICES)
occupation = serializers.ChoiceField(choices=OCCUPATION_CHOICES)
source_of_fund = serializers.ChoiceField(choices=SOURCE_OF_FUND_CHOICES)
class Stepper2Serializer(DynamicFieldSerializer):
is_nominee_required = serializers.BooleanField(default=False)
nominee_name = serializers.CharField(allow_blank=True, required=False,allow_null=True)
nominee_date_of_birth = serializers.CharField(allow_blank=True, required=False,allow_null=True)
nominee_age = serializers.CharField(allow_blank=True, required=False,allow_null=True)
is_nominee_minor = serializers.BooleanField(required=False, default=False)
nominee_relation = serializers.ChoiceField(
choices=RELATIONSHIP_WITH_APPLICANT_CHOICES, allow_blank=True
)
nominee_address = serializers.CharField(required=False, allow_null=True, allow_blank=True)
nominee_address1 = serializers.CharField(allow_blank=True, required=False,allow_null=True)
nominee_address2 = serializers.CharField(allow_blank=True, required=False,allow_null=True)
nominee_city = serializers.CharField(required=False,allow_null=True,allow_blank=True)
nominee_state = serializers.CharField(required=False,allow_null=True,allow_blank=True)
nominee_pincode = serializers.CharField(required=False,allow_null=True,allow_blank=True)
nominee_country = serializers.CharField(required=False,allow_null=True,allow_blank=True)
guardian_name = serializers.CharField(required=False,allow_null=True,allow_blank=True)
guardian_address = serializers.CharField(required=False,allow_null=True,allow_blank=True)
guardian_address1 = serializers.CharField(required=False,allow_null=True,allow_blank=True)
guardian_address2 = serializers.CharField(required=False,allow_null=True,allow_blank=True)
guardian_city = serializers.CharField(required=False,allow_null=True,allow_blank=True)
guardian_state = serializers.CharField(required=False,allow_null=True,allow_blank=True)
guardian_country = serializers.CharField(required=False,allow_null=True,allow_blank=True)
guardian_pin = serializers.CharField(required=False,allow_null=True,allow_blank=True)
guardian_age = serializers.CharField(required=False,allow_null=True,allow_blank=True)
guardian_date_of_birth = serializers.CharField(required=False,allow_null=True,allow_blank=True)
add_existing_nominee = serializers.BooleanField(required=False,default=False, allow_null=True)
class Stepper3Serializer(DynamicFieldSerializer):
facta_crs = serializers.BooleanField(required=False,default=False)
place_of_birth = serializers.CharField(required=False,allow_null=True, allow_blank=True)
country_of_birth = serializers.CharField(required=False,allow_null=True, allow_blank=True)
communication_address = serializers.CharField(required=False,allow_null=True, allow_blank=True)
address_type=serializers.ChoiceField(choices=ADDRESS_TYPE_NRO_CHOICES,required=False)
permenant_address = serializers.CharField(required=False,allow_null=True, allow_blank=True)
resident_of_India =serializers.BooleanField(required=False,default=False)
tax_residentsCountryList = serializers.ListField(default=list,required=False)
class Stepper4Serializer(DynamicFieldSerializer):
estimated_agricultural_income = serializers.CharField(required=True)
estimated_non_agricultural_income = serializers.CharField(required=True)
pan_card_applied = serializers.BooleanField(required=False)
pan_acknowledgemnt_number = serializers.CharField(required=False,allow_null=True, allow_blank=True)
pan_applied_date =serializers.CharField(required=False,allow_null=True, allow_blank=True)
class Stepper5Serializer(DynamicFieldSerializer):
aof_download = serializers.BooleanField(default=False)
class OtpRequestSerializer(serializers.Serializer):
mobilenum = serializers.RegexField(
regex=r"^[0-9]+$",
required=True,
error_messages={
"required": "Mobile number is required.",
"blank": "Mobile number is required and cannot be empty.",
"invalid": "Invalid mobile number provided.",
},
)
email = serializers.EmailField(max_length=50, required=False, allow_blank=True)
# def validate(self, attrs):
# if "mobilenum" not in attrs or "email" not in attrs:
# raise serializers.ValidationError("Mobile or Email ID is mandatory")
class CustomerRequestSerializer(OtpRequestSerializer):
name = serializers.RegexField(
regex=r"^[a-zA-Z ]+$",
required=False,
error_messages={
"required": "Customer name is mandatory.",
"blank": "Customer name is required and cannot be empty.",
"invalid": "Invalid name provided.",
},
validators=[
MinLengthValidator(
limit_value=3, message="Name must be at least 2 characters."
),
MaxLengthValidator(
limit_value=60, message="Name cannot exceed 60 characters."
),
],
)
class NROCustomerRequestSerializer(serializers.Serializer):
account_number = serializers.RegexField(
regex=r"^[0-9 ]+$",
required=False,
error_messages={
"required": "Account Number is mandatory",
"blank": "Account Number is mandatory is required and cannot be empty.",
"invalid": "Invalid Account Number is provided.",
},
validators=[
MinLengthValidator(
limit_value=14,
message="Account Number must be at least 14 characters.",
),
MaxLengthValidator(
limit_value=14,
message="Account Number is mandatory cannot exceed 14 characters.",
),
],
)
Editor is loading...
Leave a Comment