302 lines
8.3 KiB
Python
302 lines
8.3 KiB
Python
"""
|
|
Repository for Invoice model operations.
|
|
"""
|
|
from typing import List, Optional, Dict, Any
|
|
from datetime import date, timedelta
|
|
from django.db.models import QuerySet, Sum
|
|
from django.utils import timezone
|
|
from backend.core.models import Invoice, Account, Project
|
|
from backend.core.repositories.base import BaseRepository
|
|
|
|
|
|
class InvoiceRepository(BaseRepository[Invoice]):
|
|
"""
|
|
Repository for Invoice model operations.
|
|
"""
|
|
model = Invoice
|
|
|
|
@classmethod
|
|
def get_by_customer(cls, customer_id: str) -> QuerySet[Invoice]:
|
|
"""
|
|
Get invoices by customer.
|
|
|
|
Args:
|
|
customer_id: The customer ID
|
|
|
|
Returns:
|
|
QuerySet of invoices for the customer
|
|
"""
|
|
return Invoice.objects.filter(customer_id=customer_id)
|
|
|
|
@classmethod
|
|
def get_by_status(cls, status: str) -> QuerySet[Invoice]:
|
|
"""
|
|
Get invoices by status.
|
|
|
|
Args:
|
|
status: The invoice status
|
|
|
|
Returns:
|
|
QuerySet of invoices with the specified status
|
|
"""
|
|
return Invoice.objects.filter(status=status)
|
|
|
|
@classmethod
|
|
def get_by_date_range(cls, start_date: date = None, end_date: date = None) -> QuerySet[Invoice]:
|
|
"""
|
|
Get invoices within a date range.
|
|
|
|
Args:
|
|
start_date: Start date (inclusive)
|
|
end_date: End date (inclusive)
|
|
|
|
Returns:
|
|
QuerySet of invoices within the date range
|
|
"""
|
|
return cls.filter_by_date_range(start_date, end_date)
|
|
|
|
@classmethod
|
|
def get_overdue(cls) -> QuerySet[Invoice]:
|
|
"""
|
|
Get overdue invoices.
|
|
|
|
Returns:
|
|
QuerySet of overdue invoices
|
|
"""
|
|
thirty_days_ago = timezone.now().date() - timedelta(days=30)
|
|
return Invoice.objects.filter(
|
|
status__in=['sent', 'overdue'], # Include both sent and already marked overdue
|
|
sent_at__isnull=False, # Must have been sent
|
|
).exclude(
|
|
status__in=['paid', 'cancelled'] # Not paid or cancelled
|
|
).filter(
|
|
sent_at__date__lt=thirty_days_ago # Sent more than 30 days ago
|
|
)
|
|
|
|
@classmethod
|
|
def get_unpaid(cls) -> QuerySet[Invoice]:
|
|
"""
|
|
Get unpaid invoices.
|
|
|
|
Returns:
|
|
QuerySet of unpaid invoices (sent but not paid)
|
|
"""
|
|
return Invoice.objects.filter(status='sent')
|
|
|
|
@classmethod
|
|
def search(cls, search_term: str, search_fields: List[str] = None) -> QuerySet[Invoice]:
|
|
"""
|
|
Search invoices by customer.
|
|
|
|
Args:
|
|
search_term: The search term
|
|
search_fields: Optional list of fields to search (ignored, using predefined fields)
|
|
|
|
Returns:
|
|
QuerySet of matching invoices
|
|
"""
|
|
return super().search(
|
|
search_term,
|
|
['customer__name']
|
|
)
|
|
|
|
@classmethod
|
|
def filter_invoices(
|
|
cls,
|
|
customer_id: str = None,
|
|
status: str = None,
|
|
date_from: date = None,
|
|
date_to: date = None,
|
|
account_id: str = None,
|
|
project_id: str = None
|
|
) -> QuerySet[Invoice]:
|
|
"""
|
|
Filter invoices by multiple criteria.
|
|
|
|
Args:
|
|
customer_id: Filter by customer ID
|
|
status: Filter by status
|
|
date_from: Filter by start date (inclusive)
|
|
date_to: Filter by end date (inclusive)
|
|
account_id: Filter by account ID
|
|
project_id: Filter by project ID
|
|
|
|
Returns:
|
|
QuerySet of matching invoices
|
|
"""
|
|
queryset = Invoice.objects.all()
|
|
|
|
if customer_id:
|
|
queryset = queryset.filter(customer_id=customer_id)
|
|
|
|
if status:
|
|
queryset = queryset.filter(status=status)
|
|
|
|
if date_from:
|
|
queryset = queryset.filter(date__gte=date_from)
|
|
|
|
if date_to:
|
|
queryset = queryset.filter(date__lte=date_to)
|
|
|
|
if account_id:
|
|
queryset = queryset.filter(accounts__id=account_id)
|
|
|
|
if project_id:
|
|
queryset = queryset.filter(projects__id=project_id)
|
|
|
|
return queryset
|
|
|
|
@classmethod
|
|
def create_with_items(
|
|
cls,
|
|
data: Dict[str, Any],
|
|
account_ids: List[str] = None,
|
|
project_ids: List[str] = None
|
|
) -> Invoice:
|
|
"""
|
|
Create an invoice with related items.
|
|
|
|
Args:
|
|
data: Invoice data
|
|
account_ids: List of account IDs
|
|
project_ids: List of project IDs
|
|
|
|
Returns:
|
|
The created invoice
|
|
"""
|
|
# Create the invoice
|
|
invoice = cls.create(data)
|
|
|
|
# Add accounts
|
|
if account_ids:
|
|
accounts = Account.objects.filter(id__in=account_ids)
|
|
invoice.accounts.set(accounts)
|
|
|
|
# Add projects
|
|
if project_ids:
|
|
projects = Project.objects.filter(id__in=project_ids)
|
|
invoice.projects.set(projects)
|
|
|
|
return invoice
|
|
|
|
@classmethod
|
|
def update_status(cls, invoice_id: str, status: str) -> Optional[Invoice]:
|
|
"""
|
|
Update invoice status.
|
|
|
|
Args:
|
|
invoice_id: The invoice ID
|
|
status: The new status
|
|
|
|
Returns:
|
|
The updated invoice or None if not found
|
|
"""
|
|
invoice = cls.get_by_id(invoice_id)
|
|
if not invoice:
|
|
return None
|
|
|
|
invoice.status = status
|
|
|
|
# Set sent_at if status is 'sent'
|
|
if status == 'sent' and not invoice.sent_at:
|
|
invoice.sent_at = timezone.now()
|
|
|
|
# Set date_paid if status is 'paid'
|
|
if status == 'paid' and not invoice.date_paid:
|
|
invoice.date_paid = timezone.now().date()
|
|
|
|
invoice.save()
|
|
return invoice
|
|
|
|
@classmethod
|
|
def mark_as_paid(cls, invoice_id: str, payment_type: str) -> Optional[Invoice]:
|
|
"""
|
|
Mark an invoice as paid.
|
|
|
|
Args:
|
|
invoice_id: The invoice ID
|
|
payment_type: The payment type
|
|
|
|
Returns:
|
|
The updated invoice or None if not found
|
|
"""
|
|
invoice = cls.get_by_id(invoice_id)
|
|
if not invoice:
|
|
return None
|
|
|
|
invoice.status = 'paid'
|
|
invoice.date_paid = timezone.now().date()
|
|
invoice.payment_type = payment_type
|
|
invoice.save()
|
|
|
|
return invoice
|
|
|
|
@classmethod
|
|
def get_total_paid(cls, customer_id: str = None, date_from: date = None, date_to: date = None) -> float:
|
|
"""
|
|
Get total paid invoice amount.
|
|
|
|
Args:
|
|
customer_id: Filter by customer ID
|
|
date_from: Filter by start date (inclusive)
|
|
date_to: Filter by end date (inclusive)
|
|
|
|
Returns:
|
|
Total paid amount
|
|
"""
|
|
queryset = Invoice.objects.filter(status='paid')
|
|
|
|
if customer_id:
|
|
queryset = queryset.filter(customer_id=customer_id)
|
|
|
|
if date_from:
|
|
queryset = queryset.filter(date_paid__gte=date_from)
|
|
|
|
if date_to:
|
|
queryset = queryset.filter(date_paid__lte=date_to)
|
|
|
|
result = queryset.aggregate(total=Sum('total_amount'))
|
|
return float(result['total'] or 0)
|
|
|
|
@classmethod
|
|
def get_total_outstanding(cls, customer_id: str = None) -> float:
|
|
"""
|
|
Get total outstanding invoice amount.
|
|
|
|
Args:
|
|
customer_id: Filter by customer ID
|
|
|
|
Returns:
|
|
Total outstanding amount
|
|
"""
|
|
queryset = Invoice.objects.filter(status__in=['sent', 'overdue'])
|
|
|
|
if customer_id:
|
|
queryset = queryset.filter(customer_id=customer_id)
|
|
|
|
result = queryset.aggregate(total=Sum('total_amount'))
|
|
return float(result['total'] or 0)
|
|
|
|
@classmethod
|
|
def mark_overdue(cls, invoice_id: str = None) -> int:
|
|
"""
|
|
Mark invoice(s) as overdue.
|
|
|
|
Args:
|
|
invoice_id: Optional invoice ID. If not provided, all overdue invoices will be marked.
|
|
|
|
Returns:
|
|
Number of invoices marked as overdue
|
|
"""
|
|
thirty_days_ago = timezone.now().date() - timedelta(days=30)
|
|
query = Invoice.objects.filter(
|
|
status='sent',
|
|
sent_at__isnull=False,
|
|
sent_at__date__lt=thirty_days_ago
|
|
)
|
|
|
|
if invoice_id:
|
|
query = query.filter(id=invoice_id)
|
|
|
|
count = query.update(status='overdue')
|
|
return count |