chore: refactor invoice generation into a util fn
parent
9c239ad78b
commit
438e34f7d6
|
@ -12,18 +12,30 @@
|
|||
#
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
from enum import Enum, unique
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from payments import get_payment_model, RedirectNeeded, PaymentStatus
|
||||
from django.core.mail import send_mail
|
||||
from django.urls import reverse
|
||||
from django.template.loader import render_to_string
|
||||
from django.contrib.auth.models import User
|
||||
from django.conf import settings
|
||||
from django.shortcuts import get_object_or_404
|
||||
|
||||
from dash.models import Instance
|
||||
|
||||
|
||||
def payment_fullfilled(instance: Instance) -> bool:
|
||||
Payment = get_payment_model()
|
||||
|
||||
|
||||
def __get_delta():
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
delta = now - timedelta(seconds=(60 * 60 * 24 * 30)) # one month
|
||||
return delta
|
||||
|
||||
|
||||
def payment_fullfilled(instance: Instance) -> bool:
|
||||
delta = __get_delta()
|
||||
|
||||
payment = None
|
||||
for p in Payment.objects.filter(date__gt=(delta), instance_name=instance.name):
|
||||
|
@ -31,3 +43,70 @@ def payment_fullfilled(instance: Instance) -> bool:
|
|||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
@unique
|
||||
class GenerateInvoiceErrorCode(Enum):
|
||||
ALREADY_PAID = "already paid"
|
||||
DUPLICATE_PAYMENT = "DUPLICATE PAYMENT"
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.name
|
||||
|
||||
|
||||
class GenerateInvoiceException(Exception):
|
||||
error: str
|
||||
code: GenerateInvoiceErrorCode
|
||||
|
||||
def __init__(self, code: GenerateInvoiceErrorCode):
|
||||
self.error = str(code)
|
||||
self.code = code
|
||||
|
||||
def __str__(self):
|
||||
return self.error
|
||||
|
||||
|
||||
def generate_invoice(instance: Instance) -> Payment:
|
||||
delta = __get_delta()
|
||||
|
||||
payment = None
|
||||
for p in Payment.objects.filter(date__gt=(delta), instance_name=instance.name):
|
||||
if p.status == PaymentStatus.CONFIRMED:
|
||||
raise GenerateInvoiceException(code=GenerateInvoiceErrorCode.ALREADY_PAID)
|
||||
if any([p.status == PaymentStatus.INPUT, p.status == PaymentStatus.WAITING]):
|
||||
if payment is None:
|
||||
payment = p
|
||||
else:
|
||||
print(f"Duplicate payment {p}, deleting in favor of {payment}")
|
||||
p.delete()
|
||||
|
||||
if payment is None:
|
||||
print("Payment not found, generating new payment")
|
||||
payment = Payment.objects.create(
|
||||
variant="stripe", # this is the variant from PAYMENT_VARIANTS
|
||||
instance=instance,
|
||||
)
|
||||
|
||||
invoice_link = reverse("billing.invoice.details", args=(payment.public_ref,))
|
||||
ctx = {
|
||||
"username": instance.owned_by.username,
|
||||
"link": invoice_link,
|
||||
"payment": payment,
|
||||
}
|
||||
|
||||
body = render_to_string(
|
||||
"billing/emails/payment-notification.txt",
|
||||
context=ctx,
|
||||
)
|
||||
|
||||
email = instance.owned_by.email
|
||||
sender = settings.DEFAULT_FROM_EMAIL
|
||||
|
||||
send_mail(
|
||||
subject="[Hostea] An invoice is generated for your Hostea VM",
|
||||
message=body,
|
||||
from_email=f"No reply Hostea<{sender}>", # TODO read from settings.py
|
||||
recipient_list=[email],
|
||||
)
|
||||
|
||||
return payment
|
||||
|
|
|
@ -29,6 +29,12 @@ from dash.models import Instance
|
|||
from infrastructure.utils import create_vm_if_not_exists
|
||||
from dash.utils import footer_ctx
|
||||
|
||||
from .utils import (
|
||||
generate_invoice as generate_invoice_util,
|
||||
GenerateInvoiceErrorCode,
|
||||
GenerateInvoiceException,
|
||||
)
|
||||
|
||||
|
||||
def default_ctx(title: str, username: str):
|
||||
"""
|
||||
|
@ -69,51 +75,13 @@ def paid_invoices(request):
|
|||
def generate_invoice(request, instance_name: str):
|
||||
instance = get_object_or_404(Instance, name=instance_name, owned_by=request.user)
|
||||
|
||||
Payment = get_payment_model()
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
delta = now - timedelta(seconds=(60 * 60 * 24 * 30)) # one month
|
||||
|
||||
payment = None
|
||||
for p in Payment.objects.filter(date__gt=(delta), instance_name=instance_name):
|
||||
if p.status == PaymentStatus.CONFIRMED:
|
||||
try:
|
||||
payment = generate_invoice_util(instance=instance)
|
||||
return redirect(reverse("billing.invoice.details", args=(payment.public_ref,)))
|
||||
except GenerateInvoiceException as e:
|
||||
if e.code == GenerateInvoiceErrorCode.ALREADY_PAID:
|
||||
return HttpResponse("BAD REQUEST: Already paid", status=400)
|
||||
elif any([p.status == PaymentStatus.INPUT, p.status == PaymentStatus.WAITING]):
|
||||
if payment is None:
|
||||
payment = p
|
||||
else:
|
||||
print(f"Duplicate payment {p}, deleting in favor of {payment}")
|
||||
p.delete()
|
||||
|
||||
if payment is None:
|
||||
print("Payment not found, generating new payment")
|
||||
payment = Payment.objects.create(
|
||||
variant="stripe", # this is the variant from PAYMENT_VARIANTS
|
||||
instance=instance,
|
||||
)
|
||||
|
||||
invoice_link = reverse("billing.invoice.details", args=(payment.public_ref,))
|
||||
ctx = {
|
||||
"username": request.user.username,
|
||||
"link": invoice_link,
|
||||
"payment": payment,
|
||||
}
|
||||
|
||||
body = render_to_string(
|
||||
"billing/emails/payment-notification.txt",
|
||||
context=ctx,
|
||||
)
|
||||
|
||||
email = request.user.email
|
||||
sender = settings.DEFAULT_FROM_EMAIL
|
||||
|
||||
send_mail(
|
||||
subject="[Hostea] An invoice is generated for your Hostea VM",
|
||||
message=body,
|
||||
from_email=f"No reply Hostea<{sender}>", # TODO read from settings.py
|
||||
recipient_list=[email],
|
||||
)
|
||||
|
||||
return redirect(invoice_link)
|
||||
return redirect(reverse("dash.home"))
|
||||
|
||||
|
||||
@login_required
|
||||
|
|
Loading…
Reference in New Issue