156 lines
6.5 KiB
Python
156 lines
6.5 KiB
Python
import logging
|
|
import decimal
|
|
import datetime
|
|
|
|
from . import stripe as uncloud_stripe
|
|
import stripe
|
|
from .models import PricingPlan, BillingAddress
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
eu_countries = ['at', 'be', 'bg', 'ch', 'cy', 'cz', 'hr', 'dk',
|
|
'ee', 'fi', 'fr', 'mc', 'de', 'gr', 'hu', 'ie', 'it',
|
|
'lv', 'lu', 'mt', 'nl', 'po', 'pt', 'ro','sk', 'si', 'es',
|
|
'se', 'gb']
|
|
|
|
def validate_vat_number(stripe_customer_id, billing_address_id):
|
|
try:
|
|
billing_address = BillingAddress.objects.get(id=billing_address_id)
|
|
except BillingAddress.DoesNotExist as dne:
|
|
billing_address = None
|
|
except BillingAddress.MultipleObjectsReturned as mor:
|
|
billing_address = BillingAddress.objects.filter(id=billing_address_id).order_by('-id').first()
|
|
if billing_address is not None:
|
|
logger.debug("BillingAddress found: %s %s" % (
|
|
billing_address_id, str(billing_address)))
|
|
if billing_address.country.lower().strip() not in eu_countries:
|
|
return {
|
|
"validated_on": "",
|
|
"status": "not_needed"
|
|
}
|
|
if billing_address.vat_number_validated_on and billing_address.vat_number_verified:
|
|
return {
|
|
"validated_on": billing_address.vat_number_validated_on,
|
|
"status": "verified"
|
|
}
|
|
else:
|
|
if billing_address.stripe_tax_id:
|
|
logger.debug("We have a tax id %s" % billing_address.stripe_tax_id)
|
|
tax_id_obj = stripe.Customer.retrieve_tax_id(
|
|
stripe_customer_id,
|
|
billing_address.stripe_tax_id,
|
|
)
|
|
if tax_id_obj.verification.status == "verified":
|
|
logger.debug("Latest status on Stripe=%s. Updating" %
|
|
tax_id_obj.verification.status)
|
|
# update billing address
|
|
billing_address.vat_number_validated_on = datetime.datetime.now()
|
|
billing_address.vat_number_verified = True
|
|
billing_address.save()
|
|
return {
|
|
"status": "verified",
|
|
"validated_on": billing_address.vat_number_validated_on
|
|
}
|
|
else:
|
|
billing_address.vat_number_validated_on = datetime.datetime.now()
|
|
billing_address.vat_number_verified = False
|
|
billing_address.save()
|
|
else:
|
|
logger.debug("Creating a tax id")
|
|
tax_id_obj = create_tax_id(
|
|
stripe_customer_id, billing_address_id,
|
|
"ch_vat" if billing_address.country.lower() == "ch" else "eu_vat",
|
|
)
|
|
else:
|
|
logger.debug("invalid billing address")
|
|
return {
|
|
"status": "invalid billing address",
|
|
"validated_on": ""
|
|
}
|
|
return {
|
|
"status": tax_id_obj.verification.status if 'verification' in tax_id_obj else "unknown",
|
|
"validated_on": datetime.datetime.now() if tax_id_obj.verification.status == "verified" else ""
|
|
}
|
|
|
|
def create_tax_id(stripe_customer_id, billing_address_id, type):
|
|
try:
|
|
billing_address = BillingAddress.objects.get(id=billing_address_id)
|
|
except BillingAddress.DoesNotExist as dne:
|
|
billing_address = None
|
|
logger.debug("BillingAddress does not exist for %s" % billing_address_id)
|
|
except BillingAddress.MultipleObjectsReturned as mor:
|
|
logger.debug("Multiple BillingAddress exist for %s" % billing_address_id)
|
|
billing_address = BillingAddress.objects.filter(billing_address_id).order_by('-id').first()
|
|
|
|
tax_id_obj = None
|
|
if billing_address:
|
|
try:
|
|
tax_id_obj = uncloud_stripe.get_or_create_tax_id_for_user(
|
|
stripe_customer_id,
|
|
vat_number=billing_address.vat_number,
|
|
type=type,
|
|
country=billing_address.country
|
|
)
|
|
billing_address.stripe_tax_id = tax_id_obj.id
|
|
billing_address.vat_number_verified = True if tax_id_obj.verification.status == "verified" else False
|
|
billing_address.save()
|
|
return tax_id_obj
|
|
except Exception as e:
|
|
logger.debug("Received none in tax_id_obj")
|
|
return {
|
|
'verification': None,
|
|
'error': str(e)
|
|
}
|
|
|
|
def apply_vat_discount(subtotal, pricing_plan, vat_rate=False, vat_validation_status=False):
|
|
vat_percent = vat_rate or pricing_plan.vat_percentage
|
|
if pricing_plan.vat_inclusive or (vat_validation_status and vat_validation_status in ["verified", "not_needed"]):
|
|
vat_percent = decimal.Decimal(0)
|
|
vat = decimal.Decimal(0)
|
|
else:
|
|
vat = subtotal * decimal.Decimal(vat_rate) * decimal.Decimal(0.01)
|
|
discount_amount = 0
|
|
discount_amount_with_vat = 0
|
|
if pricing_plan.discount_amount:
|
|
discount_amount = round(float(pricing_plan.discount_amount), 2)
|
|
discount_amount_with_vat = decimal.Decimal(discount_amount) * (1 + decimal.Decimal(vat_rate) * decimal.Decimal(0.01))
|
|
discount_amount_with_vat = discount_amount_with_vat
|
|
|
|
subtotal = round(float(subtotal), 2)
|
|
vat_percent = round(float(vat_percent), 2)
|
|
discount = {
|
|
'name': pricing_plan.discount_name,
|
|
'amount': discount_amount,
|
|
'amount_with_vat': round(float(discount_amount_with_vat), 2)
|
|
}
|
|
subtotal_after_discount = subtotal - discount["amount"]
|
|
price_after_discount_with_vat = round((subtotal - discount['amount']) * (1 + vat_percent * 0.01), 2)
|
|
|
|
return (subtotal, round(float(subtotal_after_discount), 2), price_after_discount_with_vat,
|
|
round(float(vat), 2), vat_percent, discount)
|
|
|
|
|
|
def get_order_total_with_vat(cores, memory, storage,
|
|
pricing_name='default', vat_rate=False, vat_validation_status=False):
|
|
try:
|
|
pricing = PricingPlan.objects.get(name=pricing_name)
|
|
except Exception as ex:
|
|
logger.error(
|
|
"Error getting PricingPlan object for {pricing_name}."
|
|
"Details: {details}".format(
|
|
pricing_name=pricing_name, details=str(ex)
|
|
)
|
|
)
|
|
return None
|
|
|
|
subtotal = (
|
|
pricing.set_up_fees +
|
|
(decimal.Decimal(cores) * pricing.cores_unit_price) +
|
|
(decimal.Decimal(memory) * pricing.ram_unit_price) +
|
|
(decimal.Decimal(storage) * (pricing.storage_unit_price))
|
|
)
|
|
return apply_vat_discount(subtotal, pricing, vat_rate, vat_validation_status)
|
|
|
|
|
|
|