122 lines
6.8 KiB
Python
122 lines
6.8 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status, Query
|
|
from sqlalchemy.orm import Session
|
|
from typing import Optional
|
|
from datetime import datetime
|
|
from ..config.database import get_db
|
|
from ..middleware.auth import get_current_user, authorize_roles
|
|
from ..models.user import User
|
|
from ..models.invoice import Invoice, InvoiceStatus
|
|
from ..models.booking import Booking
|
|
from ..services.invoice_service import InvoiceService
|
|
router = APIRouter(prefix='/invoices', tags=['invoices'])
|
|
|
|
@router.get('/')
|
|
async def get_invoices(booking_id: Optional[int]=Query(None), status_filter: Optional[str]=Query(None, alias='status'), page: int=Query(1, ge=1), limit: int=Query(10, ge=1, le=100), current_user: User=Depends(get_current_user), db: Session=Depends(get_db)):
|
|
try:
|
|
user_id = None if current_user.role_id == 1 else current_user.id
|
|
result = InvoiceService.get_invoices(db=db, user_id=user_id, booking_id=booking_id, status=status_filter, page=page, limit=limit)
|
|
return {'status': 'success', 'data': result}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@router.get('/{id}')
|
|
async def get_invoice_by_id(id: int, current_user: User=Depends(get_current_user), db: Session=Depends(get_db)):
|
|
try:
|
|
invoice = InvoiceService.get_invoice(id, db)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail='Invoice not found')
|
|
if current_user.role_id != 1 and invoice['user_id'] != current_user.id:
|
|
raise HTTPException(status_code=403, detail='Forbidden')
|
|
return {'status': 'success', 'data': {'invoice': invoice}}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@router.post('/')
|
|
async def create_invoice(invoice_data: dict, current_user: User=Depends(get_current_user), db: Session=Depends(get_db)):
|
|
try:
|
|
if current_user.role_id not in [1, 2]:
|
|
raise HTTPException(status_code=403, detail='Forbidden')
|
|
booking_id = invoice_data.get('booking_id')
|
|
if not booking_id:
|
|
raise HTTPException(status_code=400, detail='booking_id is required')
|
|
try:
|
|
booking_id = int(booking_id)
|
|
except (ValueError, TypeError):
|
|
raise HTTPException(status_code=400, detail='booking_id must be a valid integer')
|
|
booking = db.query(Booking).filter(Booking.id == booking_id).first()
|
|
if not booking:
|
|
raise HTTPException(status_code=404, detail='Booking not found')
|
|
invoice_kwargs = {'company_name': invoice_data.get('company_name'), 'company_address': invoice_data.get('company_address'), 'company_phone': invoice_data.get('company_phone'), 'company_email': invoice_data.get('company_email'), 'company_tax_id': invoice_data.get('company_tax_id'), 'company_logo_url': invoice_data.get('company_logo_url'), 'customer_tax_id': invoice_data.get('customer_tax_id'), 'notes': invoice_data.get('notes'), 'terms_and_conditions': invoice_data.get('terms_and_conditions'), 'payment_instructions': invoice_data.get('payment_instructions')}
|
|
invoice_notes = invoice_kwargs.get('notes', '')
|
|
if booking.promotion_code:
|
|
promotion_note = f'Promotion Code: {booking.promotion_code}'
|
|
invoice_notes = f'{promotion_note}\n{invoice_notes}'.strip() if invoice_notes else promotion_note
|
|
invoice_kwargs['notes'] = invoice_notes
|
|
invoice = InvoiceService.create_invoice_from_booking(booking_id=booking_id, db=db, created_by_id=current_user.id, tax_rate=invoice_data.get('tax_rate', 0.0), discount_amount=invoice_data.get('discount_amount', 0.0), due_days=invoice_data.get('due_days', 30), **invoice_kwargs)
|
|
return {'status': 'success', 'message': 'Invoice created successfully', 'data': {'invoice': invoice}}
|
|
except HTTPException:
|
|
raise
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@router.put('/{id}')
|
|
async def update_invoice(id: int, invoice_data: dict, current_user: User=Depends(authorize_roles('admin', 'staff')), db: Session=Depends(get_db)):
|
|
try:
|
|
invoice = db.query(Invoice).filter(Invoice.id == id).first()
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail='Invoice not found')
|
|
updated_invoice = InvoiceService.update_invoice(invoice_id=id, db=db, updated_by_id=current_user.id, **invoice_data)
|
|
return {'status': 'success', 'message': 'Invoice updated successfully', 'data': {'invoice': updated_invoice}}
|
|
except HTTPException:
|
|
raise
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@router.post('/{id}/mark-paid')
|
|
async def mark_invoice_as_paid(id: int, payment_data: dict, current_user: User=Depends(authorize_roles('admin', 'staff')), db: Session=Depends(get_db)):
|
|
try:
|
|
amount = payment_data.get('amount')
|
|
updated_invoice = InvoiceService.mark_invoice_as_paid(invoice_id=id, db=db, amount=amount, updated_by_id=current_user.id)
|
|
return {'status': 'success', 'message': 'Invoice marked as paid successfully', 'data': {'invoice': updated_invoice}}
|
|
except HTTPException:
|
|
raise
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@router.delete('/{id}')
|
|
async def delete_invoice(id: int, current_user: User=Depends(authorize_roles('admin')), db: Session=Depends(get_db)):
|
|
try:
|
|
invoice = db.query(Invoice).filter(Invoice.id == id).first()
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail='Invoice not found')
|
|
db.delete(invoice)
|
|
db.commit()
|
|
return {'status': 'success', 'message': 'Invoice deleted successfully'}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
db.rollback()
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@router.get('/booking/{booking_id}')
|
|
async def get_invoices_by_booking(booking_id: int, current_user: User=Depends(get_current_user), db: Session=Depends(get_db)):
|
|
try:
|
|
booking = db.query(Booking).filter(Booking.id == booking_id).first()
|
|
if not booking:
|
|
raise HTTPException(status_code=404, detail='Booking not found')
|
|
if current_user.role_id != 1 and booking.user_id != current_user.id:
|
|
raise HTTPException(status_code=403, detail='Forbidden')
|
|
result = InvoiceService.get_invoices(db=db, booking_id=booking_id)
|
|
return {'status': 'success', 'data': result}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e)) |