forked from erp-dev/erp
302 lines
11 KiB
Python
302 lines
11 KiB
Python
from rest_framework import serializers
|
||
from decimal import Decimal
|
||
from typing import List, Dict, Any
|
||
from stock import models as stock_models
|
||
from basic_info import models as basic_info_models
|
||
from api_v1.utils.media import build_public_media_url
|
||
from . import models as api_models
|
||
from .enums import PrintCountObjectType
|
||
|
||
|
||
class ProductStockChangeSerializer(serializers.Serializer):
|
||
"""产品库存变动序列化器"""
|
||
product = serializers.IntegerField(help_text="产品ID")
|
||
quantity = serializers.ListField(
|
||
child=serializers.DecimalField(max_digits=10, decimal_places=2, min_value=Decimal('0.01')),
|
||
min_length=1,
|
||
required=False,
|
||
help_text="数量列表，每个数量对应一条明细记录"
|
||
)
|
||
consume_with = serializers.ListField(
|
||
child=serializers.IntegerField(min_value=1),
|
||
required=False,
|
||
allow_empty=False,
|
||
help_text="严进严出模式（出库）指定的入库明细ID列表"
|
||
)
|
||
|
||
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
|
||
|
||
def validate_consume_with(self, value: List[int]) -> List[int]:
|
||
if len(value) != len(set(value)):
|
||
raise serializers.ValidationError("consume_with 中存在重复的入库明细ID")
|
||
return value
|
||
|
||
|
||
class CreateStockChangeSerializer(serializers.Serializer):
|
||
"""创建库存变动记录序列化器"""
|
||
stock_change_record_id = serializers.IntegerField(
|
||
required=False,
|
||
allow_null=True,
|
||
help_text="库存变动记录ID，如果不提供则创建新记录"
|
||
)
|
||
products = ProductStockChangeSerializer(many=True, help_text="产品列表")
|
||
|
||
def validate_stock_change_record_id(self, value: int) -> int:
|
||
"""验证库存变动记录是否存在"""
|
||
if value is not None:
|
||
try:
|
||
stock_models.StockChangeRecord.objects.get(id=value)
|
||
except stock_models.StockChangeRecord.DoesNotExist:
|
||
raise serializers.ValidationError(f"库存变动记录ID {value} 不存在")
|
||
return value
|
||
|
||
def validate_products(self, value: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||
"""验证产品列表"""
|
||
if not value:
|
||
raise serializers.ValidationError("产品列表不能为空")
|
||
|
||
# 检查是否有重复的产品ID
|
||
product_ids = [item['product'] for item in value]
|
||
if len(product_ids) != len(set(product_ids)):
|
||
raise serializers.ValidationError("产品列表中存在重复的产品ID")
|
||
|
||
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):
|
||
"""库存变动记录响应序列化器"""
|
||
|
||
class Meta:
|
||
model = stock_models.StockChangeRecord
|
||
fields = [
|
||
'id', 'type', 'warehouse', 'source_type', 'source_id',
|
||
'is_finished', 'finished_at', 'cancelled', 'is_offset', 'is_reversed',
|
||
'created_at', 'updated_at'
|
||
]
|
||
|
||
|
||
class StockChangeDetailResponseSerializer(serializers.ModelSerializer):
|
||
"""库存变动明细响应序列化器"""
|
||
product_name = serializers.CharField(source='product.name', read_only=True)
|
||
unit_display = serializers.CharField(source='get_unit_display', read_only=True)
|
||
consume_with_id = serializers.IntegerField(read_only=True)
|
||
|
||
class Meta:
|
||
model = stock_models.StockChangeDetail
|
||
fields = [
|
||
'id', 'product', 'product_name', 'quantity',
|
||
'unit', 'unit_display', 'stock_change_record',
|
||
'is_consumed', 'consume_with_id',
|
||
]
|
||
|
||
|
||
class CreateStockChangeResponseSerializer(serializers.Serializer):
|
||
"""创建库存变动记录响应序列化器"""
|
||
stock_change_record = StockChangeRecordResponseSerializer()
|
||
details = StockChangeDetailResponseSerializer(many=True)
|
||
message = serializers.CharField()
|
||
created_details_count = serializers.IntegerField()
|
||
|
||
|
||
class SetMerchantAutoCompleteStockChangeSerializer(serializers.Serializer):
|
||
"""设置商户自动确认库存变动序列化器"""
|
||
auto_complete = serializers.BooleanField(
|
||
help_text="是否自动确认库存变动"
|
||
)
|
||
|
||
|
||
class StockTransferProductSerializer(serializers.Serializer):
|
||
product_id = serializers.IntegerField()
|
||
quantities = serializers.ListField(
|
||
child=serializers.DecimalField(max_digits=10, decimal_places=2),
|
||
required=False,
|
||
allow_empty=False,
|
||
)
|
||
value = serializers.DecimalField(max_digits=10, decimal_places=2, required=False)
|
||
num_of_rolls = serializers.IntegerField(required=False)
|
||
unit_count = serializers.DecimalField(max_digits=10, decimal_places=2, required=False)
|
||
consume_detail_ids = serializers.ListField(
|
||
child=serializers.IntegerField(min_value=1),
|
||
required=False,
|
||
allow_empty=False,
|
||
)
|
||
|
||
def validate(self, attrs):
|
||
has_quantities = bool(attrs.get('quantities'))
|
||
has_value = attrs.get('value') is not None
|
||
has_consumes = bool(attrs.get('consume_detail_ids'))
|
||
if not (has_quantities or has_value or has_consumes):
|
||
raise serializers.ValidationError('必须提供 quantities、value 或 consume_detail_ids 之一')
|
||
return attrs
|
||
|
||
|
||
class StockTransferSerializer(serializers.Serializer):
|
||
from_warehouse = serializers.IntegerField()
|
||
to_warehouse = serializers.IntegerField()
|
||
transfer_date = serializers.DateField(required=False)
|
||
remarks = serializers.CharField(required=False, allow_blank=True, allow_null=True, max_length=500)
|
||
request_id = serializers.CharField(required=False, allow_blank=True, allow_null=True, max_length=64)
|
||
products = StockTransferProductSerializer(many=True)
|
||
|
||
def validate_products(self, value):
|
||
if not value:
|
||
raise serializers.ValidationError('产品列表不能为空')
|
||
return value
|
||
|
||
|
||
class TransferOrderItemSerializer(serializers.ModelSerializer):
|
||
product_name = serializers.CharField(source='product.name', read_only=True)
|
||
unit_display = serializers.CharField(source='get_unit_display', read_only=True)
|
||
|
||
class Meta:
|
||
model = stock_models.TransferOrderItem
|
||
fields = ['id', 'product', 'product_name', 'total_quantity', 'unit', 'unit_display', 'num_of_rolls']
|
||
|
||
|
||
class StockTransferResponseSerializer(serializers.ModelSerializer):
|
||
items = TransferOrderItemSerializer(many=True, read_only=True)
|
||
from_warehouse_name = serializers.CharField(source='from_warehouse.name', read_only=True)
|
||
to_warehouse_name = serializers.CharField(source='to_warehouse.name', read_only=True)
|
||
mode_display = serializers.CharField(source='get_mode_display', read_only=True)
|
||
status_display = serializers.CharField(source='get_status_display', read_only=True)
|
||
outgoing_record_id = serializers.IntegerField(source='outgoing_record.id', read_only=True)
|
||
incoming_record_id = serializers.IntegerField(source='incoming_record.id', read_only=True)
|
||
|
||
class Meta:
|
||
model = stock_models.TransferOrder
|
||
fields = [
|
||
'id',
|
||
'merchant',
|
||
'from_warehouse',
|
||
'from_warehouse_name',
|
||
'to_warehouse',
|
||
'to_warehouse_name',
|
||
'mode',
|
||
'mode_display',
|
||
'operator',
|
||
'transfer_date',
|
||
'status',
|
||
'status_display',
|
||
'remarks',
|
||
'request_id',
|
||
'outgoing_record_id',
|
||
'incoming_record_id',
|
||
'items',
|
||
]
|
||
|
||
|
||
# ==================== 通用文件上传序列化器 ====================
|
||
|
||
class UploadedFileSerializer(serializers.ModelSerializer):
|
||
"""上传文件序列化器"""
|
||
file_url = serializers.SerializerMethodField()
|
||
owner_username = serializers.CharField(source='owner.username', read_only=True)
|
||
|
||
class Meta:
|
||
model = api_models.UploadedFile
|
||
fields = [
|
||
'id', 'path', 'file_url', 'original_filename', 'file_size',
|
||
'content_type', 'owner', 'owner_username', 'is_deleted',
|
||
'created_at', 'updated_at'
|
||
]
|
||
read_only_fields = ['id', 'owner', 'file_size', 'content_type', 'created_at', 'updated_at']
|
||
|
||
def get_file_url(self, obj):
|
||
"""获取文件完整URL"""
|
||
if obj.path:
|
||
request = self.context.get('request')
|
||
try:
|
||
raw_value = obj.path.url
|
||
except ValueError:
|
||
raw_value = obj.path.name
|
||
return build_public_media_url(raw_value, request=request)
|
||
return None
|
||
|
||
|
||
class FileUploadSerializer(serializers.Serializer):
|
||
"""文件上传请求序列化器"""
|
||
file = serializers.FileField(
|
||
required=True,
|
||
help_text='要上传的文件'
|
||
)
|
||
|
||
def validate_file(self, value):
|
||
"""验证文件"""
|
||
# 限制文件大小为 100MB
|
||
max_size = 100 * 1024 * 1024
|
||
if value.size > max_size:
|
||
raise serializers.ValidationError(f'文件大小不能超过 {max_size // (1024*1024)}MB')
|
||
|
||
return value
|
||
|
||
|
||
class PrintCountDeltaSerializer(serializers.Serializer):
|
||
"""打印次数增量请求"""
|
||
|
||
object_type = serializers.ChoiceField(choices=PrintCountObjectType.choices)
|
||
object_id = serializers.IntegerField(min_value=1)
|
||
delta = serializers.CharField(required=False, allow_null=True, allow_blank=True)
|
||
|
||
def validate(self, attrs: Dict[str, Any]) -> Dict[str, Any]:
|
||
raw_delta = attrs.get('delta')
|
||
try:
|
||
delta_value = int(raw_delta)
|
||
except (TypeError, ValueError):
|
||
delta_value = 1
|
||
|
||
if delta_value < 1:
|
||
delta_value = 1
|
||
|
||
attrs['delta'] = delta_value
|
||
return attrs |