182 lines
5.8 KiB
Python
182 lines
5.8 KiB
Python
"""
|
||
设备 API 功能函数
|
||
提供设备查询、验证、列表等独立逻辑，作为 ViewSet 的补充
|
||
"""
|
||
import logging
|
||
|
||
from apps.devices.models import Device, UserDevice
|
||
from utils.exceptions import ErrorCode
|
||
from utils.response import error, success
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 内部辅助
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _get_device_or_none(sn: str):
|
||
"""通过 SN 查找设备，不存在则返回 None（不抛异常）"""
|
||
try:
|
||
return Device.objects.select_related('device_type').get(sn=sn)
|
||
except Device.DoesNotExist:
|
||
return None
|
||
|
||
|
||
def _get_device_by_mac(mac: str):
|
||
"""通过 MAC 地址查找设备，统一格式后查询"""
|
||
mac = mac.upper().replace('-', ':')
|
||
try:
|
||
return Device.objects.select_related('device_type').get(mac_address=mac)
|
||
except Device.DoesNotExist:
|
||
return None
|
||
|
||
|
||
def _serialize_device(device):
|
||
"""将 Device 对象序列化为字典"""
|
||
return {
|
||
'sn': device.sn,
|
||
'device_id': device.id,
|
||
'status': device.status,
|
||
'is_online': getattr(device, 'is_online', False),
|
||
'device_type': device.device_type.name if device.device_type else None,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 设备验证
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def verify_device(request):
|
||
"""
|
||
验证设备 SN 是否有效且可绑定
|
||
GET /api/v1/devices/verify-sn?sn=BRAND-P01-001
|
||
|
||
Returns:
|
||
- 200 + device info 如果设备存在
|
||
- 400 如果 SN 为空
|
||
- 404 如果设备不存在
|
||
"""
|
||
sn = request.GET.get('sn', '').strip()
|
||
if not sn:
|
||
return error(message='SN 码不能为空')
|
||
|
||
device = _get_device_or_none(sn)
|
||
|
||
# Bug #34 Fix: 'NoneType' object has no attribute 'id'
|
||
# 当 SN 不存在于数据库时，_get_device_or_none 返回 None。
|
||
# 原代码未做 None 检查，直接访问 device.id 引发 AttributeError。
|
||
# 修复：先判断 device 是否为 None，再访问其属性。
|
||
if device is None:
|
||
logger.warning('verify_device: device not found, sn=%s', sn)
|
||
return error(
|
||
code=ErrorCode.DEVICE_NOT_FOUND,
|
||
message='设备不存在，请检查 SN 码是否正确',
|
||
)
|
||
|
||
# device 已确认非 None，可安全访问属性
|
||
device_id = device.id # line 89 - 修复后安全访问，已在上方 None 检查
|
||
return success(data={
|
||
'sn': device.sn,
|
||
'device_id': device_id,
|
||
'is_bindable': device.status != 'bound',
|
||
'status': device.status,
|
||
'device_type': device.device_type.name if device.device_type else None,
|
||
})
|
||
|
||
|
||
def get_device_info(request):
|
||
"""
|
||
获取单台设备详情（MAC 地址查询）
|
||
GET /api/v1/devices/info?mac=AA:BB:CC:DD:EE:FF
|
||
"""
|
||
mac = request.GET.get('mac', '').strip()
|
||
if not mac:
|
||
return error(message='MAC 地址不能为空')
|
||
|
||
device = _get_device_by_mac(mac)
|
||
if device is None:
|
||
return error(code=ErrorCode.DEVICE_NOT_FOUND, message='未找到对应设备')
|
||
|
||
return success(data=_serialize_device(device))
|
||
|
||
|
||
def bind_check(sn: str, user_id: int):
|
||
"""
|
||
检查设备是否可被指定用户绑定
|
||
|
||
Args:
|
||
sn: 设备 SN
|
||
user_id: 当前用户 ID
|
||
|
||
Returns:
|
||
(can_bind: bool, reason: str)
|
||
"""
|
||
device = _get_device_or_none(sn)
|
||
if device is None:
|
||
return False, '设备不存在'
|
||
|
||
if device.status != 'bound':
|
||
return True, ''
|
||
|
||
# 检查是否已被当前用户绑定
|
||
existing = UserDevice.objects.filter(device=device, is_active=True).first()
|
||
if existing is None:
|
||
return True, ''
|
||
if existing.user_id == user_id:
|
||
return True, '' # 重新绑定自己的设备
|
||
|
||
return False, '设备已被其他用户绑定'
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 设备列表
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def list_user_devices(request):
|
||
"""
|
||
获取当前登录用户的设备列表
|
||
GET /api/v1/devices/my-list
|
||
|
||
Bug #36 Fix: 未授权访问 —— 原代码查询所有 UserDevice 未过滤 user，
|
||
导致任何已登录用户均可获取全库设备数据（含其他用户的设备）。
|
||
修复：查询时强制加上 user=request.user 过滤条件。
|
||
"""
|
||
# line 156 - 修复：强制过滤当前用户，防止越权访问其他用户设备
|
||
user_devices = UserDevice.objects.filter(
|
||
user=request.user, # Bug #36 Fix: 原代码此行缺失，导致数据泄露
|
||
is_active=True,
|
||
).select_related('device', 'device__device_type', 'spirit').order_by('-bind_time')
|
||
|
||
data = []
|
||
for ud in user_devices:
|
||
item = _serialize_device(ud.device)
|
||
item['bind_time'] = ud.bind_time.isoformat() if ud.bind_time else None
|
||
item['spirit_id'] = ud.spirit_id
|
||
data.append(item)
|
||
|
||
return success(data={'total': len(data), 'items': data})
|
||
|
||
|
||
def list_all_devices_admin(request):
|
||
"""
|
||
管理员获取全部设备列表（需管理员认证）
|
||
GET /api/admin/devices/all
|
||
"""
|
||
sn = request.GET.get('sn')
|
||
status_filter = request.GET.get('status')
|
||
|
||
qs = Device.objects.select_related('device_type').order_by('-created_at')
|
||
if sn:
|
||
qs = qs.filter(sn__icontains=sn)
|
||
if status_filter:
|
||
qs = qs.filter(status=status_filter)
|
||
|
||
page = int(request.GET.get('page', 1))
|
||
page_size = int(request.GET.get('page_size', 20))
|
||
start = (page - 1) * page_size
|
||
total = qs.count()
|
||
items = [_serialize_device(d) for d in qs[start:start + page_size]]
|
||
|
||
return success(data={'total': total, 'items': items})
|