ASJServer/AdminController/ProductsSchemeManageController.py

# -*- encoding: utf-8 -*-
"""
@File    : ProductsSchemeManageController.py
@Time    : 2025/5/13 11:52
@Author  : stephen
@Email   : zhangdongming@asj6.wecom.work
@Software: PyCharm
"""
import datetime
import json
import secrets
import string
import time
from io import BytesIO

import qrcode
from django.core.paginator import Paginator, EmptyPage
from django.db import transaction, IntegrityError
from django.db.models import Q
from django.http import QueryDict, HttpResponse
from django.views import View

from Ansjer.config import LOGGER
from Model.models import ProductsScheme, DeviceScheme
from Object.ResponseObject import ResponseObject
from Object.TokenObject import TokenObject


class ProductsSchemeManageView(View):
    def get(self, request, *args, **kwargs):
        request.encoding = 'utf-8'
        operation = kwargs.get('operation')
        return self.validation(request.GET, request, operation)

    def post(self, request, *args, **kwargs):
        request.encoding = 'utf-8'
        operation = kwargs.get('operation')
        return self.validation(request.POST, request, operation)

    def delete(self, request, *args, **kwargs):
        request.encoding = 'utf-8'
        operation = kwargs.get('operation')
        delete = QueryDict(request.body)
        if not delete:
            delete = request.GET
        return self.validation(delete, request, operation)

    def put(self, request, *args, **kwargs):
        request.encoding = 'utf-8'
        operation = kwargs.get('operation')
        put = QueryDict(request.body)
        return self.validation(put, request, operation)

    def validation(self, request_dict, request, operation):
        """请求验证路由"""
        # 初始化响应对象
        language = request_dict.get('language', 'cn')
        response = ResponseObject(language, 'pc')

        # Token验证
        try:
            tko = TokenObject(
                request.META.get('HTTP_AUTHORIZATION'),
                returntpye='pc')
            if tko.code != 0:
                return response.json(tko.code)
            response.lang = tko.lang
            user_id = tko.userID

        except Exception as e:
            LOGGER.error(f"Token验证失败: {str(e)}")
            return response.json(444)

        # 操作路由映射
        operation_handlers = {
            'queryList': self.query_list,  # 查询列表
            'add': self.add_scheme,  # 添加方案
            'edit': self.edit_scheme,  # 编辑方案
            'delete': self.delete_scheme,  # 删除方案
            'generateQR': self.generate_qr_code,  # 生成二维码
            'deviceSchemeList': self.device_scheme_list
        }

        handler = operation_handlers.get(operation)
        if not handler:
            return response.json(444, 'operation')

        try:
            return handler(user_id, request_dict, response)
        except Exception as e:
            LOGGER.error(f"操作{operation}执行异常:{repr(e)}")
            return response.json(500, "服务器内部错误")

    def query_list(self, user_id, request_dict, response):
        """查询方案列表（优化点：分页封装/字段映射）"""
        # 参数处理与验证
        try:
            page = int(request_dict.get('page', 1))
            page_size = min(int(request_dict.get('pageSize', 10)), 100)  # 限制最大分页大小
        except ValueError:
            return response.json(444, "分页参数错误")

        # 构建查询条件
        query = Q(deleted=False)

        # 特殊字段处理示例（如果需要）
        if order_number := request_dict.get('orderNumber'):
            query &= Q(order_number__icontains=order_number)
        if storage_code := request_dict.get('storageCode'):
            query &= Q(storage_code__icontains=storage_code)
        if device_type := request_dict.get('deviceType'):
            query &= Q(device_type=int(device_type))
        if flash := request_dict.get('flash'):
            query &= Q(flash__icontains=flash)
        if ddr := request_dict.get('ddr'):
            query &= Q(ddr__icontains=ddr)
        if main_controller := request_dict.get('mainController'):
            query &= Q(main_controller__icontains=main_controller)
        if wifi := request_dict.get('wifi'):
            query &= Q(wifi__icontains=wifi)
        if four_g := request_dict.get('fourG'):
            query &= Q(four_g__icontains=four_g)
        if ad := request_dict.get('ad'):
            query &= Q(ad__icontains=ad)
        if sensor := request_dict.get('sensor'):
            query &= Q(sensor__icontains=sensor)

        # 使用分页器
        queryset = ProductsScheme.objects.filter(query).order_by('-created_time')
        paginator = Paginator(queryset, page_size)

        try:
            page_obj = paginator.page(page)
        except EmptyPage:
            return response.json(444, "页码超出范围")

        # 序列化数据
        data = [self._scheme_to_dict(scheme) for scheme in page_obj]

        return response.json(0, {
            'list': data,
            'total': paginator.count,
            'currentPage': page_obj.number,
            'totalPages': paginator.num_pages
        })

    @staticmethod
    def generate_timestamp_code(device_type: int = 0) -> str:
        """
        方案1：基于时间戳的编码
        格式：PC+设备类型+年月日+时分秒+6位随机大写字符
        示例：PC020250519143022-ABCDEF
        """
        try:
            current_time = datetime.datetime.now()
            date_part = current_time.strftime("%Y%m%d%H%M%S")
            # 使用string.ascii_uppercase生成大写随机字符
            random_part = ''.join(secrets.choice(string.ascii_uppercase) for _ in range(3))
            return f"{date_part}-{device_type}-{random_part}"
        except Exception as e:
            LOGGER.error(f"生成编码失败: {repr(e)}")
            raise

    def add_scheme(self, user_id, request_dict, response):
        """新增方案（集成自动编码生成）"""
        try:
            device_type = int(request_dict.get('deviceType', 0))

            # 生成唯一storage_code
            storage_code = self.generate_timestamp_code(device_type)

            # 构造方案数据
            scheme_data = {
                'storage_code': storage_code,  # 使用生成的编码
                'order_number': request_dict.get('orderNumber', ''),
                'device_type': device_type,
                'flash': request_dict.get('flash', ''),
                'ddr': request_dict.get('ddr', ''),
                'main_controller': request_dict.get('mainController', ''),
                'wifi': request_dict.get('wifi', ''),
                'four_g': request_dict.get('fourG', ''),
                'ad': request_dict.get('ad', ''),
                'sensor': request_dict.get('sensor', ''),
                'order_quantity': int(request_dict.get('orderQuantity', 0)),
                'customer_code': request_dict.get('customerCode', ''),
                'phy': request_dict.get('phy', ''),
                'remark': request_dict.get('remark', ''),
                'created_time': int(time.time()),
                'updated_time': int(time.time()),
                'created_by': user_id,
                'updated_by': user_id
            }

            scheme = ProductsScheme.objects.create(**scheme_data)
            return response.json(0, {
                'id': scheme.id,
                'storageCode': storage_code  # 返回生成的编码
            })

        except IntegrityError as e:
            LOGGER.error(f"唯一性冲突: {str(e)}")
            return response.json(173, "数据已存在")
        except ValueError as e:
            return response.json(444, "参数类型错误")
        except Exception as e:
            LOGGER.exception(f"添加方案异常:{repr(e)}")
            return response.json(500, "生成编码失败")

        # 可编辑字段白名单（前端字段名: 模型字段名）

    EDITABLE_FIELDS = {
        'orderNumber': 'order_number',
        'deviceType': 'device_type',
        'flash': 'flash',
        'ddr': 'ddr',
        'mainController': 'main_controller',
        'wifi': 'wifi',
        'fourG': 'four_g',
        'ad': 'ad',
        'sensor': 'sensor',
        'orderQuantity': 'order_quantity',
        'customerCode': 'customer_code',
        'phy': 'phy',
        'remark': 'remark'
    }

    # 需要类型转换的字段配置（字段名: 转换函数）
    FIELD_CONVERTERS = {
        'deviceType': int,
        'orderQuantity': int
    }

    @transaction.atomic
    def edit_scheme(self, user_id, request_dict, response):
        """
        方案编辑接口（事务原子性保证）
        优化特性：
        1. 前端字段到模型字段的自动映射
        2. 字段级白名单过滤
        3. 智能类型转换
        4. 最小化更新字段
        5. 并发安全控制
        """
        # ================= 参数校验阶段 =================
        if 'id' not in request_dict:
            return response.json(444, "缺少方案ID")

        try:
            scheme_id = int(request_dict['id'])
        except (ValueError, TypeError):
            LOGGER.warning(f"非法方案ID: {request_dict.get('id')}")
            return response.json(444, "方案ID格式错误")

        # ================= 数据获取阶段 =================
        try:
            # 使用select_for_update锁定记录，防止并发修改
            scheme = ProductsScheme.objects.select_for_update().get(
                id=scheme_id,
                deleted=False
            )
        except ProductsScheme.DoesNotExist:
            LOGGER.info(f"方案不存在: {scheme_id}")
            return response.json(173, "方案不存在")
        except Exception as e:
            LOGGER.error(f"数据库查询异常: {str(e)}")
            return response.json(500, "系统错误")

        # ================= 数据处理阶段 =================
        update_fields = []
        update_data = {'updated_time': int(time.time()), 'updated_by': user_id}

        # 遍历所有请求参数
        for frontend_field, value in request_dict.items():
            # 1. 白名单校验
            if frontend_field not in self.EDITABLE_FIELDS:
                continue

            # 2. 获取对应的模型字段名
            model_field = self.EDITABLE_FIELDS[frontend_field]

            # 3. 类型转换处理
            if frontend_field in self.FIELD_CONVERTERS:
                try:
                    value = self.FIELD_CONVERTERS[frontend_field](value)
                except (ValueError, TypeError):
                    LOGGER.warning(f"字段类型错误 {frontend_field}={value}")
                    return response.json(444, f"{frontend_field}类型不合法")

            # 4. 值变更检查（避免无意义更新）
            if getattr(scheme, model_field) != value:
                update_data[model_field] = value
                update_fields.append(model_field)

        # 无实际修改时快速返回
        if not update_fields:
            LOGGER.debug("无字段变更")
            return response.json(0)

        # ================= 数据持久化阶段 =================
        try:
            # 动态生成更新SQL：UPDATE ... SET field1=%s, field2=%s
            ProductsScheme.objects.filter(id=scheme_id).update(**update_data)

            LOGGER.info(f"方案更新成功: {scheme_id} 修改字段: {update_fields}")
            return response.json(0)
        except IntegrityError as e:
            LOGGER.error(f"数据唯一性冲突: {str(e)}")
            return response.json(177)
        except Exception as e:
            LOGGER.exception("方案更新异常")
            return response.json(500, "系统错误")

    def delete_scheme(self, user_id, request_dict, response):
        """删除方案（优化点：逻辑删除优化）"""
        # 参数校验
        if 'id' not in request_dict:
            return response.json(444, "缺少方案ID")

        try:
            # 使用update直接进行逻辑删除，提高效率
            rows = ProductsScheme.objects.filter(
                id=request_dict['id'],
                deleted=False
            ).update(
                deleted=True,
                updated_by=user_id,
                updated_time=int(time.time())
            )

            if rows == 0:
                return response.json(173, "方案不存在")

            return response.json(0)
        except ValueError:
            return response.json(500, "方案ID格式错误")

    def _scheme_to_dict(self, scheme):
        """方案对象序列化（优化点：集中管理序列化逻辑）"""
        return {
            'id': scheme.id,
            'orderNumber': scheme.order_number,
            'storageCode': scheme.storage_code,
            'deviceType': scheme.device_type,
            'flash': scheme.flash,
            'ddr': scheme.ddr,
            'mainController': scheme.main_controller,
            'wifi': scheme.wifi,
            'fourG': scheme.four_g,
            'ad': scheme.ad,
            'sensor': scheme.sensor,
            'orderQuantity': scheme.order_quantity,
            'customerCode': scheme.customer_code,
            'phy': scheme.phy,
            'remark': scheme.remark,
            'createdTime': scheme.created_time,
            'createdBy': scheme.created_by
        }

    def generate_qr_code(self, user_id, request_dict, response):
        """生成方案二维码"""
        try:
            scheme_id = int(request_dict.get('scheme_id'))
            scheme = ProductsScheme.objects.get(id=scheme_id, deleted=False)

            # 创建二维码（示例生成包含方案ID+名称）
            qr = qrcode.QRCode(
                version=1,
                error_correction=qrcode.constants.ERROR_CORRECT_L,
                box_size=10,
                border=4,
            )

            # 组织二维码内容（根据业务需求自定义）
            qr_data = json.dumps({
                "storageCode": scheme.storage_code,
                "type": 'NVR' if scheme.device_type == 1 else 'IPC',
                "flash": scheme.flash,
                "DDR": scheme.ddr,
                "mainController": scheme.main_controller,
                "wifi": scheme.wifi,
                "4G": scheme.four_g,
                "AD": scheme.ad,
                "sensor": scheme.sensor,
                'customerCode': scheme.customer_code,
                'phy': scheme.phy,
                "orderQuantity": scheme.order_quantity,
                "created_time": datetime.datetime.fromtimestamp(scheme.created_time).strftime("%Y-%m-%d %H:%M:%S")
            })
            qr.add_data(qr_data)
            qr.make(fit=True)

            # 生成图片
            img = qr.make_image(fill_color="black", back_color="white")

            # 将图片转为字节流
            buffer = BytesIO()
            img.save(buffer, format="PNG")

            # 返回文件响应
            return HttpResponse(
                buffer.getvalue(),
                content_type="image/png",
                headers={
                    'Content-Disposition': f'attachment; filename="scheme_{scheme.id}_qr.png"'
                }
            )

        except ProductsScheme.DoesNotExist:
            return response.json(173, "方案不存在")
        except Exception as e:
            LOGGER.error(f"生成二维码失败: {str(e)}")
            return response.json(500, "生成失败")

    @staticmethod
    def device_scheme_list(user_id, request_dict, response):
        """
        查询设备方案列表
        @param request_dict: 请求参数
        @param response: 响应对象
        @return: 响应对象包含设备方案列表
        """
        serial_number = request_dict.get("serialNumber", None)
        storage_code = request_dict.get("storageCode", None)
        device_type = request_dict.get("deviceType", None)
        phone_model = request_dict.get("phoneModel", None)
        page = request_dict.get("page", 1)
        page_size = request_dict.get("pageSize", 20)

        try:
            # 获取设备方案查询集
            device_scheme_qs = DeviceScheme.objects.all()

            # 过滤条件
            if serial_number:
                device_scheme_qs = device_scheme_qs.filter(serial_number=serial_number)
            if storage_code:
                device_scheme_qs = device_scheme_qs.filter(storage_code__icontains=storage_code)
            if device_type:
                device_scheme_qs = device_scheme_qs.filter(device_type=device_type)
            if phone_model:
                device_scheme_qs = device_scheme_qs.filter(phone_model__icontains=phone_model)

            # 分页
            paginator = Paginator(device_scheme_qs.order_by('-created_time'), page_size)
            device_schemes = paginator.page(page)

            # 构建返回数据
            device_list = []
            for device in device_schemes.object_list:
                # 获取关联的产品方案信息
                product_scheme = ProductsScheme.objects.filter(storage_code=device.storage_code).first()

                device_data = {
                    'id': device.id,
                    'serialNumber': device.serial_number,
                    'deviceType': device.device_type,
                    'phoneModel': device.phone_model,
                    'storageCode': device.storage_code,
                    'createdTime': device.created_time,
                    'updatedTime': device.updated_time,
                }
                if product_scheme:
                    device_data.update({
                        'orderNumber': product_scheme.order_number if product_scheme.order_number else '',
                        'type': 'NVR' if product_scheme.device_type == 1 else 'IPC',
                        'flash': product_scheme.flash if product_scheme.flash else '',
                        'ddr': product_scheme.ddr if product_scheme.ddr else '',
                        'mainController': product_scheme.main_controller if product_scheme.main_controller else '',
                        'wifi': product_scheme.wifi if product_scheme.wifi else '',
                        'fourG': product_scheme.four_g if product_scheme.four_g else '',
                        'ad': product_scheme.ad if product_scheme.ad else '',
                        'phy': product_scheme.phy if product_scheme.phy else '',
                        'sensor': product_scheme.sensor if product_scheme.sensor else '',
                        'orderQuantity': product_scheme.order_quantity if product_scheme.order_quantity else '',
                        'customerCode': product_scheme.customer_code if product_scheme.customer_code else '',
                        'remark': product_scheme.remark if product_scheme.remark else ''
                    })
                device_list.append(device_data)

            data = {
                'list': device_list,
                'total': paginator.count,
            }

            return response.json(0, data)
        except Exception as e:
            print(e)
            return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))