forked from erp-dev/erp
326 lines
12 KiB
Python
326 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from decimal import Decimal, ROUND_HALF_UP
|
|
from typing import Iterable, List
|
|
|
|
from rest_framework import views
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from rest_framework.response import Response
|
|
|
|
from basic_info import models as basic_models
|
|
from business import models as business_models
|
|
from api_v1.views.stock_change_views.mixins import StockChangeViewMixin
|
|
from . import serializers as statement_serializers
|
|
|
|
TWO_PLACES = Decimal('0.01')
|
|
ZERO = Decimal('0')
|
|
|
|
|
|
class StatementViewBase(StockChangeViewMixin, views.APIView):
|
|
"""
|
|
提供对账单视图的公共实现：权限校验、序列化上下文及记录构建方法。
|
|
"""
|
|
|
|
permission_classes = [IsAuthenticated]
|
|
serializer_class = statement_serializers.StatementResponseSerializer
|
|
|
|
def get_serializer_context(self):
|
|
return {
|
|
'request': self.request,
|
|
'view': self,
|
|
'summary_builder': self._build_summary,
|
|
}
|
|
|
|
def _build_response(self, counterparty, records: List[dict]):
|
|
serializer = self.serializer_class(
|
|
instance={
|
|
'counterparty': counterparty.id,
|
|
'counterparty_name': counterparty.name,
|
|
'records': records,
|
|
},
|
|
context=self.get_serializer_context(),
|
|
)
|
|
return Response(serializer.data)
|
|
|
|
def _build_record(
|
|
self,
|
|
*,
|
|
counterparty_id: int,
|
|
counterparty_name: str,
|
|
source_type: str,
|
|
source_label: str,
|
|
source_id: int,
|
|
occurred_at,
|
|
recorded_at,
|
|
status: int,
|
|
status_label: str,
|
|
positive_amount,
|
|
negative_amount,
|
|
extra: dict | None = None,
|
|
) -> dict:
|
|
record = {
|
|
'counterparty': counterparty_id,
|
|
'counterparty_name': counterparty_name,
|
|
'source_type': source_type,
|
|
'source_label': source_label,
|
|
'source_id': source_id,
|
|
'occurred_at': occurred_at,
|
|
'recorded_at': recorded_at,
|
|
'status': status,
|
|
'status_label': status_label,
|
|
'positive_amount': self._normalize_amount(positive_amount),
|
|
'negative_amount': self._normalize_amount(negative_amount),
|
|
}
|
|
if extra:
|
|
record['extra'] = extra
|
|
return record
|
|
|
|
def _sort_records(self, records: Iterable[dict]) -> List[dict]:
|
|
return sorted(
|
|
records,
|
|
key=lambda item: (item['occurred_at'], item['recorded_at'], item['source_id']),
|
|
reverse=True,
|
|
)
|
|
|
|
def _normalize_amount(self, value) -> Decimal:
|
|
if isinstance(value, Decimal):
|
|
decimal_value = value
|
|
else:
|
|
decimal_value = Decimal(str(value))
|
|
return decimal_value.quantize(TWO_PLACES, rounding=ROUND_HALF_UP)
|
|
|
|
def _decimal_to_string(self, value: Decimal) -> str:
|
|
normalized = self._normalize_amount(value)
|
|
return format(normalized, 'f')
|
|
|
|
def _build_summary(self, payload: dict):
|
|
records = payload.get('records', [])
|
|
total_positive = sum((record.get('positive_amount', ZERO) for record in records), ZERO)
|
|
total_negative = sum((record.get('negative_amount', ZERO) for record in records), ZERO)
|
|
return {
|
|
'positive_total': self._decimal_to_string(total_positive),
|
|
'negative_total': self._decimal_to_string(total_negative),
|
|
}
|
|
|
|
|
|
class CustomerStatementView(StatementViewBase):
|
|
"""
|
|
返回指定客户的业务单据对账单，包含销售单、销售退货单与收款单。
|
|
"""
|
|
|
|
def get(self, request, customer_id: int):
|
|
if not self.check_employee_permission(request):
|
|
return self.permission_error_response('无权限访问')
|
|
|
|
merchant = request.user.employee.merchant
|
|
try:
|
|
customer = basic_models.Customer.objects.get(id=customer_id, merchant=merchant)
|
|
except basic_models.Customer.DoesNotExist:
|
|
return self.not_found_response('客户不存在')
|
|
|
|
records = self._collect_customer_records(merchant, customer)
|
|
return self._build_response(customer, records)
|
|
|
|
def _collect_customer_records(self, merchant, customer) -> List[dict]:
|
|
records: List[dict] = []
|
|
records.extend(self._build_sales_records(merchant, customer))
|
|
records.extend(self._build_sales_return_records(merchant, customer))
|
|
records.extend(self._build_receipt_records(merchant, customer))
|
|
return self._sort_records(records)
|
|
|
|
def _build_sales_records(self, merchant, customer) -> List[dict]:
|
|
qs = (
|
|
business_models.SalesOrder.objects.filter(
|
|
merchant=merchant,
|
|
customer=customer,
|
|
status=business_models.SalesOrderStatusEnum.APPROVED,
|
|
)
|
|
.select_related('customer')
|
|
)
|
|
records = []
|
|
for order in qs:
|
|
records.append(
|
|
self._build_record(
|
|
counterparty_id=order.customer_id,
|
|
counterparty_name=order.customer.name,
|
|
source_type='sales_order',
|
|
source_label='销售单',
|
|
source_id=order.id,
|
|
occurred_at=order.sales_date,
|
|
recorded_at=order.created_at,
|
|
status=order.status,
|
|
status_label=order.get_status_display(),
|
|
positive_amount=order.get_total_amount(),
|
|
negative_amount=ZERO,
|
|
)
|
|
)
|
|
return records
|
|
|
|
def _build_sales_return_records(self, merchant, customer) -> List[dict]:
|
|
qs = (
|
|
business_models.SalesReturnOrder.objects.filter(
|
|
merchant=merchant,
|
|
customer=customer,
|
|
status=business_models.SalesReturnStatusEnum.APPROVED,
|
|
)
|
|
.select_related('customer')
|
|
)
|
|
records = []
|
|
for order in qs:
|
|
records.append(
|
|
self._build_record(
|
|
counterparty_id=order.customer_id,
|
|
counterparty_name=order.customer.name,
|
|
source_type='sales_return_order',
|
|
source_label='销售退货单',
|
|
source_id=order.id,
|
|
occurred_at=order.return_date,
|
|
recorded_at=order.created_at,
|
|
status=order.status,
|
|
status_label=order.get_status_display(),
|
|
positive_amount=ZERO,
|
|
negative_amount=order.get_total_amount(),
|
|
)
|
|
)
|
|
return records
|
|
|
|
def _build_receipt_records(self, merchant, customer) -> List[dict]:
|
|
qs = (
|
|
business_models.ReceiptOrder.objects.filter(
|
|
merchant=merchant,
|
|
customer=customer,
|
|
status=business_models.ReceiptOrderStatusEnum.APPROVED,
|
|
)
|
|
.select_related('customer')
|
|
)
|
|
records = []
|
|
for order in qs:
|
|
records.append(
|
|
self._build_record(
|
|
counterparty_id=order.customer_id,
|
|
counterparty_name=order.customer.name,
|
|
source_type='receipt_order',
|
|
source_label='收款单',
|
|
source_id=order.id,
|
|
occurred_at=order.receipt_date,
|
|
recorded_at=order.created_at,
|
|
status=order.status,
|
|
status_label=order.get_status_display(),
|
|
positive_amount=ZERO,
|
|
negative_amount=order.get_total_amount(),
|
|
)
|
|
)
|
|
return records
|
|
|
|
|
|
class SupplierStatementView(StatementViewBase):
|
|
"""
|
|
返回指定供应商的业务单据对账单，包含采购单、采购退货单与付款单。
|
|
"""
|
|
|
|
def get(self, request, supplier_id: int):
|
|
if not self.check_employee_permission(request):
|
|
return self.permission_error_response('无权限访问')
|
|
|
|
merchant = request.user.employee.merchant
|
|
try:
|
|
supplier = basic_models.Supplier.objects.get(id=supplier_id, merchant=merchant)
|
|
except basic_models.Supplier.DoesNotExist:
|
|
return self.not_found_response('供应商不存在')
|
|
|
|
records = self._collect_supplier_records(merchant, supplier)
|
|
return self._build_response(supplier, records)
|
|
|
|
def _collect_supplier_records(self, merchant, supplier) -> List[dict]:
|
|
records: List[dict] = []
|
|
records.extend(self._build_purchase_records(merchant, supplier))
|
|
records.extend(self._build_purchase_return_records(merchant, supplier))
|
|
records.extend(self._build_payment_records(merchant, supplier))
|
|
return self._sort_records(records)
|
|
|
|
def _build_purchase_records(self, merchant, supplier) -> List[dict]:
|
|
qs = (
|
|
business_models.PurchaseOrder.objects.filter(
|
|
merchant=merchant,
|
|
supplier=supplier,
|
|
status=business_models.PurchaseOrderStatusEnum.APPROVED,
|
|
)
|
|
.select_related('supplier')
|
|
)
|
|
records = []
|
|
for order in qs:
|
|
records.append(
|
|
self._build_record(
|
|
counterparty_id=order.supplier_id,
|
|
counterparty_name=order.supplier.name,
|
|
source_type='purchase_order',
|
|
source_label='采购单',
|
|
source_id=order.id,
|
|
occurred_at=order.purchase_date,
|
|
recorded_at=order.created_at,
|
|
status=order.status,
|
|
status_label=order.get_status_display(),
|
|
positive_amount=order.get_total_amount(),
|
|
negative_amount=ZERO,
|
|
)
|
|
)
|
|
return records
|
|
|
|
def _build_purchase_return_records(self, merchant, supplier) -> List[dict]:
|
|
qs = (
|
|
business_models.PurchaseReturnOrder.objects.filter(
|
|
merchant=merchant,
|
|
supplier=supplier,
|
|
status=business_models.PurchaseReturnStatusEnum.APPROVED,
|
|
)
|
|
.select_related('supplier')
|
|
)
|
|
records = []
|
|
for order in qs:
|
|
records.append(
|
|
self._build_record(
|
|
counterparty_id=order.supplier_id,
|
|
counterparty_name=order.supplier.name,
|
|
source_type='purchase_return_order',
|
|
source_label='采购退货单',
|
|
source_id=order.id,
|
|
occurred_at=order.return_date,
|
|
recorded_at=order.created_at,
|
|
status=order.status,
|
|
status_label=order.get_status_display(),
|
|
positive_amount=ZERO,
|
|
negative_amount=order.get_total_amount(),
|
|
)
|
|
)
|
|
return records
|
|
|
|
def _build_payment_records(self, merchant, supplier) -> List[dict]:
|
|
qs = (
|
|
business_models.PaymentOrder.objects.filter(
|
|
merchant=merchant,
|
|
supplier=supplier,
|
|
status=business_models.PaymentOrderStatusEnum.APPROVED,
|
|
)
|
|
.select_related('supplier')
|
|
)
|
|
records = []
|
|
for order in qs:
|
|
records.append(
|
|
self._build_record(
|
|
counterparty_id=order.supplier_id,
|
|
counterparty_name=order.supplier.name,
|
|
source_type='payment_order',
|
|
source_label='付款单',
|
|
source_id=order.id,
|
|
occurred_at=order.payment_date,
|
|
recorded_at=order.created_at,
|
|
status=order.status,
|
|
status_label=order.get_status_display(),
|
|
positive_amount=ZERO,
|
|
negative_amount=order.get_total_amount(),
|
|
)
|
|
)
|
|
return records
|
|
|
|
|