v6ole
d222978ae4
feat(import): 添加区域字段支持和改进模板路径处理 - 在OLT设备导入功能中添加region字段支持，从Excel模板读取区域信息 - 修复模板文件路径问题，使用相对路径动态构建模板文件路径 - 更新导入服务中的验证错误格式，包含行号和MAC地址信息 feat(check): 增强设备检测服务的端口信息同步 - 在CheckService中添加端口和OLT归属信息的同步逻辑 - 改进设备状态检查时的端口信息更新策略，避免用None覆盖现有值 - 扩展返回数据结构，包含端口ID、槽位号、端口号和OLT位置信息 feat(frontend): 添加设备列表刷新冷却机制和端口验证 - 实现设备状态刷新的60秒冷却时间限制，防止频繁操作 - 改进业务下发按钮的启用条件，同时支持port_id或slot_number+port_number组合 - 优化导入结果显示，显示具体的MAC地址信息 feat(olt): 添加重复MAC记录批量清除功能 - 新增批量清除重复MAC记录的功能，支持按OLT分组处理 - 实现SSH端口清除和自动忽略的批量操作流程 - 添加批量操作的进度提示和错误处理机制 ```
80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
"""数据导入 API"""
|
||
from fastapi import APIRouter, UploadFile, File, Depends, Response
|
||
from sqlalchemy.orm import Session
|
||
from app.core.database import get_db
|
||
from app.middleware.permission_middleware import require_permission
|
||
from app.services.import_service import ImportService
|
||
import shutil
|
||
import io
|
||
from openpyxl import Workbook
|
||
|
||
router = APIRouter(prefix="/api/import", tags=["数据导入"])
|
||
|
||
|
||
@router.get("/template")
|
||
def download_template():
|
||
"""下载导入数据模板"""
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "ONU设备导入模板"
|
||
|
||
# 表头：序号|区域|学校名称|楼宇|场所类型|房间号|MAC地址|备注
|
||
headers = ["mac_address", "region", "school_name", "building", "place_type", "room_number", "notes"]
|
||
ws.append(headers)
|
||
|
||
# 示例数据
|
||
example_data = [
|
||
["AA:BB:CC:DD:EE:01", "区域1", "学校1", "1号楼", "教室", "101", ""],
|
||
["AA:BB:CC:DD:EE:02", "区域1", "学校1", "1号楼", "办公室", "102", ""],
|
||
]
|
||
for row in example_data:
|
||
ws.append(row)
|
||
|
||
# 设置列宽
|
||
ws.column_dimensions['A'].width = 20 # mac_address
|
||
ws.column_dimensions['B'].width = 12 # region
|
||
ws.column_dimensions['C'].width = 18 # school_name
|
||
ws.column_dimensions['D'].width = 12 # building
|
||
ws.column_dimensions['E'].width = 12 # place_type
|
||
ws.column_dimensions['F'].width = 12 # room_number
|
||
ws.column_dimensions['G'].width = 20 # notes
|
||
|
||
# 保存到内存
|
||
output = io.BytesIO()
|
||
wb.save(output)
|
||
output.seek(0)
|
||
|
||
return Response(
|
||
content=output.getvalue(),
|
||
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||
headers={"Content-Disposition": "attachment; filename=onu_import_template.xlsx"}
|
||
)
|
||
|
||
|
||
@router.post("/upload")
|
||
async def upload_excel(
|
||
file: UploadFile = File(...),
|
||
olt_id: int = None,
|
||
db: Session = Depends(get_db),
|
||
_: dict = Depends(require_permission('device.import')),
|
||
):
|
||
"""上传并导入 Excel 文件（仅导入 MAC 信息，不关联 OLT）"""
|
||
file_path = f"/tmp/{file.filename}"
|
||
with open(file_path, "wb") as buffer:
|
||
shutil.copyfileobj(file.file, buffer)
|
||
|
||
service = ImportService(db)
|
||
records = service.parse_excel(file_path)
|
||
validation = service.validate_data(records)
|
||
|
||
success_count = 0
|
||
if validation['valid']:
|
||
result = service.import_devices(validation['valid'], olt_id)
|
||
success_count = result['success']
|
||
|
||
return {
|
||
"success": success_count,
|
||
"failed": validation['invalid']
|
||
}
|
||
|