from pathlib import Path from rest_framework.views import APIView from rest_framework.permissions import AllowAny from rest_framework import status from rest_framework import viewsets from django_filters.rest_framework import DjangoFilterBackend from rest_framework.filters import OrderingFilter from django.utils import timezone from django.db.models import Q, F, Case, When from django.core.cache import cache import requests from collections import defaultdict from utils.page import MyPageNumberPagination from .models import * from .serializers import * from .filter import * from .parsers import TextJSONParser from .parsers import TextJSONRenderer import logging import time from django.db import transaction from bound.models import BoundListModel, BoundBatchModel, OutBatchModel,BoundDetailModel,OutBoundDetailModel,BatchLogModel from warehouse.models import ProductListModel from rest_framework.response import Response logger = logging.getLogger('wms.boundBill') class WMSResponse: """ 入库申请专用响应格式 使用示例： return WMSResponse.success(data=serializer.data, total=2, success=2) return WMSResponse.error(message='验证失败', total=2, fail_count=1, fail_materials=[{...}]) """ @staticmethod def success(data, total, success, status=status.HTTP_200_OK): """成功响应格式""" logger.info('成功响应 | 数据: %s', data) return Response({ "status": True, "errorCode": status, "message": "success", "data": { "failCount": 0, "totalCount": total, "successCount": success, "fail_materials": [] } }, status=status) @staticmethod def error(message, total, fail_count, fail_materials=None, status=status.HTTP_200_OK, exception=None): """错误响应格式""" if fail_materials is None: fail_materials = [] # 记录详细错误日志 if exception: logger.error(f"入库申请错误: {message}", exc_info=exception) return Response({ "status": False, "errorCode": 400, "message": message, "data": { "failCount": fail_count, "totalCount": total, "successCount": total - fail_count, "fail_materials": fail_materials } }, status=status) """入库申请""" class InboundApplyCreate(APIView): """ 生产入库申请 """ authentication_classes = [] permission_classes = [AllowAny] def post(self, request): logger.info('生产入库申请请求 | 原始数据: %s', request.data) try: total_count = len(request.data.get('materials', [])) if total_count == 0 : return WMSResponse.error( message="物料清单不能为空", total=0, fail_count=0 ) if total_count != request.data.get('totalCount', 0): return WMSResponse.error( message="物料数量不匹配", total=total_count, fail_count=total_count ) serializer = boundPostSerializer(data=request.data) if not serializer.is_valid(): return self._handle_validation_error(serializer.errors, total_count) unique_result,error_details = self.find_unique_billid_and_number(serializer) if not unique_result: return WMSResponse.error( message="单据编号或原始单据ID重复", total=total_count, fail_count=total_count, fail_materials=[{ "entryIds": None, "production_batch": None, "errors": error_details }] ) # 保存或更新入库单 bound_bill = self.save_or_update_inbound_bill(serializer) # 保存或更新物料明细 self.save_or_update_material_detail(bound_bill, serializer) return WMSResponse.success( data=serializer.data, total=total_count, success=total_count ) except Exception as e: logger.exception("服务器内部错误") return WMSResponse.error( message="系统处理异常", total=total_count, fail_count=total_count, status=status.HTTP_500_INTERNAL_SERVER_ERROR, exception=e ) def _handle_validation_error(self, errors, total_count): """增强错误解析""" fail_materials = [] # 提取嵌套错误信息 material_errors = errors.get('materials', []) for error in material_errors: # 解析DRF的错误结构 if isinstance(error, dict) and 'metadata' in error: fail_materials.append({ "entryIds": error['metadata'].get('entryIds'), "production_batch": error['metadata'].get('production_batch'), "errors": { "missing_fields": error['metadata']['missing_fields'], "message": error['detail'] } }) return WMSResponse.error( message="物料数据不完整", total=total_count, fail_count=len(fail_materials), fail_materials=fail_materials ) def _format_material_errors(self, error_dict): """格式化单个物料的错误信息""" return { field: details[0] if isinstance(details, list) else details for field, details in error_dict.items() } def find_unique_billid_and_number(self, serializer): """增强版唯一性验证""" bill_id = serializer.validated_data['billId'] number = serializer.validated_data['number'] # 使用Q对象进行联合查询 duplicates_id = InboundBill.objects.filter( Q(billId=bill_id) ).only('billId') # 使用Q对象进行联合查询 duplicates_nu = InboundBill.objects.filter( Q(number=number) ).only( 'number') error_details = {} # 检查单据编号重复 if any(obj.billId != bill_id for obj in duplicates_nu): error_details['number'] = ["number入库单编码已存在,但是系统中与之前入库单单据ID不一致，请检查"] # 检查业务编号重复 if any(obj.number != number for obj in duplicates_id): error_details['billId'] = ["billId入库单单据ID已存在,但是系统中与之前入库申请单编码不一致，请检查"] if error_details: return False,error_details return True,None def save_or_update_inbound_bill(self, serializer): """保存或更新入库单""" # 保存或更新入库单 try: bound_bill = InboundBill.objects.get(billId=serializer.validated_data['billId']) bound_bill.number = serializer.validated_data['number'] bound_bill.type = serializer.validated_data['type'] bound_bill.date = serializer.validated_data['date'] bound_bill.department = (serializer.validated_data['department'] if 'department' in serializer.validated_data else '') bound_bill.warehouse = serializer.validated_data['warehouse'] bound_bill.creater = serializer.validated_data['creater'] bound_bill.note = (serializer.validated_data['note'] if 'note' in serializer.validated_data else '') bound_bill.totalCount = serializer.validated_data['totalCount'] if bound_bill.type != 1: bound_bill.erp_audit_id = bound_bill.number bound_bill.erp_save_id = bound_bill.billId bound_bill.update_time = timezone.now() bound_bill.save() except InboundBill.DoesNotExist: bound_bill = InboundBill.objects.create( billId=serializer.validated_data['billId'], number=serializer.validated_data['number'], type=serializer.validated_data['type'], date=serializer.validated_data['date'], department=(serializer.validated_data['department'] if 'department' in serializer.validated_data else ''), warehouse=serializer.validated_data['warehouse'], creater=serializer.validated_data['creater'], note=(serializer.validated_data['note'] if 'note' in serializer.validated_data else ''), totalCount=serializer.validated_data['totalCount'], erp_save_id = serializer.validated_data['billId'], create_time=timezone.now(), update_time=timezone.now() ) if bound_bill.type != 1: bound_bill.erp_audit_id = bound_bill.number bound_bill.save() return bound_bill def save_or_update_material_detail(self, bound_bill, serializer): """保存或更新物料明细""" # 保存或更新物料明细 for item in serializer.validated_data['materials']: try: material_detail = MaterialDetail.objects.get(bound_billId=bound_bill, entryIds=item['entryIds']) material_detail.production_batch = item['production_batch'] material_detail.goods_code = item['goods_code'] material_detail.goods_name = item['goods_name'] material_detail.goods_std = (item['goods_std'] if 'goods_std' in item else '') material_detail.plan_qty = item['plan_qty'] material_detail.goods_total_weight = item['plan_qty'] material_detail.goods_unit = item['goods_unit'] material_detail.note = (item['note'] if 'note' in item else '') material_detail.update_time = timezone.now() material_detail.save() except MaterialDetail.DoesNotExist: material_detail = MaterialDetail.objects.create( bound_billId=bound_bill, entryIds=item['entryIds'], production_batch=item['production_batch'], goods_code=item['goods_code'], goods_name=item['goods_name'], goods_std=(item['goods_std'] if 'goods_std' in item else ''), goods_weight=1, plan_qty=item['plan_qty'], goods_total_weight=item['plan_qty'], goods_unit=item['goods_unit'], note=(item['note'] if 'note' in item else ''), create_time=timezone.now(), update_time=timezone.now() ) return material_detail """入库单生成""" class GenerateInbound(APIView): """ 生产入库单生成接口 功能特性： 1. 防重复创建校验（状态校验+关联单据校验） 2. 事务级数据一致性保障 3. 批量操作优化 4. 完整日志追踪 """ def post(self, request): try: bill_id = request.data.get('billId') if not bill_id: return Response({"error": "缺少必要参数: billId"}, status=status.HTTP_400_BAD_REQUEST) # 开启原子事务 with transaction.atomic(): # 锁定原始单据并校验 bill_obj, bound_list = self.validate_and_lock(bill_id) # 创建出入库主单 bound_list = self.create_bound_list(bill_obj) logger.info(f"创建出入库主单成功 | bound_code: {bound_list.bound_code}") # 处理物料明细（批量操作优化） self.process_materials(bill_obj, bound_list) # 更新原始单据状态 bill_obj.bound_status = 1 bill_obj.save(update_fields=['bound_status']) logger.info(f"入库单生成成功 | billId: {bill_id} -> boundCode: {bound_list.bound_code}") return Response({ "code": 200, "count": 1, "next": "null", "previous": "null", "results":{ "bound_code": bound_list.bound_code, "batch_count": bill_obj.totalCount } }, status=status.HTTP_200_OK) except InboundBill.DoesNotExist: logger.error(f"原始单据不存在 | billId: {bill_id}") return Response({ "code": 404, "error": "原始单据不存在" }, status=status.HTTP_404_NOT_FOUND) except Exception as e: logger.exception(f"入库单生成异常 | billId: {bill_id}") return Response({ "code": 400, "error": str(e) }, status=status.HTTP_200_OK) def validate_and_lock(self, bill_id): """验证并锁定相关资源""" # 锁定原始单据 bill_obj = InboundBill.objects.select_for_update().get( billId=bill_id, is_delete=False ) logger.info(f"锁定原始单据成功 | billId: {bill_id}") logger.info(f"原始单据状态: {bill_obj.bound_status}") # 状态校验 if bill_obj.bound_status == 1: logger.warning(f"单据已生成过入库单 | status: {bill_obj.bound_status}") raise Exception("该单据已生成过入库单") # 关联单据校验（双重校验机制） existing_bound = BoundListModel.objects.filter( Q(bound_desc__contains=f"生产入库单{bill_obj.number}") | Q(relate_bill=bill_obj) ).first() if existing_bound: logger.warning(f"发现重复关联单据 | existingCode: {existing_bound.bound_code}") raise Exception(f"已存在关联入库单[{existing_bound.bound_code}]") return bill_obj, None def process_materials(self, bill_obj, bound_list): """批量处理物料明细""" materials = MaterialDetail.objects.filter( bound_billId=bill_obj, is_delete=False ).select_related('bound_billId') if not materials: raise Exception("入库单没有有效物料明细") # 批量创建对象列表 batch_list = [] detail_list = [] log_list = [] goods_counter = defaultdict(int) order_day=str(timezone.now().strftime('-%Y%m')) order_month=str(timezone.now().strftime('%Y%m')) for idx, material in enumerate(materials, 1): # 生成批次 data = {} qs_set = BoundBatchModel.objects.filter( goods_code=material.goods_code, bound_month=order_month, is_delete=False) goods_code = material.goods_code goods_counter[goods_code] += 1 len_qs_set = len(qs_set) + goods_counter[goods_code] print("len_qs_set", len_qs_set) data['bound_batch_order'] = int(order_day.split('-')[-1])*100 + len_qs_set data['bound_number'] = material.goods_code + order_day + str(len_qs_set).zfill(2) batch = BoundBatchModel( bound_number=data['bound_number'], sourced_number = material.production_batch, bound_month=bound_list.bound_month, bound_batch_order=data['bound_batch_order'], warehouse_code='W01', warehouse_name='立体仓', goods_code=material.goods_code, goods_desc=material.goods_name, goods_std=(material.goods_std if material.goods_std else ''), goods_unit=material.goods_unit, goods_qty=material.plan_qty, goods_weight=float(material.goods_weight), goods_total_weight=float(material.goods_total_weight), creater=bound_list.creater, openid=bound_list.openid, relate_material=material ) batch_list.append(batch) # 生成明细 detail_list.append(BoundDetailModel( bound_list=bound_list, bound_batch=batch, detail_code=f"{batch.bound_number}-DET", creater=bound_list.creater, openid=bound_list.openid )) # 生成日志 log_list.append(BatchLogModel( batch_id=batch, log_type=0, log_date=timezone.now(), goods_code=batch.goods_code, goods_desc=batch.goods_desc, goods_qty=batch.goods_qty, log_content=f"生产入库批次创建，来源单据：{bill_obj.number}", creater=batch.creater, openid=batch.openid )) # 批量写入数据库 BoundBatchModel.objects.bulk_create(batch_list) BoundDetailModel.objects.bulk_create(detail_list) BatchLogModel.objects.bulk_create(log_list) def create_bound_list(self, bill_obj): """创建出入库主单（带来源标识）""" if BoundListModel.objects.filter(relate_bill=bill_obj).exists(): return BoundListModel.objects.get(relate_bill=bill_obj) return BoundListModel.objects.create( bound_month=timezone.now().strftime("%Y%m"), bound_date=timezone.now().strftime("%Y-%m-%d"), bound_code=bill_obj.number, bound_code_type=bill_obj.type, bound_bs_type='B01', bound_type='in', bound_desc=f"生产入库单{bill_obj.number}", bound_department=(bill_obj.department if bill_obj.department else 'D99'), base_type=0, bound_status='101', creater=bill_obj.creater, openid='ERP', relate_bill=bill_obj ) def handle_exception(self, exc): """统一异常处理""" if isinstance(exc, InboundBill.DoesNotExist): return Response({"error": "原始单据不存在或已被删除"}, status=status.HTTP_404_NOT_FOUND) elif "重复" in str(exc): return Response({"error": str(exc)}, status=status.HTTP_409_CONFLICT) return super().handle_exception(exc) """出库单生成""" class GenerateOutbound(APIView): """ 生产出库单生成接口 功能特性： 1. 防重复创建校验（状态校验+关联单据校验） 2. 事务级数据一致性保障 3. 批量操作优化 4. 完整日志追踪 """ def post(self, request): try: bill_id = request.data.get('billId') if not bill_id: return Response({"error": "缺少必要参数: billId"}, status=status.HTTP_400_BAD_REQUEST) # 开启原子事务 with transaction.atomic(): # 锁定原始单据并校验 bill_obj, bound_list = self.validate_and_lock(bill_id) # 创建出库主单 bound_list = self.create_bound_list(bill_obj) logger.info(f"创建出入库主单成功 | bound_code: {bound_list.bound_code}") # 处理物料明细（批量操作优化） self.process_materials(bill_obj, bound_list) # 更新原始单据状态 bill_obj.bound_status = 1 bill_obj.save(update_fields=['bound_status']) logger.info(f"出库单生成成功 | billId: {bill_id} -> boundCode: {bound_list.bound_code}") return Response({ "code": 200, "count": 1, "next": "null", "previous": "null", "results":{ "bound_code": bound_list.bound_code, "batch_count": bill_obj.totalCount } }, status=status.HTTP_200_OK) except InboundBill.DoesNotExist: logger.error(f"原始单据不存在 | billId: {bill_id}") return Response({ "code": 404, "error": "原始单据不存在" }, status=status.HTTP_404_NOT_FOUND) except Exception as e: logger.exception(f"出库单生成异常 | billId: {bill_id}") return Response({ "code": 400, "error": str(e) }, status=status.HTTP_200_OK) def validate_and_lock(self, bill_id): """验证并锁定相关资源""" # 锁定原始单据 bill_obj = OutboundBill.objects.select_for_update().get( billId=bill_id, is_delete=False ) logger.info(f"锁定原始单据成功 | billId: {bill_id}") logger.info(f"原始单据状态: {bill_obj.bound_status}") # 状态校验 if bill_obj.bound_status == 1: logger.warning(f"单据已生成过出库单 | status: {bill_obj.bound_status}") raise Exception("该单据已生成过出库单") # 关联单据校验（双重校验机制） existing_bound = BoundListModel.objects.filter( Q(bound_desc__contains=f"生产出库单{bill_obj.number}") | Q(relate_out_bill=bill_obj) ).first() if existing_bound: logger.warning(f"发现重复关联单据 | existingCode: {existing_bound.bound_code}") raise Exception(f"已存在关联出库单[{existing_bound.bound_code}]") return bill_obj, None def process_materials(self, bill_obj, bound_list): """批量处理物料明细""" materials = OutMaterialDetail.objects.filter( bound_billId=bill_obj, is_delete=False ).select_related('bound_billId') if not materials: raise Exception("出库单没有有效物料明细") # 批量创建对象列表 batch_list = [] detail_list = [] log_list = [] for idx, material in enumerate(materials, 1): # 生成批次 MaterialDetail_obj = MaterialDetail.objects.get(entryIds=material.Material_entryIds.entryIds) batch_obj = BoundBatchModel.objects.get(relate_material=MaterialDetail_obj) batch = OutBatchModel( out_number = MaterialDetail_obj.production_batch, batch_number = batch_obj, out_date = timezone.now().strftime("%Y-%m-%d %H:%M:%S"), out_type = bill_obj.type, out_note = bill_obj.note, warehouse_code='W01', warehouse_name='立体仓', goods_code=MaterialDetail_obj.goods_code, goods_desc=MaterialDetail_obj.goods_name, goods_std=MaterialDetail_obj.goods_std, goods_unit=MaterialDetail_obj.goods_unit, goods_qty=batch_obj.goods_qty, goods_out_qty=material.goods_out_qty, status = 0, container_number = 0, goods_weight = 1, goods_total_weight = material.goods_out_qty, creater=bill_obj.creater, openid='ERP', relate_material=material ) batch_list.append(batch) # 生成明细 detail_list.append(OutBoundDetailModel( bound_list=bound_list, bound_batch=batch, bound_batch_number = batch_obj, detail_code=f"{batch.out_number}-ODET", creater=bound_list.creater, openid=bound_list.openid )) # 生成日志 log_list.append(BatchLogModel( batch_id=batch_obj, log_type=1, log_date=timezone.now(), goods_code=batch_obj.goods_code, goods_desc=batch_obj.goods_desc, goods_qty=batch.goods_out_qty, log_content=f"生产出库批次创建，来源单据：{bill_obj.number},出库件数：{batch.goods_out_qty}", creater=batch.creater, openid=batch.openid )) # 批量写入数据库 OutBatchModel.objects.bulk_create(batch_list) OutBoundDetailModel.objects.bulk_create(detail_list) BatchLogModel.objects.bulk_create(log_list) def create_bound_list(self, bill_obj): """创建出入库主单（带来源标识）""" if BoundListModel.objects.filter(relate_out_bill=bill_obj).exists(): return BoundListModel.objects.get(relate_out_bill=bill_obj) return BoundListModel.objects.create( bound_month=timezone.now().strftime("%Y%m"), bound_date=timezone.now().strftime("%Y-%m-%d"), bound_code=bill_obj.number, bound_code_type=bill_obj.type, bound_bs_type='B01', bound_type='out', bound_desc=f"生产出库单{bill_obj.number}", bound_department=(bill_obj.department if bill_obj.department else 'D99'), base_type=1, bound_status='201', creater=bill_obj.creater, openid='ERP', relate_out_bill=bill_obj ) def handle_exception(self, exc): """统一异常处理""" if isinstance(exc, OutboundBill.DoesNotExist): return Response({"error": "原始单据不存在或已被删除"}, status=status.HTTP_404_NOT_FOUND) elif "重复" in str(exc): return Response({"error": str(exc)}, status=status.HTTP_409_CONFLICT) return super().handle_exception(exc) """出库申请""" class OutboundApplyCreate(APIView): """ 生产出库申请 """ authentication_classes = [] permission_classes = [AllowAny] def post(self, request): logger.info('生产出库申请请求 | 原始数据: %s', request.data) try: total_count = len(request.data.get('materials', [])) if total_count == 0 : return WMSResponse.error( message="物料清单不能为空", total=0, fail_count=0 ) if total_count != request.data.get('totalCount', 0): return WMSResponse.error( message="物料数量不匹配", total=total_count, fail_count=total_count ) serializer = outboundPostSerializer(data=request.data) if not serializer.is_valid(): print("出错",serializer.errors) return self._handle_validation_error(serializer.errors, total_count) unique_result,error_details = self.find_unique_billid_and_number(serializer) if not unique_result: return WMSResponse.error( message="单据编号或原始单据ID重复", total=total_count, fail_count=total_count, fail_materials=[{ "entryIds": None, "production_batch": None, "errors": error_details }] ) # 保存或更新入库单 bound_bill = self.save_or_update_inbound_bill(serializer) # 保存或更新物料明细 self.save_or_update_material_detail(bound_bill, serializer) return WMSResponse.success( data=serializer.data, total=total_count, success=total_count ) except Exception as e: logger.exception("服务器内部错误") return WMSResponse.error( message="系统处理异常", total=total_count, fail_count=total_count, status=status.HTTP_500_INTERNAL_SERVER_ERROR, exception=e ) def _handle_validation_error(self, errors, total_count): """增强错误解析""" fail_materials = [] # 提取嵌套错误信息 material_errors = errors.get('materials', []) for error in material_errors: # 解析DRF的错误结构 if isinstance(error, dict) and 'metadata' in error: fail_materials.append({ "entryIds": error['metadata'].get('entryIds'), "production_batch": error['metadata'].get('production_batch'), "errors": { "missing_fields": error['metadata']['missing_fields'], "message": error['detail'] } }) return WMSResponse.error( message="物料数据不完整", total=total_count, fail_count=len(fail_materials), fail_materials=fail_materials) def _format_material_errors(self, error_dict): """格式化单个物料的错误信息""" return { field: details[0] if isinstance(details, list) else details for field, details in error_dict.items() } def find_unique_billid_and_number(self, serializer): """增强版唯一性验证""" bill_id = serializer.validated_data['billId'] number = serializer.validated_data['number'] # 使用Q对象进行联合查询 duplicates_id = OutboundBill.objects.filter( Q(billId=bill_id) ).only('billId') # 使用Q对象进行联合查询 duplicates_nu = OutboundBill.objects.filter( Q(number=number) ).only( 'number') error_details = {} # 检查单据编号重复 if any(obj.billId != bill_id for obj in duplicates_nu): error_details['number'] = ["number出库单编码已存在,但是系统中与之前出库单单据ID不一致，请检查"] # 检查业务编号重复 if any(obj.number != number for obj in duplicates_id): error_details['billId'] = ["billId出库库单单据ID已存在,但是系统中与之前出库申请单编码不一致，请检查"] if error_details: return False,error_details return True,None def save_or_update_inbound_bill(self, serializer): """保存或更新出库单""" try: bound_bill = OutboundBill.objects.get(billId=serializer.validated_data['billId']) bound_bill.number = serializer.validated_data['number'] bound_bill.type = serializer.validated_data['type'] bound_bill.date = serializer.validated_data['date'] bound_bill.department = (serializer.validated_data['department'] if 'department' in serializer.validated_data else '') bound_bill.warehouse = serializer.validated_data['warehouse'] bound_bill.creater = serializer.validated_data['creater'] bound_bill.note = (serializer.validated_data['note'] if 'note' in serializer.validated_data else '') bound_bill.totalCount = serializer.validated_data['totalCount'] bound_bill.erp_audit_id = bound_bill.number bound_bill.erp_save_id = bound_bill.billId bound_bill.update_time = timezone.now() bound_bill.save() except OutboundBill.DoesNotExist: bound_bill = OutboundBill.objects.create( billId=serializer.validated_data['billId'], number=serializer.validated_data['number'], type=serializer.validated_data['type'], date=serializer.validated_data['date'], department=(serializer.validated_data['department'] if 'department' in serializer.validated_data else ''), warehouse=serializer.validated_data['warehouse'], creater=serializer.validated_data['creater'], note=(serializer.validated_data['note'] if 'note' in serializer.validated_data else ''), totalCount=serializer.validated_data['totalCount'], erp_audit_id = serializer.validated_data['number'], erp_save_id = serializer.validated_data['billId'], create_time=timezone.now(), update_time=timezone.now() ) return bound_bill def save_or_update_material_detail(self, bound_bill, serializer): """保存或更新物料明细""" for item in serializer.validated_data['materials']: try: material_detail = OutMaterialDetail.objects.get(bound_billId=bound_bill, entryIds=item['entryIds']) Material_entryIds = MaterialDetail.objects.filter( goods_code = item['goods_code'], production_batch = item['production_batch'] ).first() if not Material_entryIds: logger.info("出库单号%s,更新——物料明细不存在",bound_bill.number) material_detail.Material_entryIds = Material_entryIds material_detail.production_batch = item['production_batch'] material_detail.goods_code = item['goods_code'] material_detail.goods_name = item['goods_name'] material_detail.goods_out_qty = item['goods_out_qty'] material_detail.goods_total_weight = item['goods_out_qty'] material_detail.goods_unit = item['goods_unit'] material_detail.note = (item['note'] if 'note' in item else '') material_detail.update_time = timezone.now() material_detail.save() except OutMaterialDetail.DoesNotExist: Material_entryIds = MaterialDetail.objects.filter( goods_code = item['goods_code'], production_batch = item['production_batch'] ).first() if not Material_entryIds: logger.info("出库单号%s,创建——物料明细不存在",bound_bill.number) material_detail = OutMaterialDetail.objects.create( bound_billId=bound_bill, entryIds=item['entryIds'], Material_entryIds=Material_entryIds, production_batch=item['production_batch'], goods_code=item['goods_code'], goods_name=item['goods_name'], goods_weight=1, goods_out_qty=item['goods_out_qty'], goods_total_weight=item['goods_out_qty'], goods_unit=item['goods_unit'], note=(item['note'] if 'note' in item else ''), create_time=timezone.now(), update_time=timezone.now() ) return material_detail """产品信息""" class ProductInfo(APIView): """ 批次信息更新 """ authentication_classes = [] # 禁用所有认证类 permission_classes = [AllowAny] # 允许任意访问 # parser_classes = [TextJSONParser] # 强制使用 text/json # renderer_classes = [TextJSONRenderer] # 强制使用 text/json def post(self, request): data = request.data logger.info('批次信息更新 | 原始数据: %s', data) total_count = data.get('totalCount', 0) materials = data.get('materials', []) success_count = 0 fail_count = 0 fail_materials = [] try: with transaction.atomic(): # 开启事务确保原子性 # 预查询已存在的产品ID existing_ids = set(ProductListModel.objects.filter( id__in=[m.get('id') for m in materials if m.get('id')] ).values_list('id', flat=True)) for material in materials: material_id = material.get('id') try: if material_id and material_id in existing_ids: instance = ProductListModel.objects.get(id=material_id) created = False else: instance = ProductListModel() created = True # 字段映射与校验 instance.id = material_id instance.product_code = material.get('product_code', instance.product_code) instance.product_name = material.get('product_name', instance.product_name) instance.product_std = material.get('product_std', instance.product_std or 'on std') instance.product_unit = material.get('product_unit', instance.product_unit or 'KG') # 必填字段校验[7](@ref) required_fields = ['product_code', 'product_name'] if any(not getattr(instance, field) for field in required_fields): raise ValueError(f"Missing required fields: {required_fields}") instance.is_delete = False # 强制重置删除标记[1](@ref) instance.full_clean() # 触发模型验证[3](@ref) instance.save() success_count += 1 except Exception as e: fail_count += 1 error_msg = f"{type(e).__name__}: {str(e)}" logger.warning(f"Material processing failed: {material} | Error: {error_msg}") fail_materials.append({ **material, "error": error_msg }) # 检查总数一致性[7](@ref) if success_count != total_count: return WMSResponse.error( message="物料数量不匹配", total=total_count, fail_count=total_count, fail_materials=fail_materials ) except Exception as e: logger.error(f"Batch update transaction failed: {str(e)}", exc_info=True) return WMSResponse.error( message=f"批量处理失败: {str(e)}", total=total_count, fail_count=fail_count, fail_materials=fail_materials, exception=e ) return WMSResponse.success( data=[], # 根据需求可返回处理后的数据 total=total_count, success=success_count ) """批次更新""" class BatchUpdate(APIView): """ 商品信息查询 """ authentication_classes = [] # 禁用所有认证类 permission_classes = [AllowAny] # 允许任意访问 def post(self, request): data = request.data total_count = data.get('totalCount', 0) material_audit_id = data.get('billnos') materials = data.get('materials', []) success_count = 0 fail_count = 0 fail_materials = [] try: with transaction.atomic(): # 开启事务确保原子性 if total_count != len(materials): return WMSResponse.error( message="物料数量不匹配", total=total_count, fail_count=total_count ) for material in materials: material_billId = material.get('billId') material_entryId = material.get('entryIds') material_status = material.get('status') if material_status == 'passing': try: instance = MaterialDetail.objects.get(bound_billId_id=material_billId,entryIds=material_entryId) if not instance: logger.info("物料明细不存在") fail_materials.append({ **material, "error": "物料明细不存在" }) fail_count += 1 continue instance.status = 1 bill_obj = InboundBill.objects.get(billId=material_billId) if not bill_obj: logger.info("入库单不存在") fail_materials.append({ **material, "error": "入库单不存在" }) fail_count += 1 continue bill_obj.erp_audit_id = material_audit_id logger.info("[1]入库单号%s,物料明细%s,更新状态,审核通过%s",bill_obj.number,material_entryId,material_audit_id) logger.info("[2]入库单号%s,物料明细%s,更新状态,审核通过%s",bill_obj.number,material_entryId,bill_obj.erp_audit_id) instance.save() bill_obj.save() success_count += 1 except Exception as e: fail_count += 1 error_msg = f"{type(e).__name__}: {str(e)}" logger.warning(f"Material processing failed: {material} | Error: {error_msg}") fail_materials.append({ **material, "error": error_msg }) # 检查总数一致性[7](@ref) if success_count != total_count: return WMSResponse.error( message="单据编码与详细编码不匹配", total=total_count, fail_count=total_count, fail_materials=fail_materials ) except Exception as e: logger.error(f"Batch update transaction failed: {str(e)}", exc_info=True) return WMSResponse.error( message=f"批量处理失败: {str(e)}", total=total_count, fail_count=fail_count, fail_materials=fail_materials, exception=e ) return WMSResponse.success( data=[], total=total_count, success=success_count ) """token""" class AccessToken(APIView): """ 获取ERP的access_token 方法：post到 https://okyy.test.kdgalaxy.com/kapi/oauth2/getToken 参数： { "client_id" : "WMS", "client_secret" : "1Ca~2Tu-3Fx$3Rg@", "username" : "xs", "accountId" : "2154719510106474496", "nonce" : "2025-04-27 11:36:08", "timestamp" : "2025-04-27 11:36:08", "language" : "zh_CN" } 返回： { "data": { "access_token": "CanAzMDM=", "token_type": "Bearer", "refresh_token": "4297d40e-b2ac-48b4-98a2-b50665e6faaf", "scope": "API", "expires_in": 7199990, "id_token": "d09UWXhOakV4TnpjMkE9EVGVV", "id_token_expires_in": 7199990, "language": "zh_CN" }, "errorCode": "0", "message": "", "status": true } """ authentication_classes = [] # 禁用所有认证类 permission_classes = [AllowAny] # 允许任意访问 @classmethod def get_token(cls): try: """获取access_token""" url = "https://okyy.test.kdgalaxy.com/kapi/oauth2/getToken" data = { "client_id" : "WMS", "client_secret" : "1Ca~2Tu-3Fx$3Rg@", "username" : "xs", "accountId" : "2154719510106474496", "nonce" : timezone.now().strftime("%Y-%m-%d %H:%M:%S"), "timestamp" : timezone.now().strftime("%Y-%m-%d %H:%M:%S"), "language" : "zh_CN" } print("请求参数",data) response = requests.post(url, json=data,timeout=10) if response.status_code == 200: result = response.json() if result.get('status'): logger.info(f"获取access_token成功 | access_token: {result.get('data',{}).get('access_token')}") return result.get('data',{}).get('access_token') return None except Exception as e: print("获取access_token异常",e) logger.exception("获取access_token异常") return None @classmethod def get_current_token(cls): """获取当前有效Token（带缓存机制）""" cache_key = 'erp_access_token' cached_token = cache.get(cache_key) if not cached_token: new_token = cls.get_token() if new_token: # 缓存时间略短于实际有效期 cache.set(cache_key, new_token, timeout=1000) return new_token return cached_token """基本同步业务类""" class ERPSyncBase: """ERP同步基类（作为发送方）""" max_retries = 30 # 最大重试次数 retry_delay = 3 # 重试间隔秒数 def __init__(self, wms_bill): self.wms_bill = wms_bill # WMS单据对象 self.erp_id_field = None # 需要更新的ERP ID字段名 def build_erp_payload(self): """构造ERP请求数据（需子类实现）""" raise NotImplementedError def get_erp_endpoint(self): """获取ERP接口地址（需子类实现）""" raise NotImplementedError def process_erp_response(self, response): """处理ERP响应（需子类实现）返回erp_id""" raise NotImplementedError def execute_sync(self): """执行同步操作""" # 测试环境 @ todo headers = { 'accessToken': f'{AccessToken.get_current_token()}', 'x-acgw-identity': 'djF8MTk2M2QzMWEzMjUwMTZlMzA3MDF8NDg5ODM4MzM4NjE3OHwYjYJyvo-DbkhOliEpFtiFOsCgKKo6braaiQGE9qdNx3w=' } for attempt in range(self.max_retries): try: print("请求头：",headers) print("请求体：",self.build_erp_payload()) print("请求地址：",self.get_erp_endpoint()) response = requests.post( self.get_erp_endpoint(), json=self.build_erp_payload(), headers=headers, timeout=10 ) response.raise_for_status() erp_id = self.process_erp_response(response.json()) return True except requests.exceptions.HTTPError as http_err: if response.status_code == 519: print("特定HTTP错误 519:", http_err) logger.error(f"ERP接口HTTP错误 519 第{attempt+1}次重试 | 单据:{self.wms_bill.number} | 错误: {http_err}") else: print("HTTP错误:", http_err) logger.error(f"ERP接口HTTP错误 第{attempt+1}次重试 | 单据:{self.wms_bill.number} | 错误: {http_err}") time.sleep(self.retry_delay) except requests.exceptions.RequestException as e: print("ERP接口请求异常:",e) print(f"ERP接口请求失败 第{attempt+1}次重试 | 单据:{self.wms_bill.number}") logger.error(f"ERP接口请求失败 第{attempt+1}次重试 | 单据:{self.wms_bill.number}") time.sleep(self.retry_delay) logger.error(f"ERP同步最终失败 | 单据:{self.wms_bill.number}") return False # ==================== 业务接口 ==================== """生产入库审核""" class ProductionInboundAuditSync(ERPSyncBase): erp_id_field = 'erp_audit_id' # 请求地址 def get_erp_endpoint(self): return "https://okyy.test.kdgalaxy.com/kapi/v2/l772/im/im_mdc_mftmanuinbill/audit" # 请求参数 # { # "data":{ # "billnos":[ # "AgTSC", # "fgBAH" # ] # } # } def build_erp_payload(self): return { "data": { "billnos": [ self.wms_bill.erp_audit_id, ] } } # 处理响应 def process_erp_response(self, response): print("ERP审核响应：",response) if response['status'] != True: raise ValueError(f"ERP审核失败: {response['message']}") return response['data'] """采购收料入库审核""" class PurchaseInboundAuditSync(ERPSyncBase): erp_id_field = 'erp_audit_id' # 请求地址 def get_erp_endpoint(self): return "https://okyy.test.kdgalaxy.com/kapi/v2/l772/im/im_purreceivebill/audit" # 请求参数 # { # "data":{ # "billnos":[ # "AgTSC", # "fgBAH" # ] # } # } def build_erp_payload(self): return { "data": { "billnos": [ self.wms_bill.erp_audit_id, ] } } # 处理响应 def process_erp_response(self, response): print("ERP审核响应：",response) if response['status'] != True: raise ValueError(f"ERP审核失败: {response['message']}") return response['data'] """其他入库审核""" class OtherInboundAuditSync(ERPSyncBase): erp_id_field = 'erp_audit_id' # 请求地址 def get_erp_endpoint(self): return "https://okyy.test.kdgalaxy.com/kapi/v2/l772/im/im_otherinbill/audit" # 请求参数 # { # "data":{ # "billnos":[ # "AgTSC", # "fgBAH" # ] # } # } def build_erp_payload(self): return { "data": { "billnos": [ self.wms_bill.erp_audit_id, ] } } # 处理响应 def process_erp_response(self, response): print("ERP审核响应：",response) if response['status'] != True: raise ValueError(f"ERP审核失败: {response['message']}") return response['data'] """调拨入库审核""" class TransferInboundAuditSync(ERPSyncBase): erp_id_field = 'erp_audit_id' # 请求地址 def get_erp_endpoint(self): return "https://okyy.test.kdgalaxy.com/kapi/v2/l772/im/im_transdirbill/audit" # 请求参数 # { # "data":{ # "billnos":[ # "AgTSC", # "fgBAH" # ] # } # } def build_erp_payload(self): return { "data": { "billnos": [ self.wms_bill.erp_audit_id, ] } } # 处理响应 def process_erp_response(self, response): print("ERP审核响应：",response) if response['status'] != True: raise ValueError(f"ERP审核失败: {response['message']}") return response['data'] """其他出库审核同步""" class OtherOutboundAuditSync(ERPSyncBase): erp_id_field = 'erp_audit_id' # 请求地址 def get_erp_endpoint(self): return "https://okyy.test.kdgalaxy.com/kapi/v2/l772/im/im_otheroutbill/audit" # 请求参数 # { # "data":{ # "billnos":[ # "AgTSC", # "fgBAH" # ] # } # } def build_erp_payload(self): return { "data": { "billnos": [ self.wms_bill.erp_audit_id, ] } } # 处理响应 def process_erp_response(self, response): print("ERP审核响应：",response) if response['status'] != True: raise ValueError(f"ERP审核失败: {response['message']}") return response['data'] """生产领料出库审核""" class ProductionOutboundAuditSync(ERPSyncBase): erp_id_field = 'erp_audit_id' # 请求地址 def get_erp_endpoint(self): return "https://okyy.test.kdgalaxy.com/kapi/v2/l772/im/im_mdc_mftproorder/audit" # 请求参数 # { # "data":{ # "billnos":[ # "AgTSC", # "fgBAH" # ] # } # } def build_erp_payload(self): return { "data": { "billnos": [ self.wms_bill.erp_audit_id, ] } } # 处理响应 def process_erp_response(self, response): print("ERP审核响应：",response) if response['status'] != True: raise ValueError(f"ERP审核失败: {response['message']}") return response['data'] """采购入库保存""" class PurchaseInboundSaveSync(ERPSyncBase): erp_id_field = 'erp_save_id' def get_erp_endpoint(self): return "https://okyy.test.kdgalaxy.com/kapi/v2/l772/im/im_purinbill/save" def build_erp_payload(self): # { # "purinbill":{ # "billId":"1745732021174", # "entryIds":[ # "1745732021087" # ] # } # } return { "purinbill": { "billId":self.wms_bill.erp_audit_id, "entryIds": [ self.wms_bill.number, ] } } def process_erp_response(self, response): print("ERP审核响应：",response) return response['data']['purchase_order_id'] """销售出库保存""" class SaleOutboundSaveSync(ERPSyncBase): erp_id_field = 'erp_save_id' def get_erp_endpoint(self): return "https://okyy.test.kdgalaxy.com/kapi/v2/l772/im/im_saloutbill/save" def build_erp_payload(self): # { # "purinbill":{ # "billId":"1745732021174", # "entryIds":[ # "1745732021087" # ] # } # } return { "purinbill": { "billId":self.wms_bill.erp_audit_id, "entryIds": [ self.wms_bill.number, ] } } def process_erp_response(self, response): print("ERP审核响应：",response) return response['data']['purchase_order_id'] """前端视图类·ERP入库""" class InboundBills(viewsets.ModelViewSet): """ retrieve: Response a data list（get） list: Response a data list（all） """ # authentication_classes = [] # 禁用所有认证类 # permission_classes = [AllowAny] # 允许任意访问 pagination_class = MyPageNumberPagination filter_backends = [DjangoFilterBackend, OrderingFilter, ] ordering_fields = ["update_time", "create_time"] filter_class = InboundBillFilter def get_project(self): # 获取项目ID，如果不存在则返回None try: id = self.kwargs.get('pk') return id except: return None def get_queryset(self): # 根据请求用户过滤查询集 id = self.get_project() if self.request.user: if id is None: return InboundBill.objects.filter(is_delete=False) else: return InboundBill.objects.filter(billId=id, is_delete=False) else: return InboundBill.objects.none() def get_serializer_class(self): # 根据操作类型选择合适的序列化器 if self.action in ['list', 'retrieve', ]: return InboundApplySerializer else: return self.http_method_not_allowed(request=self.request) """前端视图类·ERP出库""" class OutboundBills(viewsets.ModelViewSet): """ retrieve: Response a data list（get） list: Response a data list（all） """ # authentication_classes = [] # 禁用所有认证类 # permission_classes = [AllowAny] # 允许任意访问 pagination_class = MyPageNumberPagination filter_backends = [DjangoFilterBackend, OrderingFilter, ] ordering_fields = ["update_time", "create_time"] filter_class = OutboundBillFilter def get_project(self): # 获取项目ID，如果不存在则返回None try: id = self.kwargs.get('pk') return id except: return None def get_queryset(self): # 根据请求用户过滤查询集 id = self.get_project() if self.request.user: if id is None: return OutboundBill.objects.filter(is_delete=False) else: return OutboundBill.objects.filter(billId=id, is_delete=False) else: return OutboundBill.objects.none() def get_serializer_class(self): # 根据操作类型选择合适的序列化器 if self.action in ['list', 'retrieve', ]: return OutboundApplySerializer else: return self.http_method_not_allowed(request=self.request) class Materials(viewsets.ModelViewSet): """ retrieve: Response a data list（get） list: Response a data list（all） """ # authentication_classes = [] # 禁用所有认证类 # permission_classes = [AllowAny] # 允许任意访问 pagination_class = MyPageNumberPagination filter_backends = [DjangoFilterBackend, OrderingFilter, ] ordering_fields = ['id', "create_time", "update_time", ] filter_class = MaterialDetailFilter def get_project(self): try: id = self.kwargs.get('pk') return id except: return None def get_queryset(self): id = self.get_project() if self.request.user: if id is None: return MaterialDetail.objects.filter(is_delete=False) else: return MaterialDetail.objects.filter(id=id) else: return MaterialDetail.objects.none() def get_serializer_class(self): if self.action in ['retrieve', 'list']: return MaterialDetailSerializer else: return self.http_method_not_allowed(request=self.request) class OutMaterials(viewsets.ModelViewSet): """ retrieve: Response a data list（get） list: Response a data list（all） """ # authentication_classes = [] # 禁用所有认证类 # permission_classes = [AllowAny] # 允许任意访问 pagination_class = MyPageNumberPagination filter_backends = [DjangoFilterBackend, OrderingFilter, ] ordering_fields = ['id', "create_time", "update_time", ] filter_class = OutMaterialDetailFilter def get_project(self): try: id = self.kwargs.get('pk') return id except: return None def get_queryset(self): id = self.get_project() if self.request.user: if id is None: return OutMaterialDetail.objects.filter(is_delete=False) else: return OutMaterialDetail.objects.filter(id=id) else: return OutMaterialDetail.objects.none() def get_serializer_class(self): if self.action in ['retrieve', 'list']: return OutMaterialDetailSerializer else: return self.http_method_not_allowed(request=self.request)