v6ole
8a8ae4ed57
1. devices.py: /tags 和 /export/csv 移到 /{device_id}(int) 之前，
避免 FastAPI 将 'tags'/'export' 当作 device_id 解析失败返回 422
2. audit.py: /logs/export/csv 移到 /logs/{log_id}(int) 之前，同上
3. auth.py: Header(...) → Header(None)，统一为手动 401 返回，
消除全局最后一个 Header(...) 导致的 422 隐患
Co-Authored-By: Claude <noreply@anthropic.com>
753 lines
27 KiB
Python
753 lines
27 KiB
Python
"""设备管理 API"""
|
||
import csv
|
||
import io
|
||
from fastapi import APIRouter, Depends, Query, HTTPException
|
||
from fastapi.responses import StreamingResponse
|
||
from sqlalchemy.orm import Session, joinedload
|
||
from sqlalchemy import asc, desc, distinct, or_
|
||
from pydantic import BaseModel
|
||
from typing import Optional
|
||
from app.core.database import get_db
|
||
from app.middleware.permission_middleware import require_permission
|
||
from app.models.device import ONUDevice, DeviceStatusHistory, OLTDevice, DeviceReplacement
|
||
from app.schemas.device import DeviceListResponse, ONUDeviceResponse, RebootResponse, OpticalPowerResponse
|
||
|
||
router = APIRouter(prefix="/api/devices", tags=["设备管理"])
|
||
|
||
|
||
@router.get("", response_model=DeviceListResponse)
|
||
def get_devices(
|
||
skip: int = Query(0, ge=0),
|
||
limit: int = Query(20, ge=1, le=100),
|
||
region: str = None,
|
||
school_name: str = None,
|
||
keyword: str = None,
|
||
status: str = None,
|
||
tag: str = None,
|
||
db: Session = Depends(get_db),
|
||
current: dict = Depends(require_permission('device.view')),
|
||
):
|
||
"""获取设备列表"""
|
||
# 子查询：每台设备最新一条状态记录
|
||
from sqlalchemy import func
|
||
latest_subq = (
|
||
db.query(
|
||
DeviceStatusHistory.onu_device_id,
|
||
func.max(DeviceStatusHistory.checked_at).label("max_checked_at")
|
||
)
|
||
.group_by(DeviceStatusHistory.onu_device_id)
|
||
.subquery()
|
||
)
|
||
latest_history = (
|
||
db.query(DeviceStatusHistory)
|
||
.join(
|
||
latest_subq,
|
||
(DeviceStatusHistory.onu_device_id == latest_subq.c.onu_device_id) &
|
||
(DeviceStatusHistory.checked_at == latest_subq.c.max_checked_at)
|
||
)
|
||
.subquery()
|
||
)
|
||
|
||
query = db.query(ONUDevice)
|
||
|
||
# 数据范围过滤：区域管理员只能看自己分配的区域，学校管理员只能看自己分配的学校
|
||
role = current.get('role', 'user')
|
||
if role == 'area_admin':
|
||
assigned = current.get('assigned_area') or ''
|
||
areas = [a.strip() for a in assigned.split(',') if a.strip()]
|
||
if areas:
|
||
query = query.filter(ONUDevice.region.in_(areas))
|
||
else:
|
||
# 未分配区域则看不到任何设备
|
||
query = query.filter(False)
|
||
elif role == 'school_admin':
|
||
assigned = current.get('assigned_school') or ''
|
||
schools = [s.strip() for s in assigned.split(',') if s.strip()]
|
||
if schools:
|
||
query = query.filter(ONUDevice.school_name.in_(schools))
|
||
else:
|
||
query = query.filter(False)
|
||
|
||
if region:
|
||
query = query.filter(ONUDevice.region == region)
|
||
if school_name:
|
||
query = query.filter(ONUDevice.school_name.contains(school_name))
|
||
if tag:
|
||
query = query.filter(ONUDevice.tags.contains(tag))
|
||
if keyword:
|
||
query = query.filter(
|
||
or_(
|
||
ONUDevice.mac_address.contains(keyword.lower()),
|
||
ONUDevice.building.contains(keyword),
|
||
ONUDevice.place_type.contains(keyword),
|
||
ONUDevice.school_name.contains(keyword),
|
||
ONUDevice.region.contains(keyword),
|
||
ONUDevice.room_number.contains(keyword),
|
||
)
|
||
)
|
||
if status in ("online", "offline", "unknown"):
|
||
if status in ("online", "offline"):
|
||
query = query.join(
|
||
latest_history,
|
||
ONUDevice.id == latest_history.c.onu_device_id
|
||
).filter(latest_history.c.status == status)
|
||
else:
|
||
# unknown：最新状态为 unknown，或没有任何状态记录
|
||
query = query.outerjoin(
|
||
latest_history,
|
||
ONUDevice.id == latest_history.c.onu_device_id
|
||
).filter(
|
||
or_(
|
||
latest_history.c.onu_device_id == None,
|
||
latest_history.c.status == "unknown"
|
||
)
|
||
)
|
||
|
||
# 多级排序：区域 > 学校名称 > 楼宇 > 房间号（均为升序）
|
||
query = query.order_by(
|
||
asc(ONUDevice.region),
|
||
asc(ONUDevice.school_name),
|
||
asc(ONUDevice.building),
|
||
asc(ONUDevice.room_number)
|
||
)
|
||
|
||
total = query.count()
|
||
items = query.offset(skip).limit(limit).all()
|
||
|
||
# 获取每个设备最新的状态（批量，避免 N+1）
|
||
device_ids = [item.id for item in items]
|
||
history_map = {}
|
||
if device_ids:
|
||
histories = (
|
||
db.query(DeviceStatusHistory)
|
||
.join(
|
||
latest_subq,
|
||
(DeviceStatusHistory.onu_device_id == latest_subq.c.onu_device_id) &
|
||
(DeviceStatusHistory.checked_at == latest_subq.c.max_checked_at)
|
||
)
|
||
.filter(DeviceStatusHistory.onu_device_id.in_(device_ids))
|
||
.all()
|
||
)
|
||
history_map = {h.onu_device_id: h for h in histories}
|
||
|
||
# 批量加载 OLT 信息
|
||
olt_ids = {item.olt_id for item in items if item.olt_id}
|
||
olt_map = {}
|
||
if olt_ids:
|
||
olts = db.query(OLTDevice).filter(OLTDevice.id.in_(olt_ids)).all()
|
||
olt_map = {o.id: o for o in olts}
|
||
|
||
result_items = []
|
||
for item in items:
|
||
latest_status = history_map.get(item.id)
|
||
olt = olt_map.get(item.olt_id)
|
||
|
||
item_dict = {
|
||
"id": item.id,
|
||
"mac_address": item.mac_address,
|
||
"olt_id": item.olt_id,
|
||
"region": item.region,
|
||
"school_name": item.school_name,
|
||
"building": item.building,
|
||
"place_type": item.place_type,
|
||
"room_number": item.room_number,
|
||
"notes": item.notes,
|
||
"status": latest_status.status if latest_status else None,
|
||
"distance_m": latest_status.distance_m if latest_status else None,
|
||
"slot_number": item.slot_number,
|
||
"port_number": item.port_number,
|
||
"port_id": item.port_id,
|
||
"model": item.model,
|
||
"olt_location": olt.location if olt else None,
|
||
"created_at": item.created_at
|
||
}
|
||
result_items.append(ONUDeviceResponse(**item_dict))
|
||
|
||
return {"total": total, "items": result_items}
|
||
|
||
|
||
@router.get("/regions")
|
||
def get_regions(
|
||
db: Session = Depends(get_db),
|
||
current: dict = Depends(require_permission('device.view')),
|
||
):
|
||
"""获取所有区域列表（受角色数据范围限制）"""
|
||
role = current.get('role', 'user')
|
||
query = db.query(distinct(ONUDevice.region)).filter(
|
||
ONUDevice.region.isnot(None),
|
||
ONUDevice.region != ''
|
||
)
|
||
if role == 'area_admin':
|
||
assigned = current.get('assigned_area') or ''
|
||
areas = [a.strip() for a in assigned.split(',') if a.strip()]
|
||
if areas:
|
||
query = query.filter(ONUDevice.region.in_(areas))
|
||
else:
|
||
return []
|
||
elif role == 'school_admin':
|
||
assigned = current.get('assigned_school') or ''
|
||
schools = [s.strip() for s in assigned.split(',') if s.strip()]
|
||
if schools:
|
||
query = query.filter(ONUDevice.school_name.in_(schools))
|
||
else:
|
||
return []
|
||
return [r[0] for r in query.order_by(ONUDevice.region).all()]
|
||
|
||
|
||
@router.get("/schools")
|
||
def get_schools(
|
||
region: str = None,
|
||
db: Session = Depends(get_db),
|
||
_: dict = Depends(require_permission('device.view')),
|
||
):
|
||
"""获取所有学校列表（可按区域筛选）"""
|
||
query = db.query(distinct(ONUDevice.school_name)).filter(
|
||
ONUDevice.school_name.isnot(None),
|
||
ONUDevice.school_name != ''
|
||
)
|
||
if region:
|
||
query = query.filter(ONUDevice.region == region)
|
||
return [r[0] for r in query.order_by(ONUDevice.school_name).all()]
|
||
|
||
|
||
@router.get("/replacements")
|
||
def get_all_replacements(
|
||
region: Optional[str] = Query(None),
|
||
start_date: Optional[str] = Query(None),
|
||
end_date: Optional[str] = Query(None),
|
||
keyword: Optional[str] = Query(None),
|
||
skip: int = Query(0, ge=0),
|
||
limit: int = Query(200, ge=1, le=1000),
|
||
db: Session = Depends(get_db),
|
||
current: dict = Depends(require_permission('device.view')),
|
||
):
|
||
"""获取全量设备更换记录（带位置信息），支持筛选"""
|
||
from datetime import datetime
|
||
query = (
|
||
db.query(DeviceReplacement, ONUDevice)
|
||
.join(ONUDevice, DeviceReplacement.onu_device_id == ONUDevice.id)
|
||
)
|
||
if current.get('role') == 'area_admin' and current.get('assigned_area'):
|
||
areas = [a.strip() for a in current['assigned_area'].split(',') if a.strip()]
|
||
if areas:
|
||
query = query.filter(ONUDevice.region.in_(areas))
|
||
else:
|
||
return {"total": 0, "items": []}
|
||
if region:
|
||
query = query.filter(ONUDevice.region == region)
|
||
if start_date:
|
||
query = query.filter(DeviceReplacement.replaced_at >= datetime.fromisoformat(start_date))
|
||
if end_date:
|
||
query = query.filter(DeviceReplacement.replaced_at <= datetime.fromisoformat(end_date + 'T23:59:59'))
|
||
if keyword:
|
||
kw = f'%{keyword}%'
|
||
query = query.filter(or_(
|
||
ONUDevice.school_name.ilike(kw),
|
||
ONUDevice.region.ilike(kw),
|
||
DeviceReplacement.old_mac.ilike(kw),
|
||
DeviceReplacement.new_mac.ilike(kw),
|
||
DeviceReplacement.operator_name.ilike(kw),
|
||
))
|
||
total = query.count()
|
||
rows = query.order_by(DeviceReplacement.replaced_at.desc()).offset(skip).limit(limit).all()
|
||
return {
|
||
"total": total,
|
||
"items": [
|
||
{
|
||
"id": r.id,
|
||
"replaced_at": r.replaced_at,
|
||
"old_mac": r.old_mac,
|
||
"new_mac": r.new_mac,
|
||
"reason": r.reason,
|
||
"operator_name": r.operator_name,
|
||
"region": d.region,
|
||
"school_name": d.school_name,
|
||
"building": d.building,
|
||
"room_number": d.room_number,
|
||
"onu_device_id": r.onu_device_id,
|
||
}
|
||
for r, d in rows
|
||
],
|
||
}
|
||
|
||
|
||
@router.get("/replacements/export")
|
||
def export_replacements(
|
||
region: Optional[str] = Query(None),
|
||
start_date: Optional[str] = Query(None),
|
||
end_date: Optional[str] = Query(None),
|
||
keyword: Optional[str] = Query(None),
|
||
db: Session = Depends(get_db),
|
||
current: dict = Depends(require_permission('device.view')),
|
||
):
|
||
"""导出更换记录为 CSV"""
|
||
import csv, io
|
||
from datetime import datetime, timedelta
|
||
from fastapi.responses import StreamingResponse
|
||
|
||
query = (
|
||
db.query(DeviceReplacement, ONUDevice)
|
||
.join(ONUDevice, DeviceReplacement.onu_device_id == ONUDevice.id)
|
||
)
|
||
if current.get('role') == 'area_admin' and current.get('assigned_area'):
|
||
areas = [a.strip() for a in current['assigned_area'].split(',') if a.strip()]
|
||
if areas:
|
||
query = query.filter(ONUDevice.region.in_(areas))
|
||
else:
|
||
query = query.filter(False)
|
||
if region:
|
||
query = query.filter(ONUDevice.region == region)
|
||
if start_date:
|
||
query = query.filter(DeviceReplacement.replaced_at >= datetime.fromisoformat(start_date))
|
||
if end_date:
|
||
query = query.filter(DeviceReplacement.replaced_at <= datetime.fromisoformat(end_date + 'T23:59:59'))
|
||
if keyword:
|
||
kw = f'%{keyword}%'
|
||
query = query.filter(or_(
|
||
ONUDevice.school_name.ilike(kw),
|
||
ONUDevice.region.ilike(kw),
|
||
DeviceReplacement.old_mac.ilike(kw),
|
||
DeviceReplacement.new_mac.ilike(kw),
|
||
DeviceReplacement.operator_name.ilike(kw),
|
||
))
|
||
rows = query.order_by(DeviceReplacement.replaced_at.desc()).all()
|
||
|
||
output = io.StringIO()
|
||
writer = csv.writer(output)
|
||
writer.writerow(['更换时间(北京)', '区域', '学校', '楼宇', '房间', '旧MAC', '新MAC', '更换原因', '操作人'])
|
||
for r, d in rows:
|
||
bj_time = (r.replaced_at + timedelta(hours=8)).strftime('%Y-%m-%d %H:%M:%S') if r.replaced_at else ''
|
||
writer.writerow([
|
||
bj_time,
|
||
d.region or '',
|
||
d.school_name or '',
|
||
d.building or '',
|
||
d.room_number or '',
|
||
r.old_mac,
|
||
r.new_mac,
|
||
r.reason or '',
|
||
r.operator_name or '',
|
||
])
|
||
|
||
output.seek(0)
|
||
filename = f"replacement_records_{datetime.now().strftime('%Y%m%d%H%M%S')}.csv"
|
||
return StreamingResponse(
|
||
iter([output.getvalue().encode('utf-8-sig')]),
|
||
media_type='text/csv',
|
||
headers={'Content-Disposition': f'attachment; filename="{filename}"'},
|
||
)
|
||
|
||
|
||
@router.get("/tags")
|
||
def get_all_tags(db: Session = Depends(get_db)):
|
||
"""获取所有不重复的设备标签"""
|
||
from sqlalchemy import func as _func
|
||
rows = db.query(ONUDevice.tags).filter(
|
||
ONUDevice.tags.isnot(None), ONUDevice.tags != ''
|
||
).all()
|
||
tags = set()
|
||
for (tag_str,) in rows:
|
||
for t in tag_str.split(','):
|
||
t = t.strip()
|
||
if t:
|
||
tags.add(t)
|
||
return sorted(tags)
|
||
|
||
|
||
@router.get("/export/csv")
|
||
def export_devices_csv(
|
||
region: Optional[str] = Query(None),
|
||
school_name: Optional[str] = Query(None),
|
||
db: Session = Depends(get_db),
|
||
_: dict = Depends(require_permission('device.view')),
|
||
):
|
||
"""导出设备列表为 CSV"""
|
||
from sqlalchemy import func as _func
|
||
|
||
query = db.query(ONUDevice)
|
||
if region:
|
||
query = query.filter(ONUDevice.region == region)
|
||
if school_name:
|
||
query = query.filter(ONUDevice.school_name == school_name)
|
||
devices = query.order_by(ONUDevice.region, ONUDevice.school_name).all()
|
||
|
||
output = io.StringIO()
|
||
writer = csv.writer(output)
|
||
writer.writerow(["MAC地址", "区域", "学校", "楼宇", "场所类型", "房间号", "端口", "型号", "LOID", "距离(m)", "备注"])
|
||
for d in devices:
|
||
writer.writerow([d.mac_address, d.region or "", d.school_name or "", d.building or "",
|
||
d.place_type or "", d.room_number or "", d.port_id or "", d.model or "",
|
||
d.loid or "", d.distance_m or "", d.notes or ""])
|
||
|
||
output.seek(0)
|
||
return StreamingResponse(
|
||
iter([output.getvalue()]),
|
||
media_type="text/csv",
|
||
headers={"Content-Disposition": "attachment; filename=onu_devices.csv"}
|
||
)
|
||
|
||
|
||
@router.get("/{device_id}", response_model=ONUDeviceResponse)
|
||
def get_device(
|
||
device_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: dict = Depends(require_permission('device.view')),
|
||
):
|
||
"""获取设备详情"""
|
||
device = db.query(ONUDevice).filter(ONUDevice.id == device_id).first()
|
||
if not device:
|
||
from fastapi import HTTPException
|
||
raise HTTPException(status_code=404, detail="设备不存在")
|
||
|
||
latest_status = db.query(DeviceStatusHistory).filter(
|
||
DeviceStatusHistory.onu_device_id == device.id
|
||
).order_by(desc(DeviceStatusHistory.checked_at)).first()
|
||
|
||
olt = db.query(OLTDevice).filter(OLTDevice.id == device.olt_id).first() if device.olt_id else None
|
||
|
||
return ONUDeviceResponse(
|
||
id=device.id,
|
||
mac_address=device.mac_address,
|
||
olt_id=device.olt_id,
|
||
region=device.region,
|
||
school_name=device.school_name,
|
||
building=device.building,
|
||
place_type=device.place_type,
|
||
room_number=device.room_number,
|
||
notes=device.notes,
|
||
status=latest_status.status if latest_status else None,
|
||
distance_m=latest_status.distance_m if latest_status else None,
|
||
slot_number=device.slot_number,
|
||
port_number=device.port_number,
|
||
port_id=device.port_id,
|
||
model=device.model,
|
||
olt_location=olt.location if olt else None,
|
||
created_at=device.created_at
|
||
)
|
||
|
||
|
||
@router.post("/{device_id}/refresh")
|
||
def refresh_device_status(
|
||
device_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: dict = Depends(require_permission('device.check')),
|
||
):
|
||
"""通过 SSH 单独更新一台设备的状态和距离"""
|
||
from app.services.check_service import CheckService
|
||
try:
|
||
service = CheckService(db)
|
||
result = service.check_single_device(device_id)
|
||
return result
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
|
||
class DeviceUpdate(BaseModel):
|
||
region: Optional[str] = None
|
||
school_name: Optional[str] = None
|
||
building: Optional[str] = None
|
||
room_number: Optional[str] = None
|
||
place_type: Optional[str] = None
|
||
notes: Optional[str] = None
|
||
tags: Optional[str] = None
|
||
|
||
|
||
class DeviceReplaceRequest(BaseModel):
|
||
new_mac: str
|
||
reason: Optional[str] = None
|
||
|
||
|
||
@router.delete("/status/all")
|
||
def clear_all_status(
|
||
db: Session = Depends(get_db),
|
||
_: dict = Depends(require_permission('device.delete')),
|
||
):
|
||
"""清空所有设备状态历史记录"""
|
||
db.query(DeviceStatusHistory).delete()
|
||
db.commit()
|
||
return {"message": "已清空所有设备状态"}
|
||
|
||
|
||
@router.put("/{device_id}")
|
||
def update_device(
|
||
device_id: int,
|
||
body: DeviceUpdate,
|
||
db: Session = Depends(get_db),
|
||
_: dict = Depends(require_permission('device.edit')),
|
||
):
|
||
"""更新设备信息（区域、学校、楼宇、房间号、备注）"""
|
||
device = db.query(ONUDevice).filter(ONUDevice.id == device_id).first()
|
||
if not device:
|
||
raise HTTPException(status_code=404, detail="设备不存在")
|
||
device.region = body.region
|
||
device.school_name = body.school_name
|
||
device.building = body.building or None
|
||
device.room_number = body.room_number or None
|
||
device.place_type = body.place_type or None
|
||
device.notes = body.notes or None
|
||
device.tags = body.tags or None
|
||
|
||
# 若该设备 MAC 在 new_devices 待入库列表中，自动移除（已在设备列表中补全信息）
|
||
from app.models.device import NewDevice
|
||
dup_new = db.query(NewDevice).join(
|
||
ONUDevice, NewDevice.onu_device_id == ONUDevice.id
|
||
).filter(ONUDevice.mac_address == device.mac_address).all()
|
||
for nd in dup_new:
|
||
db.delete(nd)
|
||
|
||
db.commit()
|
||
return {"message": "更新成功"}
|
||
|
||
|
||
@router.post("/{device_id}/replace")
|
||
def replace_device(
|
||
device_id: int,
|
||
body: DeviceReplaceRequest,
|
||
db: Session = Depends(get_db),
|
||
current: dict = Depends(require_permission('device.edit')),
|
||
):
|
||
"""更换设备 MAC 地址，并记录更换历史"""
|
||
from datetime import datetime
|
||
device = db.query(ONUDevice).filter(ONUDevice.id == device_id).first()
|
||
if not device:
|
||
raise HTTPException(status_code=404, detail="设备不存在")
|
||
|
||
new_mac_raw = body.new_mac.strip()
|
||
# 接受 xxxx-xxxx-xxxx、xx:xx:xx:xx:xx:xx、xxxxxxxxxxxx 三种格式，统一标准化为小写 xxxx-xxxx-xxxx
|
||
import re
|
||
hex_only = re.sub(r'[:\-]', '', new_mac_raw).lower()
|
||
if not re.match(r'^[0-9a-f]{12}$', hex_only):
|
||
raise HTTPException(status_code=400, detail="MAC 地址格式不正确，支持 xxxx-xxxx-xxxx、xx:xx:xx:xx:xx:xx 或 xxxxxxxxxxxx")
|
||
new_mac = f"{hex_only[0:4]}-{hex_only[4:8]}-{hex_only[8:12]}"
|
||
|
||
# 同时匹配大小写和各种分隔符格式，兼容数据库旧数据
|
||
hex_variants = [
|
||
new_mac,
|
||
hex_only,
|
||
':'.join(hex_only[i:i+2] for i in range(0, 12, 2)),
|
||
'-'.join(hex_only[i:i+2] for i in range(0, 12, 2)),
|
||
new_mac.upper(),
|
||
hex_only.upper(),
|
||
':'.join(hex_only[i:i+2].upper() for i in range(0, 12, 2)),
|
||
'-'.join(hex_only[i:i+2].upper() for i in range(0, 12, 2)),
|
||
]
|
||
|
||
# 若新 MAC 在 new_devices 待入库列表中，先删除（更换后该记录已无意义）
|
||
from app.models.device import NewDevice
|
||
conflict_news = db.query(NewDevice).join(
|
||
ONUDevice, NewDevice.onu_device_id == ONUDevice.id
|
||
).filter(
|
||
ONUDevice.mac_address.in_(hex_variants),
|
||
ONUDevice.id != device_id
|
||
).all()
|
||
conflict_new_onu_ids = {nd.onu_device_id for nd in conflict_news}
|
||
for nd in conflict_news:
|
||
db.delete(nd)
|
||
# 先 flush，让 NewDevice 的 ORM 删除落库，解除对 onu_devices 的外键引用
|
||
db.flush()
|
||
# 同时删除对应的空白 ONU 记录，避免设备列表出现重复 MAC
|
||
if conflict_new_onu_ids:
|
||
from app.models.device import DeviceStatusHistory
|
||
db.query(DeviceStatusHistory).filter(
|
||
DeviceStatusHistory.onu_device_id.in_(conflict_new_onu_ids)
|
||
).delete(synchronize_session=False)
|
||
db.query(ONUDevice).filter(ONUDevice.id.in_(conflict_new_onu_ids)).delete(synchronize_session=False)
|
||
db.flush()
|
||
|
||
# 检查新 MAC 是否已被其他 ONU 设备使用（排除刚刚从 new_devices 删除的临时 ONU）
|
||
existing = db.query(ONUDevice).filter(
|
||
ONUDevice.mac_address.in_(hex_variants),
|
||
ONUDevice.id != device_id,
|
||
ONUDevice.id.notin_(conflict_new_onu_ids) if conflict_new_onu_ids else True,
|
||
).first()
|
||
if existing:
|
||
raise HTTPException(status_code=400, detail="该 MAC 地址已被其他设备使用")
|
||
|
||
# 同步更新库存序列号设备的 onu_device_id 关联（如有）
|
||
from app.models.inventory import SerialDevice
|
||
old_serial = db.query(SerialDevice).filter(SerialDevice.onu_device_id == device_id).first()
|
||
if old_serial:
|
||
old_serial.onu_device_id = None
|
||
old_serial.status = "returned"
|
||
new_serial = db.query(SerialDevice).filter(SerialDevice.mac_address == new_mac).first()
|
||
if new_serial:
|
||
new_serial.onu_device_id = device_id
|
||
new_serial.status = "in_use"
|
||
|
||
record = DeviceReplacement(
|
||
onu_device_id=device_id,
|
||
old_mac=device.mac_address,
|
||
new_mac=new_mac,
|
||
reason=body.reason or None,
|
||
operator_id=current.get("sub", ""),
|
||
operator_name=current.get("username", ""),
|
||
replaced_at=datetime.utcnow(),
|
||
)
|
||
db.add(record)
|
||
device.mac_address = new_mac
|
||
db.commit()
|
||
return {"message": "更换成功", "old_mac": record.old_mac, "new_mac": new_mac}
|
||
|
||
|
||
@router.get("/{device_id}/replacements")
|
||
def get_device_replacements(
|
||
device_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: dict = Depends(require_permission('device.view')),
|
||
):
|
||
"""获取设备更换历史"""
|
||
records = db.query(DeviceReplacement).filter(
|
||
DeviceReplacement.onu_device_id == device_id
|
||
).order_by(DeviceReplacement.replaced_at.desc()).all()
|
||
return [
|
||
{
|
||
"id": r.id,
|
||
"old_mac": r.old_mac,
|
||
"new_mac": r.new_mac,
|
||
"reason": r.reason,
|
||
"operator_name": r.operator_name,
|
||
"replaced_at": r.replaced_at,
|
||
}
|
||
for r in records
|
||
]
|
||
|
||
|
||
|
||
|
||
@router.post("/{device_id}/reboot", response_model=RebootResponse)
|
||
def reboot_device(
|
||
device_id: int,
|
||
db: Session = Depends(get_db),
|
||
current: dict = Depends(require_permission('device.check')),
|
||
):
|
||
"""远程重启 ONU 设备（通过 iMC REST API）"""
|
||
device = db.query(ONUDevice).filter(ONUDevice.id == device_id).first()
|
||
if not device:
|
||
raise HTTPException(status_code=404, detail="设备不存在")
|
||
|
||
role = current.get('role', 'user')
|
||
if role == 'area_admin':
|
||
areas = [a.strip() for a in (current.get('assigned_area') or '').split(',') if a.strip()]
|
||
if device.region not in areas:
|
||
raise HTTPException(status_code=403, detail="无权限操作此区域的设备")
|
||
elif role == 'school_admin':
|
||
schools = [s.strip() for s in (current.get('assigned_school') or '').split(',') if s.strip()]
|
||
if device.school_name not in schools:
|
||
raise HTTPException(status_code=403, detail="无权限操作此学校的设备")
|
||
|
||
try:
|
||
from app.services.imc_service import IMCService
|
||
result = IMCService().reboot_onu(device.mac_address)
|
||
return RebootResponse(**result)
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"重启失败: {str(e)}")
|
||
|
||
|
||
@router.get("/{device_id}/optical-power", response_model=OpticalPowerResponse)
|
||
def get_device_optical_power(
|
||
device_id: int,
|
||
db: Session = Depends(get_db),
|
||
current: dict = Depends(require_permission('device.view')),
|
||
):
|
||
"""获取 ONU 设备光功率信息（通过 iMC REST API）"""
|
||
device = db.query(ONUDevice).filter(ONUDevice.id == device_id).first()
|
||
if not device:
|
||
raise HTTPException(status_code=404, detail="设备不存在")
|
||
|
||
role = current.get('role', 'user')
|
||
if role == 'area_admin':
|
||
areas = [a.strip() for a in (current.get('assigned_area') or '').split(',') if a.strip()]
|
||
if device.region not in areas:
|
||
raise HTTPException(status_code=403, detail="无权限操作此区域的设备")
|
||
elif role == 'school_admin':
|
||
schools = [s.strip() for s in (current.get('assigned_school') or '').split(',') if s.strip()]
|
||
if device.school_name not in schools:
|
||
raise HTTPException(status_code=403, detail="无权限操作此学校的设备")
|
||
|
||
try:
|
||
from app.services.imc_service import IMCService
|
||
data = IMCService().get_optical_power(device.mac_address)
|
||
if data is None:
|
||
raise HTTPException(status_code=502, detail="获取光功率失败，iMC 接口无响应")
|
||
# 记录光功率历史
|
||
try:
|
||
from app.models.device import OpticalPowerHistory
|
||
db.add(OpticalPowerHistory(
|
||
onu_device_id=device_id,
|
||
power_in=data.get("powerIn"),
|
||
power_out=data.get("powerOut"),
|
||
))
|
||
db.commit()
|
||
except Exception:
|
||
pass
|
||
|
||
return OpticalPowerResponse(
|
||
power_in=data.get("powerIn"),
|
||
power_out=data.get("powerOut"),
|
||
bind_mac=data.get("bindMac"),
|
||
dev_id=data.get("devId"),
|
||
epon_dev_name=data.get("eponDevName"),
|
||
olt_if_name=data.get("oltIfName"),
|
||
onu_if_desc=data.get("onuIfDesc"),
|
||
)
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"获取光功率失败: {str(e)}")
|
||
|
||
|
||
@router.get("/{device_id}/onu-events")
|
||
def get_onu_events(
|
||
device_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: dict = Depends(require_permission('device.view')),
|
||
):
|
||
"""查询 ONU 上下线事件记录（SSH 到所属 OLT 执行命令）"""
|
||
device = db.query(ONUDevice).filter(ONUDevice.id == device_id).first()
|
||
if not device:
|
||
raise HTTPException(status_code=404, detail="设备不存在")
|
||
if not device.olt_id:
|
||
raise HTTPException(status_code=400, detail="该设备未关联 OLT，无法查询")
|
||
if not device.port_id:
|
||
raise HTTPException(status_code=400, detail="端口信息缺失，请先更新设备状态")
|
||
|
||
olt = db.query(OLTDevice).filter(OLTDevice.id == device.olt_id).first()
|
||
if not olt:
|
||
raise HTTPException(status_code=404, detail="关联的 OLT 不存在")
|
||
|
||
from app.services.ssh_service import SSHService
|
||
try:
|
||
with SSHService(olt.ip_address, olt.username, olt.password) as ssh:
|
||
events = ssh.get_onu_events(device.port_id)
|
||
return {
|
||
"interface": f"Onu{device.port_id}",
|
||
"olt_location": olt.location,
|
||
"events": events,
|
||
}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"查询失败: {str(e)}")
|
||
|
||
|
||
@router.get("/{device_id}/optical-power-history")
|
||
def get_optical_power_history(
|
||
device_id: int,
|
||
limit: int = Query(20, ge=1, le=100),
|
||
db: Session = Depends(get_db),
|
||
_: dict = Depends(require_permission('device.view')),
|
||
):
|
||
"""获取设备光功率历史记录"""
|
||
from app.models.device import OpticalPowerHistory
|
||
rows = (
|
||
db.query(OpticalPowerHistory)
|
||
.filter(OpticalPowerHistory.onu_device_id == device_id)
|
||
.order_by(OpticalPowerHistory.recorded_at.desc())
|
||
.limit(limit)
|
||
.all()
|
||
)
|
||
return [
|
||
{"power_in": r.power_in, "power_out": r.power_out,
|
||
"recorded_at": r.recorded_at.isoformat() if r.recorded_at else None}
|
||
for r in rows
|
||
]
|