feat: added printing module

api_v1/urls.py

@@ -1,15 +1,21 @@
 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.stock_change_views.snapshot import StockSnapshotListView
+from printing.views import PrintingOrderViewSet
 
-# 创建 DRF Router
-router = DefaultRouter()
-router.register(r'states', stateflow.StateViewSet, basename='state')
-router.register(r'processes', stateflow.ProcessViewSet, basename='process')
-router.register(r'business-objects', stateflow.BusinessObjectViewSet, basename='business-object')
+# 创建 DRF Router for Stateflow
+stateflow_router = DefaultRouter()
+stateflow_router.register(r'states', stateflow.StateViewSet, basename='state')
+stateflow_router.register(r'processes', stateflow.ProcessViewSet, basename='process')
+
+# 创建主 Router
+main_router = DefaultRouter()
+main_router.register(r'printing-orders', PrintingOrderViewSet, basename='printing-order')
 
 urlpatterns = [
     # 库存变动相关API
+    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/<int:record_id>/', stock_change_views.get_stock_change, name='get_stock_change'),
@@ -29,5 +35,8 @@ urlpatterns = [
     path('products/<int:product_id>/image/', product_image.ProductImageUploadView.as_view(), name='product_image_upload'),
     
     # Stateflow API (使用 Router)
-    path('stateflow/', include(router.urls)),
+    path('stateflow/', include(stateflow_router.urls)),
+    
+    # 主 Router (printing-orders 等)
+    path('', include(main_router.urls)),
 ]

api_v1/views/printing/test_api.py

@@ -0,0 +1,400 @@
+"""
+PrintingOrder API 测试
+"""
+from django.test import TestCase
+from rest_framework.test import APIClient
+from rest_framework import status
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import Permission
+from basic_info import models as basic_models
+from printing import models as printing_models
+
+User = get_user_model()
+
+
+class PrintingOrderAPITestCase(TestCase):
+    """测试 PrintingOrder API"""
+    
+    def setUp(self):
+        self.client = APIClient()
+        
+        # 创建商户
+        self.merchant = basic_models.Merchant.objects.create(
+            name='测试印花厂',
+            type=basic_models.MerchantTypeEnum.FACTORY
+        )
+        
+        # 创建用户
+        self.user = User.objects.create_user(
+            username='testuser',
+            password='testpass123',
+            email='test@example.com'
+        )
+        
+        # 创建员工并关联商户
+        self.employee = basic_models.Employee.objects.create(
+            sys_user=self.user,
+            merchant=self.merchant,
+            name='测试员工',
+            mobile='13800138000',
+            job_type=basic_models.EmployeeTypeEnum.PRINTER,
+            status=basic_models.EmployeeStatusEnum.ACTIVE
+        )
+        
+        # 创建客户
+        self.customer = basic_models.Customer.objects.create(
+            merchant=self.merchant,
+            name='测试客户',
+            mobile='13900139000',
+            area='测试地区'
+        )
+        
+        # 认证用户
+        self.client.force_authenticate(user=self.user)
+        
+        # 给用户添加基础权限
+        view_perm = Permission.objects.get(codename='view_printingorder')
+        add_perm = Permission.objects.get(codename='add_printingorder')
+        change_perm = Permission.objects.get(codename='change_printingorder')
+        self.user.user_permissions.add(view_perm, add_perm, change_perm)
+    
+    def test_create_printing_order(self):
+        """测试创建印染订单"""
+        data = {
+            'customer': self.customer.id,
+            'fabric': '纯棉布料',
+            'width': '150cm',
+            'is_urgent': True,
+            'area': '广州',
+            'address': '白云区xxx',
+            'fabric_source': '客户提供',
+            'is_fabric_received': False,
+            'craft': '活性印花',
+            'description': '测试订单描述',
+            'outgoing_date': '2025-11-20',
+            'printing_warn': '注意颜色',
+            'rolling_warn': '注意温度',
+            'production_warn': '质量检查'
+        }
+        
+        response = self.client.post('/api/v1/printing-orders/', data, format='json')
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        
+        # 检查响应数据
+        self.assertIn('fabric', response.data)
+        self.assertEqual(response.data['fabric'], '纯棉布料')
+        self.assertEqual(response.data['is_urgent'], True)
+        
+        # 验证订单已创建 - 通过fabric查找，因为响应可能不包含id
+        order = printing_models.PrintingOrder.objects.filter(fabric='纯棉布料').first()
+        self.assertIsNotNone(order)
+        self.assertEqual(order.customer.id, self.customer.id)
+        self.assertEqual(order.fabric, '纯棉布料')
+    
+    def test_list_printing_orders(self):
+        """测试获取订单列表"""
+        # 创建测试订单
+        printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='布料1',
+            width='150cm'
+        )
+        printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='布料2',
+            width='160cm'
+        )
+        
+        response = self.client.get('/api/v1/printing-orders/')
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        # 不使用分页参数时，返回列表
+        self.assertEqual(len(response.data), 2)
+    
+    def test_retrieve_printing_order(self):
+        """测试获取订单详情"""
+        order = printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='测试布料',
+            width='150cm',
+            craft='活性印花',
+            description='详情测试'
+        )
+        
+        response = self.client.get(f'/api/v1/printing-orders/{order.id}/')
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data['fabric'], '测试布料')
+        self.assertEqual(response.data['craft'], '活性印花')
+        self.assertIn('customer_name', response.data)
+        self.assertEqual(response.data['customer_name'], self.customer.name)
+    
+    def test_update_printing_order(self):
+        """测试更新订单"""
+        order = printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='旧布料',
+            width='150cm'
+        )
+        
+        update_data = {
+            'customer': self.customer.id,
+            'fabric': '新布料',
+            'width': '160cm',
+            'is_urgent': True,
+            'craft': '更新的工艺'
+        }
+        
+        response = self.client.put(
+            f'/api/v1/printing-orders/{order.id}/',
+            update_data,
+            format='json'
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        
+        order.refresh_from_db()
+        self.assertEqual(order.fabric, '新布料')
+        self.assertEqual(order.width, '160cm')
+        self.assertEqual(order.is_urgent, True)
+    
+    def test_partial_update_printing_order(self):
+        """测试部分更新订单"""
+        order = printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='原布料',
+            width='150cm',
+            is_urgent=False
+        )
+        
+        patch_data = {
+            'is_urgent': True,
+            'craft': '新工艺'
+        }
+        
+        response = self.client.patch(
+            f'/api/v1/printing-orders/{order.id}/',
+            patch_data,
+            format='json'
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        
+        order.refresh_from_db()
+        self.assertEqual(order.is_urgent, True)
+        self.assertEqual(order.craft, '新工艺')
+        self.assertEqual(order.fabric, '原布料')  # 未修改字段保持不变
+    
+    def test_delete_printing_order_forbidden(self):
+        """测试删除订单被禁用"""
+        # 添加删除权限以便测试destroy方法的自定义逻辑
+        delete_perm = Permission.objects.get(codename='delete_printingorder')
+        self.user.user_permissions.add(delete_perm)
+        
+        order = printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='测试布料',
+            width='150cm'
+        )
+        
+        response = self.client.delete(f'/api/v1/printing-orders/{order.id}/')
+        self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)
+        self.assertIn('不支持删除', response.data['detail'])
+        
+        # 验证订单仍然存在
+        self.assertTrue(
+            printing_models.PrintingOrder.objects.filter(id=order.id).exists()
+        )
+    
+    def test_invalidate_printing_order(self):
+        """测试作废订单"""
+        # 添加作废权限
+        invalidate_perm = Permission.objects.get(codename='can_invalidate_printingorder')
+        self.user.user_permissions.add(invalidate_perm)
+        
+        order = printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='测试布料',
+            width='150cm',
+            is_invalid=False
+        )
+        
+        response = self.client.post(f'/api/v1/printing-orders/{order.id}/invalidate/')
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertIn('已作废', response.data['detail'])
+        
+        order.refresh_from_db()
+        self.assertTrue(order.is_invalid)
+    
+    def test_invalidate_without_permission(self):
+        """测试没有权限时作废订单失败"""
+        order = printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='测试布料',
+            width='150cm'
+        )
+        
+        response = self.client.post(f'/api/v1/printing-orders/{order.id}/invalidate/')
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        self.assertIn('没有权限', response.data['detail'])
+    
+    def test_activate_printing_order(self):
+        """测试恢复订单"""
+        # 添加恢复权限
+        activate_perm = Permission.objects.get(codename='can_activate_printingorder')
+        self.user.user_permissions.add(activate_perm)
+        
+        order = printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='测试布料',
+            width='150cm',
+            is_invalid=True
+        )
+        
+        response = self.client.post(f'/api/v1/printing-orders/{order.id}/activate/')
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertIn('已恢复', response.data['detail'])
+        
+        order.refresh_from_db()
+        self.assertFalse(order.is_invalid)
+    
+    def test_activate_without_permission(self):
+        """测试没有权限时恢复订单失败"""
+        order = printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='测试布料',
+            width='150cm',
+            is_invalid=True
+        )
+        
+        response = self.client.post(f'/api/v1/printing-orders/{order.id}/activate/')
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        self.assertIn('没有权限', response.data['detail'])
+    
+    def test_mark_fabric_received(self):
+        """测试标记布料已收"""
+        order = printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='测试布料',
+            width='150cm',
+            is_fabric_received=False
+        )
+        
+        response = self.client.post(f'/api/v1/printing-orders/{order.id}/mark_fabric_received/')
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertIn('已标记布料已收', response.data['detail'])
+        
+        order.refresh_from_db()
+        self.assertTrue(order.is_fabric_received)
+    
+    def test_mark_fabric_received_already_received(self):
+        """测试重复标记布料已收"""
+        order = printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='测试布料',
+            width='150cm',
+            is_fabric_received=True
+        )
+        
+        response = self.client.post(f'/api/v1/printing-orders/{order.id}/mark_fabric_received/')
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('已经标记', response.data['detail'])
+    
+    def test_filter_by_customer(self):
+        """测试按客户过滤"""
+        customer2 = basic_models.Customer.objects.create(
+            merchant=self.merchant,
+            name='客户2',
+            mobile='13900139001'
+        )
+        
+        printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='布料1',
+            width='150cm'
+        )
+        printing_models.PrintingOrder.objects.create(
+            customer=customer2,
+            fabric='布料2',
+            width='160cm'
+        )
+        
+        response = self.client.get(f'/api/v1/printing-orders/?customer={self.customer.id}')
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(response.data), 1)
+    
+    def test_filter_by_urgent(self):
+        """测试按紧急状态过滤"""
+        printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='布料1',
+            width='150cm',
+            is_urgent=True
+        )
+        printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='布料2',
+            width='160cm',
+            is_urgent=False
+        )
+        
+        response = self.client.get('/api/v1/printing-orders/?is_urgent=true')
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(response.data), 1)
+        self.assertTrue(response.data[0]['is_urgent'])
+    
+    def test_search_by_fabric(self):
+        """测试按布料搜索"""
+        printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='纯棉布料',
+            width='150cm'
+        )
+        printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='涤纶布料',
+            width='160cm'
+        )
+        
+        response = self.client.get('/api/v1/printing-orders/?search=纯棉')
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(response.data), 1)
+        self.assertIn('纯棉', response.data[0]['fabric'])
+    
+    def test_ordering(self):
+        """测试排序"""
+        order1 = printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='布料1',
+            width='150cm'
+        )
+        order2 = printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='布料2',
+            width='160cm'
+        )
+        
+        # 按 ID 升序
+        response = self.client.get('/api/v1/printing-orders/?ordering=id')
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        results = response.data
+        self.assertEqual(results[0]['id'], order1.id)
+        self.assertEqual(results[1]['id'], order2.id)
+        
+        # 按 ID 降序
+        response = self.client.get('/api/v1/printing-orders/?ordering=-id')
+        results = response.data
+        self.assertEqual(results[0]['id'], order2.id)
+        self.assertEqual(results[1]['id'], order1.id)
+    
+    def test_human_id_generation(self):
+        """测试 human_id 自动生成"""
+        from datetime import date
+        
+        order = printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='测试布料',
+            width='150cm'
+        )
+        
+        # human_id 格式: YYYYMMDD000001
+        self.assertIsNotNone(order.human_id)
+        self.assertTrue(len(order.human_id) >= 14)
+        today = date.today().strftime('%Y%m%d')
+        self.assertTrue(order.human_id.startswith(today))

api_v1/views/stateflow/business_object.py

@@ -139,7 +139,7 @@ class BusinessObjectViewSet(viewsets.ModelViewSet):
     def reset(self, request, pk=None):
         """重置进度"""
         business_object = self.get_object()
-        services.reset_order_progress(business_object)
+        services.reset_business_object_progress(business_object)
         
         return Response({
             'success': True,

api_v1/views/stock_change_views/snapshot.py

@@ -0,0 +1,98 @@
+from rest_framework.generics import ListAPIView
+from rest_framework.pagination import LimitOffsetPagination
+from rest_framework import filters, serializers
+from django_filters.rest_framework import DjangoFilterBackend
+from django_filters import rest_framework as django_filters
+from stock import models
+
+
+class StockSnapshotSerializer(serializers.ModelSerializer):
+    """库存快照序列化器"""
+    product_name = serializers.CharField(source='product.name', read_only=True)
+    product_code = serializers.CharField(source='product.code', read_only=True)
+    warehouse_name = serializers.CharField(source='warehouse.name', read_only=True)
+    unit_display = serializers.CharField(source='get_unit_display', read_only=True)
+    stock_change_record_type = serializers.CharField(
+        source='stock_change_record.get_type_display', 
+        read_only=True
+    )
+    stock_change_record_source = serializers.CharField(
+        source='stock_change_record.get_source_type_display', 
+        read_only=True
+    )
+    
+    class Meta:
+        model = models.StockSnapshot
+        fields = [
+            'id', 'product', 'product_name', 'product_code',
+            'warehouse', 'warehouse_name',
+            'delta', 'quantity_before', 'quantity_after',
+            'stock_change_record', 'stock_change_record_type', 'stock_change_record_source',
+            'unit', 'unit_display', 'num_of_rolls',
+            'offset_to', 'offset_at', 'cancelled', 'cancelled_at', 'offset_id',
+            'created_at', 'updated_at'
+        ]
+
+
+class StockSnapshotFilterSet(django_filters.FilterSet):
+    """库存快照过滤器"""
+    product_name = django_filters.CharFilter(field_name='product__name', lookup_expr='icontains')
+    product_code = django_filters.CharFilter(field_name='product__code', lookup_expr='icontains')
+    warehouse_name = django_filters.CharFilter(field_name='warehouse__name', lookup_expr='icontains')
+    cancelled = django_filters.BooleanFilter()
+    has_offset = django_filters.BooleanFilter(method='filter_has_offset')
+    date_from = django_filters.DateFilter(field_name='created_at', lookup_expr='gte')
+    date_to = django_filters.DateFilter(field_name='created_at', lookup_expr='lte')
+    
+    def filter_has_offset(self, queryset, name, value):
+        """过滤是否有冲抵记录"""
+        if value:
+            return queryset.exclude(offset_to__isnull=True)
+        else:
+            return queryset.filter(offset_to__isnull=True)
+    
+    class Meta:
+        model = models.StockSnapshot
+        fields = [
+            'product', 'warehouse', 'stock_change_record',
+            'unit', 'cancelled'
+        ]
+
+
+class StockSnapshotListView(ListAPIView):
+    """
+    库存快照列表视图（只读）
+    
+    提供库存变动快照的查询功能
+    
+    查询参数:
+    - product: 产品ID
+    - product_name: 产品名称（模糊查询）
+    - product_code: 产品编码（模糊查询）
+    - warehouse: 仓库ID
+    - warehouse_name: 仓库名称（模糊查询）
+    - stock_change_record: 关联的库存变动记录ID
+    - unit: 单位
+    - cancelled: 是否已取消（true/false）
+    - has_offset: 是否有冲抵记录（true/false）
+    - date_from: 开始日期（格式：YYYY-MM-DD）
+    - date_to: 结束日期（格式：YYYY-MM-DD）
+    - search: 全文搜索（产品名称、产品编码、仓库名称）
+    - ordering: 排序字段
+    """
+    serializer_class = StockSnapshotSerializer
+    pagination_class = LimitOffsetPagination
+    filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
+    filterset_class = StockSnapshotFilterSet
+    search_fields = ['product__name', 'product__code', 'warehouse__name']
+    ordering_fields = [
+        'id', 'delta', 'quantity_before', 'quantity_after',
+        'created_at', 'updated_at', 'offset_at', 'cancelled_at'
+    ]
+    ordering = ['-created_at']
+    
+    def get_queryset(self):
+        """优化查询，预加载关联数据"""
+        return models.StockSnapshot.objects.select_related(
+            'product', 'warehouse', 'stock_change_record'
+        ).all()