from django.db.models import Count from rest_framework import viewsets from rest_framework.decorators import action from rest_framework.exceptions import MethodNotAllowed, PermissionDenied, ValidationError from rest_framework.response import Response from drf_spectacular.utils import extend_schema, OpenApiParameter from django.utils import timezone from apps.core.tenancy import TenantScopedMixin, get_active_organization from .models import Currency, TaxRate, BankAccount, BankStatementImport, BudgetCategory, BudgetItem, Invoice, to_letters from .serializers import ( CurrencySerializer, TaxRateSerializer, BankAccountSerializer, BankStatementImportSerializer, BudgetCategorySerializer, BudgetItemSerializer, InvoiceSerializer, ) class CurrencyViewSet(TenantScopedMixin, viewsets.ModelViewSet): """Org-wide currencies used in finance forms.""" rbac_domain = "finance" queryset = Currency.objects.all() serializer_class = CurrencySerializer class BankAccountViewSet(viewsets.ModelViewSet): """Org bank accounts for receiving money. Filter by currency.""" rbac_domain = "finance" serializer_class = BankAccountSerializer def get_queryset(self): qs = BankAccount.objects.filter(organization=get_active_organization(self.request)) currency = self.request.query_params.get("currency") if currency: qs = qs.filter(currency=currency) return qs def perform_create(self, serializer): obj = serializer.save(organization=get_active_organization(self.request)) self._enforce_single_default(obj) def perform_update(self, serializer): obj = serializer.save() self._enforce_single_default(obj) @staticmethod def _enforce_single_default(obj): # One default per currency, scoped to the same org. if obj.is_default: BankAccount.objects.exclude(pk=obj.pk).filter( organization=obj.organization, currency=obj.currency, is_default=True, ).update(is_default=False) def destroy(self, request, *args, **kwargs): account = self.get_object() account.is_active = False account.save(update_fields=["is_active", "updated_at"]) return Response(status=204) class BankStatementImportViewSet(TenantScopedMixin, viewsets.ModelViewSet): rbac_domain = "finance" rbac_action_map = {"confirm": "approve", "close": "approve"} serializer_class = BankStatementImportSerializer queryset = BankStatementImport.objects.select_related("bank_account", "uploaded_by", "confirmed_by") http_method_names = ["get", "post", "head", "options"] def perform_create(self, serializer): method = serializer.validated_data["import_method"] obj = serializer.save(uploaded_by=self.request.user, status="needs_review", organization=get_active_organization(self.request)) if method == "ocr_pdf": from .ocr import extract_bank_statement_pdf try: extraction, confidence = extract_bank_statement_pdf(obj.statement_file.path) except (NotImplementedError, AttributeError): extraction, confidence = ({"lines": [], "warnings": ["OCR requires local file storage."], "provider_status": "unavailable"}, None) obj.extraction, obj.confidence = extraction, confidence obj.save(update_fields=["extraction", "confidence"]) @action(detail=True, methods=["post"]) def confirm(self, request, pk=None): obj = self.get_object() if obj.status != "needs_review": raise ValidationError("Only a reviewed extraction can be confirmed.") extraction = request.data.get("extraction") if not isinstance(extraction, dict) or not isinstance(extraction.get("lines"), list): raise ValidationError({"extraction": "Confirmed extraction must contain a lines array."}) obj.extraction = extraction obj.status = "confirmed" obj.confirmed_by = request.user obj.confirmed_at = timezone.now() obj.save(update_fields=["extraction", "status", "confirmed_by", "confirmed_at"]) return Response(self.get_serializer(obj).data) @action(detail=True, methods=["post"]) def close(self, request, pk=None): obj = self.get_object() if obj.status != "confirmed": raise ValidationError("Confirm all reconciliation lines before closing the statement.") obj.status = "closed" obj.save(update_fields=["status"]) return Response(self.get_serializer(obj).data) class TaxRateViewSet(TenantScopedMixin, viewsets.ModelViewSet): """Org-wide tax rates (percentages) used on invoices.""" rbac_domain = "finance" queryset = TaxRate.objects.all() serializer_class = TaxRateSerializer def perform_create(self, serializer): obj = serializer.save(organization=get_active_organization(self.request)) self._enforce_single_default(obj) def perform_update(self, serializer): obj = serializer.save() self._enforce_single_default(obj) @staticmethod def _enforce_single_default(obj): # Only one rate may be the default per org; unset the others. if obj.is_default: TaxRate.objects.exclude(pk=obj.pk).filter(organization=obj.organization, is_default=True).update(is_default=False) class BudgetCategoryViewSet(TenantScopedMixin, viewsets.ModelViewSet): """Org-wide budget categories.""" rbac_domain = "finance" # item_count_anno feeds the serializer; counting in SQL beats prefetching # every item row just to len() it. queryset = BudgetCategory.objects.annotate(item_count_anno=Count("items")) serializer_class = BudgetCategorySerializer def perform_create(self, serializer): # Auto-assign the next free letter code (A, B, C, …), per org. org = get_active_organization(self.request) n = BudgetCategory.objects.filter(organization=org).count() while True: n += 1 code = to_letters(n) if not BudgetCategory.objects.filter(organization=org, code=code).exists(): break serializer.save(code=code, organization=org) class BudgetItemViewSet(TenantScopedMixin, viewsets.ModelViewSet): """Budget line items, optionally filtered by category.""" rbac_domain = "finance" organization_lookup = "category__organization" queryset = BudgetItem.objects.select_related("category").all() serializer_class = BudgetItemSerializer def perform_create(self, serializer): # Code is . at the top level (B.1, B.2), or # . when nested under a parent item (B.1.1, B.1.2). category = serializer.validated_data["category"] org = get_active_organization(self.request) if category.organization_id != org.id: raise PermissionDenied("The budget category is outside your organization.") parent = serializer.validated_data.get("parent") if parent is not None: prefix = parent.code siblings = parent.children.count() else: prefix = category.code siblings = category.items.filter(parent__isnull=True).count() n = siblings while True: n += 1 code = f"{prefix}.{n}" if not BudgetItem.objects.filter(category=category, code=code).exists(): break serializer.save(code=code) @extend_schema( parameters=[OpenApiParameter(name="category", type=str, description="Filter by category ID")] ) def list(self, request, *args, **kwargs): qs = self.get_queryset() category = request.query_params.get("category") if category: qs = qs.filter(category_id=category) serializer = self.get_serializer(qs, many=True) return Response(serializer.data) class InvoiceViewSet(viewsets.ModelViewSet): """Receivable invoices we issue to donors. Filter by status/donor/project.""" rbac_domain = "finance" serializer_class = InvoiceSerializer rbac_action_map = {"approve": "approve", "issue": "update", "void": "approve"} def get_queryset(self): qs = ( Invoice.objects.filter(organization=get_active_organization(self.request)) .select_related("donor", "project", "event") .prefetch_related("lines") ) params = self.request.query_params for field in ("status", "donor", "project", "event"): val = params.get(field) if val: qs = qs.filter(**{field if field == "status" else f"{field}_id": val}) return qs def perform_create(self, serializer): # Auto-assign a sequential invoice number: INV-YYYY-NNNN, per org. from django.utils import timezone org = get_active_organization(self.request) year = timezone.now().year prefix = f"INV-{year}-" last = ( Invoice.objects.filter(organization=org, invoice_number__startswith=prefix) .order_by("-invoice_number") .values_list("invoice_number", flat=True) .first() ) seq = (int(last.rsplit("-", 1)[-1]) + 1) if last else 1 serializer.save(invoice_number=f"{prefix}{seq:04d}", created_by=self.request.user, status="draft", organization=org) def update(self, request, *args, **kwargs): invoice = self.get_object() if "status" in request.data: raise ValidationError({"status": "Status changes must use a lifecycle action."}) if invoice.status != "draft": raise ValidationError("An approved or issued invoice is immutable; use void or credit note.") return super().update(request, *args, **kwargs) def destroy(self, request, *args, **kwargs): raise MethodNotAllowed("DELETE", detail="Customer invoices are voided or credited, never deleted after creation.") @action(detail=True, methods=["post"]) def approve(self, request, pk=None): invoice = self.get_object() if invoice.status != "draft": raise ValidationError("Only a draft invoice can be approved.") if invoice.created_by_id == request.user.id: raise PermissionDenied("The invoice creator cannot approve the same invoice.") if not invoice.lines.exists() or invoice.total <= 0: raise ValidationError("Invoice approval requires at least one positive line item.") invoice.status, invoice.approved_by, invoice.approved_at = "approved", request.user, timezone.now() invoice.save(update_fields=["status", "approved_by", "approved_at", "updated_at"]) return Response(self.get_serializer(invoice).data) @action(detail=True, methods=["post"]) def issue(self, request, pk=None): invoice = self.get_object() if invoice.status != "approved": raise ValidationError("Only an approved invoice can be issued.") invoice.status, invoice.issued_by, invoice.issue_date = "issued", request.user, invoice.issue_date or timezone.localdate() invoice.save(update_fields=["status", "issued_by", "issue_date", "updated_at"]) return Response(self.get_serializer(invoice).data) @action(detail=True, methods=["post"]) def void(self, request, pk=None): invoice = self.get_object(); reason=(request.data.get("reason") or "").strip() if invoice.status in ("paid", "void"): raise ValidationError("A paid or void invoice cannot be voided.") if not reason: raise ValidationError({"reason": "A void reason is required."}) invoice.status, invoice.voided_by, invoice.void_reason = "void", request.user, reason invoice.save(update_fields=["status", "voided_by", "void_reason", "updated_at"]) return Response(self.get_serializer(invoice).data)