1
0
forked from erp-dev/erp

feat: settlement first api beta

This commit is contained in:
2026-02-27 18:26:12 +08:00
parent c564b1af32
commit 86f5dab32b
16 changed files with 1854 additions and 10 deletions

View File

@@ -34,6 +34,7 @@ from .views.shipment import (
ShipmentDetailView,
ShipmentExternalCreateView,
)
from .views.settlement.views import PlateOrderSummaryView
# 创建 DRF Router for Stateflow
stateflow_router = DefaultRouter()
@@ -156,6 +157,13 @@ urlpatterns = [
name='sales_items_by_printing_order'
),
# Settlement API
path(
'settlement/plate-orders/summary/',
PlateOrderSummaryView.as_view(),
name='plate_order_summary'
),
# 主 Router (printing-orders 等)
path('', include(main_router.urls)),
]

View File

@@ -0,0 +1 @@
"""Settlement API views"""

View File

@@ -0,0 +1,320 @@
"""Settlement API 测试"""
from datetime import date, datetime
from django.test import TestCase
from django.utils import timezone
from rest_framework.test import APIClient
from rest_framework import status
from django.contrib.auth import get_user_model
from basic_info import models as basic_models
from printing import models as printing_models
User = get_user_model()
class PlateOrderSummaryAPITestCase(TestCase):
"""测试开版订单统计 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'
)
self.employee = basic_models.Employee.objects.create(
sys_user=self.user,
merchant=self.merchant,
name='测试员工'
)
self.customer = basic_models.Customer.objects.create(
merchant=self.merchant,
name='测试客户',
created_by=self.employee
)
self.client.force_authenticate(user=self.user)
def test_requires_date_parameter(self):
"""测试必须提供 date 参数"""
response = self.client.get('/api/v1/settlement/plate-orders/summary/')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn('缺少 date 参数', response.data['error'])
def test_invalid_date_format(self):
"""测试无效的日期格式"""
response = self.client.get(
'/api/v1/settlement/plate-orders/summary/?date=2026/02/08'
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn('日期格式错误', response.data['error'])
def test_invalid_date(self):
"""测试不存在的日期"""
response = self.client.get(
'/api/v1/settlement/plate-orders/summary/?date=2026-02-30'
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.assertIn('日期不存在', response.data['error'])
def test_requires_merchant(self):
"""测试用户必须关联商户"""
user_without_merchant = User.objects.create_user(
username='no-merchant',
password='pass123'
)
self.client.force_authenticate(user=user_without_merchant)
response = self.client.get(
'/api/v1/settlement/plate-orders/summary/?date=2026-02-08'
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.assertIn('用户未关联商户', response.data['error'])
def test_returns_summary_for_valid_date(self):
"""测试返回有效的统计数据"""
today = date(2026, 2, 8)
printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
plate_type='首版',
production_method='定位',
plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time()))
)
response = self.client.get(
'/api/v1/settlement/plate-orders/summary/?date=2026-02-08'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('data', response.data)
self.assertEqual(len(response.data['data']), 1)
self.assertEqual(response.data['data'][0]['client_id'], self.customer.id)
self.assertEqual(response.data['data'][0]['client_name'], '测试客户')
self.assertEqual(len(response.data['data'][0]['plate_order_count']), 1)
self.assertEqual(
response.data['data'][0]['plate_order_count'][0]['type'],
'首版-定位'
)
self.assertEqual(
response.data['data'][0]['plate_order_count'][0]['today'],
1
)
def test_calculates_current_month(self):
"""测试计算本月累计数量"""
today = date(2026, 2, 8)
month_start = date(2026, 2, 1)
printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
plate_type='首版',
production_method='定位',
plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time()))
)
printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
plate_type='首版',
production_method='定位',
plate_date=timezone.make_aware(datetime.combine(month_start, datetime.min.time()))
)
response = self.client.get(
'/api/v1/settlement/plate-orders/summary/?date=2026-02-08'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
response.data['data'][0]['plate_order_count'][0]['current_month'],
2
)
def test_filters_null_plate_type(self):
"""测试过滤 plate_type 为空的订单"""
today = date(2026, 2, 8)
printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
plate_type='首版',
production_method='定位',
plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time()))
)
printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
plate_type=None,
production_method='定位',
plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time()))
)
response = self.client.get(
'/api/v1/settlement/plate-orders/summary/?date=2026-02-08'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
response.data['data'][0]['plate_order_count'][0]['today'],
1
)
def test_filters_null_production_method(self):
"""测试过滤 production_method 为空的订单"""
today = date(2026, 2, 8)
printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
plate_type='首版',
production_method='定位',
plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time()))
)
printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
plate_type='首版',
production_method=None,
plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time()))
)
response = self.client.get(
'/api/v1/settlement/plate-orders/summary/?date=2026-02-08'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['data']), 1)
self.assertEqual(
response.data['data'][0]['plate_order_count'][0]['type'],
'首版-定位'
)
self.assertEqual(
response.data['data'][0]['plate_order_count'][0]['today'],
1
)
def test_filters_zero_data(self):
"""测试过滤全0数据"""
today = date(2026, 2, 8)
printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
plate_type='首版',
production_method='定位',
plate_date=timezone.make_aware(datetime.combine(date(2026, 1, 1), datetime.min.time()))
)
response = self.client.get(
'/api/v1/settlement/plate-orders/summary/?date=2026-02-08'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['data']), 0)
def test_applies_merchant_isolation(self):
"""测试应用商户隔离"""
other_merchant = basic_models.Merchant.objects.create(
name='其他商户',
type=basic_models.MerchantTypeEnum.FACTORY
)
other_customer = basic_models.Customer.objects.create(
merchant=other_merchant,
name='其他客户'
)
today = date(2026, 2, 8)
printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
plate_type='首版',
production_method='定位',
plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time()))
)
printing_models.PlateOrder.objects.create(
merchant=other_merchant,
customer=other_customer,
plate_type='首版',
production_method='定位',
plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time()))
)
response = self.client.get(
'/api/v1/settlement/plate-orders/summary/?date=2026-02-08'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['data']), 1)
self.assertEqual(response.data['data'][0]['client_id'], self.customer.id)
self.assertEqual(response.data['data'][0]['client_name'], '测试客户')
def test_applies_customer_visibility(self):
"""测试应用客户可见性过滤"""
other_employee = basic_models.Employee.objects.create(
sys_user=User.objects.create_user('other', 'pass123'),
merchant=self.merchant,
name='其他员工'
)
customer_visible = basic_models.Customer.objects.create(
merchant=self.merchant,
name='可见客户',
created_by=other_employee
)
customer_visible.visible_employees.set([self.employee])
customer_invisible = basic_models.Customer.objects.create(
merchant=self.merchant,
name='不可见客户',
created_by=other_employee
)
today = date(2026, 2, 8)
printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=customer_visible,
plate_type='首版',
production_method='定位',
plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time()))
)
printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=customer_invisible,
plate_type='首版',
production_method='定位',
plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time()))
)
response = self.client.get(
'/api/v1/settlement/plate-orders/summary/?date=2026-02-08'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['data']), 1)
self.assertEqual(response.data['data'][0]['client_id'], customer_visible.id)
self.assertEqual(response.data['data'][0]['client_name'], '可见客户')
def test_requires_authentication(self):
"""测试需要认证"""
self.client.force_authenticate(user=None)
response = self.client.get(
'/api/v1/settlement/plate-orders/summary/?date=2026-02-08'
)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

View File

@@ -0,0 +1,75 @@
"""Settlement API views"""
import logging
import re
from datetime import datetime, date
from django.utils import timezone
from rest_framework import status
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from settlement.services import get_plate_order_summary_by_customer
logger = logging.getLogger(__name__)
class PlateOrderSummaryView(APIView):
"""
开版订单统计 API
GET /api/v1/settlement/plate-orders/summary/
参数:
date: 统计日期YYYY-MM-DD
"""
permission_classes = [IsAuthenticated]
def get(self, request):
"""
获取开版订单统计
"""
date_str = request.query_params.get('date')
if not date_str:
return Response(
{'error': '缺少 date 参数'},
status=status.HTTP_400_BAD_REQUEST
)
if not re.match(r'^\d{4}-\d{2}-\d{2}$', date_str):
return Response(
{'error': '日期格式错误,请使用 YYYY-MM-DD 格式'},
status=status.HTTP_400_BAD_REQUEST
)
try:
settlement_date = date.fromisoformat(date_str)
except ValueError:
return Response(
{'error': '日期不存在'},
status=status.HTTP_403_FORBIDDEN
)
emp = getattr(request.user, 'employee', None)
if emp is None or emp.merchant is None:
return Response(
{'error': '用户未关联商户'},
status=status.HTTP_403_FORBIDDEN
)
try:
data = get_plate_order_summary_by_customer(
merchant_id=emp.merchant.id,
settlement_date=settlement_date,
user=request.user
)
return Response({'data': data})
except Exception as e:
logger.exception(
f'[settlement.views] 获取开版订单统计失败: {e}'
)
return Response(
{'error': '获取统计数据失败'},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)