304 lines
9.0 KiB
Python
304 lines
9.0 KiB
Python
"""Router for contract resource"""
|
|
import io
|
|
import zipfile
|
|
|
|
import src.contracts.service as service
|
|
import src.forms.service as form_service
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from fastapi.responses import StreamingResponse
|
|
from sqlmodel import Session
|
|
from src import messages, models
|
|
from src.auth.auth import get_current_user
|
|
from src.contracts.generate_contract import (generate_html_contract,
|
|
generate_recap)
|
|
from src.database import get_session
|
|
|
|
router = APIRouter(prefix='/contracts')
|
|
|
|
|
|
@router.post('')
|
|
async def create_contract(
|
|
contract: models.ContractCreate,
|
|
session: Session = Depends(get_session),
|
|
):
|
|
"""Create contract route"""
|
|
new_contract = service.create_one(session, contract)
|
|
occasional_contract_products = list(
|
|
filter(
|
|
lambda contract_product: (
|
|
contract_product.product.type == models.ProductType.OCCASIONAL
|
|
),
|
|
new_contract.products
|
|
)
|
|
)
|
|
occasionals = service.create_occasional_dict(occasional_contract_products)
|
|
recurrents = list(
|
|
map(
|
|
lambda x: {'product': x.product, 'quantity': x.quantity},
|
|
filter(
|
|
lambda contract_product: (
|
|
contract_product.product.type ==
|
|
models.ProductType.RECCURENT
|
|
),
|
|
new_contract.products
|
|
)
|
|
)
|
|
)
|
|
prices = service.generate_products_prices(
|
|
occasionals,
|
|
recurrents,
|
|
new_contract.form.shipments
|
|
)
|
|
recurrent_price = prices['recurrent']
|
|
total_price = prices['total']
|
|
cheques = list(
|
|
map(
|
|
lambda x: {'name': x.name, 'value': x.value},
|
|
new_contract.cheques
|
|
)
|
|
)
|
|
try:
|
|
pdf_bytes = generate_html_contract(
|
|
new_contract,
|
|
cheques,
|
|
occasionals,
|
|
recurrents,
|
|
'{:10.2f}'.format(recurrent_price),
|
|
'{:10.2f}'.format(total_price)
|
|
)
|
|
pdf_file = io.BytesIO(pdf_bytes)
|
|
contract_id = (
|
|
f'{new_contract.firstname}_'
|
|
f'{new_contract.lastname}_'
|
|
f'{new_contract.form.productor.type}_'
|
|
f'{new_contract.form.season}'
|
|
)
|
|
service.add_contract_file(
|
|
session, new_contract.id, pdf_bytes, total_price)
|
|
except Exception as error:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=messages.pdferror
|
|
) from error
|
|
return StreamingResponse(
|
|
pdf_file,
|
|
media_type='application/pdf',
|
|
headers={
|
|
'Content-Disposition': (
|
|
f'attachment; filename=contract_{contract_id}.pdf'
|
|
)
|
|
}
|
|
)
|
|
|
|
|
|
@router.get('/{form_id}/base')
|
|
async def get_base_contract_template(
|
|
form_id: int,
|
|
session: Session = Depends(get_session),
|
|
):
|
|
"""Get contract template route"""
|
|
form = form_service.get_one(session, form_id)
|
|
recurrents = [
|
|
{'product': product, 'quantity': None}
|
|
for product in form.productor.products
|
|
if product.type == models.ProductType.RECCURENT
|
|
]
|
|
occasionals = [{
|
|
'shipment': sh,
|
|
'price': None,
|
|
'products': [{'product': pr, 'quantity': None} for pr in sh.products]
|
|
} for sh in form.shipments]
|
|
empty_contract = models.ContractPublic(
|
|
firstname='',
|
|
form=form,
|
|
lastname='',
|
|
email='',
|
|
phone='',
|
|
products=[],
|
|
payment_method='cheque',
|
|
cheque_quantity=3,
|
|
total_price=0,
|
|
id=1
|
|
)
|
|
cheques = [
|
|
{'name': None, 'value': None},
|
|
{'name': None, 'value': None},
|
|
{'name': None, 'value': None}
|
|
]
|
|
try:
|
|
pdf_bytes = generate_html_contract(
|
|
empty_contract,
|
|
cheques,
|
|
occasionals,
|
|
recurrents,
|
|
)
|
|
pdf_file = io.BytesIO(pdf_bytes)
|
|
contract_id = (
|
|
f'{empty_contract.form.productor.type}_'
|
|
f'{empty_contract.form.season}'
|
|
)
|
|
except Exception as error:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=messages.pdferror
|
|
) from error
|
|
return StreamingResponse(
|
|
pdf_file,
|
|
media_type='application/pdf',
|
|
headers={
|
|
'Content-Disposition': (
|
|
f'attachment; filename=contract_{contract_id}.pdf'
|
|
)
|
|
}
|
|
)
|
|
|
|
|
|
@router.get('', response_model=list[models.ContractPublic])
|
|
def get_contracts(
|
|
forms: list[str] = Query([]),
|
|
session: Session = Depends(get_session),
|
|
user: models.User = Depends(get_current_user)
|
|
):
|
|
"""Get all contracts route"""
|
|
return service.get_all(session, user, forms)
|
|
|
|
|
|
@router.get('/{_id}/file')
|
|
def get_contract_file(
|
|
_id: int,
|
|
session: Session = Depends(get_session),
|
|
user: models.User = Depends(get_current_user)
|
|
):
|
|
"""Get a contract file (in pdf) route"""
|
|
if not service.is_allowed(session, user, _id):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=messages.Messages.not_allowed('contract', 'get')
|
|
)
|
|
contract = service.get_one(session, _id)
|
|
if contract is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=messages.Messages.not_found('contract')
|
|
)
|
|
filename = (
|
|
f'{contract.form.name.replace(' ', '_')}_'
|
|
f'{contract.form.season}_'
|
|
f'{contract.firstname}_'
|
|
f'{contract.lastname}'
|
|
)
|
|
return StreamingResponse(
|
|
io.BytesIO(contract.file),
|
|
media_type='application/pdf',
|
|
headers={
|
|
'Content-Disposition': f'attachment; filename={filename}.pdf'
|
|
}
|
|
)
|
|
|
|
|
|
@router.get('/{form_id}/files')
|
|
def get_contract_files(
|
|
form_id: int,
|
|
session: Session = Depends(get_session),
|
|
user: models.User = Depends(get_current_user)
|
|
):
|
|
"""Get all contract files for a given form"""
|
|
if not form_service.is_allowed(session, user, form_id):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=messages.Messages.not_allowed('contracts', 'get')
|
|
)
|
|
form = form_service.get_one(session, form_id=form_id)
|
|
contracts = service.get_all(session, user, forms=[form.name])
|
|
zipped_contracts = io.BytesIO()
|
|
with zipfile.ZipFile(
|
|
zipped_contracts,
|
|
'a',
|
|
zipfile.ZIP_DEFLATED,
|
|
False
|
|
) as zip_file:
|
|
for contract in contracts:
|
|
contract_filename = (
|
|
f'{contract.form.name.replace(' ', '_')}_'
|
|
f'{contract.form.season}_'
|
|
f'{contract.firstname}_'
|
|
f'{contract.lastname}'
|
|
)
|
|
zip_file.writestr(contract_filename, contract.file)
|
|
filename = f'{form.name.replace(' ', '_')}_{form.season}'
|
|
return StreamingResponse(
|
|
io.BytesIO(zipped_contracts.getvalue()),
|
|
media_type='application/zip',
|
|
headers={
|
|
'Content-Disposition': f'attachment; filename={filename}.zip'
|
|
}
|
|
)
|
|
|
|
|
|
@router.get('/{form_id}/recap')
|
|
def get_contract_recap(
|
|
form_id: int,
|
|
session: Session = Depends(get_session),
|
|
user: models.User = Depends(get_current_user)
|
|
):
|
|
"""Get a contract recap for a given form"""
|
|
if not form_service.is_allowed(session, user, form_id):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=messages.Messages.not_allowed('contract recap', 'get')
|
|
)
|
|
form = form_service.get_one(session, form_id=form_id)
|
|
contracts = service.get_all(session, user, forms=[form.name])
|
|
filename = f'{form.name}_recapitulatif_contrats.ods'
|
|
return StreamingResponse(
|
|
io.BytesIO(generate_recap(contracts, form)),
|
|
media_type='application/vnd.oasis.opendocument.spreadsheet',
|
|
headers={
|
|
'Content-Disposition': (
|
|
f'attachment; filename={filename}'
|
|
)
|
|
}
|
|
)
|
|
|
|
|
|
@router.get('/{_id}', response_model=models.ContractPublic)
|
|
def get_contract(
|
|
_id: int,
|
|
session: Session = Depends(get_session),
|
|
user: models.User = Depends(get_current_user)
|
|
):
|
|
"""Get a contract route"""
|
|
if not service.is_allowed(session, user, _id):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=messages.Messages.not_allowed('contract', 'get')
|
|
)
|
|
result = service.get_one(session, _id)
|
|
if result is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=messages.Messages.not_found('contract')
|
|
)
|
|
return result
|
|
|
|
|
|
@router.delete('/{_id}', response_model=models.ContractPublic)
|
|
def delete_contract(
|
|
_id: int,
|
|
session: Session = Depends(get_session),
|
|
user: models.User = Depends(get_current_user)
|
|
):
|
|
"""Delete contract route"""
|
|
if not service.is_allowed(session, user, _id):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=messages.Messages.not_allowed('contract', 'delete')
|
|
)
|
|
result = service.delete_one(session, _id)
|
|
if result is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=messages.Messages.not_found('contract')
|
|
)
|
|
return result
|