Mop/amap
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

add functionnal recap ready for tests

Browse Source
This commit is contained in:
Julien Aldon
2026-03-05 17:17:23 +01:00
parent 3cfa60507e
commit ff19448991
5 changed files with 359 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files

97
backend/src/contracts/contracts.py
View File

@@ -17,88 +17,6 @@ from src.database import get_session
router = APIRouter(prefix='/contracts')
def compute_recurrent_prices(
products_quantities: list[dict],
nb_shipment: int
):
"""Compute price for recurrent products"""
result = 0
for product_quantity in products_quantities:
product = product_quantity['product']
quantity = product_quantity['quantity']
result += compute_product_price(product, quantity, nb_shipment)
return result
def compute_occasional_prices(occasionals: list[dict]):
"""Compute prices for occassional products"""
result = 0
for occasional in occasionals:
result += occasional['price']
return result
def compute_product_price(
product: models.Product,
quantity: int,
nb_shipment: int = 1
):
"""Compute price for a product"""
product_quantity_unit = (
1 if product.unit == models.Unit.KILO else 1000
)
final_quantity = (
quantity if product.price else quantity / product_quantity_unit
)
final_price = (
product.price if product.price else product.price_kg
)
return final_price * final_quantity * nb_shipment
def find_dict_in_list(lst, key, value):
"""Find the index of a dictionnary in a list of dictionnaries given a key
and a value.
"""
for i, dic in enumerate(lst):
if dic[key].id == value:
return i
return -1
def create_occasional_dict(contract_products: list[models.ContractProduct]):
"""Create a dictionnary of occasional products"""
result = []
for contract_product in contract_products:
existing_id = find_dict_in_list(
result,
'shipment',
contract_product.shipment.id
)
if existing_id < 0:
result.append({
'shipment': contract_product.shipment,
'price': compute_product_price(
contract_product.product,
contract_product.quantity
),
'products': [{
'product': contract_product.product,
'quantity': contract_product.quantity
}]
})
else:
result[existing_id]['products'].append({
'product': contract_product.product,
'quantity': contract_product.quantity
})
result[existing_id]['price'] += compute_product_price(
contract_product.product,
contract_product.quantity
)
return result
@router.post('')
async def create_contract(
contract: models.ContractCreate,
@@ -114,7 +32,7 @@ async def create_contract(
new_contract.products
)
)
occasionals = create_occasional_dict(occasional_contract_products)
occasionals = service.create_occasional_dict(occasional_contract_products)
recurrents = list(
map(
lambda x: {'product': x.product, 'quantity': x.quantity},
@@ -127,11 +45,13 @@ async def create_contract(
)
)
)
recurrent_price = compute_recurrent_prices(
prices = service.generate_products_prices(
occasionals,
recurrents,
len(new_contract.form.shipments)
new_contract.form.shipments
)
price = recurrent_price + compute_occasional_prices(occasionals)
recurrent_price = prices['recurrent']
total_price = prices['total']
cheques = list(
map(
lambda x: {'name': x.name, 'value': x.value},
@@ -145,7 +65,7 @@ async def create_contract(
occasionals,
recurrents,
'{:10.2f}'.format(recurrent_price),
'{:10.2f}'.format(price)
'{:10.2f}'.format(total_price)
)
pdf_file = io.BytesIO(pdf_bytes)
contract_id = (
@@ -154,7 +74,8 @@ async def create_contract(
f'{new_contract.form.productor.type}_'
f'{new_contract.form.season}'
)
service.add_contract_file(session, new_contract.id, pdf_bytes, price)
service.add_contract_file(
session, new_contract.id, pdf_bytes, total_price)
except Exception as error:
raise HTTPException(
status_code=400,