""" 出入库模块服务层 - SN码生成、Excel导入导出 """ import io from datetime import datetime from django.db import transaction from openpyxl import Workbook, load_workbook from openpyxl.drawing.image import Image as XLImage import qrcode from io import BytesIO from apps.devices.models import DeviceType, DeviceBatch, Device def generate_production_week(date): """ 生成生产周格式: YYWXX :param date: 日期对象 :return: 如 25W45 """ year = date.year % 100 week = date.isocalendar()[1] return f"{year}W{week:02d}" def generate_sn_code(brand, product_code, production_week, batch_no, sequence): """ 生成SN码 格式: {品牌}-{产品代号}-{生产周}-{批次号}-{5位序列号} 示例: AL-DZBJ-ON-25W45-A01-00001 """ return f"{brand}-{product_code}-{production_week}-{batch_no}-{sequence:05d}" def generate_batch_devices(batch): """ 为批次生成设备和SN码 :param batch: DeviceBatch对象 :return: 生成的Device列表 """ device_type = batch.device_type brand = device_type.brand product_code = device_type.product_code production_week = batch.production_week or generate_production_week(batch.production_date) batch_no = batch.batch_no # 更新生产周 if not batch.production_week: batch.production_week = production_week batch.save(update_fields=['production_week']) devices = [] for i in range(1, batch.quantity + 1): sn = generate_sn_code(brand, product_code, production_week, batch_no, i) device = Device( sn=sn, device_type=device_type, batch=batch, status='in_stock' ) devices.append(device) # 批量创建 Device.objects.bulk_create(devices) # 更新批次状态 batch.status = 'generated' batch.save(update_fields=['status']) return devices def generate_qrcode_image(data, size=100): """ 生成二维码图片 :param data: 二维码内容 :param size: 尺寸 :return: BytesIO对象 """ qr = qrcode.QRCode( version=1, error_correction=qrcode.constants.ERROR_CORRECT_L, box_size=4, border=1, ) qr.add_data(data) qr.make(fit=True) img = qr.make_image(fill_color="black", back_color="white") img_io = BytesIO() img.save(img_io, format='PNG') img_io.seek(0) return img_io def export_batch_excel(batch): """ 导出批次SN码Excel 包含四列: SN码 / MAC地址 / 创建时间 / SN码二维码 :param batch: DeviceBatch对象 :return: BytesIO对象(Excel文件) """ wb = Workbook() ws = wb.active ws.title = f"批次{batch.batch_no}" # 设置表头 headers = ['SN码', 'MAC地址', '创建时间', 'SN码二维码'] for col, header in enumerate(headers, 1): ws.cell(row=1, column=col, value=header) # 设置列宽 ws.column_dimensions['A'].width = 35 ws.column_dimensions['B'].width = 20 ws.column_dimensions['C'].width = 22 ws.column_dimensions['D'].width = 15 # 获取批次设备 devices = Device.objects.filter(batch=batch).order_by('sn') for row, device in enumerate(devices, 2): ws.cell(row=row, column=1, value=device.sn) ws.cell(row=row, column=2, value=device.mac_address or '') ws.cell(row=row, column=3, value=device.created_at.strftime('%Y-%m-%d %H:%M:%S')) # 生成二维码并嵌入 try: qr_img = generate_qrcode_image(device.sn) img = XLImage(qr_img) img.width = 60 img.height = 60 ws.add_image(img, f'D{row}') ws.row_dimensions[row].height = 50 except Exception: ws.cell(row=row, column=4, value='[QR]') # 保存到内存 output = BytesIO() wb.save(output) output.seek(0) return output def import_mac_addresses(batch, file_obj): """ 导入MAC地址 :param batch: DeviceBatch对象 :param file_obj: Excel文件对象 :return: 导入结果字典 """ result = { 'total': 0, 'success': 0, 'failed': 0, 'errors': [] } try: wb = load_workbook(file_obj) ws = wb.active except Exception as e: result['errors'].append({'row': 0, 'error': f'文件格式错误: {str(e)}'}) return result # 跳过表头 rows = list(ws.iter_rows(min_row=2, values_only=True)) result['total'] = len(rows) for row_num, row in enumerate(rows, 2): if len(row) < 2: continue sn = row[0] mac = row[1] if not sn or not mac: continue # 统一MAC地址格式 mac = str(mac).upper().replace('-', ':').strip() # 验证MAC地址格式 import re if not re.match(r'^([0-9A-F]{2}:){5}[0-9A-F]{2}$', mac): result['failed'] += 1 result['errors'].append({ 'row': row_num, 'sn': sn, 'mac': mac, 'error': 'MAC地址格式不正确' }) continue try: device = Device.objects.get(sn=sn, batch=batch) except Device.DoesNotExist: result['failed'] += 1 result['errors'].append({ 'row': row_num, 'sn': sn, 'mac': mac, 'error': 'SN码不存在或不属于当前批次' }) continue # 检查MAC地址是否已被使用 existing = Device.objects.filter(mac_address=mac).exclude(id=device.id).first() if existing: result['failed'] += 1 result['errors'].append({ 'row': row_num, 'sn': sn, 'mac': mac, 'error': f'MAC地址已被设备 {existing.sn} 使用' }) continue # 更新MAC地址 device.mac_address = mac device.save(update_fields=['mac_address']) result['success'] += 1 return result