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

feat: added pre sales order and pre purchase order models and crud

Browse Source
This commit is contained in:
colaftc
2026-01-31 23:26:01 +08:00
parent a093174913
commit f643b643b7
12 changed files with 1008 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

122
api_v1/test_pre_purchase_order_api.py Normal file
View File

@@ -0,0 +1,122 @@
import copy
from django.contrib.auth import get_user_model
from django.test import TestCase, override_settings
from rest_framework import status
from rest_framework.test import APIClient
from basic_info.models import (
Merchant,
MerchantTypeEnum,
Supplier,
WareHouse,
WareHouseModeEnum,
ProductCategory,
Product,
ProductUnitEnum,
Employee,
EmployeeStatusEnum,
)
from business import models as business_models
@override_settings(
CELERY_TASK_ALWAYS_EAGER=True,
CELERY_TASK_EAGER_PROPAGATES=True,
)
class PrePurchaseOrderAPITestCase(TestCase):
def setUp(self):
self.merchant = Merchant.objects.create(name='预采购商户', type=MerchantTypeEnum.FACTORY)
self.supplier = Supplier.objects.create(
merchant=self.merchant,
name='预采购供应商',
)
self.warehouse = WareHouse.objects.create(
merchant=self.merchant,
name='预采购仓库',
mode=WareHouseModeEnum.UNRESTRICTED,
)
category = ProductCategory.objects.create(
merchant=self.merchant,
name='预采购品类',
product_prefix='PPR',
)
self.product = Product.objects.create(
merchant=self.merchant,
category=category,
name='预采购产品',
human_id='PPR-001',
unit=ProductUnitEnum.METER,
)
User = get_user_model()
self.user = User.objects.create_user(username='pre_purchase_user', password='pass123')
self.employee = Employee.objects.create(
merchant=self.merchant,
sys_user=self.user,
name='预采购员',
status=EmployeeStatusEnum.ACTIVE,
)
self.client = APIClient()
self.client.force_authenticate(user=self.user)
self.payload = {
'supplier': self.supplier.id,
'warehouse': self.warehouse.id,
'kind': business_models.PurchaseOrderKindEnum.WHOLESALE,
'remarks': '预采购单备注',
'items': [
{
'product_id': self.product.id,
'quantity': '12.5',
'unit': '米',
'order_quantity': 7,
'remarks': '明细备注',
}
],
}
def _create(self, payload=None) -> int:
resp = self.client.post('/api/v1/pre-purchase-orders/', payload or self.payload, format='json')
self.assertEqual(resp.status_code, status.HTTP_201_CREATED)
return resp.data['id']
def test_create_pre_purchase_order_success(self):
resp = self.client.post('/api/v1/pre-purchase-orders/', self.payload, format='json')
self.assertEqual(resp.status_code, status.HTTP_201_CREATED)
self.assertIn('id', resp.data)
self.assertIn('human_id', resp.data)
self.assertEqual(resp.data['supplier'], self.supplier.id)
self.assertEqual(resp.data['warehouse'], self.warehouse.id)
self.assertEqual(len(resp.data['items']), 1)
def test_list_pre_purchase_orders(self):
self._create()
resp = self.client.get('/api/v1/pre-purchase-orders/')
self.assertEqual(resp.status_code, status.HTTP_200_OK)
self.assertIn('results', resp.data)
self.assertGreaterEqual(len(resp.data['results']), 1)
def test_get_detail(self):
order_id = self._create()
resp = self.client.get(f'/api/v1/pre-purchase-orders/{order_id}/')
self.assertEqual(resp.status_code, status.HTTP_200_OK)
self.assertEqual(resp.data['id'], order_id)
self.assertIn('items', resp.data)
def test_update_pre_purchase_order(self):
order_id = self._create()
payload = copy.deepcopy(self.payload)
payload['remarks'] = '更新备注'
payload['items'][0]['quantity'] = '20'
resp = self.client.put(f'/api/v1/pre-purchase-orders/{order_id}/', payload, format='json')
self.assertEqual(resp.status_code, status.HTTP_200_OK)
self.assertEqual(resp.data['remarks'], '更新备注')
self.assertEqual(str(resp.data['items'][0]['quantity']), '20.00')
def test_delete_pre_purchase_order(self):
order_id = self._create()
resp = self.client.delete(f'/api/v1/pre-purchase-orders/{order_id}/')
self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
self.assertFalse(business_models.PrePurchaseOrder.objects.filter(id=order_id).exists())

32
api_v1/tests.py
View File

@@ -1542,6 +1542,13 @@ class PrintingJobWorkStateAPITestCase(TestCase):
job_id = resp.data['id']
detail = self.client.get(f'/api/v1/printing-jobs/{job_id}/')
# 客户可见性过滤：customer 未绑定 created_by/visible_employees 时，任务对该员工不可见 => 404
self.assertEqual(detail.status_code, status.HTTP_404_NOT_FOUND)
# 将客户显式授权给当前 employee 后应可见 => 200
self.customer.created_by = self.employee
self.customer.save(update_fields=['created_by'])
detail = self.client.get(f'/api/v1/printing-jobs/{job_id}/')
self.assertEqual(detail.status_code, status.HTTP_200_OK)
self.assertEqual(detail.data['work_state_display'], '待送货')
@@ -1558,10 +1565,29 @@ class PrintingJobWorkStateAPITestCase(TestCase):
{'work_state': printing_models.PrintingJobWorkStateEnum.FINISHED},
format='json',
)
# merchant 隔离：job.merchant 为空时，不属于当前员工商户 => 404
self.assertEqual(patch_resp.status_code, status.HTTP_404_NOT_FOUND)
# 创建一个显式属于当前商户的 job，并让 customer 对 employee 可见后应可更新/可查询
self.customer.created_by = self.employee
self.customer.save(update_fields=['created_by'])
visible_job = printing_models.PrintingJob.objects.create(
merchant=self.merchant,
printing_order=self.printing_order,
product=self.product,
quantity=5,
unit='米',
)
visible_url = f'/api/v1/printing-jobs/{visible_job.id}/'
patch_resp = self.client.patch(
visible_url,
{'work_state': printing_models.PrintingJobWorkStateEnum.FINISHED},
format='json',
)
self.assertEqual(patch_resp.status_code, status.HTTP_200_OK)
job.refresh_from_db()
self.assertEqual(job.work_state, printing_models.PrintingJobWorkStateEnum.FINISHED)
detail = self.client.get(url)
visible_job.refresh_from_db()
self.assertEqual(visible_job.work_state, printing_models.PrintingJobWorkStateEnum.FINISHED)
detail = self.client.get(visible_url)
self.assertEqual(detail.status_code, status.HTTP_200_OK)
self.assertEqual(detail.data['work_state_display'], '已完结')

3
api_v1/urls.py
View File

@@ -20,6 +20,7 @@ from .views.business.purchase_return import views as purchase_return_views
from .views.business.sales_return import views as sales_return_views
from .views.business.statements import views as statement_views
from .views.business.pre_sales import views as pre_sales_views
from .views.business.pre_purchase import views as pre_purchase_views
from .views.stock_change_views.snapshot import StockSnapshotListView
from .views.printing.views import PrintingOrderViewSet, PrintingJobViewSet, PlateOrderViewSet
from .views.upload import UploadFileViewSet
@@ -90,6 +91,8 @@ urlpatterns = [
path('sales-return-orders/<int:pk>/review/', sales_return_views.SalesReturnOrderReviewView.as_view(), name='sales_return_order_review'),
path('pre-sales-orders/', pre_sales_views.PreSalesOrderView.as_view(), name='pre_sales_orders'),
path('pre-sales-orders/<int:pk>/', pre_sales_views.PreSalesOrderDetailView.as_view(), name='pre_sales_order_detail'),
path('pre-purchase-orders/', pre_purchase_views.PrePurchaseOrderView.as_view(), name='pre_purchase_orders'),
path('pre-purchase-orders/<int:pk>/', pre_purchase_views.PrePurchaseOrderDetailView.as_view(), name='pre_purchase_order_detail'),
path('payment-orders/', payment_views.PaymentOrderView.as_view(), name='payment_orders'),
path('payment-orders/<int:pk>/review/', payment_views.PaymentOrderReviewView.as_view(), name='payment_order_review'),
path('receipt-orders/', receipt_views.ReceiptOrderView.as_view(), name='receipt_orders'),

0
api_v1/views/business/pre_purchase/__init__.py Normal file
View File

179
api_v1/views/business/pre_purchase/views.py Normal file
View File

@@ -0,0 +1,179 @@
from rest_framework import status, views, serializers, pagination
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from api_v1.views.stock_change_views.mixins import StockChangeViewMixin
from business import models as business_models
from business import pre_order_services
class PrePurchaseOrderItemSerializer(serializers.ModelSerializer):
class Meta:
model = business_models.PrePurchaseOrderItem
fields = [
'id',
'product_id',
'product_name',
'color',
'quantity',
'unit',
'spec',
'quantity_of_rolls',
'num_of_rolls',
'order_quantity',
'remarks',
'created_at',
'updated_at',
]
read_only_fields = ['id', 'created_at', 'updated_at']
class PrePurchaseOrderSerializer(serializers.ModelSerializer):
human_id = serializers.CharField(read_only=True)
merchant_name = serializers.CharField(source='merchant.name', read_only=True)
supplier_name = serializers.CharField(source='supplier.name', read_only=True)
warehouse_name = serializers.CharField(source='warehouse.name', read_only=True)
created_by_username = serializers.CharField(source='created_by.username', read_only=True)
operator_name = serializers.CharField(source='operator.name', read_only=True)
items = PrePurchaseOrderItemSerializer(many=True, read_only=True)
class Meta:
model = business_models.PrePurchaseOrder
fields = [
'id',
'human_id',
'merchant',
'merchant_name',
'supplier',
'supplier_name',
'warehouse',
'warehouse_name',
'created_by',
'created_by_username',
'operator',
'operator_name',
'kind',
'remarks',
'created_at',
'updated_at',
'items',
]
read_only_fields = [
'id',
'human_id',
'created_at',
'updated_at',
'merchant_name',
'supplier_name',
'warehouse_name',
'created_by_username',
'operator_name',
'items',
]
class PrePurchaseOrderPagination(pagination.LimitOffsetPagination):
default_limit = 20
max_limit = 100
class PrePurchaseOrderView(StockChangeViewMixin, views.APIView):
"""预采购单查询与创建"""
permission_classes = [IsAuthenticated]
pagination_class = PrePurchaseOrderPagination
def get(self, request):
if not self.check_employee_permission(request):
return self.permission_error_response('无权限访问')
merchant = request.user.employee.merchant
queryset = pre_order_services.list_pre_purchase_orders(merchant=merchant)
paginator = self.pagination_class()
page = paginator.paginate_queryset(queryset.order_by('-created_at'), request, view=self)
serializer = PrePurchaseOrderSerializer(page, many=True)
return paginator.get_paginated_response(serializer.data)
def post(self, request):
if not self.check_employee_permission(request):
return self.permission_error_response('无权限访问')
merchant = request.user.employee.merchant
employee = request.user.employee
data = request.data or {}
try:
pre_purchase_order = pre_order_services.create_pre_purchase_order(
merchant=merchant,
supplier_id=data.get('supplier') or data.get('supplier_id'),
warehouse_id=data.get('warehouse') or data.get('warehouse_id'),
operator=employee,
created_by=request.user,
kind=data.get('kind'),
items=data.get('items'),
remarks=data.get('remarks', ''),
)
except ValueError as exc:
return Response({'error': str(exc)}, status=status.HTTP_400_BAD_REQUEST)
return Response(PrePurchaseOrderSerializer(pre_purchase_order).data, status=status.HTTP_201_CREATED)
class PrePurchaseOrderDetailView(StockChangeViewMixin, views.APIView):
permission_classes = [IsAuthenticated]
def _get_order(self, request, pk: int):
if not self.check_employee_permission(request):
return None, self.permission_error_response('无权限访问')
merchant = request.user.employee.merchant
try:
order = pre_order_services.get_pre_purchase_order(
merchant=merchant,
pre_purchase_order_id=pk,
)
return order, None
except ValueError:
return None, self.not_found_response('预采购单不存在')
def get(self, request, pk: int):
order, error_response = self._get_order(request, pk)
if error_response:
return error_response
return Response(PrePurchaseOrderSerializer(order).data, status=status.HTTP_200_OK)
def put(self, request, pk: int):
return self._update(request, pk)
def patch(self, request, pk: int):
return self._update(request, pk)
def _update(self, request, pk: int):
order, error_response = self._get_order(request, pk)
if error_response:
return error_response
data = request.data or {}
try:
updated = pre_order_services.update_pre_purchase_order(
pre_purchase_order=order,
supplier_id=data.get('supplier') or data.get('supplier_id'),
warehouse_id=data.get('warehouse') or data.get('warehouse_id'),
kind=data.get('kind'),
items=data.get('items'),
remarks=data.get('remarks'),
)
except ValueError as exc:
return Response({'error': str(exc)}, status=status.HTTP_400_BAD_REQUEST)
return Response(PrePurchaseOrderSerializer(updated).data, status=status.HTTP_200_OK)
def delete(self, request, pk: int):
order, error_response = self._get_order(request, pk)
if error_response:
return error_response
pre_order_services.delete_pre_purchase_order(pre_purchase_order=order)
return Response(status=status.HTTP_204_NO_CONTENT)

13
api_v1/views/business/pre_sales/views.py
View File

@@ -33,7 +33,8 @@ class PreSalesOrderSerializer(serializers.ModelSerializer):
merchant_name = serializers.CharField(source='merchant.name', read_only=True)
customer_name = serializers.CharField(source='customer.name', read_only=True)
warehouse_name = serializers.CharField(source='warehouse.name', read_only=True)
created_by_name = serializers.CharField(source='created_by.name', read_only=True)
created_by_username = serializers.CharField(source='created_by.username', read_only=True)
operator_name = serializers.CharField(source='operator.name', read_only=True)
items = PreSalesOrderItemSerializer(many=True, read_only=True)
class Meta:
@@ -48,7 +49,9 @@ class PreSalesOrderSerializer(serializers.ModelSerializer):
'warehouse',
'warehouse_name',
'created_by',
'created_by_name',
'created_by_username',
'operator',
'operator_name',
'kind',
'remarks',
'created_at',
@@ -63,7 +66,8 @@ class PreSalesOrderSerializer(serializers.ModelSerializer):
'merchant_name',
'customer_name',
'warehouse_name',
'created_by_name',
'created_by_username',
'operator_name',
'items',
]
@@ -104,7 +108,8 @@ class PreSalesOrderView(StockChangeViewMixin, views.APIView):
merchant=merchant,
customer_id=data.get('customer') or data.get('customer_id'),
warehouse_id=data.get('warehouse') or data.get('warehouse_id'),
created_by=employee,
operator=employee,
created_by=request.user,
kind=data.get('kind'),
items=data.get('items'),
remarks=data.get('remarks', ''),

56
business/migrations/0023_prepurchaseorder_prepurchaseorderitem.py Normal file
View File

@@ -0,0 +1,56 @@
# Generated by Django 5.2.7 on 2026-01-31 14:27
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('basic_info', '0024_frontend_page_and_visible_pages'),
('business', '0022_presalesorder_presalesorderitem'),
]
operations = [
migrations.CreateModel(
name='PrePurchaseOrder',
fields=[
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')),
('updated_at', models.DateTimeField(auto_now=True, verbose_name='更新时间')),
('id', models.BigAutoField(primary_key=True, serialize=False)),
('kind', models.IntegerField(choices=[(1, '大货'), (2, '样板')], default=1, verbose_name='预采购单类型')),
('remarks', models.TextField(blank=True, null=True, verbose_name='备注')),
('created_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='created_pre_purchase_orders', to='basic_info.employee', verbose_name='创建者')),
('merchant', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='pre_purchase_orders', to='basic_info.merchant', verbose_name='所属商户')),
('supplier', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='pre_purchase_orders', to='basic_info.supplier', verbose_name='供应商')),
('warehouse', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='pre_purchase_orders', to='basic_info.warehouse', verbose_name='仓库')),
],
options={
'verbose_name': '预采购单',
'verbose_name_plural': '预采购单',
},
),
migrations.CreateModel(
name='PrePurchaseOrderItem',
fields=[
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')),
('updated_at', models.DateTimeField(auto_now=True, verbose_name='更新时间')),
('id', models.BigAutoField(primary_key=True, serialize=False)),
('product_id', models.BigIntegerField(blank=True, null=True, verbose_name='产品ID')),
('product_name', models.CharField(blank=True, default='', max_length=100, verbose_name='产品名称')),
('color', models.CharField(blank=True, max_length=50, null=True, verbose_name='颜色')),
('quantity', models.DecimalField(blank=True, decimal_places=2, max_digits=10, null=True, verbose_name='数量')),
('unit', models.CharField(blank=True, default='', max_length=50, verbose_name='单位')),
('spec', models.CharField(blank=True, max_length=100, null=True, verbose_name='规格')),
('quantity_of_rolls', models.TextField(blank=True, null=True, verbose_name='各条数数量')),
('num_of_rolls', models.PositiveIntegerField(default=1, verbose_name='条数')),
('order_quantity', models.IntegerField(blank=True, null=True, verbose_name='下单数量')),
('remarks', models.TextField(blank=True, null=True, verbose_name='备注')),
('pre_purchase_order', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='items', to='business.prepurchaseorder', verbose_name='预采购单')),
],
options={
'verbose_name': '预采购单明细',
'verbose_name_plural': '预采购单明细',
},
),
]

86
business/migrations/0024_prepurchaseorder_operator_presalesorder_operator_and_more.py Normal file
View File

@@ -0,0 +1,86 @@
# Generated by Django 5.2.7 on 2026-01-31 15:07
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
def forwards_copy_created_by_from_employee(apps, schema_editor):
PreSalesOrder = apps.get_model('business', 'PreSalesOrder')
PrePurchaseOrder = apps.get_model('business', 'PrePurchaseOrder')
for order in (
PreSalesOrder.objects.select_related('created_by_employee')
.filter(created_by__isnull=True)
.exclude(created_by_employee__isnull=True)
):
employee = getattr(order, 'created_by_employee', None)
if employee and getattr(employee, 'sys_user_id', None):
order.created_by_id = employee.sys_user_id
order.save(update_fields=['created_by'])
for order in (
PrePurchaseOrder.objects.select_related('created_by_employee')
.filter(created_by__isnull=True)
.exclude(created_by_employee__isnull=True)
):
employee = getattr(order, 'created_by_employee', None)
if employee and getattr(employee, 'sys_user_id', None):
order.created_by_id = employee.sys_user_id
order.save(update_fields=['created_by'])
def backwards_noop(apps, schema_editor):
# We intentionally do not attempt to restore legacy employee ids.
return
class Migration(migrations.Migration):
dependencies = [
('basic_info', '0024_frontend_page_and_visible_pages'),
('business', '0023_prepurchaseorder_prepurchaseorderitem'),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.RenameField(
model_name='prepurchaseorder',
old_name='created_by',
new_name='created_by_employee',
),
migrations.AddField(
model_name='prepurchaseorder',
name='operator',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='operated_pre_purchase_orders', to='basic_info.employee', verbose_name='经办人'),
),
migrations.AddField(
model_name='prepurchaseorder',
name='created_by',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='created_pre_purchase_orders', to=settings.AUTH_USER_MODEL, verbose_name='创建者用户'),
),
migrations.RenameField(
model_name='presalesorder',
old_name='created_by',
new_name='created_by_employee',
),
migrations.AddField(
model_name='presalesorder',
name='operator',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='operated_pre_sales_orders', to='basic_info.employee', verbose_name='经办人'),
),
migrations.AddField(
model_name='presalesorder',
name='created_by',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='created_pre_sales_orders', to=settings.AUTH_USER_MODEL, verbose_name='创建者用户'),
),
migrations.RunPython(forwards_copy_created_by_from_employee, backwards_noop),
migrations.RemoveField(
model_name='prepurchaseorder',
name='created_by_employee',
),
migrations.RemoveField(
model_name='presalesorder',
name='created_by_employee',
),
]

187
business/models.py
View File

@@ -3,6 +3,7 @@ from __future__ import annotations
from decimal import Decimal
from typing import List
from django.db import models
from django.conf import settings
from django.core.exceptions import ValidationError
from flower.common import ModelBase
from basic_info import models as basic_info_models
@@ -386,12 +387,20 @@ class PreSalesOrder(ModelBase):
verbose_name='客户',
)
created_by = models.ForeignKey(
basic_info_models.Employee,
on_delete=models.PROTECT,
related_name='created_pre_sales_orders',
settings.AUTH_USER_MODEL,
on_delete=models.SET_NULL,
null=True,
blank=True,
verbose_name='创建者',
related_name='created_pre_sales_orders',
verbose_name='创建者用户',
)
operator = models.ForeignKey(
basic_info_models.Employee,
on_delete=models.PROTECT,
related_name='operated_pre_sales_orders',
null=True,
blank=True,
verbose_name='经办人',
)
warehouse = models.ForeignKey(
basic_info_models.WareHouse,
@@ -422,9 +431,9 @@ class PreSalesOrder(ModelBase):
if self.warehouse.merchant_id != self.merchant_id:
errors['warehouse'] = '仓库所属商户与预销售单所属商户不一致'
if getattr(self, 'created_by_id', None) and getattr(self, 'merchant_id', None):
if self.created_by.merchant_id != self.merchant_id:
errors['created_by'] = '创建者所属商户与预销售单所属商户不一致'
if getattr(self, 'operator_id', None) and getattr(self, 'merchant_id', None):
if self.operator.merchant_id != self.merchant_id:
errors['operator'] = '经办人所属商户与预销售单所属商户不一致'
if errors:
raise ValidationError(errors)
@@ -442,9 +451,15 @@ class PreSalesOrder(ModelBase):
return ''
@property
def created_by_name(self) -> str:
def created_by_username(self) -> str:
if getattr(self, 'created_by_id', None):
return self.created_by.name
return getattr(self.created_by, 'username', '') or ''
return ''
@property
def operator_name(self) -> str:
if getattr(self, 'operator_id', None):
return self.operator.name
return ''
@property
@@ -501,6 +516,160 @@ class PreSalesOrderItem(ModelBase):
verbose_name_plural = '预销售单明细'
class PrePurchaseOrder(ModelBase):
"""预采购单
预采购单用于在向供应商下达正式采购单之前，用于沟通与确认货品、数量、仓库等信息。
当前阶段：仅定义 PrePurchaseOrder 与 PrePurchaseOrderItem 的一对多关系，
并关联 supplier / warehouse / created_by。
"""
id = models.BigAutoField(primary_key=True)
merchant = models.ForeignKey(
basic_info_models.Merchant,
on_delete=models.PROTECT,
related_name='pre_purchase_orders',
verbose_name='所属商户',
)
supplier = models.ForeignKey(
basic_info_models.Supplier,
on_delete=models.PROTECT,
related_name='pre_purchase_orders',
verbose_name='供应商',
)
created_by = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name='created_pre_purchase_orders',
verbose_name='创建者用户',
)
operator = models.ForeignKey(
basic_info_models.Employee,
on_delete=models.PROTECT,
related_name='operated_pre_purchase_orders',
null=True,
blank=True,
verbose_name='经办人',
)
warehouse = models.ForeignKey(
basic_info_models.WareHouse,
on_delete=models.PROTECT,
related_name='pre_purchase_orders',
verbose_name='仓库',
)
kind = models.IntegerField(
choices=PurchaseOrderKindEnum.choices,
default=PurchaseOrderKindEnum.WHOLESALE,
verbose_name='预采购单类型',
)
remarks = models.TextField(blank=True, null=True, verbose_name='备注')
def __str__(self):
return f'预采购单 {self.human_id or self.id}'
def clean(self):
super().clean()
errors = {}
if getattr(self, 'merchant_id', None) and getattr(self, 'supplier_id', None):
if self.supplier.merchant_id != self.merchant_id:
errors['supplier'] = '供应商所属商户与预采购单所属商户不一致'
if getattr(self, 'merchant_id', None) and getattr(self, 'warehouse_id', None):
if self.warehouse.merchant_id != self.merchant_id:
errors['warehouse'] = '仓库所属商户与预采购单所属商户不一致'
if getattr(self, 'operator_id', None) and getattr(self, 'merchant_id', None):
if self.operator.merchant_id != self.merchant_id:
errors['operator'] = '经办人所属商户与预采购单所属商户不一致'
if errors:
raise ValidationError(errors)
@property
def supplier_name(self) -> str:
if getattr(self, 'supplier_id', None):
return self.supplier.name
return ''
@property
def merchant_name(self) -> str:
if getattr(self, 'merchant_id', None):
return self.merchant.name
return ''
@property
def created_by_username(self) -> str:
if getattr(self, 'created_by_id', None):
return getattr(self.created_by, 'username', '') or ''
return ''
@property
def operator_name(self) -> str:
if getattr(self, 'operator_id', None):
return self.operator.name
return ''
@property
def warehouse_name(self) -> str:
if getattr(self, 'warehouse_id', None):
return self.warehouse.name
return ''
@property
def human_id(self) -> str:
"""人类可读编号（虚拟编号，不落库）。
规则：YC + YYYYMMDD{6位id}
"""
if self.id is None or self.created_at is None:
return ''
return f"YC{self.created_at.year}{self.created_at.month:02d}{self.created_at.day:02d}{self.id:06d}"
class Meta:
verbose_name = '预采购单'
verbose_name_plural = '预采购单'
class PrePurchaseOrderItem(ModelBase):
"""预采购单明细"""
id = models.BigAutoField(primary_key=True)
pre_purchase_order = models.ForeignKey(
PrePurchaseOrder,
on_delete=models.CASCADE,
related_name='items',
verbose_name='预采购单',
)
# 暂不与 Product 建立外键，先用弱关联字段承载。
product_id = models.BigIntegerField(null=True, blank=True, verbose_name='产品ID')
product_name = models.CharField(max_length=100, blank=True, default='', verbose_name='产品名称')
color = models.CharField(max_length=50, null=True, blank=True, verbose_name='颜色')
quantity = models.DecimalField(max_digits=10, decimal_places=2, null=True, blank=True, verbose_name='数量')
unit = models.CharField(max_length=50, blank=True, default='', verbose_name='单位')
spec = models.CharField(max_length=100, null=True, blank=True, verbose_name='规格')
quantity_of_rolls = models.TextField(
null=True,
blank=True,
verbose_name='各条数数量',
)
num_of_rolls = models.PositiveIntegerField(default=1, verbose_name='条数')
order_quantity = models.IntegerField(null=True, blank=True, verbose_name='下单数量')
remarks = models.TextField(blank=True, null=True, verbose_name='备注')
class Meta:
verbose_name = '预采购单明细'
verbose_name_plural = '预采购单明细'
class PurchaseReturnStatusEnum(models.IntegerChoices):
PENDING = 1, '审批中'
APPROVED = 2, '审批通过'

214
business/pre_order_services.py
View File

@@ -101,6 +101,15 @@ def _normalize_pre_sales_items(
return normalized
def _normalize_pre_purchase_items(
*,
merchant: basic_info_models.Merchant,
items: List[Dict[str, Any]],
) -> List[Dict[str, Any]]:
# 与预销售单明细字段保持一致的弱关联产品输入结构
return _normalize_pre_sales_items(merchant=merchant, items=items)
def list_pre_sales_orders(*, merchant: basic_info_models.Merchant):
return models.PreSalesOrder.objects.filter(merchant=merchant).select_related(
'merchant', 'customer', 'warehouse', 'created_by'
@@ -121,7 +130,8 @@ def create_pre_sales_order(
merchant: basic_info_models.Merchant,
customer_id: int | str | None,
warehouse_id: int | str | None,
created_by: basic_info_models.Employee,
operator: basic_info_models.Employee,
created_by=None,
kind: int | str | None = None,
items: List[Dict[str, Any]],
remarks: str | None = '',
@@ -160,8 +170,8 @@ def create_pre_sales_order(
except basic_info_models.WareHouse.DoesNotExist as exc:
raise ValueError(f'仓库 {warehouse_id_int} 不存在') from exc
if created_by.merchant_id != merchant.id:
raise ValueError('创建者所属商户与预销售单所属商户不一致')
if operator.merchant_id != merchant.id:
raise ValueError('经办人所属商户与预销售单所属商户不一致')
normalized_items = _normalize_pre_sales_items(merchant=merchant, items=items)
@@ -171,6 +181,7 @@ def create_pre_sales_order(
customer=customer,
warehouse=warehouse,
created_by=created_by,
operator=operator,
kind=kind_int or models.SalesOrderKindEnum.WHOLESALE,
remarks=remarks,
)
@@ -295,3 +306,200 @@ def update_pre_sales_order(
def delete_pre_sales_order(*, pre_sales_order: models.PreSalesOrder) -> None:
with transaction.atomic():
pre_sales_order.delete()
def list_pre_purchase_orders(*, merchant: basic_info_models.Merchant):
return models.PrePurchaseOrder.objects.filter(merchant=merchant).select_related(
'merchant', 'supplier', 'warehouse', 'created_by'
).prefetch_related('items')
def get_pre_purchase_order(*, merchant: basic_info_models.Merchant, pre_purchase_order_id: int) -> models.PrePurchaseOrder:
try:
return models.PrePurchaseOrder.objects.select_related(
'merchant', 'supplier', 'warehouse', 'created_by'
).prefetch_related('items').get(id=pre_purchase_order_id, merchant=merchant)
except models.PrePurchaseOrder.DoesNotExist as exc:
raise ValueError('预采购单不存在') from exc
def create_pre_purchase_order(
*,
merchant: basic_info_models.Merchant,
supplier_id: int | str | None,
warehouse_id: int | str | None,
operator: basic_info_models.Employee,
created_by=None,
kind: int | str | None = None,
items: List[Dict[str, Any]],
remarks: str | None = '',
) -> models.PrePurchaseOrder:
if not supplier_id:
raise ValueError('缺少供应商 ID')
if not warehouse_id:
raise ValueError('缺少仓库 ID')
try:
supplier_id_int = int(supplier_id)
except (TypeError, ValueError) as exc:
raise ValueError('supplier 必须为整数') from exc
try:
warehouse_id_int = int(warehouse_id)
except (TypeError, ValueError) as exc:
raise ValueError('warehouse 必须为整数') from exc
kind_int: int | None
if kind is None or kind == '':
kind_int = None
else:
try:
kind_int = int(kind)
except (TypeError, ValueError) as exc:
raise ValueError('kind 必须为整数') from exc
try:
supplier = basic_info_models.Supplier.objects.get(id=supplier_id_int, merchant=merchant)
except basic_info_models.Supplier.DoesNotExist as exc:
raise ValueError(f'供应商 {supplier_id_int} 不存在') from exc
try:
warehouse = basic_info_models.WareHouse.objects.get(id=warehouse_id_int, merchant=merchant)
except basic_info_models.WareHouse.DoesNotExist as exc:
raise ValueError(f'仓库 {warehouse_id_int} 不存在') from exc
if operator.merchant_id != merchant.id:
raise ValueError('经办人所属商户与预采购单所属商户不一致')
normalized_items = _normalize_pre_purchase_items(merchant=merchant, items=items)
with transaction.atomic():
pre_purchase_order = models.PrePurchaseOrder.objects.create(
merchant=merchant,
supplier=supplier,
warehouse=warehouse,
created_by=created_by,
operator=operator,
kind=kind_int or models.PurchaseOrderKindEnum.WHOLESALE,
remarks=remarks,
)
try:
pre_purchase_order.full_clean()
except ValidationError as exc:
raise ValueError(str(exc)) from exc
models.PrePurchaseOrderItem.objects.bulk_create(
[
models.PrePurchaseOrderItem(
pre_purchase_order=pre_purchase_order,
product_id=item['product_id'],
product_name=item.get('product_name') or '',
color=item.get('color'),
quantity=item.get('quantity'),
unit=item.get('unit') or '',
spec=item.get('spec'),
quantity_of_rolls=item.get('quantity_of_rolls'),
num_of_rolls=item.get('num_of_rolls') or 1,
order_quantity=item.get('order_quantity'),
remarks=item.get('remarks'),
)
for item in normalized_items
]
)
pre_purchase_order.refresh_from_db()
return pre_purchase_order
def update_pre_purchase_order(
*,
pre_purchase_order: models.PrePurchaseOrder,
supplier_id: int | str | None = None,
warehouse_id: int | str | None = None,
kind: int | str | None = None,
items: List[Dict[str, Any]] | None = None,
remarks: str | None = None,
) -> models.PrePurchaseOrder:
merchant = pre_purchase_order.merchant
new_supplier = pre_purchase_order.supplier
if supplier_id is not None:
if supplier_id == '':
raise ValueError('supplier 必须为整数')
try:
supplier_id_int = int(supplier_id)
except (TypeError, ValueError) as exc:
raise ValueError('supplier 必须为整数') from exc
try:
new_supplier = basic_info_models.Supplier.objects.get(id=supplier_id_int, merchant=merchant)
except basic_info_models.Supplier.DoesNotExist as exc:
raise ValueError(f'供应商 {supplier_id_int} 不存在') from exc
new_warehouse = pre_purchase_order.warehouse
if warehouse_id is not None:
if warehouse_id == '':
raise ValueError('warehouse 必须为整数')
try:
warehouse_id_int = int(warehouse_id)
except (TypeError, ValueError) as exc:
raise ValueError('warehouse 必须为整数') from exc
try:
new_warehouse = basic_info_models.WareHouse.objects.get(id=warehouse_id_int, merchant=merchant)
except basic_info_models.WareHouse.DoesNotExist as exc:
raise ValueError(f'仓库 {warehouse_id_int} 不存在') from exc
if kind is None or kind == '':
new_kind = pre_purchase_order.kind
else:
try:
new_kind = int(kind)
except (TypeError, ValueError) as exc:
raise ValueError('kind 必须为整数') from exc
new_remarks = remarks if remarks is not None else pre_purchase_order.remarks
if items is None:
raise ValueError('items 需要为非空数组')
normalized_items = _normalize_pre_purchase_items(merchant=merchant, items=items)
with transaction.atomic():
pre_purchase_order.supplier = new_supplier
pre_purchase_order.warehouse = new_warehouse
pre_purchase_order.kind = new_kind
pre_purchase_order.remarks = new_remarks
try:
pre_purchase_order.full_clean()
except ValidationError as exc:
raise ValueError(str(exc)) from exc
pre_purchase_order.save(update_fields=['supplier', 'warehouse', 'kind', 'remarks', 'updated_at'])
pre_purchase_order.items.all().delete()
models.PrePurchaseOrderItem.objects.bulk_create(
[
models.PrePurchaseOrderItem(
pre_purchase_order=pre_purchase_order,
product_id=item['product_id'],
product_name=item.get('product_name') or '',
color=item.get('color'),
quantity=item.get('quantity'),
unit=item.get('unit') or '',
spec=item.get('spec'),
quantity_of_rolls=item.get('quantity_of_rolls'),
num_of_rolls=item.get('num_of_rolls') or 1,
order_quantity=item.get('order_quantity'),
remarks=item.get('remarks'),
)
for item in normalized_items
]
)
pre_purchase_order.refresh_from_db()
return pre_purchase_order
def delete_pre_purchase_order(*, pre_purchase_order: models.PrePurchaseOrder) -> None:
with transaction.atomic():
pre_purchase_order.delete()

120
business/tests/test_pre_purchase_order.py Normal file
View File

@@ -0,0 +1,120 @@
from django.test import TestCase
from django.contrib.auth import get_user_model
from basic_info import models as basic_models
from business import models as business_models
from business import pre_order_services
from .fixtures import create_basic_fixtures
class PrePurchaseOrderServiceTestCase(TestCase):
def setUp(self):
(
self.merchant,
self.supplier,
self.warehouse_strict,
self.warehouse_relaxed,
self.product,
self.operator,
) = create_basic_fixtures()
self.user = get_user_model().objects.create_user(username='purchase-user', password='pass')
self.operator.sys_user = self.user
self.operator.save(update_fields=['sys_user'])
def test_create_pre_purchase_order_success(self):
pre_purchase_order = pre_order_services.create_pre_purchase_order(
merchant=self.merchant,
supplier_id=self.supplier.id,
warehouse_id=self.warehouse_relaxed.id,
created_by=self.user,
operator=self.operator,
kind=business_models.PurchaseOrderKindEnum.WHOLESALE,
items=[
{
'product_id': self.product.id,
'quantity': '12.5',
'unit': '米',
'order_quantity': 8,
'remarks': 'item-remarks',
}
],
remarks='order-remarks',
)
self.assertEqual(pre_purchase_order.merchant_id, self.merchant.id)
self.assertEqual(pre_purchase_order.supplier_id, self.supplier.id)
self.assertEqual(pre_purchase_order.warehouse_id, self.warehouse_relaxed.id)
self.assertEqual(pre_purchase_order.created_by_id, self.user.id)
self.assertEqual(pre_purchase_order.operator_id, self.operator.id)
self.assertEqual(pre_purchase_order.items.count(), 1)
item = pre_purchase_order.items.first()
self.assertEqual(item.product_id, self.product.id)
self.assertEqual(item.unit, '米')
self.assertEqual(item.order_quantity, 8)
def test_create_pre_purchase_order_rejects_cross_merchant_supplier(self):
other_merchant = basic_models.Merchant.objects.create(
name='其他商户',
type=basic_models.MerchantTypeEnum.FACTORY,
)
other_supplier = basic_models.Supplier.objects.create(
merchant=other_merchant,
name='其他供应商',
)
with self.assertRaises(ValueError):
pre_order_services.create_pre_purchase_order(
merchant=self.merchant,
supplier_id=other_supplier.id,
warehouse_id=self.warehouse_relaxed.id,
created_by=self.user,
operator=self.operator,
items=[{'product_id': self.product.id, 'quantity': '1', 'unit': '米'}],
)
def test_update_pre_purchase_order_replaces_items(self):
pre_purchase_order = pre_order_services.create_pre_purchase_order(
merchant=self.merchant,
supplier_id=self.supplier.id,
warehouse_id=self.warehouse_relaxed.id,
created_by=self.user,
operator=self.operator,
items=[{'product_id': self.product.id, 'quantity': '10', 'unit': '米'}],
)
updated = pre_order_services.update_pre_purchase_order(
pre_purchase_order=pre_purchase_order,
warehouse_id=self.warehouse_strict.id,
items=[
{
'product_id': self.product.id,
'quantity': '20',
'unit': '米',
'order_quantity': 3,
}
],
remarks='updated-remarks',
)
self.assertEqual(updated.warehouse_id, self.warehouse_strict.id)
self.assertEqual(updated.remarks, 'updated-remarks')
self.assertEqual(updated.items.count(), 1)
self.assertEqual(str(updated.items.first().quantity), '20.00')
self.assertEqual(updated.items.first().order_quantity, 3)
def test_delete_pre_purchase_order(self):
pre_purchase_order = pre_order_services.create_pre_purchase_order(
merchant=self.merchant,
supplier_id=self.supplier.id,
warehouse_id=self.warehouse_relaxed.id,
created_by=self.user,
operator=self.operator,
items=[{'product_id': self.product.id, 'quantity': '10', 'unit': '米'}],
)
order_id = pre_purchase_order.id
pre_order_services.delete_pre_purchase_order(pre_purchase_order=pre_purchase_order)
self.assertFalse(business_models.PrePurchaseOrder.objects.filter(id=order_id).exists())

20
business/tests/test_pre_sales_order.py
View File

@@ -1,4 +1,5 @@
from django.test import TestCase
from django.contrib.auth import get_user_model
from basic_info import models as basic_models
from business import models as business_models
@@ -19,12 +20,17 @@ class PreSalesOrderServiceTestCase(TestCase):
self.operator,
) = create_sales_fixtures()
self.user = get_user_model().objects.create_user(username='sales-user', password='pass')
self.operator.sys_user = self.user
self.operator.save(update_fields=['sys_user'])
def test_create_pre_sales_order_success(self):
pre_sales_order = pre_order_services.create_pre_sales_order(
merchant=self.merchant,
customer_id=self.customer.id,
warehouse_id=self.warehouse_relaxed.id,
created_by=self.operator,
created_by=self.user,
operator=self.operator,
kind=business_models.SalesOrderKindEnum.WHOLESALE,
items=[
{
@@ -41,7 +47,8 @@ class PreSalesOrderServiceTestCase(TestCase):
self.assertEqual(pre_sales_order.merchant_id, self.merchant.id)
self.assertEqual(pre_sales_order.customer_id, self.customer.id)
self.assertEqual(pre_sales_order.warehouse_id, self.warehouse_relaxed.id)
self.assertEqual(pre_sales_order.created_by_id, self.operator.id)
self.assertEqual(pre_sales_order.created_by_id, self.user.id)
self.assertEqual(pre_sales_order.operator_id, self.operator.id)
self.assertEqual(pre_sales_order.items.count(), 1)
item = pre_sales_order.items.first()
@@ -65,7 +72,8 @@ class PreSalesOrderServiceTestCase(TestCase):
merchant=self.merchant,
customer_id=other_customer.id,
warehouse_id=self.warehouse_relaxed.id,
created_by=self.operator,
created_by=self.user,
operator=self.operator,
items=[{'product_id': self.product.id, 'quantity': '1', 'unit': '米'}],
)
@@ -74,7 +82,8 @@ class PreSalesOrderServiceTestCase(TestCase):
merchant=self.merchant,
customer_id=self.customer.id,
warehouse_id=self.warehouse_relaxed.id,
created_by=self.operator,
created_by=self.user,
operator=self.operator,
items=[{'product_id': self.product.id, 'quantity': '10', 'unit': '米'}],
)
@@ -103,7 +112,8 @@ class PreSalesOrderServiceTestCase(TestCase):
merchant=self.merchant,
customer_id=self.customer.id,
warehouse_id=self.warehouse_relaxed.id,
created_by=self.operator,
created_by=self.user,
operator=self.operator,
items=[{'product_id': self.product.id, 'quantity': '10', 'unit': '米'}],
)
order_id = pre_sales_order.id