zuowei/erpnew
1
0
Fork 0
forked from erp-dev/erp
Code Pull Requests Activity

feat: shipment customer aou

Browse Source
This commit is contained in:
colaftc
2026-03-30 22:03:14 +08:00
parent 38f40644a3
commit aff10b8925
7 changed files with 690 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
api_v1/urls.py
View File

@@ -33,8 +33,10 @@ from .views.parameters import StateParameterViewSet
from .views.users import CreateUserWithProfileView
from .views.mingdaoyun import MDYPlateOrderStagingViewSet
from .views.shipment import (
SalesItemByCustomerView,
SalesItemByPrintingOrderView,
SalesItemCreateView,
ShipmentSalesItemCustomerListView,
ShipmentListCreateView,
ShipmentDetailView,
ShipmentExternalCreateView,
@@ -305,6 +307,16 @@ urlpatterns = [
ShipmentExternalCreateView.as_view(),
name="shipment_create_external",
),
path(
"shipment/sales-items/customers/",
ShipmentSalesItemCustomerListView.as_view(),
name="sales_item_customers",
),
path(
"shipment/sales-items/by-customer/<int:customer_id>/",
SalesItemByCustomerView.as_view(),
name="sales_items_by_customer",
),
path(
"shipment/sales-items/by-printing-order/<int:printing_order_id>/",
SalesItemByPrintingOrderView.as_view(),

4
api_v1/views/shipment/__init__.py
View File

@@ -4,16 +4,20 @@ Shipment API 模块
提供出货单和销售品相关的 API 接口
"""
from .views import (
SalesItemByCustomerView,
SalesItemByPrintingOrderView,
SalesItemCreateView,
ShipmentSalesItemCustomerListView,
ShipmentListCreateView,
ShipmentDetailView,
ShipmentExternalCreateView,
)
__all__ = [
'SalesItemByCustomerView',
'SalesItemByPrintingOrderView',
'SalesItemCreateView',
'ShipmentSalesItemCustomerListView',
'ShipmentListCreateView',
'ShipmentDetailView',
'ShipmentExternalCreateView',

32
api_v1/views/shipment/serializers.py
View File

@@ -248,7 +248,9 @@ class SalesItemSerializer(serializers.Serializer):
position = serializers.CharField(read_only=True)
remark = serializers.CharField(read_only=True)
printing_job_id = serializers.IntegerField(read_only=True)
printing_order_id = serializers.SerializerMethodField()
customer_id = serializers.IntegerField(read_only=True)
customer_name = serializers.SerializerMethodField()
shipment_id = serializers.IntegerField(
source="shipment.id", read_only=True, allow_null=True
)
@@ -264,6 +266,24 @@ class SalesItemSerializer(serializers.Serializer):
def get_unit_display(self, obj):
return obj.get_unit_display()
def get_printing_order_id(self, obj):
printing_order_map = self.context.get("printing_order_map", {})
if obj.printing_job_id in printing_order_map:
return printing_order_map[obj.printing_job_id]
printing_job = obj.get_printing_job()
if printing_job and getattr(printing_job, "printing_order_id", None):
return printing_job.printing_order_id
return None
def get_customer_name(self, obj):
customer_name_map = self.context.get("customer_name_map", {})
if obj.customer_id in customer_name_map:
return customer_name_map[obj.customer_id]
customer = obj.get_customer()
return customer.name if customer else None
def get_created_by_name(self, obj):
if obj.created_by:
employee = getattr(obj.created_by, "employee", None)
@@ -273,6 +293,18 @@ class SalesItemSerializer(serializers.Serializer):
return None
class ShipmentSalesItemCustomerSerializer(serializers.Serializer):
"""
由未出货销售品反推的客户摘要序列化器。
"""
customer_id = serializers.IntegerField(source="id", read_only=True)
customer_name = serializers.CharField(source="name", read_only=True)
mobile = serializers.CharField(read_only=True, allow_null=True)
area = serializers.CharField(read_only=True, allow_null=True)
unshipped_sales_items_count = serializers.IntegerField(read_only=True)
class SalesItemCreateSerializer(serializers.Serializer):
"""
销售品创建序列化器

307
api_v1/views/shipment/test_api.py
View File

@@ -6,7 +6,7 @@ from decimal import Decimal
from django.test import TestCase
from django.conf import settings
from rest_framework.test import APIClient
from rest_framework.test import APIClient, APITestCase
from rest_framework import status
from django.contrib.auth import get_user_model
@@ -279,6 +279,311 @@ class SalesItemByPrintingOrderAPITestCase(TestCase):
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
class ShipmentSalesItemCustomersAPITestCase(TestCase):
"""测试由未出货销售品汇总客户 API"""
def setUp(self):
self.client = APIClient()
self.merchant = basic_models.Merchant.objects.create(
name="测试印花厂", type=basic_models.MerchantTypeEnum.FACTORY
)
self.other_merchant = basic_models.Merchant.objects.create(
name="其它商户", type=basic_models.MerchantTypeEnum.FACTORY
)
self.user = User.objects.create_user(
username="shipment_customer_user",
password="testpass123",
email="shipment_customer@example.com",
)
self.employee = basic_models.Employee.objects.create(
sys_user=self.user,
merchant=self.merchant,
name="测试员工",
mobile="13800138020",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
self.customer1 = basic_models.Customer.objects.create(
merchant=self.merchant,
name="客户A",
mobile="13900139020",
area="杭州",
)
self.customer2 = basic_models.Customer.objects.create(
merchant=self.merchant,
name="客户B",
mobile="13900139021",
area="绍兴",
)
self.other_customer = basic_models.Customer.objects.create(
merchant=self.other_merchant,
name="客户C",
mobile="13900139022",
area="苏州",
)
shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="客户A销售品1",
quantity=Decimal("10.00"),
unit=shipment_models.UnitChoices.METER,
customer_id=self.customer1.id,
created_by=self.user,
)
shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="客户A销售品2",
quantity=Decimal("20.00"),
unit=shipment_models.UnitChoices.METER,
customer_id=self.customer1.id,
created_by=self.user,
)
shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="客户B销售品1",
quantity=Decimal("30.00"),
unit=shipment_models.UnitChoices.METER,
customer_id=self.customer2.id,
created_by=self.user,
)
shipped = shipment_models.Shipment.objects.create(
merchant=self.merchant,
customer=self.customer1,
shipment_date="2026-01-14",
created_by=self.user,
)
shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="客户A已出货销售品",
quantity=Decimal("40.00"),
unit=shipment_models.UnitChoices.METER,
customer_id=self.customer1.id,
shipment=shipped,
created_by=self.user,
)
shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="无客户销售品",
quantity=Decimal("50.00"),
unit=shipment_models.UnitChoices.METER,
created_by=self.user,
)
shipment_models.SalesItem.objects.create(
merchant=self.other_merchant,
name="其它商户销售品",
quantity=Decimal("60.00"),
unit=shipment_models.UnitChoices.METER,
customer_id=self.other_customer.id,
created_by=self.user,
)
self.client.force_authenticate(user=self.user)
def test_list_customers_with_unshipped_sales_items(self):
resp = self.client.get("/api/v1/shipment/sales-items/customers/")
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 2)
self.assertIn("next", data)
self.assertIn("previous", data)
by_customer_id = {
item["customer_id"]: item
for item in data["results"]
}
self.assertEqual(set(by_customer_id.keys()), {self.customer1.id, self.customer2.id})
self.assertEqual(by_customer_id[self.customer1.id]["customer_name"], "客户A")
self.assertEqual(by_customer_id[self.customer1.id]["unshipped_sales_items_count"], 2)
self.assertEqual(by_customer_id[self.customer2.id]["unshipped_sales_items_count"], 1)
def test_list_customers_with_unshipped_sales_items_supports_limit_offset(self):
resp = self.client.get("/api/v1/shipment/sales-items/customers/?limit=1&offset=1")
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 2)
self.assertEqual(len(data["results"]), 1)
self.assertEqual(data["results"][0]["customer_id"], self.customer2.id)
def test_list_customers_with_unshipped_sales_items_unauthenticated(self):
self.client.logout()
resp = self.client.get("/api/v1/shipment/sales-items/customers/")
self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED)
class SalesItemByCustomerAPITestCase(TestCase):
"""测试按客户查询销售品 API"""
def setUp(self):
self.client = APIClient()
self.merchant = basic_models.Merchant.objects.create(
name="测试印花厂", type=basic_models.MerchantTypeEnum.FACTORY
)
self.other_merchant = basic_models.Merchant.objects.create(
name="其它印花厂", type=basic_models.MerchantTypeEnum.FACTORY
)
self.user = User.objects.create_user(
username="sales_item_by_customer_user",
password="testpass123",
email="sales_item_by_customer@example.com",
)
self.employee = basic_models.Employee.objects.create(
sys_user=self.user,
merchant=self.merchant,
name="测试员工",
mobile="13800138030",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
self.customer = basic_models.Customer.objects.create(
merchant=self.merchant,
name="测试客户",
mobile="13900139030",
area="杭州",
)
self.other_customer = basic_models.Customer.objects.create(
merchant=self.merchant,
name="其它客户",
mobile="13900139031",
area="绍兴",
)
self.foreign_customer = basic_models.Customer.objects.create(
merchant=self.other_merchant,
name="外部客户",
mobile="13900139032",
area="苏州",
)
self.state = stateflow_models.State.objects.create(name="待生产")
self.process = stateflow_models.Process.objects.create(name="生产流程")
self.process.replace_nodes([self.state])
self.category = basic_models.ProductCategory.objects.create(
merchant=self.merchant,
name="测试分类",
)
self.product = basic_models.Product.objects.create(
merchant=self.merchant,
category=self.category,
name="测试产品",
human_id="PROD001",
)
self.printing_order = printing_models.PrintingOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
fabric="测试面料",
width="150cm",
process=self.process,
created_by=self.user,
)
self.printing_job = printing_models.PrintingJob.objects.create(
merchant=self.merchant,
printing_order=self.printing_order,
product=self.product,
quantity=100,
unit="米",
created_by=self.user,
)
self.sales_item1 = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="销售品1",
quantity=Decimal("10.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=self.printing_job.id,
customer_id=self.customer.id,
position="A1-01",
created_by=self.user,
)
self.sales_item2 = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="销售品2",
quantity=Decimal("20.00"),
unit=shipment_models.UnitChoices.PIECE,
printing_job_id=self.printing_job.id,
customer_id=self.customer.id,
created_by=self.user,
)
self.shipment = shipment_models.Shipment.objects.create(
merchant=self.merchant,
customer=self.customer,
shipment_date="2026-01-14",
created_by=self.user,
)
self.sales_item_shipped = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="已出货销售品",
quantity=Decimal("30.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=self.printing_job.id,
customer_id=self.customer.id,
shipment=self.shipment,
created_by=self.user,
)
shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="其它客户销售品",
quantity=Decimal("40.00"),
unit=shipment_models.UnitChoices.METER,
customer_id=self.other_customer.id,
created_by=self.user,
)
self.client.force_authenticate(user=self.user)
def test_get_sales_items_by_customer_exclude_shipped(self):
resp = self.client.get(
f"/api/v1/shipment/sales-items/by-customer/{self.customer.id}/"
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 2)
self.assertEqual(len(data["results"]), 2)
item = next(it for it in data["results"] if it["id"] == self.sales_item1.id)
self.assertEqual(item["customer_id"], self.customer.id)
self.assertEqual(item["customer_name"], self.customer.name)
self.assertEqual(item["printing_job_id"], self.printing_job.id)
self.assertEqual(item["printing_order_id"], self.printing_order.id)
self.assertEqual(item["position"], "A1-01")
def test_get_sales_items_by_customer_include_shipped_and_paginate(self):
resp = self.client.get(
f"/api/v1/shipment/sales-items/by-customer/{self.customer.id}/"
"?include_already_has_shipment=true&limit=2&offset=1"
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 3)
self.assertEqual(len(data["results"]), 2)
result_ids = [item["id"] for item in data["results"]]
self.assertEqual(result_ids, [self.sales_item2.id, self.sales_item_shipped.id])
def test_get_sales_items_by_customer_not_found(self):
resp = self.client.get("/api/v1/shipment/sales-items/by-customer/99999/")
self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
def test_get_sales_items_by_customer_other_merchant_404(self):
resp = self.client.get(
f"/api/v1/shipment/sales-items/by-customer/{self.foreign_customer.id}/"
)
self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
class ShipmentCreateAPITestCase(TestCase):
"""测试创建出货单 API"""

143
api_v1/views/shipment/views.py
View File

@@ -3,11 +3,11 @@ Shipment API ViewSet
"""
from rest_framework import status
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from rest_framework.generics import GenericAPIView
from rest_framework.mixins import ListModelMixin, RetrieveModelMixin
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from flower.viewsets import LimitedLimitOffsetPagination
from shipment.models import Shipment
@@ -17,10 +17,39 @@ from .serializers import (
ShipmentSerializer,
ShipmentCreateNormalSerializer,
ShipmentCreateExternalSerializer,
ShipmentSalesItemCustomerSerializer,
ShipmentUpdateSerializer,
)
def _build_sales_item_serializer_context(items):
customer_ids = {item.customer_id for item in items if item.customer_id}
printing_job_ids = {item.printing_job_id for item in items if item.printing_job_id}
customer_name_map = {}
if customer_ids:
from basic_info.models import Customer
customer_name_map = dict(
Customer.objects.filter(id__in=customer_ids).values_list("id", "name")
)
printing_order_map = {}
if printing_job_ids:
from printing.models import PrintingJob
printing_order_map = dict(
PrintingJob.objects.filter(id__in=printing_job_ids).values_list(
"id", "printing_order_id"
)
)
return {
"customer_name_map": customer_name_map,
"printing_order_map": printing_order_map,
}
class ShipmentListCreateView(ListModelMixin, GenericAPIView):
"""
出货单：查询列表 / 创建
@@ -259,6 +288,50 @@ class ShipmentExternalCreateView(APIView):
return Response(response_serializer.data, status=status.HTTP_201_CREATED)
class ShipmentSalesItemCustomerListView(GenericAPIView):
"""
从未出货销售品反推出“当前可出货客户”列表。
GET /api/v1/shipment/sales-items/customers/
"""
permission_classes = [IsAuthenticated]
serializer_class = ShipmentSalesItemCustomerSerializer
pagination_class = LimitedLimitOffsetPagination
def get_queryset(self):
user = self.request.user
if getattr(user, "is_superuser", False):
from basic_info.models import Customer, Merchant
merchant_id = self.request.query_params.get("merchant")
if not merchant_id:
return Customer.objects.none()
try:
merchant = Merchant.objects.get(id=merchant_id)
except Merchant.DoesNotExist:
return Customer.objects.none()
else:
from basic_info.models import Customer
emp = getattr(user, "employee", None)
merchant = getattr(emp, "merchant", None) if emp else None
if not merchant:
return Customer.objects.none()
from shipment.services import get_customers_with_unshipped_sales_items
return get_customers_with_unshipped_sales_items(merchant=merchant)
def get(self, request):
queryset = self.get_queryset()
page = self.paginate_queryset(queryset)
serializer = self.get_serializer(page if page is not None else queryset, many=True)
if page is not None:
return self.get_paginated_response(serializer.data)
return Response(serializer.data)
class SalesItemByPrintingOrderView(APIView):
"""
通过生产订单查询销售品
@@ -324,11 +397,73 @@ class SalesItemByPrintingOrderView(APIView):
)
# 序列化返回
serializer = SalesItemSerializer(sales_items, many=True)
items = list(sales_items)
serializer = SalesItemSerializer(
items,
many=True,
context=_build_sales_item_serializer_context(items),
)
return Response({"count": len(serializer.data), "results": serializer.data})
class SalesItemByCustomerView(GenericAPIView):
"""
通过客户查询销售品。
GET /api/v1/shipment/sales-items/by-customer/<customer_id>/
"""
permission_classes = [IsAuthenticated]
serializer_class = SalesItemSerializer
pagination_class = LimitedLimitOffsetPagination
def get(self, request, customer_id: int):
from basic_info.models import Customer
user = request.user
if getattr(user, "is_superuser", False):
customer = Customer.objects.filter(id=customer_id).first()
else:
emp = getattr(user, "employee", None)
merchant = getattr(emp, "merchant", None) if emp else None
if not merchant:
return Response(
{"detail": f"客户 {customer_id} 不存在"},
status=status.HTTP_404_NOT_FOUND,
)
customer = Customer.objects.filter(id=customer_id, merchant=merchant).first()
if customer is None:
return Response(
{"detail": f"客户 {customer_id} 不存在"},
status=status.HTTP_404_NOT_FOUND,
)
include_already_has_shipment = (
request.query_params.get("include_already_has_shipment", "false").lower()
== "true"
)
from shipment.services import get_sales_items_by_customer
queryset = get_sales_items_by_customer(
merchant=customer.merchant,
customer_id=customer.id,
include_already_has_shipment=include_already_has_shipment,
)
page = self.paginate_queryset(queryset)
items = list(page) if page is not None else list(queryset)
serializer = self.get_serializer(
items,
many=True,
context=_build_sales_item_serializer_context(items),
)
if page is not None:
return self.get_paginated_response(serializer.data)
return Response(serializer.data)
class SalesItemCreateView(APIView):
"""
手动创建销售品