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

feat: relaxed in restrict out

Browse Source
This commit is contained in:
colaftc
2025-11-24 15:57:19 +08:00
parent 113818f73d
commit cd6a2370fc
11 changed files with 840 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
api_v1/serializers.py
View File

@@ -55,6 +55,51 @@ class CreateStockChangeSerializer(serializers.Serializer):
return value
class RelaxedQuantitySerializer(serializers.Serializer):
"""宽松模式数量结构"""
value = serializers.DecimalField(
max_digits=10,
decimal_places=2,
min_value=Decimal('0.01'),
help_text="产品总数量"
)
unit_count = serializers.DecimalField(
max_digits=10,
decimal_places=2,
min_value=Decimal('0.01'),
default=Decimal('1.00'),
help_text="单条（匹）数量，默认 1"
)
class RelaxedProductStockChangeSerializer(serializers.Serializer):
"""宽松模式产品序列化器"""
product = serializers.IntegerField(help_text="产品ID")
quantity = RelaxedQuantitySerializer(help_text="数量定义（总数与单条数）")
def validate_product(self, value: int) -> int:
try:
basic_info_models.Product.objects.get(id=value)
except basic_info_models.Product.DoesNotExist:
raise serializers.ValidationError(f"产品ID {value} 不存在")
return value
class CreateStockChangeRelaxedSerializer(serializers.Serializer):
"""宽松模式创建库存变动记录"""
products = RelaxedProductStockChangeSerializer(many=True, help_text="产品列表")
def validate_products(self, value: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
if not value:
raise serializers.ValidationError("产品列表不能为空")
product_ids = [item['product'] for item in value]
if len(product_ids) != len(set(product_ids)):
raise serializers.ValidationError("产品列表中存在重复的产品ID")
return value
class StockChangeRecordResponseSerializer(serializers.ModelSerializer):
"""库存变动记录响应序列化器"""

111
api_v1/tests.py
View File

@@ -1,3 +1,112 @@
from django.test import TestCase
from django.contrib.auth.models import User, Permission
from rest_framework.test import APIClient
from rest_framework import status
from basic_info.models import Merchant, UserProfile
# Create your tests here.
class UserCreationAPITestCase(TestCase):
"""测试用户创建 API"""
def setUp(self):
"""设置测试数据"""
# 创建商户
self.merchant = Merchant.objects.create(
name='测试商户',
type=1 # 假设1是有效的MerchantType
)
# 创建管理员用户
self.admin_user = User.objects.create_user(
username='admin',
password='adminpass123',
is_staff=True
)
# 授予创建用户所需的权限
add_user_perm = Permission.objects.get(codename='add_user')
self.admin_user.user_permissions.add(add_user_perm)
self.admin_user.save()
# 设置 API 客户端
self.client = APIClient()
self.client.force_authenticate(user=self.admin_user)
def test_create_user_with_profile(self):
"""测试创建用户和用户资料"""
data = {
'username': 'testuser',
'email': 'test@example.com',
'password': 'testpass123',
'is_staff': False,
'description': '测试用户资料',
'merchant_id': self.merchant.id
}
response = self.client.post('/api/v1/users/create/', data, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
# 验证返回的数据
self.assertIn('user', response.data)
self.assertIn('profile', response.data)
self.assertEqual(response.data['user']['username'], 'testuser')
self.assertEqual(response.data['user']['email'], 'test@example.com')
self.assertEqual(response.data['user']['is_staff'], False)
self.assertEqual(response.data['profile']['description'], '测试用户资料')
self.assertEqual(response.data['profile']['merchant'], self.merchant.id)
# 验证数据库中的用户
user = User.objects.get(username='testuser')
self.assertEqual(user.email, 'test@example.com')
self.assertFalse(user.is_staff)
# 验证数据库中的用户资料
profile = UserProfile.objects.get(user=user)
self.assertEqual(profile.merchant, self.merchant)
self.assertEqual(profile.description, '测试用户资料')
def test_create_user_with_duplicate_username(self):
"""测试创建用户时使用重复的用户名"""
# 先创建一个用户
User.objects.create_user(username='existinguser', password='pass123')
data = {
'username': 'existinguser', # 重复的用户名
'password': 'testpass123',
'merchant_id': self.merchant.id
}
response = self.client.post('/api/v1/users/create/', data, format='json')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn('username', response.data)
def test_create_user_with_invalid_merchant(self):
"""测试创建用户时使用无效的merchant_id"""
data = {
'username': 'testuser',
'password': 'testpass123',
'merchant_id': 999 # 不存在的merchant_id
}
response = self.client.post('/api/v1/users/create/', data, format='json')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn('merchant_id', response.data)
def test_create_user_unauthenticated(self):
"""测试未认证用户创建用户"""
self.client.force_authenticate(user=None)
data = {
'username': 'testuser',
'password': 'testpass123',
'merchant_id': self.merchant.id
}
response = self.client.post('/api/v1/users/create/', data, format='json')
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
def test_create_user_with_short_password(self):
"""测试创建用户时密码过短"""
data = {
'username': 'testuser',
'password': '123', # 密码过短
'merchant_id': self.merchant.id
}
response = self.client.post('/api/v1/users/create/', data, format='json')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn('password', response.data)

7
api_v1/urls.py
View File

@@ -1,11 +1,12 @@
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import stock_change_views, user_info, inventory, product_image, stateflow
from .views import stock_change_views, user_info, inventory, product_image, stateflow, users
from .views.stock_change_views.snapshot import StockSnapshotListView
from .views.printing.views import PrintingOrderViewSet, PrintingJobViewSet, PlateOrderViewSet
from .views.upload import UploadFileViewSet
from .views.products import ProductQuickViewSet
from .views.parameters import StateParameterViewSet
from .views.users import CreateUserWithProfileView
# 创建 DRF Router for Stateflow
stateflow_router = DefaultRouter()
@@ -27,6 +28,7 @@ urlpatterns = [
path('stock-snapshots/', StockSnapshotListView.as_view(), name='list_stock_snapshots'),
path('stock-changes/', stock_change_views.list_stock_changes, name='list_stock_changes'),
path('stock-change/', stock_change_views.create_full_stock_change, name='create_full_stock_change'),
path('stock-change/relaxed/', stock_change_views.create_relaxed_stock_change, name='create_relaxed_stock_change'),
path('stock-change/<int:record_id>/', stock_change_views.get_stock_change, name='get_stock_change'),
path(
'set-merchant-auto-complete-stock-change/',
@@ -37,6 +39,9 @@ urlpatterns = [
# 用户信息 API
path('user-info/', user_info.user_info, name='user_info'),
# 用户创建 API
path('users/create/', CreateUserWithProfileView.as_view(), name='create_user_with_profile'),
# 库存查询 API
path('inventory/', inventory.InventoryAPIView.as_view(), name='inventory'),

1
api_v1/views/__init__.py
View File

@@ -20,4 +20,5 @@ __all__ = [
'get_stock_change',
'set_merchant_auto_complete_stock_change',
'stateflow',
'create_user_with_profile',
]

4
api_v1/views/stock_change_views/README.md
View File

@@ -63,6 +63,7 @@ stock_change_views/
```python
# 这些接口保持不变，URL 配置无需修改
create_full_stock_change = CreateStockChangeView.as_view()
create_relaxed_stock_change = CreateStockChangeRelaxedView.as_view()
list_stock_changes = ListStockChangesView.as_view()
get_stock_change = GetStockChangeView.as_view()
set_merchant_auto_complete_stock_change = SetMerchantAutoCompleteView.as_view()
@@ -87,6 +88,7 @@ from api_v1.views import stock_change_views
urlpatterns = [
path('stock-change/', stock_change_views.create_full_stock_change),
path('stock-change/relaxed/', stock_change_views.create_relaxed_stock_change),
path('stock-changes/', stock_change_views.list_stock_changes),
path('stock-change/<int:record_id>/', stock_change_views.get_stock_change),
path('set-merchant-auto-complete-stock-change/', stock_change_views.set_merchant_auto_complete_stock_change),
@@ -98,6 +100,7 @@ urlpatterns = [
```python
from api_v1.views.stock_change_views import (
CreateStockChangeView,
CreateStockChangeRelaxedView,
ListStockChangesView,
GetStockChangeView,
SetMerchantAutoCompleteView,
@@ -105,6 +108,7 @@ from api_v1.views.stock_change_views import (
urlpatterns = [
path('stock-change/', CreateStockChangeView.as_view()),
path('stock-change/relaxed/', CreateStockChangeRelaxedView.as_view()),
path('stock-changes/', ListStockChangesView.as_view()),
path('stock-change/<int:record_id>/', GetStockChangeView.as_view()),
path('set-merchant-auto-complete-stock-change/', SetMerchantAutoCompleteView.as_view()),

5
api_v1/views/stock_change_views/__init__.py
View File

@@ -5,13 +5,14 @@
"""
from .mixins import StockChangeViewMixin
from .create import CreateStockChangeView
from .create import CreateStockChangeView, CreateStockChangeRelaxedView
from .list import ListStockChangesView
from .detail import GetStockChangeView
from .settings import SetMerchantAutoCompleteView
# 向后兼容：保持原有的函数式接口
create_full_stock_change = CreateStockChangeView.as_view()
create_relaxed_stock_change = CreateStockChangeRelaxedView.as_view()
list_stock_changes = ListStockChangesView.as_view()
get_stock_change = GetStockChangeView.as_view()
set_merchant_auto_complete_stock_change = SetMerchantAutoCompleteView.as_view()
@@ -19,10 +20,12 @@ set_merchant_auto_complete_stock_change = SetMerchantAutoCompleteView.as_view()
__all__ = [
'StockChangeViewMixin',
'CreateStockChangeView',
'CreateStockChangeRelaxedView',
'ListStockChangesView',
'GetStockChangeView',
'SetMerchantAutoCompleteView',
'create_full_stock_change',
'create_relaxed_stock_change',
'list_stock_changes',
'get_stock_change',
'set_merchant_auto_complete_stock_change',

86
api_v1/views/stock_change_views/create.py
View File

@@ -126,3 +126,89 @@ class CreateStockChangeView(StockChangeViewMixin, views.APIView):
'error': '创建库存变动记录失败',
'message': str(e)
}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
class CreateStockChangeRelaxedView(StockChangeViewMixin, views.APIView):
"""宽松模式：根据总量与单条数量拆分的出入库记录"""
permission_classes = [IsAuthenticated]
@extend_schema(
tags=['创建出入库'],
request=serializers.CreateStockChangeRelaxedSerializer,
responses={201: serializers.CreateStockChangeResponseSerializer},
summary="创建库存变动记录（宽松模式）",
description="根据总数量与单条数量自动拆分明细"
)
def post(self, request):
if not self.check_employee_permission(request):
return self.permission_error_response('无权限访问')
record_data = {
'type': request.data.get('type'),
'warehouse': request.data.get('warehouse'),
'source_type': request.data.get('source_type'),
'source_id': request.data.get('source_id'),
}
if not all([record_data['type'], record_data['warehouse'], record_data['source_type']]):
return Response({
'error': '缺少必要参数',
'message': '请提供 type, warehouse, source_type'
}, status=status.HTTP_400_BAD_REQUEST)
products_data = request.data.get('products', [])
if not products_data:
return Response({
'error': '产品列表不能为空'
}, status=status.HTTP_400_BAD_REQUEST)
if not self.validate_warehouse_visibility(record_data['warehouse'], request):
return Response({
'error': f'仓库ID {record_data["warehouse"]} 对当前用户不可见'
}, status=status.HTTP_403_FORBIDDEN)
for p in products_data:
product_id = p.get('product')
if not self.validate_product_visibility(product_id, request):
return Response({
'error': f'产品ID {product_id} 对当前用户不可见'
}, status=status.HTTP_403_FORBIDDEN)
serializer = serializers.CreateStockChangeRelaxedSerializer(data={'products': products_data})
if not serializer.is_valid():
return Response({
'error': '产品数据验证失败',
'details': serializer.errors
}, status=status.HTTP_400_BAD_REQUEST)
try:
stock_change_record, created_details, created_count = stock_services.create_stock_change_record_relaxed(
merchant=request.user.employee.merchant,
created_by=request.user,
type=record_data['type'],
warehouse_id=record_data['warehouse'],
source_type=record_data['source_type'],
source_id=record_data['source_id'],
products=products_data,
)
response_serializer = serializers.CreateStockChangeResponseSerializer({
'stock_change_record': stock_change_record,
'details': created_details,
'message': f'成功创建库存变动记录及 {created_count} 条明细',
'created_details_count': created_count
})
return Response(response_serializer.data, status=status.HTTP_201_CREATED)
except ValueError as e:
return Response({
'error': str(e)
}, status=status.HTTP_400_BAD_REQUEST)
except Exception as e:
logger.error(f"创建宽松模式库存变动记录失败: {str(e)}", exc_info=True)
return Response({
'error': '创建库存变动记录失败',
'message': str(e)
}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)

123
api_v1/views/users.py Normal file
View File

@@ -0,0 +1,123 @@
from django.contrib.auth.models import User
from rest_framework.decorators import permission_classes
from rest_framework.permissions import IsAuthenticated, DjangoModelPermissions
from rest_framework.response import Response
from rest_framework.pagination import LimitOffsetPagination
from rest_framework import status, serializers
from rest_framework.generics import GenericAPIView
from basic_info.models import UserProfile
class UserCreationSerializer(serializers.Serializer):
"""用户创建序列化器"""
username = serializers.CharField(max_length=150)
email = serializers.EmailField(required=False)
password = serializers.CharField(min_length=6, write_only=True)
is_staff = serializers.BooleanField(default=False)
description = serializers.CharField(required=False, allow_blank=True)
merchant_id = serializers.IntegerField()
def validate_username(self, value):
"""验证用户名是否已存在"""
if User.objects.filter(username=value).exists():
raise serializers.ValidationError("用户名已存在")
return value
def validate_merchant_id(self, value):
"""验证merchant_id是否存在"""
from basic_info.models import Merchant
if not Merchant.objects.filter(id=value).exists():
raise serializers.ValidationError("商户不存在")
return value
class UserSerializer(serializers.ModelSerializer):
"""用户信息序列化器"""
class Meta:
model = User
fields = ['id', 'username', 'email', 'is_staff', 'is_active', 'date_joined']
class UserProfileDetailSerializer(serializers.ModelSerializer):
"""用户资料详细信息序列化器"""
user = UserSerializer(read_only=True)
class Meta:
model = UserProfile
fields = ['id', 'user', 'merchant', 'description', 'created_at', 'updated_at']
class CreateUserWithProfileView(GenericAPIView):
"""创建用户和关联的用户资料信息视图"""
queryset = User.objects.all()
permission_classes = [IsAuthenticated, DjangoModelPermissions]
pagination_class = LimitOffsetPagination
def post(self, request):
"""
创建用户和用户资料
POST /api/v1/users/create/
请求参数:
- username: 用户名 (必需)
- email: 邮箱 (可选)
- password: 密码 (必需, 最少6位)
- is_staff: 是否为员工 (可选, 默认为False)
- description: 用户资料描述 (可选)
- merchant_id: 商户ID (必需)
返回:
- user: 创建的用户信息
- profile: 创建的用户资料信息
"""
# 检查当前用户是否有权限（超级用户或有员工身份）
if not request.user.is_authenticated:
return Response({'error': '未授权'}, status=status.HTTP_401_UNAUTHORIZED)
# 序列化和验证请求数据
serializer = UserCreationSerializer(data=request.data)
if not serializer.is_valid():
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
username = serializer.validated_data['username']
email = serializer.validated_data.get('email', '')
password = serializer.validated_data['password']
is_staff = serializer.validated_data.get('is_staff', False)
description = serializer.validated_data.get('description', '')
merchant_id = serializer.validated_data['merchant_id']
try:
# 创建用户
user = User.objects.create_user(
username=username,
email=email,
password=password,
is_staff=is_staff
)
# 创建用户资料
from basic_info.models import Merchant
merchant = Merchant.objects.get(id=merchant_id)
profile = UserProfile.objects.create(
user=user,
merchant=merchant,
description=description
)
# 返回创建的用户和用户资料信息
user_data = UserSerializer(user).data
profile_data = UserProfileDetailSerializer(profile).data
return Response({
'user': user_data,
'profile': profile_data
}, status=status.HTTP_201_CREATED)
except Exception as e:
# 如果创建过程中出现错误，删除已创建的用户
if 'user' in locals():
user.delete()
return Response({
'error': f'创建用户失败: {str(e)}'
}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)