ismail/haikal
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity
haikal/inventory/utils.py

2886 lines
102 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import os
import json
import secrets
import logging
import datetime
import requests
from PIL import Image
from io import BytesIO
from decimal import Decimal
from inventory import models
from django.urls import reverse
from django.conf import settings
from urllib.parse import urljoin
from django.utils import timezone
from django.db import transaction
from django_ledger.io import roles
from django.contrib import messages
from django.db import IntegrityError
from django.shortcuts import redirect
from django_q.tasks import async_task
from django.core.mail import send_mail
from plans.models import AbstractOrder
from django_ledger.models import (
EstimateModel,
InvoiceModel,
BillModel,
VendorModel,
AccountModel,
EntityModel,
ChartOfAccountModel,
)
from django.core.files.base import ContentFile
from django_ledger.models.items import ItemModel
from django.utils.translation import get_language
from django.core.exceptions import ObjectDoesNotExist
from django.utils.translation import gettext_lazy as _
from django_q.models import Schedule as DjangoQSchedule
from django.contrib.contenttypes.models import ContentType
from django_ledger.models.transactions import TransactionModel
from django_ledger.models.journal_entry import JournalEntryModel
# from .tasks import generate_car_image_task
logger = logging.getLogger(__name__)
def make_random_password(
length=10, allowed_chars="abcdefghjkmnpqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ23456789"
):
return "".join(secrets.choice(allowed_chars) for i in range(length))
def get_jwt_token():
"""
Fetches a JWT token from an external authentication API.
This function sends a POST request to the provided authentication API endpoint with
the required headers and payload. It retrieves a JSON Web Token (JWT) if the request
is successful or returns None in case of a failure.
:raises: This function does not propagate exceptions but catches and logs
``requests.exceptions.RequestException`` if the request fails.
:return: A JWT token as a string if the request is successful, or None if
an error occurs during the process.
:rtype: str or None
"""
url = "https://carapi.app/api/auth/login"
headers = {
"accept": "text/plain",
"Content-Type": "application/json",
}
data = {
"api_token": settings.CAR_API_TOKEN,
"api_secret": settings.CAR_API_SECRET,
}
logger.debug(f"Attempting to fetch JWT token from: {url}")
try:
response = requests.post(url, headers=headers, json=data)
response.raise_for_status()
# logging for success
logger.info("Successfully fetched JWT token.")
return response.text
except requests.exceptions.RequestException as e:
# logging for error
logger.error(f"HTTP error fetching JWT token from {url}: ", exc_info=True)
print(f"Error obtaining JWT token: {e}")
return None
def localize_some_words():
"""
Localize predefined words and phrases for display purposes.
This function is used to localize a set of predefined words or phrases
that may be displayed in a user interface or other contexts. These
localized words or phrases include terms like "success," "error," and
"Forgot Password?".
:return: None
"""
success = _("success")
error = _("error")
forget = _("Forgot Password?")
return None
def get_calculations(quotation):
"""
Calculates and summarizes financial services data related to the cars in a given
quotation. It aggregates costs, VAT, and total amounts based on the services
available for the cars linked to the quotation.
:param quotation: The quotation object containing the cars and their related
financial details.
:type quotation: Quotation
:return: A dictionary with computed financial details including the services
data, total service costs, total VAT computed, and the total including VAT.
:rtype: dict
"""
context = {}
qc_len = quotation.quotation_cars.count()
cars = [x.car for x in quotation.quotation_cars.all()]
finances = models.CarFinance.objects.filter(car__in=cars)
services = ItemModel.objects.filter(additional_finances__in=finances).all()
data = [
{
"name": x.name,
"price": x.default_amount,
"total_price": x.default_amount * qc_len,
"vated": float(x.default_amount) * 0.15 * float(qc_len),
"total_price_vat": float(x.default_amount)
+ (float(x.default_amount) * 0.15 * float(qc_len)),
}
for x in services
]
context["services"] = data
context["total_cost"] = 0
context["total_vat"] = 0
context["total_cost_vat"] = 0
for k in context["services"]:
context["total_cost"] += k["total_price"]
context["total_vat"] += k["vated"]
context["total_cost_vat"] = float(context["total_cost"]) + float(
context["total_vat"]
)
return context
def send_email(from_, to_, subject, message):
"""
Send an email with the specified subject and message.
This function sends an email from the given sender to the specified
recipient with a subject and a message body. It utilizes the provided
parameters to construct and send the email.
:param from_: str
The sender's email address.
:param to_: str
The recipient's email address.
:param subject: str
The subject of the email.
:param message: str
The body/content of the email.
:return: None
"""
subject = subject
message = message
from_email = from_
recipient_list = [to_]
async_task(send_mail, subject, message, from_email, recipient_list)
def get_user_type(request):
"""
Determine the type of user based on the given request object.
This function identifies the type of user from the provided request. It
checks if the user is a dealer, staff member, or neither based on the
attributes of the request object, returning the appropriate associated
user type or `None` if the determination cannot be made.
:param request: The HTTP request object containing user and role
information.
:type request: Any
:return: A dealer object if the user is a dealer, a dealer object
associated with the staff member if the user is a staff member, or
None if the user type cannot be identified.
:rtype: Optional[Any]
"""
if request.is_dealer:
return request.user.dealer
elif request.is_staff:
return request.user.staff.dealer
return None
def get_dealer_from_instance(instance):
"""
Retrieve the dealer object from the given instance.
This function checks whether the given instance's dealer has an attribute
`staff`. If the attribute exists, it directly returns the dealer object
associated with the instance.
:param instance: The instance from which the dealer object is retrieved.
:type instance: Any
:return: The dealer object associated with the instance.
:rtype: Any
"""
if instance.dealer.staff:
return instance.dealer
else:
return instance.dealer
def reserve_car(car, request):
"""
Reserve a car for a user for 24 hours and update its status to reserved.
The function creates a reservation record for the specified car, sets the
reservation expiration time, updates the car's status, and notifies the user
about the operation outcome through success or error messages. It then redirects
to the car detail page.
:param car: The car object to be reserved.
:param request: The HTTP request object containing the user making the reservation.
:return: Redirection to the car's detail page.
"""
try:
# reserved_until = timezone.now() + timezone.timedelta(hours=24)
reserved_until = timezone.now() + timezone.timedelta(minutes=1)
reservation = models.CarReservation.objects.create(
car=car, reserved_by=request.user, reserved_until=reserved_until
)
car.status = models.CarStatusChoices.RESERVED
car.save()
# --- Logging for Success ---
DjangoQSchedule.objects.create(
name=f"remove_reservation_for_car_with_vin_{car.vin}",
func="inventory.tasks.remove_reservation_by_id",
args=reservation.pk,
schedule_type=DjangoQSchedule.ONCE,
next_run=reserved_until,
)
logger.info(
f"Car {car.pk} ('{car.id_car_make} {car.id_car_model}') reserved successfully "
f"by user {request.user}. "
f"Reserved until: {reserved_until}."
)
messages.success(request, _("Car reserved successfully."))
except Exception as e:
# --- Logging for Error ---
logger.error(
f"Error reserving car {car.pk} ('{car.id_car_make} {car.id_car_model}') "
f"for user {request.user} . "
f"Error: {e}",
exc_info=True,
)
messages.error(request, f"Error reserving car: {e}")
return redirect("car_detail", dealer_slug=request.dealer.slug, slug=car.slug)
def calculate_vat_amount(amount):
"""
Calculates the VAT (Value Added Tax) amount for a given monetary value. The method retrieves
the first active VAT rate from the database and applies it to the provided amount. If no
active VAT rate is found, the function will return the original amount.
:param amount: The monetary value to which the VAT should be applied.
:type amount: Decimal
:return: A tuple containing the computed VAT amount and the VAT rate as a decimal, or
the original amount if no VAT rate is active.
:rtype: Tuple[Decimal, Decimal] or Decimal
"""
vat = models.VatRate.objects.filter(is_active=True).first()
if vat:
return ((amount * Decimal(vat.rate)).quantize(Decimal("0.01")), vat.rate)
return amount
def get_car_finance_data(model):
"""
Fetches car finance data from the provided model instance and calculates
various related details including total prices, VAT, discounts, and
additional services information.
:param model: The model instance that contains car and finance-related
transaction data.
:type model: Model
:return: A dictionary containing detailed information about the car finance
transactions, including individual car details, total quantity, calculated
totals for prices, VAT, discounts, additional services, and the VAT rate.
:rtype: dict
"""
vat = models.VatRate.objects.filter(is_active=True).first()
data = model.get_itemtxs_data()[0].all()
total = sum(
[
Decimal(item.item_model.additional_info["car_finance"]["marked_price"])
* Decimal(item.ce_quantity or item.quantity)
for item in data
]
)
additional_services = []
for i in data:
if i.item_model.additional_info["additional_services"]:
additional_services.extend(
[
{"name": x.get("name"), "price": x.get("price")}
for x in i.item_model.additional_info["additional_services"]
]
)
return {
"cars": [
{
"vin": x.item_model.additional_info["car_info"]["vin"],
"make": x.item_model.additional_info["car_info"]["make"],
"model": x.item_model.additional_info["car_info"]["model"],
"year": x.item_model.additional_info["car_info"]["year"],
"trim": x.item_model.additional_info["car_info"]["mileage"],
"cost_price": x.item_model.additional_info["car_finance"]["cost_price"],
"marked_price": x.item_model.additional_info["car_finance"][
"marked_price"
],
"discount": x.item_model.additional_info["car_finance"][
"discount_amount"
],
"quantity": x.ce_quantity or x.quantity,
"unit_price": Decimal(
x.item_model.additional_info["car_finance"]["total"]
),
"total": Decimal(x.item_model.additional_info["car_finance"]["total"])
* Decimal(x.quantity or x.ce_quantity),
"total_vat": x.item_model.additional_info["car_finance"]["total_vat"],
"additional_services": x.item_model.additional_info[
"additional_services"
],
}
for x in data
],
"quantity": sum((x.quantity or x.ce_quantity) for x in data),
"total_price": total,
"total_vat": (total * vat.rate) + total,
"total_discount": sum(
Decimal(x.item_model.additional_info["car_finance"]["discount_amount"])
for x in data
),
"grand_total": Decimal(total * vat.rate)
+ total
- Decimal(
sum(
Decimal(x.item_model.additional_info["car_finance"]["discount_amount"])
for x in data
)
),
"additionals": additional_services,
"vat": vat.rate,
}
def get_financial_values(model):
"""
Calculates financial values for a given model, including VAT, discounts,
grand total, and additional services.
This function processes financial data related to items in the input model
and computes aggregated values like total amounts, VAT amounts, discounts,
and more. It also provides detailed information on cars and items as well
as any applicable additional services.
:param model: The model instance containing financial and item data.
It should have the necessary methods and attributes to fetch
transactional records and relevant details.
:type model: Any
:return: A dictionary containing calculated financial details including:
- "vat_amount": The computed VAT amount.
- "total": The total value of all items before VAT and discounts.
- "grand_total": The total after applying discounts and adding VAT.
- "discount_amount": The aggregated discount amount.
- "vat": The applicable VAT rate.
- "car_and_item_info": A list of detailed car and item information.
- "additional_services": A list of additional services and their prices.
:rtype: dict
"""
vat = models.VatRate.objects.filter(is_active=True).first()
if not model.get_itemtxs_data()[0].exists():
return {
"vat_amount": 0,
"total": 0,
"grand_total": 0,
"discount_amount": 0,
"vat": 0,
"car_and_item_info": [],
"additional_services": [],
}
data = model.get_itemtxs_data()[0].all()
for item in data:
if not item.item_model.additional_info.get("car_finance"):
return {
"vat_amount": 0,
"total": 0,
"grand_total": 0,
"discount_amount": 0,
"vat": 0,
"car_and_item_info": [],
"additional_services": [],
}
if isinstance(model, InvoiceModel):
if model.ce_model:
data = model.ce_model.get_itemtxs_data()[0].all()
else:
data = model.get_itemtxs_data()[0].all()
total = sum(
[
Decimal(item.item_model.additional_info["car_finance"]["marked_price"])
* Decimal(item.ce_quantity or item.quantity)
for item in data
]
)
discount_amount = sum(
Decimal(i.item_model.additional_info["car_finance"]["discount_amount"])
for i in data
)
additional_services = []
for i in data:
if i.item_model.additional_info["additional_services"]:
additional_services.extend(
[
{"name": x["name"], "price": x["price"]}
for x in i.item_model.additional_info["additional_services"]
]
)
grand_total = Decimal(total) - Decimal(discount_amount)
vat_amount = round(Decimal(grand_total) * Decimal(vat.rate), 2)
car_and_item_info = [
{
"info": x.item_model.additional_info["car_info"],
"finances": x.item_model.additional_info["car_finance"],
"quantity": x.ce_quantity or x.quantity,
"total": Decimal(
x.item_model.additional_info["car_finance"]["marked_price"]
)
* Decimal(x.ce_quantity or x.quantity),
}
for x in data
]
return {
"total": total,
"discount_amount": discount_amount,
"car_and_item_info": car_and_item_info,
"additional_services": additional_services,
"grand_total": grand_total + vat_amount,
"vat_amount": vat_amount,
"vat": vat.rate,
}
# def set_invoice_payment(dealer, entity, invoice, amount, payment_method):
# """
# Processes and applies a payment for a specified invoice. This function calculates
# finance details, handles associated account transactions, and updates the invoice
# status accordingly.
# :param dealer: Dealer object responsible for processing the payment
# :type dealer: Dealer
# :param entity: Entity object associated with the invoice and payment
# :type entity: Entity
# :param invoice: The invoice object for which the payment is being made
# :type invoice: Invoice
# :param amount: The amount being paid towards the invoice
# :type amount: Decimal
# :param payment_method: The payment method used for the transaction
# :type payment_method: str
# :return: None
# """
# calculator = CarFinanceCalculator(invoice)
# finance_data = calculator.get_finance_data()
# handle_account_process(invoice, amount, finance_data)
# if invoice.can_migrate():
# invoice.migrate_state(
# user_model=dealer.user,
# entity_slug=entity.slug
# )
# invoice.make_payment(amount)
# invoice.save()
def set_bill_payment(dealer, entity, bill, amount, payment_method):
"""
Sets the payment for a given bill by creating journal entries for the
transaction and updating the respective accounts and the bill's status.
The function handles the transaction by creating a new journal entry
linked with the specified bill, then records debit and credit
transactions using the entity’s cash and accounts payable accounts.
It finally updates the bill's payment status and persists changes.
:param dealer: The dealer making or receiving the payment.
:type dealer: Any
:param entity: The business entity involved in the payment transaction.
:type entity: Any
:param bill: The bill object representing the invoice to be paid.
:type bill: Any
:param amount: The amount to be paid for the bill.
:type amount: Decimal
:param payment_method: The method used to make the payment (e.g., cash, credit).
:type payment_method: Any
:return: None
"""
total_amount = 0
for x in bill.get_itemtxs_data()[0].all():
total_amount += Decimal(x.unit_cost) * Decimal(x.quantity)
journal = JournalEntryModel.objects.create(
posted=False,
description=f"Payment for bill {bill.bill_number}",
ledger=bill.ledger,
locked=False,
origin="Payment",
)
cash_account = entity.get_default_coa_accounts().get(name="Cash", active=True)
account_payable = entity.get_default_coa_accounts().get(
name="Accounts Payable", active=True
)
TransactionModel.objects.create(
journal_entry=journal,
account=cash_account, # Debit Cash
amount=amount, # Payment amount
tx_type="debit",
description="Bill Payment Received",
)
TransactionModel.objects.create(
journal_entry=journal,
account=account_payable, # Credit Accounts Receivable
amount=amount, # Payment amount
tx_type="credit",
description="Bill Payment Received",
)
bill.make_payment(amount)
bill.save()
def transfer_to_dealer(request, cars, to_dealer, remarks=None):
"""
Transfers a list of cars from one dealer to another, ensuring that all cars
originate from the same dealer and logging the transfer.
:param request: The HTTP request object which contains information about
the current user and context.
:type request: HttpRequest
:param cars: A list of car objects to transfer.
:type cars: list[Car]
:param to_dealer: The dealer object representing the destination dealer.
:type to_dealer: Dealer
:param remarks: Optional remarks to include in the transfer log.
Defaults to None.
:type remarks: str, optional
:return: None
:rtype: NoneType
:raises ValueError: If no cars are selected for transfer.
:raises ValueError: If the cars are not all from the same dealer.
:raises ValueError: If the source and destination dealers are the same.
"""
dealer = get_user_type(request)
if not cars:
raise ValueError("No cars selected for transfer.")
from_dealer = cars[0].dealer # Assume all cars are from the same dealer
# Validate that all cars are from the same dealer
for car in cars:
if car.dealer != from_dealer:
raise ValueError("All cars must be from the same dealer.")
if from_dealer == to_dealer:
raise ValueError("Cannot transfer cars to the same dealer.")
# Log the transfer
transfer_log = models.CarTransferLog.objects.create(
from_dealer=from_dealer,
to_dealer=to_dealer,
remarks=remarks,
)
transfer_log.cars.set(cars) # Associate the cars with the transfer log
# Update the dealer for all cars
for car in cars:
car.dealer = to_dealer
car.save()
class CarTransfer:
"""
Handles the process of transferring a car between dealers, automating associated tasks
such as creating customers, invoices, products, vendors, and bills, as well as reflecting
changes in relevant entities and systems.
The purpose of this class is to facilitate a smooth transfer of car ownership from one
dealer to another. It ensures that all necessary financial, logistic, and informational
processes are executed and updated within the system. The car's status and related
entities are adjusted accordingly post-transfer.
:ivar car: Car object being transferred.
:ivar transfer: Represents the transfer details, including source and destination dealers.
:ivar from_dealer: The dealer transferring the car out.
:ivar to_dealer: The dealer receiving the car.
:ivar customer: Customer entity corresponding to the receiving dealer.
:ivar invoice: Invoice entity created for the car transfer.
:ivar ledger: Ledger associated with the created invoice.
:ivar item: The inventory item corresponding to the car being transferred.
:ivar product: The product created in the receiving dealer's ledger.
:ivar vendor: Vendor entity related to the transferring dealer.
:ivar bill: Bill entity created for the receiving dealer.
"""
def __init__(self, car, transfer):
self.car = car
self.transfer = transfer
self.from_dealer = transfer.from_dealer
self.to_dealer = transfer.to_dealer
self.customer = None
self.invoice = None
self.ledger = None
self.item = None
self.product = None
self.vendor = None
self.bill = None
def transfer_car(self):
self._create_customer()
self._create_invoice()
self._create_product_in_receiver_ledger()
self._create_vendor_and_bill()
self._finalize_car_transfer()
return True
def _create_customer(self):
self.customer = self._find_or_create_customer()
if self.customer is None:
self.customer = self._create_new_customer()
def _find_or_create_customer(self):
return (
self.from_dealer.entity.get_customers()
.filter(email=self.to_dealer.user.email)
.first()
)
def _create_new_customer(self):
customer = self.from_dealer.entity.create_customer(
customer_model_kwargs={
"customer_name": self.to_dealer.name,
"email": self.to_dealer.user.email,
"address_1": self.to_dealer.address,
}
)
customer.additional_info = {}
customer.additional_info.update({"type": "organization"})
customer.save()
return customer
def _create_invoice(self):
self.invoice = self.from_dealer.entity.create_invoice(
customer_model=self.customer,
terms=InvoiceModel.TERMS_NET_30,
cash_account=self.from_dealer.entity.get_default_coa_accounts().get(
name="Cash", active=True
),
prepaid_account=self.from_dealer.entity.get_default_coa_accounts().get(
name="Accounts Receivable", active=True
),
coa_model=self.from_dealer.entity.get_default_coa(),
)
self.ledger = self.from_dealer.entity.create_ledger(name=str(self.invoice.pk))
self.invoice.ledgar = self.ledger
self.ledger.invoicemodel = self.invoice
self.ledger.save()
self.invoice.save()
self._add_car_item_to_invoice()
def _add_car_item_to_invoice(self):
self.item = (
self.from_dealer.entity.get_items_products()
.filter(name=self.car.vin)
.first()
)
if not self.item:
return
invoice_itemtxs = {
self.item.item_number: {
"unit_cost": self.transfer.total_price,
"quantity": self.transfer.quantity,
"total_amount": self.transfer.total_price,
}
}
invoice_itemtxs = self.invoice.migrate_itemtxs(
itemtxs=invoice_itemtxs,
commit=True,
operation=InvoiceModel.ITEMIZE_APPEND,
)
if self.invoice.can_review():
self.invoice.mark_as_review()
self.invoice.mark_as_approved(
self.from_dealer.entity.slug, self.from_dealer.entity.admin
)
self.invoice.save()
def _create_product_in_receiver_ledger(self):
uom = (
self.to_dealer.entity.get_uom_all().filter(name=self.item.uom.name).first()
)
self.product = self.to_dealer.entity.create_item_product(
name=self.item.name,
uom_model=uom,
item_type=self.item.item_type,
coa_model=self.to_dealer.entity.get_default_coa(),
)
self.product.additional_info = {}
self.product.additional_info.update({"car_info": self.car.to_dict()})
self.product.save()
def _create_vendor_and_bill(self):
self.vendor = self._find_or_create_vendor()
self.bill = self.to_dealer.entity.create_bill(
vendor_model=self.vendor,
terms=BillModel.TERMS_NET_30,
cash_account=self.to_dealer.entity.get_default_coa_accounts().get(
name="Cash", active=True
),
prepaid_account=self.to_dealer.entity.get_default_coa_accounts().get(
name="Prepaid Expenses", active=True
),
coa_model=self.to_dealer.entity.get_default_coa(),
)
self._add_car_item_to_bill()
def _find_or_create_vendor(self):
vendor = (
self.to_dealer.entity.get_vendors()
.filter(vendor_name=self.from_dealer.name)
.first()
)
if not vendor:
vendor = VendorModel.objects.create(
entity_model=self.to_dealer.entity,
vendor_name=self.from_dealer.name,
additional_info={"info": to_dict(self.from_dealer)},
)
return vendor
def _add_car_item_to_bill(self):
bill_itemtxs = {
self.product.item_number: {
"unit_cost": self.transfer.total_price,
"quantity": self.transfer.quantity,
"total_amount": self.transfer.total_price,
}
}
bill_itemtxs = self.bill.migrate_itemtxs(
itemtxs=bill_itemtxs, commit=True, operation=BillModel.ITEMIZE_APPEND
)
self.bill.additional_info = {}
self.bill.additional_info.update({"car_info": self.car.to_dict()})
self.bill.mark_as_review()
self.bill.mark_as_approved(
self.to_dealer.entity.slug, self.to_dealer.entity.admin
)
self.bill.save()
def _finalize_car_transfer(self):
self.car.dealer = self.to_dealer
self.car.vendor = self.vendor
self.car.receiving_date = datetime.datetime.now()
self.car.additional_services.clear()
if hasattr(self.car, "custom_cards"):
self.car.custom_cards.delete()
self.car.cost_price = self.transfer.total_price
self.car.marked_price = 0
self.car.discount_amount = 0
self.car.save()
self.car.location.owner = self.to_dealer
self.car.location.showroom = self.to_dealer
self.car.location.description = ""
self.car.location.save()
self.car.status = models.CarStatusChoices.AVAILABLE
self.transfer.status = models.CarTransferStatusChoices.success
self.transfer.active = False
self.transfer.save()
self.car.save()
# def transfer_car(car, transfer):
# from_dealer = transfer.from_dealer
# to_dealer = transfer.to_dealer
# # add transfer.to_dealer as customer in transfer.from_dealer entity
# customer = (
# from_dealer.entity.get_customers().filter(email=to_dealer.user.email).first()
# )
# if not customer:
# customer = from_dealer.entity.create_customer(
# customer_model_kwargs={
# "customer_name": to_dealer.name,
# "email": to_dealer.user.email,
# "address_1": to_dealer.address,
# }
# )
# customer.additional_info.update({"type": "organization"})
# customer.save()
# invoice = from_dealer.entity.create_invoice(
# customer_model=customer,
# terms=InvoiceModel.TERMS_NET_30,
# cash_account=from_dealer.entity.get_default_coa_accounts().get(
# name="Cash", active=True
# ),
# prepaid_account=from_dealer.entity.get_default_coa_accounts().get(
# name="Accounts Receivable", active=True
# ),
# coa_model=from_dealer.entity.get_default_coa(),
# )
# ledger = from_dealer.entity.create_ledger(name=str(invoice.pk))
# invoice.ledgar = ledger
# ledger.invoicemodel = invoice
# ledger.save()
# invoice.save()
# item = from_dealer.entity.get_items_products().filter(name=car.vin).first()
# if not item:
# return
# invoice_itemtxs = {
# item.item_number: {
# "unit_cost": transfer.total_price,
# "quantity": transfer.quantity,
# "total_amount": transfer.total_price,
# }
# }
# invoice_itemtxs = invoice.migrate_itemtxs(
# itemtxs=invoice_itemtxs,
# commit=True,
# operation=InvoiceModel.ITEMIZE_APPEND,
# )
# invoice.save()
# invoice.mark_as_review()
# invoice.mark_as_approved(from_dealer.entity.slug, from_dealer.entity.admin)
# # invoice.mark_as_paid(from_dealer.entity.slug, from_dealer.entity.admin)
# invoice.save()
# # create car item product in to_dealer entity
# uom = to_dealer.entity.get_uom_all().filter(name=item.uom.name).first()
# # create item product in the reciever ledger
# product = to_dealer.entity.create_item_product(
# name=item.name,
# uom_model=uom,
# item_type=item.item_type,
# coa_model=to_dealer.entity.get_default_coa(),
# )
# product.additional_info.update({"car_info": car.to_dict()})
# product.save()
# # add the sender as vendor and create a bill for it
# vendor = None
# vendor = to_dealer.entity.get_vendors().filter(vendor_name=from_dealer.name).first()
# if not vendor:
# vendor = VendorModel.objects.create(
# entity_model=to_dealer.entity,
# vendor_name=from_dealer.name,
# additional_info={"info": to_dict(from_dealer)},
# )
# # transfer the car to to_dealer and create items record
# bill = to_dealer.entity.create_bill(
# vendor_model=vendor,
# terms=BillModel.TERMS_NET_30,
# cash_account=to_dealer.entity.get_default_coa_accounts().get(
# name="Cash", active=True
# ),
# prepaid_account=to_dealer.entity.get_default_coa_accounts().get(
# name="Prepaid Expenses", active=True
# ),
# coa_model=to_dealer.entity.get_default_coa(),
# )
# bill.additional_info = {}
# bill_itemtxs = {
# product.item_number: {
# "unit_cost": transfer.total_price,
# "quantity": transfer.quantity,
# "total_amount": transfer.total_price,
# }
# }
# bill_itemtxs = bill.migrate_itemtxs(
# itemtxs=bill_itemtxs, commit=True, operation=BillModel.ITEMIZE_APPEND
# )
# bill.additional_info.update({"car_info": car.to_dict()})
# bill.additional_info.update({"car_finance": car.to_dict()})
# bill.mark_as_review()
# bill.mark_as_approved(to_dealer.entity.slug, to_dealer.entity.admin)
# bill.save()
# car.dealer = to_dealer
# car.vendor = vendor
# car.receiving_date = datetime.datetime.now()
# car.additional_services.clear()
# if hasattr(car, "custom_cards"):
# car.custom_cards.delete()
# car.cost_price = transfer.total_price
# car.selling_price = 0
# car.discount_amount = 0
# car.save()
# car.location.owner = to_dealer
# car.location.showroom = to_dealer
# car.location.description = ""
# car.location.save()
# car.status = models.CarStatusChoices.AVAILABLE
# transfer.status = models.CarTransferStatusChoices.success
# transfer.active = False
# transfer.save()
# car.save()
# return True
def to_dict(obj):
"""
Convert a Python object's attributes and associated values to a dictionary,
with special handling for specific data types such as datetime and object references.
:param obj: The Python object to be converted into a dictionary. Object attributes
are dynamically introspected to create the dictionary. For objects containing
references to other objects, specific attributes (e.g., "pk" or "id") may be used
for conversion and representation.
:return: A dictionary containing the key-value pairs for the object's attributes.
Datetime values are formatted as strings in "YYYY-MM-DD HH:MM:SS" format. All other
attributes, including nested object references, are converted to strings.
"""
obj_dict = vars(obj).copy()
if "_state" in vars(obj):
del obj_dict["_state"]
for key, value in obj_dict.items():
if isinstance(value, datetime.datetime):
obj_dict[key] = value.strftime("%Y-%m-%d %H:%M:%S")
elif hasattr(value, "pk") or hasattr(value, "id"):
try:
obj_dict[key] = value.name
except AttributeError:
obj_dict[key] = str(value)
else:
obj_dict[key] = str(value)
return obj_dict
class CarFinanceCalculator1:
"""
Class responsible for calculating car financing details.
This class provides methods and attributes required for calculating various
aspects related to car financing, such as VAT calculation, pricing, discounts,
and additional services. It processes data about cars, computes totals (e.g.,
price, VAT, discounts), and aggregates the financial data for reporting or
further processing.
:ivar model: The data model passed to the calculator for retrieving transaction data.
:type model: Any
:ivar vat_rate: The current active VAT rate retrieved from the database.
:type vat_rate: Decimal
:ivar item_transactions: A collection of item transactions retrieved from the model.
:type item_transactions: list
:ivar additional_services: A list of additional services with details (e.g., name, price, taxable status).
:type additional_services: list
"""
VAT_OBJ_NAME = "vat_rate"
CAR_FINANCE_KEY = "car_finance"
CAR_INFO_KEY = "car_info"
ADDITIONAL_SERVICES_KEY = "additional_services"
def __init__(self, model):
if isinstance(model, InvoiceModel):
self.dealer = models.Dealer.objects.get(entity=model.ce_model.entity)
self.extra_info = models.ExtraInfo.objects.get(
dealer=self.dealer,
content_type=ContentType.objects.get_for_model(model.ce_model),
object_id=model.ce_model.pk,
)
elif isinstance(model, EstimateModel):
self.dealer = models.Dealer.objects.get(entity=model.entity)
self.extra_info = models.ExtraInfo.objects.get(
dealer=self.dealer,
content_type=ContentType.objects.get_for_model(model),
object_id=model.pk,
)
self.model = model
self.vat_rate = self._get_vat_rate()
self.item_transactions = self._get_item_transactions()
# self.additional_services = self._get_additional_services()
def _get_vat_rate(self):
vat = models.VatRate.objects.filter(dealer=self.dealer, is_active=True).first()
if not vat:
raise ObjectDoesNotExist("No active VAT rate found")
return vat.rate
def _get_additional_services(self):
return [
x
for item in self.item_transactions
for x in item.item_model.car.additional_services
]
def _get_item_transactions(self):
return self.model.get_itemtxs_data()[0].all()
def get_items(self):
return self._get_item_transactions()
@staticmethod
def _get_quantity(item):
return item.ce_quantity or item.quantity
# def _get_nested_value(self, item, *keys):
# current = item.item_model.additional_info
# for key in keys:
# current = current.get(key, {})
# return current
def _get_car_data(self, item):
quantity = self._get_quantity(item)
car = item.item_model.car
unit_price = Decimal(car.marked_price)
discount = self.extra_info.data.get("discount", 0)
sell_price = unit_price - Decimal(discount)
return {
"item_number": item.item_model.item_number,
"vin": car.vin, # car_info.get("vin"),
"make": car.id_car_make, # car_info.get("make"),
"model": car.id_car_model, # car_info.get("model"),
"year": car.year, # car_info.get("year"),
"logo": car.logo, # getattr(car.id_car_make, "logo", ""),
"trim": car.id_car_trim, # car_info.get("trim"),
"mileage": car.mileage, # car_info.get("mileage"),
"cost_price": car.cost_price,
"selling_price": car.selling_price,
"marked_price": car.marked_price,
"discount": car.discount_amount,
"quantity": quantity,
"unit_price": unit_price,
"sell_price": sell_price,
"total": unit_price,
"total_vat": sell_price * self.vat_rate,
"total_discount": discount,
"final_price": sell_price + (sell_price * self.vat_rate),
"total_additionals": car.total_additional_services,
"grand_total": sell_price
+ (sell_price * self.vat_rate)
+ car.total_additional_services,
"additional_services": car.additional_services, # self._get_nested_value(
# item, self.ADDITIONAL_SERVICES_KEY
# ),
}
def calculate_totals(self):
total_price = sum(
Decimal(item.item_model.car.marked_price) for item in self.item_transactions
)
total_additionals = sum(
Decimal(item.price_) for item in self._get_additional_services()
)
total_discount = self.extra_info.data.get("discount", 0)
total_price_discounted = total_price
if total_discount:
total_price_discounted = total_price - Decimal(total_discount)
print(total_price_discounted)
total_vat_amount = total_price_discounted * self.vat_rate
return {
"total_price_discounted": total_price_discounted,
"total_price_before_discount": total_price,
"total_price": total_price_discounted,
"total_vat_amount": total_vat_amount,
"total_discount": Decimal(total_discount),
"total_additionals": total_additionals,
"grand_total": total_price_discounted
+ total_vat_amount
+ total_additionals,
}
def get_finance_data(self):
totals = self.calculate_totals()
return {
"cars": [self._get_car_data(item) for item in self.item_transactions],
"quantity": sum(
self._get_quantity(item) for item in self.item_transactions
),
"total_price": round(totals["total_price"], 2),
"total_price_discounted": round(totals["total_price_discounted"], 2),
"total_price_before_discount": round(
totals["total_price_before_discount"], 2
),
"total_vat": round(totals["total_vat_amount"] + totals["total_price"], 2),
"total_vat_amount": round(totals["total_vat_amount"], 2),
"total_discount": round(totals["total_discount"], 2),
"total_additionals": round(totals["total_additionals"], 2),
"grand_total": round(totals["grand_total"], 2),
"additionals": self._get_additional_services(),
"vat": round(self.vat_rate, 2),
}
class CarFinanceCalculator:
"""
Class responsible for calculating car financing details.
This class provides methods and attributes required for calculating various
aspects related to car financing, such as VAT calculation, pricing, discounts,
and additional services. It processes data about cars, computes totals (e.g.,
price, VAT, discounts), and aggregates the financial data for reporting or
further processing.
:ivar model: The data model passed to the calculator for retrieving transaction data.
:type model: Any
:ivar vat_rate: The current active VAT rate retrieved from the database.
:type vat_rate: Decimal
:ivar item_transactions: A collection of item transactions retrieved from the model.
:type item_transactions: list
:ivar additional_services: A list of additional services with details (e.g., name, price, taxable status).
:type additional_services: list
"""
VAT_OBJ_NAME = "vat_rate"
CAR_FINANCE_KEY = "car_finance"
CAR_INFO_KEY = "car_info"
ADDITIONAL_SERVICES_KEY = "additional_services"
def __init__(self, model):
if isinstance(model, InvoiceModel):
self.dealer = models.Dealer.objects.get(entity=model.ce_model.entity)
self.extra_info = models.ExtraInfo.objects.get(
dealer=self.dealer,
content_type=ContentType.objects.get_for_model(model.ce_model),
object_id=model.ce_model.pk,
)
elif isinstance(model, EstimateModel):
self.dealer = models.Dealer.objects.get(entity=model.entity)
self.extra_info = models.ExtraInfo.objects.get(
dealer=self.dealer,
content_type=ContentType.objects.get_for_model(model),
object_id=model.pk,
)
self.model = model
self.vat_rate = self._get_vat_rate()
self.item_transactions = self._get_item_transactions()
# self.additional_services = self._get_additional_services()
def _get_vat_rate(self):
vat = models.VatRate.objects.filter(dealer=self.dealer, is_active=True).first()
if not vat:
raise ObjectDoesNotExist("No active VAT rate found")
return vat.rate
def _get_additional_services(self):
return [
x
for item in self.item_transactions
for x in item.item_model.car.additional_services
]
def _get_item_transactions(self):
return self.model.get_itemtxs_data()[0].all()
def get_items(self):
return self._get_item_transactions()
@staticmethod
def _get_quantity(item):
return item.ce_quantity or item.quantity
# def _get_nested_value(self, item, *keys):
# current = item.item_model.additional_info
# for key in keys:
# current = current.get(key, {})
# return current
def _get_car_data(self, item):
quantity = self._get_quantity(item)
car = item.item_model.car
unit_price = Decimal(car.marked_price)
discount = self.extra_info.data.get("discount", 0)
sell_price = unit_price - Decimal(discount)
return {
"item_number": item.item_model.item_number,
"vin": car.vin, # car_info.get("vin"),
"make": car.id_car_make, # car_info.get("make"),
"model": car.id_car_model, # car_info.get("model"),
"year": car.year, # car_info.get("year"),
"logo": car.logo, # getattr(car.id_car_make, "logo", ""),
"trim": car.id_car_trim, # car_info.get("trim"),
"mileage": car.mileage, # car_info.get("mileage"),
"cost_price": car.cost_price,
"selling_price": car.selling_price,
"marked_price": car.marked_price,
"discount": car.discount_amount,
"quantity": quantity,
"unit_price": unit_price,
"sell_price": sell_price,
"total": unit_price,
"total_vat": sell_price * self.vat_rate,
"total_discount": discount,
"final_price": sell_price + (sell_price * self.vat_rate),
"total_additionals": car.total_additional_services,
"grand_total": sell_price
+ (sell_price * self.vat_rate)
+ car.total_additional_services,
"additional_services": car.additional_services, # self._get_nested_value(
# item, self.ADDITIONAL_SERVICES_KEY
# ),
}
def calculate_totals(self):
total_price = sum(
Decimal(item.item_model.car.marked_price) for item in self.item_transactions
)
total_additionals = sum(
Decimal(item.price_) for item in self._get_additional_services()
)
total_discount = self.extra_info.data.get("discount", 0)
total_price_discounted = total_price
if total_discount:
total_price_discounted = total_price - Decimal(total_discount)
print(total_price_discounted)
total_vat_amount = total_price_discounted * self.vat_rate
return {
"total_price_discounted": total_price_discounted,
"total_price_before_discount": total_price,
"total_price": total_price_discounted,
"total_vat_amount": total_vat_amount,
"total_discount": Decimal(total_discount),
"total_additionals": total_additionals,
"grand_total": total_price_discounted
+ total_vat_amount
+ total_additionals,
}
def get_finance_data(self):
totals = self.calculate_totals()
return {
"car": [self._get_car_data(item) for item in self.item_transactions],
"quantity": sum(
self._get_quantity(item) for item in self.item_transactions
),
"total_price": round(totals["total_price"], 2),
"total_price_discounted": round(totals["total_price_discounted"], 2),
"total_price_before_discount": round(
totals["total_price_before_discount"], 2
),
"total_vat": round(totals["total_vat_amount"] + totals["total_price"], 2),
"total_vat_amount": round(totals["total_vat_amount"], 2),
"total_discount": round(totals["total_discount"], 2),
"total_additionals": round(totals["total_additionals"], 2),
"grand_total": round(totals["grand_total"], 2),
"additionals": self._get_additional_services(),
"vat": round(self.vat_rate, 2),
}
def get_finance_data(estimate, dealer):
vat = models.VatRate.objects.filter(dealer=dealer, is_active=True).first()
item = estimate.get_itemtxs_data()[0].first()
car = item.item_model.car
if isinstance(estimate, InvoiceModel) and hasattr(estimate, "ce_model"):
estimate = estimate.ce_model
extra_info = models.ExtraInfo.objects.get(
dealer=dealer,
content_type=ContentType.objects.get_for_model(EstimateModel),
object_id=estimate.pk,
)
discount = extra_info.data.get("discount", 0)
discount = Decimal(discount)
additional_services = car.get_additional_services()
discounted_price = Decimal(car.marked_price) - discount
vat_amount = discounted_price * vat.rate
total_services_amount = additional_services.get("total")
total_services_vat = sum([x[1] for x in additional_services.get("services")])
total_services_amount_ = additional_services.get("total_")
total_vat = vat_amount + total_services_vat
return {
"car": car,
"discounted_price": discounted_price or 0,
"price_before_discount": car.marked_price,
"vat_amount": vat_amount,
"vat_rate": vat.rate,
"discount_amount": discount,
"additional_services": additional_services,
"final_price": discounted_price + vat_amount,
"total_services_vat": total_services_vat,
"total_services_amount": total_services_amount,
"total_services_amount_": total_services_amount_,
"total_vat": total_vat,
"grand_total": discounted_price + total_vat + additional_services.get("total"),
}
# totals = self.calculate_totals()
# return {
# "car": [self._get_car_data(item) for item in self.item_transactions],
# "quantity": sum(
# self._get_quantity(item) for item in self.item_transactions
# ),
# "total_price": round(totals["total_price"], 2),
# "total_price_discounted": round(totals["total_price_discounted"], 2),
# "total_price_before_discount": round(totals["total_price_before_discount"], 2),
# "total_vat": round(totals["total_vat_amount"] + totals["total_price"], 2),
# "total_vat_amount": round(totals["total_vat_amount"], 2),
# "total_discount": round(totals["total_discount"], 2),
# "total_additionals": round(totals["total_additionals"], 2),
# "grand_total": round(totals["grand_total"], 2),
# "additionals": self._get_additional_services(),
# "vat": round(self.vat_rate, 2),
# }
# class CarFinanceCalculator:
# """
# Class responsible for calculating car financing details.
# This class provides methods and attributes required for calculating various
# aspects related to car financing, such as VAT calculation, pricing, discounts,
# and additional services. It processes data about cars, computes totals (e.g.,
# price, VAT, discounts), and aggregates the financial data for reporting or
# further processing.
# :ivar model: The data model passed to the calculator for retrieving transaction data.
# :type model: Any
# :ivar vat_rate: The current active VAT rate retrieved from the database.
# :type vat_rate: Decimal
# :ivar item_transactions: A collection of item transactions retrieved from the model.
# :type item_transactions: list
# :ivar additional_services: A list of additional services with details (e.g., name, price, taxable status).
# :type additional_services: list
# """
# VAT_OBJ_NAME = "vat_rate"
# CAR_FINANCE_KEY = "car_finance"
# CAR_INFO_KEY = "car_info"
# ADDITIONAL_SERVICES_KEY = "additional_services"
# def __init__(self, model):
# if isinstance(model, InvoiceModel):
# self.dealer = models.Dealer.objects.get(entity=model.ce_model.entity)
# self.extra_info = models.ExtraInfo.objects.get(
# dealer=self.dealer,
# content_type=ContentType.objects.get_for_model(model.ce_model),
# object_id=model.ce_model.pk,
# )
# elif isinstance(model, EstimateModel):
# self.dealer = models.Dealer.objects.get(entity=model.entity)
# self.extra_info = models.ExtraInfo.objects.get(
# dealer=self.dealer,
# content_type=ContentType.objects.get_for_model(model),
# object_id=model.pk,
# )
# self.model = model
# self.vat_rate = self._get_vat_rate()
# self.item_transactions = self._get_item_transactions()
# self.additional_services = self._get_additional_services()
# def _get_vat_rate(self):
# vat = models.VatRate.objects.filter(dealer=self.dealer, is_active=True).first()
# if not vat:
# raise ObjectDoesNotExist("No active VAT rate found")
# return vat.rate
# def _get_item_transactions(self):
# return self.model.get_itemtxs_data()[0].all()
# @staticmethod
# def _get_quantity(item):
# return item.ce_quantity or item.quantity
# def _get_nested_value(self, item, *keys):
# current = item.item_model.additional_info
# for key in keys:
# current = current.get(key, {})
# return current
# def _get_car_data(self, item):
# quantity = self._get_quantity(item)
# car_finance = self._get_nested_value(item, self.CAR_FINANCE_KEY)
# car_info = self._get_nested_value(item, self.CAR_INFO_KEY)
# unit_price = Decimal(car_finance.get("marked_price", 0))
# return {
# "item_number": item.item_model.item_number,
# "vin": car_info.get("vin"),
# "make": car_info.get("make"),
# "model": car_info.get("model"),
# "year": car_info.get("year"),
# "logo": getattr(item.item_model.car.id_car_make, "logo", ""),
# "trim": car_info.get("trim"),
# "mileage": car_info.get("mileage"),
# "cost_price": car_finance.get("cost_price"),
# "selling_price": car_finance.get("selling_price"),
# "marked_price": car_finance.get("marked_price"),
# "discount": car_finance.get("discount_amount"),
# "quantity": quantity,
# "unit_price": unit_price,
# "total": unit_price * Decimal(quantity),
# "total_vat": car_finance.get("total_vat"),
# "additional_services": self._get_nested_value(
# item, self.ADDITIONAL_SERVICES_KEY
# ),
# }
# def _get_additional_services(self):
# return [
# {
# "name": service.get("name"),
# "price": service.get("price"),
# "taxable": service.get("taxable"),
# "price_": service.get("price_"),
# }
# for item in self.item_transactions
# for service in self._get_nested_value(item, self.ADDITIONAL_SERVICES_KEY)
# or []
# ]
# def calculate_totals(self):
# total_price = sum(
# Decimal(self._get_nested_value(item, self.CAR_FINANCE_KEY, "marked_price"))
# * int(self._get_quantity(item))
# for item in self.item_transactions
# )
# total_additionals = sum(
# Decimal(x.get("price_")) for x in self._get_additional_services()
# )
# total_discount = self.extra_info.data.get("discount", 0)
# # total_discount = sum(
# # Decimal(
# # self._get_nested_value(item, self.CAR_FINANCE_KEY, "discount_amount")
# # )
# # for item in self.item_transactions
# # )
# total_price_discounted = total_price
# if total_discount:
# total_price_discounted = total_price - Decimal(total_discount)
# total_vat_amount = total_price_discounted * self.vat_rate
# return {
# "total_price_before_discount": round(
# total_price, 2
# ), # total_price_before_discount,
# "total_price": round(total_price_discounted, 2), # total_price_discounted,
# "total_vat_amount": round(total_vat_amount, 2), # total_vat_amount,
# "total_discount": round(Decimal(total_discount)),
# "total_additionals": round(total_additionals, 2), # total_additionals,
# "grand_total": round(
# total_price_discounted + total_vat_amount + total_additionals, 2
# ),
# }
# def get_finance_data(self):
# totals = self.calculate_totals()
# return {
# "cars": [self._get_car_data(item) for item in self.item_transactions],
# "quantity": sum(
# self._get_quantity(item) for item in self.item_transactions
# ),
# "total_price": totals["total_price"],
# "total_price_before_discount": totals["total_price_before_discount"],
# "total_vat": totals["total_vat_amount"] + totals["total_price"],
# "total_vat_amount": totals["total_vat_amount"],
# "total_discount": totals["total_discount"],
# "total_additionals": totals["total_additionals"],
# "grand_total": totals["grand_total"],
# "additionals": self.additional_services,
# "vat": self.vat_rate,
# }
def get_item_transactions(txs):
"""
Extracts and compiles relevant transaction details from a list of transactions,
including information about cars, financing, estimates, and invoices. The extracted
details for each transaction are stored as dictionaries in a list.
:param txs: A list of transaction objects from which information is extracted.
:type txs: list
:return: A list of dictionaries, each containing extracted transaction details.
:rtype: list
"""
transactions = []
for tx in txs:
data = {}
if tx.item_model.additional_info.get("car_info"):
data["info"] = tx.item_model.additional_info.get("car_info")
if tx.item_model.additional_info.get("car_finance"):
data["finance"] = tx.item_model.additional_info.get("car_finance")
if tx.has_estimate():
data["estimate"] = tx.ce_model
data["has_estimate"] = True
data["customer"] = tx.ce_model.customer
if tx.has_invoice():
data["invoice"] = tx.invoice_model
data["has_invoice"] = True
data["customer"] = tx.invoice_model.customer
if bool(data):
transactions.append(data)
return transactions
def get_local_name(self):
"""
Retrieve the localized name of the instance based on the current language.
This method returns the name attribute of an instance based on the currently
active language. If the language is Arabic ('ar'), it attempts to return the
value of the attribute `arabic_name`. If the language is not Arabic or the
`arabic_name` attribute is not defined, it defaults to returning the value of
the `name` attribute.
:param self:
Reference to the instance of the class containing `name` and
optionally `arabic_name` attributes.
:return:
The localized name based on the current language. Returns the value of
`arabic_name` for Arabic ('ar') language, or the value of `name` for any
other language or if `arabic_name` is not defined.
"""
if get_language() == "ar":
return getattr(self, "arabic_name", None)
return getattr(self, "name", None)
@transaction.atomic
def set_invoice_payment(dealer, entity, invoice, amount, payment_method):
"""
Records the customer payment (`make_payment`) and posts the full
accounting (sales + VAT + COGS + Inventory).
"""
invoice.make_payment(amount)
invoice.save()
_post_sale_and_cogs(invoice, dealer)
def _post_sale_and_cogs(invoice, dealer):
"""
For every car line on the invoice:
1) Cash / A-R / VAT / Revenue journal
2) COGS / Inventory journal
"""
entity: EntityModel = invoice.ledger.entity
# calc = CarFinanceCalculator(invoice)
data = get_finance_data(invoice, dealer)
car = data.get("car")
coa: ChartOfAccountModel = entity.get_default_coa()
cash_acc = invoice.cash_account or dealer.settings.invoice_cash_account
vat_acc = (
dealer.settings.invoice_tax_payable_account
or entity.get_default_coa_accounts()
.filter(role_default=True, role=roles.LIABILITY_CL_TAXES_PAYABLE)
.first()
)
car_rev = (
dealer.settings.invoice_vehicle_sale_account
or entity.get_default_coa_accounts()
.filter(role_default=True, role=roles.INCOME_OPERATIONAL)
.first()
)
add_rev = dealer.settings.invoice_additional_services_account
if not add_rev:
try:
add_rev = (
entity.get_default_coa_accounts()
.filter(name="After-Sales Services", active=True)
.first()
)
if not add_rev:
add_rev = coa.create_account(
code="4020",
name="After-Sales Services",
role=roles.INCOME_OPERATIONAL,
balance_type=roles.CREDIT,
active=True,
)
add_rev.role_default = False
add_rev.save(update_fields=["role_default"])
dealer.settings.invoice_additional_services_account = add_rev
dealer.settings.save()
except Exception as e:
logger.error(f"error find or create additional services account {e}")
if car.get_additional_services_amount > 0 and not add_rev:
raise Exception(
"additional services exist but not account found,please create account for the additional services and set as default in the settings"
)
cogs_acc = (
dealer.settings.invoice_cost_of_good_sold_account
or entity.get_default_coa_accounts()
.filter(role_default=True, role=roles.COGS)
.first()
)
inv_acc = (
dealer.settings.invoice_inventory_account
or entity.get_default_coa_accounts()
.filter(role_default=True, role=roles.ASSET_CA_INVENTORY)
.first()
)
net_car_price = Decimal(data["discounted_price"])
net_additionals_price = Decimal(data["additional_services"]["total"])
vat_amount = Decimal(data["vat_amount"])
grand_total = net_car_price + car.get_additional_services_amount_ + vat_amount
cost_total = Decimal(car.cost_price)
discount_amount = Decimal(data["discount_amount"])
# ------------------------------------------------------------------
# 2A. Journal: Cash / A-R / VAT / Sales
# ------------------------------------------------------------------
je_sale = JournalEntryModel.objects.create(
ledger=invoice.ledger,
description=f"Sale {car.vin}",
origin=f"Invoice {invoice.invoice_number}",
locked=False,
posted=False,
)
# Dr Cash (what the customer paid)
TransactionModel.objects.create(
journal_entry=je_sale,
account=cash_acc,
amount=grand_total,
tx_type="debit",
description="Debit to Cash on Hand",
)
# Cr Sales – Car
TransactionModel.objects.create(
journal_entry=je_sale,
account=car_rev,
amount=net_car_price,
tx_type="credit",
description=" Credit to Car Sales",
)
# Cr VAT Payable
TransactionModel.objects.create(
journal_entry=je_sale,
account=vat_acc,
amount=vat_amount + car.get_additional_services_vat,
tx_type="credit",
description="Credit to Tax Payable",
)
# # Cr A/R (clear the receivable)
# TransactionModel.objects.create(
# journal_entry=je_sale,
# account=ar_acc,
# amount=grand_total,
# tx_type='credit'
# )
if car.get_additional_services_amount > 0:
# Cr Sales – Additional Services
if not add_rev:
logger.warning(
f"Additional Services account not set for dealer {dealer}. Skipping additional services revenue entry."
)
else:
TransactionModel.objects.create(
journal_entry=je_sale,
account=add_rev,
amount=car.get_additional_services_amount,
tx_type="credit",
description="Credit to After-Sales Services",
)
# TransactionModel.objects.create(
# journal_entry=je_sale,
# account=vat_acc,
# amount=car.get_additional_services_vat,
# tx_type="credit",
# description="Credit to Tax Payable (Additional Services)",
# )
# ------------------------------------------------------------------
# 2B. Journal: COGS / Inventory reduction
# ------------------------------------------------------------------
je_cogs = JournalEntryModel.objects.create(
ledger=invoice.ledger,
description=f"COGS {car.vin}",
origin=f"Invoice {invoice.invoice_number}",
locked=False,
posted=False,
)
# Dr COGS
TransactionModel.objects.create(
journal_entry=je_cogs,
account=cogs_acc,
amount=cost_total,
tx_type="debit",
)
# Cr Inventory
TransactionModel.objects.create(
journal_entry=je_cogs, account=inv_acc, amount=cost_total, tx_type="credit"
)
# ------------------------------------------------------------------
# 2C. Update car state flags inside the same transaction
# ------------------------------------------------------------------
entity.get_items_inventory().filter(name=car.vin).update(for_inventory=False)
# car.item_model.for_inventory = False
# car.item_model.save(update_fields=['for_inventory'])
car.discount_amount = discount_amount
car.selling_price = grand_total
# car.is_sold = True
car.save()
# def handle_account_process(invoice, amount, finance_data):
# """
# Processes accounting transactions based on an invoice, financial data,
# and related entity accounts configuration. This function handles the
# creation of accounts if they do not already exist, and processes journal
# entries and transactions.
# :param invoice: The invoice object to process transactions for.
# :type invoice: InvoiceModel
# :param amount: Total monetary value for the transaction.
# :type amount: Decimal
# :param finance_data: Dictionary containing financial details such as
# 'grand_total', 'total_vat_amount', and other related data.
# :type finance_data: dict
# :return: None
# """
# for i in invoice.get_itemtxs_data()[0]:
# # car = models.Car.objects.get(vin=invoice.get_itemtxs_data()[0].first().item_model.name)
# car = i.item_model.car
# entity = invoice.ledger.entity
# coa = entity.get_default_coa()
# cash_account = (
# entity.get_all_accounts()
# .filter(role_default=True, role=roles.ASSET_CA_CASH)
# .first()
# )
# inventory_account = car.get_inventory_account()
# revenue_account = car.get_revenue_account()
# cogs_account = car.get_cogs_account()
# journal = JournalEntryModel.objects.create(
# posted=False,
# description=f"Payment for Invoice {invoice.invoice_number}",
# ledger=invoice.ledger,
# locked=False,
# origin=f"Sale of {car.id_car_make.name}{car.vin}: Invoice {invoice.invoice_number}",
# )
# TransactionModel.objects.create(
# journal_entry=journal,
# account=cash_account,
# amount=Decimal(finance_data.get("grand_total")),
# tx_type="debit",
# description="",
# )
# TransactionModel.objects.create(
# journal_entry=journal,
# account=revenue_account,
# amount=Decimal(finance_data.get("grand_total")),
# tx_type="credit",
# description="",
# )
# journal_cogs = JournalEntryModel.objects.create(
# posted=False,
# description=f"COGS of {car.id_car_make.name}{car.vin}: Invoice {invoice.invoice_number}",
# ledger=invoice.ledger,
# locked=False,
# origin="Payment",
# )
# TransactionModel.objects.create(
# journal_entry=journal_cogs,
# account=cogs_account,
# amount=Decimal(car.finances.cost_price),
# tx_type="debit",
# description="",
# )
# TransactionModel.objects.create(
# journal_entry=journal_cogs,
# account=inventory_account,
# amount=Decimal(car.finances.cost_price),
# tx_type="credit",
# description="",
# )
# try:
# car.item_model.for_inventory = False
# logger.debug(f"Set item_model.for_inventory to False for car {car.vin}.")
# except Exception as e:
# logger.error(
# f"Error updating item_model.for_inventory for car {car.vin} (Invoice {invoice.invoice_number}): {e}",
# exc_info=True,
# )
# car.finances.is_sold = True
# car.finances.save()
# car.item_model.save()
# TransactionModel.objects.create(
# journal_entry=journal,
# account=additional_services_account, # Debit Additional Services
# amount=Decimal(car.finances.total_additionals),
# tx_type="debit",
# description="Additional Services",
# )
# TransactionModel.objects.create(
# journal_entry=journal,
# account=inventory_account, # Credit Inventory account
# amount=Decimal(finance_data.get("grand_total")),
# tx_type="credit",
# description="Account Adjustment",
# )
# TransactionModel.objects.create(
# journal_entry=journal,
# account=vat_payable_account, # Credit VAT Payable
# amount=finance_data.get("total_vat_amount"),
# tx_type="credit",
# description="VAT Payable on Invoice",
# )
def create_make_accounts(dealer):
"""
Creates accounts for dealer's car makes and associates them with a default
Chart of Accounts (CoA) if they don't already exist. The function iterates
through all car makes associated with the dealer and creates unique inventory
accounts for each car make. Accounts are created with a specified naming
convention and are assigned a new account code if required.
:param dealer: The dealer for whom to create make accounts.
:type dealer: Dealer
:return: None
:rtype: None
"""
entity = dealer.entity
coa = entity.get_default_coa()
# Create a unique account name for the dealer and car make combination
makes = models.DealersMake.objects.filter(dealer=dealer).all()
for make in makes:
account_name = f"{make.car_make.name} Inventory Account"
account = (
entity.get_all_accounts().filter(coa_model=coa, name=account_name).first()
)
if not account:
last_account = (
entity.get_all_accounts()
.filter(role=roles.ASSET_CA_INVENTORY)
.order_by("-created")
.first()
)
if len(last_account.code) == 4:
code = f"{int(last_account.code)}{1:03d}"
elif len(last_account.code) > 4:
code = f"{int(last_account.code) + 1}"
account = entity.create_account(
name=account_name,
code=code,
role=roles.ASSET_CA_INVENTORY,
coa_model=coa,
balance_type="credit",
active=True,
)
def handle_payment(request, dealer):
url = "https://api.moyasar.com/v1/payments"
callback_url = request.build_absolute_uri(
reverse("payment_callback", kwargs={"dealer_slug": dealer.slug})
)
email = request.POST.get("email")
first_name = request.POST.get("first_name")
last_name = request.POST.get("last_name")
phone = request.POST.get("phone")
card_name = request.POST.get("card_name")
card_number = str(request.POST.get("card_number", "")).replace(" ", "").strip()
expiry = request.POST.get("card_expiry", "").split("/")
month = int(expiry[0].strip()) if len(expiry) > 0 else 0
year = int(expiry[1].strip()) if len(expiry) > 1 else 0
cvv = request.POST.get("card_cvv")
user_data = {
"email": email,
"first_name": first_name,
"last_name": last_name,
"phone": phone,
}
# Get selected plan from session
selected_plan_id = request.session.get("pending_plan_id")
if not selected_plan_id:
raise ValueError("No pending plan found in session")
from plans.models import PlanPricing
pp = PlanPricing.objects.get(pk=selected_plan_id)
# Calculate amount without creating order
amount_sar = pp.price # assuming price is in SAR
tax_amount = amount_sar * 15 / 100
total = int((amount_sar + tax_amount) * 100) # convert to halalas
# Pass plan & dealer info via metadata
metadata = {
**user_data,
"plan_pricing_id": selected_plan_id,
"dealer_slug": dealer.slug,
}
payload = json.dumps(
{
"amount": total,
"currency": "SAR",
"description": f"Payment for plan {pp.plan.name}",
"callback_url": callback_url,
"source": {
"type": "creditcard",
"name": card_name,
"number": card_number,
"month": month,
"year": year,
"cvc": cvv,
"statement_descriptor": "Century Store",
"3ds": True,
"manual": False,
"save_card": False,
},
"metadata": metadata,
}
)
headers = {"Content-Type": "application/json", "Accept": "application/json"}
auth = (settings.MOYASAR_SECRET_KEY, "")
response = requests.post(url, auth=auth, headers=headers, data=payload)
if response.status_code != 201:
logger.error(f"Payment initiation failed: {response.text}")
return None, response.get("message")
data = response.json()
# Create PaymentHistory WITHOUT linking to order (since order doesn't exist yet)
amount_decimal = Decimal("{0:.2f}".format(Decimal(total) / Decimal(100)))
models.PaymentHistory.objects.create(
user=request.user,
user_data=json.dumps(metadata),
amount=amount_decimal,
currency=data["currency"],
status=data["status"],
transaction_id=data["id"],
payment_date=data["created_at"],
gateway_response=data,
)
logger.info(f"Payment initiated: {data}")
return data["source"]["transaction_url"], None
# def handle_payment(request, order):
# logger.info(f"Handling payment for order {order}")
# url = "https://api.moyasar.com/v1/payments"
# callback_url = request.build_absolute_uri(
# reverse("payment_callback", kwargs={"dealer_slug": request.dealer.slug})
# )
# logger.info(f"Got callback_url: {callback_url}")
# if request.user.is_authenticated:
# email = request.POST["email"]
# first_name = request.POST["first_name"]
# last_name = request.POST["last_name"]
# phone = request.POST["phone"]
# logger.info(f"Got user data: {email}, {first_name}, {last_name}, {phone}")
# card_name = request.POST["card_name"]
# card_number = str(request.POST["card_number"]).replace(" ", "").strip()
# month = int(request.POST["card_expiry"].split("/")[0].strip())
# year = int(request.POST["card_expiry"].split("/")[1].strip())
# cvv = request.POST["card_cvv"]
# logger.info(f"Got card data: {card_name}, {card_number[:4]}****, {month}, {year}, {cvv}")
# user_data = {
# "email": email,
# "first_name": first_name,
# "last_name": last_name,
# "phone": phone,
# }
# try:
# total = int((order.total() + order.tax * order.total() / 100) * 100)
# except (AttributeError, TypeError):
# raise ValueError("Order total or tax is invalid")
# logger.info(f"Calculated total: {total}")
# payload = json.dumps(
# {
# "amount": total,
# "currency": "SAR",
# "description": f"payment issued for {email}",
# "callback_url": callback_url,
# "source": {
# "type": "creditcard",
# "name": card_name,
# "number": card_number,
# "month": month,
# "year": year,
# "cvc": cvv,
# "statement_descriptor": "Century Store",
# "3ds": False,
# "manual": False,
# "save_card": False,
# },
# "metadata": user_data,
# }
# )
# logger.info(f"Generated payload: {payload}")
# headers = {"Content-Type": "application/json", "Accept": "application/json"}
# auth = (settings.MOYASAR_SECRET_KEY, "")
# response = requests.request("POST", url, auth=auth, headers=headers, data=payload)
# logger.info(f"Sent request to Moyasar API: {response.status_code}")
# if response.status_code == 400:
# data = response.json()
# if data["type"] == "validation_error":
# errors = data.get("errors", {})
# if "source.year" in errors:
# raise Exception("Invalid expiry year")
# else:
# raise Exception("Validation Error: ", errors)
# else:
# logger.error(f"Failed to process payment: {data}")
# #
# data = response.json()
# # order.status = AbstractOrder.STATUS.NEW
# order.save()
# #
# amount = Decimal("{0:.2f}".format(Decimal(total) / Decimal(100)))
# models.PaymentHistory.objects.create(
# user=request.user,
# user_data=user_data,
# amount=amount,
# currency=data["currency"],
# status=data["status"],
# transaction_id=data["id"],
# payment_date=data["created_at"],
# gateway_response=data,
# )
# transaction_url = data["source"]["transaction_url"]
# logger.info(f"Created payment history and got transaction_url: {transaction_url}")
# return transaction_url
# def get_user_quota(user):
# return user.dealer.quota
def get_accounts_data():
return [
# Current Assets (must start with 1)
{
"code": "1010",
"name": "Cash on Hand",
"role": roles.ASSET_CA_CASH,
"balance_type": roles.DEBIT,
"locked": False,
"default": True, # Default for ASSET_CA_CASH
},
{
"code": "1020",
"name": "Bank",
"role": roles.ASSET_CA_CASH,
"balance_type": roles.DEBIT,
"locked": False,
"default": False,
},
{
"code": "1030",
"name": "Accounts Receivable",
"role": roles.ASSET_CA_RECEIVABLES,
"balance_type": roles.DEBIT,
"locked": False,
"default": True, # Default for ASSET_CA_RECEIVABLES
},
{
"code": "1040",
"name": "Inventory (Cars)",
"role": roles.ASSET_CA_INVENTORY,
"balance_type": roles.DEBIT,
"locked": False,
"default": True, # Default for ASSET_CA_INVENTORY
},
{
"code": "1045",
"name": "Spare Parts Inventory",
"role": roles.ASSET_CA_INVENTORY,
"balance_type": roles.DEBIT,
"locked": False,
"default": False,
},
{
"code": "1050",
"name": "Employee Advances",
"role": roles.ASSET_CA_RECEIVABLES,
"balance_type": roles.DEBIT,
"locked": False,
"default": False,
},
{
"code": "1060",
"name": "Prepaid Expenses",
"role": roles.ASSET_CA_PREPAID,
"balance_type": roles.DEBIT,
"locked": False,
"default": True, # Default for ASSET_CA_PREPAID
},
{
"code": "1070",
"name": "Notes Receivable",
"role": roles.ASSET_LTI_NOTES_RECEIVABLE,
"balance_type": roles.DEBIT,
"locked": False,
"default": True, # Default for ASSET_LTI_NOTES_RECEIVABLE
},
# Fixed Assets (must also start with 1)
{
"code": "1110",
"name": "Lands",
"role": roles.ASSET_LTI_LAND,
"balance_type": roles.DEBIT,
"locked": False,
"default": True, # Default for ASSET_LTI_LAND
},
{
"code": "1111",
"name": "Buildings",
"role": roles.ASSET_PPE_BUILDINGS,
"balance_type": roles.DEBIT,
"locked": False,
"default": True, # Default for ASSET_PPE_BUILDINGS
},
{
"code": "1112",
"name": "Company Vehicles",
"role": roles.ASSET_PPE_EQUIPMENT,
"balance_type": roles.DEBIT,
"locked": False,
"default": True, # Default for ASSET_PPE_EQUIPMENT
},
{
"code": "1113",
"name": "Equipment & Tools",
"role": roles.ASSET_PPE_EQUIPMENT,
"balance_type": roles.DEBIT,
"locked": False,
"default": False,
},
{
"code": "1114",
"name": "Furniture & Fixtures",
"role": roles.ASSET_PPE_EQUIPMENT,
"balance_type": roles.DEBIT,
"locked": False,
"default": False,
},
{
"code": "1115",
"name": "Other Fixed Assets",
"role": roles.ASSET_PPE_EQUIPMENT,
"balance_type": roles.DEBIT,
"locked": False,
"default": False,
},
{
"code": "1120",
"name": "Long-term Investments",
"role": roles.ASSET_LTI_SECURITIES,
"balance_type": roles.DEBIT,
"locked": False,
"default": True, # Default for ASSET_LTI_SECURITIES
},
{
"code": "1130",
"name": "Intangible Assets",
"role": roles.ASSET_INTANGIBLE_ASSETS,
"balance_type": roles.DEBIT,
"locked": False,
"default": True, # Default for ASSET_INTANGIBLE_ASSETS
},
# Current Liabilities (must start with 2)
{
"code": "2010",
"name": "Accounts Payable",
"role": roles.LIABILITY_CL_ACC_PAYABLE,
"balance_type": roles.CREDIT,
"locked": True,
"default": True, # Default for LIABILITY_CL_ACC_PAYABLE
},
{
"code": "2020",
"name": "Notes Payable",
"role": roles.LIABILITY_CL_ST_NOTES_PAYABLE,
"balance_type": roles.CREDIT,
"locked": False,
"default": True, # Default for LIABILITY_CL_ST_NOTES_PAYABLE
},
{
"code": "2030",
"name": "Short-term Loans",
"role": roles.LIABILITY_CL_ST_NOTES_PAYABLE,
"balance_type": roles.CREDIT,
"locked": False,
"default": False,
},
{
"code": "2040",
"name": "Employee Payables",
"role": roles.LIABILITY_CL_WAGES_PAYABLE,
"balance_type": roles.CREDIT,
"locked": False,
"default": True, # Default for LIABILITY_CL_WAGES_PAYABLE
},
{
"code": "2050",
"name": "Accrued Expenses",
"role": roles.LIABILITY_CL_OTHER,
"balance_type": roles.CREDIT,
"locked": False,
"default": True, # Default for LIABILITY_CL_OTHER
},
{
"code": "2060",
"name": "Accrued Taxes",
"role": roles.LIABILITY_CL_TAXES_PAYABLE,
"balance_type": roles.CREDIT,
"locked": False,
"default": False, # Default for LIABILITY_CL_TAXES_PAYABLE
},
{
"code": "2070",
"name": "Provisions",
"role": roles.LIABILITY_CL_OTHER,
"balance_type": roles.CREDIT,
"locked": False,
"default": False,
},
# Long-term Liabilities (must also start with 2)
{
"code": "2103",
"name": "Deferred Revenue",
"role": roles.LIABILITY_CL_DEFERRED_REVENUE,
"balance_type": roles.CREDIT,
"locked": False,
"default": True, # Default for LIABILITY_CL_DEFERRED_REVENUE
},
{
"code": "2200",
"name": "Tax Payable",
"role": roles.LIABILITY_CL_TAXES_PAYABLE,
"balance_type": roles.CREDIT,
"locked": False,
"default": True,
},
{
"code": "2210",
"name": "Long-term Bank Loans",
"role": roles.LIABILITY_LTL_NOTES_PAYABLE,
"balance_type": roles.CREDIT,
"locked": False,
"default": True, # Default for LIABILITY_LTL_NOTES_PAYABLE
},
{
"code": "2220",
"name": "Lease Liabilities",
"role": roles.LIABILITY_LTL_NOTES_PAYABLE,
"balance_type": roles.CREDIT,
"locked": False,
"default": False,
},
{
"code": "2230",
"name": "Other Long-term Liabilities",
"role": roles.LIABILITY_LTL_NOTES_PAYABLE,
"balance_type": roles.CREDIT,
"locked": False,
"default": False,
},
# Equity (must start with 3)
{
"code": "3010",
"name": "Capital",
"role": roles.EQUITY_CAPITAL,
"balance_type": roles.CREDIT,
"locked": True,
"default": True, # Default for EQUITY_CAPITAL
},
{
"code": "3020",
"name": "Statutory Reserve",
"role": roles.EQUITY_ADJUSTMENT,
"balance_type": roles.CREDIT,
"locked": False,
"default": True, # Default for EQUITY_ADJUSTMENT
},
{
"code": "3030",
"name": "Retained Earnings",
"role": roles.EQUITY_ADJUSTMENT,
"balance_type": roles.CREDIT,
"locked": False,
"default": False,
},
{
"code": "3040",
"name": "Profit & Loss for the Period",
"role": roles.EQUITY_ADJUSTMENT,
"balance_type": roles.CREDIT,
"locked": False,
"default": False,
},
# Revenue (must start with 4)
{
"code": "4010",
"name": "Car Sales",
"role": roles.INCOME_OPERATIONAL,
"balance_type": roles.CREDIT,
"locked": True,
"default": True, # Default for INCOME_OPERATIONAL
},
{
"code": "4020",
"name": "After-Sales Services",
"role": roles.INCOME_OPERATIONAL,
"balance_type": roles.CREDIT,
"locked": False,
"default": False,
},
{
"code": "4030",
"name": "Car Rental Income",
"role": roles.INCOME_PASSIVE,
"balance_type": roles.CREDIT,
"locked": False,
"default": True, # Default for INCOME_PASSIVE
},
{
"code": "4040",
"name": "Other Income",
"role": roles.INCOME_OTHER,
"balance_type": roles.CREDIT,
"locked": False,
"default": True, # Default for INCOME_OTHER
},
# Expenses (must start with 5 for COGS, 6 for others)
{
"code": "5010",
"name": "Cost of Goods Sold",
"role": roles.COGS,
"balance_type": roles.DEBIT,
"locked": True,
"default": True, # Default for COGS
},
{
"code": "5015",
"name": "Spare Parts Cost Consumed",
"role": roles.COGS,
"balance_type": roles.DEBIT,
"locked": False,
"default": False,
},
{
"code": "6010",
"name": "Salaries & Wages",
"role": roles.EXPENSE_OPERATIONAL,
"balance_type": roles.DEBIT,
"locked": False,
"default": True, # Default for EXPENSE_OPERATIONAL
},
{
"code": "6020",
"name": "Rent",
"role": roles.EXPENSE_OPERATIONAL,
"balance_type": roles.DEBIT,
"locked": False,
"default": False,
},
{
"code": "6030",
"name": "Utilities",
"role": roles.EXPENSE_OPERATIONAL,
"balance_type": roles.DEBIT,
"locked": False,
"default": False,
},
{
"code": "6040",
"name": "Advertising & Marketing",
"role": roles.EXPENSE_OPERATIONAL,
"balance_type": roles.DEBIT,
"locked": False,
"default": False,
},
{
"code": "6050",
"name": "Maintenance",
"role": roles.EXPENSE_OPERATIONAL,
"balance_type": roles.DEBIT,
"locked": False,
"default": False,
},
{
"code": "6060",
"name": "Operating Expenses",
"role": roles.EXPENSE_OPERATIONAL,
"balance_type": roles.DEBIT,
"locked": False,
"default": False,
},
{
"code": "6070",
"name": "Depreciation",
"role": roles.EXPENSE_DEPRECIATION,
"balance_type": roles.DEBIT,
"locked": False,
"default": True, # Default for EXPENSE_DEPRECIATION
},
{
"code": "6080",
"name": "Fees & Taxes",
"role": roles.EXPENSE_OPERATIONAL,
"balance_type": roles.DEBIT,
"locked": False,
"default": False,
},
{
"code": "6090",
"name": "Bank Charges",
"role": roles.EXPENSE_OPERATIONAL,
"balance_type": roles.DEBIT,
"locked": False,
"default": False,
},
{
"code": "6100",
"name": "Other Expenses",
"role": roles.EXPENSE_OTHER,
"balance_type": roles.DEBIT,
"locked": False,
"default": True, # Default for EXPENSE_OTHER
},
]
def create_account(entity, coa, account_data):
logger.info(f"Creating account: {account_data['code']}")
logger.info(f"COA: {coa}")
try:
# Skip if exists
if coa.get_coa_accounts().filter(code=account_data["code"]).exists():
logger.info(f"Account already exists: {account_data['code']}")
return True
logger.info(f"Account does not exist: {account_data['code']},creating...")
account = coa.create_account(
code=account_data["code"],
name=account_data["name"],
role=account_data["role"],
balance_type=account_data["balance_type"],
active=True,
)
logger.info(f"Created account: {account}")
if account:
account.role_default = account_data.get("default", False)
account.save(update_fields=["role_default"])
return True
except IntegrityError:
return True # Already created by race condition
except Exception as e:
logger.error(f"❌ Error creating {account_data['code']}: {e}")
return False
# def create_account(entity, coa, account_data):
# try:
# account = entity.create_account(
# coa_model=coa,
# code=account_data["code"],
# name=account_data["name"],
# role=account_data["role"],
# balance_type=_(account_data["balance_type"]),
# active=True,
# )
# logger.info(f"Created account: {account}")
# account.role_default = account_data["default"]
# account.save()
# logger.info(f"Created default account: {account}")
# except Exception as e:
# logger.error(f"Error creating default account: {account_data['code']}, {e}")
def get_or_generate_car_image(car):
"""
Utility function to get or generate car image asynchronously
"""
try:
car_image, created = models.CarImage.objects.get_or_create(car=car)
if created:
car_image.image_hash = car.get_hash
car_image.save()
if car_image.image:
return car_image.image.url
# Check for existing image with same hash
existing = (
models.CarImage.objects.filter(
image_hash=car_image.image_hash, image__isnull=False
)
.exclude(car=car)
.first()
)
if existing:
car_image.image.save(existing.image.name, existing.image.file, save=True)
return car_image.image.url
# If no image exists and not already generating, schedule generation
if not car_image.is_generating:
car_image.schedule_image_generation()
return None # Return None while image is being generated
except Exception as e:
logger.error(f"Error getting/generating car image: {e}")
return None
def force_regenerate_car_image(car):
"""
Force regeneration of car image (useful for admin actions)
"""
try:
car_image, created = models.CarImage.objects.get_or_create(car=car)
car_image.image_hash = car.get_hash
car_image.image.delete(save=False) # Remove old image
car_image.is_generating = False
car_image.generation_attempts = 0
car_image.last_generation_error = None
car_image.save()
car_image.schedule_image_generation()
return True
except Exception as e:
logger.error(f"Error forcing image regeneration: {e}")
return False
class CarImageAPIClient:
"""Simple client to handle authenticated requests to the car image API"""
BASE_URL = settings.TENHAL_IMAGE_GENERATOR_URL
USERNAME = settings.TENHAL_IMAGE_GENERATOR_USERNAME
PASSWORD = settings.TENHAL_IMAGE_GENERATOR_PASSWORD
def __init__(self):
self.session = None
self.csrf_token = None
def login(self):
"""Login to the API and maintain session"""
try:
# Start fresh session
self.session = requests.Session()
# Step 1: Get CSRF token
response = self.session.get(f"{self.BASE_URL}/login")
response.raise_for_status()
# Get CSRF token from cookies
self.csrf_token = self.session.cookies.get("csrftoken")
if not self.csrf_token:
raise Exception("CSRF token not found in cookies")
# Step 2: Login
login_data = {
"username": self.USERNAME,
"password": self.PASSWORD,
"csrfmiddlewaretoken": self.csrf_token,
}
login_response = self.session.post(
f"{self.BASE_URL}/login", data=login_data
)
if login_response.status_code != 200:
raise Exception(
f"Login failed with status {login_response.status_code}"
)
logger.info("Successfully logged in to car image API")
return True
except Exception as e:
logger.error(f"Login failed: {e}")
self.session = None
self.csrf_token = None
return False
def generate_image(self, payload):
"""Generate car image using authenticated session"""
if not self.session or not self.csrf_token:
if not self.login():
raise Exception("Cannot generate image: Login failed")
try:
headers = {
"X-CSRFToken": self.csrf_token,
"Referer": self.BASE_URL,
}
logger.info(f"Generating image with payload: {payload}")
generate_data = {
"year": payload["year"],
"make": payload["make"],
"model": payload["model"],
"exterior_color": payload["color"],
"angle": "front three-quarter",
"reference_image": "",
}
response = self.session.post(
f"{self.BASE_URL}/generate",
json=generate_data,
headers=headers,
timeout=160,
)
response.raise_for_status()
# Parse response
result = response.json()
image_url = result.get("url")
logger.info(f"Generated image URL: {image_url}")
if not image_url:
raise Exception("No image URL in response")
# Download the actual image
image_response = self.session.get(
f"{self.BASE_URL}{image_url}", timeout=160
)
image_response.raise_for_status()
return image_response.content, None
except requests.RequestException as e:
error_msg = f"API request failed: {e}"
logger.error(error_msg)
return None, error_msg
except Exception as e:
error_msg = f"Unexpected error: {e}"
logger.error(error_msg)
return None, error_msg
# Global client instance
api_client = CarImageAPIClient()
def resize_image(image_data, max_size=(800, 600)):
"""
Resize image to make it smaller while maintaining aspect ratio
Returns resized image data in the same format
"""
try:
img = Image.open(BytesIO(image_data))
original_format = img.format
original_mode = img.mode
# Calculate new size maintaining aspect ratio
img.thumbnail(max_size, Image.Resampling.LANCZOS)
# Save back to bytes in original format
output_buffer = BytesIO()
if original_format and original_format.upper() in ["JPEG", "JPG"]:
img.save(output_buffer, format="JPEG", quality=95, optimize=True)
elif original_format and original_format.upper() == "PNG":
# Preserve transparency for PNG
if original_mode == "RGBA":
img.save(output_buffer, format="PNG", optimize=True)
else:
img.save(output_buffer, format="PNG", optimize=True)
else:
# Default to JPEG for other formats
if img.mode in ("RGBA", "LA", "P"):
# Convert to RGB if image has transparency
background = Image.new("RGB", img.size, (255, 255, 255))
if img.mode == "RGBA":
background.paste(img, mask=img.split()[3])
else:
background.paste(img, (0, 0))
img = background
img.save(output_buffer, format="JPEG", quality=95, optimize=True)
resized_data = output_buffer.getvalue()
logger.info(
f"Resized image from {len(image_data)} to {len(resized_data)} bytes"
)
return resized_data, None
except Exception as e:
error_msg = f"Image resizing failed: {e}"
logger.error(error_msg)
return None, error_msg
def generate_car_image_simple(car):
"""
Simple function to generate car image with authentication and resizing
"""
# Prepare payload
payload = {
"make": car.id_car_make.name if car.id_car_make else "",
"model": car.id_car_model.name if car.id_car_model else "",
"year": car.year,
"color": car.colors.exterior.name,
}
logger.info(f"Generating image for car {car.vin}")
# Generate image using API client
image_data, error = api_client.generate_image(payload)
if error:
return {"success": False, "error": error}
if not image_data:
return {"success": False, "error": "No image data received"}
try:
# Resize the image to make it smaller
resized_data, resize_error = resize_image(image_data, max_size=(800, 600))
if resize_error:
# If resizing fails, use original image but log warning
logger.warning(f"Resizing failed, using original image: {resize_error}")
resized_data = image_data
# Determine file extension based on content
try:
img = Image.open(BytesIO(resized_data))
file_extension = img.format.lower() if img.format else "jpg"
except:
file_extension = "jpg"
# Save the resized image
logger.info(f" {car.vin}")
with open(
os.path.join(
settings.MEDIA_ROOT, f"car_images/{car.get_hash}.{file_extension}"
),
"wb",
) as f:
f.write(resized_data)
logger.info(f"Saved image for car {car.vin}")
# Update CarImage record
logger.info(f"Successfully generated and resized image for car {car.vin}")
return {"success": True}
except Exception as e:
error_msg = f"Image processing failed: {e}"
logger.error(error_msg)
return {"success": False, "error": error_msg}
def create_estimate_(dealer, car, customer):
entity = dealer.entity
title = f"Estimate for {car.vin}-{car.id_car_make.name}-{car.id_car_model.name}-{car.year} for customer {customer.first_name} {customer.last_name}"
estimate = entity.create_estimate(
estimate_title=title,
customer_model=customer.customer_model,
contract_terms="fixed",
)
estimate_itemtxs = {
car.item_model.item_number: {
"unit_cost": round(float(car.marked_price)),
"unit_revenue": round(float(car.marked_price)),
"quantity": 1,
"total_amount": round(float(car.final_price_plus_vat)),
}
}
try:
estimate.migrate_itemtxs(
itemtxs=estimate_itemtxs,
commit=True,
operation=EstimateModel.ITEMIZE_APPEND,
)
except Exception as e:
estimate.delete()
raise e
return estimate
Reference in New Issue View Git Blame Copy Permalink