operation_management_center/src/app/api/device.py

import json
import os

import xlrd
from flask import request, jsonify, current_app, Response
from flask_restx import Resource, Namespace, reqparse
from sqlalchemy import select, insert, update, delete
from sqlalchemy.orm import Session
from werkzeug.datastructures import FileStorage

from app.defines import StatesCode, Module, OperationType
from app.modle.device import DeviceType, SecurityDevice
from app.utils.jwt_util import login_required
from app.utils.save_log import save_log
from app.utils.util import to_dict
from config import Config

ns = Namespace('device', description='设备管理接口')

config = Config()

device_type = reqparse.RequestParser(bundle_errors=True)
device_type.add_argument(name='id', type=str, location='form', required=True, help='设备类别id')
device_type.add_argument(name='name', type=str, location='form', required=True, help='设备类别名称')


@ns.route('/device_type')
class DeviceTypeApi(Resource):
    method_decorators = [login_required]

    @ns.doc(id='device_list', description='设备类别列表')
    def get(self):
        """设备类型列表"""

        try:
            with Session(current_app.engine) as session:
                stmt = select(DeviceType)
                results = session.execute(stmt).scalars().all()

            save_log(request, Module.DEVICE, OperationType.INQUIRE, StatesCode.SUCCESS)

            return jsonify(code=StatesCode.SUCCESS, message='成功', data=to_dict(results))


        except Exception as e:
            save_log(request, Module.DEVICE, OperationType.INQUIRE, StatesCode.UNKNOWN_ERROR)
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))

    @ns.doc(id='add_device', description='添加设备类别')
    @ns.expect(device_type)
    def post(self):
        """添加设备类型"""

        device_name = request.form.get('name')

        if device_name is None:
            return jsonify(code=StatesCode.PARA_ERROR, message='设备类别不能为空')

        try:
            with Session(current_app.engine) as session:
                stmt = insert(DeviceType).values(
                    name=device_name
                )
                session.execute(stmt)
                session.commit()

            save_log(request, Module.DEVICE, OperationType.ADD, StatesCode.SUCCESS)

            return jsonify(code=StatesCode.SUCCESS, message='成功')

        except Exception as e:
            save_log(request, Module.DEVICE, OperationType.ADD, StatesCode.UNKNOWN_ERROR)
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))

    @ns.doc(id='update_device', description='修改设备类别')
    @ns.expect(device_type)
    def put(self):
        """修改设备类型"""

        device_type_id = request.form.get('id')
        device_name = request.form.get('name')

        if device_type_id is None or device_name is None:
            return jsonify(code=StatesCode.PARA_ERROR, message='设备类别或设备类别id不能为空')

        try:
            with Session(current_app.engine) as session:
                stmt = update(DeviceType).where(DeviceType.id == device_type_id).values(
                    name=device_name
                )
                session.execute(stmt)
                session.commit()

            save_log(request, Module.DEVICE, OperationType.UPDATE, StatesCode.SUCCESS)

            return jsonify(code=StatesCode.SUCCESS, message='成功')

        except Exception as e:
            save_log(request, Module.DEVICE, OperationType.UPDATE, StatesCode.UNKNOWN_ERROR)
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))


device_list = reqparse.RequestParser(bundle_errors=True)
device_list.add_argument(name='id', type=str, location='args', required=True, help='设备类别id')


@ns.route('/device_list')
class DeviceListApi(Resource):
    method_decorators = [login_required]

    @ns.doc(id='device_list', description='设备列表')
    @ns.expect(device_list)
    def get(self):
        """获取设备列表"""

        device_type_id = request.args.get('id')

        if device_type_id is None:
            return jsonify(code=StatesCode.PARA_ERROR, message='设备类别或设备类别id不能为空')

        try:
            with Session(current_app.engine) as session:
                stmt = select(SecurityDevice).where(SecurityDevice.device_group == device_type_id)
                results = session.execute(stmt).scalars().all()

            save_log(request, Module.DEVICE, OperationType.INQUIRE, StatesCode.SUCCESS)

            return jsonify(code=StatesCode.SUCCESS, message='成功', data=to_dict(results))

        except Exception as e:
            save_log(request, Module.DEVICE, OperationType.INQUIRE, StatesCode.UNKNOWN_ERROR)
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))


device = reqparse.RequestParser(bundle_errors=True)
device.add_argument(name='id', type=str, location='args', required=True, help='设备id')

device_details = reqparse.RequestParser(bundle_errors=True)
device_details.add_argument(name='id', type=str, location='form', required=False, help='设备id')
device_details.add_argument(name='device_name', type=str, location='form', required=False, help='设备名称')
device_details.add_argument(name='device_id', type=str, location='form', required=False, help='设备id')
device_details.add_argument(name='device_type', type=str, location='form', required=False, help='设备类型(1-摄像头)')
device_details.add_argument(name='device_location', type=str, location='form', required=False, help='设备位置')
device_details.add_argument(name='device_age', type=str, location='form', required=False, help='设备年限')
device_details.add_argument(name='company_id', type=str, location='form', required=False, help='所属单位ID')
device_details.add_argument(name='user_id', type=str, location='form', required=False, help='设备责任人')
device_details.add_argument(name='device_group', type=str, location='form', required=False, help='设备分组')
device_details.add_argument(name='status', type=str, location='form', required=False, help='设备状态(1-开启)')
device_details.add_argument(name='floor_id', type=str, location='form', required=False, help='楼层id')
device_details.add_argument(name='third_id', type=str, location='form', required=False, help='')


@ns.route('/device')
class DeviceApi(Resource):
    method_decorators = [login_required]

    @ns.doc(id='device', description='设备详情')
    @ns.expect(device)
    def get(self):
        """获取设备详情"""

        device_id = request.args.get('id')

        if device_id is None:
            return jsonify(code=StatesCode.PARA_ERROR, message='设备id不能为空')

        try:
            with Session(current_app.engine) as session:
                stmt = select(SecurityDevice).where(SecurityDevice.id == device_id)
                results = session.execute(stmt).scalars().all()

            save_log(request, Module.DEVICE, OperationType.INQUIRE, StatesCode.SUCCESS)

            return jsonify(code=StatesCode.SUCCESS, message='成功', data=to_dict(results))

        except Exception as e:
            save_log(request, Module.DEVICE, OperationType.INQUIRE, StatesCode.UNKNOWN_ERROR)
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))

    @ns.doc(id='add_device', description='添加设备详情')
    @ns.expect(device_details)
    def post(self):
        """添加设备"""

        device_name = request.form.get('device_name')
        device_id = request.form.get('device_id')
        device_type = request.form.get('device_type')
        device_location = request.form.get('device_location')
        device_age = request.form.get('device_age')
        company_id = request.form.get('company_id')
        user_id = request.form.get('user_id')
        device_group = request.form.get('device_group')
        status = request.form.get('status')
        floor_id = request.form.get('floor_id')
        third_id = request.form.get('third_id')

        if device_name is None:
            return jsonify(code=StatesCode.PARA_ERROR, message='设备名称不能为空')
        try:
            with Session(current_app.engine) as session:
                stmt = insert(SecurityDevice).values(
                    device_name=device_name,
                    device_id=device_id,
                    device_type=device_type,
                    device_location=device_location,
                    device_age=device_age,
                    company_id=company_id,
                    user_id=user_id,
                    device_group=device_group,
                    status=status,
                    floor_id=floor_id,
                    third_id=third_id
                )
                session.execute(stmt)
                session.commit()

            save_log(request, Module.DEVICE, OperationType.ADD, StatesCode.SUCCESS)

            return jsonify(code=StatesCode.SUCCESS, message='成功')

        except Exception as e:
            save_log(request, Module.DEVICE, OperationType.ADD, StatesCode.UNKNOWN_ERROR)
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))

    @ns.doc(id='update_device', description='添加设备')
    @ns.expect(device_details)
    def put(self):
        """修改设备"""

        id = request.form.get('id')
        device_name = request.form.get('device_name')
        device_id = request.form.get('device_id')
        device_type = request.form.get('device_type')
        device_location = request.form.get('device_location')
        device_age = request.form.get('device_age')
        company_id = request.form.get('company_id')
        user_id = request.form.get('user_id')
        device_group = request.form.get('device_group')
        status = request.form.get('status')
        floor_id = request.form.get('floor_id')
        third_id = request.form.get('third_id')

        if device_name is None:
            return jsonify(code=StatesCode.PARA_ERROR, message='设备名称不能为空')
        try:
            with Session(current_app.engine) as session:
                stmt = update(SecurityDevice).where(SecurityDevice.id == id).values(
                    device_name=device_name,
                    device_id=device_id,
                    device_type=device_type,
                    device_location=device_location,
                    device_age=device_age,
                    company_id=company_id,
                    user_id=user_id,
                    device_group=device_group,
                    status=status,
                    floor_id=floor_id,
                    third_id=third_id
                )
                session.execute(stmt)
                session.commit()

            save_log(request, Module.DEVICE, OperationType.UPDATE, StatesCode.SUCCESS)

            return jsonify(code=StatesCode.SUCCESS, message='成功')

        except Exception as e:
            save_log(request, Module.DEVICE, OperationType.UPDATE, StatesCode.UNKNOWN_ERROR)
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))

    @ns.doc(id='delete_device', description='删除设备')
    @ns.expect(device_details)
    def delete(self):
        """删除设备"""

        id = request.form.get('id')

        if id is None:
            return jsonify(code=StatesCode.PARA_ERROR, message='设备id不能为空')

        try:
            with Session(current_app.engine) as session:
                stmt = delete(SecurityDevice).where(SecurityDevice.id == id)
                session.execute(stmt)
                session.commit()

            save_log(request, Module.DEVICE, OperationType.DELETE, StatesCode.SUCCESS)

            return jsonify(code=StatesCode.SUCCESS, message='成功')

        except Exception as e:
            save_log(request, Module.DEVICE, OperationType.DELETE, StatesCode.UNKNOWN_ERROR)
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))


batch_delete_user = reqparse.RequestParser(bundle_errors=True)
batch_delete_user.add_argument(name='devices_id', type=str, required=True, location='form', help='设备id列表')


@ns.route('/batch_delete_device')
class BatchDeleteDeviceApi(Resource):
    method_decorators = [login_required]

    @ns.doc(id='batch_delete_device', description='批量删除设备')
    @ns.expect(batch_delete_user)
    def delete(self):
        """批量删除设备"""
        device_id = request.form.get('devices_id')

        if device_id is not None:
            device_id = json.loads(device_id)
        else:
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message='设备id不能为空')

        try:
            with Session(current_app.engine) as session:
                stmt = select(SecurityDevice).where(SecurityDevice.id.in_(device_id))
                results = session.execute(stmt).scalars().all()

                for result in results:
                    session.delete(result)
                session.commit()

            save_log(request, Module.DEVICE, OperationType.BATCH_DELETE, StatesCode.SUCCESS)

            return jsonify(code=StatesCode.SUCCESS, message='批量删除成功')

        except Exception as e:
            save_log(request, Module.DEVICE, OperationType.DELETE, StatesCode.UNKNOWN_ERROR)
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))


import_device = reqparse.RequestParser(bundle_errors=True)
import_device.add_argument(name='files', type=FileStorage, required=True, location='files', help='导入设备xls文件')
import_device.add_argument(name='group_id', type=str, required=True, location='form', help='设备类别id')


@ns.route('/import_device')
class ImportDeviceApi(Resource):
    method_decorators = [login_required]

    @ns.doc(id='import_device', description='导入模版下载')
    def get(self):
        """导入模版下载"""
        try:
            manage_path = os.path.dirname(os.path.abspath(__file__))
            config_yml_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(manage_path))), 'config',
                                           'device.xls')

            file_data = open(config_yml_path, 'rb')
            buf = file_data.read()
            file_data.close()

            response = Response(buf)
            response.headers['Content-Type'] = 'application/octet-stream'
            response.headers['Content-Disposition'] = 'attachment; filename=%s' % 'device.xls'

            save_log(request, Module.DEVICE, OperationType.EXPORT, StatesCode.SUCCESS)

            return response
        except Exception as e:
            save_log(request, Module.DEVICE, OperationType.EXPORT, StatesCode.UNKNOWN_ERROR)
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))

    @ns.doc(id='import_device', description='设备导入')
    @ns.expect(import_device)
    def post(self):
        """设备导入"""
        device_file = request.files.get('files')
        device_group_id = request.form.get('group_id')

        if os.path.splitext(device_file.name)[-1] != '.xls':
            return jsonify(code=StatesCode.PARA_ERROR, message='不支持的文件格式，仅支持 .xls 文件')

        if device_group_id is None:
            return jsonify(code=StatesCode.PARA_ERROR, message='设备类型id不能为空')

        try:
            xlsx = xlrd.open_workbook(file_contents=device_file.read())

            with Session(current_app.engine) as session:

                for sheet in xlsx.sheets():

                    # 添加设备数据
                    for i in range(1, sheet.nrows):  # 逐行打印sheet数据
                        value = sheet.row_values(i)
                        add_device = insert(SecurityDevice).values(
                            device_name=value[0],
                            device_id=value[1],
                            device_type=int(value[2]) if value[2] != '' else None,
                            device_location=value[3],
                            device_age=int(value[4]) if value[4] != '' else None,
                            company_id=value[5],
                            user_id=value[6],
                            device_group=device_group_id,
                            status=int(value[7]) if value[7] != '' else None,
                            floor_id=int(value[8]) if value[8] != '' else None,
                            third_id=value[9]
                        )
                        session.execute(add_device)
                        session.commit()

            save_log(request, Module.DEVICE, OperationType.IMPORT, StatesCode.SUCCESS)

            return jsonify(code=StatesCode.SUCCESS, message='成功')
        except Exception as e:
            save_log(request, Module.DEVICE, OperationType.IMPORT, StatesCode.UNKNOWN_ERROR)
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))