445 lines
14 KiB
Python
445 lines
14 KiB
Python
"""
|
|
Commands for invoice-related operations.
|
|
"""
|
|
from typing import Any, Dict, List, Optional, Union
|
|
from datetime import date
|
|
|
|
from backend.core.models.invoices.invoices import Invoice
|
|
from backend.core.repositories.invoices.invoices import InvoiceRepository
|
|
from backend.core.repositories.customers.customers import CustomerRepository
|
|
from backend.core.utils.validators import (
|
|
is_valid_uuid, is_valid_date, validate_required_fields,
|
|
validate_model_exists
|
|
)
|
|
from backend.core.utils.helpers import generate_uuid, parse_date
|
|
from backend.core.commands.base import Command, CommandResult
|
|
|
|
|
|
class CreateInvoiceCommand(Command):
|
|
"""
|
|
Command to create a new invoice.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
invoice_repo: InvoiceRepository,
|
|
customer_repo: CustomerRepository,
|
|
customer_id: str,
|
|
invoice_date: str,
|
|
account_ids: Optional[List[str]] = None,
|
|
project_ids: Optional[List[str]] = None,
|
|
total_amount: Optional[float] = None,
|
|
status: str = 'draft'
|
|
):
|
|
"""
|
|
Initialize the create invoice command.
|
|
"""
|
|
self.invoice_repo = invoice_repo
|
|
self.customer_repo = customer_repo
|
|
self.customer_id = customer_id
|
|
self.date = invoice_date
|
|
self.account_ids = account_ids or []
|
|
self.project_ids = project_ids or []
|
|
self.total_amount = total_amount or 0
|
|
self.status = status
|
|
|
|
def validate(self) -> Dict[str, Any]:
|
|
"""
|
|
Validate the invoice creation data.
|
|
"""
|
|
errors = []
|
|
|
|
# Check required fields
|
|
missing_fields = validate_required_fields(
|
|
{'customer_id': self.customer_id, 'date': self.date},
|
|
['customer_id', 'date']
|
|
)
|
|
|
|
if missing_fields:
|
|
errors.append(f"Required fields missing: {', '.join(missing_fields)}")
|
|
|
|
# Validate customer exists
|
|
if not errors and self.customer_id:
|
|
customer_validation = validate_model_exists(
|
|
self.customer_id, 'customer', self.customer_repo.get_by_id
|
|
)
|
|
if not customer_validation['valid']:
|
|
errors.append(customer_validation['error'])
|
|
|
|
# Validate date format
|
|
if not errors and self.date and not is_valid_date(self.date):
|
|
errors.append("Invalid date format. Use YYYY-MM-DD.")
|
|
|
|
# Validate status
|
|
valid_statuses = ['draft', 'sent', 'paid', 'overdue', 'cancelled']
|
|
if not errors and self.status not in valid_statuses:
|
|
errors.append(f"Invalid status. Must be one of: {', '.join(valid_statuses)}")
|
|
|
|
# If no accounts or projects are specified, warn about empty invoice
|
|
if not errors and not self.account_ids and not self.project_ids:
|
|
errors.append("Invoice must include at least one account or project")
|
|
|
|
return {
|
|
'is_valid': len(errors) == 0,
|
|
'errors': errors
|
|
}
|
|
|
|
def execute(self) -> CommandResult[Invoice]:
|
|
"""
|
|
Execute the invoice creation command.
|
|
"""
|
|
# Validate command data
|
|
validation = self.validate()
|
|
if not validation['is_valid']:
|
|
return CommandResult.failure_result(validation['errors'])
|
|
|
|
try:
|
|
# Create invoice data
|
|
invoice_id = generate_uuid()
|
|
|
|
# Setup invoice data
|
|
invoice_data = {
|
|
'id': invoice_id,
|
|
'customer_id': self.customer_id,
|
|
'date': self.date,
|
|
'status': self.status,
|
|
'total_amount': self.total_amount
|
|
}
|
|
|
|
# Create invoice with related items
|
|
created_invoice = self.invoice_repo.create_with_items(
|
|
invoice_data,
|
|
self.account_ids,
|
|
self.project_ids
|
|
)
|
|
|
|
return CommandResult.success_result(
|
|
created_invoice,
|
|
f"Invoice created successfully for customer {self.customer_id}"
|
|
)
|
|
|
|
except Exception as e:
|
|
return CommandResult.failure_result(
|
|
str(e),
|
|
"Failed to create invoice"
|
|
)
|
|
|
|
|
|
class SendInvoiceCommand(Command):
|
|
"""
|
|
Command to send an invoice.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
invoice_repo: InvoiceRepository,
|
|
invoice_id: str
|
|
):
|
|
"""
|
|
Initialize the send invoice command.
|
|
"""
|
|
self.invoice_repo = invoice_repo
|
|
self.invoice_id = invoice_id
|
|
|
|
def validate(self) -> Dict[str, Any]:
|
|
"""
|
|
Validate the invoice sending request.
|
|
"""
|
|
errors = []
|
|
|
|
# Validate invoice exists
|
|
if not is_valid_uuid(self.invoice_id):
|
|
errors.append("Invalid invoice ID format")
|
|
else:
|
|
invoice = self.invoice_repo.get_by_id(self.invoice_id)
|
|
if not invoice:
|
|
errors.append(f"Invoice with ID {self.invoice_id} not found")
|
|
|
|
# Check if invoice is in draft status
|
|
if not errors and invoice.status != 'draft':
|
|
errors.append(f"Only invoices in 'draft' status can be sent. Current status: {invoice.status}")
|
|
|
|
return {
|
|
'is_valid': len(errors) == 0,
|
|
'errors': errors
|
|
}
|
|
|
|
def execute(self) -> CommandResult[Invoice]:
|
|
"""
|
|
Execute the send invoice command.
|
|
"""
|
|
# Validate command data
|
|
validation = self.validate()
|
|
if not validation['is_valid']:
|
|
return CommandResult.failure_result(validation['errors'])
|
|
|
|
try:
|
|
# Update invoice status to 'sent'
|
|
updated_invoice = self.invoice_repo.update_status(self.invoice_id, 'sent')
|
|
|
|
if updated_invoice:
|
|
return CommandResult.success_result(
|
|
updated_invoice,
|
|
f"Invoice {self.invoice_id} sent successfully"
|
|
)
|
|
else:
|
|
return CommandResult.failure_result(
|
|
"Failed to send invoice",
|
|
f"Invoice {self.invoice_id} could not be sent"
|
|
)
|
|
|
|
except Exception as e:
|
|
return CommandResult.failure_result(
|
|
str(e),
|
|
"Failed to send invoice"
|
|
)
|
|
|
|
|
|
class MarkInvoicePaidCommand(Command):
|
|
"""
|
|
Command to mark an invoice as paid.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
invoice_repo: InvoiceRepository,
|
|
invoice_id: str,
|
|
payment_type: str,
|
|
date_paid: Optional[str] = None
|
|
):
|
|
"""
|
|
Initialize the mark invoice paid command.
|
|
"""
|
|
self.invoice_repo = invoice_repo
|
|
self.invoice_id = invoice_id
|
|
self.payment_type = payment_type
|
|
self.date_paid = date_paid
|
|
|
|
def validate(self) -> Dict[str, Any]:
|
|
"""
|
|
Validate the mark invoice paid request.
|
|
"""
|
|
errors = []
|
|
|
|
# Validate invoice exists
|
|
if not is_valid_uuid(self.invoice_id):
|
|
errors.append("Invalid invoice ID format")
|
|
else:
|
|
invoice = self.invoice_repo.get_by_id(self.invoice_id)
|
|
if not invoice:
|
|
errors.append(f"Invoice with ID {self.invoice_id} not found")
|
|
|
|
# Check if invoice can be marked as paid
|
|
if not errors and invoice.status not in ['sent', 'overdue']:
|
|
errors.append(
|
|
f"Only invoices in 'sent' or 'overdue' status can be marked as paid. Current status: {invoice.status}")
|
|
|
|
# Validate payment type
|
|
valid_payment_types = ['check', 'credit_card', 'bank_transfer', 'cash']
|
|
if not errors and self.payment_type not in valid_payment_types:
|
|
errors.append(f"Invalid payment type. Must be one of: {', '.join(valid_payment_types)}")
|
|
|
|
# Validate date_paid format if provided
|
|
if not errors and self.date_paid and not is_valid_date(self.date_paid):
|
|
errors.append("Invalid date_paid format. Use YYYY-MM-DD.")
|
|
|
|
return {
|
|
'is_valid': len(errors) == 0,
|
|
'errors': errors
|
|
}
|
|
|
|
def execute(self) -> CommandResult[Invoice]:
|
|
"""
|
|
Execute the mark invoice paid command.
|
|
"""
|
|
# Validate command data
|
|
validation = self.validate()
|
|
if not validation['is_valid']:
|
|
return CommandResult.failure_result(validation['errors'])
|
|
|
|
try:
|
|
# If date_paid is provided, update with that date
|
|
if self.date_paid:
|
|
update_data = {
|
|
'status': 'paid',
|
|
'date_paid': self.date_paid,
|
|
'payment_type': self.payment_type
|
|
}
|
|
updated_invoice = self.invoice_repo.update(self.invoice_id, update_data)
|
|
else:
|
|
# Use the repository method to mark as paid with today's date
|
|
updated_invoice = self.invoice_repo.mark_as_paid(self.invoice_id, self.payment_type)
|
|
|
|
return CommandResult.success_result(
|
|
updated_invoice,
|
|
f"Invoice {self.invoice_id} marked as paid successfully"
|
|
)
|
|
|
|
except Exception as e:
|
|
return CommandResult.failure_result(
|
|
str(e),
|
|
"Failed to mark invoice as paid"
|
|
)
|
|
|
|
|
|
class CancelInvoiceCommand(Command):
|
|
"""
|
|
Command to cancel an invoice.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
invoice_repo: InvoiceRepository,
|
|
invoice_id: str
|
|
):
|
|
"""
|
|
Initialize the cancel invoice command.
|
|
"""
|
|
self.invoice_repo = invoice_repo
|
|
self.invoice_id = invoice_id
|
|
|
|
def validate(self) -> Dict[str, Any]:
|
|
"""
|
|
Validate the invoice cancellation request.
|
|
"""
|
|
errors = []
|
|
|
|
# Validate invoice exists
|
|
if not is_valid_uuid(self.invoice_id):
|
|
errors.append("Invalid invoice ID format")
|
|
else:
|
|
invoice = self.invoice_repo.get_by_id(self.invoice_id)
|
|
if not invoice:
|
|
errors.append(f"Invoice with ID {self.invoice_id} not found")
|
|
|
|
# Check if invoice can be cancelled
|
|
if not errors and invoice.status == 'paid':
|
|
errors.append("Paid invoices cannot be cancelled")
|
|
|
|
return {
|
|
'is_valid': len(errors) == 0,
|
|
'errors': errors
|
|
}
|
|
|
|
def execute(self) -> CommandResult[Invoice]:
|
|
"""
|
|
Execute the cancel invoice command.
|
|
"""
|
|
# Validate command data
|
|
validation = self.validate()
|
|
if not validation['is_valid']:
|
|
return CommandResult.failure_result(validation['errors'])
|
|
|
|
try:
|
|
# Update invoice status to 'cancelled'
|
|
updated_invoice = self.invoice_repo.update_status(self.invoice_id, 'cancelled')
|
|
|
|
return CommandResult.success_result(
|
|
updated_invoice,
|
|
f"Invoice {self.invoice_id} cancelled successfully"
|
|
)
|
|
|
|
except Exception as e:
|
|
return CommandResult.failure_result(
|
|
str(e),
|
|
"Failed to cancel invoice"
|
|
)
|
|
|
|
|
|
class FilterInvoicesCommand(Command):
|
|
"""
|
|
Command to filter invoices by multiple criteria.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
invoice_repo: InvoiceRepository,
|
|
customer_id: Optional[str] = None,
|
|
status: Optional[str] = None,
|
|
date_from: Optional[Union[str, date]] = None,
|
|
date_to: Optional[Union[str, date]] = None,
|
|
account_id: Optional[str] = None,
|
|
project_id: Optional[str] = None
|
|
):
|
|
"""
|
|
Initialize the filter invoices command.
|
|
"""
|
|
self.invoice_repo = invoice_repo
|
|
self.customer_id = customer_id
|
|
self.status = status
|
|
|
|
# Convert string dates to date objects if needed
|
|
if isinstance(date_from, str):
|
|
self.date_from = parse_date(date_from)
|
|
else:
|
|
self.date_from = date_from
|
|
|
|
if isinstance(date_to, str):
|
|
self.date_to = parse_date(date_to)
|
|
else:
|
|
self.date_to = date_to
|
|
|
|
self.account_id = account_id
|
|
self.project_id = project_id
|
|
|
|
def validate(self) -> Dict[str, Any]:
|
|
"""
|
|
Validate the filter invoices request.
|
|
"""
|
|
errors = []
|
|
|
|
# Validate UUID formats
|
|
if self.customer_id and not is_valid_uuid(self.customer_id):
|
|
errors.append("Invalid customer ID format")
|
|
|
|
if self.account_id and not is_valid_uuid(self.account_id):
|
|
errors.append("Invalid account ID format")
|
|
|
|
if self.project_id and not is_valid_uuid(self.project_id):
|
|
errors.append("Invalid project ID format")
|
|
|
|
# Validate status if provided
|
|
if self.status:
|
|
valid_statuses = ['draft', 'sent', 'paid', 'overdue', 'cancelled']
|
|
if self.status not in valid_statuses:
|
|
errors.append(f"Invalid status. Must be one of: {', '.join(valid_statuses)}")
|
|
|
|
# Validate date range if both provided
|
|
if self.date_from and self.date_to and self.date_from > self.date_to:
|
|
errors.append("Start date must be before end date")
|
|
|
|
return {
|
|
'is_valid': len(errors) == 0,
|
|
'errors': errors
|
|
}
|
|
|
|
def execute(self) -> CommandResult[List[Invoice]]:
|
|
"""
|
|
Execute the filter invoices command.
|
|
"""
|
|
# Validate command data
|
|
validation = self.validate()
|
|
if not validation['is_valid']:
|
|
return CommandResult.failure_result(validation['errors'])
|
|
|
|
try:
|
|
# Filter invoices
|
|
filtered_invoices = self.invoice_repo.filter_invoices(
|
|
customer_id=self.customer_id,
|
|
status=self.status,
|
|
date_from=self.date_from,
|
|
date_to=self.date_to,
|
|
account_id=self.account_id,
|
|
project_id=self.project_id
|
|
)
|
|
|
|
return CommandResult.success_result(
|
|
list(filtered_invoices),
|
|
f"Found {filtered_invoices.count()} matching invoices"
|
|
)
|
|
|
|
except Exception as e:
|
|
return CommandResult.failure_result(
|
|
str(e),
|
|
"Failed to filter invoices"
|
|
) |